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_bloated.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"
41 class ConnectionReceiveThread;
42 class ConnectionSendThread;
44 typedef enum MTProtocols {
47 MTP_MINETEST_RELIABLE_UDP
50 #define MAX_UDP_PEERS 65535
52 #define SEQNUM_MAX 65535
54 inline bool seqnum_higher(u16 totest, u16 base)
58 if ((totest - base) > (SEQNUM_MAX/2))
64 if ((base - totest) > (SEQNUM_MAX/2))
70 inline bool seqnum_in_window(u16 seqnum, u16 next,u16 window_size)
72 u16 window_start = next;
73 u16 window_end = ( next + window_size ) % (SEQNUM_MAX+1);
75 if (window_start < window_end) {
76 return ((seqnum >= window_start) && (seqnum < window_end));
80 return ((seqnum < window_end) || (seqnum >= window_start));
83 static inline float CALC_DTIME(u64 lasttime, u64 curtime)
85 float value = ( curtime - lasttime) / 1000.0;
86 return MYMAX(MYMIN(value,0.1),0.0);
91 BufferedPacket(u8 *a_data, u32 a_size):
94 BufferedPacket(u32 a_size):
97 Buffer<u8> data; // Data of the packet, including headers
98 float time = 0.0f; // Seconds from buffering the packet or re-sending
99 float totaltime = 0.0f; // Seconds from buffering the packet
100 u64 absolute_send_time = -1;
101 Address address; // Sender or destination
102 unsigned int resend_count = 0;
105 // This adds the base headers to the data and makes a packet out of it
106 BufferedPacket makePacket(Address &address, SharedBuffer<u8> data,
107 u32 protocol_id, session_t sender_peer_id, u8 channel);
109 // Depending on size, make a TYPE_ORIGINAL or TYPE_SPLIT packet
110 // Increments split_seqnum if a split packet is made
111 void makeAutoSplitPacket(SharedBuffer<u8> data, u32 chunksize_max,
112 u16 &split_seqnum, std::list<SharedBuffer<u8>> *list);
114 // Add the TYPE_RELIABLE header to the data
115 SharedBuffer<u8> makeReliablePacket(SharedBuffer<u8> data, u16 seqnum);
117 struct IncomingSplitPacket
119 IncomingSplitPacket(u32 cc, bool r):
120 chunk_count(cc), reliable(r) {}
122 IncomingSplitPacket() = delete;
124 // Key is chunk number, value is data without headers
125 std::map<u16, SharedBuffer<u8>> chunks;
127 float time = 0.0f; // Seconds from adding
128 bool reliable = false; // If true, isn't deleted on timeout
130 bool allReceived() const
132 return (chunks.size() == chunk_count);
139 A packet is sent through a channel to a peer with a basic header:
140 TODO: Should we have a receiver_peer_id also?
143 [4] session_t sender_peer_id
147 value 0 (PEER_ID_INEXISTENT) is reserved for making new connections
148 value 1 (PEER_ID_SERVER) is reserved for server
149 these constants are defined in constants.h
151 The lower the number, the higher the priority is.
152 Only channels 0, 1 and 2 exist.
154 #define BASE_HEADER_SIZE 7
155 #define CHANNEL_COUNT 3
159 CONTROL: This is a packet used by the protocol.
160 - When this is processed, nothing is handed to the user.
164 controltype and data description:
167 CONTROLTYPE_SET_PEER_ID
168 [2] session_t peer_id_new
170 - There is no actual reply, but this can be sent in a reliable
171 packet to get a reply
174 //#define TYPE_CONTROL 0
175 #define CONTROLTYPE_ACK 0
176 #define CONTROLTYPE_SET_PEER_ID 1
177 #define CONTROLTYPE_PING 2
178 #define CONTROLTYPE_DISCO 3
181 ORIGINAL: This is a plain packet with no control and no error
183 - When this is processed, it is directly handed to the user.
187 //#define TYPE_ORIGINAL 1
188 #define ORIGINAL_HEADER_SIZE 1
190 SPLIT: These are sequences of packets forming one bigger piece of
192 - When processed and all the packet_nums 0...packet_count-1 are
193 present (this should be buffered), the resulting data shall be
194 directly handed to the user.
195 - If the data fails to come up in a reasonable time, the buffer shall
196 be silently discarded.
197 - These can be sent as-is or atop of a RELIABLE packet stream.
204 //#define TYPE_SPLIT 2
206 RELIABLE: Delivery of all RELIABLE packets shall be forced by ACKs,
207 and they shall be delivered in the same order as sent. This is done
208 with a buffer in the receiving and transmitting end.
209 - When this is processed, the contents of each packet is recursively
210 processed as packets.
216 //#define TYPE_RELIABLE 3
217 #define RELIABLE_HEADER_SIZE 3
218 #define SEQNUM_INITIAL 65500
220 enum PacketType: u8 {
221 PACKET_TYPE_CONTROL = 0,
222 PACKET_TYPE_ORIGINAL = 1,
223 PACKET_TYPE_SPLIT = 2,
224 PACKET_TYPE_RELIABLE = 3,
228 A buffer which stores reliable packets and sorts them internally
229 for fast access to the smallest one.
232 typedef std::list<BufferedPacket>::iterator RPBSearchResult;
234 class ReliablePacketBuffer
237 ReliablePacketBuffer() = default;
239 bool getFirstSeqnum(u16& result);
241 BufferedPacket popFirst();
242 BufferedPacket popSeqnum(u16 seqnum);
243 void insert(BufferedPacket &p,u16 next_expected);
245 void incrementTimeouts(float dtime);
246 std::list<BufferedPacket> getTimedOuts(float timeout,
247 unsigned int max_packets);
251 bool containsPacket(u16 seqnum);
252 RPBSearchResult notFound();
257 RPBSearchResult findPacket(u16 seqnum);
259 std::list<BufferedPacket> m_list;
262 u16 m_oldest_non_answered_ack;
264 std::mutex m_list_mutex;
268 A buffer for reconstructing split packets
271 class IncomingSplitBuffer
274 ~IncomingSplitBuffer();
276 Returns a reference counted buffer of length != 0 when a full split
277 packet is constructed. If not, returns one of length 0.
279 SharedBuffer<u8> insert(const BufferedPacket &p, bool reliable);
281 void removeUnreliableTimedOuts(float dtime, float timeout);
285 std::map<u16, IncomingSplitPacket*> m_buf;
287 std::mutex m_map_mutex;
290 struct OutgoingPacket
294 SharedBuffer<u8> data;
298 OutgoingPacket(session_t peer_id_, u8 channelnum_, const SharedBuffer<u8> &data_,
299 bool reliable_,bool ack_=false):
301 channelnum(channelnum_),
309 enum ConnectionCommandType{
314 CONNCMD_DISCONNECT_PEER,
321 struct ConnectionCommand
323 enum ConnectionCommandType type = CONNCMD_NONE;
325 session_t peer_id = PEER_ID_INEXISTENT;
328 bool reliable = false;
331 ConnectionCommand() = default;
332 ConnectionCommand &operator=(const ConnectionCommand &other)
335 address = other.address;
336 peer_id = other.peer_id;
337 channelnum = other.channelnum;
338 // We must copy the buffer here to prevent race condition
339 data = SharedBuffer<u8>(*other.data, other.data.getSize());
340 reliable = other.reliable;
345 void serve(Address address_)
347 type = CONNCMD_SERVE;
350 void connect(Address address_)
352 type = CONNCMD_CONNECT;
357 type = CONNCMD_DISCONNECT;
359 void disconnect_peer(session_t peer_id_)
361 type = CONNCMD_DISCONNECT_PEER;
365 void send(session_t peer_id_, u8 channelnum_, NetworkPacket *pkt, bool reliable_);
367 void ack(session_t peer_id_, u8 channelnum_, const SharedBuffer<u8> &data_)
371 channelnum = channelnum_;
376 void createPeer(session_t peer_id_, const SharedBuffer<u8> &data_)
378 type = CONCMD_CREATE_PEER;
387 /* maximum window size to use, 0xFFFF is theoretical maximum don't think about
388 * touching it, the less you're away from it the more likely data corruption
391 #define MAX_RELIABLE_WINDOW_SIZE 0x8000
392 /* starting value for window size */
393 #define MIN_RELIABLE_WINDOW_SIZE 0x40
399 u16 readNextIncomingSeqNum();
400 u16 incNextIncomingSeqNum();
402 u16 getOutgoingSequenceNumber(bool& successfull);
403 u16 readOutgoingSequenceNumber();
404 bool putBackSequenceNumber(u16);
406 u16 readNextSplitSeqNum();
407 void setNextSplitSeqNum(u16 seqnum);
409 // This is for buffering the incoming packets that are coming in
411 ReliablePacketBuffer incoming_reliables;
412 // This is for buffering the sent packets so that the sender can
413 // re-send them if no ACK is received
414 ReliablePacketBuffer outgoing_reliables_sent;
416 //queued reliable packets
417 std::queue<BufferedPacket> queued_reliables;
419 //queue commands prior splitting to packets
420 std::deque<ConnectionCommand> queued_commands;
422 IncomingSplitBuffer incoming_splits;
425 ~Channel() = default;
427 void UpdatePacketLossCounter(unsigned int count);
428 void UpdatePacketTooLateCounter();
429 void UpdateBytesSent(unsigned int bytes,unsigned int packages=1);
430 void UpdateBytesLost(unsigned int bytes);
431 void UpdateBytesReceived(unsigned int bytes);
433 void UpdateTimers(float dtime);
435 const float getCurrentDownloadRateKB()
436 { MutexAutoLock lock(m_internal_mutex); return cur_kbps; };
437 const float getMaxDownloadRateKB()
438 { MutexAutoLock lock(m_internal_mutex); return max_kbps; };
440 const float getCurrentLossRateKB()
441 { MutexAutoLock lock(m_internal_mutex); return cur_kbps_lost; };
442 const float getMaxLossRateKB()
443 { MutexAutoLock lock(m_internal_mutex); return max_kbps_lost; };
445 const float getCurrentIncomingRateKB()
446 { MutexAutoLock lock(m_internal_mutex); return cur_incoming_kbps; };
447 const float getMaxIncomingRateKB()
448 { MutexAutoLock lock(m_internal_mutex); return max_incoming_kbps; };
450 const float getAvgDownloadRateKB()
451 { MutexAutoLock lock(m_internal_mutex); return avg_kbps; };
452 const float getAvgLossRateKB()
453 { MutexAutoLock lock(m_internal_mutex); return avg_kbps_lost; };
454 const float getAvgIncomingRateKB()
455 { MutexAutoLock lock(m_internal_mutex); return avg_incoming_kbps; };
457 const unsigned int getWindowSize() const { return window_size; };
459 void setWindowSize(unsigned int size) { window_size = size; };
461 std::mutex m_internal_mutex;
462 int window_size = MIN_RELIABLE_WINDOW_SIZE;
464 u16 next_incoming_seqnum = SEQNUM_INITIAL;
466 u16 next_outgoing_seqnum = SEQNUM_INITIAL;
467 u16 next_outgoing_split_seqnum = SEQNUM_INITIAL;
469 unsigned int current_packet_loss = 0;
470 unsigned int current_packet_too_late = 0;
471 unsigned int current_packet_successful = 0;
472 float packet_loss_counter = 0.0f;
474 unsigned int current_bytes_transfered = 0;
475 unsigned int current_bytes_received = 0;
476 unsigned int current_bytes_lost = 0;
477 float max_kbps = 0.0f;
478 float cur_kbps = 0.0f;
479 float avg_kbps = 0.0f;
480 float max_incoming_kbps = 0.0f;
481 float cur_incoming_kbps = 0.0f;
482 float avg_incoming_kbps = 0.0f;
483 float max_kbps_lost = 0.0f;
484 float cur_kbps_lost = 0.0f;
485 float avg_kbps_lost = 0.0f;
486 float bpm_counter = 0.0f;
488 unsigned int rate_samples = 0;
496 PeerHelper() = default;
497 PeerHelper(Peer* peer);
500 PeerHelper& operator=(Peer* peer);
501 Peer* operator->() const;
503 Peer* operator&() const;
504 bool operator!=(void* ptr);
507 Peer *m_peer = nullptr;
523 friend class PeerHelper;
525 Peer(Address address_,u16 id_,Connection* connection) :
527 m_connection(connection),
529 m_last_timeout_check(porting::getTimeMs())
534 MutexAutoLock usage_lock(m_exclusive_access_mutex);
535 FATAL_ERROR_IF(m_usage != 0, "Reference counting failure");
538 // Unique id of the peer
543 virtual void PutReliableSendCommand(ConnectionCommand &c,
544 unsigned int max_packet_size) {};
546 virtual bool getAddress(MTProtocols type, Address& toset) = 0;
548 bool isPendingDeletion()
549 { MutexAutoLock lock(m_exclusive_access_mutex); return m_pending_deletion; };
552 {MutexAutoLock lock(m_exclusive_access_mutex); m_timeout_counter = 0.0; };
554 bool isTimedOut(float timeout);
556 unsigned int m_increment_packets_remaining = 9;
557 unsigned int m_increment_bytes_remaining = 0;
559 virtual u16 getNextSplitSequenceNumber(u8 channel) { return 0; };
560 virtual void setNextSplitSequenceNumber(u8 channel, u16 seqnum) {};
561 virtual SharedBuffer<u8> addSplitPacket(u8 channel, const BufferedPacket &toadd,
564 fprintf(stderr,"Peer: addSplitPacket called, this is supposed to be never called!\n");
565 return SharedBuffer<u8>(0);
568 virtual bool Ping(float dtime, SharedBuffer<u8>& data) { return false; };
570 virtual float getStat(rtt_stat_type type) const {
573 return m_rtt.min_rtt;
575 return m_rtt.max_rtt;
577 return m_rtt.avg_rtt;
579 return m_rtt.jitter_min;
581 return m_rtt.jitter_max;
583 return m_rtt.jitter_avg;
584 case TIMEOUT_COUNTER:
585 return m_timeout_counter;
590 virtual void reportRTT(float rtt) {};
592 void RTTStatistics(float rtt, const std::string &profiler_id = "");
597 std::mutex m_exclusive_access_mutex;
599 bool m_pending_deletion = false;
601 Connection* m_connection;
603 // Address of the peer
607 float m_ping_timer = 0.0f;
611 float jitter_min = FLT_MAX;
612 float jitter_max = 0.0f;
613 float jitter_avg = -1.0f;
614 float min_rtt = FLT_MAX;
615 float max_rtt = 0.0f;
616 float avg_rtt = -1.0f;
618 rttstats() = default;
622 float m_last_rtt = -1.0f;
624 // current usage count
625 unsigned int m_usage = 0;
627 // Seconds from last receive
628 float m_timeout_counter = 0.0f;
630 u64 m_last_timeout_check;
633 class UDPPeer : public Peer
637 friend class PeerHelper;
638 friend class ConnectionReceiveThread;
639 friend class ConnectionSendThread;
640 friend class Connection;
642 UDPPeer(u16 a_id, Address a_address, Connection* connection);
643 virtual ~UDPPeer() = default;
645 void PutReliableSendCommand(ConnectionCommand &c,
646 unsigned int max_packet_size);
648 bool getAddress(MTProtocols type, Address& toset);
650 u16 getNextSplitSequenceNumber(u8 channel);
651 void setNextSplitSequenceNumber(u8 channel, u16 seqnum);
653 SharedBuffer<u8> addSplitPacket(u8 channel, const BufferedPacket &toadd,
658 Calculates avg_rtt and resend_timeout.
659 rtt=-1 only recalculates resend_timeout
661 void reportRTT(float rtt);
663 void RunCommandQueues(
664 unsigned int max_packet_size,
665 unsigned int maxcommands,
666 unsigned int maxtransfer);
668 float getResendTimeout()
669 { MutexAutoLock lock(m_exclusive_access_mutex); return resend_timeout; }
671 void setResendTimeout(float timeout)
672 { MutexAutoLock lock(m_exclusive_access_mutex); resend_timeout = timeout; }
673 bool Ping(float dtime,SharedBuffer<u8>& data);
675 Channel channels[CHANNEL_COUNT];
676 bool m_pending_disconnect = false;
678 // This is changed dynamically
679 float resend_timeout = 0.5;
681 bool processReliableSendCommand(
682 ConnectionCommand &c,
683 unsigned int max_packet_size);
690 enum ConnectionEventType{
692 CONNEVENT_DATA_RECEIVED,
693 CONNEVENT_PEER_ADDED,
694 CONNEVENT_PEER_REMOVED,
695 CONNEVENT_BIND_FAILED,
698 struct ConnectionEvent
700 enum ConnectionEventType type = CONNEVENT_NONE;
701 session_t peer_id = 0;
703 bool timeout = false;
706 ConnectionEvent() = default;
708 std::string describe()
712 return "CONNEVENT_NONE";
713 case CONNEVENT_DATA_RECEIVED:
714 return "CONNEVENT_DATA_RECEIVED";
715 case CONNEVENT_PEER_ADDED:
716 return "CONNEVENT_PEER_ADDED";
717 case CONNEVENT_PEER_REMOVED:
718 return "CONNEVENT_PEER_REMOVED";
719 case CONNEVENT_BIND_FAILED:
720 return "CONNEVENT_BIND_FAILED";
722 return "Invalid ConnectionEvent";
725 void dataReceived(session_t peer_id_, const SharedBuffer<u8> &data_)
727 type = CONNEVENT_DATA_RECEIVED;
731 void peerAdded(session_t peer_id_, Address address_)
733 type = CONNEVENT_PEER_ADDED;
737 void peerRemoved(session_t peer_id_, bool timeout_, Address address_)
739 type = CONNEVENT_PEER_REMOVED;
746 type = CONNEVENT_BIND_FAILED;
755 friend class ConnectionSendThread;
756 friend class ConnectionReceiveThread;
758 Connection(u32 protocol_id, u32 max_packet_size, float timeout, bool ipv6,
759 PeerHandler *peerhandler);
763 ConnectionEvent waitEvent(u32 timeout_ms);
764 void putCommand(ConnectionCommand &c);
766 void SetTimeoutMs(u32 timeout) { m_bc_receive_timeout = timeout; }
767 void Serve(Address bind_addr);
768 void Connect(Address address);
771 void Receive(NetworkPacket* pkt);
772 void Send(session_t peer_id, u8 channelnum, NetworkPacket *pkt, bool reliable);
773 session_t GetPeerID() const { return m_peer_id; }
774 Address GetPeerAddress(session_t peer_id);
775 float getPeerStat(session_t peer_id, rtt_stat_type type);
776 float getLocalStat(rate_stat_type type);
777 const u32 GetProtocolID() const { return m_protocol_id; };
778 const std::string getDesc();
779 void DisconnectPeer(session_t peer_id);
782 PeerHelper getPeerNoEx(session_t peer_id);
783 u16 lookupPeer(Address& sender);
785 u16 createPeer(Address& sender, MTProtocols protocol, int fd);
786 UDPPeer* createServerPeer(Address& sender);
787 bool deletePeer(session_t peer_id, bool timeout);
789 void SetPeerID(session_t id) { m_peer_id = id; }
791 void sendAck(session_t peer_id, u8 channelnum, u16 seqnum);
793 void PrintInfo(std::ostream &out);
795 std::list<session_t> getPeerIDs()
797 MutexAutoLock peerlock(m_peers_mutex);
801 UDPSocket m_udpSocket;
802 MutexedQueue<ConnectionCommand> m_command_queue;
804 void putEvent(ConnectionEvent &e);
808 MutexedQueue<ConnectionEvent> m_event_queue;
810 session_t m_peer_id = 0;
813 std::map<session_t, Peer *> m_peers;
814 std::list<session_t> m_peer_ids;
815 std::mutex m_peers_mutex;
817 std::unique_ptr<ConnectionSendThread> m_sendThread;
818 std::unique_ptr<ConnectionReceiveThread> m_receiveThread;
820 std::mutex m_info_mutex;
822 // Backwards compatibility
823 PeerHandler *m_bc_peerhandler;
824 u32 m_bc_receive_timeout = 0;
826 bool m_shutting_down = false;
828 session_t m_next_remote_peer_id = 2;