src/libswift/swift.h

   1 /*
   2  *  swift.h
   3  *  the main header file for libswift, normally you should only read this one
   4  *
   5  *  Created by Victor Grishchenko on 3/6/09.
   6  *  Copyright 2009-2012 TECHNISCHE UNIVERSITEIT DELFT. All rights reserved.
   7  *
   8  */
   9 /*
  10
  11   The swift protocol
  12
  13   Messages
  14
  15   HANDSHAKE    00, channelid
  16   Communicates the channel id of the sender. The
  17   initial handshake packet also has the root hash
  18   (a HASH message).
  19
  20   DATA        01, bin_32, buffer
  21   1K of data.
  22
  23   ACK        02, bin_32, timestamp_32
  24   HAVE       03, bin_32
  25   Confirms successfull delivery of data. Used for
  26   congestion control, as well.
  27
  28   HINT        08, bin_32
  29   Practical value of "hints" is to avoid overlap, mostly.
  30   Hints might be lost in the network or ignored.
  31   Peer might send out data without a hint.
  32   Hint which was not responded (by DATA) in some RTTs
  33   is considered to be ignored.
  34   As peers cant pick randomly kilobyte here and there,
  35   they send out "long hints" for non-base bins.
  36
  37   HASH        04, bin_32, sha1hash
  38   SHA1 hash tree hashes for data verification. The
  39   connection to a fresh peer starts with bootstrapping
  40   him with peak hashes. Later, before sending out
  41   any data, a peer sends the necessary uncle hashes.
  42
  43   PEX+/PEX-    05/06, ipv4 addr, port
  44   Peer exchange messages; reports all connected and
  45   disconected peers. Might has special meaning (as
  46   in the case with swarm supervisors).
  47
  48 */
  49 #ifndef SWIFT_H
  50 #define SWIFT_H
  51
  52 #include <deque>
  53 #include <vector>
  54 #include <set>
  55 #include <algorithm>
  56 #include <string>
  57 #include <map>
  58 #include <math.h>
  59
  60 #include "compat.h"
  61 #include <event2/event.h>
  62 #include <event2/event_struct.h>
  63 #include <event2/buffer.h>
  64 #include "bin.h"
  65 #include "binmap.h"
  66 #include "hashtree.h"
  67 #include "avgspeed.h"
  68 #include "availability.h"
  69 #include "../kernel/mptp.h"
  70
  71 namespace swift {
  72
  73 #define SWIFT_MAX_UDP_OVER_ETH_PAYLOAD          (1500-20-8)
  74 // Arno: Maximum size of non-DATA messages in a UDP packet we send.
  75 #define SWIFT_MAX_NONDATA_DGRAM_SIZE            (SWIFT_MAX_UDP_OVER_ETH_PAYLOAD-SWIFT_DEFAULT_CHUNK_SIZE-1-4)
  76 // Arno: Maximum size of a UDP packet we send. Note: depends on CHUNKSIZE 8192
  77 #define SWIFT_MAX_SEND_DGRAM_SIZE                       (SWIFT_MAX_NONDATA_DGRAM_SIZE+1+4+8192)
  78 // Arno: Maximum size of a UDP packet we are willing to accept. Note: depends on CHUNKSIZE 8192
  79 #define SWIFT_MAX_RECV_DGRAM_SIZE                       (SWIFT_MAX_SEND_DGRAM_SIZE*2)
  80
  81 #define layer2bytes(ln,cs)      (uint64_t)( ((double)cs)*pow(2.0,(double)ln))
  82 #define bytes2layer(bn,cs)  (int)log2(  ((double)bn)/((double)cs) )
  83
  84 // Arno, 2011-12-22: Enable Riccardo's VodPiecePicker
  85 #define ENABLE_VOD_PIECEPICKER          1
  86
  87
  88 /** IPv4 address, just a nice wrapping around struct sockaddr_in. */
  89     struct Address {
  90         struct sockaddr_mptp *addr;
  91         static uint32_t LOCALHOST;
  92         void set_port (uint16_t port) {
  93             addr->dests[0].port = htons(port);
  94         }
  95         void set_port (const char* port_str) {
  96             int p;
  97             if (sscanf(port_str,"%i",&p))
  98                 set_port(p);
  99         }
 100         void set_ipv4 (uint32_t ipv4) {
 101             addr->dests[0].addr = htonl(ipv4);
 102         }
 103         void set_ipv4 (const char* ipv4_str) ;
 104         //{    inet_aton(ipv4_str,&(addr.sin_addr));    }
 105         void clear () {
 106                 addr = (struct sockaddr_mptp *)calloc(1, sizeof(struct sockaddr_mptp) + sizeof(struct mptp_dest));
 107                 addr->count = 1;
 108         }
 109         Address() {
 110             clear();
 111         }
 112         Address(const Address &b) {
 113             clear();
 114                 addr->dests[0].addr = b.addr->dests[0].addr;
 115                 addr->dests[0].port = b.addr->dests[0].port;
 116         }
 117         Address(const char* ip, uint16_t port)  {
 118             clear();
 119             set_ipv4(ip);
 120             set_port(port);
 121         }
 122         Address(const char* ip_port);
 123         Address(uint16_t port) {
 124             clear();
 125             set_ipv4((uint32_t)INADDR_ANY);
 126             set_port(port);
 127         }
 128         Address(uint32_t ipv4addr, uint16_t port) {
 129             clear();
 130             set_ipv4(ipv4addr);
 131             set_port(port);
 132         }
 133         Address(const struct sockaddr_in& address) {
 134                 clear();
 135                 addr->dests[0].addr = address.sin_addr.s_addr;
 136                 addr->dests[0].port = address.sin_port;
 137         }
 138         ~Address(){
 139                 free(addr);
 140         }
 141         uint32_t ipv4 () const { return ntohl(addr->dests[0].addr); }
 142         uint16_t port () const { return ntohs(addr->dests[0].port); }
 143         Address& operator = (const Address& b) {
 144                 if (this != &b) {
 145                         free(addr);
 146                         clear();
 147                         addr->dests[0].addr = b.addr->dests[0].addr;
 148                         addr->dests[0].port = b.addr->dests[0].port;
 149                 }
 150                 return *this;
 151         }
 152         bool operator == (const Address& b) const {
 153             return addr->count == b.addr->count &&
 154                 addr->dests[0].port==b.addr->dests[0].port &&
 155                 addr->dests[0].addr==b.addr->dests[0].addr;
 156         }
 157         const char* str () const {
 158                 // Arno, 2011-10-04: not thread safe, replace.
 159             static char rs[4][32];
 160             static int i;
 161             i = (i+1) & 3;
 162             sprintf(rs[i],"%i.%i.%i.%i:%i",ipv4()>>24,(ipv4()>>16)&0xff,
 163                     (ipv4()>>8)&0xff,ipv4()&0xff,port());
 164             return rs[i];
 165         }
 166         const char* ipv4str () const {
 167                 // Arno, 2011-10-04: not thread safe, replace.
 168             static char rs[4][32];
 169             static int i;
 170             i = (i+1) & 3;
 171             sprintf(rs[i],"%i.%i.%i.%i",ipv4()>>24,(ipv4()>>16)&0xff,
 172                     (ipv4()>>8)&0xff,ipv4()&0xff);
 173             return rs[i];
 174         }
 175         bool operator != (const Address& b) const { return !(*this==b); }
 176         bool is_private() const {
 177                 // TODO IPv6
 178                 uint32_t no = ipv4(); uint8_t no0 = no>>24,no1 = (no>>16)&0xff;
 179                 if (no0 == 10) return true;
 180                 else if (no0 == 172 && no1 >= 16 && no1 <= 31) return true;
 181                 else if (no0 == 192 && no1 == 168) return true;
 182                 else return false;
 183         }
 184     };
 185
 186 // Arno, 2011-10-03: Use libevent callback functions, no on_error?
 187 #define sockcb_t                event_callback_fn
 188     struct sckrwecb_t {
 189         sckrwecb_t (evutil_socket_t s=0, sockcb_t mr=NULL, sockcb_t mw=NULL,
 190                     sockcb_t oe=NULL) :
 191             sock(s), may_read(mr), may_write(mw), on_error(oe) {}
 192         evutil_socket_t sock;
 193         sockcb_t   may_read;
 194         sockcb_t   may_write;
 195         sockcb_t   on_error;
 196     };
 197
 198     struct now_t  {
 199         static tint now;
 200     };
 201
 202 #define NOW now_t::now
 203
 204     /** tintbin is basically a pair<tint,bin64_t> plus some nice operators.
 205         Most frequently used in different queues (acknowledgements, requests,
 206         etc). */
 207     struct tintbin {
 208         tint    time;
 209         bin_t bin;
 210         tintbin(const tintbin& b) : time(b.time), bin(b.bin) {}
 211         tintbin() : time(TINT_NEVER), bin(bin_t::NONE) {}
 212         tintbin(tint time_, bin_t bin_) : time(time_), bin(bin_) {}
 213         tintbin(bin_t bin_) : time(NOW), bin(bin_) {}
 214         bool operator < (const tintbin& b) const
 215                         { return time > b.time; }
 216         bool operator == (const tintbin& b) const
 217                         { return time==b.time && bin==b.bin; }
 218         bool operator != (const tintbin& b) const
 219                         { return !(*this==b); }
 220     };
 221
 222     typedef std::deque<tintbin> tbqueue;
 223     typedef std::deque<bin_t> binqueue;
 224     typedef Address   Address;
 225
 226     /** A heap (priority queue) for timestamped bin numbers (tintbins). */
 227     class tbheap {
 228         tbqueue data_;
 229     public:
 230         int size () const { return data_.size(); }
 231         bool is_empty () const { return data_.empty(); }
 232         tintbin         pop() {
 233             tintbin ret = data_.front();
 234             std::pop_heap(data_.begin(),data_.end());
 235             data_.pop_back();
 236             return ret;
 237         }
 238         void            push(const tintbin& tb) {
 239             data_.push_back(tb);
 240             push_heap(data_.begin(),data_.end());
 241         }
 242         const tintbin&  peek() const {
 243             return data_.front();
 244         }
 245     };
 246
 247     /** swift protocol message types; these are used on the wire. */
 248     typedef enum {
 249         SWIFT_HANDSHAKE = 0,
 250         SWIFT_DATA = 1,
 251         SWIFT_ACK = 2,
 252         SWIFT_HAVE = 3,
 253         SWIFT_HASH = 4,
 254         SWIFT_PEX_ADD = 5,
 255         SWIFT_PEX_REQ = 6,
 256         SWIFT_SIGNED_HASH = 7,
 257         SWIFT_HINT = 8,
 258         SWIFT_MSGTYPE_RCVD = 9,
 259         SWIFT_RANDOMIZE = 10, //FRAGRAND
 260         SWIFT_VERSION = 11, // Arno, 2011-10-19: TODO to match RFC-rev-03
 261         SWIFT_MESSAGE_COUNT = 12
 262     } messageid_t;
 263
 264     typedef enum {
 265         DDIR_UPLOAD,
 266         DDIR_DOWNLOAD
 267     } data_direction_t;
 268
 269     class PiecePicker;
 270     //class CongestionController; // Arno: Currently part of Channel. See ::NextSendTime
 271     class PeerSelector;
 272     class Channel;
 273     typedef void (*ProgressCallback) (int transfer, bin_t bin);
 274
 275     /** A class representing single file transfer. */
 276     class    FileTransfer {
 277
 278     public:
 279
 280         /** A constructor. Open/submit/retrieve a file.
 281          *  @param file_name    the name of the file
 282          *  @param root_hash    the root hash of the file; zero hash if the file
 283                  *                          is newly submitted
 284                  */
 285         FileTransfer(const char *file_name, const Sha1Hash& root_hash=Sha1Hash::ZERO,bool force_check_diskvshash=true,bool check_netwvshash=true,uint32_t chunk_size=SWIFT_DEFAULT_CHUNK_SIZE);
 286
 287         /**    Close everything. */
 288         ~FileTransfer();
 289
 290
 291         /** While we need to feed ACKs to every peer, we try (1) avoid
 292             unnecessary duplication and (2) keep minimum state. Thus,
 293             we use a rotating queue of bin completion events. */
 294         //bin64_t         RevealAck (uint64_t& offset);
 295         /** Rotating queue read for channels of this transmission. */
 296         // Jori
 297         int             RevealChannel (int& i);
 298         // Gertjan
 299         int             RandomChannel (int own_id);
 300
 301
 302         /** Find transfer by the root hash. */
 303         static FileTransfer* Find (const Sha1Hash& hash);
 304         /** Find transfer by the file descriptor. */
 305         static FileTransfer* file (int fd) {
 306             return fd<files.size() ? files[fd] : NULL;
 307         }
 308
 309         /** The binmap for data already retrieved and checked. */
 310         binmap_t&           ack_out ()  { return file_.ack_out(); }
 311         /** Piece picking strategy used by this transfer. */
 312         PiecePicker&    picker () { return *picker_; }
 313         /** The number of channels working for this transfer. */
 314         int             channel_count () const { return hs_in_.size(); }
 315         /** Hash tree checked file; all the hashes and data are kept here. */
 316         HashTree&       file() { return file_; }
 317         /** File descriptor for the data file. */
 318         int             fd () const { return file_.file_descriptor(); }
 319         /** Root SHA1 hash of the transfer (and the data file). */
 320         const Sha1Hash& root_hash () const { return file_.root_hash(); }
 321         /** Ric: the availability in the swarm */
 322         Availability&   availability() { return *availability_; }
 323
 324                 // RATELIMIT
 325         /** Arno: Call when n bytes are received. */
 326         void                    OnRecvData(int n);
 327         /** Arno: Call when n bytes are sent. */
 328         void                    OnSendData(int n);
 329         /** Arno: Call when no bytes are sent due to rate limiting. */
 330         void                    OnSendNoData();
 331         /** Arno: Return current speed for the given direction in bytes/s */
 332                 double                  GetCurrentSpeed(data_direction_t ddir);
 333                 /** Arno: Return maximum speed for the given direction in bytes/s */
 334                 double                  GetMaxSpeed(data_direction_t ddir);
 335                 /** Arno: Set maximum speed for the given direction in bytes/s */
 336                 void                    SetMaxSpeed(data_direction_t ddir, double m);
 337                 /** Arno: Return the number of non-seeders current channeled with. */
 338                 uint32_t                GetNumLeechers();
 339                 /** Arno: Return the number of seeders current channeled with. */
 340                 uint32_t                GetNumSeeders();
 341                 /** Arno: Return the set of Channels for this transfer. MORESTATS */
 342                 std::set<Channel *> GetChannels() { return mychannels_; }
 343
 344                 /** Arno: set the tracker for this transfer. Reseting it won't kill
 345                  * any existing connections.
 346                  */
 347                 void SetTracker(Address tracker) { tracker_ = tracker; }
 348
 349                 /** Arno: (Re)Connect to tracker for this transfer, or global Channel::tracker if not set */
 350                 void ConnectToTracker();
 351
 352                 /** Arno: Reconnect to the tracker if no established peers and
 353                  * exp backoff allows it.
 354                  */
 355                 void ReConnectToTrackerIfAllowed(bool hasestablishedpeers);
 356
 357                 /** Arno: Return the Channel to peer "addr" that is not equal to "notc". */
 358                 Channel * FindChannel(const Address &addr, Channel *notc);
 359
 360                 // SAFECLOSE
 361                 static void LibeventCleanCallback(int fd, short event, void *arg);
 362     protected:
 363
 364         HashTree        file_;
 365
 366         /** Piece picker strategy. */
 367         PiecePicker*    picker_;
 368
 369         /** Channels working for this transfer. */
 370         binqueue        hs_in_;                 // Arno, 2011-10-03: Should really be queue of channel ID (=uint32_t)
 371
 372         /** Messages we are accepting.    */
 373         uint64_t        cap_out_;
 374
 375         tint            init_time_;
 376
 377         // Ric: PPPLUG
 378         /** Availability in the swarm */
 379         Availability*   availability_;
 380
 381 #define SWFT_MAX_TRANSFER_CB 8
 382         ProgressCallback callbacks[SWFT_MAX_TRANSFER_CB];
 383         uint8_t         cb_agg[SWFT_MAX_TRANSFER_CB];
 384         int             cb_installed;
 385
 386                 // RATELIMIT
 387         std::set<Channel *>     mychannels_; // Arno, 2012-01-31: May be duplicate of hs_in_
 388         MovingAverageSpeed      cur_speed_[2];
 389         double                          max_speed_[2];
 390         int                                     speedzerocount_;
 391
 392         // SAFECLOSE
 393         struct event            evclean_;
 394
 395         Address                         tracker_; // Tracker for this transfer
 396         tint                            tracker_retry_interval_;
 397         tint                            tracker_retry_time_;
 398
 399     public:
 400         void            OnDataIn (bin_t pos);
 401         // Gertjan fix: return bool
 402         bool            OnPexIn (const Address& addr);
 403
 404         static std::vector<FileTransfer*> files;
 405
 406
 407         friend class Channel;
 408         // Ric: maybe not really needed
 409         friend class Availability;
 410         friend uint64_t  Size (int fdes);
 411         friend bool      IsComplete (int fdes);
 412         friend uint64_t  Complete (int fdes);
 413         friend uint64_t  SeqComplete (int fdes);
 414         friend int     Open (const char* filename, const Sha1Hash& hash, Address tracker, bool force_check_diskvshash, bool check_netwvshash, uint32_t chunk_size);
 415         friend void    Close (int fd) ;
 416         friend void AddProgressCallback (int transfer,ProgressCallback cb,uint8_t agg);
 417         friend void RemoveProgressCallback (int transfer,ProgressCallback cb);
 418         friend void ExternallyRetrieved (int transfer,bin_t piece);
 419     };
 420
 421
 422     /** PiecePicker implements some strategy of choosing (picking) what
 423         to request next, given the possible range of choices:
 424         data acknowledged by the peer minus data already retrieved.
 425         May pick sequentially, do rarest first or in some other way. */
 426     class PiecePicker {
 427     public:
 428         virtual void Randomize (uint64_t twist) = 0;
 429         /** The piece picking method itself.
 430          *  @param  offered     the data acknowledged by the peer
 431          *  @param  max_width   maximum number of packets to ask for
 432          *  @param  expires     (not used currently) when to consider request expired
 433          *  @return             the bin number to request */
 434         virtual bin_t Pick (binmap_t& offered, uint64_t max_width, tint expires) = 0;
 435         virtual void LimitRange (bin_t range) = 0;
 436         /** updates the playback position for streaming piece picking.
 437          *  @param  amount              amount to increment in bin unit size (1KB default) */
 438         virtual void updatePlaybackPos (int amount=1) = 0;
 439         virtual ~PiecePicker() {}
 440     };
 441
 442
 443     class PeerSelector { // Arno: partically unused
 444     public:
 445         virtual void AddPeer (const Address& addr, const Sha1Hash& root) = 0;
 446         virtual Address GetPeer (const Sha1Hash& for_root) = 0;
 447     };
 448
 449
 450     /* class DataStorer { // Arno: never implemented
 451     public:
 452         DataStorer (const Sha1Hash& id, size_t size);
 453         virtual size_t    ReadData (bin_t pos,uint8_t** buf) = 0;
 454         virtual size_t    WriteData (bin_t pos, uint8_t* buf, size_t len) = 0;
 455     }; */
 456
 457
 458     /**    swift channel's "control block"; channels loosely correspond to TCP
 459            connections or FTP sessions; one channel is created for one file
 460            being transferred between two peers. As we don't need buffers and
 461            lots of other TCP stuff, sizeof(Channel+members) must be below 1K.
 462            Normally, API users do not deal with this class. */
 463         class MessageQueue;
 464     class Channel {
 465
 466 #define DGRAM_MAX_SOCK_OPEN 128
 467         static int sock_count;
 468         static sckrwecb_t sock_open[DGRAM_MAX_SOCK_OPEN];
 469
 470     public:
 471         Channel    (FileTransfer* file, int socket=INVALID_SOCKET, Address peer=Address());
 472         ~Channel();
 473
 474         typedef enum {
 475             KEEP_ALIVE_CONTROL,
 476             PING_PONG_CONTROL,
 477             SLOW_START_CONTROL,
 478             AIMD_CONTROL,
 479             LEDBAT_CONTROL,
 480             CLOSE_CONTROL
 481         } send_control_t;
 482
 483         static Address  tracker; // Global tracker for all transfers
 484         struct event *evsend_ptr_; // Arno: timer per channel // SAFECLOSE
 485         static struct event_base *evbase;
 486         static struct event evrecv;
 487         static const char* SEND_CONTROL_MODES[];
 488
 489                 static MessageQueue messageQueue;
 490
 491             static tint epoch, start;
 492             static uint64_t global_dgrams_up, global_dgrams_down, global_raw_bytes_up, global_raw_bytes_down, global_bytes_up, global_bytes_down,
 493                                                 global_buffers_up, global_syscalls_up, global_buffers_down, global_syscalls_down;
 494         static void CloseChannelByAddress(const Address &addr);
 495
 496         // SOCKMGMT
 497         // Arno: channel is also a "singleton" class that manages all sockets
 498         // for a swift process
 499         static void LibeventSendCallback(int fd, short event, void *arg);
 500         static void LibeventReceiveCallback(int fd, short event, void *arg);
 501         static void RecvDatagram (evutil_socket_t socket); // Called by LibeventReceiveCallback
 502             static int RecvFrom(evutil_socket_t sock, Address& addr, struct evbuffer **evb); // Called by RecvDatagram
 503             static int SendTo(evutil_socket_t sock, const Address& addr, struct evbuffer **evb); // Called by Channel::Send()
 504             static evutil_socket_t Bind(Address address, sckrwecb_t callbacks=sckrwecb_t());
 505             static Address BoundAddress(evutil_socket_t sock);
 506             static evutil_socket_t default_socket()
 507             { return sock_count ? sock_open[0].sock : INVALID_SOCKET; }
 508
 509             /** close the port */
 510             static void CloseSocket(evutil_socket_t sock);
 511             static void Shutdown ();
 512             /** the current time */
 513             static tint Time();
 514
 515             // Arno: Per instance methods
 516         void        Recv (struct evbuffer *evb);
 517         void        Send ();  // Called by LibeventSendCallback
 518         void        Close ();
 519
 520         void        OnAck (struct evbuffer *evb);
 521         void        OnHave (struct evbuffer *evb);
 522         bin_t       OnData (struct evbuffer *evb);
 523         void        OnHint (struct evbuffer *evb);
 524         void        OnHash (struct evbuffer *evb);
 525         void        OnPex (struct evbuffer *evb);
 526         void        OnHandshake (struct evbuffer *evb);
 527         void        OnRandomize (struct evbuffer *evb); //FRAGRAND
 528         void        AddHandshake (struct evbuffer *evb);
 529         bin_t       AddData (struct evbuffer **evb);
 530         void        AddAck (struct evbuffer *evb);
 531         void        AddHave (struct evbuffer *evb);
 532         void        AddHint (struct evbuffer *evb);
 533         void        AddUncleHashes (struct evbuffer *evb, bin_t pos);
 534         void        AddPeakHashes (struct evbuffer *evb);
 535         void        AddPex (struct evbuffer *evb);
 536         void        OnPexReq(void);
 537         void        AddPexReq(struct evbuffer *evb);
 538         void        BackOffOnLosses (float ratio=0.5);
 539         tint        SwitchSendControl (int control_mode);
 540         tint        NextSendTime ();
 541         tint        KeepAliveNextSendTime ();
 542         tint        PingPongNextSendTime ();
 543         tint        CwndRateNextSendTime ();
 544         tint        SlowStartNextSendTime ();
 545         tint        AimdNextSendTime ();
 546         tint        LedbatNextSendTime ();
 547         /** Arno: return true if this peer has complete file. May be fuzzy if Peak Hashes not in */
 548         bool            IsComplete();
 549         /** Arno: return (UDP) port for this channel */
 550         uint16_t        GetMyPort();
 551         bool            IsDiffSenderOrDuplicate(Address addr, uint32_t chid);
 552
 553         static int  MAX_REORDERING;
 554         static tint TIMEOUT;
 555         static tint MIN_DEV;
 556         static tint MAX_SEND_INTERVAL;
 557         static tint LEDBAT_TARGET;
 558         static float LEDBAT_GAIN;
 559         static tint LEDBAT_DELAY_BIN;
 560         static bool SELF_CONN_OK;
 561         static tint MAX_POSSIBLE_RTT;
 562         static tint MIN_PEX_REQUEST_INTERVAL;
 563         static FILE* debug_file;
 564
 565         const std::string id_string () const;
 566         /** A channel is "established" if had already sent and received packets. */
 567         bool        is_established () { return peer_channel_id_ && own_id_mentioned_; }
 568         FileTransfer& transfer() { return *transfer_; }
 569         HashTree&   file () { return transfer_->file(); }
 570         const Address& peer() const { return peer_; }
 571         const Address& recv_peer() const { return recv_peer_; }
 572         tint ack_timeout () {
 573                 tint dev = dev_avg_ < MIN_DEV ? MIN_DEV : dev_avg_;
 574                 tint tmo = rtt_avg_ + dev * 4;
 575                 return tmo < 30*TINT_SEC ? tmo : 30*TINT_SEC;
 576         }
 577         uint32_t    id () const { return id_; }
 578
 579         // MORESTATS
 580         uint64_t raw_bytes_up() { return raw_bytes_up_; }
 581         uint64_t raw_bytes_down() { return raw_bytes_down_; }
 582         uint64_t bytes_up() { return bytes_up_; }
 583         uint64_t bytes_down() { return bytes_down_; }
 584
 585         static int  DecodeID(int scrambled);
 586         static int  EncodeID(int unscrambled);
 587         static Channel* channel(int i) {
 588             return i<channels.size()?channels[i]:NULL;
 589         }
 590         static void CloseTransfer (FileTransfer* trans);
 591
 592         // SAFECLOSE
 593         void            ClearEvents();
 594         void            Schedule4Close() { scheduled4close_ = true; }
 595         bool            IsScheduled4Close() { return scheduled4close_; }
 596
 597                 void Sent(int bytes, evbuffer *evb, bool tofree);
 598
 599
 600     protected:
 601         /** Channel id: index in the channel array. */
 602         uint32_t    id_;
 603         /**    Socket address of the peer. */
 604         Address     peer_;
 605         /**    The UDP socket fd. */
 606         evutil_socket_t      socket_;
 607         /**    Descriptor of the file in question. */
 608         FileTransfer*    transfer_;
 609         /**    Peer channel id; zero if we are trying to open a channel. */
 610         uint32_t    peer_channel_id_;
 611         bool        own_id_mentioned_;
 612         /**    Peer's progress, based on acknowledgements. */
 613         binmap_t    ack_in_;
 614         /**    Last data received; needs to be acked immediately. */
 615         tintbin     data_in_;
 616         bin_t       data_in_dbl_;
 617         /** The history of data sent and still unacknowledged. */
 618         tbqueue     data_out_;
 619         /** Timeouted data (potentially to be retransmitted). */
 620         tbqueue     data_out_tmo_;
 621         bin_t       data_out_cap_;
 622         /** Index in the history array. */
 623         binmap_t    have_out_;
 624         /**    Transmit schedule: in most cases filled with the peer's hints */
 625         tbqueue     hint_in_;
 626         /** Hints sent (to detect and reschedule ignored hints). */
 627         tbqueue     hint_out_;
 628         uint64_t    hint_out_size_;
 629         /** Types of messages the peer accepts. */
 630         uint64_t    cap_in_;
 631         /** For repeats. */
 632         //tint        last_send_time, last_recv_time;
 633         /** PEX progress */
 634         bool        pex_requested_;
 635         tint        last_pex_request_time_;
 636         tint        next_pex_request_time_;
 637         bool        pex_request_outstanding_;
 638         tbqueue     reverse_pex_out_;           // Arno, 2011-10-03: should really be a queue of (tint,channel id(= uint32_t)) pairs.
 639         int         useless_pex_count_;
 640         /** Smoothed averages for RTT, RTT deviation and data interarrival periods. */
 641         tint        rtt_avg_, dev_avg_, dip_avg_;
 642         tint        last_send_time_;
 643         tint        last_recv_time_;
 644         tint        last_data_out_time_;
 645         tint        last_data_in_time_;
 646         tint        last_loss_time_;
 647         tint        next_send_time_;
 648         /** Congestion window; TODO: int, bytes. */
 649         float       cwnd_;
 650         int         cwnd_count1_;
 651         /** Data sending interval. */
 652         tint        send_interval_;
 653         /** The congestion control strategy. */
 654         int         send_control_;
 655         /** Datagrams (not data) sent since last recv.    */
 656         int         sent_since_recv_;
 657
 658         /** Arno: Fix for KEEP_ALIVE_CONTROL */
 659         bool            lastrecvwaskeepalive_;
 660         bool            lastsendwaskeepalive_;
 661
 662         /** Recent acknowlegements for data previously sent.    */
 663         int         ack_rcvd_recent_;
 664         /** Recent non-acknowlegements (losses) of data previously sent.    */
 665         int         ack_not_rcvd_recent_;
 666         /** LEDBAT one-way delay machinery */
 667         tint        owd_min_bins_[4];
 668         int         owd_min_bin_;
 669         tint        owd_min_bin_start_;
 670         tint        owd_current_[4];
 671         int         owd_cur_bin_;
 672         /** Stats */
 673         int         dgrams_sent_;
 674         int         dgrams_rcvd_;
 675         // Arno, 2011-11-28: for detailed, per-peer stats. MORESTATS
 676         uint64_t raw_bytes_up_, raw_bytes_down_, bytes_up_, bytes_down_;
 677
 678         // SAFECLOSE
 679         bool            scheduled4close_;
 680         /** Arno: Socket address of the peer where packets are received from,
 681          * when an IANA private address, otherwise 0.
 682          * May not be equal to peer_. 2PEERSBEHINDSAMENAT */
 683         Address     recv_peer_;
 684
 685         int         PeerBPS() const {
 686             return TINT_SEC / dip_avg_ * 1024;
 687         }
 688         /** Get a request for one packet from the queue of peer's requests. */
 689         bin_t       DequeueHint(bool *retransmitptr);
 690         bin_t       ImposeHint();
 691         void        TimeoutDataOut ();
 692         void        CleanStaleHintOut();
 693         void        CleanHintOut(bin_t pos);
 694         void        Reschedule();
 695         void            UpdateDIP(bin_t pos); // RETRANSMIT
 696
 697
 698         static PeerSelector* peer_selector;
 699
 700         static tint     last_tick;
 701         //static tbheap   send_queue;
 702
 703         static std::vector<Channel*> channels;
 704
 705         friend int      Listen (Address addr);
 706         friend void     Shutdown (int sock_des);
 707         friend void     AddPeer (Address address, const Sha1Hash& root);
 708         friend void     SetTracker(const Address& tracker);
 709         friend int      Open (const char*, const Sha1Hash&, Address tracker, bool force_check_diskvshash, bool check_netwvshash, uint32_t chunk_size) ; // FIXME
 710     };
 711
 712
 713
 714     /*************** The top-level API ****************/
 715     /** Start listening a port. Returns socket descriptor. */
 716     int     Listen (Address addr);
 717     /** Stop listening to a port. */
 718     void    Shutdown (int sock_des=-1);
 719
 720     /** Open a file, start a transmission; fill it with content for a given
 721         root hash and tracker (optional). If "force_check_diskvshash" is true, the
 722         hashtree state will be (re)constructed from the file on disk (if any).
 723         If not, open will try to reconstruct the hashtree state from
 724         the .mhash and .mbinmap files on disk. .mhash files are created
 725         automatically, .mbinmap files must be written by checkpointing the
 726         transfer by calling FileTransfer::serialize(). If the reconstruction
 727         fails, it will hashcheck anyway. Roothash is optional for new files or
 728         files already hashchecked and checkpointed. If "check_netwvshash" is
 729         false, no uncle hashes will be sent and no data will be verified against
 730         then on receipt. In this mode, checking disk contents against hashes
 731         no longer works on restarts, unless checkpoints are used.
 732         */
 733     int     Open (const char* filename, const Sha1Hash& hash=Sha1Hash::ZERO,Address tracker=Address(), bool force_check_diskvshash=true, bool check_netwvshash=true, uint32_t chunk_size=SWIFT_DEFAULT_CHUNK_SIZE);
 734     /** Get the root hash for the transmission. */
 735     const Sha1Hash& RootMerkleHash (int file) ;
 736     /** Close a file and a transmission. */
 737     void    Close (int fd) ;
 738     /** Add a possible peer which participares in a given transmission. In the case
 739         root hash is zero, the peer might be talked to regarding any transmission
 740         (likely, a tracker, cache or an archive). */
 741     void    AddPeer (Address address, const Sha1Hash& root=Sha1Hash::ZERO);
 742
 743     void    SetTracker(const Address& tracker);
 744     /** Set the default tracker that is used when Open is not passed a tracker
 745         address. */
 746
 747     /** Returns size of the file in bytes, 0 if unknown. Might be rounded up to a kilobyte
 748         before the transmission is complete. */
 749     uint64_t  Size (int fdes);
 750     /** Returns the amount of retrieved and verified data, in bytes.
 751         A 100% complete transmission has Size()==Complete(). */
 752     uint64_t  Complete (int fdes);
 753     bool      IsComplete (int fdes);
 754     /** Returns the number of bytes that are complete sequentially, starting from the
 755         beginning, till the first not-yet-retrieved packet. */
 756     uint64_t  SeqComplete (int fdes);
 757     /***/
 758     int       Find (Sha1Hash hash);
 759     /** Returns the number of bytes in a chunk for this transmission */
 760     uint32_t      ChunkSize(int fdes);
 761
 762     /** Get the address bound to the socket descriptor returned by Listen() */
 763     Address BoundAddress(evutil_socket_t sock);
 764
 765     void AddProgressCallback (int transfer,ProgressCallback cb,uint8_t agg);
 766     void RemoveProgressCallback (int transfer,ProgressCallback cb);
 767     void ExternallyRetrieved (int transfer,bin_t piece);
 768
 769
 770     /** Must be called by any client using the library */
 771     void LibraryInit(void);
 772
 773     int evbuffer_add_string(struct evbuffer *evb, std::string str);
 774     int evbuffer_add_8(struct evbuffer *evb, uint8_t b);
 775     int evbuffer_add_16be(struct evbuffer *evb, uint16_t w);
 776     int evbuffer_add_32be(struct evbuffer *evb, uint32_t i);
 777     int evbuffer_add_64be(struct evbuffer *evb, uint64_t l);
 778     int evbuffer_add_hash(struct evbuffer *evb, const Sha1Hash& hash);
 779
 780     uint8_t evbuffer_remove_8(struct evbuffer *evb);
 781     uint16_t evbuffer_remove_16be(struct evbuffer *evb);
 782     uint32_t evbuffer_remove_32be(struct evbuffer *evb);
 783     uint64_t evbuffer_remove_64be(struct evbuffer *evb);
 784     Sha1Hash evbuffer_remove_hash(struct evbuffer* evb);
 785
 786     const char* tintstr(tint t=0);
 787     std::string sock2str (struct sockaddr_in addr);
 788  #define SWIFT_MAX_CONNECTIONS 20
 789
 790     void nat_test_update(void);
 791
 792     // Arno: Save transfer's binmap for zero-hashcheck restart
 793     void Checkpoint(int fdes);
 794
 795 #define MAX_QUEUE_LENGTH 1
 796 #define TIMER_USEC 10000
 797
 798         class MessageQueue
 799         {
 800         public:
 801                 class Entry
 802                 {
 803                 public:
 804                         Entry(evbuffer *ievb, const Address &iaddr, Channel *ichannel, bool itofree)
 805                                 :
 806                                         evb(ievb),
 807                                         addr(iaddr),
 808                                         channel(ichannel),
 809                                         tofree(itofree)
 810                         {
 811                         }
 812
 813                         evbuffer *evb;
 814                         Address addr;
 815                         Channel *channel;
 816                         bool tofree;
 817                 };
 818
 819                 typedef std::deque<Entry> EntryList;
 820                 typedef std::map<int, EntryList> EntryLists;
 821
 822                 void AddBuffer(int sock, evbuffer *evb, const Address &addr, Channel *channel, bool tofree = true)
 823                 {
 824                         EntryList &list = lists[sock];
 825                         list.push_back(Entry(evb, addr, channel, tofree));
 826                         if (list.size() == MAX_QUEUE_LENGTH)
 827                                 Flush(sock);
 828                 }
 829
 830                 void Flush(int sock)
 831                 {
 832                         EntryList &list = lists[sock];
 833                         if (list.empty())
 834                                 return;
 835
 836                         Address addr;
 837                         free(addr.addr);
 838                         addr.addr = (struct sockaddr_mptp *) calloc(1, sizeof(struct sockaddr_mptp) + list.size() * sizeof(struct mptp_dest));
 839                         addr.addr->count = list.size();
 840                         evbuffer *evbs[list.size()];
 841                         int i = 0;
 842                         for (EntryList::iterator it = list.begin(); it != list.end(); ++it, ++i) {
 843                                 addr.addr->dests[i].addr = (*it).addr.addr->dests[0].addr;
 844                                 addr.addr->dests[i].port = (*it).addr.addr->dests[0].port;
 845                                 evbs[i] = (*it).evb;
 846                         }
 847
 848                         int r = Channel::SendTo(sock, addr, evbs);
 849                         if (r > 0) {
 850                                 i = 0;
 851                                 for (EntryList::iterator it = list.begin(); it != list.end(); ++it, ++i)
 852                                         (*it).channel->Sent(evbuffer_get_length((*it).evb), (*it).evb, (*it).tofree);
 853                         }
 854                         list.clear();
 855                 }
 856
 857                 void Flush()
 858                 {
 859                         for (EntryLists::iterator it = lists.begin(); it != lists.end(); ++it)
 860                                 Flush(it->first);
 861                 }
 862
 863         private:
 864                 EntryLists lists;
 865         };
 866
 867 } // namespace end
 868
 869 // #define SWIFT_MUTE
 870
 871 #ifndef SWIFT_MUTE
 872 #define dprintf(...) do { if (Channel::debug_file) fprintf(Channel::debug_file,__VA_ARGS__); } while (0)
 873 #define dflush() fflush(Channel::debug_file)
 874 #else
 875 #define dprintf(...) do {} while(0)
 876 #define dflush() do {} while(0)
 877 #endif
 878 #define eprintf(...) fprintf(stderr,__VA_ARGS__)
 879
 880 #endif