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 #pragma once
  21
  22 #include "irrlichttypes.h"
  23 #include "peerhandler.h"
  24 #include "socket.h"
  25 #include "constants.h"
  26 #include "util/pointer.h"
  27 #include "util/container.h"
  28 #include "util/thread.h"
  29 #include "util/numeric.h"
  30 #include "networkprotocol.h"
  31 #include <iostream>
  32 #include <vector>
  33 #include <map>
  34
  35 #define MAX_UDP_PEERS 65535
  36
  37 /*
  38 === NOTES ===
  39
  40 A packet is sent through a channel to a peer with a basic header:
  41         Header (7 bytes):
  42         [0] u32 protocol_id
  43         [4] session_t sender_peer_id
  44         [6] u8 channel
  45 sender_peer_id:
  46         Unique to each peer.
  47         value 0 (PEER_ID_INEXISTENT) is reserved for making new connections
  48         value 1 (PEER_ID_SERVER) is reserved for server
  49         these constants are defined in constants.h
  50 channel:
  51         Channel numbers have no intrinsic meaning. Currently only 0, 1, 2 exist.
  52 */
  53 #define BASE_HEADER_SIZE 7
  54 #define CHANNEL_COUNT 3
  55
  56 /*
  57 Packet types:
  58
  59 CONTROL: This is a packet used by the protocol.
  60 - When this is processed, nothing is handed to the user.
  61         Header (2 byte):
  62         [0] u8 type
  63         [1] u8 controltype
  64 controltype and data description:
  65         CONTROLTYPE_ACK
  66                 [2] u16 seqnum
  67         CONTROLTYPE_SET_PEER_ID
  68                 [2] session_t peer_id_new
  69         CONTROLTYPE_PING
  70         - There is no actual reply, but this can be sent in a reliable
  71           packet to get a reply
  72         CONTROLTYPE_DISCO
  73 */
  74 enum ControlType : u8 {
  75         CONTROLTYPE_ACK = 0,
  76         CONTROLTYPE_SET_PEER_ID = 1,
  77         CONTROLTYPE_PING = 2,
  78         CONTROLTYPE_DISCO = 3,
  79 };
  80
  81 /*
  82 ORIGINAL: This is a plain packet with no control and no error
  83 checking at all.
  84 - When this is processed, it is directly handed to the user.
  85         Header (1 byte):
  86         [0] u8 type
  87 */
  88 //#define TYPE_ORIGINAL 1
  89 #define ORIGINAL_HEADER_SIZE 1
  90
  91 /*
  92 SPLIT: These are sequences of packets forming one bigger piece of
  93 data.
  94 - When processed and all the packet_nums 0...packet_count-1 are
  95   present (this should be buffered), the resulting data shall be
  96   directly handed to the user.
  97 - If the data fails to come up in a reasonable time, the buffer shall
  98   be silently discarded.
  99 - These can be sent as-is or atop of a RELIABLE packet stream.
 100         Header (7 bytes):
 101         [0] u8 type
 102         [1] u16 seqnum
 103         [3] u16 chunk_count
 104         [5] u16 chunk_num
 105 */
 106 //#define TYPE_SPLIT 2
 107
 108 /*
 109 RELIABLE: Delivery of all RELIABLE packets shall be forced by ACKs,
 110 and they shall be delivered in the same order as sent. This is done
 111 with a buffer in the receiving and transmitting end.
 112 - When this is processed, the contents of each packet is recursively
 113   processed as packets.
 114         Header (3 bytes):
 115         [0] u8 type
 116         [1] u16 seqnum
 117
 118 */
 119 //#define TYPE_RELIABLE 3
 120 #define RELIABLE_HEADER_SIZE 3
 121 #define SEQNUM_INITIAL 65500
 122 #define SEQNUM_MAX 65535
 123
 124 class NetworkPacket;
 125
 126 namespace con
 127 {
 128
 129 class ConnectionReceiveThread;
 130 class ConnectionSendThread;
 131
 132 typedef enum MTProtocols {
 133         MTP_PRIMARY,
 134         MTP_UDP,
 135         MTP_MINETEST_RELIABLE_UDP
 136 } MTProtocols;
 137
 138 enum PacketType : u8 {
 139         PACKET_TYPE_CONTROL = 0,
 140         PACKET_TYPE_ORIGINAL = 1,
 141         PACKET_TYPE_SPLIT = 2,
 142         PACKET_TYPE_RELIABLE = 3,
 143         PACKET_TYPE_MAX
 144 };
 145
 146 inline bool seqnum_higher(u16 totest, u16 base)
 147 {
 148         if (totest > base)
 149         {
 150                 if ((totest - base) > (SEQNUM_MAX/2))
 151                         return false;
 152
 153                 return true;
 154         }
 155
 156         if ((base - totest) > (SEQNUM_MAX/2))
 157                 return true;
 158
 159         return false;
 160 }
 161
 162 inline bool seqnum_in_window(u16 seqnum, u16 next,u16 window_size)
 163 {
 164         u16 window_start = next;
 165         u16 window_end   = ( next + window_size ) % (SEQNUM_MAX+1);
 166
 167         if (window_start < window_end) {
 168                 return ((seqnum >= window_start) && (seqnum < window_end));
 169         }
 170
 171
 172         return ((seqnum < window_end) || (seqnum >= window_start));
 173 }
 174
 175 static inline float CALC_DTIME(u64 lasttime, u64 curtime)
 176 {
 177         float value = ( curtime - lasttime) / 1000.0;
 178         return MYMAX(MYMIN(value,0.1),0.0);
 179 }
 180
 181 /*
 182         Struct for all kinds of packets. Includes following data:
 183                 BASE_HEADER
 184                 u8[] packet data (usually copied from SharedBuffer<u8>)
 185 */
 186 struct BufferedPacket {
 187         BufferedPacket(u32 a_size)
 188         {
 189                 m_data.resize(a_size);
 190                 data = &m_data[0];
 191         }
 192
 193         DISABLE_CLASS_COPY(BufferedPacket)
 194
 195         u16 getSeqnum() const;
 196
 197         inline size_t size() const { return m_data.size(); }
 198
 199         u8 *data; // Direct memory access
 200         float time = 0.0f; // Seconds from buffering the packet or re-sending
 201         float totaltime = 0.0f; // Seconds from buffering the packet
 202         u64 absolute_send_time = -1;
 203         Address address; // Sender or destination
 204         unsigned int resend_count = 0;
 205
 206 private:
 207         std::vector<u8> m_data; // Data of the packet, including headers
 208 };
 209
 210 typedef std::shared_ptr<BufferedPacket> BufferedPacketPtr;
 211
 212
 213 // This adds the base headers to the data and makes a packet out of it
 214 BufferedPacketPtr makePacket(Address &address, const SharedBuffer<u8> &data,
 215                 u32 protocol_id, session_t sender_peer_id, u8 channel);
 216
 217 // Depending on size, make a TYPE_ORIGINAL or TYPE_SPLIT packet
 218 // Increments split_seqnum if a split packet is made
 219 void makeAutoSplitPacket(const SharedBuffer<u8> &data, u32 chunksize_max,
 220                 u16 &split_seqnum, std::list<SharedBuffer<u8>> *list);
 221
 222 // Add the TYPE_RELIABLE header to the data
 223 SharedBuffer<u8> makeReliablePacket(const SharedBuffer<u8> &data, u16 seqnum);
 224
 225 struct IncomingSplitPacket
 226 {
 227         IncomingSplitPacket(u32 cc, bool r):
 228                 chunk_count(cc), reliable(r) {}
 229
 230         IncomingSplitPacket() = delete;
 231
 232         float time = 0.0f; // Seconds from adding
 233         u32 chunk_count;
 234         bool reliable; // If true, isn't deleted on timeout
 235
 236         bool allReceived() const
 237         {
 238                 return (chunks.size() == chunk_count);
 239         }
 240         bool insert(u32 chunk_num, SharedBuffer<u8> &chunkdata);
 241         SharedBuffer<u8> reassemble();
 242
 243 private:
 244         // Key is chunk number, value is data without headers
 245         std::map<u16, SharedBuffer<u8>> chunks;
 246 };
 247
 248 /*
 249         A buffer which stores reliable packets and sorts them internally
 250         for fast access to the smallest one.
 251 */
 252
 253 typedef std::list<BufferedPacketPtr>::iterator RPBSearchResult;
 254
 255 class ReliablePacketBuffer
 256 {
 257 public:
 258         ReliablePacketBuffer() = default;
 259
 260         bool getFirstSeqnum(u16& result);
 261
 262         BufferedPacketPtr popFirst();
 263         BufferedPacketPtr popSeqnum(u16 seqnum);
 264         void insert(BufferedPacketPtr &p_ptr, u16 next_expected);
 265
 266         void incrementTimeouts(float dtime);
 267         std::list<ConstSharedPtr<BufferedPacket>> getTimedOuts(float timeout, u32 max_packets);
 268
 269         void print();
 270         bool empty();
 271         u32 size();
 272
 273
 274 private:
 275         RPBSearchResult findPacketNoLock(u16 seqnum);
 276
 277         std::list<BufferedPacketPtr> m_list;
 278
 279         u16 m_oldest_non_answered_ack;
 280
 281         std::mutex m_list_mutex;
 282 };
 283
 284 /*
 285         A buffer for reconstructing split packets
 286 */
 287
 288 class IncomingSplitBuffer
 289 {
 290 public:
 291         ~IncomingSplitBuffer();
 292         /*
 293                 Returns a reference counted buffer of length != 0 when a full split
 294                 packet is constructed. If not, returns one of length 0.
 295         */
 296         SharedBuffer<u8> insert(BufferedPacketPtr &p_ptr, bool reliable);
 297
 298         void removeUnreliableTimedOuts(float dtime, float timeout);
 299
 300 private:
 301         // Key is seqnum
 302         std::map<u16, IncomingSplitPacket*> m_buf;
 303
 304         std::mutex m_map_mutex;
 305 };
 306
 307 enum ConnectionCommandType{
 308         CONNCMD_NONE,
 309         CONNCMD_SERVE,
 310         CONNCMD_CONNECT,
 311         CONNCMD_DISCONNECT,
 312         CONNCMD_DISCONNECT_PEER,
 313         CONNCMD_SEND,
 314         CONNCMD_SEND_TO_ALL,
 315         CONCMD_ACK,
 316         CONCMD_CREATE_PEER
 317 };
 318
 319 struct ConnectionCommand;
 320 typedef std::shared_ptr<ConnectionCommand> ConnectionCommandPtr;
 321
 322 // This is very similar to ConnectionEvent
 323 struct ConnectionCommand
 324 {
 325         const ConnectionCommandType type;
 326         Address address;
 327         session_t peer_id = PEER_ID_INEXISTENT;
 328         u8 channelnum = 0;
 329         Buffer<u8> data;
 330         bool reliable = false;
 331         bool raw = false;
 332
 333         DISABLE_CLASS_COPY(ConnectionCommand);
 334
 335         static ConnectionCommandPtr serve(Address address);
 336         static ConnectionCommandPtr connect(Address address);
 337         static ConnectionCommandPtr disconnect();
 338         static ConnectionCommandPtr disconnect_peer(session_t peer_id);
 339         static ConnectionCommandPtr send(session_t peer_id, u8 channelnum, NetworkPacket *pkt, bool reliable);
 340         static ConnectionCommandPtr ack(session_t peer_id, u8 channelnum, const Buffer<u8> &data);
 341         static ConnectionCommandPtr createPeer(session_t peer_id, const Buffer<u8> &data);
 342
 343 private:
 344         ConnectionCommand(ConnectionCommandType type_) :
 345                 type(type_) {}
 346
 347         static ConnectionCommandPtr create(ConnectionCommandType type);
 348 };
 349
 350 /* maximum window size to use, 0xFFFF is theoretical maximum. don't think about
 351  * touching it, the less you're away from it the more likely data corruption
 352  * will occur
 353  */
 354 #define MAX_RELIABLE_WINDOW_SIZE 0x8000
 355 /* starting value for window size */
 356 #define START_RELIABLE_WINDOW_SIZE 0x400
 357 /* minimum value for window size */
 358 #define MIN_RELIABLE_WINDOW_SIZE 0x40
 359
 360 class Channel
 361 {
 362
 363 public:
 364         u16 readNextIncomingSeqNum();
 365         u16 incNextIncomingSeqNum();
 366
 367         u16 getOutgoingSequenceNumber(bool& successful);
 368         u16 readOutgoingSequenceNumber();
 369         bool putBackSequenceNumber(u16);
 370
 371         u16 readNextSplitSeqNum();
 372         void setNextSplitSeqNum(u16 seqnum);
 373
 374         // This is for buffering the incoming packets that are coming in
 375         // the wrong order
 376         ReliablePacketBuffer incoming_reliables;
 377         // This is for buffering the sent packets so that the sender can
 378         // re-send them if no ACK is received
 379         ReliablePacketBuffer outgoing_reliables_sent;
 380
 381         //queued reliable packets
 382         std::queue<BufferedPacketPtr> queued_reliables;
 383
 384         //queue commands prior splitting to packets
 385         std::deque<ConnectionCommandPtr> queued_commands;
 386
 387         IncomingSplitBuffer incoming_splits;
 388
 389         Channel() = default;
 390         ~Channel() = default;
 391
 392         void UpdatePacketLossCounter(unsigned int count);
 393         void UpdatePacketTooLateCounter();
 394         void UpdateBytesSent(unsigned int bytes,unsigned int packages=1);
 395         void UpdateBytesLost(unsigned int bytes);
 396         void UpdateBytesReceived(unsigned int bytes);
 397
 398         void UpdateTimers(float dtime);
 399
 400         float getCurrentDownloadRateKB()
 401                 { MutexAutoLock lock(m_internal_mutex); return cur_kbps; };
 402         float getMaxDownloadRateKB()
 403                 { MutexAutoLock lock(m_internal_mutex); return max_kbps; };
 404
 405         float getCurrentLossRateKB()
 406                 { MutexAutoLock lock(m_internal_mutex); return cur_kbps_lost; };
 407         float getMaxLossRateKB()
 408                 { MutexAutoLock lock(m_internal_mutex); return max_kbps_lost; };
 409
 410         float getCurrentIncomingRateKB()
 411                 { MutexAutoLock lock(m_internal_mutex); return cur_incoming_kbps; };
 412         float getMaxIncomingRateKB()
 413                 { MutexAutoLock lock(m_internal_mutex); return max_incoming_kbps; };
 414
 415         float getAvgDownloadRateKB()
 416                 { MutexAutoLock lock(m_internal_mutex); return avg_kbps; };
 417         float getAvgLossRateKB()
 418                 { MutexAutoLock lock(m_internal_mutex); return avg_kbps_lost; };
 419         float getAvgIncomingRateKB()
 420                 { MutexAutoLock lock(m_internal_mutex); return avg_incoming_kbps; };
 421
 422         u16 getWindowSize() const { return m_window_size; };
 423
 424         void setWindowSize(long size)
 425         {
 426                 m_window_size = (u16)rangelim(size, MIN_RELIABLE_WINDOW_SIZE, MAX_RELIABLE_WINDOW_SIZE);
 427         }
 428
 429 private:
 430         std::mutex m_internal_mutex;
 431         u16 m_window_size = MIN_RELIABLE_WINDOW_SIZE;
 432
 433         u16 next_incoming_seqnum = SEQNUM_INITIAL;
 434
 435         u16 next_outgoing_seqnum = SEQNUM_INITIAL;
 436         u16 next_outgoing_split_seqnum = SEQNUM_INITIAL;
 437
 438         unsigned int current_packet_loss = 0;
 439         unsigned int current_packet_too_late = 0;
 440         unsigned int current_packet_successful = 0;
 441         float packet_loss_counter = 0.0f;
 442
 443         unsigned int current_bytes_transfered = 0;
 444         unsigned int current_bytes_received = 0;
 445         unsigned int current_bytes_lost = 0;
 446         float max_kbps = 0.0f;
 447         float cur_kbps = 0.0f;
 448         float avg_kbps = 0.0f;
 449         float max_incoming_kbps = 0.0f;
 450         float cur_incoming_kbps = 0.0f;
 451         float avg_incoming_kbps = 0.0f;
 452         float max_kbps_lost = 0.0f;
 453         float cur_kbps_lost = 0.0f;
 454         float avg_kbps_lost = 0.0f;
 455         float bpm_counter = 0.0f;
 456
 457         unsigned int rate_samples = 0;
 458 };
 459
 460 class Peer;
 461
 462 class PeerHelper
 463 {
 464 public:
 465         PeerHelper() = default;
 466         PeerHelper(Peer* peer);
 467         ~PeerHelper();
 468
 469         PeerHelper&   operator=(Peer* peer);
 470         Peer*         operator->() const;
 471         bool          operator!();
 472         Peer*         operator&() const;
 473         bool          operator!=(void* ptr);
 474
 475 private:
 476         Peer *m_peer = nullptr;
 477 };
 478
 479 class Connection;
 480
 481 typedef enum {
 482         CUR_DL_RATE,
 483         AVG_DL_RATE,
 484         CUR_INC_RATE,
 485         AVG_INC_RATE,
 486         CUR_LOSS_RATE,
 487         AVG_LOSS_RATE,
 488 } rate_stat_type;
 489
 490 class Peer {
 491         public:
 492                 friend class PeerHelper;
 493
 494                 Peer(Address address_,session_t id_,Connection* connection) :
 495                         id(id_),
 496                         m_connection(connection),
 497                         address(address_),
 498                         m_last_timeout_check(porting::getTimeMs())
 499                 {
 500                 };
 501
 502                 virtual ~Peer() {
 503                         MutexAutoLock usage_lock(m_exclusive_access_mutex);
 504                         FATAL_ERROR_IF(m_usage != 0, "Reference counting failure");
 505                 };
 506
 507                 // Unique id of the peer
 508                 const session_t id;
 509
 510                 void Drop();
 511
 512                 virtual void PutReliableSendCommand(ConnectionCommandPtr &c,
 513                                                 unsigned int max_packet_size) {};
 514
 515                 virtual bool getAddress(MTProtocols type, Address& toset) = 0;
 516
 517                 bool isPendingDeletion()
 518                 { MutexAutoLock lock(m_exclusive_access_mutex); return m_pending_deletion; };
 519
 520                 void ResetTimeout()
 521                         {MutexAutoLock lock(m_exclusive_access_mutex); m_timeout_counter = 0.0; };
 522
 523                 bool isTimedOut(float timeout);
 524
 525                 unsigned int m_increment_packets_remaining = 0;
 526
 527                 virtual u16 getNextSplitSequenceNumber(u8 channel) { return 0; };
 528                 virtual void setNextSplitSequenceNumber(u8 channel, u16 seqnum) {};
 529                 virtual SharedBuffer<u8> addSplitPacket(u8 channel, BufferedPacketPtr &toadd,
 530                                 bool reliable)
 531                 {
 532                         errorstream << "Peer::addSplitPacket called,"
 533                                         << " this is supposed to be never called!" << std::endl;
 534                         return SharedBuffer<u8>(0);
 535                 };
 536
 537                 virtual bool Ping(float dtime, SharedBuffer<u8>& data) { return false; };
 538
 539                 virtual float getStat(rtt_stat_type type) const {
 540                         switch (type) {
 541                                 case MIN_RTT:
 542                                         return m_rtt.min_rtt;
 543                                 case MAX_RTT:
 544                                         return m_rtt.max_rtt;
 545                                 case AVG_RTT:
 546                                         return m_rtt.avg_rtt;
 547                                 case MIN_JITTER:
 548                                         return m_rtt.jitter_min;
 549                                 case MAX_JITTER:
 550                                         return m_rtt.jitter_max;
 551                                 case AVG_JITTER:
 552                                         return m_rtt.jitter_avg;
 553                         }
 554                         return -1;
 555                 }
 556         protected:
 557                 virtual void reportRTT(float rtt) {};
 558
 559                 void RTTStatistics(float rtt,
 560                                                         const std::string &profiler_id = "",
 561                                                         unsigned int num_samples = 1000);
 562
 563                 bool IncUseCount();
 564                 void DecUseCount();
 565
 566                 mutable std::mutex m_exclusive_access_mutex;
 567
 568                 bool m_pending_deletion = false;
 569
 570                 Connection* m_connection;
 571
 572                 // Address of the peer
 573                 Address address;
 574
 575                 // Ping timer
 576                 float m_ping_timer = 0.0f;
 577         private:
 578
 579                 struct rttstats {
 580                         float jitter_min = FLT_MAX;
 581                         float jitter_max = 0.0f;
 582                         float jitter_avg = -1.0f;
 583                         float min_rtt = FLT_MAX;
 584                         float max_rtt = 0.0f;
 585                         float avg_rtt = -1.0f;
 586
 587                         rttstats() = default;
 588                 };
 589
 590                 rttstats m_rtt;
 591                 float m_last_rtt = -1.0f;
 592
 593                 // current usage count
 594                 unsigned int m_usage = 0;
 595
 596                 // Seconds from last receive
 597                 float m_timeout_counter = 0.0f;
 598
 599                 u64 m_last_timeout_check;
 600 };
 601
 602 class UDPPeer : public Peer
 603 {
 604 public:
 605
 606         friend class PeerHelper;
 607         friend class ConnectionReceiveThread;
 608         friend class ConnectionSendThread;
 609         friend class Connection;
 610
 611         UDPPeer(u16 a_id, Address a_address, Connection* connection);
 612         virtual ~UDPPeer() = default;
 613
 614         void PutReliableSendCommand(ConnectionCommandPtr &c,
 615                                                         unsigned int max_packet_size);
 616
 617         bool getAddress(MTProtocols type, Address& toset);
 618
 619         u16 getNextSplitSequenceNumber(u8 channel);
 620         void setNextSplitSequenceNumber(u8 channel, u16 seqnum);
 621
 622         SharedBuffer<u8> addSplitPacket(u8 channel, BufferedPacketPtr &toadd,
 623                 bool reliable);
 624
 625 protected:
 626         /*
 627                 Calculates avg_rtt and resend_timeout.
 628                 rtt=-1 only recalculates resend_timeout
 629         */
 630         void reportRTT(float rtt);
 631
 632         void RunCommandQueues(
 633                                         unsigned int max_packet_size,
 634                                         unsigned int maxcommands,
 635                                         unsigned int maxtransfer);
 636
 637         float getResendTimeout()
 638                 { MutexAutoLock lock(m_exclusive_access_mutex); return resend_timeout; }
 639
 640         void setResendTimeout(float timeout)
 641                 { MutexAutoLock lock(m_exclusive_access_mutex); resend_timeout = timeout; }
 642         bool Ping(float dtime,SharedBuffer<u8>& data);
 643
 644         Channel channels[CHANNEL_COUNT];
 645         bool m_pending_disconnect = false;
 646 private:
 647         // This is changed dynamically
 648         float resend_timeout = 0.5;
 649
 650         bool processReliableSendCommand(
 651                                         ConnectionCommandPtr &c_ptr,
 652                                         unsigned int max_packet_size);
 653 };
 654
 655 /*
 656         Connection
 657 */
 658
 659 enum ConnectionEventType {
 660         CONNEVENT_NONE,
 661         CONNEVENT_DATA_RECEIVED,
 662         CONNEVENT_PEER_ADDED,
 663         CONNEVENT_PEER_REMOVED,
 664         CONNEVENT_BIND_FAILED,
 665 };
 666
 667 struct ConnectionEvent;
 668 typedef std::shared_ptr<ConnectionEvent> ConnectionEventPtr;
 669
 670 // This is very similar to ConnectionCommand
 671 struct ConnectionEvent
 672 {
 673         const ConnectionEventType type;
 674         session_t peer_id = 0;
 675         Buffer<u8> data;
 676         bool timeout = false;
 677         Address address;
 678
 679         // We don't want to copy "data"
 680         DISABLE_CLASS_COPY(ConnectionEvent);
 681
 682         static ConnectionEventPtr create(ConnectionEventType type);
 683         static ConnectionEventPtr dataReceived(session_t peer_id, const Buffer<u8> &data);
 684         static ConnectionEventPtr peerAdded(session_t peer_id, Address address);
 685         static ConnectionEventPtr peerRemoved(session_t peer_id, bool is_timeout, Address address);
 686         static ConnectionEventPtr bindFailed();
 687
 688         const char *describe() const;
 689
 690 private:
 691         ConnectionEvent(ConnectionEventType type_) :
 692                 type(type_) {}
 693 };
 694
 695 class PeerHandler;
 696
 697 class Connection
 698 {
 699 public:
 700         friend class ConnectionSendThread;
 701         friend class ConnectionReceiveThread;
 702
 703         Connection(u32 protocol_id, u32 max_packet_size, float timeout, bool ipv6,
 704                         PeerHandler *peerhandler);
 705         ~Connection();
 706
 707         /* Interface */
 708         ConnectionEventPtr waitEvent(u32 timeout_ms);
 709
 710         void putCommand(ConnectionCommandPtr c);
 711
 712         void SetTimeoutMs(u32 timeout) { m_bc_receive_timeout = timeout; }
 713         void Serve(Address bind_addr);
 714         void Connect(Address address);
 715         bool Connected();
 716         void Disconnect();
 717         void Receive(NetworkPacket* pkt);
 718         bool TryReceive(NetworkPacket *pkt);
 719         void Send(session_t peer_id, u8 channelnum, NetworkPacket *pkt, bool reliable);
 720         session_t GetPeerID() const { return m_peer_id; }
 721         Address GetPeerAddress(session_t peer_id);
 722         float getPeerStat(session_t peer_id, rtt_stat_type type);
 723         float getLocalStat(rate_stat_type type);
 724         u32 GetProtocolID() const { return m_protocol_id; };
 725         const std::string getDesc();
 726         void DisconnectPeer(session_t peer_id);
 727
 728 protected:
 729         PeerHelper getPeerNoEx(session_t peer_id);
 730         u16   lookupPeer(Address& sender);
 731
 732         u16 createPeer(Address& sender, MTProtocols protocol, int fd);
 733         UDPPeer*  createServerPeer(Address& sender);
 734         bool deletePeer(session_t peer_id, bool timeout);
 735
 736         void SetPeerID(session_t id) { m_peer_id = id; }
 737
 738         void sendAck(session_t peer_id, u8 channelnum, u16 seqnum);
 739
 740         std::vector<session_t> getPeerIDs()
 741         {
 742                 MutexAutoLock peerlock(m_peers_mutex);
 743                 return m_peer_ids;
 744         }
 745
 746         UDPSocket m_udpSocket;
 747         // Command queue: user -> SendThread
 748         MutexedQueue<ConnectionCommandPtr> m_command_queue;
 749
 750         bool Receive(NetworkPacket *pkt, u32 timeout);
 751
 752         void putEvent(ConnectionEventPtr e);
 753
 754         void TriggerSend();
 755
 756         bool ConnectedToServer()
 757         {
 758                 return getPeerNoEx(PEER_ID_SERVER) != nullptr;
 759         }
 760 private:
 761         // Event queue: ReceiveThread -> user
 762         MutexedQueue<ConnectionEventPtr> m_event_queue;
 763
 764         session_t m_peer_id = 0;
 765         u32 m_protocol_id;
 766
 767         std::map<session_t, Peer *> m_peers;
 768         std::vector<session_t> m_peer_ids;
 769         std::mutex m_peers_mutex;
 770
 771         std::unique_ptr<ConnectionSendThread> m_sendThread;
 772         std::unique_ptr<ConnectionReceiveThread> m_receiveThread;
 773
 774         mutable std::mutex m_info_mutex;
 775
 776         // Backwards compatibility
 777         PeerHandler *m_bc_peerhandler;
 778         u32 m_bc_receive_timeout = 0;
 779
 780         bool m_shutting_down = false;
 781
 782         session_t m_next_remote_peer_id = 2;
 783 };
 784
 785 } // namespace