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 is reserved for making new connections
 195         value 1 is reserved for server
 196 channel:
 197         The lower the number, the higher the priority is.
 198         Only channels 0, 1 and 2 exist.
 199 */
 200 #define BASE_HEADER_SIZE 7
 201 #define PEER_ID_INEXISTENT 0
 202 #define PEER_ID_SERVER 1
 203 #define CHANNEL_COUNT 3
 204 /*
 205 Packet types:
 206
 207 CONTROL: This is a packet used by the protocol.
 208 - When this is processed, nothing is handed to the user.
 209         Header (2 byte):
 210         [0] u8 type
 211         [1] u8 controltype
 212 controltype and data description:
 213         CONTROLTYPE_ACK
 214                 [2] u16 seqnum
 215         CONTROLTYPE_SET_PEER_ID
 216                 [2] u16 peer_id_new
 217         CONTROLTYPE_PING
 218         - There is no actual reply, but this can be sent in a reliable
 219           packet to get a reply
 220         CONTROLTYPE_DISCO
 221 */
 222 #define TYPE_CONTROL 0
 223 #define CONTROLTYPE_ACK 0
 224 #define CONTROLTYPE_SET_PEER_ID 1
 225 #define CONTROLTYPE_PING 2
 226 #define CONTROLTYPE_DISCO 3
 227 /*
 228 ORIGINAL: This is a plain packet with no control and no error
 229 checking at all.
 230 - When this is processed, it is directly handed to the user.
 231         Header (1 byte):
 232         [0] u8 type
 233 */
 234 #define TYPE_ORIGINAL 1
 235 #define ORIGINAL_HEADER_SIZE 1
 236 /*
 237 SPLIT: These are sequences of packets forming one bigger piece of
 238 data.
 239 - When processed and all the packet_nums 0...packet_count-1 are
 240   present (this should be buffered), the resulting data shall be
 241   directly handed to the user.
 242 - If the data fails to come up in a reasonable time, the buffer shall
 243   be silently discarded.
 244 - These can be sent as-is or atop of a RELIABLE packet stream.
 245         Header (7 bytes):
 246         [0] u8 type
 247         [1] u16 seqnum
 248         [3] u16 chunk_count
 249         [5] u16 chunk_num
 250 */
 251 #define TYPE_SPLIT 2
 252 /*
 253 RELIABLE: Delivery of all RELIABLE packets shall be forced by ACKs,
 254 and they shall be delivered in the same order as sent. This is done
 255 with a buffer in the receiving and transmitting end.
 256 - When this is processed, the contents of each packet is recursively
 257   processed as packets.
 258         Header (3 bytes):
 259         [0] u8 type
 260         [1] u16 seqnum
 261
 262 */
 263 #define TYPE_RELIABLE 3
 264 #define RELIABLE_HEADER_SIZE 3
 265 //#define SEQNUM_INITIAL 0x10
 266 #define SEQNUM_INITIAL 65500
 267
 268 /*
 269         A buffer which stores reliable packets and sorts them internally
 270         for fast access to the smallest one.
 271 */
 272
 273 typedef std::list<BufferedPacket>::iterator RPBSearchResult;
 274
 275 class ReliablePacketBuffer
 276 {
 277 public:
 278         ReliablePacketBuffer();
 279         void print();
 280         bool empty();
 281         u32 size();
 282         RPBSearchResult findPacket(u16 seqnum);
 283         RPBSearchResult notFound();
 284         u16 getFirstSeqnum();
 285         BufferedPacket popFirst();
 286         BufferedPacket popSeqnum(u16 seqnum);
 287         void insert(BufferedPacket &p);
 288         void incrementTimeouts(float dtime);
 289         void resetTimedOuts(float timeout);
 290         bool anyTotaltimeReached(float timeout);
 291         std::list<BufferedPacket> getTimedOuts(float timeout);
 292
 293 private:
 294         std::list<BufferedPacket> m_list;
 295         u16 m_list_size;
 296 };
 297
 298 /*
 299         A buffer for reconstructing split packets
 300 */
 301
 302 class IncomingSplitBuffer
 303 {
 304 public:
 305         ~IncomingSplitBuffer();
 306         /*
 307                 Returns a reference counted buffer of length != 0 when a full split
 308                 packet is constructed. If not, returns one of length 0.
 309         */
 310         SharedBuffer<u8> insert(BufferedPacket &p, bool reliable);
 311
 312         void removeUnreliableTimedOuts(float dtime, float timeout);
 313
 314 private:
 315         // Key is seqnum
 316         std::map<u16, IncomingSplitPacket*> m_buf;
 317 };
 318
 319 class Connection;
 320
 321 struct Channel
 322 {
 323         Channel();
 324         ~Channel();
 325
 326         u16 next_outgoing_seqnum;
 327         u16 next_incoming_seqnum;
 328         u16 next_outgoing_split_seqnum;
 329
 330         // This is for buffering the incoming packets that are coming in
 331         // the wrong order
 332         ReliablePacketBuffer incoming_reliables;
 333         // This is for buffering the sent packets so that the sender can
 334         // re-send them if no ACK is received
 335         ReliablePacketBuffer outgoing_reliables;
 336
 337         IncomingSplitBuffer incoming_splits;
 338 };
 339
 340 class Peer;
 341
 342 class PeerHandler
 343 {
 344 public:
 345         PeerHandler()
 346         {
 347         }
 348         virtual ~PeerHandler()
 349         {
 350         }
 351
 352         /*
 353                 This is called after the Peer has been inserted into the
 354                 Connection's peer container.
 355         */
 356         virtual void peerAdded(Peer *peer) = 0;
 357         /*
 358                 This is called before the Peer has been removed from the
 359                 Connection's peer container.
 360         */
 361         virtual void deletingPeer(Peer *peer, bool timeout) = 0;
 362 };
 363
 364 class Peer
 365 {
 366 public:
 367
 368         Peer(u16 a_id, Address a_address);
 369         virtual ~Peer();
 370
 371         /*
 372                 Calculates avg_rtt and resend_timeout.
 373
 374                 rtt=-1 only recalculates resend_timeout
 375         */
 376         void reportRTT(float rtt);
 377
 378         Channel channels[CHANNEL_COUNT];
 379
 380         // Address of the peer
 381         Address address;
 382         // Unique id of the peer
 383         u16 id;
 384         // Seconds from last receive
 385         float timeout_counter;
 386         // Ping timer
 387         float ping_timer;
 388         // This is changed dynamically
 389         float resend_timeout;
 390         // Updated when an ACK is received
 391         float avg_rtt;
 392         // This is set to true when the peer has actually sent something
 393         // with the id we have given to it
 394         bool has_sent_with_id;
 395
 396         float m_sendtime_accu;
 397         float m_max_packets_per_second;
 398         int m_num_sent;
 399         int m_max_num_sent;
 400
 401         // Updated from configuration by Connection
 402         float congestion_control_aim_rtt;
 403         float congestion_control_max_rate;
 404         float congestion_control_min_rate;
 405 private:
 406 };
 407
 408 /*
 409         Connection
 410 */
 411
 412 struct OutgoingPacket
 413 {
 414         u16 peer_id;
 415         u8 channelnum;
 416         SharedBuffer<u8> data;
 417         bool reliable;
 418
 419         OutgoingPacket(u16 peer_id_, u8 channelnum_, SharedBuffer<u8> data_,
 420                         bool reliable_):
 421                 peer_id(peer_id_),
 422                 channelnum(channelnum_),
 423                 data(data_),
 424                 reliable(reliable_)
 425         {
 426         }
 427 };
 428
 429 enum ConnectionEventType{
 430         CONNEVENT_NONE,
 431         CONNEVENT_DATA_RECEIVED,
 432         CONNEVENT_PEER_ADDED,
 433         CONNEVENT_PEER_REMOVED,
 434         CONNEVENT_BIND_FAILED,
 435 };
 436
 437 struct ConnectionEvent
 438 {
 439         enum ConnectionEventType type;
 440         u16 peer_id;
 441         Buffer<u8> data;
 442         bool timeout;
 443         Address address;
 444
 445         ConnectionEvent(): type(CONNEVENT_NONE) {}
 446
 447         std::string describe()
 448         {
 449                 switch(type){
 450                 case CONNEVENT_NONE:
 451                         return "CONNEVENT_NONE";
 452                 case CONNEVENT_DATA_RECEIVED:
 453                         return "CONNEVENT_DATA_RECEIVED";
 454                 case CONNEVENT_PEER_ADDED:
 455                         return "CONNEVENT_PEER_ADDED";
 456                 case CONNEVENT_PEER_REMOVED:
 457                         return "CONNEVENT_PEER_REMOVED";
 458                 case CONNEVENT_BIND_FAILED:
 459                         return "CONNEVENT_BIND_FAILED";
 460                 }
 461                 return "Invalid ConnectionEvent";
 462         }
 463
 464         void dataReceived(u16 peer_id_, SharedBuffer<u8> data_)
 465         {
 466                 type = CONNEVENT_DATA_RECEIVED;
 467                 peer_id = peer_id_;
 468                 data = data_;
 469         }
 470         void peerAdded(u16 peer_id_, Address address_)
 471         {
 472                 type = CONNEVENT_PEER_ADDED;
 473                 peer_id = peer_id_;
 474                 address = address_;
 475         }
 476         void peerRemoved(u16 peer_id_, bool timeout_, Address address_)
 477         {
 478                 type = CONNEVENT_PEER_REMOVED;
 479                 peer_id = peer_id_;
 480                 timeout = timeout_;
 481                 address = address_;
 482         }
 483         void bindFailed()
 484         {
 485                 type = CONNEVENT_BIND_FAILED;
 486         }
 487 };
 488
 489 enum ConnectionCommandType{
 490         CONNCMD_NONE,
 491         CONNCMD_SERVE,
 492         CONNCMD_CONNECT,
 493         CONNCMD_DISCONNECT,
 494         CONNCMD_SEND,
 495         CONNCMD_SEND_TO_ALL,
 496         CONNCMD_DELETE_PEER,
 497 };
 498
 499 struct ConnectionCommand
 500 {
 501         enum ConnectionCommandType type;
 502         u16 port;
 503         Address address;
 504         u16 peer_id;
 505         u8 channelnum;
 506         Buffer<u8> data;
 507         bool reliable;
 508
 509         ConnectionCommand(): type(CONNCMD_NONE) {}
 510
 511         void serve(u16 port_)
 512         {
 513                 type = CONNCMD_SERVE;
 514                 port = port_;
 515         }
 516         void connect(Address address_)
 517         {
 518                 type = CONNCMD_CONNECT;
 519                 address = address_;
 520         }
 521         void disconnect()
 522         {
 523                 type = CONNCMD_DISCONNECT;
 524         }
 525         void send(u16 peer_id_, u8 channelnum_,
 526                         SharedBuffer<u8> data_, bool reliable_)
 527         {
 528                 type = CONNCMD_SEND;
 529                 peer_id = peer_id_;
 530                 channelnum = channelnum_;
 531                 data = data_;
 532                 reliable = reliable_;
 533         }
 534         void sendToAll(u8 channelnum_, SharedBuffer<u8> data_, bool reliable_)
 535         {
 536                 type = CONNCMD_SEND_TO_ALL;
 537                 channelnum = channelnum_;
 538                 data = data_;
 539                 reliable = reliable_;
 540         }
 541         void deletePeer(u16 peer_id_)
 542         {
 543                 type = CONNCMD_DELETE_PEER;
 544                 peer_id = peer_id_;
 545         }
 546 };
 547
 548 class Connection: public SimpleThread
 549 {
 550 public:
 551         Connection(u32 protocol_id, u32 max_packet_size, float timeout);
 552         Connection(u32 protocol_id, u32 max_packet_size, float timeout,
 553                         PeerHandler *peerhandler);
 554         ~Connection();
 555         void * Thread();
 556
 557         /* Interface */
 558
 559         ConnectionEvent getEvent();
 560         ConnectionEvent waitEvent(u32 timeout_ms);
 561         void putCommand(ConnectionCommand &c);
 562
 563         void SetTimeoutMs(int timeout){ m_bc_receive_timeout = timeout; }
 564         void Serve(unsigned short port);
 565         void Connect(Address address);
 566         bool Connected();
 567         void Disconnect();
 568         u32 Receive(u16 &peer_id, SharedBuffer<u8> &data);
 569         void SendToAll(u8 channelnum, SharedBuffer<u8> data, bool reliable);
 570         void Send(u16 peer_id, u8 channelnum, SharedBuffer<u8> data, bool reliable);
 571         void RunTimeouts(float dtime); // dummy
 572         u16 GetPeerID(){ return m_peer_id; }
 573         Address GetPeerAddress(u16 peer_id);
 574         float GetPeerAvgRTT(u16 peer_id);
 575         void DeletePeer(u16 peer_id);
 576
 577 private:
 578         void putEvent(ConnectionEvent &e);
 579         void processCommand(ConnectionCommand &c);
 580         void send(float dtime);
 581         void receive();
 582         void runTimeouts(float dtime);
 583         void serve(u16 port);
 584         void connect(Address address);
 585         void disconnect();
 586         void sendToAll(u8 channelnum, SharedBuffer<u8> data, bool reliable);
 587         void send(u16 peer_id, u8 channelnum, SharedBuffer<u8> data, bool reliable);
 588         void sendAsPacket(u16 peer_id, u8 channelnum,
 589                         SharedBuffer<u8> data, bool reliable);
 590         void rawSendAsPacket(u16 peer_id, u8 channelnum,
 591                         SharedBuffer<u8> data, bool reliable);
 592         void rawSend(const BufferedPacket &packet);
 593         Peer* getPeer(u16 peer_id);
 594         Peer* getPeerNoEx(u16 peer_id);
 595         std::list<Peer*> getPeers();
 596         bool getFromBuffers(u16 &peer_id, SharedBuffer<u8> &dst);
 597         // Returns next data from a buffer if possible
 598         // If found, returns true; if not, false.
 599         // If found, sets peer_id and dst
 600         bool checkIncomingBuffers(Channel *channel, u16 &peer_id,
 601                         SharedBuffer<u8> &dst);
 602         /*
 603                 Processes a packet with the basic header stripped out.
 604                 Parameters:
 605                         packetdata: Data in packet (with no base headers)
 606                         peer_id: peer id of the sender of the packet in question
 607                         channelnum: channel on which the packet was sent
 608                         reliable: true if recursing into a reliable packet
 609         */
 610         SharedBuffer<u8> processPacket(Channel *channel,
 611                         SharedBuffer<u8> packetdata, u16 peer_id,
 612                         u8 channelnum, bool reliable);
 613         bool deletePeer(u16 peer_id, bool timeout);
 614
 615         Queue<OutgoingPacket> m_outgoing_queue;
 616         MutexedQueue<ConnectionEvent> m_event_queue;
 617         MutexedQueue<ConnectionCommand> m_command_queue;
 618
 619         u32 m_protocol_id;
 620         u32 m_max_packet_size;
 621         float m_timeout;
 622         UDPSocket m_socket;
 623         u16 m_peer_id;
 624
 625         std::map<u16, Peer*> m_peers;
 626         JMutex m_peers_mutex;
 627
 628         // Backwards compatibility
 629         PeerHandler *m_bc_peerhandler;
 630         int m_bc_receive_timeout;
 631
 632         void SetPeerID(u16 id){ m_peer_id = id; }
 633         u32 GetProtocolID(){ return m_protocol_id; }
 634         void PrintInfo(std::ostream &out);
 635         void PrintInfo();
 636         std::string getDesc();
 637         u16 m_indentation;
 638 };
 639
 640 } // namespace
 641
 642 #endif
 643