X-Git-Url: http://p2p-next.cs.pub.ro/gitweb/?a=blobdiff_plain;f=src%2Flibswift%2Fsendrecv.cpp;h=cbdf264bb56a96b7e777023a85dc0be20f0f1d59;hb=HEAD;hp=5961ad53f30aeccbb7bd3fa006df7328a8ad7327;hpb=722d42608bfc8f90d258ecd9c2e6a591c47602fa;p=swifty.git

diff --git a/src/libswift/sendrecv.cpp b/src/libswift/sendrecv.cpp
index 5961ad5..cbdf264 100644
--- a/src/libswift/sendrecv.cpp
+++ b/src/libswift/sendrecv.cpp
@@ -18,6 +18,7 @@ using namespace swift;
 using namespace std;
 
 struct event_base *Channel::evbase;
+MessageQueue Channel::messageQueue;
 struct event Channel::evrecv;
 
 #define DEBUGTRAFFIC 	0
@@ -180,7 +181,7 @@ void    Channel::Send () {
 			}
 			AddPex(evb);
 			TimeoutDataOut();
-			data = AddData(evb);
+			data = AddData(&evb);
     	} else {
     		// Arno: send explicit close
     		AddHandshake(evb);
@@ -200,16 +201,20 @@ void    Channel::Send () {
     dprintf("%s #%u sent %ib %s:%x\n",
             tintstr(),id_,(int)evbuffer_get_length(evb),peer().str(),
             peer_channel_id_);
-    int r = SendTo(socket_,peer(),evb);
-    if (r==-1)
-        print_error("can't send datagram");
-    else
-    	raw_bytes_up_ += r;
-    last_send_time_ = NOW;
-    sent_since_recv_++;
-    dgrams_sent_++;
-    evbuffer_free(evb);
-    Reschedule();
+
+	messageQueue.AddBuffer(socket_, evb, peer(), this); 
+}
+
+void Channel::Sent(int bytes, evbuffer *evb, bool tofree)
+{
+	raw_bytes_up_ += bytes;
+	if (tofree) {
+		last_send_time_ = NOW;
+		sent_since_recv_++;
+		dgrams_sent_++;
+		evbuffer_free(evb);
+		Reschedule();
+	}
 }
 
 void    Channel::AddHint (struct evbuffer *evb) {
@@ -291,7 +296,7 @@ void    Channel::AddHint (struct evbuffer *evb) {
 }
 
 
-bin_t        Channel::AddData (struct evbuffer *evb) {
+bin_t        Channel::AddData (struct evbuffer **evb) {
 	// RATELIMIT
 	if (transfer().GetCurrentSpeed(DDIR_UPLOAD) > transfer().GetMaxSpeed(DDIR_UPLOAD)) {
 		transfer().OnSendNoData();
@@ -320,11 +325,11 @@ bin_t        Channel::AddData (struct evbuffer *evb) {
         return bin_t::NONE; // once in a while, empty data is sent just to check rtt FIXED
 
     if (ack_in_.is_empty() && file().size())
-        AddPeakHashes(evb);
+        AddPeakHashes(*evb);
 
     //NETWVSHASH
     if (file().get_check_netwvshash())
-    	AddUncleHashes(evb,tosend);
+    	AddUncleHashes(*evb,tosend);
 
     if (!ack_in_.is_empty()) // TODO: cwnd_>1
         data_out_cap_ = tosend;
@@ -334,14 +339,13 @@ bin_t        Channel::AddData (struct evbuffer *evb) {
     // frame with DATA. Send 2 datagrams then, one with peaks so they have
     // a better chance of arriving. Optimistic violation of atomic datagram
     // principle.
-    if (file().chunk_size() == SWIFT_DEFAULT_CHUNK_SIZE && evbuffer_get_length(evb) > SWIFT_MAX_NONDATA_DGRAM_SIZE) {
+    if (file().chunk_size() == SWIFT_DEFAULT_CHUNK_SIZE && evbuffer_get_length(*evb) > SWIFT_MAX_NONDATA_DGRAM_SIZE) {
         dprintf("%s #%u fsent %ib %s:%x\n",
-                tintstr(),id_,(int)evbuffer_get_length(evb),peer().str(),
+                tintstr(),id_,(int)evbuffer_get_length(*evb),peer().str(),
                 peer_channel_id_);
-    	int ret = Channel::SendTo(socket_,peer(),evb); // kind of fragmentation
-    	if (ret > 0)
-    		raw_bytes_up_ += ret;
-        evbuffer_add_32be(evb, peer_channel_id_);
+		messageQueue.AddBuffer(socket_, *evb, peer(), this, false);
+		*evb = evbuffer_new();
+        evbuffer_add_32be(*evb, peer_channel_id_);
     }
 
     if (file().chunk_size() != SWIFT_DEFAULT_CHUNK_SIZE && isretransmit) {
@@ -358,15 +362,15 @@ bin_t        Channel::AddData (struct evbuffer *evb) {
     	 */
  	     char binstr[32];
          fprintf(stderr,"AddData: retransmit of randomized chunk %s\n",tosend.str(binstr) );
-         evbuffer_add_8(evb, SWIFT_RANDOMIZE);
-         evbuffer_add_32be(evb, (int)rand() );
+         evbuffer_add_8(*evb, SWIFT_RANDOMIZE);
+         evbuffer_add_32be(*evb, (int)rand() );
     }
 
-    evbuffer_add_8(evb, SWIFT_DATA);
-    evbuffer_add_32be(evb, bin_toUInt32(tosend));
+    evbuffer_add_8(*evb, SWIFT_DATA);
+    evbuffer_add_32be(*evb, bin_toUInt32(tosend));
 
     struct evbuffer_iovec vec;
-    if (evbuffer_reserve_space(evb, file().chunk_size(), &vec, 1) < 0) {
+    if (evbuffer_reserve_space(*evb, file().chunk_size(), &vec, 1) < 0) {
 	print_error("error on evbuffer_reserve_space");
 	return bin_t::NONE;
     }
@@ -376,12 +380,12 @@ bin_t        Channel::AddData (struct evbuffer *evb) {
     if (r<0) {
         print_error("error on reading");
         vec.iov_len = 0;
-        evbuffer_commit_space(evb, &vec, 1);
+        evbuffer_commit_space(*evb, &vec, 1);
         return bin_t::NONE;
     }
     // assert(dgram.space()>=r+4+1);
     vec.iov_len = r;
-    if (evbuffer_commit_space(evb, &vec, 1) < 0) {
+    if (evbuffer_commit_space(*evb, &vec, 1) < 0) {
         print_error("error on evbuffer_commit_space");
         return bin_t::NONE;
     }
@@ -483,7 +487,7 @@ void    Channel::Recv (struct evbuffer *evb) {
         uint8_t type = evbuffer_remove_8(evb);
 
         if (DEBUGTRAFFIC)
-        	fprintf(stderr," %d", type);
+        	fprintf(stderr," %d\n", type);
 
         switch (type) {
             case SWIFT_HANDSHAKE: OnHandshake(evb); break;
@@ -929,32 +933,35 @@ void    Channel::RecvDatagram (evutil_socket_t socket) {
 	int i = 0;
 	for (; i<addr.addr->count; ++i) {
 		struct evbuffer *evb = pevb[i];
+		Address fromi;
+		fromi.addr->dests[0].addr = addr.addr->dests[i].addr;
+		fromi.addr->dests[0].port = addr.addr->dests[i].port;
 		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());
+			return_log("socket layer weird: datagram shorter than 4 bytes from %s (prob ICMP unreach)\n",fromi.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());
+						tintstr(),(int)evbuffer_get_length(evb),fromi.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());
+						tintstr(),fromi.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());
+				return_log ("%s #0 that is not the root hash %s\n",tintstr(),fromi.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());
+				return_log ("%s #0 hash %s unknown, requested by %s\n",tintstr(),hash.hex().c_str(),fromi.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);
+			Channel* existchannel = ft->FindChannel(fromi,NULL);
 			if (existchannel)
 			{
 				// Arno: 2011-10-13: Ignore if established, otherwise consider
@@ -964,7 +971,7 @@ void    Channel::RecvDatagram (evutil_socket_t socket) {
 					// 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());
+					return_log("%s #0 have a channel already to %s\n",tintstr(),fromi.str());
 				} else {
 					channel = existchannel;
 					//fprintf(stderr,"Channel::RecvDatagram: HANDSHAKE: reuse channel %s\n", channel->peer_.str() );
@@ -972,18 +979,18 @@ void    Channel::RecvDatagram (evutil_socket_t socket) {
 			}
 			if (channel == NULL) {
 				//fprintf(stderr,"Channel::RecvDatagram: HANDSHAKE: create new channel %s\n", addr.str() );
-				channel = new Channel(ft, socket, addr);
+				channel = new Channel(ft, socket, fromi);
 			}
 			//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());
+				return_log("%s invalid channel #%u, %s\n",tintstr(),mych,fromi.str());
 			channel = channels[mych];
 			if (!channel)
 				return_log ("%s #%u is already closed\n",tintstr(),mych);
-			if (channel->IsDiffSenderOrDuplicate(addr,mych)) {
+			if (channel->IsDiffSenderOrDuplicate(fromi,mych)) {
 				channel->Schedule4Close();
 				return;
 			}
@@ -993,7 +1000,7 @@ void    Channel::RecvDatagram (evutil_socket_t socket) {
 		//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(), fromi.str() );
 
 		channel->Recv(evb);