From 722d42608bfc8f90d258ecd9c2e6a591c47602fa Mon Sep 17 00:00:00 2001 From: Adrian Bondrescu Date: Fri, 1 Jun 2012 04:11:24 +0300 Subject: [PATCH] Implement receiving from multiple sources using a single system call. --- src/libswift/channel.cpp | 37 +++++---- src/libswift/sendrecv.cpp | 154 +++++++++++++++++++++----------------- src/libswift/swift.h | 2 +- 3 files changed, 107 insertions(+), 86 deletions(-) diff --git a/src/libswift/channel.cpp b/src/libswift/channel.cpp index 2828e86..4fed7a5 100644 --- 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; icount; 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; icount; ++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 --- 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; icount = NUM_DATAGRAMS; + RecvFrom(socket, addr, pevb); + int i = 0; + for (; icount; ++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