supafly is a synthetic application for "proving" the point that evaluating

apps/protocols that use X cpu on PlanetLab can also add network artifacts.

sfmiddleman sits in the middle as a compute-intensive proxy.  sfsender
sends the middleman n "messages" each composed of m "blocks", and can
pause for t1 and t2 times at microsecond granularity after each block send
and msg send.  sfmiddleman reads blocks and does operations on them
(default is decrypt/encrypt, can also do only encrypt), then forwards them
to all connected receivers.  You guessed it, sfreceiver connects to
sfmiddleman and reads blocks and prints timestamps.  For each op on a
block, sfmiddleman notes the number of microseconds taken by the crypto
ops, and the timestamp when it/they finished, and prints those.

There is one known bug.  I'll fix it in the morning.  If there are others,
I'll get them then too.

pelab/supafly/GNUmakefile

+CC = gcc
+CFLAGS += -g -lssl
+
+all: sfmiddleman sfsender sfreceiver
+
+sfmiddleman: sfmiddleman.c util.c crypto.c
+	$(CC) -o sfmiddleman $(CFLAGS) sfmiddleman.c util.c crypto.c
+
+sfsender: sfsender.c util.c crypto.c
+	$(CC) -o sfsender $(CFLAGS) sfsender.c util.c crypto.c
+
+sfreceiver: sfreceiver.c util.c
+	$(CC) -o sfreceiver $(CFLAGS) sfreceiver.c util.c
+
+clean:
+	rm -f sfmiddleman sfsender sfreceiver

pelab/supafly/crypto.c

+#include <stdio.h>
+#include <time.h>
+#include <sys/types.h>
+#include <stdlib.h>
+#include <string.h>
+#include <unistd.h>
+#include <errno.h>
+
+#include "crypto.h"
+
+DES_cblock *sgenkey() {
+    DES_cblock *retval;
+
+    retval = (DES_cblock *)malloc(sizeof(DES_cblock));
+    if (retval != NULL) {
+	DES_random_key(retval);
+    }
+
+    return retval;
+}
+
+DES_cblock *sgeniv() {
+    DES_cblock *retval;
+
+    retval = (DES_cblock *)malloc(sizeof(DES_cblock));
+    if (retval != NULL) {
+	DES_random_key(retval);
+    }
+
+    return retval;
+}
+
+/* note: if IV is null, don't do cbc; otherwise, do it. */
+int sencrypt(unsigned char *input,unsigned char *output,
+	     unsigned int len,DES_cblock *k1,DES_cblock *k2,
+	     DES_cblock *iv
+	     ) {
+    DES_key_schedule s1,s2;
+    int i;
+    
+    if (len%8 != 0) {
+	return -1;
+    }
+
+    /* the idea is to process the keys each time to keep them out of memory
+     * if possible (besides, it's easier)
+     */
+
+    DES_set_key_checked(k1,&s1);
+    DES_set_key_checked(k2,&s2);
+    
+    if (iv == NULL) {
+	/* actually encrypt, now */
+	//printf("%d\n",len/8);
+	for (i = 0; i < len; i+=8) {
+	    //printf("encrypt block %d\n",i);
+	    DES_ecb2_encrypt(input+i,output+i,&s1,&s2,DES_ENCRYPT);
+	}
+    }
+    else {
+	DES_ede2_cbc_encrypt(input,output,len,&s1,&s2,iv,DES_ENCRYPT);
+    }
+
+    return 0;
+
+}
+
+int sdecrypt(unsigned char *input,unsigned char *output,
+	     unsigned int len,DES_cblock *k1,DES_cblock *k2,
+	     DES_cblock *iv
+	     ) {
+    DES_key_schedule s1,s2;
+    int i;
+
+    if (len%8 != 0) {
+	return -1;
+    }
+
+    /* the idea is to process the keys each time to keep them out of memory
+     * if possible (besides, it's easier)
+     */
+
+    DES_set_key_checked(k1,&s1);
+    DES_set_key_checked(k2,&s2);
+    //printf("blah2\n");
+    if (iv == NULL) {
+	//printf("blah3\n");
+	/* actually encrypt, now */
+	for (i = 0; i < len; i+=8) {
+	    //printf("decrypt block %d\n",i);
+	    DES_ecb2_encrypt(input+i,output+i,&s1,&s2,DES_DECRYPT);
+	}
+    }
+    else {
+	//printf("blah4\n");
+	DES_ede2_cbc_encrypt(input,output,len,&s1,&s2,iv,DES_DECRYPT);
+    }
+
+    return 0;
+}

pelab/supafly/crypto.h

+#ifndef __CRYPTO_H__
+#define __CRYPTO_H__
+
+#include <openssl/rand.h>
+#include <openssl/des.h>
+
+DES_cblock *sgenkey();
+DES_cblock *sgeniv();
+/* note: if IV is null, don't do cbc; otherwise, do it. */
+int sencrypt(unsigned char *input,unsigned char *output,
+	     unsigned int len,DES_cblock *k1,DES_cblock *k2,
+	     DES_cblock *iv);
+int sdecrypt(unsigned char *input,unsigned char *output,
+	     unsigned int len,DES_cblock *k1,DES_cblock *k2,
+	     DES_cblock *iv);
+
+
+#endif /* __CRYPTO_H__ */

pelab/supafly/defs.h

+#ifndef __DEFS_H__
+#define __DEFS_H__
+
+#include <sys/time.h>
+#include <sys/types.h>
+#include <unistd.h>
+#include <stdlib.h>
+#include <netinet/in.h>
+#include <arpa/inet.h>
+#include <netdb.h>
+#include <sys/socket.h>
+#include <stdio.h>
+#include <string.h>
+#include <errno.h>
+#include <sys/select.h>
+
+/** sane systems honor this as 32 bits regardless */
+/**
+ * each sent packet group (allowing for app/ip frag) must prefix a new
+ * unit with the length of crap being sent.  This might be useful...
+ */
+typedef uint32_t slen_t;
+
+#define MIDDLEMAN_SEND_CLIENT_PORT 6878
+#define MIDDLEMAN_RECV_CLIENT_PORT 6888
+
+#define MIDDLEMAN_MAX_CLIENTS 32
+
+#define MAX_NAME_LEN 255
+
+#endif /* __DEFS_H__ */

pelab/supafly/sfmiddleman.c

+#include <stdio.h>
+#include <fcntl.h>
+
+#include "defs.h"
+#include "crypto.h"
+
+/** 
+ * globals
+ */
+char *optarg;
+int optind,opterr,optopt;
+
+int block_size = 4096;
+int block_count = 1024;
+int only_encrypt = 0;
+int debug = 0;
+char *srv_host = "localhost";
+short srv_send_port = MIDDLEMAN_SEND_CLIENT_PORT;
+short srv_recv_port = MIDDLEMAN_RECV_CLIENT_PORT;
+
+char *deadbeef = "deadbeef";
+
+/**
+ * defs
+ */
+void usage(char *bin) {
+    fprintf(stdout,
+	    "USAGE: %s -scohSRud  (option defaults in parens)\n",
+	    bin
+	    );
+}
+
+void parse_args(int argc,char **argv) {
+    int c;
+    char *ep = NULL;
+
+    while ((c = getopt(argc,argv,"s:c:oud")) != -1) {
+	switch(c) {
+	case 's':
+	    block_size = (int)strtol(optarg,&ep,10);
+	    if (ep == optarg) {
+		usage(argv[0]);
+		exit(-1);
+	    }
+	    break;
+	case 'c':
+	    block_count = (int)strtol(optarg,&ep,10);
+	    if (ep == optarg) {
+		usage(argv[0]);
+		exit(-1);
+	    }
+	    break;
+	case 'h':
+	    srv_host = optarg;
+	    break;
+	case 'S':
+	    srv_send_port = (short)strtol(optarg,&ep,10);
+	    if (ep == optarg) {
+		usage(argv[0]);
+		exit(-1);
+	    }
+	    break;
+	case 'R':
+	    srv_recv_port = (short)strtol(optarg,&ep,10);
+	    if (ep == optarg) {
+		usage(argv[0]);
+		exit(-1);
+	    }
+	    break;
+	case 'o':
+	    only_encrypt = 1;
+	    break;
+	case 'u':
+	    usage(argv[0]);
+	    exit(0);
+	case 'd':
+	    ++debug;
+	    break;
+	default:
+	    break;
+	}
+    }
+
+    /** arg check */
+    if (block_size % 8 != 0) {
+	fprintf(stderr,"block_size must be divisible by 8!\n");
+	exit(-1);
+    }
+
+    return;
+
+}
+
+int main(int argc,char **argv) {
+    int srv_send_sock,srv_recv_sock;
+    int retval,ptype;
+    int i,j;
+    struct sockaddr_in srv_send_sa,srv_recv_sa;
+    int recv_clientfds[MIDDLEMAN_MAX_CLIENTS];
+    char *recv_clientsas[MIDDLEMAN_MAX_CLIENTS];
+    int recv_client_cnt = 0;
+    int max_fd = 0;
+    int send_client_fd = -1;
+    int c;
+    char *buf = NULL;
+    char *outbuf = NULL;
+    fd_set rfds;
+    fd_set static_rfds;
+    int current_send_rc;
+
+    /* grab some quick args */
+    parse_args(argc,argv);
+
+    /* initialize ourself... */
+    srand(time(NULL));
+
+    /* grab the bufs... */
+    if ((buf = (char *)malloc(sizeof(char)*block_size)) == NULL) {
+	efatal("no memory for data buf");
+    }
+    if ((outbuf = (char *)malloc(sizeof(char)*block_size)) == NULL) {
+	efatal("no memory for output data buf");
+    }
+
+    /* fill with deadbeef */
+    for (i = 0; i < block_size; i += 8) {
+	memcpy(&buf[i],deadbeef,8);
+    }
+    
+    /* setup the sockaddrs */
+    if ((retval = fill_sockaddr(srv_host,srv_send_port,&srv_send_sa)) != 0) {
+	if (retval == -1) {
+	    fatal("bad port");
+	}
+	else {
+	    efatal("host lookup failed");
+	}
+    }
+    if ((retval = fill_sockaddr(srv_host,srv_recv_port,&srv_recv_sa)) != 0) {
+	if (retval == -1) {
+	    fatal("bad port");
+	}
+	else {
+	    efatal("host lookup failed");
+	}
+    }
+
+    /* startup server... */
+    if ((srv_send_sock = socket(PF_INET,SOCK_STREAM,0)) == -1) {
+	efatal("could not get send listen socket");
+    }
+    if ((srv_recv_sock = socket(PF_INET,SOCK_STREAM,0)) == -1) {
+	efatal("could not get send listen socket");
+    }
+
+    if (bind(srv_send_sock,
+	     (struct sockaddr *)&srv_send_sa,
+	     sizeof(struct sockaddr_in)
+	     ) < 0) {
+	efatal("could not bind");
+    }
+    if (bind(srv_recv_sock,
+	     (struct sockaddr *)&srv_recv_sa,
+	     sizeof(struct sockaddr_in)
+	     ) < 0) {
+	efatal("could not bind");
+    }
+
+    /* only listen for len 1 cause we won't accept on this guy if we 
+     * already have a sender
+     */
+    if (listen(srv_send_sock,1) < 0) {
+	efatal("could not listen");
+    }
+    /* more sane listen queue for this one */
+    if (listen(srv_recv_sock,8) < 0) {
+	efatal("could not listen");
+    }
+
+    /* daemonize... */
+    if (!debug) {
+	daemon(0,0);
+    }
+
+    for (i = 0; i < MIDDLEMAN_MAX_CLIENTS; ++i) {
+	recv_clientfds[i] = -1;
+	recv_clientsas[i] = (char *)malloc(MAX_NAME_LEN + 1);
+    }
+
+    FD_ZERO(&static_rfds);
+    FD_SET(srv_send_sock,&static_rfds);
+    FD_SET(srv_recv_sock,&static_rfds);
+    max_fd = (srv_send_sock > srv_recv_sock) ? srv_send_sock : srv_recv_sock;
+    
+    /* listen and read forever */
+    while (1) {
+	/* reset fdsets */
+	memcpy(&rfds,&static_rfds,sizeof(static_rfds));
+
+	retval = select(max_fd+1,&rfds,NULL,NULL,NULL);
+	
+	if (retval > 0) {
+	    if (FD_ISSET(srv_send_sock,&rfds) 
+		&& send_client_fd < 0) {
+		struct sockaddr_in client_sin;
+		socklen_t slen;
+		
+		slen = sizeof(client_sin);
+		
+		if ((send_client_fd = accept(srv_send_sock,
+					     (struct sockaddr *)&client_sin,
+					     &slen)) < 0) {
+		    warn("accept failed");
+		}
+		
+		if (send_client_fd > max_fd) {
+		    max_fd = send_client_fd;
+		}
+		
+		char *addr = inet_ntoa(client_sin.sin_addr);
+		int addrlen = strlen(addr);
+		
+		if (debug > 1) {
+		    fprintf(stdout,
+			    "DEBUG: (send client!) connect from %s:%d\n",
+			    addr,
+			    ntohs(client_sin.sin_port));
+		}
+
+		/* add the new send client for consideration */
+		FD_SET(send_client_fd,&static_rfds);
+
+		/* zero the counter */
+		current_send_rc = 0;
+	    }
+	    else if (FD_ISSET(srv_recv_sock,&rfds)) {
+		struct sockaddr_in client_sin;
+		socklen_t slen;
+		int client_fd;
+
+		slen = sizeof(client_sin);
+
+		if ((client_fd = accept(srv_recv_sock,
+					(struct sockaddr *)&client_sin,
+					&slen)) < 0) {
+		    warn("accept failed");
+		}
+		else if (recv_client_cnt >= MIDDLEMAN_MAX_CLIENTS) {
+		    warn("already at max clients");
+		    close(client_fd);
+		}
+		else {
+		    /* add new recv client... */
+		    for (i = 0; i < MIDDLEMAN_MAX_CLIENTS; ++i) {
+			if (recv_clientfds[i] == -1) {
+			    break;
+			}
+		    }
+		    
+		    recv_clientfds[i] = client_fd;
+		    if (client_fd > max_fd) {
+			max_fd = client_fd;
+		    }
+		    
+		    FD_SET(client_fd,&static_rfds);
+		    
+		    char *addr = inet_ntoa(client_sin.sin_addr);
+		    int addrlen = strlen(addr);
+		    
+		    if (debug > 1) {
+			fprintf(stdout,
+				"DEBUG: connect from %s:%d\n",
+				addr,
+				ntohs(client_sin.sin_port));
+		    }
+		    
+		    strncpy(recv_clientsas[i],
+			    addr,
+			    (addrlen > MAX_NAME_LEN)?MAX_NAME_LEN:addrlen);
+		    /* null term if strncpy couldn't */
+		    if (addrlen > MAX_NAME_LEN) {
+			recv_clientsas[i][MAX_NAME_LEN] = '\0';
+		    }
+		    
+		    ++recv_client_cnt;
+		}
+	    }
+	    else if (send_client_fd > 0 
+		     && FD_ISSET(send_client_fd,&rfds)) {
+		/**
+		 * ok, here we need to read a block from this guy for awhile,
+		 * then [decrypt/munge]/encrypt (according to command line
+		 * arg), then forward the block to all the recv clients.
+		 */
+		
+		/**
+		 * we don't fork or spin a thread for this, since none of 
+		 * the clients will be writing and I don't have time to
+		 * write any app-level buffering to process the sender's
+		 * packets.  The sender should just have a fixed rate.  :-)
+		 * We'll see how that works out.
+		 */
+
+		/**
+		 * Of course, we put send_client_fd in nonblocking mode, so
+		 * we read what we can when we can...
+		 */
+
+		retval = read(send_client_fd,
+			      &buf[current_send_rc],
+			      block_size - current_send_rc);
+
+		if (retval < 0) {
+		    if (errno == ECONNRESET) {
+			/* dead sender */
+			close(send_client_fd);
+			FD_CLR(send_client_fd,&static_rfds);
+			send_client_fd = -1;
+
+			current_send_rc = 0;
+
+			if (debug > 1) {
+			    fprintf(stderr,
+				    "DEBUG: (send client!) disconnect from %s\n",
+				    recv_clientsas[i]);
+			}
+		    }
+		    else if (errno == EAGAIN) {
+			// ignore, doh
+		    }
+		}
+		else {
+		    current_send_rc += retval;
+		}
+
+
+		if (current_send_rc == block_size) {
+		    current_send_rc = 0;
+
+		    /* end of block, do the encryption op and note times... */
+		    if (debug > 1) {
+			fprintf(stderr,
+				"DEBUG: read %d bytes (a block)\n",
+				block_size);
+		    }
+
+		    DES_cblock *iv = sgeniv();
+		    DES_cblock *k1 = sgenkey();
+		    DES_cblock *k2 = sgenkey();
+		    struct timeval t1,t2,t3;
+
+		    if (only_encrypt) {
+			gettimeofday(&t1,NULL);
+			sencrypt(buf,outbuf,block_size,
+				 k1,k2,iv);
+			gettimeofday(&t2,NULL);
+		    }
+		    else {
+			/* do a decrypt, then choose new iv, then encrypt. */
+			gettimeofday(&t1,NULL);
+			sdecrypt(buf,outbuf,block_size,
+				 k1,k2,iv);
+			iv = sgeniv();
+			sencrypt(buf,outbuf,block_size,
+				 k1,k2,iv);
+			gettimeofday(&t2,NULL);
+		    }
+
+		    t3.tv_sec = t2.tv_sec - t1.tv_sec;
+		    t3.tv_usec = t2.tv_usec - t1.tv_usec;
+		    if (t3.tv_usec < 0) {
+			t3.tv_sec--;
+			t3.tv_usec += 1000000;
+		    }
+		    fprintf(stdout,
+			    "BLOCKTIME(%s): %d at %.6f\n",
+			    (only_encrypt)?"e":"de",
+			    t3.tv_sec * 1000 + t3.tv_usec,
+			    t2.tv_sec + t2.tv_usec / 1000000.0f);
+		    fflush(stdout);
+		    
+		    if (debug > 1) {
+			fprintf(stderr,
+				"DEBUG: sending to clients\n");
+		    }
+
+		    /* send to clients */
+		    for (i = 0; i < MIDDLEMAN_MAX_CLIENTS; ++i) {
+			if (recv_clientfds[i] > -1) {
+			    retval = write(recv_clientfds[i],
+					   outbuf,
+					   block_size);
+			    if (retval < 0) {
+				ewarn("client write failed");
+			    }
+			    else if (retval != block_size) {
+				fprintf(stderr,
+					"ERROR: wrote only %d bytes to client %d!\n",
+					retval,i);
+			    }
+			    else {
+				if (debug > 1) {
+				    fprintf(stderr,
+					    "DEBUG: wrote all block_size %d bytes to client %d!\n",
+					    block_size,i);
+				}
+			    }
+			}
+		    }
+
+		}
+				
+	    }
+	    else {
+		/**
+		 * If we get here, assume that one of the recv clients
+		 * has died (cause they are not supposed to write anything
+		 * to us)... so nuke the doofus!
+		 */
+		for (i = 0; i < MIDDLEMAN_MAX_CLIENTS; ++i) {
+		    if (recv_clientfds[i] > -1) {
+			if (FD_ISSET(recv_clientfds[i],&rfds)) {
+			    /* read a block, or as much as possible */
+			    //retval = read(recv_clientfds[i],buf,block_size);
+
+			    /* dead client, pretty much */
+			    //if (retval <= 0) {
+				if (debug > 1) {
+				    fprintf(stderr,
+					    "DEBUG: disconnect from %s\n",
+					    recv_clientsas[i]);
+				}
+
+				close(recv_clientfds[i]);
+				FD_CLR(recv_clientfds[i],&static_rfds);
+				recv_clientfds[i] = -1;
+
+				--recv_client_cnt;
+			    //}
+			    //else if (debug > 2 ) {
+				//fprintf(stdout,
+				//"DEBUG: read %d bytes from %s\n",
+				//retval,
+				//recv_clientsas[i]);
+			    //}
+			}
+		    }
+		}
+	    }
+	}
+	else if (retval < 0) {
+	    /* error... */
+	    ewarn("error in select");
+	}
+    }
+
+    return -1;
+
+}

pelab/supafly/sfreceiver.c

+#include <stdio.h>
+
+#include "defs.h"
+#include "crypto.h"
+
+/** 
+ * globals
+ */
+char *optarg;
+int optind,opterr,optopt;
+
+/** -s <bytes>*/
+int block_size = 4096;
+/** -m <middleman hostname> */
+char *middleman_host = "localhost";
+/** -M <middleman hostport> */
+short middleman_port = MIDDLEMAN_RECV_CLIENT_PORT;