sendrecv.cpp

   1 /*
   2  *  datasendrecv.cpp
   3  *  serp++
   4  *
   5  *  Created by Victor Grishchenko on 3/6/09.
   6  *  Copyright 2009 Delft University of Technology. All rights reserved.
   7  *
   8  */
   9 #include <algorithm>
  10 //#include <glog/logging.h>
  11 #include "p2tp.h"
  12 #include "compat/util.h"
  13
  14
  15 using namespace p2tp;
  16 using namespace std; // FIXME remove
  17
  18 /*
  19  TODO  25 Oct 18:55
  20  - move hint_out_, piece picking to piece picker (needed e.g. for the case of channel drop)
  21  - ANY_LAYER
  22  - range: ALL
  23  - randomized testing of advanced ops (new testcase)
  24  - PeerCwnd()
  25  - bins hint_out_, tbqueue hint_out_ts_
  26
  27  */
  28
  29 void    Channel::AddPeakHashes (Datagram& dgram) {
  30         for(int i=0; i<file().peak_count(); i++) {
  31         bin64_t peak = file().peak(i);
  32                 dgram.Push8(P2TP_HASH);
  33                 dgram.Push32((uint32_t)peak);
  34                 dgram.PushHash(file().peak_hash(i));
  35         //DLOG(INFO)<<"#"<<id<<" +pHASH"<<file().peak(i);
  36         dprintf("%s #%i +phash (%i,%lli)\n",tintstr(),id,peak.layer(),peak.offset());
  37         }
  38 }
  39
  40
  41 void    Channel::AddUncleHashes (Datagram& dgram, bin64_t pos) {
  42     bin64_t peak = file().peak_for(pos);
  43     while (pos!=peak && ack_in_.get(pos.parent())==bins::EMPTY) {
  44         bin64_t uncle = pos.sibling();
  45                 dgram.Push8(P2TP_HASH);
  46                 dgram.Push32((uint32_t)uncle);
  47                 dgram.PushHash( file().hash(uncle) );
  48         //DLOG(INFO)<<"#"<<id<<" +uHASH"<<uncle;
  49         dprintf("%s #%i +hash (%i,%lli)\n",tintstr(),id,uncle.layer(),uncle.offset());
  50         pos = pos.parent();
  51     }
  52 }
  53
  54
  55 bin64_t         Channel::DequeueHint () { // TODO: resilience
  56     bin64_t send = bin64_t::NONE;
  57     while (!hint_in_.empty() && send==bin64_t::NONE) {
  58         bin64_t hint = hint_in_.front();
  59         hint_in_.pop_front();
  60         send = file().ack_out().find_filtered
  61             (ack_in_,hint,0,bins::FILLED);
  62         dprintf("%s #%i may_send %lli\n",tintstr(),id,send.base_offset());
  63         if (send!=bin64_t::NONE)
  64             while (send!=hint) {
  65                 hint = hint.towards(send);
  66                 hint_in_.push_front(hint.sibling());
  67             }
  68     }
  69     return send;
  70 }
  71
  72
  73 /*void  Channel::CleanStaleHints () {
  74         while ( !hint_out.empty() && file().ack_out().get(hint_out.front().bin)==bins::FILLED )
  75                 hint_out.pop_front();  // FIXME must normally clear fulfilled entries
  76         tint timed_out = NOW - cc_->RoundTripTime()*8;
  77         while ( !hint_out.empty() && hint_out.front().time < timed_out ) {
  78         file().picker()->Snubbed(hint_out.front().bin);
  79                 hint_out.pop_front();
  80         }
  81 }*/
  82
  83
  84 void    Channel::AddHandshake (Datagram& dgram) {
  85         if (!peer_channel_id_) { // initiating
  86                 dgram.Push8(P2TP_HASH);
  87                 dgram.Push32(bin64_t::ALL32);
  88                 dgram.PushHash(file().root_hash());
  89         dprintf("%s #%i +hash ALL %s\n",
  90                 tintstr(),id,file().root_hash().hex().c_str());
  91         }
  92         dgram.Push8(P2TP_HANDSHAKE);
  93         dgram.Push32(EncodeID(id));
  94     dprintf("%s #%i +hs\n",tintstr(),id);
  95     ack_out_.clear();
  96     AddAck(dgram);
  97 }
  98
  99
 100 void    Channel::ClearStaleDataOut() {
 101     int oldsize = data_out_.size();
 102     while ( data_out_.size() && data_out_.front().time <
 103            NOW - rtt_avg_ - dev_avg_*4 )
 104         data_out_.pop_front();
 105     if (data_out_.size()!=oldsize)
 106         cc_->OnAckRcvd(bin64_t::NONE);
 107 }
 108
 109
 110 void    Channel::Send () {
 111     Datagram dgram(socket_,peer());
 112     dgram.Push32(peer_channel_id_);
 113     bin64_t data = bin64_t::NONE;
 114     if ( is_established() ) {
 115         AddAck(dgram);
 116         AddHint(dgram);
 117         AddPex(dgram);
 118         ClearStaleDataOut();
 119         if (cc_->MaySendData())
 120             data = AddData(dgram);
 121         else
 122             dprintf("%s #%i no cwnd\n",tintstr(),id);
 123     } else {
 124         AddHandshake(dgram);
 125         AddAck(dgram);
 126     }
 127     dprintf("%s #%i sent %ib %s\n",tintstr(),id,dgram.size(),peer().str().c_str());
 128         if (dgram.Send()==-1)
 129         print_error("can't send datagram");
 130     if (dgram.size()==4) // only the channel id; bare keep-alive
 131         data = bin64_t::ALL;
 132     cc_->OnDataSent(data);
 133     last_send_time_ = NOW;
 134     RequeueSend(cc_->NextSendTime());
 135 }
 136
 137
 138 void    Channel::AddHint (Datagram& dgram) {
 139
 140     while (!hint_out_.empty()) {
 141         tintbin f = hint_out_.front();
 142         if (f.time<NOW-rtt_avg_*8) {
 143             hint_out_.pop_front();
 144         } else {
 145             int status = file().ack_out().get(f.bin);
 146             if (status==bins::EMPTY) {
 147                 break;
 148             } else if (status==bins::FILLED) {
 149                 hint_out_.pop_front();
 150             } else { // mixed
 151                 hint_out_.front().bin = f.bin.right();
 152                 f.bin = f.bin.left();
 153                 hint_out_.push_front(f);
 154             }
 155         }
 156     }
 157     /*while (!hint_out_.empty() &&
 158             (hint_out_.front().time<NOW-TINT_SEC ||
 159             file().ack_out().get(hint_out_.front().bin)==bins::FILLED ) ) {
 160         file().picker().Expired(hint_out_.front().bin);
 161         hint_out_.pop_front();
 162     }*/
 163     uint64_t hinted = 0;
 164     for(tbqueue::iterator i=hint_out_.begin(); i!=hint_out_.end(); i++)
 165         hinted += i->bin.width();
 166     //int bps = PeerBPS();
 167     double peer_cwnd = rtt_avg_ / dip_avg_;
 168     if (peer_cwnd<1)
 169         peer_cwnd = 1;
 170     dprintf("%s #%i hinted %lli peer_cwnd %lli/%lli=%f\n",
 171             tintstr(),id,hinted,rtt_avg_,dip_avg_,((float)rtt_avg_/dip_avg_));
 172
 173     if ( 4*peer_cwnd > hinted ) { //hinted*1024 < peer_cwnd*4 ) {
 174
 175         uint8_t layer = 2; // actually, enough
 176         bin64_t hint = transfer().picker().Pick(ack_in_,layer);
 177         // FIXME FIXME FIXME: any layer
 178         if (hint==bin64_t::NONE)
 179             hint = transfer().picker().Pick(ack_in_,0);
 180
 181         if (hint!=bin64_t::NONE) {
 182             hint_out_.push_back(hint);
 183             dgram.Push8(P2TP_HINT);
 184             dgram.Push32(hint);
 185             dprintf("%s #%i +hint (%i,%lli)\n",tintstr(),id,hint.layer(),hint.offset());
 186         }
 187
 188     }
 189 }
 190
 191
 192 bin64_t         Channel::AddData (Datagram& dgram) {
 193         if (!file().size()) // know nothing
 194                 return bin64_t::NONE;
 195         bin64_t tosend = DequeueHint();
 196     if (tosend==bin64_t::NONE) {
 197         dprintf("%s #%i out of hints\n",tintstr(),id);
 198         return bin64_t::NONE;
 199     }
 200     if (ack_in_.is_empty() && file().size())
 201         AddPeakHashes(dgram);
 202     AddUncleHashes(dgram,tosend);
 203     uint8_t buf[1024];
 204     size_t r = pread(file().file_descriptor(),buf,1024,tosend.base_offset()<<10);
 205     // TODO: ??? corrupted data, retries
 206     if (r<0) {
 207         print_error("error on reading");
 208         return bin64_t::NONE;
 209     }
 210     assert(dgram.space()>=r+4+1);
 211     dgram.Push8(P2TP_DATA);
 212     dgram.Push32(tosend);
 213     dgram.Push(buf,r);
 214     dprintf("%s #%i +data (%lli)\n",tintstr(),id,tosend.base_offset());
 215     data_out_.push_back(tosend);
 216         return tosend;
 217 }
 218
 219
 220 void    Channel::AddTs (Datagram& dgram) {
 221     dgram.Push8(P2TP_TS);
 222     dgram.Push64(data_in_.time);
 223     dprintf("%s #%i +ts %lli\n",tintstr(),id,data_in_.time);
 224 }
 225
 226
 227 void    Channel::AddAck (Datagram& dgram) {
 228         if (data_in_.bin!=bin64_t::NONE) {
 229         AddTs(dgram);
 230         bin64_t pos = data_in_.bin;
 231                 dgram.Push8(P2TP_ACK);
 232                 dgram.Push32(pos);
 233                 //dgram.Push64(data_in_.time);
 234         ack_out_.set(pos);
 235         dprintf("%s #%i +ack (%i,%lli) %s\n",tintstr(),id,
 236                 pos.layer(),pos.offset(),tintstr(data_in_.time));
 237         data_in_ = tintbin(0,bin64_t::NONE);
 238         }
 239     for(int count=0; count<4; count++) {
 240         bin64_t ack = file().ack_out().find_filtered(ack_out_, bin64_t::ALL, 0, bins::FILLED);
 241         // TODO bins::ANY_LAYER
 242         if (ack==bin64_t::NONE)
 243             break;
 244         while (file().ack_out().get(ack.parent())==bins::FILLED)
 245             ack = ack.parent();
 246         ack_out_.set(ack);
 247         dgram.Push8(P2TP_ACK);
 248         dgram.Push32(ack);
 249         dprintf("%s #%i +ack (%i,%lli)\n",tintstr(),id,ack.layer(),ack.offset());
 250     }
 251 }
 252
 253
 254 void    Channel::Recv (Datagram& dgram) {
 255     if (last_send_time_ && rtt_avg_==TINT_SEC && dev_avg_==0) {
 256         rtt_avg_ = NOW - last_send_time_;
 257         dev_avg_ = rtt_avg_;
 258         dip_avg_ = rtt_avg_;
 259         transfer().hs_in_.push_back(id);
 260         dprintf("%s #%i rtt init %lli\n",tintstr(),id,rtt_avg_);
 261     }
 262     bin64_t data = dgram.size() ? bin64_t::NONE : bin64_t::ALL;
 263         while (dgram.size()) {
 264                 uint8_t type = dgram.Pull8();
 265                 switch (type) {
 266             case P2TP_HANDSHAKE: OnHandshake(dgram); break;
 267                         case P2TP_DATA:         data=OnData(dgram); break;
 268                         case P2TP_TS:       OnTs(dgram); break;
 269                         case P2TP_ACK:          OnAck(dgram); break;
 270                         case P2TP_HASH:         OnHash(dgram); break;
 271                         case P2TP_HINT:         OnHint(dgram); break;
 272             case P2TP_PEX_ADD:  OnPex(dgram); break;
 273                         default:
 274                                 //LOG(ERROR) << this->id_string() << " malformed datagram";
 275                                 return;
 276                 }
 277         }
 278     cc_->OnDataRecvd(data);
 279     last_recv_time_ = NOW;
 280     if (data!=bin64_t::ALL)
 281         RequeueSend(NOW);
 282 }
 283
 284
 285 void    Channel::OnHash (Datagram& dgram) {
 286         bin64_t pos = dgram.Pull32();
 287         Sha1Hash hash = dgram.PullHash();
 288         file().OfferHash(pos,hash);
 289     //DLOG(INFO)<<"#"<<id<<" .HASH"<<(int)pos;
 290     dprintf("%s #%i -hash (%i,%lli)\n",tintstr(),id,pos.layer(),pos.offset());
 291 }
 292
 293
 294 bin64_t Channel::OnData (Datagram& dgram) {
 295         bin64_t pos = dgram.Pull32();
 296     uint8_t *data;
 297     int length = dgram.Pull(&data,1024);
 298     bool ok = file().OfferData(pos, (char*)data, length) ;
 299     dprintf("%s #%i %cdata (%lli)\n",tintstr(),id,ok?'-':'!',pos.offset());
 300     if (ok) {
 301         data_in_ = tintbin(NOW,pos);
 302         if (last_recv_time_) {
 303             tint dip = NOW - last_recv_time_;
 304             dip_avg_ = ( dip_avg_*3 + dip ) >> 2;
 305         }
 306         return pos;
 307     } else
 308         return bin64_t::NONE;
 309 }
 310
 311
 312 void    Channel::OnAck (Datagram& dgram) {
 313         bin64_t ackd_pos = dgram.Pull32();
 314     if (ackd_pos.base_offset()>file().size())
 315         return;
 316     dprintf("%s #%i -ack (%i,%lli)\n",tintstr(),id,ackd_pos.layer(),ackd_pos.offset());
 317     for (int i=0; i<8 && i<data_out_.size(); i++)
 318         if (data_out_[i].bin.within(ackd_pos)) {
 319             tint rtt = NOW-data_out_[i].time;
 320             rtt_avg_ = (rtt_avg_*3 + rtt) >> 2;
 321             dev_avg_ = ( dev_avg_*3 + abs(rtt-rtt_avg_) ) >> 2;
 322             dprintf("%s #%i rtt %lli dev %lli\n",
 323                     tintstr(),id,rtt_avg_,dev_avg_);
 324             cc_->OnAckRcvd(data_out_[i].bin);
 325         }
 326         ack_in_.set(ackd_pos);
 327     while (data_out_.size() && ack_in_.get(data_out_.front().bin)==bins::FILLED)
 328         data_out_.pop_front();
 329 }
 330
 331
 332 /*void  Channel::OnAckTs (Datagram& dgram) {  // FIXME:   OnTs
 333         bin64_t pos = dgram.Pull32();
 334     tint ts = dgram.Pull64();
 335     // TODO sanity check
 336     dprintf("%s #%i -ackts (%i,%lli) %s\n",
 337             tintstr(),id,pos.layer(),pos.offset(),tintstr(ts));
 338         ack_in_.set(pos);
 339         cc_->OnAckRcvd(pos,ts);
 340 }*/
 341
 342 void Channel::OnTs (Datagram& dgram) {
 343     peer_send_time_ = dgram.Pull64();
 344     dprintf("%s #%i -ts %lli\n",tintstr(),id,peer_send_time_);
 345 }
 346
 347
 348 void    Channel::OnHint (Datagram& dgram) {
 349         bin64_t hint = dgram.Pull32();
 350         hint_in_.push_back(hint);
 351     //RequeueSend(cc_->OnHintRecvd(hint));
 352     dprintf("%s #%i -hint (%i,%lli)\n",tintstr(),id,hint.layer(),hint.offset());
 353 }
 354
 355
 356 void Channel::OnHandshake (Datagram& dgram) {
 357     peer_channel_id_ = dgram.Pull32();
 358     dprintf("%s #%i -hs %i\n",tintstr(),id,peer_channel_id_);
 359     // FUTURE: channel forking
 360 }
 361
 362
 363 void Channel::OnPex (Datagram& dgram) {
 364     uint32_t ipv4 = dgram.Pull32();
 365     uint16_t port = dgram.Pull16();
 366     Address addr(ipv4,port);
 367     dprintf("%s #%i -pex %s\n",tintstr(),id,addr.str().c_str());
 368     transfer().OnPexIn(addr);
 369 }
 370
 371
 372 void    Channel::AddPex (Datagram& dgram) {
 373     int chid = transfer().RevealChannel(pex_out_);
 374     if (chid==-1 || chid==id)
 375         return;
 376     Address a = channels[chid]->peer();
 377     dgram.Push8(P2TP_PEX_ADD);
 378     dgram.Push32(a.ipv4());
 379     dgram.Push16(a.port());
 380     dprintf("%s #%i +pex %s\n",tintstr(),id,a.str().c_str());
 381 }
 382
 383
 384 void    Channel::Recv (int socket) {
 385         Datagram data(socket);
 386         data.Recv();
 387         if (data.size()<4)
 388                 RETLOG("datagram shorter than 4 bytes");
 389         uint32_t mych = data.Pull32();
 390         Sha1Hash hash;
 391         Channel* channel = NULL;
 392         if (!mych) { // handshake initiated
 393                 if (data.size()<1+4+1+4+Sha1Hash::SIZE)
 394                         RETLOG ("incorrect size initial handshake packet");
 395                 uint8_t hashid = data.Pull8();
 396                 if (hashid!=P2TP_HASH)
 397                         RETLOG ("no hash in the initial handshake");
 398                 bin64_t pos = data.Pull32();
 399                 if (pos!=bin64_t::ALL32)
 400                         RETLOG ("that is not the root hash");
 401                 hash = data.PullHash();
 402                 FileTransfer* file = FileTransfer::Find(hash);
 403                 if (!file)
 404                         RETLOG ("hash unknown, no such file");
 405         dprintf("%s #0 -hash ALL %s\n",tintstr(),hash.hex().c_str());
 406         for(binqueue::iterator i=file->hs_in_.begin(); i!=file->hs_in_.end(); i++)
 407             if (channels[*i] && channels[*i]->peer_==data.addr)
 408                 RETLOG("have a channel already");
 409                 channel = new Channel(file, socket, data.address());
 410         } else {
 411                 mych = DecodeID(mych);
 412                 if (mych>=channels.size()) {
 413             eprintf("invalid channel #%i\n",mych);
 414             return;
 415         }
 416                 channel = channels[mych];
 417                 if (!channel)
 418                         RETLOG ("channel is closed");
 419                 if (channel->peer() != data.address())
 420                         RETLOG ("invalid peer address");
 421         channel->own_id_mentioned_ = true;
 422         }
 423     dprintf("recvd %i bytes for %i\n",data.size(),channel->id);
 424     channel->Recv(data);
 425 }
 426
 427
 428 bool tblater (const tintbin& a, const tintbin& b) {
 429     return a.time > b.time;
 430 }
 431
 432
 433 void    Channel::RequeueSend (tint next_time) {
 434     if (next_time==next_send_time_)
 435         return;
 436     next_send_time_ = next_time;
 437     send_queue.push_back(tintbin(next_time,id));
 438     push_heap(send_queue.begin(),send_queue.end(),tblater);
 439     dprintf("%s requeue #%i for %s\n",tintstr(),id,tintstr(next_time));
 440 }
 441
 442
 443 void    Channel::Loop (tint howlong) {
 444
 445     tint limit = Datagram::Time() + howlong;
 446
 447     do {
 448
 449         tint send_time(TINT_NEVER);
 450         Channel* sender(NULL);
 451         while (!send_queue.empty()) {
 452             send_time = send_queue.front().time;
 453             sender = channel((int)send_queue.front().bin);
 454             if (sender && sender->next_send_time_==send_time)
 455                 break;
 456             sender = NULL; // it was a stale entry
 457             pop_heap(send_queue.begin(), send_queue.end(), tblater);
 458             send_queue.pop_back();
 459         }
 460         if (send_time>limit)
 461             send_time = limit;
 462         if ( sender && send_time <= NOW ) {
 463             dprintf("%s #%i sch_send %s\n",tintstr(),sender->id,
 464                     tintstr(send_time));
 465             sender->Send();
 466             pop_heap(send_queue.begin(), send_queue.end(), tblater);
 467             send_queue.pop_back();
 468         } else {
 469             tint towait = send_time - NOW;
 470             dprintf("%s waiting %lliusec\n",tintstr(),towait);
 471             int rd = Datagram::Wait(socket_count,sockets,towait);
 472             if (rd!=INVALID_SOCKET)
 473                 Recv(rd);
 474         }
 475
 476     } while (Datagram::Time()<limit);
 477
 478 }
 479