Implement receiving from multiple sources using a single system call.

diff --git a/src/libswift/channel.cpp b/src/libswift/channel.cpp
index 2828e86..4fed7a5 100644 (file)
--- a/src/libswift/channel.cpp
+++ b/src/libswift/channel.cpp
@@ -280,21 +280,26 @@ int Channel::SendTo (evutil_socket_t sock, const Address& addr, struct evbuffer
     return r;
 }
 
-int Channel::RecvFrom (evutil_socket_t sock, Address& addr, struct evbuffer *evb) {
+int Channel::RecvFrom (evutil_socket_t sock, Address& addr, struct evbuffer **evb) {
     socklen_t addrlen = sizeof(struct sockaddr_mptp) + addr.addr->count * sizeof(mptp_dest);
-    struct evbuffer_iovec vec;
-    if (evbuffer_reserve_space(evb, SWIFT_MAX_RECV_DGRAM_SIZE, &vec, 1) < 0) {
-       print_error("error on evbuffer_reserve_space");
-       return 0;
-    }
-       struct iovec iov[1];
+       int count = addr.addr->count;
+    struct evbuffer_iovec vec[count];
+       for (int i=0; i<count; ++i) {
+               if (evbuffer_reserve_space(evb[i], SWIFT_MAX_RECV_DGRAM_SIZE, &vec[i], 1) < 0) {
+                       print_error("error on evbuffer_reserve_space");
+                       return 0;
+               }
+       }
+       struct iovec iov[count];
        struct msghdr msg;
        memset(&msg, 0, sizeof(msg));
        memset(&iov, 0, sizeof(iov));
-       iov[0].iov_base = vec.iov_base;
-       iov[0].iov_len = SWIFT_MAX_RECV_DGRAM_SIZE;
+       for (int i=0; i<count; ++i) {
+               iov[i].iov_base = vec[i].iov_base;
+               iov[i].iov_len = SWIFT_MAX_RECV_DGRAM_SIZE;
+       }
        msg.msg_iov = iov;
-       msg.msg_iovlen = 1;
+       msg.msg_iovlen = addr.addr->count;
        msg.msg_name = addr.addr;
        msg.msg_namelen = addrlen;
        int length = recvmsg(sock, &msg, 0);
@@ -316,11 +321,13 @@ int Channel::RecvFrom (evutil_socket_t sock, Address& addr, struct evbuffer *evb
         else
                print_error("error on recv");
     }
-    vec.iov_len = length;
-    if (evbuffer_commit_space(evb, &vec, 1) < 0)  {
-        length = 0;
-        print_error("error on evbuffer_commit_space");
-    }
+       for (int i=0; i<addr.addr->count; ++i) {
+               vec[i].iov_len = iov[i].iov_len;
+               if (evbuffer_commit_space(evb[i], &vec[i], 1) < 0)  {
+                       length = 0;
+                       print_error("error on evbuffer_commit_space");
+               }
+       }
     global_dgrams_down++;
     global_raw_bytes_down+=length;
     Time();

diff --git a/src/libswift/sendrecv.cpp b/src/libswift/sendrecv.cpp
index e64d3ee..5961ad5 100644 (file)
--- a/src/libswift/sendrecv.cpp
+++ b/src/libswift/sendrecv.cpp
@@ -914,84 +914,98 @@ void Channel::LibeventReceiveCallback(evutil_socket_t fd, short event, void *arg
     event_add(&evrecv, NULL);
 }
 
+#define NUM_DATAGRAMS 10
+
 void    Channel::RecvDatagram (evutil_socket_t socket) {
-    struct evbuffer *evb = evbuffer_new();
+    struct evbuffer *pevb[NUM_DATAGRAMS];
+       for (int i=0; i<NUM_DATAGRAMS; ++i)
+               pevb[i] = evbuffer_new();
     Address addr;
-    RecvFrom(socket, addr, evb);
-    size_t evboriglen = evbuffer_get_length(evb);
+       //FIXME: make this more readable
+       free(addr.addr);
+       addr.addr = (struct sockaddr_mptp *) calloc(1, sizeof(struct sockaddr_mptp) + NUM_DATAGRAMS * sizeof(struct mptp_dest));
+       addr.addr->count = NUM_DATAGRAMS;
+    RecvFrom(socket, addr, pevb);
+       int i = 0;
+       for (; i<addr.addr->count; ++i) {
+               struct evbuffer *evb = pevb[i];
+               size_t evboriglen = evbuffer_get_length(evb);
 #define return_log(...) { fprintf(stderr,__VA_ARGS__); evbuffer_free(evb); return; }
-    if (evbuffer_get_length(evb)<4)
-        return_log("socket layer weird: datagram shorter than 4 bytes from %s (prob ICMP unreach)\n",addr.str());
-    uint32_t mych = evbuffer_remove_32be(evb);
-    Sha1Hash hash;
-    Channel* channel = NULL;
-    if (mych==0) { // peer initiates handshake
-        if (evbuffer_get_length(evb)<1+4+1+4+Sha1Hash::SIZE)
-            return_log ("%s #0 incorrect size %i initial handshake packet %s\n",
-                        tintstr(),(int)evbuffer_get_length(evb),addr.str());
-        uint8_t hashid = evbuffer_remove_8(evb);
-        if (hashid!=SWIFT_HASH)
-            return_log ("%s #0 no hash in the initial handshake %s\n",
-                        tintstr(),addr.str());
-        bin_t pos = bin_fromUInt32(evbuffer_remove_32be(evb));
-        if (!pos.is_all())
-            return_log ("%s #0 that is not the root hash %s\n",tintstr(),addr.str());
-        hash = evbuffer_remove_hash(evb);
-        FileTransfer* ft = FileTransfer::Find(hash);
-        if (!ft)
-            return_log ("%s #0 hash %s unknown, requested by %s\n",tintstr(),hash.hex().c_str(),addr.str());
-        dprintf("%s #0 -hash ALL %s\n",tintstr(),hash.hex().c_str());
-
-        // Arno, 2012-02-27: Check for duplicate channel
-        Channel* existchannel = ft->FindChannel(addr,NULL);
-        if (existchannel)
-        {
-                       // Arno: 2011-10-13: Ignore if established, otherwise consider
-                       // it a concurrent connection attempt.
-                       if (existchannel->is_established()) {
-                               // ARNOTODO: Read complete handshake here so we know whether
-                               // attempt is to new channel or to existing. Currently read
-                               // in OnHandshake()
-                               //
-                               return_log("%s #0 have a channel already to %s\n",tintstr(),addr.str());
-                       } else {
-                               channel = existchannel;
-                               //fprintf(stderr,"Channel::RecvDatagram: HANDSHAKE: reuse channel %s\n", channel->peer_.str() );
+               if (evbuffer_get_length(evb)<4)
+                       return_log("socket layer weird: datagram shorter than 4 bytes from %s (prob ICMP unreach)\n",addr.str());
+               uint32_t mych = evbuffer_remove_32be(evb);
+               Sha1Hash hash;
+               Channel* channel = NULL;
+               if (mych==0) { // peer initiates handshake
+                       if (evbuffer_get_length(evb)<1+4+1+4+Sha1Hash::SIZE)
+                               return_log ("%s #0 incorrect size %i initial handshake packet %s\n",
+                                               tintstr(),(int)evbuffer_get_length(evb),addr.str());
+                       uint8_t hashid = evbuffer_remove_8(evb);
+                       if (hashid!=SWIFT_HASH)
+                               return_log ("%s #0 no hash in the initial handshake %s\n",
+                                               tintstr(),addr.str());
+                       bin_t pos = bin_fromUInt32(evbuffer_remove_32be(evb));
+                       if (!pos.is_all())
+                               return_log ("%s #0 that is not the root hash %s\n",tintstr(),addr.str());
+                       hash = evbuffer_remove_hash(evb);
+                       FileTransfer* ft = FileTransfer::Find(hash);
+                       if (!ft)
+                               return_log ("%s #0 hash %s unknown, requested by %s\n",tintstr(),hash.hex().c_str(),addr.str());
+                       dprintf("%s #0 -hash ALL %s\n",tintstr(),hash.hex().c_str());
+
+                       // Arno, 2012-02-27: Check for duplicate channel
+                       Channel* existchannel = ft->FindChannel(addr,NULL);
+                       if (existchannel)
+                       {
+                               // Arno: 2011-10-13: Ignore if established, otherwise consider
+                               // it a concurrent connection attempt.
+                               if (existchannel->is_established()) {
+                                       // ARNOTODO: Read complete handshake here so we know whether
+                                       // attempt is to new channel or to existing. Currently read
+                                       // in OnHandshake()
+                                       //
+                                       return_log("%s #0 have a channel already to %s\n",tintstr(),addr.str());
+                               } else {
+                                       channel = existchannel;
+                                       //fprintf(stderr,"Channel::RecvDatagram: HANDSHAKE: reuse channel %s\n", channel->peer_.str() );
+                               }
                        }
-        }
-        if (channel == NULL) {
-               //fprintf(stderr,"Channel::RecvDatagram: HANDSHAKE: create new channel %s\n", addr.str() );
-               channel = new Channel(ft, socket, addr);
-        }
-        //fprintf(stderr,"CHANNEL INCOMING DEF hass %s is id %d\n",hash.hex().c_str(),channel->id());
-
-    } else { // peer responds to my handshake (and other messages)
-        mych = DecodeID(mych);
-        if (mych>=channels.size())
-            return_log("%s invalid channel #%u, %s\n",tintstr(),mych,addr.str());
-        channel = channels[mych];
-        if (!channel)
-            return_log ("%s #%u is already closed\n",tintstr(),mych);
-               if (channel->IsDiffSenderOrDuplicate(addr,mych)) {
-                       channel->Schedule4Close();
-                       return;
+                       if (channel == NULL) {
+                               //fprintf(stderr,"Channel::RecvDatagram: HANDSHAKE: create new channel %s\n", addr.str() );
+                               channel = new Channel(ft, socket, addr);
+                       }
+                       //fprintf(stderr,"CHANNEL INCOMING DEF hass %s is id %d\n",hash.hex().c_str(),channel->id());
+
+               } else { // peer responds to my handshake (and other messages)
+                       mych = DecodeID(mych);
+                       if (mych>=channels.size())
+                               return_log("%s invalid channel #%u, %s\n",tintstr(),mych,addr.str());
+                       channel = channels[mych];
+                       if (!channel)
+                               return_log ("%s #%u is already closed\n",tintstr(),mych);
+                       if (channel->IsDiffSenderOrDuplicate(addr,mych)) {
+                               channel->Schedule4Close();
+                               return;
+                       }
+                       channel->own_id_mentioned_ = true;
                }
-        channel->own_id_mentioned_ = true;
-    }
-    channel->raw_bytes_down_ += evboriglen;
-    //dprintf("recvd %i bytes for %i\n",data.size(),channel->id);
-    bool wasestablished = channel->is_established();
+               channel->raw_bytes_down_ += evboriglen;
+               //dprintf("recvd %i bytes for %i\n",data.size(),channel->id);
+               bool wasestablished = channel->is_established();
 
-    dprintf("%s #%u peer %s recv_peer %s addr %s\n", tintstr(),mych, channel->peer().str(), channel->recv_peer().str(), addr.str() );
+               dprintf("%s #%u peer %s recv_peer %s addr %s\n", tintstr(),mych, channel->peer().str(), channel->recv_peer().str(), addr.str() );
 
-    channel->Recv(evb);
+               channel->Recv(evb);
 
-    evbuffer_free(evb);
-    //SAFECLOSE
-    if (wasestablished && !channel->is_established()) {
-       // Arno, 2012-01-26: Received an explict close, clean up channel, safely.
-       channel->Schedule4Close();
-    }
+               evbuffer_free(evb);
+               //SAFECLOSE
+               if (wasestablished && !channel->is_established()) {
+                       // Arno, 2012-01-26: Received an explict close, clean up channel, safely.
+                       channel->Schedule4Close();
+               }
+       }
+       for (; i<NUM_DATAGRAMS; ++i)
+               evbuffer_free(pevb[i]);
 }
 
 

diff --git a/src/libswift/swift.h b/src/libswift/swift.h
index 288238a..7c8d27c 100644 (file)
--- a/src/libswift/swift.h
+++ b/src/libswift/swift.h
@@ -494,7 +494,7 @@ namespace swift {
         static void LibeventSendCallback(int fd, short event, void *arg);
         static void LibeventReceiveCallback(int fd, short event, void *arg);
         static void RecvDatagram (evutil_socket_t socket); // Called by LibeventReceiveCallback
-           static int RecvFrom(evutil_socket_t sock, Address& addr, struct evbuffer *evb); // Called by RecvDatagram
+           static int RecvFrom(evutil_socket_t sock, Address& addr, struct evbuffer **evb); // Called by RecvDatagram
            static int SendTo(evutil_socket_t sock, const Address& addr, struct evbuffer *evb); // Called by Channel::Send()
            static evutil_socket_t Bind(Address address, sckrwecb_t callbacks=sckrwecb_t());
            static Address BoundAddress(evutil_socket_t sock);