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 Delft University of Technology. 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 <algorithm>
  55 #include <string>
  56 #include "bin64.h"
  57 #include "bins.h"
  58 #include "datagram.h"
  59 #include "hashtree.h"
  60
  61 namespace swift {
  62
  63     #define NOW Datagram::now
  64
  65     /** tintbin is basically a pair<tint,bin64_t> plus some nice operators.
  66         Most frequently used in different queues (acknowledgements, requests,
  67         etc). */
  68     struct tintbin {
  69         tint    time;
  70         bin64_t bin;
  71         tintbin(const tintbin& b) : time(b.time), bin(b.bin) {}
  72         tintbin() : time(TINT_NEVER), bin(bin64_t::NONE) {}
  73         tintbin(tint time_, bin64_t bin_) : time(time_), bin(bin_) {}
  74         tintbin(bin64_t bin_) : time(NOW), bin(bin_) {}
  75         bool operator < (const tintbin& b) const
  76             { return time > b.time; }
  77         bool operator == (const tintbin& b) const
  78             { return time==b.time && bin==b.bin; }
  79         bool operator != (const tintbin& b) const
  80             { return !(*this==b); }
  81     };
  82
  83     typedef std::deque<tintbin> tbqueue;
  84     typedef std::deque<bin64_t> binqueue;
  85     typedef Address   Address;
  86
  87     /** A heap (priority queue) for timestamped bin numbers (tintbins). */
  88     class tbheap {
  89         tbqueue data_;
  90     public:
  91         int size () const { return data_.size(); }
  92         bool is_empty () const { return data_.empty(); }
  93         tintbin         pop() {
  94             tintbin ret = data_.front();
  95             std::pop_heap(data_.begin(),data_.end());
  96             data_.pop_back();
  97             return ret;
  98         }
  99         void            push(const tintbin& tb) {
 100             data_.push_back(tb);
 101             push_heap(data_.begin(),data_.end());
 102         }
 103         const tintbin&  peek() const {
 104             return data_.front();
 105         }
 106     };
 107
 108     /** swift protocol message types; these are used on the wire. */
 109     typedef enum {
 110         SWIFT_HANDSHAKE = 0,
 111         SWIFT_DATA = 1,
 112         SWIFT_ACK = 2,
 113         SWIFT_HAVE = 3,
 114         SWIFT_HASH = 4,
 115         SWIFT_PEX_ADD = 5,
 116         SWIFT_PEX_RM = 6,
 117         SWIFT_SIGNED_HASH = 7,
 118         SWIFT_HINT = 8,
 119         SWIFT_MSGTYPE_RCVD = 9,
 120         SWIFT_MESSAGE_COUNT = 10
 121     } messageid_t;
 122
 123     class PiecePicker;
 124     class CongestionController;
 125     class PeerSelector;
 126     typedef void (*ProgressCallback) (int transfer, bin64_t bin);
 127
 128
 129     /** A class representing single file transfer. */
 130     class    FileTransfer {
 131
 132     public:
 133
 134         /** A constructor. Open/submit/retrieve a file.
 135          *  @param file_name    the name of the file
 136          *  @param root_hash    the root hash of the file; zero hash if the file
 137                                 is newly submitted */
 138         FileTransfer(const char *file_name, const Sha1Hash& root_hash=Sha1Hash::ZERO);
 139
 140         /**    Close everything. */
 141         ~FileTransfer();
 142
 143
 144         /** While we need to feed ACKs to every peer, we try (1) avoid
 145             unnecessary duplication and (2) keep minimum state. Thus,
 146             we use a rotating queue of bin completion events. */
 147         //bin64_t         RevealAck (uint64_t& offset);
 148         /** Rotating queue read for channels of this transmission. */
 149         int             RevealChannel (int& i);
 150
 151         /** Find transfer by the root hash. */
 152         static FileTransfer* Find (const Sha1Hash& hash);
 153         /** Find transfer by the file descriptor. */
 154         static FileTransfer* file (int fd) {
 155             return fd<files.size() ? files[fd] : NULL;
 156         }
 157
 158         /** The binmap for data already retrieved and checked. */
 159         binmap_t&           ack_out ()  { return file_.ack_out(); }
 160         /** Piece picking strategy used by this transfer. */
 161         PiecePicker&    picker () { return *picker_; }
 162         /** The number of channels working for this transfer. */
 163         int             channel_count () const { return hs_in_.size(); }
 164         /** Hash tree checked file; all the hashes and data are kept here. */
 165         HashTree&       file() { return file_; }
 166         /** File descriptor for the data file. */
 167         int             fd () const { return file_.file_descriptor(); }
 168         /** Root SHA1 hash of the transfer (and the data file). */
 169         const Sha1Hash& root_hash () const { return file_.root_hash(); }
 170
 171     private:
 172
 173         static std::vector<FileTransfer*> files;
 174
 175         HashTree        file_;
 176
 177         /** Piece picker strategy. */
 178         PiecePicker*    picker_;
 179
 180         /** Channels working for this transfer. */
 181         binqueue        hs_in_;
 182         int             hs_in_offset_;
 183         std::deque<Address> pex_in_;
 184
 185         /** Messages we are accepting.    */
 186         uint64_t        cap_out_;
 187
 188         tint            init_time_;
 189
 190         #define SWFT_MAX_TRANSFER_CB 8
 191         ProgressCallback callbacks[SWFT_MAX_TRANSFER_CB];
 192         uint8_t         cb_agg[SWFT_MAX_TRANSFER_CB];
 193         int             cb_installed;
 194
 195     public:
 196         void            OnDataIn (bin64_t pos);
 197         void            OnPexIn (const Address& addr);
 198
 199         friend class Channel;
 200         friend uint64_t  Size (int fdes);
 201         friend bool      IsComplete (int fdes);
 202         friend uint64_t  Complete (int fdes);
 203         friend uint64_t  SeqComplete (int fdes);
 204         friend int     Open (const char* filename, const Sha1Hash& hash) ;
 205         friend void    Close (int fd) ;
 206         friend void AddProgressCallback (int transfer,ProgressCallback cb,uint8_t agg);
 207         friend void RemoveProgressCallback (int transfer,ProgressCallback cb);
 208         friend void ExternallyRetrieved (int transfer,bin64_t piece);
 209     };
 210
 211
 212     /** PiecePicker implements some strategy of choosing (picking) what
 213         to request next, given the possible range of choices:
 214         data acknowledged by the peer minus data already retrieved.
 215         May pick sequentially, do rarest first or in some other way. */
 216     class PiecePicker {
 217     public:
 218         virtual void Randomize (uint64_t twist) = 0;
 219         /** The piece picking method itself.
 220          *  @param  offered     the daata acknowledged by the peer
 221          *  @param  max_width   maximum number of packets to ask for
 222          *  @param  expires     (not used currently) when to consider request expired
 223          *  @return             the bin number to request */
 224         virtual bin64_t Pick (binmap_t& offered, uint64_t max_width, tint expires) = 0;
 225         virtual void LimitRange (bin64_t range) = 0;
 226         virtual ~PiecePicker() {}
 227     };
 228
 229
 230     class PeerSelector {
 231     public:
 232         virtual void AddPeer (const Address& addr, const Sha1Hash& root) = 0;
 233         virtual Address GetPeer (const Sha1Hash& for_root) = 0;
 234     };
 235
 236
 237     class DataStorer {
 238     public:
 239         DataStorer (const Sha1Hash& id, size_t size);
 240         virtual size_t    ReadData (bin64_t pos,uint8_t** buf) = 0;
 241         virtual size_t    WriteData (bin64_t pos, uint8_t* buf, size_t len) = 0;
 242     };
 243
 244
 245     /**    swift channel's "control block"; channels loosely correspond to TCP
 246         connections or FTP sessions; one channel is created for one file
 247         being transferred between two peers. As we don't need buffers and
 248         lots of other TCP stuff, sizeof(Channel+members) must be below 1K.
 249         Normally, API users do not deal with this class. */
 250     class Channel {
 251     public:
 252         Channel    (FileTransfer* file, int socket=INVALID_SOCKET, Address peer=Address());
 253         ~Channel();
 254
 255         typedef enum {
 256             KEEP_ALIVE_CONTROL,
 257             PING_PONG_CONTROL,
 258             SLOW_START_CONTROL,
 259             AIMD_CONTROL,
 260             LEDBAT_CONTROL,
 261             CLOSE_CONTROL
 262         } send_control_t;
 263
 264         static const char* SEND_CONTROL_MODES[];
 265
 266         static void RecvDatagram (SOCKET socket);
 267         static void Loop (tint till);
 268
 269         void        Recv (Datagram& dgram);
 270         void        Send ();
 271         void        Close ();
 272
 273         void        OnAck (Datagram& dgram);
 274         void        OnHave (Datagram& dgram);
 275         bin64_t     OnData (Datagram& dgram);
 276         void        OnHint (Datagram& dgram);
 277         void        OnHash (Datagram& dgram);
 278         void        OnPex (Datagram& dgram);
 279         void        OnHandshake (Datagram& dgram);
 280         void        AddHandshake (Datagram& dgram);
 281         bin64_t     AddData (Datagram& dgram);
 282         void        AddAck (Datagram& dgram);
 283         void        AddHave (Datagram& dgram);
 284         void        AddHint (Datagram& dgram);
 285         void        AddUncleHashes (Datagram& dgram, bin64_t pos);
 286         void        AddPeakHashes (Datagram& dgram);
 287         void        AddPex (Datagram& dgram);
 288
 289         void        BackOffOnLosses (float ratio=0.5);
 290         tint        SwitchSendControl (int control_mode);
 291         tint        NextSendTime ();
 292         tint        KeepAliveNextSendTime ();
 293         tint        PingPongNextSendTime ();
 294         tint        CwndRateNextSendTime ();
 295         tint        SlowStartNextSendTime ();
 296         tint        AimdNextSendTime ();
 297         tint        LedbatNextSendTime ();
 298
 299         static int  MAX_REORDERING;
 300         static tint TIMEOUT;
 301         static tint MIN_DEV;
 302         static tint MAX_SEND_INTERVAL;
 303         static tint LEDBAT_TARGET;
 304         static float LEDBAT_GAIN;
 305         static tint LEDBAT_DELAY_BIN;
 306         static bool SELF_CONN_OK;
 307         static tint MAX_POSSIBLE_RTT;
 308         static FILE* debug_file;
 309
 310         const std::string id_string () const;
 311         /** A channel is "established" if had already sent and received packets. */
 312         bool        is_established () { return peer_channel_id_ && own_id_mentioned_; }
 313         FileTransfer& transfer() { return *transfer_; }
 314         HashTree&   file () { return transfer_->file(); }
 315         const Address& peer() const { return peer_; }
 316         tint ack_timeout () {
 317                         tint dev = dev_avg_ < MIN_DEV ? MIN_DEV : dev_avg_;
 318                         tint tmo = rtt_avg_ + dev * 4;
 319                         return tmo < 30*TINT_SEC ? tmo : 30*TINT_SEC;
 320         }
 321         uint32_t    id () const { return id_; }
 322
 323         static int  DecodeID(int scrambled);
 324         static int  EncodeID(int unscrambled);
 325         static Channel* channel(int i) {
 326             return i<channels.size()?channels[i]:NULL;
 327         }
 328         static void CloseTransfer (FileTransfer* trans);
 329
 330     protected:
 331         /** Channel id: index in the channel array. */
 332         uint32_t    id_;
 333         /**    Socket address of the peer. */
 334         Address     peer_;
 335         /**    The UDP socket fd. */
 336         SOCKET      socket_;
 337         /**    Descriptor of the file in question. */
 338         FileTransfer*    transfer_;
 339         /**    Peer channel id; zero if we are trying to open a channel. */
 340         uint32_t    peer_channel_id_;
 341         bool        own_id_mentioned_;
 342         /**    Peer's progress, based on acknowledgements. */
 343         binmap_t        ack_in_;
 344         /**    Last data received; needs to be acked immediately. */
 345         tintbin     data_in_;
 346         bin64_t     data_in_dbl_;
 347         /** The history of data sent and still unacknowledged. */
 348         tbqueue     data_out_;
 349         /** Timeouted data (potentially to be retransmitted). */
 350         tbqueue     data_out_tmo_;
 351         bin64_t     data_out_cap_;
 352         /** Index in the history array. */
 353         binmap_t        have_out_;
 354         /**    Transmit schedule: in most cases filled with the peer's hints */
 355         tbqueue     hint_in_;
 356         /** Hints sent (to detect and reschedule ignored hints). */
 357         tbqueue     hint_out_;
 358         uint64_t    hint_out_size_;
 359         /** Types of messages the peer accepts. */
 360         uint64_t    cap_in_;
 361         /** For repeats. */
 362         //tint        last_send_time, last_recv_time;
 363         /** PEX progress */
 364         int         pex_out_;
 365         /** Smoothed averages for RTT, RTT deviation and data interarrival periods. */
 366         tint        rtt_avg_, dev_avg_, dip_avg_;
 367         tint        last_send_time_;
 368         tint        last_recv_time_;
 369         tint        last_data_out_time_;
 370         tint        last_data_in_time_;
 371         tint        last_loss_time_;
 372         tint        next_send_time_;
 373         /** Congestion window; TODO: int, bytes. */
 374         float       cwnd_;
 375         /** Data sending interval. */
 376         tint        send_interval_;
 377         /** The congestion control strategy. */
 378         int         send_control_;
 379         /** Datagrams (not data) sent since last recv.    */
 380         int         sent_since_recv_;
 381         /** Recent acknowlegements for data previously sent.    */
 382         int         ack_rcvd_recent_;
 383         /** Recent non-acknowlegements (losses) of data previously sent.    */
 384         int         ack_not_rcvd_recent_;
 385         /** LEDBAT one-way delay machinery */
 386         tint        owd_min_bins_[4];
 387         int         owd_min_bin_;
 388         tint        owd_min_bin_start_;
 389         tint        owd_current_[4];
 390         int         owd_cur_bin_;
 391         /** Stats */
 392         int         dgrams_sent_;
 393         int         dgrams_rcvd_;
 394
 395         int         PeerBPS() const {
 396             return TINT_SEC / dip_avg_ * 1024;
 397         }
 398         /** Get a request for one packet from the queue of peer's requests. */
 399         bin64_t     DequeueHint();
 400         bin64_t     ImposeHint();
 401         void        TimeoutDataOut ();
 402         void        CleanStaleHintOut();
 403         void        CleanHintOut(bin64_t pos);
 404         void        Reschedule();
 405
 406         static PeerSelector* peer_selector;
 407
 408         static tint     last_tick;
 409         static tbheap   send_queue;
 410
 411         static Address  tracker;
 412         static std::vector<Channel*> channels;
 413
 414         friend int      Listen (Address addr);
 415         friend void     Shutdown (int sock_des);
 416         friend void     AddPeer (Address address, const Sha1Hash& root);
 417         friend void     SetTracker(const Address& tracker);
 418         friend int      Open (const char*, const Sha1Hash&) ; // FIXME
 419
 420     };
 421
 422
 423
 424     /*************** The top-level API ****************/
 425     /** Start listening a port. Returns socket descriptor. */
 426     int     Listen (Address addr);
 427     /** Run send/receive loop for the specified amount of time. */
 428     void    Loop (tint till);
 429     bool    Listen3rdPartySocket (sckrwecb_t);
 430     /** Stop listening to a port. */
 431     void    Shutdown (int sock_des=-1);
 432
 433     /** Open a file, start a transmission; fill it with content for a given root hash;
 434         in case the hash is omitted, the file is a fresh submit. */
 435     int     Open (const char* filename, const Sha1Hash& hash=Sha1Hash::ZERO) ;
 436     /** Get the root hash for the transmission. */
 437     const Sha1Hash& RootMerkleHash (int file) ;
 438     /** Close a file and a transmission. */
 439     void    Close (int fd) ;
 440     /** Add a possible peer which participares in a given transmission. In the case
 441         root hash is zero, the peer might be talked to regarding any transmission
 442         (likely, a tracker, cache or an archive). */
 443     void    AddPeer (Address address, const Sha1Hash& root=Sha1Hash::ZERO);
 444
 445     void    SetTracker(const Address& tracker);
 446
 447     /** Returns size of the file in bytes, 0 if unknown. Might be rounded up to a kilobyte
 448         before the transmission is complete. */
 449     uint64_t  Size (int fdes);
 450     /** Returns the amount of retrieved and verified data, in bytes.
 451         A 100% complete transmission has Size()==Complete(). */
 452     uint64_t  Complete (int fdes);
 453     bool      IsComplete (int fdes);
 454     /** Returns the number of bytes that are complete sequentially, starting from the
 455         beginning, till the first not-yet-retrieved packet. */
 456     uint64_t  SeqComplete (int fdes);
 457     /***/
 458     int       Find (Sha1Hash hash);
 459
 460     void AddProgressCallback (int transfer,ProgressCallback cb,uint8_t agg);
 461     void RemoveProgressCallback (int transfer,ProgressCallback cb);
 462     void ExternallyRetrieved (int transfer,bin64_t piece);
 463
 464     //uint32_t Width (const tbinvec& v);
 465
 466
 467     /** Must be called by any client using the library */
 468     void LibraryInit(void);
 469
 470
 471 } // namespace end
 472
 473 #ifndef SWIFT_MUTE
 474 #define dprintf(...) { if (Channel::debug_file) fprintf(Channel::debug_file,__VA_ARGS__); }
 475 #else
 476 #define dprintf(...) {}
 477 #endif
 478 #define eprintf(...) fprintf(stderr,__VA_ARGS__)
 479
 480 #endif