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