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