src/connection.h

   1 /*
   2 Minetest
   3 Copyright (C) 2013 celeron55, Perttu Ahola <celeron55@gmail.com>
   4
   5 This program is free software; you can redistribute it and/or modify
   6 it under the terms of the GNU Lesser General Public License as published by
   7 the Free Software Foundation; either version 2.1 of the License, or
   8 (at your option) any later version.
   9
  10 This program is distributed in the hope that it will be useful,
  11 but WITHOUT ANY WARRANTY; without even the implied warranty of
  12 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13 GNU Lesser General Public License for more details.
  14
  15 You should have received a copy of the GNU Lesser General Public License along
  16 with this program; if not, write to the Free Software Foundation, Inc.,
  17 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  18 */
  19
  20 #ifndef CONNECTION_HEADER
  21 #define CONNECTION_HEADER
  22
  23 #include "irrlichttypes_bloated.h"
  24 #include "socket.h"
  25 #include "exceptions.h"
  26 #include "constants.h"
  27 #include "util/pointer.h"
  28 #include "util/container.h"
  29 #include "util/thread.h"
  30 #include "util/numeric.h"
  31 #include <iostream>
  32 #include <fstream>
  33 #include <list>
  34 #include <map>
  35
  36 namespace con
  37 {
  38
  39 /*
  40         Exceptions
  41 */
  42 class NotFoundException : public BaseException
  43 {
  44 public:
  45         NotFoundException(const char *s):
  46                 BaseException(s)
  47         {}
  48 };
  49
  50 class PeerNotFoundException : public BaseException
  51 {
  52 public:
  53         PeerNotFoundException(const char *s):
  54                 BaseException(s)
  55         {}
  56 };
  57
  58 class ConnectionException : public BaseException
  59 {
  60 public:
  61         ConnectionException(const char *s):
  62                 BaseException(s)
  63         {}
  64 };
  65
  66 class ConnectionBindFailed : public BaseException
  67 {
  68 public:
  69         ConnectionBindFailed(const char *s):
  70                 BaseException(s)
  71         {}
  72 };
  73
  74 class InvalidIncomingDataException : public BaseException
  75 {
  76 public:
  77         InvalidIncomingDataException(const char *s):
  78                 BaseException(s)
  79         {}
  80 };
  81
  82 class InvalidOutgoingDataException : public BaseException
  83 {
  84 public:
  85         InvalidOutgoingDataException(const char *s):
  86                 BaseException(s)
  87         {}
  88 };
  89
  90 class NoIncomingDataException : public BaseException
  91 {
  92 public:
  93         NoIncomingDataException(const char *s):
  94                 BaseException(s)
  95         {}
  96 };
  97
  98 class ProcessedSilentlyException : public BaseException
  99 {
 100 public:
 101         ProcessedSilentlyException(const char *s):
 102                 BaseException(s)
 103         {}
 104 };
 105
 106 class ProcessedQueued : public BaseException
 107 {
 108 public:
 109         ProcessedQueued(const char *s):
 110                 BaseException(s)
 111         {}
 112 };
 113
 114 class IncomingDataCorruption : public BaseException
 115 {
 116 public:
 117         IncomingDataCorruption(const char *s):
 118                 BaseException(s)
 119         {}
 120 };
 121
 122 typedef enum MTProtocols {
 123         MTP_PRIMARY,
 124         MTP_UDP,
 125         MTP_MINETEST_RELIABLE_UDP
 126 } MTProtocols;
 127
 128 #define SEQNUM_MAX 65535
 129 inline bool seqnum_higher(u16 totest, u16 base)
 130 {
 131         if (totest > base)
 132         {
 133                 if((totest - base) > (SEQNUM_MAX/2))
 134                         return false;
 135                 else
 136                         return true;
 137         }
 138         else
 139         {
 140                 if((base - totest) > (SEQNUM_MAX/2))
 141                         return true;
 142                 else
 143                         return false;
 144         }
 145 }
 146
 147 inline bool seqnum_in_window(u16 seqnum, u16 next,u16 window_size)
 148 {
 149         u16 window_start = next;
 150         u16 window_end   = ( next + window_size ) % (SEQNUM_MAX+1);
 151
 152         if (window_start < window_end)
 153         {
 154                 return ((seqnum >= window_start) && (seqnum < window_end));
 155         }
 156         else
 157         {
 158                 return ((seqnum < window_end) || (seqnum >= window_start));
 159         }
 160 }
 161
 162 struct BufferedPacket
 163 {
 164         BufferedPacket(u8 *a_data, u32 a_size):
 165                 data(a_data, a_size), time(0.0), totaltime(0.0), absolute_send_time(-1)
 166         {}
 167         BufferedPacket(u32 a_size):
 168                 data(a_size), time(0.0), totaltime(0.0), absolute_send_time(-1)
 169         {}
 170         SharedBuffer<u8> data; // Data of the packet, including headers
 171         float time; // Seconds from buffering the packet or re-sending
 172         float totaltime; // Seconds from buffering the packet
 173         unsigned int absolute_send_time;
 174         Address address; // Sender or destination
 175 };
 176
 177 // This adds the base headers to the data and makes a packet out of it
 178 BufferedPacket makePacket(Address &address, u8 *data, u32 datasize,
 179                 u32 protocol_id, u16 sender_peer_id, u8 channel);
 180 BufferedPacket makePacket(Address &address, SharedBuffer<u8> &data,
 181                 u32 protocol_id, u16 sender_peer_id, u8 channel);
 182
 183 // Add the TYPE_ORIGINAL header to the data
 184 SharedBuffer<u8> makeOriginalPacket(
 185                 SharedBuffer<u8> data);
 186
 187 // Split data in chunks and add TYPE_SPLIT headers to them
 188 std::list<SharedBuffer<u8> > makeSplitPacket(
 189                 SharedBuffer<u8> data,
 190                 u32 chunksize_max,
 191                 u16 seqnum);
 192
 193 // Depending on size, make a TYPE_ORIGINAL or TYPE_SPLIT packet
 194 // Increments split_seqnum if a split packet is made
 195 std::list<SharedBuffer<u8> > makeAutoSplitPacket(
 196                 SharedBuffer<u8> data,
 197                 u32 chunksize_max,
 198                 u16 &split_seqnum);
 199
 200 // Add the TYPE_RELIABLE header to the data
 201 SharedBuffer<u8> makeReliablePacket(
 202                 SharedBuffer<u8> data,
 203                 u16 seqnum);
 204
 205 struct IncomingSplitPacket
 206 {
 207         IncomingSplitPacket()
 208         {
 209                 time = 0.0;
 210                 reliable = false;
 211         }
 212         // Key is chunk number, value is data without headers
 213         std::map<u16, SharedBuffer<u8> > chunks;
 214         u32 chunk_count;
 215         float time; // Seconds from adding
 216         bool reliable; // If true, isn't deleted on timeout
 217
 218         bool allReceived()
 219         {
 220                 return (chunks.size() == chunk_count);
 221         }
 222 };
 223
 224 /*
 225 === NOTES ===
 226
 227 A packet is sent through a channel to a peer with a basic header:
 228 TODO: Should we have a receiver_peer_id also?
 229         Header (7 bytes):
 230         [0] u32 protocol_id
 231         [4] u16 sender_peer_id
 232         [6] u8 channel
 233 sender_peer_id:
 234         Unique to each peer.
 235         value 0 (PEER_ID_INEXISTENT) is reserved for making new connections
 236         value 1 (PEER_ID_SERVER) is reserved for server
 237         these constants are defined in constants.h
 238 channel:
 239         The lower the number, the higher the priority is.
 240         Only channels 0, 1 and 2 exist.
 241 */
 242 #define BASE_HEADER_SIZE 7
 243 #define CHANNEL_COUNT 3
 244 /*
 245 Packet types:
 246
 247 CONTROL: This is a packet used by the protocol.
 248 - When this is processed, nothing is handed to the user.
 249         Header (2 byte):
 250         [0] u8 type
 251         [1] u8 controltype
 252 controltype and data description:
 253         CONTROLTYPE_ACK
 254                 [2] u16 seqnum
 255         CONTROLTYPE_SET_PEER_ID
 256                 [2] u16 peer_id_new
 257         CONTROLTYPE_PING
 258         - There is no actual reply, but this can be sent in a reliable
 259           packet to get a reply
 260         CONTROLTYPE_DISCO
 261 */
 262 #define TYPE_CONTROL 0
 263 #define CONTROLTYPE_ACK 0
 264 #define CONTROLTYPE_SET_PEER_ID 1
 265 #define CONTROLTYPE_PING 2
 266 #define CONTROLTYPE_DISCO 3
 267 #define CONTROLTYPE_ENABLE_BIG_SEND_WINDOW 4
 268
 269 /*
 270 ORIGINAL: This is a plain packet with no control and no error
 271 checking at all.
 272 - When this is processed, it is directly handed to the user.
 273         Header (1 byte):
 274         [0] u8 type
 275 */
 276 #define TYPE_ORIGINAL 1
 277 #define ORIGINAL_HEADER_SIZE 1
 278 /*
 279 SPLIT: These are sequences of packets forming one bigger piece of
 280 data.
 281 - When processed and all the packet_nums 0...packet_count-1 are
 282   present (this should be buffered), the resulting data shall be
 283   directly handed to the user.
 284 - If the data fails to come up in a reasonable time, the buffer shall
 285   be silently discarded.
 286 - These can be sent as-is or atop of a RELIABLE packet stream.
 287         Header (7 bytes):
 288         [0] u8 type
 289         [1] u16 seqnum
 290         [3] u16 chunk_count
 291         [5] u16 chunk_num
 292 */
 293 #define TYPE_SPLIT 2
 294 /*
 295 RELIABLE: Delivery of all RELIABLE packets shall be forced by ACKs,
 296 and they shall be delivered in the same order as sent. This is done
 297 with a buffer in the receiving and transmitting end.
 298 - When this is processed, the contents of each packet is recursively
 299   processed as packets.
 300         Header (3 bytes):
 301         [0] u8 type
 302         [1] u16 seqnum
 303
 304 */
 305 #define TYPE_RELIABLE 3
 306 #define RELIABLE_HEADER_SIZE 3
 307 #define SEQNUM_INITIAL 65500
 308
 309 /*
 310         A buffer which stores reliable packets and sorts them internally
 311         for fast access to the smallest one.
 312 */
 313
 314 typedef std::list<BufferedPacket>::iterator RPBSearchResult;
 315
 316 class ReliablePacketBuffer
 317 {
 318 public:
 319         ReliablePacketBuffer();
 320
 321         bool getFirstSeqnum(u16& result);
 322
 323         BufferedPacket popFirst();
 324         BufferedPacket popSeqnum(u16 seqnum);
 325         void insert(BufferedPacket &p,u16 next_expected);
 326
 327         void incrementTimeouts(float dtime);
 328         std::list<BufferedPacket> getTimedOuts(float timeout,
 329                         unsigned int max_packets);
 330
 331         void print();
 332         bool empty();
 333         bool containsPacket(u16 seqnum);
 334         RPBSearchResult notFound();
 335         u32 size();
 336
 337
 338 private:
 339         RPBSearchResult findPacket(u16 seqnum);
 340
 341         std::list<BufferedPacket> m_list;
 342         u32 m_list_size;
 343
 344         u16 m_oldest_non_answered_ack;
 345
 346         JMutex m_list_mutex;
 347 };
 348
 349 /*
 350         A buffer for reconstructing split packets
 351 */
 352
 353 class IncomingSplitBuffer
 354 {
 355 public:
 356         ~IncomingSplitBuffer();
 357         /*
 358                 Returns a reference counted buffer of length != 0 when a full split
 359                 packet is constructed. If not, returns one of length 0.
 360         */
 361         SharedBuffer<u8> insert(BufferedPacket &p, bool reliable);
 362
 363         void removeUnreliableTimedOuts(float dtime, float timeout);
 364
 365 private:
 366         // Key is seqnum
 367         std::map<u16, IncomingSplitPacket*> m_buf;
 368
 369         JMutex m_map_mutex;
 370 };
 371
 372 struct OutgoingPacket
 373 {
 374         u16 peer_id;
 375         u8 channelnum;
 376         SharedBuffer<u8> data;
 377         bool reliable;
 378         bool ack;
 379
 380         OutgoingPacket(u16 peer_id_, u8 channelnum_, SharedBuffer<u8> data_,
 381                         bool reliable_,bool ack_=false):
 382                 peer_id(peer_id_),
 383                 channelnum(channelnum_),
 384                 data(data_),
 385                 reliable(reliable_),
 386                 ack(ack_)
 387         {
 388         }
 389 };
 390
 391 enum ConnectionCommandType{
 392         CONNCMD_NONE,
 393         CONNCMD_SERVE,
 394         CONNCMD_CONNECT,
 395         CONNCMD_DISCONNECT,
 396         CONNCMD_DISCONNECT_PEER,
 397         CONNCMD_SEND,
 398         CONNCMD_SEND_TO_ALL,
 399         CONCMD_ACK,
 400         CONCMD_CREATE_PEER,
 401         CONCMD_DISABLE_LEGACY
 402 };
 403
 404 struct ConnectionCommand
 405 {
 406         enum ConnectionCommandType type;
 407         Address address;
 408         u16 peer_id;
 409         u8 channelnum;
 410         Buffer<u8> data;
 411         bool reliable;
 412         bool raw;
 413
 414         ConnectionCommand(): type(CONNCMD_NONE), peer_id(PEER_ID_INEXISTENT), reliable(false), raw(false) {}
 415
 416         void serve(Address address_)
 417         {
 418                 type = CONNCMD_SERVE;
 419                 address = address_;
 420         }
 421         void connect(Address address_)
 422         {
 423                 type = CONNCMD_CONNECT;
 424                 address = address_;
 425         }
 426         void disconnect()
 427         {
 428                 type = CONNCMD_DISCONNECT;
 429         }
 430         void disconnect_peer(u16 peer_id_)
 431         {
 432                 type = CONNCMD_DISCONNECT_PEER;
 433                 peer_id = peer_id_;
 434         }
 435         void send(u16 peer_id_, u8 channelnum_,
 436                         SharedBuffer<u8> data_, bool reliable_)
 437         {
 438                 type = CONNCMD_SEND;
 439                 peer_id = peer_id_;
 440                 channelnum = channelnum_;
 441                 data = data_;
 442                 reliable = reliable_;
 443         }
 444         void sendToAll(u8 channelnum_, SharedBuffer<u8> data_, bool reliable_)
 445         {
 446                 type = CONNCMD_SEND_TO_ALL;
 447                 channelnum = channelnum_;
 448                 data = data_;
 449                 reliable = reliable_;
 450         }
 451
 452         void ack(u16 peer_id_, u8 channelnum_, SharedBuffer<u8> data_)
 453         {
 454                 type = CONCMD_ACK;
 455                 peer_id = peer_id_;
 456                 channelnum = channelnum_;
 457                 data = data_;
 458                 reliable = false;
 459         }
 460
 461         void createPeer(u16 peer_id_, SharedBuffer<u8> data_)
 462         {
 463                 type = CONCMD_CREATE_PEER;
 464                 peer_id = peer_id_;
 465                 data = data_;
 466                 channelnum = 0;
 467                 reliable = true;
 468                 raw = true;
 469         }
 470
 471         void disableLegacy(u16 peer_id_, SharedBuffer<u8> data_)
 472         {
 473                 type = CONCMD_DISABLE_LEGACY;
 474                 peer_id = peer_id_;
 475                 data = data_;
 476                 channelnum = 0;
 477                 reliable = true;
 478                 raw = true;
 479         }
 480 };
 481
 482 class Channel
 483 {
 484
 485 public:
 486         u16 readNextIncomingSeqNum();
 487         u16 incNextIncomingSeqNum();
 488
 489         u16 getOutgoingSequenceNumber(bool& successfull);
 490         u16 readOutgoingSequenceNumber();
 491         bool putBackSequenceNumber(u16);
 492
 493         u16 readNextSplitSeqNum();
 494         void setNextSplitSeqNum(u16 seqnum);
 495
 496         // This is for buffering the incoming packets that are coming in
 497         // the wrong order
 498         ReliablePacketBuffer incoming_reliables;
 499         // This is for buffering the sent packets so that the sender can
 500         // re-send them if no ACK is received
 501         ReliablePacketBuffer outgoing_reliables_sent;
 502
 503         //queued reliable packets
 504         Queue<BufferedPacket> queued_reliables;
 505
 506         //queue commands prior splitting to packets
 507         Queue<ConnectionCommand> queued_commands;
 508
 509         IncomingSplitBuffer incoming_splits;
 510
 511         Channel();
 512         ~Channel();
 513
 514         void UpdatePacketLossCounter(unsigned int count);
 515         void UpdatePacketTooLateCounter();
 516         void UpdateBytesSent(unsigned int bytes,unsigned int packages=1);
 517         void UpdateBytesLost(unsigned int bytes);
 518         void UpdateBytesReceived(unsigned int bytes);
 519
 520         void UpdateTimers(float dtime, bool legacy_peer);
 521
 522         const float getCurrentDownloadRateKB()
 523                 { JMutexAutoLock lock(m_internal_mutex); return cur_kbps; };
 524         const float getMaxDownloadRateKB()
 525                 { JMutexAutoLock lock(m_internal_mutex); return max_kbps; };
 526
 527         const float getCurrentLossRateKB()
 528                 { JMutexAutoLock lock(m_internal_mutex); return cur_kbps_lost; };
 529         const float getMaxLossRateKB()
 530                 { JMutexAutoLock lock(m_internal_mutex); return max_kbps_lost; };
 531
 532         const float getCurrentIncomingRateKB()
 533                 { JMutexAutoLock lock(m_internal_mutex); return cur_incoming_kbps; };
 534         const float getMaxIncomingRateKB()
 535                 { JMutexAutoLock lock(m_internal_mutex); return max_incoming_kbps; };
 536
 537         const float getAvgDownloadRateKB()
 538                 { JMutexAutoLock lock(m_internal_mutex); return avg_kbps; };
 539         const float getAvgLossRateKB()
 540                 { JMutexAutoLock lock(m_internal_mutex); return avg_kbps_lost; };
 541         const float getAvgIncomingRateKB()
 542                 { JMutexAutoLock lock(m_internal_mutex); return avg_incoming_kbps; };
 543
 544         const unsigned int getWindowSize() const { return window_size; };
 545
 546         void setWindowSize(unsigned int size) { window_size = size; };
 547 private:
 548         JMutex m_internal_mutex;
 549         int window_size;
 550
 551         u16 next_incoming_seqnum;
 552
 553         u16 next_outgoing_seqnum;
 554         u16 next_outgoing_split_seqnum;
 555
 556         unsigned int current_packet_loss;
 557         unsigned int current_packet_too_late;
 558         unsigned int current_packet_successfull;
 559         float packet_loss_counter;
 560
 561         unsigned int current_bytes_transfered;
 562         unsigned int current_bytes_received;
 563         unsigned int current_bytes_lost;
 564         float max_kbps;
 565         float cur_kbps;
 566         float avg_kbps;
 567         float max_incoming_kbps;
 568         float cur_incoming_kbps;
 569         float avg_incoming_kbps;
 570         float max_kbps_lost;
 571         float cur_kbps_lost;
 572         float avg_kbps_lost;
 573         float bpm_counter;
 574
 575         unsigned int rate_samples;
 576 };
 577
 578 class Peer;
 579
 580 enum PeerChangeType
 581 {
 582         PEER_ADDED,
 583         PEER_REMOVED
 584 };
 585 struct PeerChange
 586 {
 587         PeerChangeType type;
 588         u16 peer_id;
 589         bool timeout;
 590 };
 591
 592 class PeerHandler
 593 {
 594 public:
 595
 596         PeerHandler()
 597         {
 598         }
 599         virtual ~PeerHandler()
 600         {
 601         }
 602
 603         /*
 604                 This is called after the Peer has been inserted into the
 605                 Connection's peer container.
 606         */
 607         virtual void peerAdded(Peer *peer) = 0;
 608         /*
 609                 This is called before the Peer has been removed from the
 610                 Connection's peer container.
 611         */
 612         virtual void deletingPeer(Peer *peer, bool timeout) = 0;
 613 };
 614
 615 class PeerHelper
 616 {
 617 public:
 618         PeerHelper();
 619         PeerHelper(Peer* peer);
 620         ~PeerHelper();
 621
 622         PeerHelper&   operator=(Peer* peer);
 623         Peer*         operator->() const;
 624         bool          operator!();
 625         Peer*         operator&() const;
 626         bool          operator!=(void* ptr);
 627
 628 private:
 629         Peer* m_peer;
 630 };
 631
 632 class Connection;
 633
 634 typedef enum {
 635         MIN_RTT,
 636         MAX_RTT,
 637         AVG_RTT,
 638         MIN_JITTER,
 639         MAX_JITTER,
 640         AVG_JITTER
 641 } rtt_stat_type;
 642
 643 typedef enum {
 644         CUR_DL_RATE,
 645         AVG_DL_RATE,
 646         CUR_INC_RATE,
 647         AVG_INC_RATE,
 648         CUR_LOSS_RATE,
 649         AVG_LOSS_RATE,
 650 } rate_stat_type;
 651
 652 class Peer {
 653         public:
 654                 friend class PeerHelper;
 655
 656                 Peer(Address address_,u16 id_,Connection* connection) :
 657                         id(id_),
 658                         m_increment_packets_remaining(9),
 659                         m_increment_bytes_remaining(0),
 660                         m_pending_deletion(false),
 661                         m_connection(connection),
 662                         address(address_),
 663                         m_ping_timer(0.0),
 664                         m_last_rtt(-1.0),
 665                         m_usage(0),
 666                         m_timeout_counter(0.0),
 667                         m_last_timeout_check(porting::getTimeMs()),
 668                         m_has_sent_with_id(false)
 669                 {
 670                         m_rtt.avg_rtt = -1.0;
 671                         m_rtt.jitter_avg = -1.0;
 672                         m_rtt.jitter_max = 0.0;
 673                         m_rtt.max_rtt = 0.0;
 674                         m_rtt.jitter_min = FLT_MAX;
 675                         m_rtt.min_rtt = FLT_MAX;
 676                 };
 677
 678                 virtual ~Peer() {
 679                         JMutexAutoLock usage_lock(m_exclusive_access_mutex);
 680                         assert(m_usage == 0);
 681                 };
 682
 683                 // Unique id of the peer
 684                 u16 id;
 685
 686                 void Drop();
 687
 688                 virtual void PutReliableSendCommand(ConnectionCommand &c,
 689                                                 unsigned int max_packet_size) {};
 690
 691                 virtual bool isActive() { return false; };
 692
 693                 virtual bool getAddress(MTProtocols type, Address& toset) = 0;
 694
 695                 void ResetTimeout()
 696                         {JMutexAutoLock lock(m_exclusive_access_mutex); m_timeout_counter=0.0; };
 697
 698                 bool isTimedOut(float timeout);
 699
 700                 void setSentWithID()
 701                 { JMutexAutoLock lock(m_exclusive_access_mutex); m_has_sent_with_id = true; };
 702
 703                 bool hasSentWithID()
 704                 { JMutexAutoLock lock(m_exclusive_access_mutex); return m_has_sent_with_id; };
 705
 706                 unsigned int m_increment_packets_remaining;
 707                 unsigned int m_increment_bytes_remaining;
 708
 709                 virtual u16 getNextSplitSequenceNumber(u8 channel) { return 0; };
 710                 virtual void setNextSplitSequenceNumber(u8 channel, u16 seqnum) {};
 711                 virtual SharedBuffer<u8> addSpiltPacket(u8 channel,
 712                                                                                                 BufferedPacket toadd,
 713                                                                                                 bool reliable)
 714                                 {
 715                                         fprintf(stderr,"Peer: addSplitPacket called, this is supposed to be never called!\n");
 716                                         return SharedBuffer<u8>(0);
 717                                 };
 718
 719                 virtual bool Ping(float dtime, SharedBuffer<u8>& data) { return false; };
 720
 721                 virtual float getStat(rtt_stat_type type) const {
 722                         switch (type) {
 723                                 case MIN_RTT:
 724                                         return m_rtt.min_rtt;
 725                                 case MAX_RTT:
 726                                         return m_rtt.max_rtt;
 727                                 case AVG_RTT:
 728                                         return m_rtt.avg_rtt;
 729                                 case MIN_JITTER:
 730                                         return m_rtt.jitter_min;
 731                                 case MAX_JITTER:
 732                                         return m_rtt.jitter_max;
 733                                 case AVG_JITTER:
 734                                         return m_rtt.jitter_avg;
 735                         }
 736                         return -1;
 737                 }
 738         protected:
 739                 virtual void reportRTT(float rtt) {};
 740
 741                 void RTTStatistics(float rtt,
 742                                                         std::string profiler_id="",
 743                                                         unsigned int num_samples=1000);
 744
 745                 bool IncUseCount();
 746                 void DecUseCount();
 747
 748                 JMutex m_exclusive_access_mutex;
 749
 750                 bool m_pending_deletion;
 751
 752                 Connection* m_connection;
 753
 754                 // Address of the peer
 755                 Address address;
 756
 757                 // Ping timer
 758                 float m_ping_timer;
 759         private:
 760
 761                 struct rttstats {
 762                         float jitter_min;
 763                         float jitter_max;
 764                         float jitter_avg;
 765                         float min_rtt;
 766                         float max_rtt;
 767                         float avg_rtt;
 768                 };
 769
 770                 rttstats m_rtt;
 771                 float    m_last_rtt;
 772
 773                 // current usage count
 774                 unsigned int m_usage;
 775
 776                 // Seconds from last receive
 777                 float m_timeout_counter;
 778
 779                 u32 m_last_timeout_check;
 780
 781                 bool m_has_sent_with_id;
 782 };
 783
 784 class UDPPeer : public Peer
 785 {
 786 public:
 787
 788         friend class PeerHelper;
 789         friend class ConnectionReceiveThread;
 790         friend class ConnectionSendThread;
 791         friend class Connection;
 792
 793         UDPPeer(u16 a_id, Address a_address, Connection* connection);
 794         virtual ~UDPPeer() {};
 795
 796         void PutReliableSendCommand(ConnectionCommand &c,
 797                                                         unsigned int max_packet_size);
 798
 799         bool isActive()
 800         { return ((hasSentWithID()) && (!m_pending_deletion)); };
 801
 802         bool getAddress(MTProtocols type, Address& toset);
 803
 804         void setNonLegacyPeer();
 805
 806         bool getLegacyPeer()
 807         { return m_legacy_peer; }
 808
 809         u16 getNextSplitSequenceNumber(u8 channel);
 810         void setNextSplitSequenceNumber(u8 channel, u16 seqnum);
 811
 812         SharedBuffer<u8> addSpiltPacket(u8 channel,
 813                                                                         BufferedPacket toadd,
 814                                                                         bool reliable);
 815
 816
 817 protected:
 818         /*
 819                 Calculates avg_rtt and resend_timeout.
 820                 rtt=-1 only recalculates resend_timeout
 821         */
 822         void reportRTT(float rtt);
 823
 824         void RunCommandQueues(
 825                                         unsigned int max_packet_size,
 826                                         unsigned int maxcommands,
 827                                         unsigned int maxtransfer);
 828
 829         float getResendTimeout()
 830                 { JMutexAutoLock lock(m_exclusive_access_mutex); return resend_timeout; }
 831
 832         void setResendTimeout(float timeout)
 833                 { JMutexAutoLock lock(m_exclusive_access_mutex); resend_timeout = timeout; }
 834         bool Ping(float dtime,SharedBuffer<u8>& data);
 835
 836         Channel channels[CHANNEL_COUNT];
 837         bool m_pending_disconnect;
 838 private:
 839         // This is changed dynamically
 840         float resend_timeout;
 841
 842         bool processReliableSendCommand(
 843                                         ConnectionCommand &c,
 844                                         unsigned int max_packet_size);
 845
 846         bool m_legacy_peer;
 847 };
 848
 849 /*
 850         Connection
 851 */
 852
 853 enum ConnectionEventType{
 854         CONNEVENT_NONE,
 855         CONNEVENT_DATA_RECEIVED,
 856         CONNEVENT_PEER_ADDED,
 857         CONNEVENT_PEER_REMOVED,
 858         CONNEVENT_BIND_FAILED,
 859 };
 860
 861 struct ConnectionEvent
 862 {
 863         enum ConnectionEventType type;
 864         u16 peer_id;
 865         Buffer<u8> data;
 866         bool timeout;
 867         Address address;
 868
 869         ConnectionEvent(): type(CONNEVENT_NONE) {}
 870
 871         std::string describe()
 872         {
 873                 switch(type){
 874                 case CONNEVENT_NONE:
 875                         return "CONNEVENT_NONE";
 876                 case CONNEVENT_DATA_RECEIVED:
 877                         return "CONNEVENT_DATA_RECEIVED";
 878                 case CONNEVENT_PEER_ADDED:
 879                         return "CONNEVENT_PEER_ADDED";
 880                 case CONNEVENT_PEER_REMOVED:
 881                         return "CONNEVENT_PEER_REMOVED";
 882                 case CONNEVENT_BIND_FAILED:
 883                         return "CONNEVENT_BIND_FAILED";
 884                 }
 885                 return "Invalid ConnectionEvent";
 886         }
 887
 888         void dataReceived(u16 peer_id_, SharedBuffer<u8> data_)
 889         {
 890                 type = CONNEVENT_DATA_RECEIVED;
 891                 peer_id = peer_id_;
 892                 data = data_;
 893         }
 894         void peerAdded(u16 peer_id_, Address address_)
 895         {
 896                 type = CONNEVENT_PEER_ADDED;
 897                 peer_id = peer_id_;
 898                 address = address_;
 899         }
 900         void peerRemoved(u16 peer_id_, bool timeout_, Address address_)
 901         {
 902                 type = CONNEVENT_PEER_REMOVED;
 903                 peer_id = peer_id_;
 904                 timeout = timeout_;
 905                 address = address_;
 906         }
 907         void bindFailed()
 908         {
 909                 type = CONNEVENT_BIND_FAILED;
 910         }
 911 };
 912
 913 class ConnectionSendThread : public JThread {
 914
 915 public:
 916         friend class UDPPeer;
 917
 918         ConnectionSendThread(Connection* parent,
 919                                                         unsigned int max_packet_size, float timeout);
 920
 921         void * Thread       ();
 922
 923         void Trigger();
 924
 925         void setPeerTimeout(float peer_timeout)
 926                 { m_timeout = peer_timeout; }
 927
 928 private:
 929         void runTimeouts    (float dtime);
 930         void rawSend        (const BufferedPacket &packet);
 931         bool rawSendAsPacket(u16 peer_id, u8 channelnum,
 932                                                         SharedBuffer<u8> data, bool reliable);
 933
 934         void processReliableCommand (ConnectionCommand &c);
 935         void processNonReliableCommand (ConnectionCommand &c);
 936         void serve          (Address bind_address);
 937         void connect        (Address address);
 938         void disconnect     ();
 939         void disconnect_peer(u16 peer_id);
 940         void send           (u16 peer_id, u8 channelnum,
 941                                                         SharedBuffer<u8> data);
 942         void sendReliable   (ConnectionCommand &c);
 943         void sendToAll      (u8 channelnum,
 944                                                         SharedBuffer<u8> data);
 945         void sendToAllReliable(ConnectionCommand &c);
 946
 947         void sendPackets    (float dtime);
 948
 949         void sendAsPacket   (u16 peer_id, u8 channelnum,
 950                                                         SharedBuffer<u8> data,bool ack=false);
 951
 952         void sendAsPacketReliable(BufferedPacket& p, Channel* channel);
 953
 954         bool packetsQueued();
 955
 956         Connection*           m_connection;
 957         unsigned int          m_max_packet_size;
 958         float                 m_timeout;
 959         Queue<OutgoingPacket> m_outgoing_queue;
 960         JSemaphore            m_send_sleep_semaphore;
 961
 962         unsigned int          m_iteration_packets_avaialble;
 963         unsigned int          m_max_commands_per_iteration;
 964         unsigned int          m_max_data_packets_per_iteration;
 965         unsigned int          m_max_packets_requeued;
 966 };
 967
 968 class ConnectionReceiveThread : public JThread {
 969 public:
 970         ConnectionReceiveThread(Connection* parent,
 971                                                         unsigned int max_packet_size);
 972
 973         void * Thread       ();
 974
 975 private:
 976         void receive        ();
 977
 978         // Returns next data from a buffer if possible
 979         // If found, returns true; if not, false.
 980         // If found, sets peer_id and dst
 981         bool getFromBuffers (u16 &peer_id, SharedBuffer<u8> &dst);
 982
 983         bool checkIncomingBuffers(Channel *channel, u16 &peer_id,
 984                                                         SharedBuffer<u8> &dst);
 985
 986         /*
 987                 Processes a packet with the basic header stripped out.
 988                 Parameters:
 989                         packetdata: Data in packet (with no base headers)
 990                         peer_id: peer id of the sender of the packet in question
 991                         channelnum: channel on which the packet was sent
 992                         reliable: true if recursing into a reliable packet
 993         */
 994         SharedBuffer<u8> processPacket(Channel *channel,
 995                                                         SharedBuffer<u8> packetdata, u16 peer_id,
 996                                                         u8 channelnum, bool reliable);
 997
 998
 999         Connection*           m_connection;
1000 };
1001
1002 class Connection
1003 {
1004 public:
1005         friend class ConnectionSendThread;
1006         friend class ConnectionReceiveThread;
1007
1008         Connection(u32 protocol_id, u32 max_packet_size, float timeout, bool ipv6);
1009         Connection(u32 protocol_id, u32 max_packet_size, float timeout, bool ipv6,
1010                         PeerHandler *peerhandler);
1011         ~Connection();
1012
1013         /* Interface */
1014         ConnectionEvent getEvent();
1015         ConnectionEvent waitEvent(u32 timeout_ms);
1016         void putCommand(ConnectionCommand &c);
1017
1018         void SetTimeoutMs(int timeout){ m_bc_receive_timeout = timeout; }
1019         void Serve(Address bind_addr);
1020         void Connect(Address address);
1021         bool Connected();
1022         void Disconnect();
1023         u32 Receive(u16 &peer_id, SharedBuffer<u8> &data);
1024         void SendToAll(u8 channelnum, SharedBuffer<u8> data, bool reliable);
1025         void Send(u16 peer_id, u8 channelnum, SharedBuffer<u8> data, bool reliable);
1026         u16 GetPeerID(){ return m_peer_id; }
1027         Address GetPeerAddress(u16 peer_id);
1028         float getPeerStat(u16 peer_id, rtt_stat_type type);
1029         float getLocalStat(rate_stat_type type);
1030         const u32 GetProtocolID() const { return m_protocol_id; };
1031         const std::string getDesc();
1032         void DisconnectPeer(u16 peer_id);
1033
1034 protected:
1035         PeerHelper getPeer(u16 peer_id);
1036         PeerHelper getPeerNoEx(u16 peer_id);
1037         u16   lookupPeer(Address& sender);
1038
1039         u16 createPeer(Address& sender, MTProtocols protocol, int fd);
1040         UDPPeer*  createServerPeer(Address& sender);
1041         bool deletePeer(u16 peer_id, bool timeout);
1042
1043         void SetPeerID(u16 id){ m_peer_id = id; }
1044
1045         void sendAck(u16 peer_id, u8 channelnum, u16 seqnum);
1046
1047         void PrintInfo(std::ostream &out);
1048         void PrintInfo();
1049
1050         std::list<u16> getPeerIDs();
1051
1052         UDPSocket m_udpSocket;
1053         MutexedQueue<ConnectionCommand> m_command_queue;
1054
1055         void putEvent(ConnectionEvent &e);
1056
1057         void TriggerSend()
1058                 { m_sendThread.Trigger(); }
1059 private:
1060         std::list<Peer*> getPeers();
1061
1062         MutexedQueue<ConnectionEvent> m_event_queue;
1063
1064         u16 m_peer_id;
1065         u32 m_protocol_id;
1066
1067         std::map<u16, Peer*> m_peers;
1068         JMutex m_peers_mutex;
1069
1070         ConnectionSendThread m_sendThread;
1071         ConnectionReceiveThread m_receiveThread;
1072
1073         JMutex m_info_mutex;
1074
1075         // Backwards compatibility
1076         PeerHandler *m_bc_peerhandler;
1077         int m_bc_receive_timeout;
1078
1079         bool m_shutting_down;
1080
1081         u16 m_next_remote_peer_id;
1082 };
1083
1084 } // namespace
1085
1086 #endif
1087