src/connection.h

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