Start working on multiple senders per thread, using epoll.

diff --git a/jam.c b/jam.c
index 081b5232d52ee0cc295765c0de60d6e8cc7b9ea2..4836410e5552cc643b187c73f243f6b8ece92c8c 100644 (file)
--- 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);
 }