sendrecv.cpp

   1 /*
   2  *  sendrecv.cpp
   3  *  most of the swift's state machine
   4  *
   5  *  Created by Victor Grishchenko on 3/6/09.
   6  *  Copyright 2009 Delft University of Technology. All rights reserved.
   7  *
   8  */
   9 #include "swift.h"
  10 #include <algorithm>  // kill it
  11
  12 using namespace swift;
  13 using namespace std;
  14
  15 /*
  16  TODO  25 Oct 18:55
  17  - range: ALL
  18  - randomized testing of advanced ops (new testcase)
  19  */
  20
  21 void    Channel::AddPeakHashes (Datagram& dgram) {
  22     for(int i=0; i<file().peak_count(); i++) {
  23         bin64_t peak = file().peak(i);
  24         dgram.Push8(SWIFT_HASH);
  25         dgram.Push32((uint32_t)peak);
  26         dgram.PushHash(file().peak_hash(i));
  27         //DLOG(INFO)<<"#"<<id<<" +pHASH"<<file().peak(i);
  28         dprintf("%s #%u +phash %s\n",tintstr(),id_,peak.str());
  29     }
  30 }
  31
  32
  33 void    Channel::AddUncleHashes (Datagram& dgram, bin64_t pos) {
  34     bin64_t peak = file().peak_for(pos);
  35     while (pos!=peak && ((NOW&3)==3 || !data_out_cap_.within(pos.parent())) &&
  36             ack_in_.get(pos.parent())==binmap_t::EMPTY  ) {
  37         bin64_t uncle = pos.sibling();
  38         dgram.Push8(SWIFT_HASH);
  39         dgram.Push32((uint32_t)uncle);
  40         dgram.PushHash( file().hash(uncle) );
  41         //DLOG(INFO)<<"#"<<id<<" +uHASH"<<uncle;
  42         dprintf("%s #%u +hash %s\n",tintstr(),id_,uncle.str());
  43         pos = pos.parent();
  44     }
  45 }
  46
  47
  48 bin64_t           Channel::ImposeHint () {
  49     uint64_t twist = peer_channel_id_;  // got no hints, send something randomly
  50     twist &= file().peak(0); // FIXME may make it semi-seq here
  51     file().ack_out().twist(twist);
  52     ack_in_.twist(twist);
  53     bin64_t my_pick =
  54         file().ack_out().find_filtered(ack_in_,bin64_t::ALL,binmap_t::FILLED);
  55     while (my_pick.width()>max(1,(int)cwnd_))
  56         my_pick = my_pick.left();
  57     file().ack_out().twist(0);
  58     ack_in_.twist(0);
  59     return my_pick.twisted(twist);
  60 }
  61
  62
  63 bin64_t        Channel::DequeueHint () {
  64     if (hint_in_.empty() && last_recv_time_>NOW-rtt_avg_-TINT_SEC) {
  65         bin64_t my_pick = ImposeHint(); // FIXME move to the loop
  66         if (my_pick!=bin64_t::NONE) {
  67             hint_in_.push_back(my_pick);
  68             dprintf("%s #%u *hint %s\n",tintstr(),id_,my_pick.str());
  69         }
  70     }
  71     bin64_t send = bin64_t::NONE;
  72     while (!hint_in_.empty() && send==bin64_t::NONE) {
  73         bin64_t hint = hint_in_.front().bin;
  74         tint time = hint_in_.front().time;
  75         hint_in_.pop_front();
  76         while (!hint.is_base()) { // FIXME optimize; possible attack
  77             hint_in_.push_front(tintbin(time,hint.right()));
  78             hint = hint.left();
  79         }
  80         //if (time < NOW-TINT_SEC*3/2 )
  81         //    continue;  bad idea
  82         if (ack_in_.get(hint)!=binmap_t::FILLED)
  83             send = hint;
  84     }
  85     uint64_t mass = 0;
  86     for(int i=0; i<hint_in_.size(); i++)
  87         mass += hint_in_[i].bin.width();
  88     dprintf("%s #%u dequeued %s [%lli]\n",tintstr(),id_,send.str(),mass);
  89     return send;
  90 }
  91
  92
  93 void    Channel::AddHandshake (Datagram& dgram) {
  94     if (!peer_channel_id_) { // initiating
  95         dgram.Push8(SWIFT_HASH);
  96         dgram.Push32(bin64_t::ALL32);
  97         dgram.PushHash(file().root_hash());
  98         dprintf("%s #%u +hash ALL %s\n",
  99                 tintstr(),id_,file().root_hash().hex().c_str());
 100     }
 101     dgram.Push8(SWIFT_HANDSHAKE);
 102     int encoded = EncodeID(id_);
 103     dgram.Push32(encoded);
 104     dprintf("%s #%u +hs %x\n",tintstr(),id_,encoded);
 105     have_out_.clear();
 106     AddHave(dgram);
 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         // FIXME: seeder check
 116         AddHave(dgram);
 117         AddAck(dgram);
 118         if (!file().is_complete())
 119             AddHint(dgram);
 120         AddPex(dgram);
 121         TimeoutDataOut();
 122         data = AddData(dgram);
 123     } else {
 124         AddHandshake(dgram);
 125         AddHave(dgram);
 126         AddAck(dgram);
 127     }
 128     dprintf("%s #%u sent %ib %s:%x\n",
 129             tintstr(),id_,dgram.size(),peer().str(),peer_channel_id_);
 130     if (dgram.size()==4) {// only the channel id; bare keep-alive
 131         data = bin64_t::ALL;
 132     }
 133     if (dgram.Send()==-1)
 134         print_error("can't send datagram");
 135     last_send_time_ = NOW;
 136     sent_since_recv_++;
 137     dgrams_sent_++;
 138 }
 139
 140
 141 void    Channel::AddHint (Datagram& dgram) {
 142
 143     tint plan_for = max(TINT_SEC,rtt_avg_*4);
 144
 145     tint timed_out = NOW - plan_for*2;
 146     while ( !hint_out_.empty() && hint_out_.front().time < timed_out ) {
 147         hint_out_size_ -= hint_out_.front().bin.width();
 148         hint_out_.pop_front();
 149     }
 150
 151     int plan_pck = max ( (tint)1, plan_for / dip_avg_ );
 152
 153     if ( hint_out_size_ < plan_pck ) {
 154
 155         int diff = plan_pck - hint_out_size_; // TODO: aggregate
 156         bin64_t hint = transfer().picker().Pick(ack_in_,diff,NOW+plan_for*2);
 157
 158         if (hint!=bin64_t::NONE) {
 159             dgram.Push8(SWIFT_HINT);
 160             dgram.Push32(hint);
 161             dprintf("%s #%u +hint %s [%lli]\n",tintstr(),id_,hint.str(),hint_out_size_);
 162             hint_out_.push_back(hint);
 163             hint_out_size_ += hint.width();
 164         } else
 165             dprintf("%s #%u Xhint\n",tintstr(),id_);
 166
 167     }
 168 }
 169
 170
 171 bin64_t        Channel::AddData (Datagram& dgram) {
 172
 173     if (!file().size()) // know nothing
 174         return bin64_t::NONE;
 175
 176     bin64_t tosend = bin64_t::NONE;
 177     tint luft = send_interval_>>4; // may wake up a bit earlier
 178     if (data_out_.size()<cwnd_ &&
 179             last_data_out_time_+send_interval_<=NOW+luft) {
 180         tosend = DequeueHint();
 181         if (tosend==bin64_t::NONE) {
 182             dprintf("%s #%u sendctrl no idea what to send\n",tintstr(),id_);
 183             if (send_control_!=KEEP_ALIVE_CONTROL)
 184                 SwitchSendControl(KEEP_ALIVE_CONTROL);
 185         }
 186     } else
 187         dprintf("%s #%u sendctrl wait cwnd %f data_out %i next %s\n",
 188                 tintstr(),id_,cwnd_,data_out_.size(),tintstr(last_data_out_time_+NOW-send_interval_));
 189
 190     if (tosend==bin64_t::NONE)// && (last_data_out_time_>NOW-TINT_SEC || data_out_.empty()))
 191         return bin64_t::NONE; // once in a while, empty data is sent just to check rtt FIXED
 192
 193     if (ack_in_.is_empty() && file().size())
 194         AddPeakHashes(dgram);
 195     AddUncleHashes(dgram,tosend);
 196     if (!ack_in_.is_empty()) // TODO: cwnd_>1
 197         data_out_cap_ = tosend;
 198
 199     if (dgram.size()>254) {
 200         dgram.Send(); // kind of fragmentation
 201         dgram.Push32(peer_channel_id_);
 202     }
 203
 204     dgram.Push8(SWIFT_DATA);
 205     dgram.Push32(tosend.to32());
 206
 207     uint8_t buf[1024];
 208     size_t r = pread(file().file_descriptor(),buf,1024,tosend.base_offset()<<10);
 209     // TODO: corrupted data, retries, caching
 210     if (r<0) {
 211         print_error("error on reading");
 212         return bin64_t::NONE;
 213     }
 214     assert(dgram.space()>=r+4+1);
 215     dgram.Push(buf,r);
 216
 217     last_data_out_time_ = NOW;
 218     data_out_.push_back(tosend);
 219     dprintf("%s #%u +data %s\n",tintstr(),id_,tosend.str());
 220
 221     return tosend;
 222 }
 223
 224
 225 void    Channel::AddAck (Datagram& dgram) {
 226     if (data_in_==tintbin())
 227         return;
 228     dgram.Push8(SWIFT_ACK);
 229     dgram.Push32(data_in_.bin.to32()); // FIXME not cover
 230     dgram.Push64(data_in_.time); // FIXME 32
 231     have_out_.set(data_in_.bin);
 232     dprintf("%s #%u +ack %s %s\n",
 233         tintstr(),id_,data_in_.bin.str(),tintstr(data_in_.time));
 234     if (data_in_.bin.layer()>2)
 235         data_in_dbl_ = data_in_.bin;
 236     data_in_ = tintbin();
 237 }
 238
 239
 240 void    Channel::AddHave (Datagram& dgram) {
 241     if (data_in_dbl_!=bin64_t::NONE) { // TODO: do redundancy better
 242         dgram.Push8(SWIFT_HAVE);
 243         dgram.Push32(data_in_dbl_.to32());
 244         data_in_dbl_=bin64_t::NONE;
 245     }
 246     for(int count=0; count<4; count++) {
 247         bin64_t ack = file().ack_out().find_filtered // FIXME: do rotating queue
 248             (have_out_, bin64_t::ALL, binmap_t::FILLED);
 249         if (ack==bin64_t::NONE)
 250             break;
 251         ack = file().ack_out().cover(ack);
 252         have_out_.set(ack);
 253         dgram.Push8(SWIFT_HAVE);
 254         dgram.Push32(ack.to32());
 255         dprintf("%s #%u +have %s\n",tintstr(),id_,ack.str());
 256     }
 257 }
 258
 259
 260 void    Channel::Recv (Datagram& dgram) {
 261     dprintf("%s #%u recvd %ib\n",tintstr(),id_,dgram.size()+4);
 262     dgrams_rcvd_++;
 263     if (last_send_time_ && rtt_avg_==TINT_SEC && dev_avg_==0) {
 264         rtt_avg_ = NOW - last_send_time_;
 265         dev_avg_ = rtt_avg_;
 266         dip_avg_ = rtt_avg_;
 267         dprintf("%s #%u sendctrl rtt init %lli\n",tintstr(),id_,rtt_avg_);
 268     }
 269     bin64_t data = dgram.size() ? bin64_t::NONE : bin64_t::ALL;
 270     while (dgram.size()) {
 271         uint8_t type = dgram.Pull8();
 272         switch (type) {
 273             case SWIFT_HANDSHAKE: OnHandshake(dgram); break;
 274             case SWIFT_DATA:      data=OnData(dgram); break;
 275             case SWIFT_HAVE:      OnHave(dgram); break;
 276             case SWIFT_ACK:       OnAck(dgram); break;
 277             case SWIFT_HASH:      OnHash(dgram); break;
 278             case SWIFT_HINT:      OnHint(dgram); break;
 279             case SWIFT_PEX_ADD:   OnPex(dgram); break;
 280             default:
 281                 eprintf("%s #%u ?msg id unknown %i\n",tintstr(),id_,(int)type);
 282                 return;
 283         }
 284     }
 285     last_recv_time_ = NOW;
 286     sent_since_recv_ = 0;
 287 }
 288
 289
 290 void    Channel::OnHash (Datagram& dgram) {
 291     bin64_t pos = dgram.Pull32();
 292     Sha1Hash hash = dgram.PullHash();
 293     file().OfferHash(pos,hash);
 294     dprintf("%s #%u -hash %s\n",tintstr(),id_,pos.str());
 295 }
 296
 297
 298 void    Channel::CleanHintOut (bin64_t pos) {
 299     int hi = 0;
 300     while (hi<hint_out_.size() && !pos.within(hint_out_[hi].bin))
 301         hi++;
 302     if (hi==hint_out_.size())
 303         return; // something not hinted or hinted in far past
 304     while (hi--) { // removing likely snubbed hints
 305         hint_out_size_ -= hint_out_.front().bin.width();
 306         hint_out_.pop_front();
 307     }
 308     while (hint_out_.front().bin!=pos) {
 309         tintbin f = hint_out_.front();
 310         f.bin = f.bin.towards(pos);
 311         hint_out_.front().bin = f.bin.sibling();
 312         hint_out_.push_front(f);
 313     }
 314     hint_out_.pop_front();
 315     hint_out_size_--;
 316 }
 317
 318
 319 bin64_t Channel::OnData (Datagram& dgram) {  // TODO: HAVE NONE for corrupted data
 320     bin64_t pos = dgram.Pull32();
 321     uint8_t *data;
 322     int length = dgram.Pull(&data,1024);
 323     bool ok = (pos==bin64_t::NONE) || file().OfferData(pos, (char*)data, length) ;
 324     dprintf("%s #%u %cdata %s\n",tintstr(),id_,ok?'-':'!',pos.str());
 325     data_in_ = tintbin(NOW,bin64_t::NONE);
 326     if (!ok)
 327         return bin64_t::NONE;
 328     data_in_.bin = pos;
 329     if (pos!=bin64_t::NONE) {
 330         if (last_data_in_time_) {
 331             tint dip = NOW - last_data_in_time_;
 332             dip_avg_ = ( dip_avg_*3 + dip ) >> 2;
 333         }
 334         last_data_in_time_ = NOW;
 335     }
 336     CleanHintOut(pos);
 337     return pos;
 338 }
 339
 340
 341 void    Channel::OnAck (Datagram& dgram) {
 342     bin64_t ackd_pos = dgram.Pull32();
 343     tint peer_time = dgram.Pull64(); // FIXME 32
 344     // FIXME FIXME: wrap around here
 345     if (ackd_pos==bin64_t::NONE)
 346         return; // likely, brocken packet / insufficient hashes
 347     if (file().size() && ackd_pos.base_offset()>=file().packet_size()) {
 348         eprintf("invalid ack: %s\n",ackd_pos.str());
 349         return;
 350     }
 351     ack_in_.set(ackd_pos);
 352     int di = 0, ri = 0;
 353     // find an entry for the send (data out) event
 354     while (  di<data_out_.size() && ( data_out_[di]==tintbin() ||
 355            !data_out_[di].bin.within(ackd_pos) )  )
 356         di++;
 357     // FUTURE: delayed acks
 358     // rule out retransmits
 359     while (  ri<data_out_tmo_.size() && !data_out_tmo_[ri].bin.within(ackd_pos) )
 360         ri++;
 361     dprintf("%s #%u %cack %s %lli\n",tintstr(),id_,
 362             di==data_out_.size()?'?':'-',ackd_pos.str(),peer_time);
 363     if (di!=data_out_.size() && ri==data_out_tmo_.size()) { // not a retransmit
 364             // round trip time calculations
 365         tint rtt = NOW-data_out_[di].time;
 366         rtt_avg_ = (rtt_avg_*7 + rtt) >> 3;
 367         dev_avg_ = ( dev_avg_*3 + abs(rtt-rtt_avg_) ) >> 2;
 368         assert(data_out_[di].time!=TINT_NEVER);
 369             // one-way delay calculations
 370         tint owd = peer_time - data_out_[di].time;
 371         owd_cur_bin_ = 0;//(owd_cur_bin_+1) & 3;
 372         owd_current_[owd_cur_bin_] = owd;
 373         if ( owd_min_bin_start_+TINT_SEC*30 < NOW ) {
 374             owd_min_bin_start_ = NOW;
 375             owd_min_bin_ = (owd_min_bin_+1) & 3;
 376             owd_min_bins_[owd_min_bin_] = TINT_NEVER;
 377         }
 378         if (owd_min_bins_[owd_min_bin_]>owd)
 379             owd_min_bins_[owd_min_bin_] = owd;
 380         dprintf("%s #%u sendctrl rtt %lli dev %lli based on %s\n",
 381                 tintstr(),id_,rtt_avg_,dev_avg_,data_out_[di].bin.str());
 382         ack_rcvd_recent_++;
 383         // early loss detection by packet reordering
 384         for (int re=0; re<di-MAX_REORDERING; re++) {
 385             if (data_out_[re]==tintbin())
 386                 continue;
 387             ack_not_rcvd_recent_++;
 388             data_out_tmo_.push_back(data_out_[re].bin);
 389             dprintf("%s #%u Rdata %s\n",tintstr(),id_,data_out_.front().bin.str());
 390             data_out_cap_ = bin64_t::ALL;
 391             data_out_[re] = tintbin();
 392         }
 393     }
 394     if (di!=data_out_.size())
 395         data_out_[di]=tintbin();
 396     // clear zeroed items
 397     while (!data_out_.empty() && ( data_out_.front()==tintbin() ||
 398             ack_in_.is_filled(data_out_.front().bin) ) )
 399         data_out_.pop_front();
 400     assert(data_out_.empty() || data_out_.front().time!=TINT_NEVER);
 401 }
 402
 403
 404 void Channel::TimeoutDataOut ( ) {
 405     // losses: timeouted packets
 406     tint timeout = NOW - ack_timeout();
 407     while (!data_out_.empty() &&
 408         ( data_out_.front().time<timeout || data_out_.front()==tintbin() ) ) {
 409         if (data_out_.front()!=tintbin() && ack_in_.is_empty(data_out_.front().bin)) {
 410             ack_not_rcvd_recent_++;
 411             data_out_cap_ = bin64_t::ALL;
 412             data_out_tmo_.push_back(data_out_.front().bin);
 413             dprintf("%s #%u Tdata %s\n",tintstr(),id_,data_out_.front().bin.str());
 414         }
 415         data_out_.pop_front();
 416     }
 417     // clear retransmit queue of older items
 418     while (!data_out_tmo_.empty() && data_out_tmo_.front().time<NOW-MAX_POSSIBLE_RTT)
 419         data_out_tmo_.pop_front();
 420 }
 421
 422
 423 void Channel::OnHave (Datagram& dgram) {
 424     bin64_t ackd_pos = dgram.Pull32();
 425     if (ackd_pos==bin64_t::NONE)
 426         return; // wow, peer has hashes
 427     ack_in_.set(ackd_pos);
 428     dprintf("%s #%u -have %s\n",tintstr(),id_,ackd_pos.str());
 429 }
 430
 431
 432 void    Channel::OnHint (Datagram& dgram) {
 433     bin64_t hint = dgram.Pull32();
 434     // FIXME: wake up here
 435     hint_in_.push_back(hint);
 436     dprintf("%s #%u -hint %s\n",tintstr(),id_,hint.str());
 437 }
 438
 439
 440 void Channel::OnHandshake (Datagram& dgram) {
 441     peer_channel_id_ = dgram.Pull32();
 442     dprintf("%s #%u -hs %x\n",tintstr(),id_,peer_channel_id_);
 443     // self-connection check
 444     if (!SELF_CONN_OK) {
 445         uint32_t try_id = DecodeID(peer_channel_id_);
 446         if (channel(try_id) && !channel(try_id)->peer_channel_id_) {
 447             peer_channel_id_ = 0;
 448             Close();
 449             return; // this is a self-connection
 450         }
 451     }
 452     // FUTURE: channel forking
 453 }
 454
 455
 456 void Channel::OnPex (Datagram& dgram) {
 457     uint32_t ipv4 = dgram.Pull32();
 458     uint16_t port = dgram.Pull16();
 459     Address addr(ipv4,port);
 460     dprintf("%s #%u -pex %s\n",tintstr(),id_,addr.str());
 461     transfer().OnPexIn(addr);
 462 }
 463
 464
 465 void    Channel::AddPex (Datagram& dgram) {
 466     int chid = transfer().RevealChannel(pex_out_);
 467     if (chid==-1 || chid==id_)
 468         return;
 469     Address a = channels[chid]->peer();
 470     dgram.Push8(SWIFT_PEX_ADD);
 471     dgram.Push32(a.ipv4());
 472     dgram.Push16(a.port());
 473     dprintf("%s #%u +pex %s\n",tintstr(),id_,a.str());
 474 }
 475
 476
 477 Channel*    Channel::RecvDatagram (int socket) {
 478     Datagram data(socket);
 479     data.Recv();
 480     const Address& addr = data.address();
 481 #define return_log(...) { printf(__VA_ARGS__); return NULL; }
 482     if (data.size()<4)
 483         return_log("datagram shorter than 4 bytes %s\n",addr.str());
 484     uint32_t mych = data.Pull32();
 485     Sha1Hash hash;
 486     Channel* channel = NULL;
 487     if (!mych) { // handshake initiated
 488         if (data.size()<1+4+1+4+Sha1Hash::SIZE)
 489             return_log ("%s #0 incorrect size %i initial handshake packet %s\n",
 490                         tintstr(),data.size(),addr.str());
 491         uint8_t hashid = data.Pull8();
 492         if (hashid!=SWIFT_HASH)
 493             return_log ("%s #0 no hash in the initial handshake %s\n",
 494                         tintstr(),addr.str());
 495         bin64_t pos = data.Pull32();
 496         if (pos!=bin64_t::ALL)
 497             return_log ("%s #0 that is not the root hash %s\n",tintstr(),addr.str());
 498         hash = data.PullHash();
 499         FileTransfer* file = FileTransfer::Find(hash);
 500         if (!file)
 501             return_log ("%s #0 hash %s unknown, no such file %s\n",tintstr(),hash.hex().c_str(),addr.str());
 502         dprintf("%s #0 -hash ALL %s\n",tintstr(),hash.hex().c_str());
 503         for(binqueue::iterator i=file->hs_in_.begin(); i!=file->hs_in_.end(); i++)
 504             if (channels[*i] && channels[*i]->peer_==data.address() &&
 505                 channels[*i]->last_recv_time_>NOW-TINT_SEC*2)
 506                 return_log("%s #0 have a channel already to %s\n",tintstr(),addr.str());
 507         channel = new Channel(file, socket, data.address());
 508     } else {
 509         mych = DecodeID(mych);
 510         if (mych>=channels.size())
 511             return_log("%s invalid channel #%u, %s\n",tintstr(),mych,addr.str());
 512         channel = channels[mych];
 513         if (!channel)
 514             return_log ("%s #%u is already closed\n",tintstr(),mych,addr.str());
 515         if (channel->peer() != addr)
 516             return_log ("%s #%u invalid peer address %s!=%s\n",
 517                         tintstr(),mych,channel->peer().str(),addr.str());
 518         channel->own_id_mentioned_ = true;
 519     }
 520     //dprintf("recvd %i bytes for %i\n",data.size(),channel->id);
 521     channel->Recv(data);
 522     return channel;
 523 }
 524
 525
 526 void    Channel::Loop (tint howlong) {
 527
 528     tint limit = Datagram::Time() + howlong;
 529
 530     do {
 531
 532         tint send_time(TINT_NEVER);
 533         Channel* sender(NULL);
 534         while (!sender && !send_queue.is_empty()) { // dequeue
 535             tintbin next = send_queue.pop();
 536             sender = channel((int)next.bin);
 537             send_time = next.time;
 538             if (sender && sender->next_send_time_!=send_time &&
 539                      sender->next_send_time_!=TINT_NEVER )
 540                 sender = NULL; // it was a stale entry
 541         }
 542
 543         if ( sender!=NULL && send_time<=NOW ) { // it's time
 544
 545             dprintf("%s #%u sch_send %s\n",tintstr(),sender->id(),
 546                     tintstr(send_time));
 547             sender->Send();
 548             sender->Reschedule();
 549
 550         } else {  // it's too early, wait
 551
 552             tint towait = min(limit,send_time) - NOW;
 553             dprintf("%s #0 waiting %lliusec\n",tintstr(),towait);
 554             int rd = Datagram::Wait(socket_count,sockets,towait);
 555             if (rd!=INVALID_SOCKET) { // in meantime, received something
 556                 Channel* receiver = RecvDatagram(rd);
 557                 if (receiver) // receiver's state may have changed
 558                     receiver->Reschedule();
 559             }
 560             if (sender)  // get back to that later
 561                 send_queue.push(tintbin(send_time,sender->id()));
 562
 563         }
 564
 565     } while (NOW<limit);
 566
 567 }
 568
 569
 570 void Channel::Close () {
 571     this->SwitchSendControl(CLOSE_CONTROL);
 572 }
 573
 574
 575 void Channel::Reschedule () {
 576     next_send_time_ = NextSendTime();
 577     if (next_send_time_!=TINT_NEVER) {
 578         assert(next_send_time_<NOW+TINT_MIN);
 579         send_queue.push(tintbin(next_send_time_,id_));
 580         dprintf("%s #%u requeue for %s\n",tintstr(),id_,tintstr(next_send_time_));
 581     } else {
 582         dprintf("%s #%u closed\n",tintstr(),id_);
 583         delete this;
 584     }
 585 }