#pragma once
-#include "irrlichttypes_bloated.h"
+#include "irrlichttypes.h"
#include "peerhandler.h"
#include "socket.h"
#include "constants.h"
#include "util/numeric.h"
#include "networkprotocol.h"
#include <iostream>
-#include <fstream>
-#include <list>
+#include <vector>
#include <map>
+#define MAX_UDP_PEERS 65535
+
+/*
+=== NOTES ===
+
+A packet is sent through a channel to a peer with a basic header:
+ Header (7 bytes):
+ [0] u32 protocol_id
+ [4] session_t sender_peer_id
+ [6] u8 channel
+sender_peer_id:
+ Unique to each peer.
+ value 0 (PEER_ID_INEXISTENT) is reserved for making new connections
+ value 1 (PEER_ID_SERVER) is reserved for server
+ these constants are defined in constants.h
+channel:
+ Channel numbers have no intrinsic meaning. Currently only 0, 1, 2 exist.
+*/
+#define BASE_HEADER_SIZE 7
+#define CHANNEL_COUNT 3
+
+/*
+Packet types:
+
+CONTROL: This is a packet used by the protocol.
+- When this is processed, nothing is handed to the user.
+ Header (2 byte):
+ [0] u8 type
+ [1] u8 controltype
+controltype and data description:
+ CONTROLTYPE_ACK
+ [2] u16 seqnum
+ CONTROLTYPE_SET_PEER_ID
+ [2] session_t peer_id_new
+ CONTROLTYPE_PING
+ - There is no actual reply, but this can be sent in a reliable
+ packet to get a reply
+ CONTROLTYPE_DISCO
+*/
+enum ControlType : u8 {
+ CONTROLTYPE_ACK = 0,
+ CONTROLTYPE_SET_PEER_ID = 1,
+ CONTROLTYPE_PING = 2,
+ CONTROLTYPE_DISCO = 3,
+};
+
+/*
+ORIGINAL: This is a plain packet with no control and no error
+checking at all.
+- When this is processed, it is directly handed to the user.
+ Header (1 byte):
+ [0] u8 type
+*/
+//#define TYPE_ORIGINAL 1
+#define ORIGINAL_HEADER_SIZE 1
+
+/*
+SPLIT: These are sequences of packets forming one bigger piece of
+data.
+- When processed and all the packet_nums 0...packet_count-1 are
+ present (this should be buffered), the resulting data shall be
+ directly handed to the user.
+- If the data fails to come up in a reasonable time, the buffer shall
+ be silently discarded.
+- These can be sent as-is or atop of a RELIABLE packet stream.
+ Header (7 bytes):
+ [0] u8 type
+ [1] u16 seqnum
+ [3] u16 chunk_count
+ [5] u16 chunk_num
+*/
+//#define TYPE_SPLIT 2
+
+/*
+RELIABLE: Delivery of all RELIABLE packets shall be forced by ACKs,
+and they shall be delivered in the same order as sent. This is done
+with a buffer in the receiving and transmitting end.
+- When this is processed, the contents of each packet is recursively
+ processed as packets.
+ Header (3 bytes):
+ [0] u8 type
+ [1] u16 seqnum
+
+*/
+//#define TYPE_RELIABLE 3
+#define RELIABLE_HEADER_SIZE 3
+#define SEQNUM_INITIAL 65500
+#define SEQNUM_MAX 65535
+
class NetworkPacket;
namespace con
MTP_MINETEST_RELIABLE_UDP
} MTProtocols;
-#define MAX_UDP_PEERS 65535
-
-#define SEQNUM_MAX 65535
+enum PacketType : u8 {
+ PACKET_TYPE_CONTROL = 0,
+ PACKET_TYPE_ORIGINAL = 1,
+ PACKET_TYPE_SPLIT = 2,
+ PACKET_TYPE_RELIABLE = 3,
+ PACKET_TYPE_MAX
+};
inline bool seqnum_higher(u16 totest, u16 base)
{
return MYMAX(MYMIN(value,0.1),0.0);
}
-struct BufferedPacket
-{
- BufferedPacket(u8 *a_data, u32 a_size):
- data(a_data, a_size)
- {}
- BufferedPacket(u32 a_size):
- data(a_size)
- {}
- Buffer<u8> data; // Data of the packet, including headers
+/*
+ Struct for all kinds of packets. Includes following data:
+ BASE_HEADER
+ u8[] packet data (usually copied from SharedBuffer<u8>)
+*/
+struct BufferedPacket {
+ BufferedPacket(u32 a_size)
+ {
+ m_data.resize(a_size);
+ data = &m_data[0];
+ }
+
+ DISABLE_CLASS_COPY(BufferedPacket)
+
+ u16 getSeqnum() const;
+
+ inline size_t size() const { return m_data.size(); }
+
+ u8 *data; // Direct memory access
float time = 0.0f; // Seconds from buffering the packet or re-sending
float totaltime = 0.0f; // Seconds from buffering the packet
u64 absolute_send_time = -1;
Address address; // Sender or destination
unsigned int resend_count = 0;
+
+private:
+ std::vector<u8> m_data; // Data of the packet, including headers
};
+typedef std::shared_ptr<BufferedPacket> BufferedPacketPtr;
+
+
// This adds the base headers to the data and makes a packet out of it
-BufferedPacket makePacket(Address &address, const SharedBuffer<u8> &data,
+BufferedPacketPtr makePacket(Address &address, const SharedBuffer<u8> &data,
u32 protocol_id, session_t sender_peer_id, u8 channel);
// Depending on size, make a TYPE_ORIGINAL or TYPE_SPLIT packet
std::map<u16, SharedBuffer<u8>> chunks;
};
-/*
-=== NOTES ===
-
-A packet is sent through a channel to a peer with a basic header:
-TODO: Should we have a receiver_peer_id also?
- Header (7 bytes):
- [0] u32 protocol_id
- [4] session_t sender_peer_id
- [6] u8 channel
-sender_peer_id:
- Unique to each peer.
- value 0 (PEER_ID_INEXISTENT) is reserved for making new connections
- value 1 (PEER_ID_SERVER) is reserved for server
- these constants are defined in constants.h
-channel:
- The lower the number, the higher the priority is.
- Only channels 0, 1 and 2 exist.
-*/
-#define BASE_HEADER_SIZE 7
-#define CHANNEL_COUNT 3
-/*
-Packet types:
-
-CONTROL: This is a packet used by the protocol.
-- When this is processed, nothing is handed to the user.
- Header (2 byte):
- [0] u8 type
- [1] u8 controltype
-controltype and data description:
- CONTROLTYPE_ACK
- [2] u16 seqnum
- CONTROLTYPE_SET_PEER_ID
- [2] session_t peer_id_new
- CONTROLTYPE_PING
- - There is no actual reply, but this can be sent in a reliable
- packet to get a reply
- CONTROLTYPE_DISCO
-*/
-//#define TYPE_CONTROL 0
-#define CONTROLTYPE_ACK 0
-#define CONTROLTYPE_SET_PEER_ID 1
-#define CONTROLTYPE_PING 2
-#define CONTROLTYPE_DISCO 3
-
-/*
-ORIGINAL: This is a plain packet with no control and no error
-checking at all.
-- When this is processed, it is directly handed to the user.
- Header (1 byte):
- [0] u8 type
-*/
-//#define TYPE_ORIGINAL 1
-#define ORIGINAL_HEADER_SIZE 1
-/*
-SPLIT: These are sequences of packets forming one bigger piece of
-data.
-- When processed and all the packet_nums 0...packet_count-1 are
- present (this should be buffered), the resulting data shall be
- directly handed to the user.
-- If the data fails to come up in a reasonable time, the buffer shall
- be silently discarded.
-- These can be sent as-is or atop of a RELIABLE packet stream.
- Header (7 bytes):
- [0] u8 type
- [1] u16 seqnum
- [3] u16 chunk_count
- [5] u16 chunk_num
-*/
-//#define TYPE_SPLIT 2
-/*
-RELIABLE: Delivery of all RELIABLE packets shall be forced by ACKs,
-and they shall be delivered in the same order as sent. This is done
-with a buffer in the receiving and transmitting end.
-- When this is processed, the contents of each packet is recursively
- processed as packets.
- Header (3 bytes):
- [0] u8 type
- [1] u16 seqnum
-
-*/
-//#define TYPE_RELIABLE 3
-#define RELIABLE_HEADER_SIZE 3
-#define SEQNUM_INITIAL 65500
-
-enum PacketType: u8 {
- PACKET_TYPE_CONTROL = 0,
- PACKET_TYPE_ORIGINAL = 1,
- PACKET_TYPE_SPLIT = 2,
- PACKET_TYPE_RELIABLE = 3,
- PACKET_TYPE_MAX
-};
/*
A buffer which stores reliable packets and sorts them internally
for fast access to the smallest one.
*/
-typedef std::list<BufferedPacket>::iterator RPBSearchResult;
+typedef std::list<BufferedPacketPtr>::iterator RPBSearchResult;
class ReliablePacketBuffer
{
bool getFirstSeqnum(u16& result);
- BufferedPacket popFirst();
- BufferedPacket popSeqnum(u16 seqnum);
- void insert(BufferedPacket &p, u16 next_expected);
+ BufferedPacketPtr popFirst();
+ BufferedPacketPtr popSeqnum(u16 seqnum);
+ void insert(BufferedPacketPtr &p_ptr, u16 next_expected);
void incrementTimeouts(float dtime);
- std::list<BufferedPacket> getTimedOuts(float timeout,
- unsigned int max_packets);
+ std::list<ConstSharedPtr<BufferedPacket>> getTimedOuts(float timeout, u32 max_packets);
void print();
bool empty();
- RPBSearchResult notFound();
u32 size();
private:
- RPBSearchResult findPacket(u16 seqnum); // does not perform locking
+ RPBSearchResult findPacketNoLock(u16 seqnum);
- std::list<BufferedPacket> m_list;
+ std::list<BufferedPacketPtr> m_list;
u16 m_oldest_non_answered_ack;
Returns a reference counted buffer of length != 0 when a full split
packet is constructed. If not, returns one of length 0.
*/
- SharedBuffer<u8> insert(const BufferedPacket &p, bool reliable);
+ SharedBuffer<u8> insert(BufferedPacketPtr &p_ptr, bool reliable);
void removeUnreliableTimedOuts(float dtime, float timeout);
std::mutex m_map_mutex;
};
-struct OutgoingPacket
-{
- session_t peer_id;
- u8 channelnum;
- SharedBuffer<u8> data;
- bool reliable;
- bool ack;
-
- OutgoingPacket(session_t peer_id_, u8 channelnum_, const SharedBuffer<u8> &data_,
- bool reliable_,bool ack_=false):
- peer_id(peer_id_),
- channelnum(channelnum_),
- data(data_),
- reliable(reliable_),
- ack(ack_)
- {
- }
-};
-
enum ConnectionCommandType{
CONNCMD_NONE,
CONNCMD_SERVE,
CONCMD_CREATE_PEER
};
+struct ConnectionCommand;
+typedef std::shared_ptr<ConnectionCommand> ConnectionCommandPtr;
+
+// This is very similar to ConnectionEvent
struct ConnectionCommand
{
- enum ConnectionCommandType type = CONNCMD_NONE;
+ const ConnectionCommandType type;
Address address;
session_t peer_id = PEER_ID_INEXISTENT;
u8 channelnum = 0;
bool reliable = false;
bool raw = false;
- ConnectionCommand() = default;
- ConnectionCommand &operator=(const ConnectionCommand &other)
- {
- type = other.type;
- address = other.address;
- peer_id = other.peer_id;
- channelnum = other.channelnum;
- // We must copy the buffer here to prevent race condition
- data = SharedBuffer<u8>(*other.data, other.data.getSize());
- reliable = other.reliable;
- raw = other.raw;
- return *this;
- }
+ DISABLE_CLASS_COPY(ConnectionCommand);
- void serve(Address address_)
- {
- type = CONNCMD_SERVE;
- address = address_;
- }
- void connect(Address address_)
- {
- type = CONNCMD_CONNECT;
- address = address_;
- }
- void disconnect()
- {
- type = CONNCMD_DISCONNECT;
- }
- void disconnect_peer(session_t peer_id_)
- {
- type = CONNCMD_DISCONNECT_PEER;
- peer_id = peer_id_;
- }
+ static ConnectionCommandPtr serve(Address address);
+ static ConnectionCommandPtr connect(Address address);
+ static ConnectionCommandPtr disconnect();
+ static ConnectionCommandPtr disconnect_peer(session_t peer_id);
+ static ConnectionCommandPtr send(session_t peer_id, u8 channelnum, NetworkPacket *pkt, bool reliable);
+ static ConnectionCommandPtr ack(session_t peer_id, u8 channelnum, const Buffer<u8> &data);
+ static ConnectionCommandPtr createPeer(session_t peer_id, const Buffer<u8> &data);
- void send(session_t peer_id_, u8 channelnum_, NetworkPacket *pkt, bool reliable_);
-
- void ack(session_t peer_id_, u8 channelnum_, const SharedBuffer<u8> &data_)
- {
- type = CONCMD_ACK;
- peer_id = peer_id_;
- channelnum = channelnum_;
- data = data_;
- reliable = false;
- }
+private:
+ ConnectionCommand(ConnectionCommandType type_) :
+ type(type_) {}
- void createPeer(session_t peer_id_, const SharedBuffer<u8> &data_)
- {
- type = CONCMD_CREATE_PEER;
- peer_id = peer_id_;
- data = data_;
- channelnum = 0;
- reliable = true;
- raw = true;
- }
+ static ConnectionCommandPtr create(ConnectionCommandType type);
};
-/* maximum window size to use, 0xFFFF is theoretical maximum don't think about
+/* maximum window size to use, 0xFFFF is theoretical maximum. don't think about
* touching it, the less you're away from it the more likely data corruption
* will occur
*/
#define MAX_RELIABLE_WINDOW_SIZE 0x8000
- /* starting value for window size */
+/* starting value for window size */
+#define START_RELIABLE_WINDOW_SIZE 0x400
+/* minimum value for window size */
#define MIN_RELIABLE_WINDOW_SIZE 0x40
class Channel
u16 readNextIncomingSeqNum();
u16 incNextIncomingSeqNum();
- u16 getOutgoingSequenceNumber(bool& successfull);
+ u16 getOutgoingSequenceNumber(bool& successful);
u16 readOutgoingSequenceNumber();
bool putBackSequenceNumber(u16);
ReliablePacketBuffer outgoing_reliables_sent;
//queued reliable packets
- std::queue<BufferedPacket> queued_reliables;
+ std::queue<BufferedPacketPtr> queued_reliables;
//queue commands prior splitting to packets
- std::deque<ConnectionCommand> queued_commands;
+ std::deque<ConnectionCommandPtr> queued_commands;
IncomingSplitBuffer incoming_splits;
void UpdateTimers(float dtime);
- const float getCurrentDownloadRateKB()
+ float getCurrentDownloadRateKB()
{ MutexAutoLock lock(m_internal_mutex); return cur_kbps; };
- const float getMaxDownloadRateKB()
+ float getMaxDownloadRateKB()
{ MutexAutoLock lock(m_internal_mutex); return max_kbps; };
- const float getCurrentLossRateKB()
+ float getCurrentLossRateKB()
{ MutexAutoLock lock(m_internal_mutex); return cur_kbps_lost; };
- const float getMaxLossRateKB()
+ float getMaxLossRateKB()
{ MutexAutoLock lock(m_internal_mutex); return max_kbps_lost; };
- const float getCurrentIncomingRateKB()
+ float getCurrentIncomingRateKB()
{ MutexAutoLock lock(m_internal_mutex); return cur_incoming_kbps; };
- const float getMaxIncomingRateKB()
+ float getMaxIncomingRateKB()
{ MutexAutoLock lock(m_internal_mutex); return max_incoming_kbps; };
- const float getAvgDownloadRateKB()
+ float getAvgDownloadRateKB()
{ MutexAutoLock lock(m_internal_mutex); return avg_kbps; };
- const float getAvgLossRateKB()
+ float getAvgLossRateKB()
{ MutexAutoLock lock(m_internal_mutex); return avg_kbps_lost; };
- const float getAvgIncomingRateKB()
+ float getAvgIncomingRateKB()
{ MutexAutoLock lock(m_internal_mutex); return avg_incoming_kbps; };
- const unsigned int getWindowSize() const { return window_size; };
+ u16 getWindowSize() const { return m_window_size; };
+
+ void setWindowSize(long size)
+ {
+ m_window_size = (u16)rangelim(size, MIN_RELIABLE_WINDOW_SIZE, MAX_RELIABLE_WINDOW_SIZE);
+ }
- void setWindowSize(unsigned int size) { window_size = size; };
private:
std::mutex m_internal_mutex;
- int window_size = MIN_RELIABLE_WINDOW_SIZE;
+ u16 m_window_size = MIN_RELIABLE_WINDOW_SIZE;
u16 next_incoming_seqnum = SEQNUM_INITIAL;
public:
friend class PeerHelper;
- Peer(Address address_,u16 id_,Connection* connection) :
+ Peer(Address address_,session_t id_,Connection* connection) :
id(id_),
m_connection(connection),
address(address_),
};
// Unique id of the peer
- u16 id;
+ const session_t id;
void Drop();
- virtual void PutReliableSendCommand(ConnectionCommand &c,
+ virtual void PutReliableSendCommand(ConnectionCommandPtr &c,
unsigned int max_packet_size) {};
virtual bool getAddress(MTProtocols type, Address& toset) = 0;
bool isTimedOut(float timeout);
- unsigned int m_increment_packets_remaining = 9;
- unsigned int m_increment_bytes_remaining = 0;
+ unsigned int m_increment_packets_remaining = 0;
virtual u16 getNextSplitSequenceNumber(u8 channel) { return 0; };
virtual void setNextSplitSequenceNumber(u8 channel, u16 seqnum) {};
- virtual SharedBuffer<u8> addSplitPacket(u8 channel, const BufferedPacket &toadd,
+ virtual SharedBuffer<u8> addSplitPacket(u8 channel, BufferedPacketPtr &toadd,
bool reliable)
{
- fprintf(stderr,"Peer: addSplitPacket called, this is supposed to be never called!\n");
+ errorstream << "Peer::addSplitPacket called,"
+ << " this is supposed to be never called!" << std::endl;
return SharedBuffer<u8>(0);
};
bool IncUseCount();
void DecUseCount();
- std::mutex m_exclusive_access_mutex;
+ mutable std::mutex m_exclusive_access_mutex;
bool m_pending_deletion = false;
struct rttstats {
float jitter_min = FLT_MAX;
float jitter_max = 0.0f;
- float jitter_avg = -2.0f;
+ float jitter_avg = -1.0f;
float min_rtt = FLT_MAX;
float max_rtt = 0.0f;
- float avg_rtt = -2.0f;
+ float avg_rtt = -1.0f;
rttstats() = default;
};
rttstats m_rtt;
- float m_last_rtt = -2.0f;
+ float m_last_rtt = -1.0f;
// current usage count
unsigned int m_usage = 0;
UDPPeer(u16 a_id, Address a_address, Connection* connection);
virtual ~UDPPeer() = default;
- void PutReliableSendCommand(ConnectionCommand &c,
+ void PutReliableSendCommand(ConnectionCommandPtr &c,
unsigned int max_packet_size);
bool getAddress(MTProtocols type, Address& toset);
u16 getNextSplitSequenceNumber(u8 channel);
void setNextSplitSequenceNumber(u8 channel, u16 seqnum);
- SharedBuffer<u8> addSplitPacket(u8 channel, const BufferedPacket &toadd,
+ SharedBuffer<u8> addSplitPacket(u8 channel, BufferedPacketPtr &toadd,
bool reliable);
protected:
float resend_timeout = 0.5;
bool processReliableSendCommand(
- ConnectionCommand &c,
+ ConnectionCommandPtr &c_ptr,
unsigned int max_packet_size);
};
Connection
*/
-enum ConnectionEventType{
+enum ConnectionEventType {
CONNEVENT_NONE,
CONNEVENT_DATA_RECEIVED,
CONNEVENT_PEER_ADDED,
CONNEVENT_BIND_FAILED,
};
+struct ConnectionEvent;
+typedef std::shared_ptr<ConnectionEvent> ConnectionEventPtr;
+
+// This is very similar to ConnectionCommand
struct ConnectionEvent
{
- enum ConnectionEventType type = CONNEVENT_NONE;
+ const ConnectionEventType type;
session_t peer_id = 0;
Buffer<u8> data;
bool timeout = false;
Address address;
- ConnectionEvent() = default;
+ // We don't want to copy "data"
+ DISABLE_CLASS_COPY(ConnectionEvent);
- std::string describe()
- {
- switch(type) {
- case CONNEVENT_NONE:
- return "CONNEVENT_NONE";
- case CONNEVENT_DATA_RECEIVED:
- return "CONNEVENT_DATA_RECEIVED";
- case CONNEVENT_PEER_ADDED:
- return "CONNEVENT_PEER_ADDED";
- case CONNEVENT_PEER_REMOVED:
- return "CONNEVENT_PEER_REMOVED";
- case CONNEVENT_BIND_FAILED:
- return "CONNEVENT_BIND_FAILED";
- }
- return "Invalid ConnectionEvent";
- }
+ static ConnectionEventPtr create(ConnectionEventType type);
+ static ConnectionEventPtr dataReceived(session_t peer_id, const Buffer<u8> &data);
+ static ConnectionEventPtr peerAdded(session_t peer_id, Address address);
+ static ConnectionEventPtr peerRemoved(session_t peer_id, bool is_timeout, Address address);
+ static ConnectionEventPtr bindFailed();
- void dataReceived(session_t peer_id_, const SharedBuffer<u8> &data_)
- {
- type = CONNEVENT_DATA_RECEIVED;
- peer_id = peer_id_;
- data = data_;
- }
- void peerAdded(session_t peer_id_, Address address_)
- {
- type = CONNEVENT_PEER_ADDED;
- peer_id = peer_id_;
- address = address_;
- }
- void peerRemoved(session_t peer_id_, bool timeout_, Address address_)
- {
- type = CONNEVENT_PEER_REMOVED;
- peer_id = peer_id_;
- timeout = timeout_;
- address = address_;
- }
- void bindFailed()
- {
- type = CONNEVENT_BIND_FAILED;
- }
+ const char *describe() const;
+
+private:
+ ConnectionEvent(ConnectionEventType type_) :
+ type(type_) {}
};
class PeerHandler;
~Connection();
/* Interface */
- ConnectionEvent waitEvent(u32 timeout_ms);
- void putCommand(ConnectionCommand &c);
+ ConnectionEventPtr waitEvent(u32 timeout_ms);
+
+ void putCommand(ConnectionCommandPtr c);
void SetTimeoutMs(u32 timeout) { m_bc_receive_timeout = timeout; }
void Serve(Address bind_addr);
Address GetPeerAddress(session_t peer_id);
float getPeerStat(session_t peer_id, rtt_stat_type type);
float getLocalStat(rate_stat_type type);
- const u32 GetProtocolID() const { return m_protocol_id; };
+ u32 GetProtocolID() const { return m_protocol_id; };
const std::string getDesc();
void DisconnectPeer(session_t peer_id);
void sendAck(session_t peer_id, u8 channelnum, u16 seqnum);
- void PrintInfo(std::ostream &out);
-
- std::list<session_t> getPeerIDs()
+ std::vector<session_t> getPeerIDs()
{
MutexAutoLock peerlock(m_peers_mutex);
return m_peer_ids;
}
UDPSocket m_udpSocket;
- MutexedQueue<ConnectionCommand> m_command_queue;
+ // Command queue: user -> SendThread
+ MutexedQueue<ConnectionCommandPtr> m_command_queue;
bool Receive(NetworkPacket *pkt, u32 timeout);
- void putEvent(ConnectionEvent &e);
+ void putEvent(ConnectionEventPtr e);
void TriggerSend();
+
+ bool ConnectedToServer()
+ {
+ return getPeerNoEx(PEER_ID_SERVER) != nullptr;
+ }
private:
- MutexedQueue<ConnectionEvent> m_event_queue;
+ // Event queue: ReceiveThread -> user
+ MutexedQueue<ConnectionEventPtr> m_event_queue;
session_t m_peer_id = 0;
u32 m_protocol_id;
std::map<session_t, Peer *> m_peers;
- std::list<session_t> m_peer_ids;
+ std::vector<session_t> m_peer_ids;
std::mutex m_peers_mutex;
std::unique_ptr<ConnectionSendThread> m_sendThread;
std::unique_ptr<ConnectionReceiveThread> m_receiveThread;
- std::mutex m_info_mutex;
+ mutable std::mutex m_info_mutex;
// Backwards compatibility
PeerHandler *m_bc_peerhandler;
projects / minetest.git / blobdiff