diff options
| -rw-r--r-- | minetest.conf.example | 5 | ||||
| -rw-r--r-- | src/client.cpp | 12 | ||||
| -rw-r--r-- | src/connection.cpp | 2813 | ||||
| -rw-r--r-- | src/connection.h | 780 | ||||
| -rw-r--r-- | src/defaultsettings.cpp | 2 | ||||
| -rw-r--r-- | src/log.cpp | 2 | ||||
| -rw-r--r-- | src/server.cpp | 22 | ||||
| -rw-r--r-- | src/server.h | 2 |
8 files changed, 2663 insertions, 975 deletions
diff --git a/minetest.conf.example b/minetest.conf.example index 674d253fb..e63bd70f7 100644 --- a/minetest.conf.example +++ b/minetest.conf.example @@ -345,7 +345,10 @@ # Number of emerge threads to use. Make this field blank, or increase this number, to use multiple threads. # On multiprocessor systems, this will improve mapgen speed greatly, at the cost of slightly buggy caves. #num_emerge_threads = 1 - +# maximum number of packets sent per send step, if you have a slow connection +# try reducing it, but don't reduce it to a number below double of targeted +# client number +#max_packets_per_iteration = 1024 # # Physics stuff # diff --git a/src/client.cpp b/src/client.cpp index 721c413c0..1c7ecf3f3 100644 --- a/src/client.cpp +++ b/src/client.cpp @@ -538,7 +538,7 @@ void Client::step(float dtime) writeU16(&data[53], CLIENT_PROTOCOL_VERSION_MAX); // Send as unreliable - Send(0, data, false); + Send(1, data, false); } // Not connected, return @@ -597,7 +597,7 @@ void Client::step(float dtime) writeV3S16(&reply[2+1+6*k], *j); k++; } - m_con.Send(PEER_ID_SERVER, 1, reply, true); + m_con.Send(PEER_ID_SERVER, 2, reply, true); if(i == deleted_blocks.end()) break; @@ -745,7 +745,7 @@ void Client::step(float dtime) reply[2] = 1; writeV3S16(&reply[3], r.p); // Send as reliable - m_con.Send(PEER_ID_SERVER, 1, reply, true); + m_con.Send(PEER_ID_SERVER, 2, reply, true); } } if(num_processed_meshes > 0) @@ -840,7 +840,7 @@ void Client::step(float dtime) std::string s = os.str(); SharedBuffer<u8> data((u8*)s.c_str(), s.size()); // Send as reliable - Send(0, data, true); + Send(1, data, true); } } } @@ -957,7 +957,7 @@ void Client::request_media(const std::list<std::string> &file_requests) std::string s = os.str(); SharedBuffer<u8> data((u8*)s.c_str(), s.size()); // Send as reliable - Send(0, data, true); + Send(1, data, true); infostream<<"Client: Sending media request list to server (" <<file_requests.size()<<" files)"<<std::endl; } @@ -970,7 +970,7 @@ void Client::received_media() std::string s = os.str(); SharedBuffer<u8> data((u8*)s.c_str(), s.size()); // Send as reliable - Send(0, data, true); + Send(1, data, true); infostream<<"Client: Notifying server that we received all media" <<std::endl; } diff --git a/src/connection.cpp b/src/connection.cpp index bc9279649..ee03b2113 100644 --- a/src/connection.cpp +++ b/src/connection.cpp @@ -17,6 +17,8 @@ with this program; if not, write to the Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. */ +#include <iomanip> +#include <errno.h> #include "connection.h" #include "main.h" #include "serialization.h" @@ -26,10 +28,47 @@ with this program; if not, write to the Free Software Foundation, Inc., #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]); @@ -170,26 +209,38 @@ SharedBuffer<u8> makeReliablePacket( 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(); @@ -207,79 +258,145 @@ RPBSearchResult ReliablePacketBuffer::notFound() { 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) { @@ -288,35 +405,22 @@ void ReliablePacketBuffer::incrementTimeouts(float dtime) } } -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; } @@ -327,6 +431,7 @@ std::list<BufferedPacket> ReliablePacketBuffer::getTimedOuts(float timeout) IncomingSplitBuffer::~IncomingSplitBuffer() { + JMutexAutoLock listlock(m_map_mutex); for(std::map<u16, IncomingSplitPacket*>::iterator i = m_buf.begin(); i != m_buf.end(); ++i) { @@ -339,6 +444,7 @@ IncomingSplitBuffer::~IncomingSplitBuffer() */ 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]); @@ -360,13 +466,13 @@ SharedBuffer<u8> IncomingSplitBuffer::insert(BufferedPacket &p, bool reliable) // 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 @@ -416,22 +522,25 @@ SharedBuffer<u8> IncomingSplitBuffer::insert(BufferedPacket &p, bool reliable) 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); } @@ -441,796 +550,1615 @@ void IncomingSplitBuffer::removeUnreliableTimedOuts(float dtime, float timeout) 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 -*/ +u16 Channel::readNextIncomingSeqNum() +{ + JMutexAutoLock internal(m_internal_mutex); + return next_incoming_seqnum; +} -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::incNextIncomingSeqNum() +{ + JMutexAutoLock internal(m_internal_mutex); + u16 retval = next_incoming_seqnum; + next_incoming_seqnum++; + return retval; +} + +u16 Channel::readNextSplitSeqNum() { + JMutexAutoLock internal(m_internal_mutex); + return next_outgoing_split_seqnum; } -Peer::~Peer() +void Channel::setNextSplitSeqNum(u16 seqnum) { + JMutexAutoLock internal(m_internal_mutex); + next_outgoing_split_seqnum = seqnum; } -void Peer::reportRTT(float rtt) +u16 Channel::getOutgoingSequenceNumber(bool& successfull) { - 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; + 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; + } + + if (outgoing_reliables_sent.getFirstSeqnum(lowest_unacked_seqnumber)) + { + 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; + } } } - 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 + next_outgoing_seqnum++; + return retval; +} + +u16 Channel::readOutgoingSequenceNumber() +{ + JMutexAutoLock internal(m_internal_mutex); + return next_outgoing_seqnum; +} + +bool Channel::putBackSequenceNumber(u16 seqnum) +{ + if (((seqnum + 1) % (SEQNUM_MAX+1)) == next_outgoing_seqnum) { + + next_outgoing_seqnum = seqnum; + return true; + } + return false; +} + +void Channel::UpdateBytesSent(unsigned int bytes, unsigned int packets) +{ + JMutexAutoLock internal(m_internal_mutex); + current_bytes_transfered += bytes; + current_packet_successfull += packets; +} + +void Channel::UpdateBytesLost(unsigned int bytes) +{ + JMutexAutoLock internal(m_internal_mutex); + current_bytes_lost += bytes; +} + + +void Channel::UpdatePacketLossCounter(unsigned int count) +{ + JMutexAutoLock internal(m_internal_mutex); + current_packet_loss += count; +} + +void Channel::UpdatePacketTooLateCounter() +{ + JMutexAutoLock internal(m_internal_mutex); + current_packet_too_late++; +} + +void Channel::UpdateTimers(float dtime) +{ + bpm_counter += dtime; + packet_loss_counter += dtime; + + if (packet_loss_counter > 1.0) + { + 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; + + { + 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; + } + + float successfull_to_lost_ratio = 0.0; + bool done = false; + + 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; + } - /*int reliable_count = 0; - for(int i=0; i<CHANNEL_COUNT; i++) + if (!done) + { + if ((successfull_to_lost_ratio < 0.01) && + (window_size < MAX_RELIABLE_WINDOW_SIZE)) + { + /* 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); + } + else if ((successfull_to_lost_ratio < 0.05) && + (window_size < MAX_RELIABLE_WINDOW_SIZE)) + { + /* 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 if (successfull_to_lost_ratio > 0.15) + { + window_size = MYMAX( + (window_size - 100), + MIN_RELIABLE_WINDOW_SIZE); + } + else if (successfull_to_lost_ratio > 0.1) + { + window_size = MYMAX( + (window_size - 50), + MIN_RELIABLE_WINDOW_SIZE); + } + } + } + + if (bpm_counter > 10.0) { - reliable_count += channels[i].outgoing_reliables.size(); + { + 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; + } + + if (cur_kbps > max_kbps) + { + max_kbps = cur_kbps; + } + + if (cur_kbps_lost > max_kbps_lost) + { + max_kbps_lost = cur_kbps_lost; + } + + avg_kbps = avg_kbps * 0.9 + cur_kbps * 0.1; + avg_kbps_lost = avg_kbps_lost * 0.9 + cur_kbps_lost * 0.1; } - 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; } - + + /* - Connection + Peer */ -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) +PeerHelper::PeerHelper() : + m_peer(0) { - m_socket.setTimeoutMs(5); +} - Start(); +PeerHelper::PeerHelper(Peer* peer) : + m_peer(peer) +{ + if (peer != NULL) + { + if (!peer->IncUseCount()) + { + m_peer = 0; + } + } } -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) +PeerHelper::~PeerHelper() { - m_socket.setTimeoutMs(5); + if (m_peer != 0) + m_peer->DecUseCount(); - Start(); + m_peer = 0; } - -Connection::~Connection() +PeerHelper& PeerHelper::operator=(Peer* peer) { - Stop(); - // Delete peers - for(std::map<u16, Peer*>::iterator - j = m_peers.begin(); - j != m_peers.end(); ++j) + m_peer = peer; + if (peer != NULL) { - delete j->second; + if (!peer->IncUseCount()) + { + m_peer = 0; + } } + return *this; } -/* Internal stuff */ +Peer* PeerHelper::operator->() const +{ + return m_peer; +} -void * Connection::Thread() +Peer* PeerHelper::operator&() const { - ThreadStarted(); - log_register_thread("Connection"); + return m_peer; +} - dout_con<<"Connection thread started"<<std::endl; - - u32 curtime = porting::getTimeMs(); - u32 lasttime = curtime; +bool PeerHelper::operator!() { + return ! m_peer; +} - while(!StopRequested()) +bool PeerHelper::operator!=(void* ptr) +{ + return ((void*) m_peer != ptr); +} + +bool Peer::IncUseCount() +{ + JMutexAutoLock lock(m_exclusive_access_mutex); + + if (!m_pending_deletion) { - 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); + 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; - //NOTE this is only thread safe for ONE consumer thread! - while(!m_command_queue.empty()){ - ConnectionCommand c = m_command_queue.pop_frontNoEx(); - processCommand(c); + 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; +} - send(dtime); +bool Peer::isTimedOut(float timeout) +{ + JMutexAutoLock lock(m_exclusive_access_mutex); + u32 current_time = porting::getTimeMs(); - receive(); - - END_DEBUG_EXCEPTION_HANDLER(derr_con); + 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; } - return NULL; + 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; } -void Connection::putEvent(ConnectionEvent &e) +UDPPeer::UDPPeer(u16 a_id, Address a_address, Connection* connection) : + Peer(a_address,a_id,connection), + resend_timeout(0.5), + m_legacy_peer(true) { - assert(e.type != CONNEVENT_NONE); - m_event_queue.push_back(e); } -void Connection::processCommand(ConnectionCommand &c) +UDPPeer::~UDPPeer() { - 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); +} + +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; } -void Connection::send(float dtime) +bool UDPPeer::Ping(float dtime,SharedBuffer<u8>& data) { - for(std::map<u16, Peer*>::iterator - j = m_peers.begin(); - j != m_peers.end(); ++j) + m_ping_timer += dtime; + if(m_ping_timer >= PING_TIMEOUT) { - 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; + // Create and send PING packet + writeU8(&data[0], TYPE_CONTROL); + writeU8(&data[1], CONTROLTYPE_PING); + m_ping_timer = 0.0; + return true; } - 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); + 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); } } - 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) + else { - 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; + 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); } } -// Receive packets from the network and buffers and create ConnectionEvents -void Connection::receive() +bool UDPPeer::processReliableSendCommand( + ConnectionCommand &c, + unsigned int max_packet_size) { - 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); + u32 chunksize_max = max_packet_size + - BASE_HEADER_SIZE + - RELIABLE_HEADER_SIZE; - bool single_wait_done = false; - - for(u32 loop_i=0; loop_i<1000; loop_i++) // Limit in case of DoS + 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) { - try{ - /* Check if some buffer has relevant data */ - { - u16 peer_id; - SharedBuffer<u8> resultdata; - bool got = getFromBuffers(peer_id, resultdata); - if(got){ - ConnectionEvent e; - e.dataReceived(peer_id, resultdata); - putEvent(e); - continue; - } - } - - if(single_wait_done){ - if(m_socket.WaitData(0) == false) - break; - } - - single_wait_done = true; + originals.push_back(c.data); + } + else { + originals = makeAutoSplitPacket(c.data, chunksize_max,split_sequence_number); + channels[c.channelnum].setNextSplitSeqNum(split_sequence_number); + } - Address sender; - s32 received_size = m_socket.Receive(sender, *packetdata, packet_maxsize); + bool have_sequence_number = true; + bool have_initial_sequence_number = false; + Queue<BufferedPacket> toadd; + volatile u16 initial_sequence_number = 0; - if(received_size < 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(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 (!have_initial_sequence_number) + { + initial_sequence_number = seqnum; + have_initial_sequence_number = true; } - if(peer_id == PEER_ID_INEXISTENT) + 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) { - /* - 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) - { - Peer *peer = j->second; - if(peer->has_sent_with_id) - continue; - if(peer->address == sender) - break; - } - - /* - 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()) - { - // Pass on to adding the peer - } - // Else: A peer was found. - else - { - 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; - } + return false; } - - /* - The peer was not found in our lists. Add it. - */ - if(peer_id == PEER_ID_INEXISTENT) + 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; + } +} + +void UDPPeer::RunCommandQueues( + unsigned int max_packet_size, + unsigned int maxcommands, + unsigned int maxtransfer) +{ + + 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)) { - // 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(;;) - { - // 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; + 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); } - peer_id_new++; } - if(out_of_ids){ - errorstream<<getDesc()<<" ran out of peer ids"<<std::endl; - continue; + catch (ItemNotFoundException e) { + // intentionally empty } + } + } +} - PrintInfo(); - dout_con<<"Receive(): Got a packet with peer_id=PEER_ID_INEXISTENT," - " giving peer_id="<<peer_id_new<<std::endl; +u16 UDPPeer::getNextSplitSequenceNumber(u8 channel) +{ + assert(channel < CHANNEL_COUNT); + return channels[channel].readNextIncomingSeqNum(); +} - // 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; +void UDPPeer::setNextSplitSequenceNumber(u8 channel, u16 seqnum) +{ + assert(channel < CHANNEL_COUNT); + channels[channel].setNextSplitSeqNum(seqnum); +} - // Go on and process whatever it sent - } +SharedBuffer<u8> UDPPeer::addSpiltPacket(u8 channel, + BufferedPacket toadd, + bool reliable) +{ + assert(channel < CHANNEL_COUNT); + return channels[channel].incoming_splits.insert(toadd,reliable); +} - std::map<u16, Peer*>::iterator node = m_peers.find(peer_id); +/******************************************************************************/ +/* Connection Threads */ +/******************************************************************************/ - if(node == m_peers.end()) - { - // 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)"); - } +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) +{ +} - Peer *peer = node->second; +void * ConnectionSendThread::Thread() +{ + ThreadStarted(); + log_register_thread("ConnectionSend"); - // Validate peer address - if(peer->address != sender) - { - PrintInfo(derr_con); - derr_con<<"Peer "<<peer_id<<" sending from different address." - " Ignoring."<<std::endl; - continue; + 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); } - - peer->timeout_counter = 0.0; - Channel *channel = &(peer->channels[channelnum]); - - // Throw the received packet to channel->processPacket() + /* send non reliable packets */ + sendPackets(dtime); - // 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); + 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; - }catch(ProcessedSilentlyException &e){ + + 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; + } } - }catch(InvalidIncomingDataException &e){ - } - catch(ProcessedSilentlyException &e){ } - } // for + + + return false; } -void Connection::runTimeouts(float dtime) +void ConnectionSendThread::runTimeouts(float dtime) { - 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"); - 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->UpdatePacketLossCounter(timed_outs.size()); - channel->outgoing_reliables.resetTimedOuts(resend_timeout); + 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])); - PrintInfo(derr_con); - derr_con<<"RE-SENDING timed-out RELIABLE to "; - j->address.print(&derr_con); - derr_con<<"(t/o="<<resend_timeout<<"): " + 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)channel&0xff) + <<", channel="<<((int)channelnum&0xff) <<", seqnum="<<seqnum - <<std::endl; + <<std::endl); rawSend(*j); - // 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); + // 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 pings - */ - peer->ping_timer += dtime; - if(peer->ping_timer >= 5.0) - { - // 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); - peer->ping_timer = 0.0; + /* 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); + } } - -nextpeer: - continue; + + 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) { - PrintInfo(derr_con); - derr_con<<"RunTimeouts(): Removing peer "<<(*i)<<std::endl; - deletePeer(*i, true); + LOG(derr_con<<m_connection->getDesc() + <<"RunTimeouts(): Removing peer "<<(*i)<<std::endl); + m_connection->deletePeer(*i, true); } } -void Connection::serve(u16 port) +void ConnectionSendThread::rawSend(const BufferedPacket &packet) { - dout_con<<getDesc()<<" serving at port "<<port<<std::endl; try{ - m_socket.Bind(port); - m_peer_id = PEER_ID_SERVER; + 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; + + case CONNCMD_SEND: + LOG(dout_con<<m_connection->getDesc()<<"UDP processing reliable CONNCMD_SEND"<<std::endl); + sendReliable(c); + return; + + case CONNCMD_SEND_TO_ALL: + LOG(dout_con<<m_connection->getDesc()<<"UDP processing CONNCMD_SEND_TO_ALL"<<std::endl); + sendToAllReliable(c); + return; + + 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)) + { + /* put to queue if we couldn't send it immediately */ + sendReliable(c); + } + return; + + 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); + } + 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); + } +} + + +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 ConnectionSendThread::serve(u16 port) +{ + LOG(dout_con<<m_connection->getDesc()<<"UDP serving at port "<<port<<std::endl); + try{ + 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; + LOG(dout_con<<m_connection->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"); - } - - 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; } - catch(AlreadyExistsException &e) + peer->m_increment_packets_remaining = m_iteration_packets_avaialble/m_connection->m_peers.size(); + + if (dynamic_cast<UDPPeer*>(&peer) == 0) { - PrintInfo(derr_con); - derr_con<<"WARNING: Going to send a reliable packet " - "seqnum="<<seqnum<<" that is already " - "in outgoing buffer"<<std::endl; - //assert(0); + continue; + } + + 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++) + { + 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); - return node->second; -} + PROFILE(std::stringstream ThreadIdentifier); + PROFILE(ThreadIdentifier << "ConnectionReceive: [" << m_connection->getDesc() << "]"); -Peer* Connection::getPeerNoEx(u16 peer_id) -{ - std::map<u16, Peer*>::iterator node = m_peers.find(peer_id); +#ifdef DEBUG_CONNECTION_KBPS + u32 curtime = porting::getTimeMs(); + u32 lasttime = curtime; + float debug_print_timer = 0.0; +#endif - if(node == m_peers.end()){ - return NULL; - } + while(!StopRequested()) { + BEGIN_DEBUG_EXCEPTION_HANDLER + PROFILE(ScopeProfiler sp(g_profiler, ThreadIdentifier.str(), SPT_AVG)); - // Error checking - assert(node->second->id == peer_id); +#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(); + } - return node->second; + 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; } -std::list<Peer*> Connection::getPeers() +// Receive packets from the network and buffers and create ConnectionEvents +void ConnectionReceiveThread::receive() { - std::list<Peer*> list; - for(std::map<u16, Peer*>::iterator j = m_peers.begin(); - j != m_peers.end(); ++j) + /* 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; + + 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))) { - Peer *peer = j->second; - list.push_back(peer); + 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){ + } } - 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){ @@ -1242,41 +2170,28 @@ bool Connection::getFromBuffers(u16 &peer_id, SharedBuffer<u8> &dst) 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); @@ -1289,17 +2204,17 @@ bool Connection::checkIncomingBuffers(Channel *channel, u16 &peer_id, 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) @@ -1307,94 +2222,120 @@ SharedBuffer<u8> Connection::processPacket(Channel *channel, 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"); } } @@ -1403,9 +2344,9 @@ SharedBuffer<u8> Connection::processPacket(Channel *channel, 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()); @@ -1413,29 +2354,40 @@ SharedBuffer<u8> Connection::processPacket(Channel *channel, } 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"); @@ -1445,71 +2397,104 @@ SharedBuffer<u8> Connection::processPacket(Channel *channel, ("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); @@ -1519,11 +2504,11 @@ SharedBuffer<u8> Connection::processPacket(Channel *channel, } 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. @@ -1531,20 +2516,174 @@ SharedBuffer<u8> Connection::processPacket(Channel *channel, 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; } @@ -1573,7 +2712,11 @@ ConnectionEvent Connection::waitEvent(u32 timeout_ms) 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) @@ -1603,7 +2746,7 @@ bool Connection::Connected() if(m_peer_id == PEER_ID_INEXISTENT) return false; - + return true; } @@ -1619,8 +2762,8 @@ u32 Connection::Receive(u16 &peer_id, SharedBuffer<u8> &data) 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"); @@ -1629,12 +2772,12 @@ u32 Connection::Receive(u16 &peer_id, SharedBuffer<u8> &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; } @@ -1672,14 +2815,78 @@ void Connection::RunTimeouts(float dtime) 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) @@ -1691,7 +2898,9 @@ 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() @@ -1699,9 +2908,59 @@ 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 diff --git a/src/connection.h b/src/connection.h index 56badc904..e43d93fb3 100644 --- a/src/connection.h +++ b/src/connection.h @@ -27,6 +27,7 @@ with this program; if not, write to the Free Software Foundation, Inc., #include "util/pointer.h" #include "util/container.h" #include "util/thread.h" +#include "util/numeric.h" #include <iostream> #include <fstream> #include <list> @@ -110,26 +111,74 @@ public: {} }; +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 }; @@ -223,6 +272,8 @@ controltype and data description: #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. @@ -261,7 +312,6 @@ with a buffer in the receiving and transmitting end. */ #define TYPE_RELIABLE 3 #define RELIABLE_HEADER_SIZE 3 -//#define SEQNUM_INITIAL 0x10 #define SEQNUM_INITIAL 65500 /* @@ -275,23 +325,35 @@ class ReliablePacketBuffer { 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; }; /* @@ -313,27 +375,207 @@ public: 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; @@ -360,71 +602,232 @@ public: 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, @@ -485,76 +888,108 @@ struct ConnectionEvent } }; -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); @@ -572,68 +1007,53 @@ public: 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 diff --git a/src/defaultsettings.cpp b/src/defaultsettings.cpp index 659682b8f..053602833 100644 --- a/src/defaultsettings.cpp +++ b/src/defaultsettings.cpp @@ -172,6 +172,8 @@ void set_default_settings(Settings *settings) // Server stuff // "map-dir" doesn't exist by default. + settings->setDefault("workaround_window_size","5"); + settings->setDefault("max_packets_per_iteration","1024"); settings->setDefault("port", "30000"); settings->setDefault("default_game", "minetest"); settings->setDefault("motd", ""); diff --git a/src/log.cpp b/src/log.cpp index 527f54480..97f25cc77 100644 --- a/src/log.cpp +++ b/src/log.cpp @@ -93,6 +93,7 @@ static std::string get_lev_string(enum LogMessageLevel lev) void log_printline(enum LogMessageLevel lev, const std::string &text) { + log_threadnamemutex.Lock(); std::string threadname = "(unknown thread)"; std::map<threadid_t, std::string>::const_iterator i; i = log_threadnames.find(get_current_thread_id()); @@ -108,6 +109,7 @@ void log_printline(enum LogMessageLevel lev, const std::string &text) out->printLog(os.str(), lev); out->printLog(lev, text); } + log_threadnamemutex.Unlock(); } class Logbuf : public std::streambuf diff --git a/src/server.cpp b/src/server.cpp index d8bedb260..b257448a1 100644 --- a/src/server.cpp +++ b/src/server.cpp @@ -90,14 +90,15 @@ public: void * ServerThread::Thread() { - ThreadStarted(); - log_register_thread("ServerThread"); DSTACK(__FUNCTION_NAME); - BEGIN_DEBUG_EXCEPTION_HANDLER + m_server->AsyncRunStep(true); + + ThreadStarted(); + while(!StopRequested()) { try{ @@ -1018,7 +1019,7 @@ void Server::step(float dtime) } } -void Server::AsyncRunStep() +void Server::AsyncRunStep(bool initial_step) { DSTACK(__FUNCTION_NAME); @@ -1035,7 +1036,7 @@ void Server::AsyncRunStep() SendBlocks(dtime); } - if(dtime < 0.001) + if((dtime < 0.001) && (initial_step == false)) return; g_profiler->add("Server::AsyncRunStep with dtime (num)", 1); @@ -1528,7 +1529,7 @@ void Server::AsyncRunStep() memcpy((char*)&reply[2], unreliable_data.c_str(), unreliable_data.size()); // Send as unreliable - m_con.Send(client->peer_id, 0, reply, false); + m_con.Send(client->peer_id, 1, reply, false); } /*if(reliable_data.size() > 0 || unreliable_data.size() > 0) @@ -3721,7 +3722,7 @@ void Server::SendHUDAdd(u16 peer_id, u32 id, HudElement *form) std::string s = os.str(); SharedBuffer<u8> data((u8*)s.c_str(), s.size()); // Send as reliable - m_con.Send(peer_id, 0, data, true); + m_con.Send(peer_id, 1, data, true); } void Server::SendHUDRemove(u16 peer_id, u32 id) @@ -3736,7 +3737,8 @@ void Server::SendHUDRemove(u16 peer_id, u32 id) std::string s = os.str(); SharedBuffer<u8> data((u8*)s.c_str(), s.size()); // Send as reliable - m_con.Send(peer_id, 0, data, true); + + m_con.Send(peer_id, 1, data, true); } void Server::SendHUDChange(u16 peer_id, u32 id, HudElementStat stat, void *value) @@ -4204,7 +4206,7 @@ void Server::SendBlockNoLock(u16 peer_id, MapBlock *block, u8 ver, u16 net_proto /* Send packet */ - m_con.Send(peer_id, 1, reply, true); + m_con.Send(peer_id, 2, reply, true); } void Server::SendBlocks(float dtime) @@ -4566,7 +4568,7 @@ void Server::sendRequestedMedia(u16 peer_id, <<" size=" <<s.size()<<std::endl; SharedBuffer<u8> data((u8*)s.c_str(), s.size()); // Send as reliable - m_con.Send(peer_id, 0, data, true); + m_con.Send(peer_id, 2, data, true); } } diff --git a/src/server.h b/src/server.h index 39e715aa3..aeae216a6 100644 --- a/src/server.h +++ b/src/server.h @@ -330,7 +330,7 @@ public: // Actual processing is done in an another thread. void step(float dtime); // This is run by ServerThread and does the actual processing - void AsyncRunStep(); + void AsyncRunStep(bool initial_step=false); void Receive(); void ProcessData(u8 *data, u32 datasize, u16 peer_id); |