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