It looks like send() can give us EAGAIN even if epoll() said the socket

[jam] / jam.c

diff --git a/jam.c b/jam.c
index 18839c834daaa4ad82ad181f81ddd3be97e42098..581d51c91200a69696aabe8d39e9519b19c250ca 100644 (file)
--- a/jam.c
+++ b/jam.c
@@ -114,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);
@@ -143,6 +143,11 @@ void parse_options(int argc, char **argv)
        }
 }
 
+struct sender {
+       int fd;
+       unsigned long long bytes_left;
+};
+
 void generate_new_sender(int ep_fd)
 {
        int sock, one = 1;
@@ -150,6 +155,13 @@ void generate_new_sender(int ep_fd)
        unsigned long long num_bytes;
        struct sockaddr_in sin;
        struct epoll_event ev;
+       struct sender *s;
+       
+       s = (struct sender *)malloc(sizeof(struct sender));
+       if (s == NULL) {
+               perror("malloc()");
+               exit(1);
+       }
 
        sock = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP);
        if (sock == -1) {
@@ -181,8 +193,10 @@ void generate_new_sender(int ep_fd)
        
        // 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");
@@ -227,20 +241,24 @@ void *sender_worker(void *arg)
                }
 
                for (i = 0; i < num_active; ++i) {
-                       unsigned long long bytes_to_send = events[i].data.u64;
+                       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
@@ -253,13 +271,14 @@ 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);
                                }
+                               close(s->fd);
 
+                               free(s);
                                generate_new_sender(ep_fd);
                        }
                }