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 "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 private:
 399 };
 400
 401 /*
 402         Connection
 403 */
 404
 405 struct OutgoingPacket
 406 {
 407         u16 peer_id;
 408         u8 channelnum;
 409         SharedBuffer<u8> data;
 410         bool reliable;
 411
 412         OutgoingPacket(u16 peer_id_, u8 channelnum_, SharedBuffer<u8> data_,
 413                         bool reliable_):
 414                 peer_id(peer_id_),
 415                 channelnum(channelnum_),
 416                 data(data_),
 417                 reliable(reliable_)
 418         {
 419         }
 420 };
 421
 422 enum ConnectionEventType{
 423         CONNEVENT_NONE,
 424         CONNEVENT_DATA_RECEIVED,
 425         CONNEVENT_PEER_ADDED,
 426         CONNEVENT_PEER_REMOVED,
 427         CONNEVENT_BIND_FAILED,
 428 };
 429
 430 struct ConnectionEvent
 431 {
 432         enum ConnectionEventType type;
 433         u16 peer_id;
 434         Buffer<u8> data;
 435         bool timeout;
 436         Address address;
 437
 438         ConnectionEvent(): type(CONNEVENT_NONE) {}
 439
 440         std::string describe()
 441         {
 442                 switch(type){
 443                 case CONNEVENT_NONE:
 444                         return "CONNEVENT_NONE";
 445                 case CONNEVENT_DATA_RECEIVED:
 446                         return "CONNEVENT_DATA_RECEIVED";
 447                 case CONNEVENT_PEER_ADDED:
 448                         return "CONNEVENT_PEER_ADDED";
 449                 case CONNEVENT_PEER_REMOVED:
 450                         return "CONNEVENT_PEER_REMOVED";
 451                 case CONNEVENT_BIND_FAILED:
 452                         return "CONNEVENT_BIND_FAILED";
 453                 }
 454                 return "Invalid ConnectionEvent";
 455         }
 456
 457         void dataReceived(u16 peer_id_, SharedBuffer<u8> data_)
 458         {
 459                 type = CONNEVENT_DATA_RECEIVED;
 460                 peer_id = peer_id_;
 461                 data = data_;
 462         }
 463         void peerAdded(u16 peer_id_, Address address_)
 464         {
 465                 type = CONNEVENT_PEER_ADDED;
 466                 peer_id = peer_id_;
 467                 address = address_;
 468         }
 469         void peerRemoved(u16 peer_id_, bool timeout_, Address address_)
 470         {
 471                 type = CONNEVENT_PEER_REMOVED;
 472                 peer_id = peer_id_;
 473                 timeout = timeout_;
 474                 address = address_;
 475         }
 476         void bindFailed()
 477         {
 478                 type = CONNEVENT_BIND_FAILED;
 479         }
 480 };
 481
 482 enum ConnectionCommandType{
 483         CONNCMD_NONE,
 484         CONNCMD_SERVE,
 485         CONNCMD_CONNECT,
 486         CONNCMD_DISCONNECT,
 487         CONNCMD_SEND,
 488         CONNCMD_SEND_TO_ALL,
 489         CONNCMD_DELETE_PEER,
 490 };
 491
 492 struct ConnectionCommand
 493 {
 494         enum ConnectionCommandType type;
 495         u16 port;
 496         Address address;
 497         u16 peer_id;
 498         u8 channelnum;
 499         Buffer<u8> data;
 500         bool reliable;
 501
 502         ConnectionCommand(): type(CONNCMD_NONE) {}
 503
 504         void serve(u16 port_)
 505         {
 506                 type = CONNCMD_SERVE;
 507                 port = port_;
 508         }
 509         void connect(Address address_)
 510         {
 511                 type = CONNCMD_CONNECT;
 512                 address = address_;
 513         }
 514         void disconnect()
 515         {
 516                 type = CONNCMD_DISCONNECT;
 517         }
 518         void send(u16 peer_id_, u8 channelnum_,
 519                         SharedBuffer<u8> data_, bool reliable_)
 520         {
 521                 type = CONNCMD_SEND;
 522                 peer_id = peer_id_;
 523                 channelnum = channelnum_;
 524                 data = data_;
 525                 reliable = reliable_;
 526         }
 527         void sendToAll(u8 channelnum_, SharedBuffer<u8> data_, bool reliable_)
 528         {
 529                 type = CONNCMD_SEND_TO_ALL;
 530                 channelnum = channelnum_;
 531                 data = data_;
 532                 reliable = reliable_;
 533         }
 534         void deletePeer(u16 peer_id_)
 535         {
 536                 type = CONNCMD_DELETE_PEER;
 537                 peer_id = peer_id_;
 538         }
 539 };
 540
 541 class Connection: public SimpleThread
 542 {
 543 public:
 544         Connection(u32 protocol_id, u32 max_packet_size, float timeout);
 545         Connection(u32 protocol_id, u32 max_packet_size, float timeout,
 546                         PeerHandler *peerhandler);
 547         ~Connection();
 548         void * Thread();
 549
 550         /* Interface */
 551
 552         ConnectionEvent getEvent();
 553         ConnectionEvent waitEvent(u32 timeout_ms);
 554         void putCommand(ConnectionCommand &c);
 555
 556         void SetTimeoutMs(int timeout){ m_bc_receive_timeout = timeout; }
 557         void Serve(unsigned short port);
 558         void Connect(Address address);
 559         bool Connected();
 560         void Disconnect();
 561         u32 Receive(u16 &peer_id, SharedBuffer<u8> &data);
 562         void SendToAll(u8 channelnum, SharedBuffer<u8> data, bool reliable);
 563         void Send(u16 peer_id, u8 channelnum, SharedBuffer<u8> data, bool reliable);
 564         void RunTimeouts(float dtime); // dummy
 565         u16 GetPeerID(){ return m_peer_id; }
 566         Address GetPeerAddress(u16 peer_id);
 567         float GetPeerAvgRTT(u16 peer_id);
 568         void DeletePeer(u16 peer_id);
 569
 570 private:
 571         void putEvent(ConnectionEvent &e);
 572         void processCommand(ConnectionCommand &c);
 573         void send(float dtime);
 574         void receive();
 575         void runTimeouts(float dtime);
 576         void serve(u16 port);
 577         void connect(Address address);
 578         void disconnect();
 579         void sendToAll(u8 channelnum, SharedBuffer<u8> data, bool reliable);
 580         void send(u16 peer_id, u8 channelnum, SharedBuffer<u8> data, bool reliable);
 581         void sendAsPacket(u16 peer_id, u8 channelnum,
 582                         SharedBuffer<u8> data, bool reliable);
 583         void rawSendAsPacket(u16 peer_id, u8 channelnum,
 584                         SharedBuffer<u8> data, bool reliable);
 585         void rawSend(const BufferedPacket &packet);
 586         Peer* getPeer(u16 peer_id);
 587         Peer* getPeerNoEx(u16 peer_id);
 588         core::list<Peer*> getPeers();
 589         bool getFromBuffers(u16 &peer_id, SharedBuffer<u8> &dst);
 590         // Returns next data from a buffer if possible
 591         // If found, returns true; if not, false.
 592         // If found, sets peer_id and dst
 593         bool checkIncomingBuffers(Channel *channel, u16 &peer_id,
 594                         SharedBuffer<u8> &dst);
 595         /*
 596                 Processes a packet with the basic header stripped out.
 597                 Parameters:
 598                         packetdata: Data in packet (with no base headers)
 599                         peer_id: peer id of the sender of the packet in question
 600                         channelnum: channel on which the packet was sent
 601                         reliable: true if recursing into a reliable packet
 602         */
 603         SharedBuffer<u8> processPacket(Channel *channel,
 604                         SharedBuffer<u8> packetdata, u16 peer_id,
 605                         u8 channelnum, bool reliable);
 606         bool deletePeer(u16 peer_id, bool timeout);
 607
 608         Queue<OutgoingPacket> m_outgoing_queue;
 609         MutexedQueue<ConnectionEvent> m_event_queue;
 610         MutexedQueue<ConnectionCommand> m_command_queue;
 611
 612         u32 m_protocol_id;
 613         u32 m_max_packet_size;
 614         float m_timeout;
 615         UDPSocket m_socket;
 616         u16 m_peer_id;
 617
 618         core::map<u16, Peer*> m_peers;
 619         JMutex m_peers_mutex;
 620
 621         // Backwards compatibility
 622         PeerHandler *m_bc_peerhandler;
 623         int m_bc_receive_timeout;
 624
 625         void SetPeerID(u16 id){ m_peer_id = id; }
 626         u32 GetProtocolID(){ return m_protocol_id; }
 627         void PrintInfo(std::ostream &out);
 628         void PrintInfo();
 629         std::string getDesc();
 630         u16 m_indentation;
 631 };
 632
 633 } // namespace
 634
 635 #endif
 636