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 <iostream>
  31 #include <fstream>
  32 #include <list>
  33 #include <map>
  34
  35 namespace con
  36 {
  37
  38 /*
  39         Exceptions
  40 */
  41 class NotFoundException : public BaseException
  42 {
  43 public:
  44         NotFoundException(const char *s):
  45                 BaseException(s)
  46         {}
  47 };
  48
  49 class PeerNotFoundException : public BaseException
  50 {
  51 public:
  52         PeerNotFoundException(const char *s):
  53                 BaseException(s)
  54         {}
  55 };
  56
  57 class ConnectionException : public BaseException
  58 {
  59 public:
  60         ConnectionException(const char *s):
  61                 BaseException(s)
  62         {}
  63 };
  64
  65 class ConnectionBindFailed : public BaseException
  66 {
  67 public:
  68         ConnectionBindFailed(const char *s):
  69                 BaseException(s)
  70         {}
  71 };
  72
  73 /*class ThrottlingException : public BaseException
  74 {
  75 public:
  76         ThrottlingException(const char *s):
  77                 BaseException(s)
  78         {}
  79 };*/
  80
  81 class InvalidIncomingDataException : public BaseException
  82 {
  83 public:
  84         InvalidIncomingDataException(const char *s):
  85                 BaseException(s)
  86         {}
  87 };
  88
  89 class InvalidOutgoingDataException : public BaseException
  90 {
  91 public:
  92         InvalidOutgoingDataException(const char *s):
  93                 BaseException(s)
  94         {}
  95 };
  96
  97 class NoIncomingDataException : public BaseException
  98 {
  99 public:
 100         NoIncomingDataException(const char *s):
 101                 BaseException(s)
 102         {}
 103 };
 104
 105 class ProcessedSilentlyException : public BaseException
 106 {
 107 public:
 108         ProcessedSilentlyException(const char *s):
 109                 BaseException(s)
 110         {}
 111 };
 112
 113 #define SEQNUM_MAX 65535
 114 inline bool seqnum_higher(u16 higher, u16 lower)
 115 {
 116         if(lower > higher && lower - higher > SEQNUM_MAX/2){
 117                 return true;
 118         }
 119         return (higher > lower);
 120 }
 121
 122 struct BufferedPacket
 123 {
 124         BufferedPacket(u8 *a_data, u32 a_size):
 125                 data(a_data, a_size), time(0.0), totaltime(0.0)
 126         {}
 127         BufferedPacket(u32 a_size):
 128                 data(a_size), time(0.0), totaltime(0.0)
 129         {}
 130         SharedBuffer<u8> data; // Data of the packet, including headers
 131         float time; // Seconds from buffering the packet or re-sending
 132         float totaltime; // Seconds from buffering the packet
 133         Address address; // Sender or destination
 134 };
 135
 136 // This adds the base headers to the data and makes a packet out of it
 137 BufferedPacket makePacket(Address &address, u8 *data, u32 datasize,
 138                 u32 protocol_id, u16 sender_peer_id, u8 channel);
 139 BufferedPacket makePacket(Address &address, SharedBuffer<u8> &data,
 140                 u32 protocol_id, u16 sender_peer_id, u8 channel);
 141
 142 // Add the TYPE_ORIGINAL header to the data
 143 SharedBuffer<u8> makeOriginalPacket(
 144                 SharedBuffer<u8> data);
 145
 146 // Split data in chunks and add TYPE_SPLIT headers to them
 147 std::list<SharedBuffer<u8> > makeSplitPacket(
 148                 SharedBuffer<u8> data,
 149                 u32 chunksize_max,
 150                 u16 seqnum);
 151
 152 // Depending on size, make a TYPE_ORIGINAL or TYPE_SPLIT packet
 153 // Increments split_seqnum if a split packet is made
 154 std::list<SharedBuffer<u8> > makeAutoSplitPacket(
 155                 SharedBuffer<u8> data,
 156                 u32 chunksize_max,
 157                 u16 &split_seqnum);
 158
 159 // Add the TYPE_RELIABLE header to the data
 160 SharedBuffer<u8> makeReliablePacket(
 161                 SharedBuffer<u8> data,
 162                 u16 seqnum);
 163
 164 struct IncomingSplitPacket
 165 {
 166         IncomingSplitPacket()
 167         {
 168                 time = 0.0;
 169                 reliable = false;
 170         }
 171         // Key is chunk number, value is data without headers
 172         std::map<u16, SharedBuffer<u8> > chunks;
 173         u32 chunk_count;
 174         float time; // Seconds from adding
 175         bool reliable; // If true, isn't deleted on timeout
 176
 177         bool allReceived()
 178         {
 179                 return (chunks.size() == chunk_count);
 180         }
 181 };
 182
 183 /*
 184 === NOTES ===
 185
 186 A packet is sent through a channel to a peer with a basic header:
 187 TODO: Should we have a receiver_peer_id also?
 188         Header (7 bytes):
 189         [0] u32 protocol_id
 190         [4] u16 sender_peer_id
 191         [6] u8 channel
 192 sender_peer_id:
 193         Unique to each peer.
 194         value 0 (PEER_ID_INEXISTENT) is reserved for making new connections
 195         value 1 (PEER_ID_SERVER) is reserved for server
 196         these constants are defined in constants.h
 197 channel:
 198         The lower the number, the higher the priority is.
 199         Only channels 0, 1 and 2 exist.
 200 */
 201 #define BASE_HEADER_SIZE 7
 202 #define CHANNEL_COUNT 3
 203 /*
 204 Packet types:
 205
 206 CONTROL: This is a packet used by the protocol.
 207 - When this is processed, nothing is handed to the user.
 208         Header (2 byte):
 209         [0] u8 type
 210         [1] u8 controltype
 211 controltype and data description:
 212         CONTROLTYPE_ACK
 213                 [2] u16 seqnum
 214         CONTROLTYPE_SET_PEER_ID
 215                 [2] u16 peer_id_new
 216         CONTROLTYPE_PING
 217         - There is no actual reply, but this can be sent in a reliable
 218           packet to get a reply
 219         CONTROLTYPE_DISCO
 220 */
 221 #define TYPE_CONTROL 0
 222 #define CONTROLTYPE_ACK 0
 223 #define CONTROLTYPE_SET_PEER_ID 1
 224 #define CONTROLTYPE_PING 2
 225 #define CONTROLTYPE_DISCO 3
 226 /*
 227 ORIGINAL: This is a plain packet with no control and no error
 228 checking at all.
 229 - When this is processed, it is directly handed to the user.
 230         Header (1 byte):
 231         [0] u8 type
 232 */
 233 #define TYPE_ORIGINAL 1
 234 #define ORIGINAL_HEADER_SIZE 1
 235 /*
 236 SPLIT: These are sequences of packets forming one bigger piece of
 237 data.
 238 - When processed and all the packet_nums 0...packet_count-1 are
 239   present (this should be buffered), the resulting data shall be
 240   directly handed to the user.
 241 - If the data fails to come up in a reasonable time, the buffer shall
 242   be silently discarded.
 243 - These can be sent as-is or atop of a RELIABLE packet stream.
 244         Header (7 bytes):
 245         [0] u8 type
 246         [1] u16 seqnum
 247         [3] u16 chunk_count
 248         [5] u16 chunk_num
 249 */
 250 #define TYPE_SPLIT 2
 251 /*
 252 RELIABLE: Delivery of all RELIABLE packets shall be forced by ACKs,
 253 and they shall be delivered in the same order as sent. This is done
 254 with a buffer in the receiving and transmitting end.
 255 - When this is processed, the contents of each packet is recursively
 256   processed as packets.
 257         Header (3 bytes):
 258         [0] u8 type
 259         [1] u16 seqnum
 260
 261 */
 262 #define TYPE_RELIABLE 3
 263 #define RELIABLE_HEADER_SIZE 3
 264 //#define SEQNUM_INITIAL 0x10
 265 #define SEQNUM_INITIAL 65500
 266
 267 /*
 268         A buffer which stores reliable packets and sorts them internally
 269         for fast access to the smallest one.
 270 */
 271
 272 typedef std::list<BufferedPacket>::iterator RPBSearchResult;
 273
 274 class ReliablePacketBuffer
 275 {
 276 public:
 277         ReliablePacketBuffer();
 278         void print();
 279         bool empty();
 280         u32 size();
 281         RPBSearchResult findPacket(u16 seqnum);
 282         RPBSearchResult notFound();
 283         bool getFirstSeqnum(u16 *result);
 284         BufferedPacket popFirst();
 285         BufferedPacket popSeqnum(u16 seqnum);
 286         void insert(BufferedPacket &p);
 287         void incrementTimeouts(float dtime);
 288         void resetTimedOuts(float timeout);
 289         bool anyTotaltimeReached(float timeout);
 290         std::list<BufferedPacket> getTimedOuts(float timeout);
 291
 292 private:
 293         std::list<BufferedPacket> m_list;
 294         u16 m_list_size;
 295 };
 296
 297 /*
 298         A buffer for reconstructing split packets
 299 */
 300
 301 class IncomingSplitBuffer
 302 {
 303 public:
 304         ~IncomingSplitBuffer();
 305         /*
 306                 Returns a reference counted buffer of length != 0 when a full split
 307                 packet is constructed. If not, returns one of length 0.
 308         */
 309         SharedBuffer<u8> insert(BufferedPacket &p, bool reliable);
 310
 311         void removeUnreliableTimedOuts(float dtime, float timeout);
 312
 313 private:
 314         // Key is seqnum
 315         std::map<u16, IncomingSplitPacket*> m_buf;
 316 };
 317
 318 class Connection;
 319
 320 struct Channel
 321 {
 322         Channel();
 323         ~Channel();
 324
 325         u16 next_outgoing_seqnum;
 326         u16 next_incoming_seqnum;
 327         u16 next_outgoing_split_seqnum;
 328
 329         // This is for buffering the incoming packets that are coming in
 330         // the wrong order
 331         ReliablePacketBuffer incoming_reliables;
 332         // This is for buffering the sent packets so that the sender can
 333         // re-send them if no ACK is received
 334         ReliablePacketBuffer outgoing_reliables;
 335
 336         IncomingSplitBuffer incoming_splits;
 337 };
 338
 339 class Peer;
 340
 341 class PeerHandler
 342 {
 343 public:
 344         PeerHandler()
 345         {
 346         }
 347         virtual ~PeerHandler()
 348         {
 349         }
 350
 351         /*
 352                 This is called after the Peer has been inserted into the
 353                 Connection's peer container.
 354         */
 355         virtual void peerAdded(Peer *peer) = 0;
 356         /*
 357                 This is called before the Peer has been removed from the
 358                 Connection's peer container.
 359         */
 360         virtual void deletingPeer(Peer *peer, bool timeout) = 0;
 361 };
 362
 363 class Peer
 364 {
 365 public:
 366
 367         Peer(u16 a_id, Address a_address);
 368         virtual ~Peer();
 369
 370         /*
 371                 Calculates avg_rtt and resend_timeout.
 372
 373                 rtt=-1 only recalculates resend_timeout
 374         */
 375         void reportRTT(float rtt);
 376
 377         Channel channels[CHANNEL_COUNT];
 378
 379         // Address of the peer
 380         Address address;
 381         // Unique id of the peer
 382         u16 id;
 383         // Seconds from last receive
 384         float timeout_counter;
 385         // Ping timer
 386         float ping_timer;
 387         // This is changed dynamically
 388         float resend_timeout;
 389         // Updated when an ACK is received
 390         float avg_rtt;
 391         // This is set to true when the peer has actually sent something
 392         // with the id we have given to it
 393         bool has_sent_with_id;
 394
 395         float m_sendtime_accu;
 396         float m_max_packets_per_second;
 397         int m_num_sent;
 398         int m_max_num_sent;
 399
 400         // Updated from configuration by Connection
 401         float congestion_control_aim_rtt;
 402         float congestion_control_max_rate;
 403         float congestion_control_min_rate;
 404 private:
 405 };
 406
 407 /*
 408         Connection
 409 */
 410
 411 struct OutgoingPacket
 412 {
 413         u16 peer_id;
 414         u8 channelnum;
 415         SharedBuffer<u8> data;
 416         bool reliable;
 417
 418         OutgoingPacket(u16 peer_id_, u8 channelnum_, SharedBuffer<u8> data_,
 419                         bool reliable_):
 420                 peer_id(peer_id_),
 421                 channelnum(channelnum_),
 422                 data(data_),
 423                 reliable(reliable_)
 424         {
 425         }
 426 };
 427
 428 enum ConnectionEventType{
 429         CONNEVENT_NONE,
 430         CONNEVENT_DATA_RECEIVED,
 431         CONNEVENT_PEER_ADDED,
 432         CONNEVENT_PEER_REMOVED,
 433         CONNEVENT_BIND_FAILED,
 434 };
 435
 436 struct ConnectionEvent
 437 {
 438         enum ConnectionEventType type;
 439         u16 peer_id;
 440         Buffer<u8> data;
 441         bool timeout;
 442         Address address;
 443
 444         ConnectionEvent(): type(CONNEVENT_NONE) {}
 445
 446         std::string describe()
 447         {
 448                 switch(type){
 449                 case CONNEVENT_NONE:
 450                         return "CONNEVENT_NONE";
 451                 case CONNEVENT_DATA_RECEIVED:
 452                         return "CONNEVENT_DATA_RECEIVED";
 453                 case CONNEVENT_PEER_ADDED:
 454                         return "CONNEVENT_PEER_ADDED";
 455                 case CONNEVENT_PEER_REMOVED:
 456                         return "CONNEVENT_PEER_REMOVED";
 457                 case CONNEVENT_BIND_FAILED:
 458                         return "CONNEVENT_BIND_FAILED";
 459                 }
 460                 return "Invalid ConnectionEvent";
 461         }
 462
 463         void dataReceived(u16 peer_id_, SharedBuffer<u8> data_)
 464         {
 465                 type = CONNEVENT_DATA_RECEIVED;
 466                 peer_id = peer_id_;
 467                 data = data_;
 468         }
 469         void peerAdded(u16 peer_id_, Address address_)
 470         {
 471                 type = CONNEVENT_PEER_ADDED;
 472                 peer_id = peer_id_;
 473                 address = address_;
 474         }
 475         void peerRemoved(u16 peer_id_, bool timeout_, Address address_)
 476         {
 477                 type = CONNEVENT_PEER_REMOVED;
 478                 peer_id = peer_id_;
 479                 timeout = timeout_;
 480                 address = address_;
 481         }
 482         void bindFailed()
 483         {
 484                 type = CONNEVENT_BIND_FAILED;
 485         }
 486 };
 487
 488 enum ConnectionCommandType{
 489         CONNCMD_NONE,
 490         CONNCMD_SERVE,
 491         CONNCMD_CONNECT,
 492         CONNCMD_DISCONNECT,
 493         CONNCMD_SEND,
 494         CONNCMD_SEND_TO_ALL,
 495         CONNCMD_DELETE_PEER,
 496 };
 497
 498 struct ConnectionCommand
 499 {
 500         enum ConnectionCommandType type;
 501         u16 port;
 502         Address address;
 503         u16 peer_id;
 504         u8 channelnum;
 505         Buffer<u8> data;
 506         bool reliable;
 507
 508         ConnectionCommand(): type(CONNCMD_NONE) {}
 509
 510         void serve(u16 port_)
 511         {
 512                 type = CONNCMD_SERVE;
 513                 port = port_;
 514         }
 515         void connect(Address address_)
 516         {
 517                 type = CONNCMD_CONNECT;
 518                 address = address_;
 519         }
 520         void disconnect()
 521         {
 522                 type = CONNCMD_DISCONNECT;
 523         }
 524         void send(u16 peer_id_, u8 channelnum_,
 525                         SharedBuffer<u8> data_, bool reliable_)
 526         {
 527                 type = CONNCMD_SEND;
 528                 peer_id = peer_id_;
 529                 channelnum = channelnum_;
 530                 data = data_;
 531                 reliable = reliable_;
 532         }
 533         void sendToAll(u8 channelnum_, SharedBuffer<u8> data_, bool reliable_)
 534         {
 535                 type = CONNCMD_SEND_TO_ALL;
 536                 channelnum = channelnum_;
 537                 data = data_;
 538                 reliable = reliable_;
 539         }
 540         void deletePeer(u16 peer_id_)
 541         {
 542                 type = CONNCMD_DELETE_PEER;
 543                 peer_id = peer_id_;
 544         }
 545 };
 546
 547 class Connection: public SimpleThread
 548 {
 549 public:
 550         Connection(u32 protocol_id, u32 max_packet_size, float timeout, bool ipv6);
 551         Connection(u32 protocol_id, u32 max_packet_size, float timeout, bool ipv6,
 552                         PeerHandler *peerhandler);
 553         ~Connection();
 554         void * Thread();
 555
 556         /* Interface */
 557
 558         ConnectionEvent getEvent();
 559         ConnectionEvent waitEvent(u32 timeout_ms);
 560         void putCommand(ConnectionCommand &c);
 561
 562         void SetTimeoutMs(int timeout){ m_bc_receive_timeout = timeout; }
 563         void Serve(unsigned short port);
 564         void Connect(Address address);
 565         bool Connected();
 566         void Disconnect();
 567         u32 Receive(u16 &peer_id, SharedBuffer<u8> &data);
 568         void SendToAll(u8 channelnum, SharedBuffer<u8> data, bool reliable);
 569         void Send(u16 peer_id, u8 channelnum, SharedBuffer<u8> data, bool reliable);
 570         void RunTimeouts(float dtime); // dummy
 571         u16 GetPeerID(){ return m_peer_id; }
 572         Address GetPeerAddress(u16 peer_id);
 573         float GetPeerAvgRTT(u16 peer_id);
 574         void DeletePeer(u16 peer_id);
 575
 576 private:
 577         void putEvent(ConnectionEvent &e);
 578         void processCommand(ConnectionCommand &c);
 579         void send(float dtime);
 580         void receive();
 581         void runTimeouts(float dtime);
 582         void serve(u16 port);
 583         void connect(Address address);
 584         void disconnect();
 585         void sendToAll(u8 channelnum, SharedBuffer<u8> data, bool reliable);
 586         void send(u16 peer_id, u8 channelnum, SharedBuffer<u8> data, bool reliable);
 587         void sendAsPacket(u16 peer_id, u8 channelnum,
 588                         SharedBuffer<u8> data, bool reliable);
 589         void rawSendAsPacket(u16 peer_id, u8 channelnum,
 590                         SharedBuffer<u8> data, bool reliable);
 591         void rawSend(const BufferedPacket &packet);
 592         Peer* getPeer(u16 peer_id);
 593         Peer* getPeerNoEx(u16 peer_id);
 594         std::list<Peer*> getPeers();
 595         bool getFromBuffers(u16 &peer_id, SharedBuffer<u8> &dst);
 596         // Returns next data from a buffer if possible
 597         // If found, returns true; if not, false.
 598         // If found, sets peer_id and dst
 599         bool checkIncomingBuffers(Channel *channel, u16 &peer_id,
 600                         SharedBuffer<u8> &dst);
 601         /*
 602                 Processes a packet with the basic header stripped out.
 603                 Parameters:
 604                         packetdata: Data in packet (with no base headers)
 605                         peer_id: peer id of the sender of the packet in question
 606                         channelnum: channel on which the packet was sent
 607                         reliable: true if recursing into a reliable packet
 608         */
 609         SharedBuffer<u8> processPacket(Channel *channel,
 610                         SharedBuffer<u8> packetdata, u16 peer_id,
 611                         u8 channelnum, bool reliable);
 612         bool deletePeer(u16 peer_id, bool timeout);
 613
 614         Queue<OutgoingPacket> m_outgoing_queue;
 615         MutexedQueue<ConnectionEvent> m_event_queue;
 616         MutexedQueue<ConnectionCommand> m_command_queue;
 617
 618         u32 m_protocol_id;
 619         u32 m_max_packet_size;
 620         float m_timeout;
 621         UDPSocket m_socket;
 622         u16 m_peer_id;
 623
 624         std::map<u16, Peer*> m_peers;
 625         JMutex m_peers_mutex;
 626
 627         // Backwards compatibility
 628         PeerHandler *m_bc_peerhandler;
 629         int m_bc_receive_timeout;
 630
 631         void SetPeerID(u16 id){ m_peer_id = id; }
 632         u32 GetProtocolID(){ return m_protocol_id; }
 633         void PrintInfo(std::ostream &out);
 634         void PrintInfo();
 635         std::string getDesc();
 636         u16 m_indentation;
 637 };
 638
 639 } // namespace
 640
 641 #endif
 642