3 Copyright (C) 2013 celeron55, Perttu Ahola <celeron55@gmail.com>
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.
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.
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.
22 #include "irrlichttypes.h"
23 #include "peerhandler.h"
25 #include "constants.h"
26 #include "util/pointer.h"
27 #include "util/container.h"
28 #include "util/thread.h"
29 #include "util/numeric.h"
30 #include "networkprotocol.h"
40 class ConnectionReceiveThread;
41 class ConnectionSendThread;
43 typedef enum MTProtocols {
46 MTP_MINETEST_RELIABLE_UDP
49 #define MAX_UDP_PEERS 65535
51 #define SEQNUM_MAX 65535
53 inline bool seqnum_higher(u16 totest, u16 base)
57 if ((totest - base) > (SEQNUM_MAX/2))
63 if ((base - totest) > (SEQNUM_MAX/2))
69 inline bool seqnum_in_window(u16 seqnum, u16 next,u16 window_size)
71 u16 window_start = next;
72 u16 window_end = ( next + window_size ) % (SEQNUM_MAX+1);
74 if (window_start < window_end) {
75 return ((seqnum >= window_start) && (seqnum < window_end));
79 return ((seqnum < window_end) || (seqnum >= window_start));
82 static inline float CALC_DTIME(u64 lasttime, u64 curtime)
84 float value = ( curtime - lasttime) / 1000.0;
85 return MYMAX(MYMIN(value,0.1),0.0);
90 BufferedPacket(u8 *a_data, u32 a_size):
93 BufferedPacket(u32 a_size):
96 Buffer<u8> data; // Data of the packet, including headers
97 float time = 0.0f; // Seconds from buffering the packet or re-sending
98 float totaltime = 0.0f; // Seconds from buffering the packet
99 u64 absolute_send_time = -1;
100 Address address; // Sender or destination
101 unsigned int resend_count = 0;
104 // This adds the base headers to the data and makes a packet out of it
105 BufferedPacket makePacket(Address &address, const SharedBuffer<u8> &data,
106 u32 protocol_id, session_t sender_peer_id, u8 channel);
108 // Depending on size, make a TYPE_ORIGINAL or TYPE_SPLIT packet
109 // Increments split_seqnum if a split packet is made
110 void makeAutoSplitPacket(const SharedBuffer<u8> &data, u32 chunksize_max,
111 u16 &split_seqnum, std::list<SharedBuffer<u8>> *list);
113 // Add the TYPE_RELIABLE header to the data
114 SharedBuffer<u8> makeReliablePacket(const SharedBuffer<u8> &data, u16 seqnum);
116 struct IncomingSplitPacket
118 IncomingSplitPacket(u32 cc, bool r):
119 chunk_count(cc), reliable(r) {}
121 IncomingSplitPacket() = delete;
123 float time = 0.0f; // Seconds from adding
125 bool reliable; // If true, isn't deleted on timeout
127 bool allReceived() const
129 return (chunks.size() == chunk_count);
131 bool insert(u32 chunk_num, SharedBuffer<u8> &chunkdata);
132 SharedBuffer<u8> reassemble();
135 // Key is chunk number, value is data without headers
136 std::map<u16, SharedBuffer<u8>> chunks;
142 A packet is sent through a channel to a peer with a basic header:
145 [4] session_t sender_peer_id
149 value 0 (PEER_ID_INEXISTENT) is reserved for making new connections
150 value 1 (PEER_ID_SERVER) is reserved for server
151 these constants are defined in constants.h
153 Channel numbers have no intrinsic meaning. Currently only 0, 1, 2 exist.
155 #define BASE_HEADER_SIZE 7
156 #define CHANNEL_COUNT 3
160 CONTROL: This is a packet used by the protocol.
161 - When this is processed, nothing is handed to the user.
165 controltype and data description:
168 CONTROLTYPE_SET_PEER_ID
169 [2] session_t peer_id_new
171 - There is no actual reply, but this can be sent in a reliable
172 packet to get a reply
175 //#define TYPE_CONTROL 0
176 #define CONTROLTYPE_ACK 0
177 #define CONTROLTYPE_SET_PEER_ID 1
178 #define CONTROLTYPE_PING 2
179 #define CONTROLTYPE_DISCO 3
182 ORIGINAL: This is a plain packet with no control and no error
184 - When this is processed, it is directly handed to the user.
188 //#define TYPE_ORIGINAL 1
189 #define ORIGINAL_HEADER_SIZE 1
191 SPLIT: These are sequences of packets forming one bigger piece of
193 - When processed and all the packet_nums 0...packet_count-1 are
194 present (this should be buffered), the resulting data shall be
195 directly handed to the user.
196 - If the data fails to come up in a reasonable time, the buffer shall
197 be silently discarded.
198 - These can be sent as-is or atop of a RELIABLE packet stream.
205 //#define TYPE_SPLIT 2
207 RELIABLE: Delivery of all RELIABLE packets shall be forced by ACKs,
208 and they shall be delivered in the same order as sent. This is done
209 with a buffer in the receiving and transmitting end.
210 - When this is processed, the contents of each packet is recursively
211 processed as packets.
217 //#define TYPE_RELIABLE 3
218 #define RELIABLE_HEADER_SIZE 3
219 #define SEQNUM_INITIAL 65500
221 enum PacketType: u8 {
222 PACKET_TYPE_CONTROL = 0,
223 PACKET_TYPE_ORIGINAL = 1,
224 PACKET_TYPE_SPLIT = 2,
225 PACKET_TYPE_RELIABLE = 3,
229 A buffer which stores reliable packets and sorts them internally
230 for fast access to the smallest one.
233 typedef std::list<BufferedPacket>::iterator RPBSearchResult;
235 class ReliablePacketBuffer
238 ReliablePacketBuffer() = default;
240 bool getFirstSeqnum(u16& result);
242 BufferedPacket popFirst();
243 BufferedPacket popSeqnum(u16 seqnum);
244 void insert(const BufferedPacket &p, u16 next_expected);
246 void incrementTimeouts(float dtime);
247 std::list<BufferedPacket> getTimedOuts(float timeout, u32 max_packets);
255 RPBSearchResult findPacket(u16 seqnum); // does not perform locking
256 inline RPBSearchResult notFound() { return m_list.end(); }
258 std::list<BufferedPacket> m_list;
260 u16 m_oldest_non_answered_ack;
262 std::mutex m_list_mutex;
266 A buffer for reconstructing split packets
269 class IncomingSplitBuffer
272 ~IncomingSplitBuffer();
274 Returns a reference counted buffer of length != 0 when a full split
275 packet is constructed. If not, returns one of length 0.
277 SharedBuffer<u8> insert(const BufferedPacket &p, bool reliable);
279 void removeUnreliableTimedOuts(float dtime, float timeout);
283 std::map<u16, IncomingSplitPacket*> m_buf;
285 std::mutex m_map_mutex;
288 struct OutgoingPacket
292 SharedBuffer<u8> data;
296 OutgoingPacket(session_t peer_id_, u8 channelnum_, const SharedBuffer<u8> &data_,
297 bool reliable_,bool ack_=false):
299 channelnum(channelnum_),
307 enum ConnectionCommandType{
312 CONNCMD_DISCONNECT_PEER,
319 struct ConnectionCommand
321 enum ConnectionCommandType type = CONNCMD_NONE;
323 session_t peer_id = PEER_ID_INEXISTENT;
326 bool reliable = false;
329 ConnectionCommand() = default;
331 void serve(Address address_)
333 type = CONNCMD_SERVE;
336 void connect(Address address_)
338 type = CONNCMD_CONNECT;
343 type = CONNCMD_DISCONNECT;
345 void disconnect_peer(session_t peer_id_)
347 type = CONNCMD_DISCONNECT_PEER;
351 void send(session_t peer_id_, u8 channelnum_, NetworkPacket *pkt, bool reliable_);
353 void ack(session_t peer_id_, u8 channelnum_, const Buffer<u8> &data_)
357 channelnum = channelnum_;
362 void createPeer(session_t peer_id_, const Buffer<u8> &data_)
364 type = CONCMD_CREATE_PEER;
373 /* maximum window size to use, 0xFFFF is theoretical maximum. don't think about
374 * touching it, the less you're away from it the more likely data corruption
377 #define MAX_RELIABLE_WINDOW_SIZE 0x8000
378 /* starting value for window size */
379 #define START_RELIABLE_WINDOW_SIZE 0x400
380 /* minimum value for window size */
381 #define MIN_RELIABLE_WINDOW_SIZE 0x40
387 u16 readNextIncomingSeqNum();
388 u16 incNextIncomingSeqNum();
390 u16 getOutgoingSequenceNumber(bool& successfull);
391 u16 readOutgoingSequenceNumber();
392 bool putBackSequenceNumber(u16);
394 u16 readNextSplitSeqNum();
395 void setNextSplitSeqNum(u16 seqnum);
397 // This is for buffering the incoming packets that are coming in
399 ReliablePacketBuffer incoming_reliables;
400 // This is for buffering the sent packets so that the sender can
401 // re-send them if no ACK is received
402 ReliablePacketBuffer outgoing_reliables_sent;
404 //queued reliable packets
405 std::queue<BufferedPacket> queued_reliables;
407 //queue commands prior splitting to packets
408 std::deque<ConnectionCommand> queued_commands;
410 IncomingSplitBuffer incoming_splits;
413 ~Channel() = default;
415 void UpdatePacketLossCounter(unsigned int count);
416 void UpdatePacketTooLateCounter();
417 void UpdateBytesSent(unsigned int bytes,unsigned int packages=1);
418 void UpdateBytesLost(unsigned int bytes);
419 void UpdateBytesReceived(unsigned int bytes);
421 void UpdateTimers(float dtime);
423 const float getCurrentDownloadRateKB()
424 { MutexAutoLock lock(m_internal_mutex); return cur_kbps; };
425 const float getMaxDownloadRateKB()
426 { MutexAutoLock lock(m_internal_mutex); return max_kbps; };
428 const float getCurrentLossRateKB()
429 { MutexAutoLock lock(m_internal_mutex); return cur_kbps_lost; };
430 const float getMaxLossRateKB()
431 { MutexAutoLock lock(m_internal_mutex); return max_kbps_lost; };
433 const float getCurrentIncomingRateKB()
434 { MutexAutoLock lock(m_internal_mutex); return cur_incoming_kbps; };
435 const float getMaxIncomingRateKB()
436 { MutexAutoLock lock(m_internal_mutex); return max_incoming_kbps; };
438 const float getAvgDownloadRateKB()
439 { MutexAutoLock lock(m_internal_mutex); return avg_kbps; };
440 const float getAvgLossRateKB()
441 { MutexAutoLock lock(m_internal_mutex); return avg_kbps_lost; };
442 const float getAvgIncomingRateKB()
443 { MutexAutoLock lock(m_internal_mutex); return avg_incoming_kbps; };
445 const unsigned int getWindowSize() const { return window_size; };
447 void setWindowSize(unsigned int size) { window_size = size; };
449 std::mutex m_internal_mutex;
450 int window_size = MIN_RELIABLE_WINDOW_SIZE;
452 u16 next_incoming_seqnum = SEQNUM_INITIAL;
454 u16 next_outgoing_seqnum = SEQNUM_INITIAL;
455 u16 next_outgoing_split_seqnum = SEQNUM_INITIAL;
457 unsigned int current_packet_loss = 0;
458 unsigned int current_packet_too_late = 0;
459 unsigned int current_packet_successful = 0;
460 float packet_loss_counter = 0.0f;
462 unsigned int current_bytes_transfered = 0;
463 unsigned int current_bytes_received = 0;
464 unsigned int current_bytes_lost = 0;
465 float max_kbps = 0.0f;
466 float cur_kbps = 0.0f;
467 float avg_kbps = 0.0f;
468 float max_incoming_kbps = 0.0f;
469 float cur_incoming_kbps = 0.0f;
470 float avg_incoming_kbps = 0.0f;
471 float max_kbps_lost = 0.0f;
472 float cur_kbps_lost = 0.0f;
473 float avg_kbps_lost = 0.0f;
474 float bpm_counter = 0.0f;
476 unsigned int rate_samples = 0;
484 PeerHelper() = default;
485 PeerHelper(Peer* peer);
488 PeerHelper& operator=(Peer* peer);
489 Peer* operator->() const;
491 Peer* operator&() const;
492 bool operator!=(void* ptr);
495 Peer *m_peer = nullptr;
511 friend class PeerHelper;
513 Peer(Address address_,u16 id_,Connection* connection) :
515 m_connection(connection),
517 m_last_timeout_check(porting::getTimeMs())
522 MutexAutoLock usage_lock(m_exclusive_access_mutex);
523 FATAL_ERROR_IF(m_usage != 0, "Reference counting failure");
526 // Unique id of the peer
531 virtual void PutReliableSendCommand(ConnectionCommand &c,
532 unsigned int max_packet_size) {};
534 virtual bool getAddress(MTProtocols type, Address& toset) = 0;
536 bool isPendingDeletion()
537 { MutexAutoLock lock(m_exclusive_access_mutex); return m_pending_deletion; };
540 {MutexAutoLock lock(m_exclusive_access_mutex); m_timeout_counter = 0.0; };
542 bool isTimedOut(float timeout);
544 unsigned int m_increment_packets_remaining = 0;
546 virtual u16 getNextSplitSequenceNumber(u8 channel) { return 0; };
547 virtual void setNextSplitSequenceNumber(u8 channel, u16 seqnum) {};
548 virtual SharedBuffer<u8> addSplitPacket(u8 channel, const BufferedPacket &toadd,
551 errorstream << "Peer::addSplitPacket called,"
552 << " this is supposed to be never called!" << std::endl;
553 return SharedBuffer<u8>(0);
556 virtual bool Ping(float dtime, SharedBuffer<u8>& data) { return false; };
558 virtual float getStat(rtt_stat_type type) const {
561 return m_rtt.min_rtt;
563 return m_rtt.max_rtt;
565 return m_rtt.avg_rtt;
567 return m_rtt.jitter_min;
569 return m_rtt.jitter_max;
571 return m_rtt.jitter_avg;
576 virtual void reportRTT(float rtt) {};
578 void RTTStatistics(float rtt,
579 const std::string &profiler_id = "",
580 unsigned int num_samples = 1000);
585 std::mutex m_exclusive_access_mutex;
587 bool m_pending_deletion = false;
589 Connection* m_connection;
591 // Address of the peer
595 float m_ping_timer = 0.0f;
599 float jitter_min = FLT_MAX;
600 float jitter_max = 0.0f;
601 float jitter_avg = -1.0f;
602 float min_rtt = FLT_MAX;
603 float max_rtt = 0.0f;
604 float avg_rtt = -1.0f;
606 rttstats() = default;
610 float m_last_rtt = -1.0f;
612 // current usage count
613 unsigned int m_usage = 0;
615 // Seconds from last receive
616 float m_timeout_counter = 0.0f;
618 u64 m_last_timeout_check;
621 class UDPPeer : public Peer
625 friend class PeerHelper;
626 friend class ConnectionReceiveThread;
627 friend class ConnectionSendThread;
628 friend class Connection;
630 UDPPeer(u16 a_id, Address a_address, Connection* connection);
631 virtual ~UDPPeer() = default;
633 void PutReliableSendCommand(ConnectionCommand &c,
634 unsigned int max_packet_size);
636 bool getAddress(MTProtocols type, Address& toset);
638 u16 getNextSplitSequenceNumber(u8 channel);
639 void setNextSplitSequenceNumber(u8 channel, u16 seqnum);
641 SharedBuffer<u8> addSplitPacket(u8 channel, const BufferedPacket &toadd,
646 Calculates avg_rtt and resend_timeout.
647 rtt=-1 only recalculates resend_timeout
649 void reportRTT(float rtt);
651 void RunCommandQueues(
652 unsigned int max_packet_size,
653 unsigned int maxcommands,
654 unsigned int maxtransfer);
656 float getResendTimeout()
657 { MutexAutoLock lock(m_exclusive_access_mutex); return resend_timeout; }
659 void setResendTimeout(float timeout)
660 { MutexAutoLock lock(m_exclusive_access_mutex); resend_timeout = timeout; }
661 bool Ping(float dtime,SharedBuffer<u8>& data);
663 Channel channels[CHANNEL_COUNT];
664 bool m_pending_disconnect = false;
666 // This is changed dynamically
667 float resend_timeout = 0.5;
669 bool processReliableSendCommand(
670 ConnectionCommand &c,
671 unsigned int max_packet_size);
678 enum ConnectionEventType{
680 CONNEVENT_DATA_RECEIVED,
681 CONNEVENT_PEER_ADDED,
682 CONNEVENT_PEER_REMOVED,
683 CONNEVENT_BIND_FAILED,
686 struct ConnectionEvent
688 enum ConnectionEventType type = CONNEVENT_NONE;
689 session_t peer_id = 0;
691 bool timeout = false;
694 ConnectionEvent() = default;
696 const char *describe() const
700 return "CONNEVENT_NONE";
701 case CONNEVENT_DATA_RECEIVED:
702 return "CONNEVENT_DATA_RECEIVED";
703 case CONNEVENT_PEER_ADDED:
704 return "CONNEVENT_PEER_ADDED";
705 case CONNEVENT_PEER_REMOVED:
706 return "CONNEVENT_PEER_REMOVED";
707 case CONNEVENT_BIND_FAILED:
708 return "CONNEVENT_BIND_FAILED";
710 return "Invalid ConnectionEvent";
713 void dataReceived(session_t peer_id_, const Buffer<u8> &data_)
715 type = CONNEVENT_DATA_RECEIVED;
719 void peerAdded(session_t peer_id_, Address address_)
721 type = CONNEVENT_PEER_ADDED;
725 void peerRemoved(session_t peer_id_, bool timeout_, Address address_)
727 type = CONNEVENT_PEER_REMOVED;
734 type = CONNEVENT_BIND_FAILED;
743 friend class ConnectionSendThread;
744 friend class ConnectionReceiveThread;
746 Connection(u32 protocol_id, u32 max_packet_size, float timeout, bool ipv6,
747 PeerHandler *peerhandler);
751 ConnectionEvent waitEvent(u32 timeout_ms);
752 // Warning: creates an unnecessary copy, prefer putCommand(T&&) if possible
753 void putCommand(const ConnectionCommand &c);
754 void putCommand(ConnectionCommand &&c);
756 void SetTimeoutMs(u32 timeout) { m_bc_receive_timeout = timeout; }
757 void Serve(Address bind_addr);
758 void Connect(Address address);
761 void Receive(NetworkPacket* pkt);
762 bool TryReceive(NetworkPacket *pkt);
763 void Send(session_t peer_id, u8 channelnum, NetworkPacket *pkt, bool reliable);
764 session_t GetPeerID() const { return m_peer_id; }
765 Address GetPeerAddress(session_t peer_id);
766 float getPeerStat(session_t peer_id, rtt_stat_type type);
767 float getLocalStat(rate_stat_type type);
768 const u32 GetProtocolID() const { return m_protocol_id; };
769 const std::string getDesc();
770 void DisconnectPeer(session_t peer_id);
773 PeerHelper getPeerNoEx(session_t peer_id);
774 u16 lookupPeer(Address& sender);
776 u16 createPeer(Address& sender, MTProtocols protocol, int fd);
777 UDPPeer* createServerPeer(Address& sender);
778 bool deletePeer(session_t peer_id, bool timeout);
780 void SetPeerID(session_t id) { m_peer_id = id; }
782 void sendAck(session_t peer_id, u8 channelnum, u16 seqnum);
784 void PrintInfo(std::ostream &out);
786 std::vector<session_t> getPeerIDs()
788 MutexAutoLock peerlock(m_peers_mutex);
792 UDPSocket m_udpSocket;
793 // Command queue: user -> SendThread
794 MutexedQueue<ConnectionCommand> m_command_queue;
796 bool Receive(NetworkPacket *pkt, u32 timeout);
798 // Warning: creates an unnecessary copy, prefer putEvent(T&&) if possible
799 void putEvent(const ConnectionEvent &e);
800 void putEvent(ConnectionEvent &&e);
804 bool ConnectedToServer()
806 return getPeerNoEx(PEER_ID_SERVER) != nullptr;
809 // Event queue: ReceiveThread -> user
810 MutexedQueue<ConnectionEvent> m_event_queue;
812 session_t m_peer_id = 0;
815 std::map<session_t, Peer *> m_peers;
816 std::vector<session_t> m_peer_ids;
817 std::mutex m_peers_mutex;
819 std::unique_ptr<ConnectionSendThread> m_sendThread;
820 std::unique_ptr<ConnectionReceiveThread> m_receiveThread;
822 std::mutex m_info_mutex;
824 // Backwards compatibility
825 PeerHandler *m_bc_peerhandler;
826 u32 m_bc_receive_timeout = 0;
828 bool m_shutting_down = false;
830 session_t m_next_remote_peer_id = 2;