X-Git-Url: https://git.sesse.net/?a=blobdiff_plain;f=jam.c;h=5faf0379d4b7eb13299dd48595641d935e6e409a;hb=2f5350f30cbfeebde0bd287d16fb118639536493;hp=f7417e373b5583e852d24b7a7378c9def90f2be3;hpb=f37bf230ab686ac4c964758666406afa1319d0b7;p=jam

diff --git a/jam.c b/jam.c
index f7417e3..5faf037 100644
--- a/jam.c
+++ b/jam.c
@@ -10,6 +10,7 @@
 #include <netinet/in.h>
 #include <sys/epoll.h>
 #include <sys/ioctl.h>
+#include <errno.h>
 
 unsigned short port = 2007;
 
@@ -113,7 +114,7 @@ void parse_options(int argc, char **argv)
 	int option_index = 0;
 
 	for ( ;; ) {
-		int c = getopt_long(argc, argv, "s:d:n:p:o", longopts, &option_index); 
+		int c = getopt_long(argc, argv, "s:d:n:N:p:o", longopts, &option_index); 
 		switch (c) {
 		case 's':
 			read_ip_list(optarg, &sources, &num_sources, &room_sources);
@@ -142,14 +143,19 @@ void parse_options(int argc, char **argv)
 	}
 }
 
-void generate_new_sender(int ep_fd)
+struct sender {
+	int fd;
+	unsigned long long bytes_left;
+};
+
+void generate_new_sender(int ep_fd, struct sender *s)
 {
 	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()");
@@ -173,15 +179,17 @@ void generate_new_sender(int ep_fd)
 	sin.sin_addr = destinations[dst_num];
 
 	// non-blocking connect (will be detected by epoll later)
-	if (connect(sock, (struct sockaddr *)&sin, sizeof(sin)) == -1) {
+	if (connect(sock, (struct sockaddr *)&sin, sizeof(sin)) == -1 && errno != EINPROGRESS) {
 		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;
+	ev.data.ptr = s;
+
+	s->fd = sock;
+	s->bytes_left = num_bytes;
 
 	if (epoll_ctl(ep_fd, EPOLL_CTL_ADD, sock, &ev) == -1) {
 		perror("EPOLL_CTL_ADD");
@@ -215,26 +223,42 @@ void *sender_worker(void *arg)
 		buf[i] = rand() & 0xff;
 
 	// allocate all the senders
-	for (i = 0; i < num_sockets_per_sender; ++i)
-		generate_new_sender(ep_fd);
+	for (i = 0; i < num_sockets_per_sender; ++i) {
+		struct sender *s = (struct sender *)malloc(sizeof(struct sender));
+		if (s == NULL) {
+			perror("malloc()");
+			exit(1);
+		}
+
+		generate_new_sender(ep_fd, s);
+	}
 
 	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;
+		int num_active = epoll_wait(ep_fd, events, num_sockets_per_sender, -1);
+		if (num_active == -1) {
+			perror("epoll_wait");
+			exit(1);
+		}
+
+		for (i = 0; i < num_active; ++i) {
+			struct sender *s = (struct sender *)events[i].data.ptr;
+			unsigned long long bytes_to_send = s->bytes_left;
 			unsigned ret;
 
 			if (bytes_to_send > 65536) {
 				bytes_to_send = 65536;	
 			}
 
-			ret = send(events[i].data.fd, buf, bytes_to_send, MSG_NOSIGNAL);
+			ret = send(s->fd, buf, bytes_to_send, MSG_NOSIGNAL);
 			if (ret == -1) {
+				if (errno == EAGAIN)
+					continue;
+
 				perror("send()");
 				exit(1);
 			}
 
-			events[i].data.u64 -= ret;
+			s->bytes_left -= ret;
 			bytes_sent += ret;
 
 			// update the central counter after every 1MB (8ms
@@ -247,14 +271,13 @@ 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) {
+			if (s->bytes_left == 0) {
+				if (epoll_ctl(ep_fd, EPOLL_CTL_DEL, s->fd, NULL) == -1) {
 					perror("EPOLL_CTL_ADD");
 					exit(1);
 				}
-
-				generate_new_sender(ep_fd);
+				close(s->fd);
+				generate_new_sender(ep_fd, s);
 			}
 		}
 	}