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 float getCurrentDownloadRateKB()
424 { MutexAutoLock lock(m_internal_mutex); return cur_kbps; };
425 float getMaxDownloadRateKB()
426 { MutexAutoLock lock(m_internal_mutex); return max_kbps; };
428 float getCurrentLossRateKB()
429 { MutexAutoLock lock(m_internal_mutex); return cur_kbps_lost; };
430 float getMaxLossRateKB()
431 { MutexAutoLock lock(m_internal_mutex); return max_kbps_lost; };
433 float getCurrentIncomingRateKB()
434 { MutexAutoLock lock(m_internal_mutex); return cur_incoming_kbps; };
435 float getMaxIncomingRateKB()
436 { MutexAutoLock lock(m_internal_mutex); return max_incoming_kbps; };
438 float getAvgDownloadRateKB()
439 { MutexAutoLock lock(m_internal_mutex); return avg_kbps; };
440 float getAvgLossRateKB()
441 { MutexAutoLock lock(m_internal_mutex); return avg_kbps_lost; };
442 float getAvgIncomingRateKB()
443 { MutexAutoLock lock(m_internal_mutex); return avg_incoming_kbps; };
445 u16 getWindowSize() const { return m_window_size; };
447 void setWindowSize(long size)
449 m_window_size = (u16)rangelim(size, MIN_RELIABLE_WINDOW_SIZE, MAX_RELIABLE_WINDOW_SIZE);
453 std::mutex m_internal_mutex;
454 u16 m_window_size = MIN_RELIABLE_WINDOW_SIZE;
456 u16 next_incoming_seqnum = SEQNUM_INITIAL;
458 u16 next_outgoing_seqnum = SEQNUM_INITIAL;
459 u16 next_outgoing_split_seqnum = SEQNUM_INITIAL;
461 unsigned int current_packet_loss = 0;
462 unsigned int current_packet_too_late = 0;
463 unsigned int current_packet_successful = 0;
464 float packet_loss_counter = 0.0f;
466 unsigned int current_bytes_transfered = 0;
467 unsigned int current_bytes_received = 0;
468 unsigned int current_bytes_lost = 0;
469 float max_kbps = 0.0f;
470 float cur_kbps = 0.0f;
471 float avg_kbps = 0.0f;
472 float max_incoming_kbps = 0.0f;
473 float cur_incoming_kbps = 0.0f;
474 float avg_incoming_kbps = 0.0f;
475 float max_kbps_lost = 0.0f;
476 float cur_kbps_lost = 0.0f;
477 float avg_kbps_lost = 0.0f;
478 float bpm_counter = 0.0f;
480 unsigned int rate_samples = 0;
488 PeerHelper() = default;
489 PeerHelper(Peer* peer);
492 PeerHelper& operator=(Peer* peer);
493 Peer* operator->() const;
495 Peer* operator&() const;
496 bool operator!=(void* ptr);
499 Peer *m_peer = nullptr;
515 friend class PeerHelper;
517 Peer(Address address_,u16 id_,Connection* connection) :
519 m_connection(connection),
521 m_last_timeout_check(porting::getTimeMs())
526 MutexAutoLock usage_lock(m_exclusive_access_mutex);
527 FATAL_ERROR_IF(m_usage != 0, "Reference counting failure");
530 // Unique id of the peer
535 virtual void PutReliableSendCommand(ConnectionCommand &c,
536 unsigned int max_packet_size) {};
538 virtual bool getAddress(MTProtocols type, Address& toset) = 0;
540 bool isPendingDeletion()
541 { MutexAutoLock lock(m_exclusive_access_mutex); return m_pending_deletion; };
544 {MutexAutoLock lock(m_exclusive_access_mutex); m_timeout_counter = 0.0; };
546 bool isTimedOut(float timeout);
548 unsigned int m_increment_packets_remaining = 0;
550 virtual u16 getNextSplitSequenceNumber(u8 channel) { return 0; };
551 virtual void setNextSplitSequenceNumber(u8 channel, u16 seqnum) {};
552 virtual SharedBuffer<u8> addSplitPacket(u8 channel, const BufferedPacket &toadd,
555 errorstream << "Peer::addSplitPacket called,"
556 << " this is supposed to be never called!" << std::endl;
557 return SharedBuffer<u8>(0);
560 virtual bool Ping(float dtime, SharedBuffer<u8>& data) { return false; };
562 virtual float getStat(rtt_stat_type type) const {
565 return m_rtt.min_rtt;
567 return m_rtt.max_rtt;
569 return m_rtt.avg_rtt;
571 return m_rtt.jitter_min;
573 return m_rtt.jitter_max;
575 return m_rtt.jitter_avg;
580 virtual void reportRTT(float rtt) {};
582 void RTTStatistics(float rtt,
583 const std::string &profiler_id = "",
584 unsigned int num_samples = 1000);
589 std::mutex m_exclusive_access_mutex;
591 bool m_pending_deletion = false;
593 Connection* m_connection;
595 // Address of the peer
599 float m_ping_timer = 0.0f;
603 float jitter_min = FLT_MAX;
604 float jitter_max = 0.0f;
605 float jitter_avg = -1.0f;
606 float min_rtt = FLT_MAX;
607 float max_rtt = 0.0f;
608 float avg_rtt = -1.0f;
610 rttstats() = default;
614 float m_last_rtt = -1.0f;
616 // current usage count
617 unsigned int m_usage = 0;
619 // Seconds from last receive
620 float m_timeout_counter = 0.0f;
622 u64 m_last_timeout_check;
625 class UDPPeer : public Peer
629 friend class PeerHelper;
630 friend class ConnectionReceiveThread;
631 friend class ConnectionSendThread;
632 friend class Connection;
634 UDPPeer(u16 a_id, Address a_address, Connection* connection);
635 virtual ~UDPPeer() = default;
637 void PutReliableSendCommand(ConnectionCommand &c,
638 unsigned int max_packet_size);
640 bool getAddress(MTProtocols type, Address& toset);
642 u16 getNextSplitSequenceNumber(u8 channel);
643 void setNextSplitSequenceNumber(u8 channel, u16 seqnum);
645 SharedBuffer<u8> addSplitPacket(u8 channel, const BufferedPacket &toadd,
650 Calculates avg_rtt and resend_timeout.
651 rtt=-1 only recalculates resend_timeout
653 void reportRTT(float rtt);
655 void RunCommandQueues(
656 unsigned int max_packet_size,
657 unsigned int maxcommands,
658 unsigned int maxtransfer);
660 float getResendTimeout()
661 { MutexAutoLock lock(m_exclusive_access_mutex); return resend_timeout; }
663 void setResendTimeout(float timeout)
664 { MutexAutoLock lock(m_exclusive_access_mutex); resend_timeout = timeout; }
665 bool Ping(float dtime,SharedBuffer<u8>& data);
667 Channel channels[CHANNEL_COUNT];
668 bool m_pending_disconnect = false;
670 // This is changed dynamically
671 float resend_timeout = 0.5;
673 bool processReliableSendCommand(
674 ConnectionCommand &c,
675 unsigned int max_packet_size);
682 enum ConnectionEventType{
684 CONNEVENT_DATA_RECEIVED,
685 CONNEVENT_PEER_ADDED,
686 CONNEVENT_PEER_REMOVED,
687 CONNEVENT_BIND_FAILED,
690 struct ConnectionEvent
692 enum ConnectionEventType type = CONNEVENT_NONE;
693 session_t peer_id = 0;
695 bool timeout = false;
698 ConnectionEvent() = default;
700 const char *describe() const
704 return "CONNEVENT_NONE";
705 case CONNEVENT_DATA_RECEIVED:
706 return "CONNEVENT_DATA_RECEIVED";
707 case CONNEVENT_PEER_ADDED:
708 return "CONNEVENT_PEER_ADDED";
709 case CONNEVENT_PEER_REMOVED:
710 return "CONNEVENT_PEER_REMOVED";
711 case CONNEVENT_BIND_FAILED:
712 return "CONNEVENT_BIND_FAILED";
714 return "Invalid ConnectionEvent";
717 void dataReceived(session_t peer_id_, const Buffer<u8> &data_)
719 type = CONNEVENT_DATA_RECEIVED;
723 void peerAdded(session_t peer_id_, Address address_)
725 type = CONNEVENT_PEER_ADDED;
729 void peerRemoved(session_t peer_id_, bool timeout_, Address address_)
731 type = CONNEVENT_PEER_REMOVED;
738 type = CONNEVENT_BIND_FAILED;
747 friend class ConnectionSendThread;
748 friend class ConnectionReceiveThread;
750 Connection(u32 protocol_id, u32 max_packet_size, float timeout, bool ipv6,
751 PeerHandler *peerhandler);
755 ConnectionEvent waitEvent(u32 timeout_ms);
756 // Warning: creates an unnecessary copy, prefer putCommand(T&&) if possible
757 void putCommand(const ConnectionCommand &c);
758 void putCommand(ConnectionCommand &&c);
760 void SetTimeoutMs(u32 timeout) { m_bc_receive_timeout = timeout; }
761 void Serve(Address bind_addr);
762 void Connect(Address address);
765 void Receive(NetworkPacket* pkt);
766 bool TryReceive(NetworkPacket *pkt);
767 void Send(session_t peer_id, u8 channelnum, NetworkPacket *pkt, bool reliable);
768 session_t GetPeerID() const { return m_peer_id; }
769 Address GetPeerAddress(session_t peer_id);
770 float getPeerStat(session_t peer_id, rtt_stat_type type);
771 float getLocalStat(rate_stat_type type);
772 u32 GetProtocolID() const { return m_protocol_id; };
773 const std::string getDesc();
774 void DisconnectPeer(session_t peer_id);
777 PeerHelper getPeerNoEx(session_t peer_id);
778 u16 lookupPeer(Address& sender);
780 u16 createPeer(Address& sender, MTProtocols protocol, int fd);
781 UDPPeer* createServerPeer(Address& sender);
782 bool deletePeer(session_t peer_id, bool timeout);
784 void SetPeerID(session_t id) { m_peer_id = id; }
786 void sendAck(session_t peer_id, u8 channelnum, u16 seqnum);
788 void PrintInfo(std::ostream &out);
790 std::vector<session_t> getPeerIDs()
792 MutexAutoLock peerlock(m_peers_mutex);
796 UDPSocket m_udpSocket;
797 // Command queue: user -> SendThread
798 MutexedQueue<ConnectionCommand> m_command_queue;
800 bool Receive(NetworkPacket *pkt, u32 timeout);
802 // Warning: creates an unnecessary copy, prefer putEvent(T&&) if possible
803 void putEvent(const ConnectionEvent &e);
804 void putEvent(ConnectionEvent &&e);
808 bool ConnectedToServer()
810 return getPeerNoEx(PEER_ID_SERVER) != nullptr;
813 // Event queue: ReceiveThread -> user
814 MutexedQueue<ConnectionEvent> m_event_queue;
816 session_t m_peer_id = 0;
819 std::map<session_t, Peer *> m_peers;
820 std::vector<session_t> m_peer_ids;
821 std::mutex m_peers_mutex;
823 std::unique_ptr<ConnectionSendThread> m_sendThread;
824 std::unique_ptr<ConnectionReceiveThread> m_receiveThread;
826 std::mutex m_info_mutex;
828 // Backwards compatibility
829 PeerHandler *m_bc_peerhandler;
830 u32 m_bc_receive_timeout = 0;
832 bool m_shutting_down = false;
834 session_t m_next_remote_peer_id = 2;