51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
*/
+#include <iomanip>
+#include <errno.h>
#include "connection.h"
#include "main.h"
#include "serialization.h"
#include "util/numeric.h"
#include "util/string.h"
#include "settings.h"
+#include "profiler.h"
+#include "main.h" // for profiling
namespace con
{
+/* this mutex is used to achieve log message consistency */
+JMutex log_message_mutex;
+#define LOG(a) \
+ { \
+ JMutexAutoLock loglock(log_message_mutex); \
+ a; \
+ }
+
+/******************************************************************************/
+/* defines used for debugging and profiling */
+/******************************************************************************/
+#define PROFILE(a) a
+//#define PROFILE(a)
+
+//#define DEBUG_CONNECTION_KBPS
+#undef DEBUG_CONNECTION_KBPS
+
+
+static inline float CALC_DTIME(unsigned int lasttime, unsigned int curtime) {
+ float value = ( curtime - lasttime) / 1000.0;
+ return MYMAX(MYMIN(value,0.1),0.0);
+}
+
+/* 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 */
+#define MIN_RELIABLE_WINDOW_SIZE 0x40
+
+#define MAX_UDP_PEERS 65535
+
+#define PING_TIMEOUT 5.0
+
static u16 readPeerId(u8 *packetdata)
{
return readU16(&packetdata[4]);
ReliablePacketBuffer
*/
-ReliablePacketBuffer::ReliablePacketBuffer(): m_list_size(0) {}
+ReliablePacketBuffer::ReliablePacketBuffer(): m_list_size(0),writeptr(0) {}
void ReliablePacketBuffer::print()
{
+ JMutexAutoLock listlock(m_list_mutex);
+ LOG(dout_con<<"Dump of ReliablePacketBuffer:" << std::endl);
+ unsigned int index = 0;
for(std::list<BufferedPacket>::iterator i = m_list.begin();
i != m_list.end();
++i)
{
u16 s = readU16(&(i->data[BASE_HEADER_SIZE+1]));
- dout_con<<s<<" ";
+ LOG(dout_con<<index<< ":" << s << std::endl);
+ index++;
}
}
bool ReliablePacketBuffer::empty()
{
+ JMutexAutoLock listlock(m_list_mutex);
return m_list.empty();
}
+
u32 ReliablePacketBuffer::size()
{
return m_list_size;
}
+
+bool ReliablePacketBuffer::containsPacket(u16 seqnum)
+{
+ return !(findPacket(seqnum) == m_list.end());
+}
+
RPBSearchResult ReliablePacketBuffer::findPacket(u16 seqnum)
{
std::list<BufferedPacket>::iterator i = m_list.begin();
{
return m_list.end();
}
-bool ReliablePacketBuffer::getFirstSeqnum(u16 *result)
+bool ReliablePacketBuffer::getFirstSeqnum(u16& result)
{
- if(empty())
+ JMutexAutoLock listlock(m_list_mutex);
+ if(m_list.empty())
return false;
BufferedPacket p = *m_list.begin();
- *result = readU16(&p.data[BASE_HEADER_SIZE+1]);
+ result = readU16(&p.data[BASE_HEADER_SIZE+1]);
return true;
}
+
BufferedPacket ReliablePacketBuffer::popFirst()
{
- if(empty())
+ JMutexAutoLock listlock(m_list_mutex);
+ if(m_list.empty())
throw NotFoundException("Buffer is empty");
BufferedPacket p = *m_list.begin();
m_list.erase(m_list.begin());
--m_list_size;
+
+ if (m_list_size == 0)
+ { m_oldest_non_answered_ack = 0; }
+ else
+ { m_oldest_non_answered_ack = readU16(&(*m_list.begin()).data[BASE_HEADER_SIZE+1]); }
return p;
}
BufferedPacket ReliablePacketBuffer::popSeqnum(u16 seqnum)
{
+ JMutexAutoLock listlock(m_list_mutex);
RPBSearchResult r = findPacket(seqnum);
if(r == notFound()){
- dout_con<<"Not found"<<std::endl;
+ LOG(dout_con<<"Sequence number: " << seqnum << " not found in reliable buffer"<<std::endl);
throw NotFoundException("seqnum not found in buffer");
}
BufferedPacket p = *r;
+
+
+ RPBSearchResult next = r;
+ next++;
+ if (next != notFound()) {
+ u16 s = readU16(&(next->data[BASE_HEADER_SIZE+1]));
+ m_oldest_non_answered_ack = s;
+ }
+
m_list.erase(r);
--m_list_size;
+
+ if (m_list_size == 0)
+ { m_oldest_non_answered_ack = 0; }
+ else
+ { m_oldest_non_answered_ack = readU16(&(*m_list.begin()).data[BASE_HEADER_SIZE+1]); }
return p;
}
-void ReliablePacketBuffer::insert(BufferedPacket &p)
+void ReliablePacketBuffer::insert(BufferedPacket &p,u16 next_expected)
{
+ JMutexAutoLock listlock(m_list_mutex);
assert(p.data.getSize() >= BASE_HEADER_SIZE+3);
u8 type = readU8(&p.data[BASE_HEADER_SIZE+0]);
assert(type == TYPE_RELIABLE);
u16 seqnum = readU16(&p.data[BASE_HEADER_SIZE+1]);
+ assert(seqnum_in_window(seqnum,next_expected,MAX_RELIABLE_WINDOW_SIZE));
+ assert(seqnum != next_expected);
+
++m_list_size;
- // Find the right place for the packet and insert it there
+ assert(m_list_size <= SEQNUM_MAX+1);
+ // Find the right place for the packet and insert it there
// If list is empty, just add it
if(m_list.empty())
{
m_list.push_back(p);
+ m_oldest_non_answered_ack = seqnum;
// Done.
return;
}
+
// Otherwise find the right place
std::list<BufferedPacket>::iterator i = m_list.begin();
// Find the first packet in the list which has a higher seqnum
- for(; i != m_list.end(); ++i){
- u16 s = readU16(&(i->data[BASE_HEADER_SIZE+1]));
- if(s == seqnum){
- --m_list_size;
- throw AlreadyExistsException("Same seqnum in list");
- }
- if(seqnum_higher(s, seqnum)){
- break;
+ u16 s = readU16(&(i->data[BASE_HEADER_SIZE+1]));
+
+ /* case seqnum is smaller then next_expected seqnum */
+ /* this is true e.g. on wrap around */
+ if (seqnum < next_expected) {
+ while(((s < seqnum) || (s >= next_expected)) && (i != m_list.end())) {
+ i++;
+ if (i != m_list.end())
+ s = readU16(&(i->data[BASE_HEADER_SIZE+1]));
}
}
- // If we're at the end of the list, add the packet to the
- // end of the list
- if(i == m_list.end())
+ /* non wrap around case (at least for incoming and next_expected */
+ else
{
+ while(((s < seqnum) && (s >= next_expected)) && (i != m_list.end())) {
+ i++;
+ if (i != m_list.end())
+ s = readU16(&(i->data[BASE_HEADER_SIZE+1]));
+ }
+ }
+
+ if (s == seqnum) {
+ if (
+ (readU16(&(i->data[BASE_HEADER_SIZE+1])) != seqnum) ||
+ (i->data.getSize() != p.data.getSize()) ||
+ (i->address != p.address)
+ )
+ {
+ /* if this happens your maximum transfer window may be to big */
+ fprintf(stderr, "Duplicated seqnum %d non matching packet detected:\n",seqnum);
+ fprintf(stderr, "Old: seqnum: %05d size: %04d, address: %s\n",
+ readU16(&(i->data[BASE_HEADER_SIZE+1])),i->data.getSize(), i->address.serializeString().c_str());
+ fprintf(stderr, "New: seqnum: %05d size: %04d, address: %s\n",
+ readU16(&(p.data[BASE_HEADER_SIZE+1])),p.data.getSize(), p.address.serializeString().c_str());
+ throw IncomingDataCorruption("duplicated packet isn't same as original one");
+ }
+
+ assert(readU16(&(i->data[BASE_HEADER_SIZE+1])) == seqnum);
+ assert(i->data.getSize() == p.data.getSize());
+ assert(i->address == p.address);
+
+ /* nothing to do this seems to be a resent packet */
+ /* for paranoia reason data should be compared */
+ --m_list_size;
+ }
+ /* insert or push back */
+ else if (i != m_list.end()) {
+ m_list.insert(i, p);
+ }
+ else {
m_list.push_back(p);
- // Done.
- return;
}
- // Insert before i
- m_list.insert(i, p);
+
+ /* update last packet number */
+ m_oldest_non_answered_ack = readU16(&(*m_list.begin()).data[BASE_HEADER_SIZE+1]);
}
void ReliablePacketBuffer::incrementTimeouts(float dtime)
{
+ JMutexAutoLock listlock(m_list_mutex);
for(std::list<BufferedPacket>::iterator i = m_list.begin();
i != m_list.end(); ++i)
{
}
}
-void ReliablePacketBuffer::resetTimedOuts(float timeout)
-{
- for(std::list<BufferedPacket>::iterator i = m_list.begin();
- i != m_list.end(); ++i)
- {
- if(i->time >= timeout)
- i->time = 0.0;
- }
-}
-
-bool ReliablePacketBuffer::anyTotaltimeReached(float timeout)
-{
- for(std::list<BufferedPacket>::iterator i = m_list.begin();
- i != m_list.end(); ++i)
- {
- if(i->totaltime >= timeout)
- return true;
- }
- return false;
-}
-
-std::list<BufferedPacket> ReliablePacketBuffer::getTimedOuts(float timeout)
+std::list<BufferedPacket> ReliablePacketBuffer::getTimedOuts(float timeout,
+ unsigned int max_packets)
{
+ JMutexAutoLock listlock(m_list_mutex);
std::list<BufferedPacket> timed_outs;
for(std::list<BufferedPacket>::iterator i = m_list.begin();
i != m_list.end(); ++i)
{
- if(i->time >= timeout)
+ if(i->time >= timeout) {
timed_outs.push_back(*i);
+
+ //this packet will be sent right afterwards reset timeout here
+ i->time = 0.0;
+ if (timed_outs.size() >= max_packets)
+ break;
+ }
}
return timed_outs;
}
IncomingSplitBuffer::~IncomingSplitBuffer()
{
+ JMutexAutoLock listlock(m_map_mutex);
for(std::map<u16, IncomingSplitPacket*>::iterator i = m_buf.begin();
i != m_buf.end(); ++i)
{
*/
SharedBuffer<u8> IncomingSplitBuffer::insert(BufferedPacket &p, bool reliable)
{
+ JMutexAutoLock listlock(m_map_mutex);
u32 headersize = BASE_HEADER_SIZE + 7;
assert(p.data.getSize() >= headersize);
u8 type = readU8(&p.data[BASE_HEADER_SIZE+0]);
// TODO: These errors should be thrown or something? Dunno.
if(chunk_count != sp->chunk_count)
- derr_con<<"Connection: WARNING: chunk_count="<<chunk_count
+ LOG(derr_con<<"Connection: WARNING: chunk_count="<<chunk_count
<<" != sp->chunk_count="<<sp->chunk_count
- <<std::endl;
+ <<std::endl);
if(reliable != sp->reliable)
- derr_con<<"Connection: WARNING: reliable="<<reliable
+ LOG(derr_con<<"Connection: WARNING: reliable="<<reliable
<<" != sp->reliable="<<sp->reliable
- <<std::endl;
+ <<std::endl);
// If chunk already exists, ignore it.
// Sometimes two identical packets may arrive when there is network
void IncomingSplitBuffer::removeUnreliableTimedOuts(float dtime, float timeout)
{
std::list<u16> remove_queue;
- for(std::map<u16, IncomingSplitPacket*>::iterator i = m_buf.begin();
- i != m_buf.end(); ++i)
{
- IncomingSplitPacket *p = i->second;
- // Reliable ones are not removed by timeout
- if(p->reliable == true)
- continue;
- p->time += dtime;
- if(p->time >= timeout)
- remove_queue.push_back(i->first);
+ JMutexAutoLock listlock(m_map_mutex);
+ for(std::map<u16, IncomingSplitPacket*>::iterator i = m_buf.begin();
+ i != m_buf.end(); ++i)
+ {
+ IncomingSplitPacket *p = i->second;
+ // Reliable ones are not removed by timeout
+ if(p->reliable == true)
+ continue;
+ p->time += dtime;
+ if(p->time >= timeout)
+ remove_queue.push_back(i->first);
+ }
}
for(std::list<u16>::iterator j = remove_queue.begin();
j != remove_queue.end(); ++j)
{
- dout_con<<"NOTE: Removing timed out unreliable split packet"
- <<std::endl;
+ JMutexAutoLock listlock(m_map_mutex);
+ LOG(dout_con<<"NOTE: Removing timed out unreliable split packet"<<std::endl);
delete m_buf[*j];
m_buf.erase(*j);
}
Channel
*/
-Channel::Channel()
+Channel::Channel() :
+ window_size(MIN_RELIABLE_WINDOW_SIZE),
+ next_incoming_seqnum(SEQNUM_INITIAL),
+ next_outgoing_seqnum(SEQNUM_INITIAL),
+ next_outgoing_split_seqnum(SEQNUM_INITIAL),
+ current_packet_loss(0),
+ current_packet_too_late(0),
+ packet_loss_counter(0),
+ current_bytes_transfered(0),
+ current_bytes_lost(0),
+ max_kbps(0.0),
+ cur_kbps(0.0),
+ avg_kbps(0.0),
+ max_kbps_lost(0.0),
+ cur_kbps_lost(0.0),
+ avg_kbps_lost(0.0),
+ bpm_counter(0.0)
{
- next_outgoing_seqnum = SEQNUM_INITIAL;
- next_incoming_seqnum = SEQNUM_INITIAL;
- next_outgoing_split_seqnum = SEQNUM_INITIAL;
}
+
Channel::~Channel()
{
}
-/*
- Peer
-*/
-
-Peer::Peer(u16 a_id, Address a_address):
- address(a_address),
- id(a_id),
- timeout_counter(0.0),
- ping_timer(0.0),
- resend_timeout(0.5),
- avg_rtt(-1.0),
- has_sent_with_id(false),
- m_sendtime_accu(0),
- m_max_packets_per_second(10),
- m_num_sent(0),
- m_max_num_sent(0),
- congestion_control_aim_rtt(0.2),
- congestion_control_max_rate(400),
- congestion_control_min_rate(10)
+u16 Channel::readNextIncomingSeqNum()
{
+ JMutexAutoLock internal(m_internal_mutex);
+ return next_incoming_seqnum;
}
-Peer::~Peer()
+
+u16 Channel::incNextIncomingSeqNum()
{
+ JMutexAutoLock internal(m_internal_mutex);
+ u16 retval = next_incoming_seqnum;
+ next_incoming_seqnum++;
+ return retval;
}
-void Peer::reportRTT(float rtt)
+u16 Channel::readNextSplitSeqNum()
{
- if(rtt >= 0.0){
- if(rtt < 0.01){
- if(m_max_packets_per_second < congestion_control_max_rate)
- m_max_packets_per_second += 10;
- } else if(rtt < congestion_control_aim_rtt){
- if(m_max_packets_per_second < congestion_control_max_rate)
- m_max_packets_per_second += 2;
- } else {
- m_max_packets_per_second *= 0.8;
- if(m_max_packets_per_second < congestion_control_min_rate)
- m_max_packets_per_second = congestion_control_min_rate;
- }
- }
-
- if(rtt < -0.999)
- {}
- else if(avg_rtt < 0.0)
- avg_rtt = rtt;
- else
- avg_rtt = rtt * 0.1 + avg_rtt * 0.9;
-
- // Calculate resend_timeout
-
- /*int reliable_count = 0;
- for(int i=0; i<CHANNEL_COUNT; i++)
- {
- reliable_count += channels[i].outgoing_reliables.size();
- }
- float timeout = avg_rtt * RESEND_TIMEOUT_FACTOR
- * ((float)reliable_count * 1);*/
-
- float timeout = avg_rtt * RESEND_TIMEOUT_FACTOR;
- if(timeout < RESEND_TIMEOUT_MIN)
- timeout = RESEND_TIMEOUT_MIN;
- if(timeout > RESEND_TIMEOUT_MAX)
- timeout = RESEND_TIMEOUT_MAX;
- resend_timeout = timeout;
+ JMutexAutoLock internal(m_internal_mutex);
+ return next_outgoing_split_seqnum;
}
-
-/*
- Connection
-*/
-
-Connection::Connection(u32 protocol_id, u32 max_packet_size, float timeout,
- bool ipv6):
- m_protocol_id(protocol_id),
- m_max_packet_size(max_packet_size),
- m_timeout(timeout),
- m_socket(ipv6),
- m_peer_id(0),
- m_bc_peerhandler(NULL),
- m_bc_receive_timeout(0),
- m_indentation(0)
+void Channel::setNextSplitSeqNum(u16 seqnum)
{
- m_socket.setTimeoutMs(5);
-
- Start();
+ JMutexAutoLock internal(m_internal_mutex);
+ next_outgoing_split_seqnum = seqnum;
}
-Connection::Connection(u32 protocol_id, u32 max_packet_size, float timeout,
- bool ipv6, PeerHandler *peerhandler):
- m_protocol_id(protocol_id),
- m_max_packet_size(max_packet_size),
- m_timeout(timeout),
- m_socket(ipv6),
- m_peer_id(0),
- m_bc_peerhandler(peerhandler),
- m_bc_receive_timeout(0),
- m_indentation(0)
+u16 Channel::getOutgoingSequenceNumber(bool& successfull)
{
- m_socket.setTimeoutMs(5);
-
- Start();
-}
+ JMutexAutoLock internal(m_internal_mutex);
+ u16 retval = next_outgoing_seqnum;
+ u16 lowest_unacked_seqnumber;
+ /* shortcut if there ain't any packet in outgoing list */
+ if (outgoing_reliables_sent.empty())
+ {
+ next_outgoing_seqnum++;
+ return retval;
+ }
-Connection::~Connection()
-{
- Stop();
- // Delete peers
- for(std::map<u16, Peer*>::iterator
- j = m_peers.begin();
- j != m_peers.end(); ++j)
+ if (outgoing_reliables_sent.getFirstSeqnum(lowest_unacked_seqnumber))
{
- delete j->second;
+ if (lowest_unacked_seqnumber < next_outgoing_seqnum) {
+ // ugly cast but this one is required in order to tell compiler we
+ // know about difference of two unsigned may be negative in general
+ // but we already made sure it won't happen in this case
+ if (((u16)(next_outgoing_seqnum - lowest_unacked_seqnumber)) > window_size) {
+ successfull = false;
+ return 0;
+ }
+ }
+ else {
+ // ugly cast but this one is required in order to tell compiler we
+ // know about difference of two unsigned may be negative in general
+ // but we already made sure it won't happen in this case
+ if ((next_outgoing_seqnum + (u16)(SEQNUM_MAX - lowest_unacked_seqnumber)) >
+ window_size) {
+ successfull = false;
+ return 0;
+ }
+ }
}
-}
-/* Internal stuff */
+ next_outgoing_seqnum++;
+ return retval;
+}
-void * Connection::Thread()
+u16 Channel::readOutgoingSequenceNumber()
{
- ThreadStarted();
- log_register_thread("Connection");
-
- dout_con<<"Connection thread started"<<std::endl;
-
- u32 curtime = porting::getTimeMs();
- u32 lasttime = curtime;
-
- while(!StopRequested())
- {
- BEGIN_DEBUG_EXCEPTION_HANDLER
-
- lasttime = curtime;
- curtime = porting::getTimeMs();
- float dtime = (float)(curtime - lasttime) / 1000.;
- if(dtime > 0.1)
- dtime = 0.1;
- if(dtime < 0.0)
- dtime = 0.0;
-
- runTimeouts(dtime);
-
- //NOTE this is only thread safe for ONE consumer thread!
- while(!m_command_queue.empty()){
- ConnectionCommand c = m_command_queue.pop_frontNoEx();
- processCommand(c);
- }
+ JMutexAutoLock internal(m_internal_mutex);
+ return next_outgoing_seqnum;
+}
- send(dtime);
+bool Channel::putBackSequenceNumber(u16 seqnum)
+{
+ if (((seqnum + 1) % (SEQNUM_MAX+1)) == next_outgoing_seqnum) {
- receive();
-
- END_DEBUG_EXCEPTION_HANDLER(derr_con);
+ next_outgoing_seqnum = seqnum;
+ return true;
}
+ return false;
+}
- return NULL;
+void Channel::UpdateBytesSent(unsigned int bytes, unsigned int packets)
+{
+ JMutexAutoLock internal(m_internal_mutex);
+ current_bytes_transfered += bytes;
+ current_packet_successfull += packets;
}
-void Connection::putEvent(ConnectionEvent &e)
+void Channel::UpdateBytesLost(unsigned int bytes)
{
- assert(e.type != CONNEVENT_NONE);
- m_event_queue.push_back(e);
+ JMutexAutoLock internal(m_internal_mutex);
+ current_bytes_lost += bytes;
}
-void Connection::processCommand(ConnectionCommand &c)
+
+void Channel::UpdatePacketLossCounter(unsigned int count)
{
- switch(c.type){
- case CONNCMD_NONE:
- dout_con<<getDesc()<<" processing CONNCMD_NONE"<<std::endl;
- return;
- case CONNCMD_SERVE:
- dout_con<<getDesc()<<" processing CONNCMD_SERVE port="
- <<c.port<<std::endl;
- serve(c.port);
- return;
- case CONNCMD_CONNECT:
- dout_con<<getDesc()<<" processing CONNCMD_CONNECT"<<std::endl;
- connect(c.address);
- return;
- case CONNCMD_DISCONNECT:
- dout_con<<getDesc()<<" processing CONNCMD_DISCONNECT"<<std::endl;
- disconnect();
- return;
- case CONNCMD_SEND:
- dout_con<<getDesc()<<" processing CONNCMD_SEND"<<std::endl;
- send(c.peer_id, c.channelnum, c.data, c.reliable);
- return;
- case CONNCMD_SEND_TO_ALL:
- dout_con<<getDesc()<<" processing CONNCMD_SEND_TO_ALL"<<std::endl;
- sendToAll(c.channelnum, c.data, c.reliable);
- return;
- case CONNCMD_DELETE_PEER:
- dout_con<<getDesc()<<" processing CONNCMD_DELETE_PEER"<<std::endl;
- deletePeer(c.peer_id, false);
- return;
- }
+ JMutexAutoLock internal(m_internal_mutex);
+ current_packet_loss += count;
}
-void Connection::send(float dtime)
+void Channel::UpdatePacketTooLateCounter()
{
- for(std::map<u16, Peer*>::iterator
- j = m_peers.begin();
- j != m_peers.end(); ++j)
- {
- Peer *peer = j->second;
- peer->m_sendtime_accu += dtime;
- peer->m_num_sent = 0;
- peer->m_max_num_sent = peer->m_sendtime_accu *
- peer->m_max_packets_per_second;
- }
- Queue<OutgoingPacket> postponed_packets;
- while(!m_outgoing_queue.empty()){
- OutgoingPacket packet = m_outgoing_queue.pop_front();
- Peer *peer = getPeerNoEx(packet.peer_id);
- if(!peer)
- continue;
- if(peer->channels[packet.channelnum].outgoing_reliables.size() >= 5){
- postponed_packets.push_back(packet);
- } else if(peer->m_num_sent < peer->m_max_num_sent){
- rawSendAsPacket(packet.peer_id, packet.channelnum,
- packet.data, packet.reliable);
- peer->m_num_sent++;
- } else {
- postponed_packets.push_back(packet);
- }
- }
- while(!postponed_packets.empty()){
- m_outgoing_queue.push_back(postponed_packets.pop_front());
- }
- for(std::map<u16, Peer*>::iterator
- j = m_peers.begin();
- j != m_peers.end(); ++j)
- {
- Peer *peer = j->second;
- peer->m_sendtime_accu -= (float)peer->m_num_sent /
- peer->m_max_packets_per_second;
- if(peer->m_sendtime_accu > 10. / peer->m_max_packets_per_second)
- peer->m_sendtime_accu = 10. / peer->m_max_packets_per_second;
- }
+ JMutexAutoLock internal(m_internal_mutex);
+ current_packet_too_late++;
}
-// Receive packets from the network and buffers and create ConnectionEvents
-void Connection::receive()
+void Channel::UpdateTimers(float dtime)
{
- u32 datasize = m_max_packet_size * 2; // Double it just to be safe
- // TODO: We can not know how many layers of header there are.
- // For now, just assume there are no other than the base headers.
- u32 packet_maxsize = datasize + BASE_HEADER_SIZE;
- SharedBuffer<u8> packetdata(packet_maxsize);
+ bpm_counter += dtime;
+ packet_loss_counter += dtime;
- bool single_wait_done = false;
-
- for(u32 loop_i=0; loop_i<1000; loop_i++) // Limit in case of DoS
+ if (packet_loss_counter > 1.0)
{
- try{
- /* Check if some buffer has relevant data */
+ packet_loss_counter -= 1.0;
+
+ unsigned int packet_loss = 11; /* use a neutral value for initialization */
+ unsigned int packets_successfull = 0;
+ unsigned int packet_too_late = 0;
+
+ bool reasonable_amount_of_data_transmitted = false;
+
{
- u16 peer_id;
- SharedBuffer<u8> resultdata;
- bool got = getFromBuffers(peer_id, resultdata);
- if(got){
- ConnectionEvent e;
- e.dataReceived(peer_id, resultdata);
- putEvent(e);
- continue;
+ JMutexAutoLock internal(m_internal_mutex);
+ packet_loss = current_packet_loss;
+ packet_too_late = current_packet_too_late;
+ packets_successfull = current_packet_successfull;
+
+ if (current_bytes_transfered > (window_size*512/2))
+ {
+ reasonable_amount_of_data_transmitted = true;
}
+ current_packet_loss = 0;
+ current_packet_too_late = 0;
+ current_packet_successfull = 0;
}
-
- if(single_wait_done){
- if(m_socket.WaitData(0) == false)
- break;
- }
-
- single_wait_done = true;
- Address sender;
- s32 received_size = m_socket.Receive(sender, *packetdata, packet_maxsize);
+ float successfull_to_lost_ratio = 0.0;
+ bool done = false;
- if(received_size < 0)
- break;
- if(received_size < BASE_HEADER_SIZE)
- continue;
- if(readU32(&packetdata[0]) != m_protocol_id)
- continue;
-
- u16 peer_id = readPeerId(*packetdata);
- u8 channelnum = readChannel(*packetdata);
- if(channelnum > CHANNEL_COUNT-1){
- PrintInfo(derr_con);
- derr_con<<"Receive(): Invalid channel "<<channelnum<<std::endl;
- throw InvalidIncomingDataException("Channel doesn't exist");
+ if (packets_successfull > 0) {
+ successfull_to_lost_ratio = packet_loss/packets_successfull;
+ }
+ else if (packet_loss > 0)
+ {
+ window_size = MYMAX(
+ (window_size - 10),
+ MIN_RELIABLE_WINDOW_SIZE);
+ done = true;
}
- if(peer_id == PEER_ID_INEXISTENT)
+ if (!done)
{
- /*
- Somebody is trying to send stuff to us with no peer id.
-
- Check if the same address and port was added to our peer
- list before.
- Allow only entries that have has_sent_with_id==false.
- */
-
- std::map<u16, Peer*>::iterator j;
- j = m_peers.begin();
- for(; j != m_peers.end(); ++j)
+ if ((successfull_to_lost_ratio < 0.01) &&
+ (window_size < MAX_RELIABLE_WINDOW_SIZE))
{
- Peer *peer = j->second;
- if(peer->has_sent_with_id)
- continue;
- if(peer->address == sender)
- break;
+ /* don't even think about increasing if we didn't even
+ * use major parts of our window */
+ if (reasonable_amount_of_data_transmitted)
+ window_size = MYMIN(
+ (window_size + 100),
+ MAX_RELIABLE_WINDOW_SIZE);
}
-
- /*
- If no peer was found with the same address and port,
- we shall assume it is a new peer and create an entry.
- */
- if(j == m_peers.end())
+ else if ((successfull_to_lost_ratio < 0.05) &&
+ (window_size < MAX_RELIABLE_WINDOW_SIZE))
{
- // Pass on to adding the peer
+ /* don't even think about increasing if we didn't even
+ * use major parts of our window */
+ if (reasonable_amount_of_data_transmitted)
+ window_size = MYMIN(
+ (window_size + 50),
+ MAX_RELIABLE_WINDOW_SIZE);
}
- // Else: A peer was found.
- else
+ else if (successfull_to_lost_ratio > 0.15)
{
- Peer *peer = j->second;
- peer_id = peer->id;
- PrintInfo(derr_con);
- derr_con<<"WARNING: Assuming unknown peer to be "
- <<"peer_id="<<peer_id<<std::endl;
+ window_size = MYMAX(
+ (window_size - 100),
+ MIN_RELIABLE_WINDOW_SIZE);
}
- }
-
- /*
- The peer was not found in our lists. Add it.
- */
- if(peer_id == PEER_ID_INEXISTENT)
- {
- // Somebody wants to make a new connection
-
- // Get a unique peer id (2 or higher)
- u16 peer_id_new = 2;
- /*
- Find an unused peer id
- */
- bool out_of_ids = false;
- for(;;)
+ else if (successfull_to_lost_ratio > 0.1)
{
- // Check if exists
- if(m_peers.find(peer_id_new) == m_peers.end())
- break;
- // Check for overflow
- if(peer_id_new == 65535){
- out_of_ids = true;
- break;
- }
- peer_id_new++;
+ window_size = MYMAX(
+ (window_size - 50),
+ MIN_RELIABLE_WINDOW_SIZE);
}
- if(out_of_ids){
- errorstream<<getDesc()<<" ran out of peer ids"<<std::endl;
- continue;
- }
-
- PrintInfo();
- dout_con<<"Receive(): Got a packet with peer_id=PEER_ID_INEXISTENT,"
- " giving peer_id="<<peer_id_new<<std::endl;
-
- // Create a peer
- Peer *peer = new Peer(peer_id_new, sender);
- m_peers[peer->id] = peer;
-
- // Create peer addition event
- ConnectionEvent e;
- e.peerAdded(peer_id_new, sender);
- putEvent(e);
-
- // Create CONTROL packet to tell the peer id to the new peer.
- SharedBuffer<u8> reply(4);
- writeU8(&reply[0], TYPE_CONTROL);
- writeU8(&reply[1], CONTROLTYPE_SET_PEER_ID);
- writeU16(&reply[2], peer_id_new);
- sendAsPacket(peer_id_new, 0, reply, true);
-
- // We're now talking to a valid peer_id
- peer_id = peer_id_new;
-
- // Go on and process whatever it sent
}
+ }
- std::map<u16, Peer*>::iterator node = m_peers.find(peer_id);
-
- if(node == m_peers.end())
+ if (bpm_counter > 10.0)
+ {
{
- // Peer not found
- // This means that the peer id of the sender is not PEER_ID_INEXISTENT
- // and it is invalid.
- PrintInfo(derr_con);
- derr_con<<"Receive(): Peer not found"<<std::endl;
- throw InvalidIncomingDataException("Peer not found (possible timeout)");
+ JMutexAutoLock internal(m_internal_mutex);
+ cur_kbps = (current_bytes_transfered/bpm_counter)/1024;
+ current_bytes_transfered = 0;
+ cur_kbps_lost = (current_bytes_lost/bpm_counter)/1024;
+ current_bytes_lost = 0;
+ bpm_counter = 0;
}
- Peer *peer = node->second;
-
- // Validate peer address
- if(peer->address != sender)
+ if (cur_kbps > max_kbps)
{
- PrintInfo(derr_con);
- derr_con<<"Peer "<<peer_id<<" sending from different address."
- " Ignoring."<<std::endl;
- continue;
+ max_kbps = cur_kbps;
}
-
- peer->timeout_counter = 0.0;
- Channel *channel = &(peer->channels[channelnum]);
-
- // Throw the received packet to channel->processPacket()
+ if (cur_kbps_lost > max_kbps_lost)
+ {
+ max_kbps_lost = cur_kbps_lost;
+ }
- // Make a new SharedBuffer from the data without the base headers
- SharedBuffer<u8> strippeddata(received_size - BASE_HEADER_SIZE);
- memcpy(*strippeddata, &packetdata[BASE_HEADER_SIZE],
- strippeddata.getSize());
-
- try{
- // Process it (the result is some data with no headers made by us)
- SharedBuffer<u8> resultdata = processPacket
- (channel, strippeddata, peer_id, channelnum, false);
-
- PrintInfo();
- dout_con<<"ProcessPacket returned data of size "
- <<resultdata.getSize()<<std::endl;
-
- ConnectionEvent e;
- e.dataReceived(peer_id, resultdata);
- putEvent(e);
- continue;
- }catch(ProcessedSilentlyException &e){
+ avg_kbps = avg_kbps * 0.9 + cur_kbps * 0.1;
+ avg_kbps_lost = avg_kbps_lost * 0.9 + cur_kbps_lost * 0.1;
+ }
+}
+
+
+/*
+ Peer
+*/
+
+PeerHelper::PeerHelper() :
+ m_peer(0)
+{
+}
+
+PeerHelper::PeerHelper(Peer* peer) :
+ m_peer(peer)
+{
+ if (peer != NULL)
+ {
+ if (!peer->IncUseCount())
+ {
+ m_peer = 0;
}
- }catch(InvalidIncomingDataException &e){
}
- catch(ProcessedSilentlyException &e){
+}
+
+PeerHelper::~PeerHelper()
+{
+ if (m_peer != 0)
+ m_peer->DecUseCount();
+
+ m_peer = 0;
+}
+
+PeerHelper& PeerHelper::operator=(Peer* peer)
+{
+ m_peer = peer;
+ if (peer != NULL)
+ {
+ if (!peer->IncUseCount())
+ {
+ m_peer = 0;
+ }
+ }
+ return *this;
+}
+
+Peer* PeerHelper::operator->() const
+{
+ return m_peer;
+}
+
+Peer* PeerHelper::operator&() const
+{
+ return m_peer;
+}
+
+bool PeerHelper::operator!() {
+ return ! m_peer;
+}
+
+bool PeerHelper::operator!=(void* ptr)
+{
+ return ((void*) m_peer != ptr);
+}
+
+bool Peer::IncUseCount()
+{
+ JMutexAutoLock lock(m_exclusive_access_mutex);
+
+ if (!m_pending_deletion)
+ {
+ this->m_usage++;
+ return true;
+ }
+
+ return false;
+}
+
+void Peer::DecUseCount()
+{
+ {
+ JMutexAutoLock lock(m_exclusive_access_mutex);
+ assert(m_usage > 0);
+ m_usage--;
+
+ if (!((m_pending_deletion) && (m_usage == 0)))
+ return;
+ }
+ delete this;
+}
+
+void Peer::RTTStatistics(float rtt,
+ std::string profiler_id,
+ unsigned int num_samples) {
+
+ if (m_last_rtt > 0) {
+ /* set min max values */
+ if (rtt < m_rtt.min_rtt)
+ m_rtt.min_rtt = rtt;
+ if (rtt >= m_rtt.max_rtt)
+ m_rtt.max_rtt = rtt;
+
+ /* do average calculation */
+ if(m_rtt.avg_rtt < 0.0)
+ m_rtt.avg_rtt = rtt;
+ else
+ m_rtt.avg_rtt = m_rtt.avg_rtt * (num_samples/(num_samples-1)) +
+ rtt * (1/num_samples);
+
+ /* do jitter calculation */
+
+ //just use some neutral value at beginning
+ float jitter = m_rtt.jitter_min;
+
+ if (rtt > m_last_rtt)
+ jitter = rtt-m_last_rtt;
+
+ if (rtt <= m_last_rtt)
+ jitter = m_last_rtt - rtt;
+
+ if (jitter < m_rtt.jitter_min)
+ m_rtt.jitter_min = jitter;
+ if (jitter >= m_rtt.jitter_max)
+ m_rtt.jitter_max = jitter;
+
+ if(m_rtt.jitter_avg < 0.0)
+ m_rtt.jitter_avg = jitter;
+ else
+ m_rtt.jitter_avg = m_rtt.jitter_avg * (num_samples/(num_samples-1)) +
+ jitter * (1/num_samples);
+
+ if (profiler_id != "")
+ {
+ g_profiler->graphAdd(profiler_id + "_rtt", rtt);
+ g_profiler->graphAdd(profiler_id + "_jitter", jitter);
+ }
+ }
+ /* save values required for next loop */
+ m_last_rtt = rtt;
+}
+
+bool Peer::isTimedOut(float timeout)
+{
+ JMutexAutoLock lock(m_exclusive_access_mutex);
+ u32 current_time = porting::getTimeMs();
+
+ float dtime = CALC_DTIME(m_last_timeout_check,current_time);
+ m_last_timeout_check = current_time;
+
+ m_timeout_counter += dtime;
+
+ return m_timeout_counter > timeout;
+}
+
+void Peer::Drop()
+{
+ {
+ JMutexAutoLock usage_lock(m_exclusive_access_mutex);
+ m_pending_deletion = true;
+ if (m_usage != 0)
+ return;
+ }
+
+ PROFILE(std::stringstream peerIdentifier1);
+ PROFILE(peerIdentifier1 << "runTimeouts[" << m_connection->getDesc() << ";" << id << ";RELIABLE]");
+ PROFILE(g_profiler->remove(peerIdentifier1.str()));
+ PROFILE(std::stringstream peerIdentifier2);
+ PROFILE(peerIdentifier2 << "sendPackets[" << m_connection->getDesc() << ";" << id << ";RELIABLE]");
+ PROFILE(ScopeProfiler peerprofiler(g_profiler, peerIdentifier2.str(), SPT_AVG));
+
+ delete this;
+}
+
+UDPPeer::UDPPeer(u16 a_id, Address a_address, Connection* connection) :
+ Peer(a_address,a_id,connection),
+ resend_timeout(0.5),
+ m_legacy_peer(true)
+{
+}
+
+UDPPeer::~UDPPeer()
+{
+}
+
+bool UDPPeer::getAddress(MTProtocols type,Address& toset)
+{
+ if ((type == UDP) || (type == MINETEST_RELIABLE_UDP) || (type == PRIMARY))
+ {
+ toset = address;
+ return true;
+ }
+
+ return false;
+}
+
+void UDPPeer::reportRTT(float rtt)
+{
+ if (rtt < 0.0) {
+ return;
+ }
+ RTTStatistics(rtt,"rudp",MAX_RELIABLE_WINDOW_SIZE*10);
+
+ float timeout = getStat(AVG_RTT) * RESEND_TIMEOUT_FACTOR;
+ if(timeout < RESEND_TIMEOUT_MIN)
+ timeout = RESEND_TIMEOUT_MIN;
+ if(timeout > RESEND_TIMEOUT_MAX)
+ timeout = RESEND_TIMEOUT_MAX;
+
+ JMutexAutoLock usage_lock(m_exclusive_access_mutex);
+ resend_timeout = timeout;
+}
+
+bool UDPPeer::Ping(float dtime,SharedBuffer<u8>& data)
+{
+ m_ping_timer += dtime;
+ if(m_ping_timer >= PING_TIMEOUT)
+ {
+ // Create and send PING packet
+ writeU8(&data[0], TYPE_CONTROL);
+ writeU8(&data[1], CONTROLTYPE_PING);
+ m_ping_timer = 0.0;
+ return true;
+ }
+ return false;
+}
+
+void UDPPeer::PutReliableSendCommand(ConnectionCommand &c,
+ unsigned int max_packet_size)
+{
+ if ( channels[c.channelnum].queued_commands.empty() &&
+ /* don't queue more packets then window size */
+ (channels[c.channelnum].queued_reliables.size()
+ < (channels[c.channelnum].getWindowSize()/2)))
+ {
+ LOG(dout_con<<m_connection->getDesc()
+ <<" processing reliable command for peer id: " << c.peer_id
+ <<" data size: " << c.data.getSize() << std::endl);
+ if (!processReliableSendCommand(c,max_packet_size))
+ {
+ channels[c.channelnum].queued_commands.push_back(c);
+ }
+ }
+ else
+ {
+ LOG(dout_con<<m_connection->getDesc()
+ <<" Queueing reliable command for peer id: " << c.peer_id
+ <<" data size: " << c.data.getSize() <<std::endl);
+ channels[c.channelnum].queued_commands.push_back(c);
+ }
+}
+
+bool UDPPeer::processReliableSendCommand(
+ ConnectionCommand &c,
+ unsigned int max_packet_size)
+{
+ u32 chunksize_max = max_packet_size
+ - BASE_HEADER_SIZE
+ - RELIABLE_HEADER_SIZE;
+
+ assert(c.data.getSize() < MAX_RELIABLE_WINDOW_SIZE*512);
+
+ std::list<SharedBuffer<u8> > originals;
+ u16 split_sequence_number = channels[c.channelnum].readNextSplitSeqNum();
+
+ if (c.raw)
+ {
+ originals.push_back(c.data);
+ }
+ else {
+ originals = makeAutoSplitPacket(c.data, chunksize_max,split_sequence_number);
+ channels[c.channelnum].setNextSplitSeqNum(split_sequence_number);
+ }
+
+ bool have_sequence_number = true;
+ bool have_initial_sequence_number = false;
+ Queue<BufferedPacket> toadd;
+ volatile u16 initial_sequence_number = 0;
+
+ for(std::list<SharedBuffer<u8> >::iterator i = originals.begin();
+ i != originals.end(); ++i)
+ {
+ u16 seqnum = channels[c.channelnum].getOutgoingSequenceNumber(have_sequence_number);
+
+ /* oops, we don't have enough sequence numbers to send this packet */
+ if (!have_sequence_number)
+ break;
+
+ if (!have_initial_sequence_number)
+ {
+ initial_sequence_number = seqnum;
+ have_initial_sequence_number = true;
+ }
+
+ SharedBuffer<u8> reliable = makeReliablePacket(*i, seqnum);
+
+ // Add base headers and make a packet
+ BufferedPacket p = con::makePacket(address, reliable,
+ m_connection->GetProtocolID(), m_connection->GetPeerID(),
+ c.channelnum);
+
+ toadd.push_back(p);
+ }
+
+ if (have_sequence_number) {
+ volatile u16 pcount = 0;
+ while(toadd.size() > 0) {
+ BufferedPacket p = toadd.pop_front();
+// LOG(dout_con<<connection->getDesc()
+// << " queuing reliable packet for peer_id: " << c.peer_id
+// << " channel: " << (c.channelnum&0xFF)
+// << " seqnum: " << readU16(&p.data[BASE_HEADER_SIZE+1])
+// << std::endl)
+ channels[c.channelnum].queued_reliables.push_back(p);
+ pcount++;
+ }
+ assert(channels[c.channelnum].queued_reliables.size() < 0xFFFF);
+ return true;
+ }
+ else {
+ volatile u16 packets_available = toadd.size();
+ /* we didn't get a single sequence number no need to fill queue */
+ if (!have_initial_sequence_number)
+ {
+ return false;
+ }
+ while(toadd.size() > 0) {
+ /* remove packet */
+ toadd.pop_front();
+
+ bool successfully_put_back_sequence_number
+ = channels[c.channelnum].putBackSequenceNumber(
+ (initial_sequence_number+toadd.size() % (SEQNUM_MAX+1)));
+
+ assert(successfully_put_back_sequence_number);
+ }
+ LOG(dout_con<<m_connection->getDesc()
+ << " Windowsize exceeded on reliable sending " << c.data.getSize() << " bytes"
+ << std::endl << "\t\tinitial_sequence_number: " << initial_sequence_number
+ << std::endl << "\t\tgot at most : " << packets_available << " packets"
+ << std::endl << "\t\tpackets queued : " << channels[c.channelnum].outgoing_reliables_sent.size()
+ << std::endl);
+ return false;
}
- } // for
}
-void Connection::runTimeouts(float dtime)
+void UDPPeer::RunCommandQueues(
+ unsigned int max_packet_size,
+ unsigned int maxcommands,
+ unsigned int maxtransfer)
{
- float congestion_control_aim_rtt
- = g_settings->getFloat("congestion_control_aim_rtt");
- float congestion_control_max_rate
- = g_settings->getFloat("congestion_control_max_rate");
- float congestion_control_min_rate
- = g_settings->getFloat("congestion_control_min_rate");
+ for (unsigned int i = 0; i < CHANNEL_COUNT; i++)
+ {
+ unsigned int commands_processed = 0;
+
+ if ((channels[i].queued_commands.size() > 0) &&
+ (channels[i].queued_reliables.size() < maxtransfer) &&
+ (commands_processed < maxcommands))
+ {
+ try {
+ ConnectionCommand c = channels[i].queued_commands.pop_front();
+ LOG(dout_con<<m_connection->getDesc()
+ <<" processing queued reliable command "<<std::endl);
+ if (!processReliableSendCommand(c,max_packet_size)) {
+ LOG(dout_con<<m_connection->getDesc()
+ << " Failed to queue packets for peer_id: " << c.peer_id
+ << ", delaying sending of " << c.data.getSize() << " bytes" << std::endl);
+ channels[i].queued_commands.push_front(c);
+ }
+ }
+ catch (ItemNotFoundException e) {
+ // intentionally empty
+ }
+ }
+ }
+}
+
+u16 UDPPeer::getNextSplitSequenceNumber(u8 channel)
+{
+ assert(channel < CHANNEL_COUNT);
+ return channels[channel].readNextIncomingSeqNum();
+}
+
+void UDPPeer::setNextSplitSequenceNumber(u8 channel, u16 seqnum)
+{
+ assert(channel < CHANNEL_COUNT);
+ channels[channel].setNextSplitSeqNum(seqnum);
+}
+
+SharedBuffer<u8> UDPPeer::addSpiltPacket(u8 channel,
+ BufferedPacket toadd,
+ bool reliable)
+{
+ assert(channel < CHANNEL_COUNT);
+ return channels[channel].incoming_splits.insert(toadd,reliable);
+}
+
+/******************************************************************************/
+/* Connection Threads */
+/******************************************************************************/
+
+ConnectionSendThread::ConnectionSendThread(Connection* parent,
+ unsigned int max_packet_size,
+ float timeout) :
+ m_connection(parent),
+ m_max_packet_size(max_packet_size),
+ m_timeout(timeout),
+ m_max_commands_per_iteration(1),
+ m_max_data_packets_per_iteration(g_settings->getU16("max_packets_per_iteration")),
+ m_max_packets_requeued(256)
+{
+}
+
+void * ConnectionSendThread::Thread()
+{
+ ThreadStarted();
+ log_register_thread("ConnectionSend");
+
+ LOG(dout_con<<m_connection->getDesc()
+ <<"ConnectionSend thread started"<<std::endl);
+
+ u32 curtime = porting::getTimeMs();
+ u32 lasttime = curtime;
+
+ PROFILE(std::stringstream ThreadIdentifier);
+ PROFILE(ThreadIdentifier << "ConnectionSend: [" << m_connection->getDesc() << "]");
+
+ /* if stop is requested don't stop immediately but try to send all */
+ /* packets first */
+ while(!StopRequested() || packetsQueued()) {
+ BEGIN_DEBUG_EXCEPTION_HANDLER
+ PROFILE(ScopeProfiler sp(g_profiler, ThreadIdentifier.str(), SPT_AVG));
+
+ m_iteration_packets_avaialble = m_max_data_packets_per_iteration;
+
+ /* wait for trigger or timeout */
+ m_send_sleep_semaphore.Wait(50);
+
+ /* remove all triggers */
+ while(m_send_sleep_semaphore.Wait(0)) {}
+
+ lasttime = curtime;
+ curtime = porting::getTimeMs();
+ float dtime = CALC_DTIME(lasttime,curtime);
+
+ /* first do all the reliable stuff */
+ runTimeouts(dtime);
+
+ /* translate commands to packets */
+ ConnectionCommand c = m_connection->m_command_queue.pop_frontNoEx(0);
+ while(c.type != CONNCMD_NONE)
+ {
+ if (c.reliable)
+ processReliableCommand(c);
+ else
+ processNonReliableCommand(c);
+
+ c = m_connection->m_command_queue.pop_frontNoEx(0);
+ }
+
+ /* send non reliable packets */
+ sendPackets(dtime);
+
+ END_DEBUG_EXCEPTION_HANDLER(derr_con);
+ }
+
+ PROFILE(g_profiler->remove(ThreadIdentifier.str()));
+ return NULL;
+}
+
+void ConnectionSendThread::Trigger()
+{
+ m_send_sleep_semaphore.Post();
+}
+
+bool ConnectionSendThread::packetsQueued()
+{
+ std::list<u16> peerIds = m_connection->getPeerIDs();
+
+ if ((this->m_outgoing_queue.size() > 0) && (peerIds.size() > 0))
+ return true;
+
+ for(std::list<u16>::iterator j = peerIds.begin();
+ j != peerIds.end(); ++j)
+ {
+ PeerHelper peer = m_connection->getPeerNoEx(*j);
+
+ if (!peer)
+ continue;
+
+ if (dynamic_cast<UDPPeer*>(&peer) == 0)
+ continue;
+
+ for(u16 i=0; i<CHANNEL_COUNT; i++)
+ {
+ Channel *channel = &(dynamic_cast<UDPPeer*>(&peer))->channels[i];
+
+ if (channel->queued_commands.size() > 0)
+ {
+ return true;
+ }
+ }
+ }
+
+
+ return false;
+}
+
+void ConnectionSendThread::runTimeouts(float dtime)
+{
std::list<u16> timeouted_peers;
- for(std::map<u16, Peer*>::iterator j = m_peers.begin();
- j != m_peers.end(); ++j)
+ std::list<u16> peerIds = m_connection->getPeerIDs();
+
+ for(std::list<u16>::iterator j = peerIds.begin();
+ j != peerIds.end(); ++j)
{
- Peer *peer = j->second;
+ PeerHelper peer = m_connection->getPeerNoEx(*j);
+
+ if (!peer)
+ continue;
+
+ if(dynamic_cast<UDPPeer*>(&peer) == 0)
+ continue;
+
+ PROFILE(std::stringstream peerIdentifier);
+ PROFILE(peerIdentifier << "runTimeouts[" << m_connection->getDesc() << ";" << *j << ";RELIABLE]");
+ PROFILE(ScopeProfiler peerprofiler(g_profiler, peerIdentifier.str(), SPT_AVG));
+
+ SharedBuffer<u8> data(2); // data for sending ping, required here because of goto
- // Update congestion control values
- peer->congestion_control_aim_rtt = congestion_control_aim_rtt;
- peer->congestion_control_max_rate = congestion_control_max_rate;
- peer->congestion_control_min_rate = congestion_control_min_rate;
-
/*
Check peer timeout
*/
- peer->timeout_counter += dtime;
- if(peer->timeout_counter > m_timeout)
+ if(peer->isTimedOut(m_timeout))
{
- PrintInfo(derr_con);
-
derr_con<<"RunTimeouts(): Peer "<<peer->id
+ LOG(derr_con<<m_connection->getDesc()
+
<<"RunTimeouts(): Peer "<<peer->id
<<" has timed out."
<<" (source=peer->timeout_counter)"
- <<std::endl;
+ <<std::endl);
// Add peer to the list
timeouted_peers.push_back(peer->id);
// Don't bother going through the buffers of this one
continue;
}
- float resend_timeout = peer->resend_timeout;
+ float resend_timeout = dynamic_cast<UDPPeer*>(&peer)->getResendTimeout();
for(u16 i=0; i<CHANNEL_COUNT; i++)
{
std::list<BufferedPacket> timed_outs;
-
- Channel *channel = &peer->channels[i];
+ Channel *channel = &(dynamic_cast<UDPPeer*>(&peer))->channels[i];
+
+ if (dynamic_cast<UDPPeer*>(&peer)->getLegacyPeer())
+ channel->setWindowSize(g_settings->getU16("workaround_window_size"));
// Remove timed out incomplete unreliable split packets
channel->incoming_splits.removeUnreliableTimedOuts(dtime, m_timeout);
-
+
// Increment reliable packet times
- channel->outgoing_reliables.incrementTimeouts(dtime);
+ channel->outgoing_reliables_sent.incrementTimeouts(dtime);
- // Check reliable packet total times, remove peer if
- // over timeout.
- if(channel->outgoing_reliables.anyTotaltimeReached(m_timeout))
- {
- PrintInfo(derr_con);
- derr_con<<"RunTimeouts(): Peer "<<peer->id
- <<" has timed out."
- <<" (source=reliable packet totaltime)"
- <<std::endl;
- // Add peer to the to-be-removed list
- timeouted_peers.push_back(peer->id);
- goto nextpeer;
- }
+ unsigned int numpeers = m_connection->m_peers.size();
+
+ if (numpeers == 0)
+ return;
// Re-send timed out outgoing reliables
-
timed_outs = channel->
- outgoing_reliables.getTimedOuts(resend_timeout);
+ outgoing_reliables_sent.getTimedOuts(resend_timeout,
+ (m_max_data_packets_per_iteration/numpeers));
- channel->outgoing_reliables.resetTimedOuts(resend_timeout);
+ channel->UpdatePacketLossCounter(timed_outs.size());
+
+ m_iteration_packets_avaialble -= timed_outs.size();
for(std::list<BufferedPacket>::iterator j = timed_outs.begin();
j != timed_outs.end(); ++j)
{
u16 peer_id = readPeerId(*(j->data));
- u8 channel = readChannel(*(j->data));
- u16 seqnum = readU16(&(j->data[BASE_HEADER_SIZE+1]));
+ u8 channelnum = readChannel(*(j->data));
+ u16 seqnum = readU16(&(j->data[BASE_HEADER_SIZE+1]));
+
+ channel->UpdateBytesLost(j->data.getSize());
+
+ LOG(derr_con<<m_connection->getDesc()
+ <<"RE-SENDING timed-out RELIABLE to "
+ << j->address.serializeString()
+ << "(t/o="<<resend_timeout<<"): "
+ <<"from_peer_id="<<peer_id
+ <<", channel="<<((int)channelnum&0xff)
+ <<", seqnum="<<seqnum
+ <<std::endl);
+
+ rawSend(*j);
+
+ // do not handle rtt here as we can't decide if this packet was
+ // lost or really takes more time to transmit
+ }
+
+ if (!dynamic_cast<UDPPeer*>(&peer)->getLegacyPeer())
+ channel->UpdateTimers(dtime);
+ }
+
+ /* send ping if necessary */
+ if (dynamic_cast<UDPPeer*>(&peer)->Ping(dtime,data)) {
+ LOG(dout_con<<m_connection->getDesc()
+ <<"Sending ping for peer_id: "
+ << dynamic_cast<UDPPeer*>(&peer)->id <<std::endl);
+ /* this may fail if there ain't a sequence number left */
+ if (!rawSendAsPacket(dynamic_cast<UDPPeer*>(&peer)->id, 0, data, true))
+ {
+ //retrigger with reduced ping interval
+ dynamic_cast<UDPPeer*>(&peer)->Ping(4.0,data);
+ }
+ }
+
+ dynamic_cast<UDPPeer*>(&peer)->RunCommandQueues(m_max_packet_size,
+ m_max_commands_per_iteration,
+ m_max_packets_requeued);
+ }
+
+ // Remove timed out peers
+ for(std::list<u16>::iterator i = timeouted_peers.begin();
+ i != timeouted_peers.end(); ++i)
+ {
+ LOG(derr_con<<m_connection->getDesc()
+ <<"RunTimeouts(): Removing peer "<<(*i)<<std::endl);
+ m_connection->deletePeer(*i, true);
+ }
+}
+
+void ConnectionSendThread::rawSend(const BufferedPacket &packet)
+{
+ try{
+ m_connection->m_udpSocket.Send(packet.address, *packet.data, packet.data.getSize());
+ LOG(dout_con <<m_connection->getDesc()
+ << " rawSend: " << packet.data.getSize() << " bytes sent" << std::endl);
+ } catch(SendFailedException &e){
+ LOG(derr_con<<m_connection->getDesc()
+ <<"Connection::rawSend(): SendFailedException: "
+ <<packet.address.serializeString()<<std::endl);
+ }
+}
+
+void ConnectionSendThread::sendAsPacketReliable(BufferedPacket& p, Channel* channel)
+{
+ try{
+ p.absolute_send_time = porting::getTimeMs();
+ // Buffer the packet
+ channel->outgoing_reliables_sent.insert(p,
+ (channel->readOutgoingSequenceNumber() - MAX_RELIABLE_WINDOW_SIZE) % (MAX_RELIABLE_WINDOW_SIZE+1));
+ }
+ catch(AlreadyExistsException &e)
+ {
+ LOG(derr_con<<m_connection->getDesc()
+ <<"WARNING: Going to send a reliable packet"
+ <<" in outgoing buffer" <<std::endl);
+ //assert(0);
+ }
+
+ // Send the packet
+ rawSend(p);
+}
+
+bool ConnectionSendThread::rawSendAsPacket(u16 peer_id, u8 channelnum,
+ SharedBuffer<u8> data, bool reliable)
+{
+ PeerHelper peer = m_connection->getPeerNoEx(peer_id);
+ if(!peer) {
+ LOG(dout_con<<m_connection->getDesc()
+ <<" INFO: dropped packet for non existent peer_id: " << peer_id << std::endl);
+ assert(reliable && "trying to send raw packet reliable but no peer found!");
+ return false;
+ }
+ Channel *channel = &(dynamic_cast<UDPPeer*>(&peer)->channels[channelnum]);
+
+ if(reliable)
+ {
+ bool have_sequence_number_for_raw_packet = true;
+ u16 seqnum = channel->getOutgoingSequenceNumber(have_sequence_number_for_raw_packet);
+
+ if (!have_sequence_number_for_raw_packet)
+ return false;
+
+ SharedBuffer<u8> reliable = makeReliablePacket(data, seqnum);
+ Address peer_address;
+ peer->getAddress(MINETEST_RELIABLE_UDP,peer_address);
+
+ // Add base headers and make a packet
+ BufferedPacket p = con::makePacket(peer_address, reliable,
+ m_connection->GetProtocolID(), m_connection->GetPeerID(),
+ channelnum);
+
+ // first check if our send window is already maxed out
+ if (channel->outgoing_reliables_sent.size()
+ < channel->getWindowSize()) {
+ LOG(dout_con<<m_connection->getDesc()
+ <<" INFO: sending a reliable packet to peer_id " << peer_id
+ <<" channel: " << channelnum
+ <<" seqnum: " << seqnum << std::endl);
+ sendAsPacketReliable(p,channel);
+ return true;
+ }
+ else {
+ LOG(dout_con<<m_connection->getDesc()
+ <<" INFO: queueing reliable packet for peer_id: " << peer_id
+ <<" channel: " << channelnum
+ <<" seqnum: " << seqnum << std::endl);
+ channel->queued_reliables.push_back(p);
+ return false;
+ }
+ }
+ else
+ {
+ Address peer_address;
+
+ if (peer->getAddress(UDP,peer_address))
+ {
+ // Add base headers and make a packet
+ BufferedPacket p = con::makePacket(peer_address, data,
+ m_connection->GetProtocolID(), m_connection->GetPeerID(),
+ channelnum);
+
+ // Send the packet
+ rawSend(p);
+ return true;
+ }
+ else {
+ LOG(dout_con<<m_connection->getDesc()
+ <<" INFO: dropped unreliable packet for peer_id: " << peer_id
+ <<" because of (yet) missing udp address" << std::endl);
+ return false;
+ }
+ }
+
+ //never reached
+ return false;
+}
+
+void ConnectionSendThread::processReliableCommand(ConnectionCommand &c)
+{
+ assert(c.reliable);
+
+ switch(c.type){
+ case CONNCMD_NONE:
+ LOG(dout_con<<m_connection->getDesc()<<"UDP processing reliable CONNCMD_NONE"<<std::endl);
+ return;
- PrintInfo(derr_con);
- derr_con<<"RE-SENDING timed-out RELIABLE to ";
- j->address.print(&derr_con);
- derr_con<<"(t/o="<<resend_timeout<<"): "
- <<"from_peer_id="<<peer_id
- <<", channel="<<((int)channel&0xff)
- <<", seqnum="<<seqnum
- <<std::endl;
+ case CONNCMD_SEND:
+ LOG(dout_con<<m_connection->getDesc()<<"UDP processing reliable CONNCMD_SEND"<<std::endl);
+ sendReliable(c);
+ return;
- rawSend(*j);
+ case CONNCMD_SEND_TO_ALL:
+ LOG(dout_con<<m_connection->getDesc()<<"UDP processing CONNCMD_SEND_TO_ALL"<<std::endl);
+ sendToAllReliable(c);
+ return;
- // Enlarge avg_rtt and resend_timeout:
- // The rtt will be at least the timeout.
- // NOTE: This won't affect the timeout of the next
- // checked channel because it was cached.
- peer->reportRTT(resend_timeout);
- }
- }
-
- /*
- Send pings
- */
- peer->ping_timer += dtime;
- if(peer->ping_timer >= 5.0)
+ case CONCMD_CREATE_PEER:
+ LOG(dout_con<<m_connection->getDesc()<<"UDP processing reliable CONCMD_CREATE_PEER"<<std::endl);
+ if (!rawSendAsPacket(c.peer_id,c.channelnum,c.data,c.reliable))
{
- // Create and send PING packet
- SharedBuffer<u8> data(2);
- writeU8(&data[0], TYPE_CONTROL);
- writeU8(&data[1], CONTROLTYPE_PING);
- rawSendAsPacket(peer->id, 0, data, true);
+ /* put to queue if we couldn't send it immediately */
+ sendReliable(c);
+ }
+ return;
- peer->ping_timer = 0.0;
+ case CONCMD_DISABLE_LEGACY:
+ LOG(dout_con<<m_connection->getDesc()<<"UDP processing reliable CONCMD_DISABLE_LEGACY"<<std::endl);
+ if (!rawSendAsPacket(c.peer_id,c.channelnum,c.data,c.reliable))
+ {
+ /* put to queue if we couldn't send it immediately */
+ sendReliable(c);
}
-
-nextpeer:
- continue;
+ return;
+
+ case CONNCMD_SERVE:
+ case CONNCMD_CONNECT:
+ case CONNCMD_DISCONNECT:
+ case CONNCMD_DELETE_PEER:
+ case CONCMD_ACK:
+ assert("Got command that shouldn't be reliable as reliable command" == 0);
+ default:
+ LOG(dout_con<<m_connection->getDesc()<<" Invalid reliable command type: " << c.type <<std::endl);
}
+}
- // Remove timed out peers
- for(std::list<u16>::iterator i = timeouted_peers.begin();
- i != timeouted_peers.end(); ++i)
- {
- PrintInfo(derr_con);
- derr_con<<"RunTimeouts(): Removing peer "<<(*i)<<std::endl;
- deletePeer(*i, true);
+
+void ConnectionSendThread::processNonReliableCommand(ConnectionCommand &c)
+{
+ assert(!c.reliable);
+
+ switch(c.type){
+ case CONNCMD_NONE:
+ LOG(dout_con<<m_connection->getDesc()<<" UDP processing CONNCMD_NONE"<<std::endl);
+ return;
+ case CONNCMD_SERVE:
+ LOG(dout_con<<m_connection->getDesc()<<" UDP processing CONNCMD_SERVE port="
+ <<c.port<<std::endl);
+ serve(c.port);
+ return;
+ case CONNCMD_CONNECT:
+ LOG(dout_con<<m_connection->getDesc()<<" UDP processing CONNCMD_CONNECT"<<std::endl);
+ connect(c.address);
+ return;
+ case CONNCMD_DISCONNECT:
+ LOG(dout_con<<m_connection->getDesc()<<" UDP processing CONNCMD_DISCONNECT"<<std::endl);
+ disconnect();
+ return;
+ case CONNCMD_DISCONNECT_PEER:
+ LOG(dout_con<<m_connection->getDesc()<<" UDP processing CONNCMD_DISCONNECT_PEER"<<std::endl);
+ disconnect_peer(c.peer_id);
+ return;
+ case CONNCMD_SEND:
+ LOG(dout_con<<m_connection->getDesc()<<" UDP processing CONNCMD_SEND"<<std::endl);
+ send(c.peer_id, c.channelnum, c.data);
+ return;
+ case CONNCMD_SEND_TO_ALL:
+ LOG(dout_con<<m_connection->getDesc()<<" UDP processing CONNCMD_SEND_TO_ALL"<<std::endl);
+ sendToAll(c.channelnum, c.data);
+ return;
+ case CONNCMD_DELETE_PEER:
+ LOG(dout_con<<m_connection->getDesc()<<" UDP processing CONNCMD_DELETE_PEER"<<std::endl);
+ m_connection->deletePeer(c.peer_id, false);
+ return;
+ case CONCMD_ACK:
+ LOG(dout_con<<m_connection->getDesc()<<" UDP processing CONCMD_ACK"<<std::endl);
+ sendAsPacket(c.peer_id,c.channelnum,c.data,true);
+ return;
+ case CONCMD_CREATE_PEER:
+ assert("Got command that should be reliable as unreliable command" == 0);
+ default:
+ LOG(dout_con<<m_connection->getDesc()<<" Invalid command type: " << c.type <<std::endl);
}
}
-void Connection::serve(u16 port)
+void ConnectionSendThread::serve(u16 port)
{
- dout_con<<getDesc()<<" serving at port "<<port<<std::endl;
+ LOG(dout_con<<m_connection->getDesc()<<"UDP serving at port "<<port<<std::endl);
try{
- m_socket.Bind(port);
- m_peer_id = PEER_ID_SERVER;
+ m_connection->m_udpSocket.Bind(port);
+ m_connection->SetPeerID(PEER_ID_SERVER);
}
catch(SocketException &e){
// Create event
ConnectionEvent ce;
ce.bindFailed();
- putEvent(ce);
+ m_connection->putEvent(ce);
}
}
-void Connection::connect(Address address)
+void ConnectionSendThread::connect(Address address)
{
- dout_con<<getDesc()<<" connecting to "<<address.serializeString()
- <<":"<<address.getPort()<<std::endl;
-
- std::map<u16, Peer*>::iterator node = m_peers.find(PEER_ID_SERVER);
- if(node != m_peers.end()){
- throw ConnectionException("Already connected to a server");
- }
+ LOG(dout_con<<m_connection->getDesc()<<" connecting to "<<address.serializeString()
+ <<":"<<address.getPort()<<std::endl);
- Peer *peer = new Peer(PEER_ID_SERVER, address);
- m_peers[peer->id] = peer;
+ UDPPeer *peer = m_connection->createServerPeer(address);
// Create event
ConnectionEvent e;
e.peerAdded(peer->id, peer->address);
- putEvent(e);
-
- m_socket.Bind(0);
-
+ m_connection->putEvent(e);
+
+ m_connection->m_udpSocket.Bind(0);
+
// Send a dummy packet to server with peer_id = PEER_ID_INEXISTENT
- m_peer_id = PEER_ID_INEXISTENT;
+ m_connection->SetPeerID(PEER_ID_INEXISTENT);
SharedBuffer<u8> data(0);
- Send(PEER_ID_SERVER, 0, data, true);
+ m_connection->Send(PEER_ID_SERVER, 0, data, true);
}
-void Connection::disconnect()
+void ConnectionSendThread::disconnect()
{
- dout_con<<getDesc()<<" disconnecting"<<std::endl;
+ LOG(dout_con<<m_connection->getDesc()<<" disconnecting"<<std::endl);
// Create and send DISCO packet
SharedBuffer<u8> data(2);
writeU8(&data[0], TYPE_CONTROL);
writeU8(&data[1], CONTROLTYPE_DISCO);
-
+
+
// Send to all
- for(std::map<u16, Peer*>::iterator j = m_peers.begin();
- j != m_peers.end(); ++j)
+ std::list<u16> peerids = m_connection->getPeerIDs();
+
+ for (std::list<u16>::iterator i = peerids.begin();
+ i != peerids.end();
+ i++)
{
- Peer *peer = j->second;
- rawSendAsPacket(peer->id, 0, data, false);
+ sendAsPacket(*i, 0,data,false);
}
}
-void Connection::sendToAll(u8 channelnum, SharedBuffer<u8> data, bool reliable)
+void ConnectionSendThread::disconnect_peer(u16 peer_id)
{
- for(std::map<u16, Peer*>::iterator j = m_peers.begin();
- j != m_peers.end(); ++j)
- {
- Peer *peer = j->second;
- send(peer->id, channelnum, data, reliable);
- }
+ LOG(dout_con<<m_connection->getDesc()<<" disconnecting peer"<<std::endl);
+
+ // Create and send DISCO packet
+ SharedBuffer<u8> data(2);
+ writeU8(&data[0], TYPE_CONTROL);
+ writeU8(&data[1], CONTROLTYPE_DISCO);
+ sendAsPacket(peer_id, 0,data,false);
}
-void Connection::send(u16 peer_id, u8 channelnum,
- SharedBuffer<u8> data, bool reliable)
+void ConnectionSendThread::send(u16 peer_id, u8 channelnum,
+ SharedBuffer<u8> data)
{
- dout_con<<getDesc()<<" sending to peer_id="<<peer_id<<std::endl;
-
assert(channelnum < CHANNEL_COUNT);
-
- Peer *peer = getPeerNoEx(peer_id);
- if(peer == NULL)
+
+ PeerHelper peer = m_connection->getPeerNoEx(peer_id);
+ if(!peer) {
+ LOG(dout_con<<m_connection->getDesc()<<" peer: peer_id="<<peer_id
+ << ">>>NOT<<< found on sending packet"
+ << ", channel " << (channelnum % 0xFF)
+ << ", size: " << data.getSize() <<std::endl);
return;
- Channel *channel = &(peer->channels[channelnum]);
+ }
- u32 chunksize_max = m_max_packet_size - BASE_HEADER_SIZE;
- if(reliable)
- chunksize_max -= RELIABLE_HEADER_SIZE;
+ LOG(dout_con<<m_connection->getDesc()<<" sending to peer_id="<<peer_id
+ << ", channel " << (channelnum % 0xFF)
+ << ", size: " << data.getSize() <<std::endl);
+
+ u16 split_sequence_number = peer->getNextSplitSequenceNumber(channelnum);
+ u32 chunksize_max = m_max_packet_size - BASE_HEADER_SIZE;
std::list<SharedBuffer<u8> > originals;
- originals = makeAutoSplitPacket(data, chunksize_max,
- channel->next_outgoing_split_seqnum);
-
+
+ originals = makeAutoSplitPacket(data, chunksize_max,split_sequence_number);
+
+ peer->setNextSplitSequenceNumber(channelnum,split_sequence_number);
+
for(std::list<SharedBuffer<u8> >::iterator i = originals.begin();
i != originals.end(); ++i)
{
SharedBuffer<u8> original = *i;
-
- sendAsPacket(peer_id, channelnum, original, reliable);
+ sendAsPacket(peer_id, channelnum, original);
}
}
-void Connection::sendAsPacket(u16 peer_id, u8 channelnum,
- SharedBuffer<u8> data, bool reliable)
+void ConnectionSendThread::sendReliable(ConnectionCommand &c)
{
- OutgoingPacket packet(peer_id, channelnum, data, reliable);
- m_outgoing_queue.push_back(packet);
+ PeerHelper peer = m_connection->getPeerNoEx(c.peer_id);
+ if (!peer)
+ return;
+
+ peer->PutReliableSendCommand(c,m_max_packet_size);
}
-void Connection::rawSendAsPacket(u16 peer_id, u8 channelnum,
- SharedBuffer<u8> data, bool reliable)
+void ConnectionSendThread::sendToAll(u8 channelnum, SharedBuffer<u8> data)
{
- Peer *peer = getPeerNoEx(peer_id);
- if(!peer)
- return;
- Channel *channel = &(peer->channels[channelnum]);
+ std::list<u16> peerids = m_connection->getPeerIDs();
- if(reliable)
+ for (std::list<u16>::iterator i = peerids.begin();
+ i != peerids.end();
+ i++)
{
- u16 seqnum = channel->next_outgoing_seqnum;
- channel->next_outgoing_seqnum++;
+ send(*i, channelnum, data);
+ }
+}
- SharedBuffer<u8> reliable = makeReliablePacket(data, seqnum);
+void ConnectionSendThread::sendToAllReliable(ConnectionCommand &c)
+{
+ std::list<u16> peerids = m_connection->getPeerIDs();
- // Add base headers and make a packet
- BufferedPacket p = makePacket(peer->address, reliable,
- m_protocol_id, m_peer_id, channelnum);
-
- try{
- // Buffer the packet
- channel->outgoing_reliables.insert(p);
+ for (std::list<u16>::iterator i = peerids.begin();
+ i != peerids.end();
+ i++)
+ {
+ PeerHelper peer = m_connection->getPeerNoEx(*i);
+
+ if (!peer)
+ continue;
+
+ peer->PutReliableSendCommand(c,m_max_packet_size);
+ }
+}
+
+void ConnectionSendThread::sendPackets(float dtime)
+{
+ std::list<u16> peerIds = m_connection->getPeerIDs();
+
+ for(std::list<u16>::iterator
+ j = peerIds.begin();
+ j != peerIds.end(); ++j)
+ {
+ PeerHelper peer = m_connection->getPeerNoEx(*j);
+ //peer may have been removed
+ if (!peer) {
+ LOG(dout_con<<m_connection->getDesc()<< " Peer not found: peer_id=" << *j << std::endl);
+ continue;
+ }
+ peer->m_increment_packets_remaining = m_iteration_packets_avaialble/m_connection->m_peers.size();
+
+ if (dynamic_cast<UDPPeer*>(&peer) == 0)
+ {
+ continue;
}
- catch(AlreadyExistsException &e)
+
+ PROFILE(std::stringstream peerIdentifier);
+ PROFILE(peerIdentifier << "sendPackets[" << m_connection->getDesc() << ";" << *j << ";RELIABLE]");
+ PROFILE(ScopeProfiler peerprofiler(g_profiler, peerIdentifier.str(), SPT_AVG));
+
+ LOG(dout_con<<m_connection->getDesc()
+ << " Handle per peer queues: peer_id=" << *j
+ << " packet quota: " << peer->m_increment_packets_remaining << std::endl);
+ // first send queued reliable packets for all peers (if possible)
+ for (unsigned int i=0; i < CHANNEL_COUNT; i++)
{
- PrintInfo(derr_con);
- derr_con<<"WARNING: Going to send a reliable packet "
- "seqnum="<<seqnum<<" that is already "
- "in outgoing buffer"<<std::endl;
- //assert(0);
+ u16 next_to_ack = 0;
+ dynamic_cast<UDPPeer*>(&peer)->channels[i].outgoing_reliables_sent.getFirstSeqnum(next_to_ack);
+ u16 next_to_receive = 0;
+ dynamic_cast<UDPPeer*>(&peer)->channels[i].incoming_reliables.getFirstSeqnum(next_to_receive);
+
+ LOG(dout_con<<m_connection->getDesc()<< "\t channel: "
+ << i << ", peer quota:"
+ << peer->m_increment_packets_remaining
+ << std::endl
+ << "\t\t\treliables on wire: "
+ << dynamic_cast<UDPPeer*>(&peer)->channels[i].outgoing_reliables_sent.size()
+ << ", waiting for ack for " << next_to_ack
+ << std::endl
+ << "\t\t\tincoming_reliables: "
+ << dynamic_cast<UDPPeer*>(&peer)->channels[i].incoming_reliables.size()
+ << ", next reliable packet: "
+ << dynamic_cast<UDPPeer*>(&peer)->channels[i].readNextIncomingSeqNum()
+ << ", next queued: " << next_to_receive
+ << std::endl
+ << "\t\t\treliables queued : "
+ << dynamic_cast<UDPPeer*>(&peer)->channels[i].queued_reliables.size()
+ << std::endl
+ << "\t\t\tqueued commands : "
+ << dynamic_cast<UDPPeer*>(&peer)->channels[i].queued_commands.size()
+ << std::endl);
+
+ while ((dynamic_cast<UDPPeer*>(&peer)->channels[i].queued_reliables.size() > 0) &&
+ (dynamic_cast<UDPPeer*>(&peer)->channels[i].outgoing_reliables_sent.size()
+ < dynamic_cast<UDPPeer*>(&peer)->channels[i].getWindowSize())&&
+ (peer->m_increment_packets_remaining > 0))
+ {
+ BufferedPacket p = dynamic_cast<UDPPeer*>(&peer)->channels[i].queued_reliables.pop_front();
+ Channel* channel = &(dynamic_cast<UDPPeer*>(&peer)->channels[i]);
+ LOG(dout_con<<m_connection->getDesc()
+ <<" INFO: sending a queued reliable packet "
+ <<" channel: " << i
+ <<", seqnum: " << readU16(&p.data[BASE_HEADER_SIZE+1])
+ << std::endl);
+ sendAsPacketReliable(p,channel);
+ peer->m_increment_packets_remaining--;
+ }
}
-
- // Send the packet
- rawSend(p);
}
- else
+
+ if (m_outgoing_queue.size())
{
- // Add base headers and make a packet
- BufferedPacket p = makePacket(peer->address, data,
- m_protocol_id, m_peer_id, channelnum);
+ LOG(dout_con<<m_connection->getDesc()
+ << " Handle non reliable queue ("
+ << m_outgoing_queue.size() << " pkts)" << std::endl);
+ }
+
+ /* send non reliable packets*/
+ for(unsigned int i=0;i < m_outgoing_queue.size();i++) {
+ OutgoingPacket packet = m_outgoing_queue.pop_front();
+
+ assert(!packet.reliable &&
+ "reliable packets are not allowed in outgoing queue!");
- // Send the packet
- rawSend(p);
+ PeerHelper peer = m_connection->getPeerNoEx(packet.peer_id);
+ if(!peer) {
+ LOG(dout_con<<m_connection->getDesc()
+ <<" Outgoing queue: peer_id="<<packet.peer_id
+ << ">>>NOT<<< found on sending packet"
+ << ", channel " << (packet.channelnum % 0xFF)
+ << ", size: " << packet.data.getSize() <<std::endl);
+ continue;
+ }
+ /* send acks immediately */
+ else if (packet.ack)
+ {
+ rawSendAsPacket(packet.peer_id, packet.channelnum,
+ packet.data, packet.reliable);
+ peer->m_increment_packets_remaining =
+ MYMIN(0,peer->m_increment_packets_remaining--);
+ }
+ else if (
+ ( peer->m_increment_packets_remaining > 0) ||
+ (StopRequested())){
+ rawSendAsPacket(packet.peer_id, packet.channelnum,
+ packet.data, packet.reliable);
+ peer->m_increment_packets_remaining--;
+ }
+ else {
+ m_outgoing_queue.push_back(packet);
+ }
}
}
-void Connection::rawSend(const BufferedPacket &packet)
+void ConnectionSendThread::sendAsPacket(u16 peer_id, u8 channelnum,
+ SharedBuffer<u8> data, bool ack)
{
- try{
- m_socket.Send(packet.address, *packet.data, packet.data.getSize());
- } catch(SendFailedException &e){
- derr_con<<"Connection::rawSend(): SendFailedException: "
- <<packet.address.serializeString()<<std::endl;
- }
+ OutgoingPacket packet(peer_id, channelnum, data, false, ack);
+ m_outgoing_queue.push_back(packet);
}
-Peer* Connection::getPeer(u16 peer_id)
+ConnectionReceiveThread::ConnectionReceiveThread(Connection* parent,
+ unsigned int max_packet_size) :
+ m_connection(parent),
+ m_max_packet_size(max_packet_size)
{
- std::map<u16, Peer*>::iterator node = m_peers.find(peer_id);
+}
- if(node == m_peers.end()){
- throw PeerNotFoundException("GetPeer: Peer not found (possible timeout)");
- }
+void * ConnectionReceiveThread::Thread()
+{
+ ThreadStarted();
+ log_register_thread("ConnectionReceive");
- // Error checking
- assert(node->second->id == peer_id);
+ LOG(dout_con<<m_connection->getDesc()
+ <<"ConnectionReceive thread started"<<std::endl);
+
+ PROFILE(std::stringstream ThreadIdentifier);
+ PROFILE(ThreadIdentifier << "ConnectionReceive: [" << m_connection->getDesc() << "]");
+
+#ifdef DEBUG_CONNECTION_KBPS
+ u32 curtime = porting::getTimeMs();
+ u32 lasttime = curtime;
+ float debug_print_timer = 0.0;
+#endif
- return node->second;
+ while(!StopRequested()) {
+ BEGIN_DEBUG_EXCEPTION_HANDLER
+ PROFILE(ScopeProfiler sp(g_profiler, ThreadIdentifier.str(), SPT_AVG));
+
+#ifdef DEBUG_CONNECTION_KBPS
+ lasttime = curtime;
+ curtime = porting::getTimeMs();
+ float dtime = CALC_DTIME(lasttime,curtime);
+#endif
+
+ /* receive packets */
+ receive();
+
+#ifdef DEBUG_CONNECTION_KBPS
+ debug_print_timer += dtime;
+ if (debug_print_timer > 20.0) {
+ debug_print_timer -= 20.0;
+
+ std::list<u16> peerids = m_connection->getPeerIDs();
+
+ for (std::list<u16>::iterator i = peerids.begin();
+ i != peerids.end();
+ i++)
+ {
+ PeerHelper peer = m_connection->getPeerNoEx(*i);
+ if (!peer)
+ continue;
+
+ float peer_current = 0.0;
+ float peer_loss = 0.0;
+ float avg_rate = 0.0;
+ float avg_loss = 0.0;
+
+ for(u16 j=0; j<CHANNEL_COUNT; j++)
+ {
+ peer_current +=peer->channels[j].getCurrentDownloadRateKB();
+ peer_loss += peer->channels[j].getCurrentLossRateKB();
+ avg_rate += peer->channels[j].getAvgDownloadRateKB();
+ avg_loss += peer->channels[j].getAvgLossRateKB();
+ }
+
+ std::stringstream output;
+ output << std::fixed << std::setprecision(1);
+ output << "OUT to Peer " << *i << " RATES (good / loss) " << std::endl;
+ output << "\tcurrent (sum): " << peer_current << "kb/s "<< peer_loss << "kb/s" << std::endl;
+ output << "\taverage (sum): " << avg_rate << "kb/s "<< avg_loss << "kb/s" << std::endl;
+ output << std::setfill(' ');
+ for(u16 j=0; j<CHANNEL_COUNT; j++)
+ {
+ output << "\tcha " << j << ":"
+ << " CUR: " << std::setw(6) << peer->channels[j].getCurrentDownloadRateKB() <<"kb/s"
+ << " AVG: " << std::setw(6) << peer->channels[j].getAvgDownloadRateKB() <<"kb/s"
+ << " MAX: " << std::setw(6) << peer->channels[j].getMaxDownloadRateKB() <<"kb/s"
+ << " /"
+ << " CUR: " << std::setw(6) << peer->channels[j].getCurrentLossRateKB() <<"kb/s"
+ << " AVG: " << std::setw(6) << peer->channels[j].getAvgLossRateKB() <<"kb/s"
+ << " MAX: " << std::setw(6) << peer->channels[j].getMaxLossRateKB() <<"kb/s"
+ << " / WS: " << peer->channels[j].getWindowSize()
+ << std::endl;
+ }
+
+ fprintf(stderr,"%s\n",output.str().c_str());
+ }
+ }
+#endif
+ END_DEBUG_EXCEPTION_HANDLER(derr_con);
+ }
+ PROFILE(g_profiler->remove(ThreadIdentifier.str()));
+ return NULL;
}
-Peer* Connection::getPeerNoEx(u16 peer_id)
+// Receive packets from the network and buffers and create ConnectionEvents
+void ConnectionReceiveThread::receive()
{
- std::map<u16, Peer*>::iterator node = m_peers.find(peer_id);
+ /* now reorder reliables */
+ u32 datasize = m_max_packet_size * 2; // Double it just to be safe
+ // TODO: We can not know how many layers of header there are.
+ // For now, just assume there are no other than the base headers.
+ u32 packet_maxsize = datasize + BASE_HEADER_SIZE;
+ SharedBuffer<u8> packetdata(packet_maxsize);
+
+ bool packet_queued = true;
- if(node == m_peers.end()){
- return NULL;
+ unsigned int loop_count = 0;
+
+ /* first of all read packets from socket */
+ /* check for incoming data available */
+ while( (loop_count < 10) &&
+ (m_connection->m_udpSocket.WaitData(50)))
+ {
+ loop_count++;
+ try{
+ if (packet_queued)
+ {
+ bool no_data_left = false;
+ u16 peer_id;
+ SharedBuffer<u8> resultdata;
+ while(!no_data_left)
+ {
+ try {
+ no_data_left = !getFromBuffers(peer_id, resultdata);
+ if (!no_data_left) {
+ ConnectionEvent e;
+ e.dataReceived(peer_id, resultdata);
+ m_connection->putEvent(e);
+ }
+ }
+ catch(ProcessedSilentlyException e) {
+ /* try reading again */
+ }
+ }
+ packet_queued = false;
+ }
+
+ Address sender;
+ s32 received_size = m_connection->m_udpSocket.Receive(sender, *packetdata, packet_maxsize);
+
+ if ((received_size < 0) ||
+ (received_size < BASE_HEADER_SIZE) ||
+ (readU32(&packetdata[0]) != m_connection->GetProtocolID()))
+ {
+ LOG(derr_con<<m_connection->getDesc()
+ <<"Receive(): Invalid incoming packet, "
+ <<"size: " << received_size
+ <<", protocol: " << readU32(&packetdata[0]) <<std::endl);
+ }
+
+ u16 peer_id = readPeerId(*packetdata);
+ u8 channelnum = readChannel(*packetdata);
+
+ if(channelnum > CHANNEL_COUNT-1){
+ LOG(derr_con<<m_connection->getDesc()
+ <<"Receive(): Invalid channel "<<channelnum<<std::endl);
+ throw InvalidIncomingDataException("Channel doesn't exist");
+ }
+
+ /* preserve original peer_id for later usage */
+ u16 packet_peer_id = peer_id;
+
+ /* Try to identify peer by sender address (may happen on join) */
+ if(peer_id == PEER_ID_INEXISTENT)
+ {
+ peer_id = m_connection->lookupPeer(sender);
+ }
+
+ /* The peer was not found in our lists. Add it. */
+ if(peer_id == PEER_ID_INEXISTENT)
+ {
+ peer_id = m_connection->createPeer(sender,MINETEST_RELIABLE_UDP,0);
+ }
+
+ PeerHelper peer = m_connection->getPeerNoEx(peer_id);
+
+ if (!peer) {
+ LOG(dout_con<<m_connection->getDesc()
+ <<" got packet from unknown peer_id: "
+ <<peer_id<<" Ignoring."<<std::endl);
+ continue;
+ }
+
+ // Validate peer address
+
+ Address peer_address;
+
+ if (peer->getAddress(UDP,peer_address)) {
+ if (peer_address != sender) {
+ LOG(derr_con<<m_connection->getDesc()
+ <<m_connection->getDesc()
+ <<" Peer "<<peer_id<<" sending from different address."
+ " Ignoring."<<std::endl);
+ continue;
+ }
+ }
+ else {
+
+ bool invalid_address = true;
+ if (invalid_address) {
+ LOG(derr_con<<m_connection->getDesc()
+ <<m_connection->getDesc()
+ <<" Peer "<<peer_id<<" unknown."
+ " Ignoring."<<std::endl);
+ continue;
+ }
+ }
+
+
+ /* mark peer as seen with id */
+ if (!(packet_peer_id == PEER_ID_INEXISTENT))
+ peer->setSentWithID();
+
+ peer->ResetTimeout();
+
+ Channel *channel = 0;
+
+ if (dynamic_cast<UDPPeer*>(&peer) != 0)
+ {
+ channel = &(dynamic_cast<UDPPeer*>(&peer)->channels[channelnum]);
+ }
+
+ // Throw the received packet to channel->processPacket()
+
+ // Make a new SharedBuffer from the data without the base headers
+ SharedBuffer<u8> strippeddata(received_size - BASE_HEADER_SIZE);
+ memcpy(*strippeddata, &packetdata[BASE_HEADER_SIZE],
+ strippeddata.getSize());
+
+ try{
+ // Process it (the result is some data with no headers made by us)
+ SharedBuffer<u8> resultdata = processPacket
+ (channel, strippeddata, peer_id, channelnum, false);
+
+ LOG(dout_con<<m_connection->getDesc()
+ <<" ProcessPacket from peer_id: " << peer_id
+ << ",channel: " << channelnum << ", returned "
+ << resultdata.getSize() << " bytes" <<std::endl);
+
+ ConnectionEvent e;
+ e.dataReceived(peer_id, resultdata);
+ m_connection->putEvent(e);
+ }catch(ProcessedSilentlyException &e){
+ }catch(ProcessedQueued &e){
+ packet_queued = true;
+ }
+ }catch(InvalidIncomingDataException &e){
+ }
+ catch(ProcessedSilentlyException &e){
}
-
- // Error checking
- assert(node->second->id == peer_id);
-
- return node->second;
-}
-
-std::list<Peer*> Connection::getPeers()
-{
- std::list<Peer*> list;
- for(std::map<u16, Peer*>::iterator j = m_peers.begin();
- j != m_peers.end(); ++j)
- {
- Peer *peer = j->second;
- list.push_back(peer);
}
- return list;
}
-bool Connection::getFromBuffers(u16 &peer_id, SharedBuffer<u8> &dst)
+bool ConnectionReceiveThread::getFromBuffers(u16 &peer_id, SharedBuffer<u8> &dst)
{
- for(std::map<u16, Peer*>::iterator j = m_peers.begin();
- j != m_peers.end(); ++j)
+ std::list<u16> peerids = m_connection->getPeerIDs();
+
+ for(std::list<u16>::iterator j = peerids.begin();
+ j != peerids.end(); ++j)
{
- Peer *peer = j->second;
+ PeerHelper peer = m_connection->getPeerNoEx(*j);
+ if (!peer)
+ continue;
+
+ if(dynamic_cast<UDPPeer*>(&peer) == 0)
+ continue;
+
for(u16 i=0; i<CHANNEL_COUNT; i++)
{
- Channel *channel = &peer->channels[i];
+ Channel *channel = &(dynamic_cast<UDPPeer*>(&peer))->channels[i];
+
SharedBuffer<u8> resultdata;
bool got = checkIncomingBuffers(channel, peer_id, resultdata);
if(got){
return false;
}
-bool Connection::checkIncomingBuffers(Channel *channel, u16 &peer_id,
+bool ConnectionReceiveThread::checkIncomingBuffers(Channel *channel, u16 &peer_id,
SharedBuffer<u8> &dst)
{
u16 firstseqnum = 0;
- // Clear old packets from start of buffer
- for(;;){
- bool found = channel->incoming_reliables.getFirstSeqnum(&firstseqnum);
- if(!found)
- break;
- if(seqnum_higher(channel->next_incoming_seqnum, firstseqnum))
- channel->incoming_reliables.popFirst();
- else
- break;
- }
- // This happens if all packets are old
-
- if(channel->incoming_reliables.empty() == false)
+ if (channel->incoming_reliables.getFirstSeqnum(firstseqnum))
{
- if(firstseqnum == channel->next_incoming_seqnum)
+ if(firstseqnum == channel->readNextIncomingSeqNum())
{
BufferedPacket p = channel->incoming_reliables.popFirst();
-
peer_id = readPeerId(*p.data);
u8 channelnum = readChannel(*p.data);
u16 seqnum = readU16(&p.data[BASE_HEADER_SIZE+1]);
- PrintInfo();
- dout_con<<"UNBUFFERING TYPE_RELIABLE"
+ LOG(dout_con<<m_connection->getDesc()
+ <<"UNBUFFERING TYPE_RELIABLE"
<<" seqnum="<<seqnum
<<" peer_id="<<peer_id
<<" channel="<<((int)channelnum&0xff)
- <<std::endl;
+ <<std::endl);
+
+ channel->incNextIncomingSeqNum();
- channel->next_incoming_seqnum++;
-
u32 headers_size = BASE_HEADER_SIZE + RELIABLE_HEADER_SIZE;
// Get out the inside packet and re-process it
SharedBuffer<u8> payload(p.data.getSize() - headers_size);
return false;
}
-SharedBuffer<u8> Connection::processPacket(Channel *channel,
+SharedBuffer<u8> ConnectionReceiveThread::processPacket(Channel *channel,
SharedBuffer<u8> packetdata, u16 peer_id,
u8 channelnum, bool reliable)
{
- IndentationRaiser iraiser(&(m_indentation));
+ PeerHelper peer = m_connection->getPeer(peer_id);
if(packetdata.getSize() < 1)
throw InvalidIncomingDataException("packetdata.getSize() < 1");
u8 type = readU8(&packetdata[0]);
-
+
if(type == TYPE_CONTROL)
{
if(packetdata.getSize() < 2)
u8 controltype = readU8(&packetdata[1]);
- if(controltype == CONTROLTYPE_ACK)
+ if( (controltype == CONTROLTYPE_ACK)
+ && (peer_id <= MAX_UDP_PEERS))
{
+ assert(channel != 0);
if(packetdata.getSize() < 4)
throw InvalidIncomingDataException
("packetdata.getSize() < 4 (ACK header size)");
u16 seqnum = readU16(&packetdata[2]);
- PrintInfo();
- dout_con<<"Got CONTROLTYPE_ACK: channelnum="
+ LOG(dout_con<<m_connection->getDesc()
+ <<" [ CONTROLTYPE_ACK: channelnum="
<<((int)channelnum&0xff)<<", peer_id="<<peer_id
- <<", seqnum="<<seqnum<<std::endl;
+ <<", seqnum="<<seqnum<< " ]"<<std::endl);
try{
- BufferedPacket p = channel->outgoing_reliables.popSeqnum(seqnum);
+ BufferedPacket p =
+ channel->outgoing_reliables_sent.popSeqnum(seqnum);
// Get round trip time
- float rtt = p.totaltime;
+ unsigned int current_time = porting::getTimeMs();
- // Let peer calculate stuff according to it
- // (avg_rtt and resend_timeout)
- Peer *peer = getPeer(peer_id);
- peer->reportRTT(rtt);
+ if (current_time > p.absolute_send_time)
+ {
+ float rtt = (current_time - p.absolute_send_time) / 1000.0;
- //PrintInfo(dout_con);
- //dout_con<<"RTT = "<<rtt<<std::endl;
+ // Let peer calculate stuff according to it
+ // (avg_rtt and resend_timeout)
+ dynamic_cast<UDPPeer*>(&peer)->reportRTT(rtt);
+ }
+ else if (p.totaltime > 0)
+ {
+ float rtt = p.totaltime;
- /*dout_con<<"OUTGOING: ";
- PrintInfo();
- channel->outgoing_reliables.print();
- dout_con<<std::endl;*/
+ // Let peer calculate stuff according to it
+ // (avg_rtt and resend_timeout)
+ dynamic_cast<UDPPeer*>(&peer)->reportRTT(rtt);
+ }
+ //put bytes for max bandwidth calculation
+ channel->UpdateBytesSent(p.data.getSize(),1);
}
catch(NotFoundException &e){
- PrintInfo(derr_con);
- derr_con<<"WARNING: ACKed packet not "
+ LOG(derr_con<<m_connection->getDesc()
+ <<"WARNING: ACKed packet not "
"in outgoing queue"
- <<std::endl;
+ <<std::endl);
+ channel->UpdatePacketTooLateCounter();
}
-
throw ProcessedSilentlyException("Got an ACK");
}
- else if(controltype == CONTROLTYPE_SET_PEER_ID)
+ else if((controltype == CONTROLTYPE_SET_PEER_ID)
+ && (peer_id <= MAX_UDP_PEERS))
{
+ // Got a packet to set our peer id
if(packetdata.getSize() < 4)
throw InvalidIncomingDataException
("packetdata.getSize() < 4 (SET_PEER_ID header size)");
u16 peer_id_new = readU16(&packetdata[2]);
- PrintInfo();
- dout_con<<"Got new peer id: "<<peer_id_new<<"... "<<std::endl;
+ LOG(dout_con<<m_connection->getDesc()
+ <<"Got new peer id: "<<peer_id_new<<"... "<<std::endl);
- if(GetPeerID() != PEER_ID_INEXISTENT)
+ if(m_connection->GetPeerID() != PEER_ID_INEXISTENT)
{
- PrintInfo(derr_con);
- derr_con<<"WARNING: Not changing"
- " existing peer id."<<std::endl;
+ LOG(derr_con<<m_connection->getDesc()
+ <<"WARNING: Not changing"
+ " existing peer id."<<std::endl);
}
else
{
- dout_con<<"changing."<<std::endl;
- SetPeerID(peer_id_new);
+ LOG(dout_con<<m_connection->getDesc()<<"changing own peer id"<<std::endl);
+ m_connection->SetPeerID(peer_id_new);
}
+
+ ConnectionCommand cmd;
+
+ SharedBuffer<u8> reply(2);
+ writeU8(&reply[0], TYPE_CONTROL);
+ writeU8(&reply[1], CONTROLTYPE_ENABLE_BIG_SEND_WINDOW);
+ cmd.disableLegacy(PEER_ID_SERVER,reply);
+ m_connection->putCommand(cmd);
+
throw ProcessedSilentlyException("Got a SET_PEER_ID");
}
- else if(controltype == CONTROLTYPE_PING)
+ else if((controltype == CONTROLTYPE_PING)
+ && (peer_id <= MAX_UDP_PEERS))
{
// Just ignore it, the incoming data already reset
// the timeout counter
- PrintInfo();
- dout_con<<"PING"<<std::endl;
+ LOG(dout_con<<m_connection->getDesc()<<"PING"<<std::endl);
throw ProcessedSilentlyException("Got a PING");
}
else if(controltype == CONTROLTYPE_DISCO)
{
// Just ignore it, the incoming data already reset
// the timeout counter
- PrintInfo();
- dout_con<<"DISCO: Removing peer "<<(peer_id)<<std::endl;
-
- if(deletePeer(peer_id, false) == false)
+ LOG(dout_con<<m_connection->getDesc()
+ <<"DISCO: Removing peer "<<(peer_id)<<std::endl);
+
+ if(m_connection->deletePeer(peer_id, false) == false)
{
- PrintInfo(derr_con);
- derr_con<<"DISCO: Peer not found"<<std::endl;
+ derr_con<<m_connection->getDesc()
+ <<"DISCO: Peer not found"<<std::endl;
}
throw ProcessedSilentlyException("Got a DISCO");
}
+ else if((controltype == CONTROLTYPE_ENABLE_BIG_SEND_WINDOW)
+ && (peer_id <= MAX_UDP_PEERS))
+ {
+ dynamic_cast<UDPPeer*>(&peer)->setNonLegacyPeer();
+ throw ProcessedSilentlyException("Got non legacy control");
+ }
else{
- PrintInfo(derr_con);
- derr_con<<"INVALID TYPE_CONTROL: invalid controltype="
- <<((int)controltype&0xff)<<std::endl;
+ LOG(derr_con<<m_connection->getDesc()
+ <<"INVALID TYPE_CONTROL: invalid controltype="
+ <<((int)controltype&0xff)<<std::endl);
throw InvalidIncomingDataException("Invalid control type");
}
}
if(packetdata.getSize() < ORIGINAL_HEADER_SIZE)
throw InvalidIncomingDataException
("packetdata.getSize() < ORIGINAL_HEADER_SIZE");
- PrintInfo();
- dout_con<<"RETURNING TYPE_ORIGINAL to user"
- <<std::endl;
+ LOG(dout_con<<m_connection->getDesc()
+ <<"RETURNING TYPE_ORIGINAL to user"
+ <<std::endl);
// Get the inside packet out and return it
SharedBuffer<u8> payload(packetdata.getSize() - ORIGINAL_HEADER_SIZE);
memcpy(*payload, &packetdata[ORIGINAL_HEADER_SIZE], payload.getSize());
}
else if(type == TYPE_SPLIT)
{
- // We have to create a packet again for buffering
- // This isn't actually too bad an idea.
- BufferedPacket packet = makePacket(
- getPeer(peer_id)->address,
- packetdata,
- GetProtocolID(),
- peer_id,
- channelnum);
- // Buffer the packet
- SharedBuffer<u8> data = channel->incoming_splits.insert(packet, reliable);
- if(data.getSize() != 0)
- {
- PrintInfo();
- dout_con<<"RETURNING TYPE_SPLIT: Constructed full data, "
- <<"size="<<data.getSize()<<std::endl;
- return data;
+ Address peer_address;
+
+ if (peer->getAddress(UDP,peer_address)) {
+
+ // We have to create a packet again for buffering
+ // This isn't actually too bad an idea.
+ BufferedPacket packet = makePacket(
+ peer_address,
+ packetdata,
+ m_connection->GetProtocolID(),
+ peer_id,
+ channelnum);
+
+ // Buffer the packet
+ SharedBuffer<u8> data =
+ peer->addSpiltPacket(channelnum,packet,reliable);
+
+ if(data.getSize() != 0)
+ {
+ LOG(dout_con<<m_connection->getDesc()
+ <<"RETURNING TYPE_SPLIT: Constructed full data, "
+ <<"size="<<data.getSize()<<std::endl);
+ return data;
+ }
+ LOG(dout_con<<m_connection->getDesc()<<"BUFFERED TYPE_SPLIT"<<std::endl);
+ throw ProcessedSilentlyException("Buffered a split packet chunk");
+ }
+ else {
+ //TODO throw some error
}
- PrintInfo();
- dout_con<<"BUFFERED TYPE_SPLIT"<<std::endl;
- throw ProcessedSilentlyException("Buffered a split packet chunk");
}
- else if(type == TYPE_RELIABLE)
+ else if((peer_id <= MAX_UDP_PEERS) && (type == TYPE_RELIABLE))
{
+ assert(channel != 0);
// Recursive reliable packets not allowed
if(reliable)
throw InvalidIncomingDataException("Found nested reliable packets");
("packetdata.getSize() < RELIABLE_HEADER_SIZE");
u16 seqnum = readU16(&packetdata[1]);
+ bool is_future_packet = false;
+ bool is_old_packet = false;
- bool is_future_packet = seqnum_higher(seqnum, channel->next_incoming_seqnum);
- bool is_old_packet = seqnum_higher(channel->next_incoming_seqnum, seqnum);
-
- PrintInfo();
- if(is_future_packet)
- dout_con<<"BUFFERING";
- else if(is_old_packet)
- dout_con<<"OLD";
- else
- dout_con<<"RECUR";
- dout_con<<" TYPE_RELIABLE seqnum="<<seqnum
- <<" next="<<channel->next_incoming_seqnum;
- dout_con<<" [sending CONTROLTYPE_ACK"
- " to peer_id="<<peer_id<<"]";
- dout_con<<std::endl;
-
- //DEBUG
- //assert(channel->incoming_reliables.size() < 100);
-
- // Send a CONTROLTYPE_ACK
- SharedBuffer<u8> reply(4);
- writeU8(&reply[0], TYPE_CONTROL);
- writeU8(&reply[1], CONTROLTYPE_ACK);
- writeU16(&reply[2], seqnum);
- rawSendAsPacket(peer_id, channelnum, reply, false);
-
- //if(seqnum_higher(seqnum, channel->next_incoming_seqnum))
- if(is_future_packet)
+ /* packet is within our receive window send ack */
+ if (seqnum_in_window(seqnum, channel->readNextIncomingSeqNum(),MAX_RELIABLE_WINDOW_SIZE))
{
- /*PrintInfo();
- dout_con<<"Buffering reliable packet (seqnum="
- <<seqnum<<")"<<std::endl;*/
-
+ m_connection->sendAck(peer_id,channelnum,seqnum);
+ }
+ else {
+ is_future_packet = seqnum_higher(seqnum, channel->readNextIncomingSeqNum());
+ is_old_packet = seqnum_higher(channel->readNextIncomingSeqNum(), seqnum);
+
+
+ /* packet is not within receive window, don't send ack. *
+ * if this was a valid packet it's gonna be retransmitted */
+ if (is_future_packet)
+ {
+ throw ProcessedSilentlyException("Received packet newer then expected, not sending ack");
+ }
+
+ /* seems like our ack was lost, send another one for a old packet */
+ if (is_old_packet)
+ {
+ LOG(dout_con<<m_connection->getDesc()
+ << "RE-SENDING ACK: peer_id: " << peer_id
+ << ", channel: " << (channelnum&0xFF)
+ << ", seqnum: " << seqnum << std::endl;)
+ m_connection->sendAck(peer_id,channelnum,seqnum);
+
+ // we already have this packet so this one was on wire at least
+ // the current timeout
+ dynamic_cast<UDPPeer*>(&peer)->reportRTT(dynamic_cast<UDPPeer*>(&peer)->getResendTimeout());
+
+ throw ProcessedSilentlyException("Retransmitting ack for old packet");
+ }
+ }
+
+ if (seqnum != channel->readNextIncomingSeqNum())
+ {
+ Address peer_address;
+
+ // this is a reliable packet so we have a udp address for sure
+ peer->getAddress(MINETEST_RELIABLE_UDP,peer_address);
// This one comes later, buffer it.
// Actually we have to make a packet to buffer one.
// Well, we have all the ingredients, so just do it.
- BufferedPacket packet = makePacket(
- getPeer(peer_id)->address,
+ BufferedPacket packet = con::makePacket(
+ peer_address,
packetdata,
- GetProtocolID(),
+ m_connection->GetProtocolID(),
peer_id,
channelnum);
try{
- channel->incoming_reliables.insert(packet);
-
- /*PrintInfo();
- dout_con<<"INCOMING: ";
- channel->incoming_reliables.print();
- dout_con<<std::endl;*/
+ channel->incoming_reliables.insert(packet,channel->readNextIncomingSeqNum());
+
+ LOG(dout_con<<m_connection->getDesc()
+ << "BUFFERING, TYPE_RELIABLE peer_id: " << peer_id
+ << ", channel: " << (channelnum&0xFF)
+ << ", seqnum: " << seqnum << std::endl;)
+
+ throw ProcessedQueued("Buffered future reliable packet");
}
catch(AlreadyExistsException &e)
{
}
-
- throw ProcessedSilentlyException("Buffered future reliable packet");
+ catch(IncomingDataCorruption &e)
+ {
+ ConnectionCommand discon;
+ discon.disconnect_peer(peer_id);
+ m_connection->putCommand(discon);
+
+ LOG(derr_con<<m_connection->getDesc()
+ << "INVALID, TYPE_RELIABLE peer_id: " << peer_id
+ << ", channel: " << (channelnum&0xFF)
+ << ", seqnum: " << seqnum
+ << "DROPPING CLIENT!" << std::endl;)
+ }
}
- //else if(seqnum_higher(channel->next_incoming_seqnum, seqnum))
- else if(is_old_packet)
+
+ /* we got a packet to process right now */
+ LOG(dout_con<<m_connection->getDesc()
+ << "RECURSIVE, TYPE_RELIABLE peer_id: " << peer_id
+ << ", channel: " << (channelnum&0xFF)
+ << ", seqnum: " << seqnum << std::endl;)
+
+
+ /* check for resend case */
+ u16 queued_seqnum = 0;
+ if (channel->incoming_reliables.getFirstSeqnum(queued_seqnum))
{
- // An old packet, dump it
- throw InvalidIncomingDataException("Got an old reliable packet");
+ if (queued_seqnum == seqnum)
+ {
+ BufferedPacket queued_packet = channel->incoming_reliables.popFirst();
+ /** TODO find a way to verify the new against the old packet */
+ }
}
- channel->next_incoming_seqnum++;
+ channel->incNextIncomingSeqNum();
// Get out the inside packet and re-process it
SharedBuffer<u8> payload(packetdata.getSize() - RELIABLE_HEADER_SIZE);
}
else
{
- PrintInfo(derr_con);
- derr_con<<"Got invalid type="<<((int)type&0xff)<<std::endl;
+ derr_con<<m_connection->getDesc()
+ <<"Got invalid type="<<((int)type&0xff)<<std::endl;
throw InvalidIncomingDataException("Invalid packet type");
}
-
+
// We should never get here.
// If you get here, add an exception or a return to some of the
// above conditionals.
throw BaseException("Error in Channel::ProcessPacket()");
}
+/*
+ Connection
+*/
+
+Connection::Connection(u32 protocol_id, u32 max_packet_size, float timeout,
+ bool ipv6):
+ m_udpSocket(ipv6),
+ m_command_queue(),
+ m_event_queue(),
+ m_peer_id(0),
+ m_protocol_id(protocol_id),
+ m_sendThread(this, max_packet_size, timeout),
+ m_receiveThread(this, max_packet_size),
+ m_info_mutex(),
+ m_bc_peerhandler(0),
+ m_bc_receive_timeout(0),
+ m_shutting_down(false)
+{
+ m_udpSocket.setTimeoutMs(5);
+
+ m_sendThread.Start();
+ m_receiveThread.Start();
+}
+
+Connection::Connection(u32 protocol_id, u32 max_packet_size, float timeout,
+ bool ipv6, PeerHandler *peerhandler):
+ m_udpSocket(ipv6),
+ m_command_queue(),
+ m_event_queue(),
+ m_peer_id(0),
+ m_protocol_id(protocol_id),
+ m_sendThread(this, max_packet_size, timeout),
+ m_receiveThread(this, max_packet_size),
+ m_info_mutex(),
+ m_bc_peerhandler(peerhandler),
+ m_bc_receive_timeout(0),
+ m_shutting_down(false)
+
+{
+ m_udpSocket.setTimeoutMs(5);
+
+ m_sendThread.Start();
+ m_receiveThread.Start();
+
+}
+
+
+Connection::~Connection()
+{
+ m_shutting_down = true;
+ // request threads to stop
+ m_sendThread.Stop();
+ m_receiveThread.Stop();
+
+ //TODO for some unkonwn reason send/receive threads do not exit as they're
+ // supposed to be but wait on peer timeout. To speed up shutdown we reduce
+ // timeout to half a second.
+ m_sendThread.setPeerTimeout(0.5);
+
+ // wait for threads to finish
+ m_sendThread.Wait();
+ m_receiveThread.Wait();
+
+ // Delete peers
+ for(std::map<u16, Peer*>::iterator
+ j = m_peers.begin();
+ j != m_peers.end(); ++j)
+ {
+ delete j->second;
+ }
+}
+
+/* Internal stuff */
+void Connection::putEvent(ConnectionEvent &e)
+{
+ assert(e.type != CONNEVENT_NONE);
+ m_event_queue.push_back(e);
+}
+
+PeerHelper Connection::getPeer(u16 peer_id)
+{
+ JMutexAutoLock peerlock(m_peers_mutex);
+ std::map<u16, Peer*>::iterator node = m_peers.find(peer_id);
+
+ if(node == m_peers.end()){
+ throw PeerNotFoundException("GetPeer: Peer not found (possible timeout)");
+ }
+
+ // Error checking
+ assert(node->second->id == peer_id);
+
+ return PeerHelper(node->second);
+}
+
+PeerHelper Connection::getPeerNoEx(u16 peer_id)
+{
+ JMutexAutoLock peerlock(m_peers_mutex);
+ std::map<u16, Peer*>::iterator node = m_peers.find(peer_id);
+
+ if(node == m_peers.end()){
+ return PeerHelper(NULL);
+ }
+
+ // Error checking
+ assert(node->second->id == peer_id);
+
+ return PeerHelper(node->second);
+}
+
+/* find peer_id for address */
+u16 Connection::lookupPeer(Address& sender)
+{
+ JMutexAutoLock peerlock(m_peers_mutex);
+ std::map<u16, Peer*>::iterator j;
+ j = m_peers.begin();
+ for(; j != m_peers.end(); ++j)
+ {
+ Peer *peer = j->second;
+ if(peer->isActive())
+ continue;
+
+ Address tocheck;
+
+ if ((peer->getAddress(MINETEST_RELIABLE_UDP,tocheck)) && (tocheck == sender))
+ return peer->id;
+
+ if ((peer->getAddress(UDP,tocheck)) && (tocheck == sender))
+ return peer->id;
+ }
+
+ return PEER_ID_INEXISTENT;
+}
+
+std::list<Peer*> Connection::getPeers()
+{
+ std::list<Peer*> list;
+ for(std::map<u16, Peer*>::iterator j = m_peers.begin();
+ j != m_peers.end(); ++j)
+ {
+ Peer *peer = j->second;
+ list.push_back(peer);
+ }
+ return list;
+}
+
bool Connection::deletePeer(u16 peer_id, bool timeout)
{
- if(m_peers.find(peer_id) == m_peers.end())
- return false;
-
- Peer *peer = m_peers[peer_id];
+ Peer *peer = 0;
+ /* lock list as short as possible */
+ {
+ JMutexAutoLock peerlock(m_peers_mutex);
+ if(m_peers.find(peer_id) == m_peers.end())
+ return false;
+ peer = m_peers[peer_id];
+ m_peers.erase(peer_id);
+ }
+
+ Address peer_address;
+ //any peer has a primary address this never fails!
+ peer->getAddress(PRIMARY,peer_address);
// Create event
ConnectionEvent e;
- e.peerRemoved(peer_id, timeout, peer->address);
+ e.peerRemoved(peer_id, timeout, peer_address);
putEvent(e);
- delete m_peers[peer_id];
- m_peers.erase(peer_id);
+
+ peer->Drop();
return true;
}
void Connection::putCommand(ConnectionCommand &c)
{
- m_command_queue.push_back(c);
+ if (!m_shutting_down)
+ {
+ m_command_queue.push_back(c);
+ m_sendThread.Trigger();
+ }
}
void Connection::Serve(unsigned short port)
if(m_peer_id == PEER_ID_INEXISTENT)
return false;
-
+
return true;
}
for(;;){
ConnectionEvent e = waitEvent(m_bc_receive_timeout);
if(e.type != CONNEVENT_NONE)
- dout_con<<getDesc()<<": Receive: got event: "
- <<e.describe()<<std::endl;
+ LOG(dout_con<<getDesc()<<": Receive: got event: "
+ <<e.describe()<<std::endl);
switch(e.type){
case CONNEVENT_NONE:
throw NoIncomingDataException("No incoming data");
data = SharedBuffer<u8>(e.data);
return e.data.getSize();
case CONNEVENT_PEER_ADDED: {
- Peer tmp(e.peer_id, e.address);
+ UDPPeer tmp(e.peer_id, e.address, this);
if(m_bc_peerhandler)
m_bc_peerhandler->peerAdded(&tmp);
continue; }
case CONNEVENT_PEER_REMOVED: {
- Peer tmp(e.peer_id, e.address);
+ UDPPeer tmp(e.peer_id, e.address, this);
if(m_bc_peerhandler)
m_bc_peerhandler->deletingPeer(&tmp, e.timeout);
continue; }
Address Connection::GetPeerAddress(u16 peer_id)
{
- JMutexAutoLock peerlock(m_peers_mutex);
- return getPeer(peer_id)->address;
+ PeerHelper peer = getPeerNoEx(peer_id);
+
+ if (!peer)
+ throw PeerNotFoundException("No address for peer found!");
+ Address peer_address;
+ peer->getAddress(PRIMARY,peer_address);
+ return peer_address;
}
float Connection::GetPeerAvgRTT(u16 peer_id)
{
- JMutexAutoLock peerlock(m_peers_mutex);
- return getPeer(peer_id)->avg_rtt;
+ PeerHelper peer = getPeerNoEx(peer_id);
+ if (!peer) return -1;
+ return peer->getStat(AVG_RTT);
+}
+
+u16 Connection::createPeer(Address& sender, MTProtocols protocol, int fd)
+{
+ // Somebody wants to make a new connection
+
+ // Get a unique peer id (2 or higher)
+ u16 peer_id_new = 2;
+ u16 overflow = MAX_UDP_PEERS;
+
+ /*
+ Find an unused peer id
+ */
+ bool out_of_ids = false;
+ for(;;)
+ {
+ // Check if exists
+ if(m_peers.find(peer_id_new) == m_peers.end())
+ break;
+ // Check for overflow
+ if(peer_id_new == overflow){
+ out_of_ids = true;
+ break;
+ }
+ peer_id_new++;
+ }
+ if(out_of_ids){
+ errorstream<<getDesc()<<" ran out of peer ids"<<std::endl;
+ return PEER_ID_INEXISTENT;
+ }
+
+ LOG(dout_con<<getDesc()
+ <<"createPeer(): giving peer_id="<<peer_id_new<<std::endl);
+
+ // Create a peer
+ Peer *peer = 0;
+
+ peer = new UDPPeer(peer_id_new, sender, this);
+
+ m_peers_mutex.Lock();
+ m_peers[peer->id] = peer;
+ m_peers_mutex.Unlock();
+
+ // Create peer addition event
+ ConnectionEvent e;
+ e.peerAdded(peer_id_new, sender);
+ putEvent(e);
+
+ ConnectionCommand cmd;
+ SharedBuffer<u8> reply(4);
+ writeU8(&reply[0], TYPE_CONTROL);
+ writeU8(&reply[1], CONTROLTYPE_SET_PEER_ID);
+ writeU16(&reply[2], peer_id_new);
+ cmd.createPeer(peer_id_new,reply);
+ this->putCommand(cmd);
+
+ // We're now talking to a valid peer_id
+ return peer_id_new;
}
void Connection::DeletePeer(u16 peer_id)
void Connection::PrintInfo(std::ostream &out)
{
+ m_info_mutex.Lock();
out<<getDesc()<<": ";
+ m_info_mutex.Unlock();
}
void Connection::PrintInfo()
PrintInfo(dout_con);
}
-std::string Connection::getDesc()
+const std::string Connection::getDesc()
+{
+ return std::string("con(")+itos(m_udpSocket.GetHandle())+"/"+itos(m_peer_id)+")";
+}
+
+void Connection::sendAck(u16 peer_id, u8 channelnum, u16 seqnum) {
+
+ assert(channelnum < CHANNEL_COUNT);
+
+ LOG(dout_con<<getDesc()
+ <<" Queuing ACK command to peer_id: " << peer_id <<
+ " channel: " << (channelnum & 0xFF) <<
+ " seqnum: " << seqnum << std::endl);
+
+ ConnectionCommand c;
+ SharedBuffer<u8> ack(4);
+ writeU8(&ack[0], TYPE_CONTROL);
+ writeU8(&ack[1], CONTROLTYPE_ACK);
+ writeU16(&ack[2], seqnum);
+
+ c.ack(peer_id, channelnum, ack);
+ putCommand(c);
+ m_sendThread.Trigger();
+}
+
+UDPPeer* Connection::createServerPeer(Address& address)
{
- return std::string("con(")+itos(m_socket.GetHandle())+"/"+itos(m_peer_id)+")";
+ if (getPeerNoEx(PEER_ID_SERVER) != 0)
+ {
+ throw ConnectionException("Already connected to a server");
+ }
+
+ UDPPeer *peer = new UDPPeer(PEER_ID_SERVER, address, this);
+
+ {
+ JMutexAutoLock lock(m_peers_mutex);
+ m_peers[peer->id] = peer;
+ }
+
+ return peer;
+}
+
+std::list<u16> Connection::getPeerIDs()
+{
+ std::list<u16> retval;
+
+ JMutexAutoLock lock(m_peers_mutex);
+ for(std::map<u16, Peer*>::iterator j = m_peers.begin();
+ j != m_peers.end(); ++j)
+ {
+ retval.push_back(j->first);
+ }
+ return retval;
}
} // namespace
#include "util/pointer.h"
#include "util/container.h"
#include "util/thread.h"
+#include "util/numeric.h"
#include <iostream>
#include <fstream>
#include <list>
{}
};
+class ProcessedQueued : public BaseException
+{
+public:
+ ProcessedQueued(const char *s):
+ BaseException(s)
+ {}
+};
+
+class IncomingDataCorruption : public BaseException
+{
+public:
+ IncomingDataCorruption(const char *s):
+ BaseException(s)
+ {}
+};
+
+typedef enum MTProtocols {
+ PRIMARY,
+ UDP,
+ MINETEST_RELIABLE_UDP
+} MTProtocols;
+
#define SEQNUM_MAX 65535
-inline bool seqnum_higher(u16 higher, u16 lower)
+inline bool seqnum_higher(u16 totest, u16 base)
+{
+ if (totest > base)
+ {
+ if((totest - base) > (SEQNUM_MAX/2))
+ return false;
+ else
+ return true;
+ }
+ else
+ {
+ if((base - totest) > (SEQNUM_MAX/2))
+ return true;
+ else
+ return false;
+ }
+}
+
+inline bool seqnum_in_window(u16 seqnum, u16 next,u16 window_size)
{
- if(lower > higher && lower - higher > SEQNUM_MAX/2){
- return true;
+ u16 window_start = next;
+ u16 window_end = ( next + window_size ) % (SEQNUM_MAX+1);
+
+ if (window_start < window_end)
+ {
+ return ((seqnum >= window_start) && (seqnum < window_end));
+ }
+ else
+ {
+ return ((seqnum < window_end) || (seqnum >= window_start));
}
- return (higher > lower);
}
struct BufferedPacket
{
BufferedPacket(u8 *a_data, u32 a_size):
- data(a_data, a_size), time(0.0), totaltime(0.0)
+ data(a_data, a_size), time(0.0), totaltime(0.0), absolute_send_time(-1)
{}
BufferedPacket(u32 a_size):
- data(a_size), time(0.0), totaltime(0.0)
+ data(a_size), time(0.0), totaltime(0.0), absolute_send_time(-1)
{}
SharedBuffer<u8> data; // Data of the packet, including headers
float time; // Seconds from buffering the packet or re-sending
float totaltime; // Seconds from buffering the packet
+ unsigned int absolute_send_time;
Address address; // Sender or destination
};
#define CONTROLTYPE_SET_PEER_ID 1
#define CONTROLTYPE_PING 2
#define CONTROLTYPE_DISCO 3
+#define CONTROLTYPE_ENABLE_BIG_SEND_WINDOW 4
+
/*
ORIGINAL: This is a plain packet with no control and no error
checking at all.
*/
#define TYPE_RELIABLE 3
#define RELIABLE_HEADER_SIZE 3
-//#define SEQNUM_INITIAL 0x10
#define SEQNUM_INITIAL 65500
/*
{
public:
ReliablePacketBuffer();
- void print();
- bool empty();
- u32 size();
- RPBSearchResult findPacket(u16 seqnum);
- RPBSearchResult notFound();
- bool getFirstSeqnum(u16 *result);
+
+ bool getFirstSeqnum(u16& result);
+
BufferedPacket popFirst();
BufferedPacket popSeqnum(u16 seqnum);
- void insert(BufferedPacket &p);
+ void insert(BufferedPacket &p,u16 next_expected);
+
void incrementTimeouts(float dtime);
- void resetTimedOuts(float timeout);
- bool anyTotaltimeReached(float timeout);
- std::list<BufferedPacket> getTimedOuts(float timeout);
+ std::list<BufferedPacket> getTimedOuts(float timeout,
+ unsigned int max_packets);
+
+ void print();
+ bool empty();
+ bool containsPacket(u16 seqnum);
+ RPBSearchResult notFound();
+ u32 size();
+
private:
+ RPBSearchResult findPacket(u16 seqnum);
+
std::list<BufferedPacket> m_list;
u16 m_list_size;
+
+ u16 m_oldest_non_answered_ack;
+
+ JMutex m_list_mutex;
+
+ unsigned int writeptr;
};
/*
private:
// Key is seqnum
std::map<u16, IncomingSplitPacket*> m_buf;
+
+ JMutex m_map_mutex;
};
-class Connection;
+struct OutgoingPacket
+{
+ u16 peer_id;
+ u8 channelnum;
+ SharedBuffer<u8> data;
+ bool reliable;
+ bool ack;
+
+ OutgoingPacket(u16 peer_id_, u8 channelnum_, SharedBuffer<u8> data_,
+ bool reliable_,bool ack_=false):
+ peer_id(peer_id_),
+ channelnum(channelnum_),
+ data(data_),
+ reliable(reliable_),
+ ack(ack_)
+ {
+ }
+};
-struct Channel
+enum ConnectionCommandType{
+ CONNCMD_NONE,
+ CONNCMD_SERVE,
+ CONNCMD_CONNECT,
+ CONNCMD_DISCONNECT,
+ CONNCMD_DISCONNECT_PEER,
+ CONNCMD_SEND,
+ CONNCMD_SEND_TO_ALL,
+ CONNCMD_DELETE_PEER,
+ CONCMD_ACK,
+ CONCMD_CREATE_PEER,
+ CONCMD_DISABLE_LEGACY
+};
+
+struct ConnectionCommand
{
- Channel();
- ~Channel();
+ enum ConnectionCommandType type;
+ u16 port;
+ Address address;
+ u16 peer_id;
+ u8 channelnum;
+ Buffer<u8> data;
+ bool reliable;
+ bool raw;
- u16 next_outgoing_seqnum;
- u16 next_incoming_seqnum;
- u16 next_outgoing_split_seqnum;
+ ConnectionCommand(): type(CONNCMD_NONE), peer_id(PEER_ID_INEXISTENT), reliable(false), raw(false) {}
+
+ void serve(u16 port_)
+ {
+ type = CONNCMD_SERVE;
+ port = port_;
+ }
+ void connect(Address address_)
+ {
+ type = CONNCMD_CONNECT;
+ address = address_;
+ }
+ void disconnect()
+ {
+ type = CONNCMD_DISCONNECT;
+ }
+ void disconnect_peer(u16 peer_id_)
+ {
+ type = CONNCMD_DISCONNECT_PEER;
+ peer_id = peer_id_;
+ }
+ void send(u16 peer_id_, u8 channelnum_,
+ SharedBuffer<u8> data_, bool reliable_)
+ {
+ type = CONNCMD_SEND;
+ peer_id = peer_id_;
+ channelnum = channelnum_;
+ data = data_;
+ reliable = reliable_;
+ }
+ void sendToAll(u8 channelnum_, SharedBuffer<u8> data_, bool reliable_)
+ {
+ type = CONNCMD_SEND_TO_ALL;
+ channelnum = channelnum_;
+ data = data_;
+ reliable = reliable_;
+ }
+ void deletePeer(u16 peer_id_)
+ {
+ type = CONNCMD_DELETE_PEER;
+ peer_id = peer_id_;
+ }
+
+ void ack(u16 peer_id_, u8 channelnum_, SharedBuffer<u8> data_)
+ {
+ type = CONCMD_ACK;
+ peer_id = peer_id_;
+ channelnum = channelnum_;
+ data = data_;
+ reliable = false;
+ }
+
+ void createPeer(u16 peer_id_, SharedBuffer<u8> data_)
+ {
+ type = CONCMD_CREATE_PEER;
+ peer_id = peer_id_;
+ data = data_;
+ channelnum = 0;
+ reliable = true;
+ raw = true;
+ }
+
+ void disableLegacy(u16 peer_id_, SharedBuffer<u8> data_)
+ {
+ type = CONCMD_DISABLE_LEGACY;
+ peer_id = peer_id_;
+ data = data_;
+ channelnum = 0;
+ reliable = true;
+ raw = true;
+ }
+};
+
+class Channel
+{
+
+public:
+ u16 readNextIncomingSeqNum();
+ u16 incNextIncomingSeqNum();
+
+ u16 getOutgoingSequenceNumber(bool& successfull);
+ u16 readOutgoingSequenceNumber();
+ bool putBackSequenceNumber(u16);
+
+ u16 readNextSplitSeqNum();
+ void setNextSplitSeqNum(u16 seqnum);
// This is for buffering the incoming packets that are coming in
// the wrong order
ReliablePacketBuffer incoming_reliables;
// This is for buffering the sent packets so that the sender can
// re-send them if no ACK is received
- ReliablePacketBuffer outgoing_reliables;
+ ReliablePacketBuffer outgoing_reliables_sent;
+
+ //queued reliable packets
+ Queue<BufferedPacket> queued_reliables;
+
+ //queue commands prior splitting to packets
+ Queue<ConnectionCommand> queued_commands;
IncomingSplitBuffer incoming_splits;
+
+ Channel();
+ ~Channel();
+
+ void UpdatePacketLossCounter(unsigned int count);
+ void UpdatePacketTooLateCounter();
+ void UpdateBytesSent(unsigned int bytes,unsigned int packages=1);
+ void UpdateBytesLost(unsigned int bytes);
+
+ void UpdateTimers(float dtime);
+
+ const float getCurrentDownloadRateKB()
+ { JMutexAutoLock lock(m_internal_mutex); return cur_kbps; };
+ const float getMaxDownloadRateKB()
+ { JMutexAutoLock lock(m_internal_mutex); return max_kbps; };
+
+ const float getCurrentLossRateKB()
+ { JMutexAutoLock lock(m_internal_mutex); return cur_kbps_lost; };
+ const float getMaxLossRateKB()
+ { JMutexAutoLock lock(m_internal_mutex); return max_kbps_lost; };
+
+ const float getAvgDownloadRateKB()
+ { JMutexAutoLock lock(m_internal_mutex); return avg_kbps; };
+ const float getAvgLossRateKB()
+ { JMutexAutoLock lock(m_internal_mutex); return avg_kbps_lost; };
+
+ const unsigned int getWindowSize() const { return window_size; };
+
+ void setWindowSize(unsigned int size) { window_size = size; };
+private:
+ JMutex m_internal_mutex;
+ unsigned int window_size;
+
+ u16 next_incoming_seqnum;
+
+ u16 next_outgoing_seqnum;
+ u16 next_outgoing_split_seqnum;
+
+ unsigned int current_packet_loss;
+ unsigned int current_packet_too_late;
+ unsigned int current_packet_successfull;
+ float packet_loss_counter;
+
+ unsigned int current_bytes_transfered;
+ unsigned int current_bytes_lost;
+ float max_kbps;
+ float cur_kbps;
+ float avg_kbps;
+ float max_kbps_lost;
+ float cur_kbps_lost;
+ float avg_kbps_lost;
+ float bpm_counter;
};
class Peer;
virtual void deletingPeer(Peer *peer, bool timeout) = 0;
};
-class Peer
+class PeerHelper
{
public:
+ PeerHelper();
+ PeerHelper(Peer* peer);
+ ~PeerHelper();
- Peer(u16 a_id, Address a_address);
- virtual ~Peer();
-
+ PeerHelper& operator=(Peer* peer);
+ Peer* operator->() const;
+ bool operator!();
+ Peer* operator&() const;
+ bool operator!=(void* ptr);
+
+private:
+ Peer* m_peer;
+};
+
+class Connection;
+
+typedef enum rtt_stat_type {
+ MIN_RTT,
+ MAX_RTT,
+ AVG_RTT,
+ MIN_JITTER,
+ MAX_JITTER,
+ AVG_JITTER
+} rtt_stat_type;
+
+class Peer {
+ public:
+ friend class PeerHelper;
+
+ Peer(Address address_,u16 id_,Connection* connection) :
+ id(id_),
+ m_increment_packets_remaining(9),
+ m_increment_bytes_remaining(0),
+ m_pending_deletion(false),
+ m_connection(connection),
+ address(address_),
+ m_ping_timer(0.0),
+ m_last_rtt(-1.0),
+ m_usage(0),
+ m_timeout_counter(0.0),
+ m_last_timeout_check(porting::getTimeMs()),
+ m_has_sent_with_id(false)
+ {
+ m_rtt.avg_rtt = -1.0;
+ m_rtt.jitter_avg = -1.0;
+ m_rtt.jitter_max = 0.0;
+ m_rtt.max_rtt = 0.0;
+ m_rtt.jitter_min = FLT_MAX;
+ m_rtt.min_rtt = FLT_MAX;
+ };
+
+ virtual ~Peer() {
+ JMutexAutoLock usage_lock(m_exclusive_access_mutex);
+ assert(m_usage == 0);
+ };
+
+ // Unique id of the peer
+ u16 id;
+
+ void Drop();
+
+ virtual void PutReliableSendCommand(ConnectionCommand &c,
+ unsigned int max_packet_size) {};
+
+ virtual bool isActive() { return false; };
+
+ virtual bool getAddress(MTProtocols type, Address& toset) = 0;
+
+ void ResetTimeout()
+ {JMutexAutoLock lock(m_exclusive_access_mutex); m_timeout_counter=0.0; };
+
+ bool isTimedOut(float timeout);
+
+ void setSentWithID()
+ { JMutexAutoLock lock(m_exclusive_access_mutex); m_has_sent_with_id = true; };
+
+ bool hasSentWithID()
+ { JMutexAutoLock lock(m_exclusive_access_mutex); return m_has_sent_with_id; };
+
+ unsigned int m_increment_packets_remaining;
+ unsigned int m_increment_bytes_remaining;
+
+ virtual u16 getNextSplitSequenceNumber(u8 channel) { return 0; };
+ virtual void setNextSplitSequenceNumber(u8 channel, u16 seqnum) {};
+ virtual SharedBuffer<u8> addSpiltPacket(u8 channel,
+ BufferedPacket toadd,
+ bool reliable)
+ {
+ fprintf(stderr,"Peer: addSplitPacket called, this is supposed to be never called!\n");
+ return SharedBuffer<u8>(0);
+ };
+
+ virtual bool Ping(float dtime, SharedBuffer<u8>& data) { return false; };
+
+ virtual float getStat(rtt_stat_type type) const {
+ switch (type) {
+ case MIN_RTT:
+ return m_rtt.min_rtt;
+ case MAX_RTT:
+ return m_rtt.max_rtt;
+ case AVG_RTT:
+ return m_rtt.avg_rtt;
+ case MIN_JITTER:
+ return m_rtt.jitter_min;
+ case MAX_JITTER:
+ return m_rtt.jitter_max;
+ case AVG_JITTER:
+ return m_rtt.jitter_avg;
+ }
+ return -1;
+ }
+ protected:
+ virtual void reportRTT(float rtt) {};
+
+ void RTTStatistics(float rtt,
+ std::string profiler_id="",
+ unsigned int num_samples=1000);
+
+ bool IncUseCount();
+ void DecUseCount();
+
+ JMutex m_exclusive_access_mutex;
+
+ bool m_pending_deletion;
+
+ Connection* m_connection;
+
+ // Address of the peer
+ Address address;
+
+ // Ping timer
+ float m_ping_timer;
+ private:
+
+ struct rttstats {
+ float jitter_min;
+ float jitter_max;
+ float jitter_avg;
+ float min_rtt;
+ float max_rtt;
+ float avg_rtt;
+ };
+
+ rttstats m_rtt;
+ float m_last_rtt;
+
+ // current usage count
+ unsigned int m_usage;
+
+ // Seconds from last receive
+ float m_timeout_counter;
+
+ u32 m_last_timeout_check;
+
+ bool m_has_sent_with_id;
+};
+
+class UDPPeer : public Peer
+{
+public:
+
+ friend class PeerHelper;
+ friend class ConnectionReceiveThread;
+ friend class ConnectionSendThread;
+
+ UDPPeer(u16 a_id, Address a_address, Connection* connection);
+ virtual ~UDPPeer();
+
+ void PutReliableSendCommand(ConnectionCommand &c,
+ unsigned int max_packet_size);
+
+ bool isActive()
+ { return ((hasSentWithID()) && (!m_pending_deletion)); };
+
+ bool getAddress(MTProtocols type, Address& toset);
+
+ void setNonLegacyPeer()
+ { m_legacy_peer = false; }
+
+ bool getLegacyPeer()
+ { return m_legacy_peer; }
+
+ u16 getNextSplitSequenceNumber(u8 channel);
+ void setNextSplitSequenceNumber(u8 channel, u16 seqnum);
+
+ SharedBuffer<u8> addSpiltPacket(u8 channel,
+ BufferedPacket toadd,
+ bool reliable);
+protected:
/*
Calculates avg_rtt and resend_timeout.
-
rtt=-1 only recalculates resend_timeout
*/
void reportRTT(float rtt);
- Channel channels[CHANNEL_COUNT];
+ void RunCommandQueues(
+ unsigned int max_packet_size,
+ unsigned int maxcommands,
+ unsigned int maxtransfer);
- // Address of the peer
- Address address;
- // Unique id of the peer
- u16 id;
- // Seconds from last receive
- float timeout_counter;
- // Ping timer
- float ping_timer;
+ float getResendTimeout()
+ { JMutexAutoLock lock(m_exclusive_access_mutex); return resend_timeout; }
+
+ void setResendTimeout(float timeout)
+ { JMutexAutoLock lock(m_exclusive_access_mutex); resend_timeout = timeout; }
+ bool Ping(float dtime,SharedBuffer<u8>& data);
+
+ Channel channels[CHANNEL_COUNT];
+private:
// This is changed dynamically
float resend_timeout;
- // Updated when an ACK is received
- float avg_rtt;
- // This is set to true when the peer has actually sent something
- // with the id we have given to it
- bool has_sent_with_id;
-
- float m_sendtime_accu;
- float m_max_packets_per_second;
- int m_num_sent;
- int m_max_num_sent;
-
- // Updated from configuration by Connection
- float congestion_control_aim_rtt;
- float congestion_control_max_rate;
- float congestion_control_min_rate;
-private:
+
+ bool processReliableSendCommand(
+ ConnectionCommand &c,
+ unsigned int max_packet_size);
+
+ bool m_legacy_peer;
};
/*
Connection
*/
-struct OutgoingPacket
-{
- u16 peer_id;
- u8 channelnum;
- SharedBuffer<u8> data;
- bool reliable;
-
- OutgoingPacket(u16 peer_id_, u8 channelnum_, SharedBuffer<u8> data_,
- bool reliable_):
- peer_id(peer_id_),
- channelnum(channelnum_),
- data(data_),
- reliable(reliable_)
- {
- }
-};
-
enum ConnectionEventType{
CONNEVENT_NONE,
CONNEVENT_DATA_RECEIVED,
}
};
-enum ConnectionCommandType{
- CONNCMD_NONE,
- CONNCMD_SERVE,
- CONNCMD_CONNECT,
- CONNCMD_DISCONNECT,
- CONNCMD_SEND,
- CONNCMD_SEND_TO_ALL,
- CONNCMD_DELETE_PEER,
+class ConnectionSendThread : public JThread {
+
+public:
+ friend class UDPPeer;
+
+ ConnectionSendThread(Connection* parent,
+ unsigned int max_packet_size, float timeout);
+
+ void * Thread ();
+
+ void Trigger();
+
+ void setPeerTimeout(float peer_timeout)
+ { m_timeout = peer_timeout; }
+
+private:
+ void runTimeouts (float dtime);
+ void rawSend (const BufferedPacket &packet);
+ bool rawSendAsPacket(u16 peer_id, u8 channelnum,
+ SharedBuffer<u8> data, bool reliable);
+
+ void processReliableCommand (ConnectionCommand &c);
+ void processNonReliableCommand (ConnectionCommand &c);
+ void serve (u16 port);
+ void connect (Address address);
+ void disconnect ();
+ void disconnect_peer(u16 peer_id);
+ void send (u16 peer_id, u8 channelnum,
+ SharedBuffer<u8> data);
+ void sendReliable (ConnectionCommand &c);
+ void sendToAll (u8 channelnum,
+ SharedBuffer<u8> data);
+ void sendToAllReliable(ConnectionCommand &c);
+
+ void sendPackets (float dtime);
+
+ void sendAsPacket (u16 peer_id, u8 channelnum,
+ SharedBuffer<u8> data,bool ack=false);
+
+ void sendAsPacketReliable(BufferedPacket& p, Channel* channel);
+
+ bool packetsQueued();
+
+ Connection* m_connection;
+ unsigned int m_max_packet_size;
+ float m_timeout;
+ Queue<OutgoingPacket> m_outgoing_queue;
+ JSemaphore m_send_sleep_semaphore;
+
+ unsigned int m_iteration_packets_avaialble;
+ unsigned int m_max_commands_per_iteration;
+ unsigned int m_max_data_packets_per_iteration;
+ unsigned int m_max_packets_requeued;
};
-struct ConnectionCommand
-{
- enum ConnectionCommandType type;
- u16 port;
- Address address;
- u16 peer_id;
- u8 channelnum;
- Buffer<u8> data;
- bool reliable;
-
- ConnectionCommand(): type(CONNCMD_NONE) {}
+class ConnectionReceiveThread : public JThread {
+public:
+ ConnectionReceiveThread(Connection* parent,
+ unsigned int max_packet_size);
- void serve(u16 port_)
- {
- type = CONNCMD_SERVE;
- port = port_;
- }
- void connect(Address address_)
- {
- type = CONNCMD_CONNECT;
- address = address_;
- }
- void disconnect()
- {
- type = CONNCMD_DISCONNECT;
- }
- void send(u16 peer_id_, u8 channelnum_,
- SharedBuffer<u8> data_, bool reliable_)
- {
- type = CONNCMD_SEND;
- peer_id = peer_id_;
- channelnum = channelnum_;
- data = data_;
- reliable = reliable_;
- }
- void sendToAll(u8 channelnum_, SharedBuffer<u8> data_, bool reliable_)
- {
- type = CONNCMD_SEND_TO_ALL;
- channelnum = channelnum_;
- data = data_;
- reliable = reliable_;
- }
- void deletePeer(u16 peer_id_)
- {
- type = CONNCMD_DELETE_PEER;
- peer_id = peer_id_;
- }
+ void * Thread ();
+
+private:
+ void receive ();
+
+ // Returns next data from a buffer if possible
+ // If found, returns true; if not, false.
+ // If found, sets peer_id and dst
+ bool getFromBuffers (u16 &peer_id, SharedBuffer<u8> &dst);
+
+ bool checkIncomingBuffers(Channel *channel, u16 &peer_id,
+ SharedBuffer<u8> &dst);
+
+ /*
+ Processes a packet with the basic header stripped out.
+ Parameters:
+ packetdata: Data in packet (with no base headers)
+ peer_id: peer id of the sender of the packet in question
+ channelnum: channel on which the packet was sent
+ reliable: true if recursing into a reliable packet
+ */
+ SharedBuffer<u8> processPacket(Channel *channel,
+ SharedBuffer<u8> packetdata, u16 peer_id,
+ u8 channelnum, bool reliable);
+
+
+ Connection* m_connection;
+ unsigned int m_max_packet_size;
};
-class Connection: public JThread
+class Connection
{
public:
+ friend class ConnectionSendThread;
+ friend class ConnectionReceiveThread;
+
Connection(u32 protocol_id, u32 max_packet_size, float timeout, bool ipv6);
Connection(u32 protocol_id, u32 max_packet_size, float timeout, bool ipv6,
PeerHandler *peerhandler);
~Connection();
- void * Thread();
/* Interface */
-
ConnectionEvent getEvent();
ConnectionEvent waitEvent(u32 timeout_ms);
void putCommand(ConnectionCommand &c);
Address GetPeerAddress(u16 peer_id);
float GetPeerAvgRTT(u16 peer_id);
void DeletePeer(u16 peer_id);
-
-private:
+ const u32 GetProtocolID() const { return m_protocol_id; };
+ const std::string getDesc();
+
+protected:
+ PeerHelper getPeer(u16 peer_id);
+ PeerHelper getPeerNoEx(u16 peer_id);
+ u16 lookupPeer(Address& sender);
+
+ u16 createPeer(Address& sender, MTProtocols protocol, int fd);
+ UDPPeer* createServerPeer(Address& sender);
+ bool deletePeer(u16 peer_id, bool timeout);
+
+ void SetPeerID(u16 id){ m_peer_id = id; }
+
+ void sendAck(u16 peer_id, u8 channelnum, u16 seqnum);
+
+ void PrintInfo(std::ostream &out);
+ void PrintInfo();
+
+ std::list<u16> getPeerIDs();
+
+ UDPSocket m_udpSocket;
+ MutexedQueue<ConnectionCommand> m_command_queue;
+
void putEvent(ConnectionEvent &e);
- void processCommand(ConnectionCommand &c);
- void send(float dtime);
- void receive();
- void runTimeouts(float dtime);
- void serve(u16 port);
- void connect(Address address);
- void disconnect();
- void sendToAll(u8 channelnum, SharedBuffer<u8> data, bool reliable);
- void send(u16 peer_id, u8 channelnum, SharedBuffer<u8> data, bool reliable);
- void sendAsPacket(u16 peer_id, u8 channelnum,
- SharedBuffer<u8> data, bool reliable);
- void rawSendAsPacket(u16 peer_id, u8 channelnum,
- SharedBuffer<u8> data, bool reliable);
- void rawSend(const BufferedPacket &packet);
- Peer* getPeer(u16 peer_id);
- Peer* getPeerNoEx(u16 peer_id);
+
+private:
std::list<Peer*> getPeers();
- bool getFromBuffers(u16 &peer_id, SharedBuffer<u8> &dst);
- // Returns next data from a buffer if possible
- // If found, returns true; if not, false.
- // If found, sets peer_id and dst
- bool checkIncomingBuffers(Channel *channel, u16 &peer_id,
- SharedBuffer<u8> &dst);
- /*
- Processes a packet with the basic header stripped out.
- Parameters:
- packetdata: Data in packet (with no base headers)
- peer_id: peer id of the sender of the packet in question
- channelnum: channel on which the packet was sent
- reliable: true if recursing into a reliable packet
- */
- SharedBuffer<u8> processPacket(Channel *channel,
- SharedBuffer<u8> packetdata, u16 peer_id,
- u8 channelnum, bool reliable);
- bool deletePeer(u16 peer_id, bool timeout);
-
- Queue<OutgoingPacket> m_outgoing_queue;
+
MutexedQueue<ConnectionEvent> m_event_queue;
- MutexedQueue<ConnectionCommand> m_command_queue;
-
- u32 m_protocol_id;
- u32 m_max_packet_size;
- float m_timeout;
- UDPSocket m_socket;
+
u16 m_peer_id;
+ u32 m_protocol_id;
std::map<u16, Peer*> m_peers;
JMutex m_peers_mutex;
+ ConnectionSendThread m_sendThread;
+ ConnectionReceiveThread m_receiveThread;
+
+ JMutex m_info_mutex;
+
// Backwards compatibility
PeerHandler *m_bc_peerhandler;
int m_bc_receive_timeout;
-
- void SetPeerID(u16 id){ m_peer_id = id; }
- u32 GetProtocolID(){ return m_protocol_id; }
- void PrintInfo(std::ostream &out);
- void PrintInfo();
- std::string getDesc();
- u16 m_indentation;
+
+ bool m_shutting_down;
};
} // namespace
projects / oweals / minetest.git / commitdiff