From 31ec74adc1a19d0b163f3c9a261ca6a601ebcafa Mon Sep 17 00:00:00 2001
From: "sgunderson@bigfoot.com" <>
Date: Tue, 13 Feb 2007 20:41:36 +0100
Subject: [PATCH] Start working on multiple senders per thread, using epoll.

---
 jam.c | 123 ++++++++++++++++++++++++++++++++++++++++++++--------------
 1 file changed, 93 insertions(+), 30 deletions(-)

diff --git a/jam.c b/jam.c
index 081b523..4836410 100644
--- a/jam.c
+++ b/jam.c
@@ -8,6 +8,8 @@
 #include <pthread.h>
 #include <sys/socket.h>
 #include <netinet/in.h>
+#include <sys/epoll.h>
+#include <sys/ioctl.h>
 
 unsigned short port = 2007;
 
@@ -20,6 +22,7 @@ unsigned num_sources = 0;
 unsigned room_sources = 0;
 
 unsigned num_senders = 128;
+unsigned num_sockets_per_sender = 16;
 unsigned do_listen = 1;
 
 unsigned long long total_bytes_received = 0;
@@ -32,6 +35,7 @@ const static struct option longopts[] = {
 	{ "source-list", required_argument, NULL, 's' },
 	{ "destination-list", required_argument, NULL, 'd' },
 	{ "num-senders", required_argument, NULL, 'n' },
+	{ "num-sockets-per-sender", required_argument, NULL, 'N' },
 	{ "port", required_argument, NULL, 'p' },
 	{ "sender-only", no_argument, NULL, 'o' },
 	{ NULL, 0, NULL, 0 }
@@ -120,6 +124,9 @@ void parse_options(int argc, char **argv)
 		case 'n':
 			num_senders = atoi(optarg);
 			break;
+		case 'N':
+			num_sockets_per_sender = atoi(optarg);
+			break;
 		case 'p':
 			port = atoi(optarg);
 			break;
@@ -135,54 +142,99 @@ void parse_options(int argc, char **argv)
 	}
 }
 
+void generate_new_sender(int ep_fd)
+{
+	int sock, one = 1;
+	unsigned src_num, dst_num;
+	unsigned long long num_bytes;
+	struct sockaddr_in sin;
+	struct epoll_event ev;
+
+	sock = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP);
+	if (sock == -1) {
+		perror("socket()");
+		exit(1);
+	}
+
+	// find the right parameters
+	src_num = (unsigned)(num_sources * gen_uniform_random());
+	dst_num = (unsigned)(num_destinations * gen_uniform_random());
+	num_bytes = (unsigned)gen_pareto_random(1048576.0, 1.0);
+
+	// FIXME: bind to the right source
+
+	if (ioctl(sock, FIONBIO, &one) == -1) {
+		perror("FIONBIO");
+		exit(1);
+	}
+
+	sin.sin_family = AF_INET;
+	sin.sin_port = htons(port);
+	sin.sin_addr = destinations[dst_num];
+
+	// non-blocking connect (will be detected by epoll later)
+	if (connect(sock, (struct sockaddr *)&sin, sizeof(sin)) == -1) {
+		perror("connect()");
+		exit(1);
+	}
+	
+	// stick it in the epoll set (FIXME: make edge-triggered?)
+	ev.events = EPOLLOUT | EPOLLHUP | EPOLLERR;
+	ev.data.fd = sock;
+	ev.data.u64 = num_bytes;
+
+	if (epoll_ctl(ep_fd, EPOLL_CTL_ADD, sock, &ev) == -1) {
+		perror("EPOLL_CTL_ADD");
+		exit(1);
+	}
+}
+
 void *sender_worker(void *arg)
 {
 	unsigned i;
-	int sock;
 	char buf[65536];
 	unsigned long long bytes_sent = 0;
+	int ep_fd;
+	struct epoll_event *events;
+
+	ep_fd = epoll_create(num_sockets_per_sender);
+	if (ep_fd == -1) {
+		perror("epoll_create");
+		exit(1);
+	}
+
+	// malloc, since there might not be enough room on the stack
+	events = (struct epoll_event *)malloc(sizeof(struct epoll_event) * num_sockets_per_sender);
+	if (events == NULL) {
+		perror("malloc");
+		exit(1);
+	}
 
+	// fill the buffer with random junk
 	for (i = 0; i < 65536; ++i)
 		buf[i] = rand() & 0xff;
 
-	for (i = 0; i < 1000; ++i) {
-		unsigned src_num = (unsigned)(num_sources * gen_uniform_random());
-		unsigned dst_num = (unsigned)(num_destinations * gen_uniform_random());
-		unsigned num_bytes = (unsigned)gen_pareto_random(1048576.0, 1.0);
-		struct sockaddr_in sin;
+	// allocate all the senders
+	for (i = 0; i < num_sockets_per_sender; ++i)
+		generate_new_sender(ep_fd);
 
-		sock = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP);
-		if (sock == -1) {
-			perror("socket()");
-			exit(1);
-		}
-		
-		// FIXME: bind to the right source
-		
-		sin.sin_family = AF_INET;
-		sin.sin_port = htons(port);
-		sin.sin_addr = destinations[dst_num];
-	
-		if (connect(sock, (struct sockaddr *)&sin, sizeof(sin)) == -1) {
-			perror("connect()");
-			exit(1);
-		}
-		
-		while (num_bytes > 0) {
-			unsigned bytes_to_send = num_bytes;
+	for ( ;; ) {
+		int ret = epoll_wait(ep_fd, events, num_sockets_per_sender, -1);
+		for (i = 0; i < ret; ++i) {
+			unsigned bytes_to_send = events[i].data.u64;
 			unsigned ret;
 
 			if (bytes_to_send > 65536) {
 				bytes_to_send = 65536;	
 			}
 
-			ret = send(sock, buf, bytes_to_send, MSG_NOSIGNAL);
+			ret = send(events[i].data.fd, buf, bytes_to_send, MSG_NOSIGNAL);
 			if (ret == -1) {
 				perror("send()");
 				exit(1);
 			}
 
-			num_bytes -= ret;
+			events[i].data.u64 -= ret;
 			bytes_sent += ret;
 
 			// update the central counter after every 1MB (8ms
@@ -194,15 +246,26 @@ void *sender_worker(void *arg)
 
 				bytes_sent = 0;
 			}
+			
+			if (events[i].data.u64 == 0) {
+				close(events[i].data.fd);
+				if (epoll_ctl(ep_fd, EPOLL_CTL_DEL, events[i].data.fd, NULL) == -1) {
+					perror("EPOLL_CTL_ADD");
+					exit(1);
+				}
+
+				generate_new_sender(ep_fd);
+			}
 		}
-
-		close(sock);
 	}
-			
+
 	pthread_mutex_lock(&send_mutex);
 	total_bytes_sent += bytes_sent;
 	pthread_mutex_unlock(&send_mutex);
 
+	free(events);
+	close(ep_fd);
+
 	pthread_exit(0);
 }
 
-- 
2.39.2