X-Git-Url: https://git.lizzy.rs/?a=blobdiff_plain;f=src%2Fnetwork%2Fconnection.cpp;h=0ba8c36b28feef2cf22babd7e2c1ab93240bae6c;hb=674d67f312c815e7f10dc00705e352bc392fc2af;hp=e11b4a953d55f9927044d1f9a85bd138491d53c5;hpb=abd68d3466b7f2155cf0f1c4172a254f10c1f02e;p=minetest.git diff --git a/src/network/connection.cpp b/src/network/connection.cpp index e11b4a953..0ba8c36b2 100644 --- a/src/network/connection.cpp +++ b/src/network/connection.cpp @@ -18,12 +18,16 @@ with this program; if not, write to the Free Software Foundation, Inc., */ #include -#include +#include +#include +#include #include "connection.h" #include "serialization.h" #include "log.h" #include "porting.h" +#include "network/connectionthreads.h" #include "network/networkpacket.h" +#include "network/peerhandler.h" #include "util/serialize.h" #include "util/numeric.h" #include "util/string.h" @@ -37,56 +41,31 @@ namespace con /* defines used for debugging and profiling */ /******************************************************************************/ #ifdef NDEBUG -#define LOG(a) a -#define PROFILE(a) -#undef DEBUG_CONNECTION_KBPS + #define LOG(a) a + #define PROFILE(a) #else -/* this mutex is used to achieve log message consistency */ -Mutex log_message_mutex; -#define LOG(a) \ - { \ - MutexAutoLock loglock(log_message_mutex); \ - a; \ - } -#define PROFILE(a) a -//#define DEBUG_CONNECTION_KBPS -#undef DEBUG_CONNECTION_KBPS + #if 0 + /* this mutex is used to achieve log message consistency */ + std::mutex log_message_mutex; + #define LOG(a) \ + { \ + MutexAutoLock loglock(log_message_mutex); \ + a; \ + } + #else + // Prevent deadlocks until a solution is found after 5.2.0 (TODO) + #define LOG(a) a + #endif + + #define PROFILE(a) a #endif - -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 -/* maximum number of retries for reliable packets */ -#define MAX_RELIABLE_RETRY 5 - -static u16 readPeerId(u8 *packetdata) +BufferedPacket makePacket(Address &address, const SharedBuffer &data, + u32 protocol_id, session_t sender_peer_id, u8 channel) { - return readU16(&packetdata[4]); -} -static u8 readChannel(u8 *packetdata) -{ - return readU8(&packetdata[6]); -} - -BufferedPacket makePacket(Address &address, u8 *data, u32 datasize, - u32 protocol_id, u16 sender_peer_id, u8 channel) -{ - u32 packet_size = datasize + BASE_HEADER_SIZE; + u32 packet_size = data.getSize() + BASE_HEADER_SIZE; BufferedPacket p(packet_size); p.address = address; @@ -94,47 +73,36 @@ BufferedPacket makePacket(Address &address, u8 *data, u32 datasize, writeU16(&p.data[4], sender_peer_id); writeU8(&p.data[6], channel); - memcpy(&p.data[BASE_HEADER_SIZE], data, datasize); + memcpy(&p.data[BASE_HEADER_SIZE], *data, data.getSize()); return p; } -BufferedPacket makePacket(Address &address, SharedBuffer &data, - u32 protocol_id, u16 sender_peer_id, u8 channel) -{ - return makePacket(address, *data, data.getSize(), - protocol_id, sender_peer_id, channel); -} - -SharedBuffer makeOriginalPacket( - SharedBuffer data) +SharedBuffer makeOriginalPacket(const SharedBuffer &data) { u32 header_size = 1; u32 packet_size = data.getSize() + header_size; SharedBuffer b(packet_size); - writeU8(&(b[0]), TYPE_ORIGINAL); + writeU8(&(b[0]), PACKET_TYPE_ORIGINAL); if (data.getSize() > 0) { memcpy(&(b[header_size]), *data, data.getSize()); } return b; } -std::list > makeSplitPacket( - SharedBuffer data, - u32 chunksize_max, - u16 seqnum) +// Split data in chunks and add TYPE_SPLIT headers to them +void makeSplitPacket(const SharedBuffer &data, u32 chunksize_max, u16 seqnum, + std::list> *chunks) { // Chunk packets, containing the TYPE_SPLIT header - std::list > chunks; - u32 chunk_header_size = 7; u32 maximum_data_size = chunksize_max - chunk_header_size; u32 start = 0; u32 end = 0; u32 chunk_num = 0; u16 chunk_count = 0; - do{ + do { end = start + maximum_data_size - 1; if (end > data.getSize() - 1) end = data.getSize() - 1; @@ -144,59 +112,47 @@ std::list > makeSplitPacket( SharedBuffer chunk(packet_size); - writeU8(&chunk[0], TYPE_SPLIT); + writeU8(&chunk[0], PACKET_TYPE_SPLIT); writeU16(&chunk[1], seqnum); // [3] u16 chunk_count is written at next stage writeU16(&chunk[5], chunk_num); memcpy(&chunk[chunk_header_size], &data[start], payload_size); - chunks.push_back(chunk); + chunks->push_back(chunk); chunk_count++; start = end + 1; chunk_num++; } - while(end != data.getSize() - 1); + while (end != data.getSize() - 1); - for(std::list >::iterator i = chunks.begin(); - i != chunks.end(); ++i) - { + for (SharedBuffer &chunk : *chunks) { // Write chunk_count - writeU16(&((*i)[3]), chunk_count); + writeU16(&(chunk[3]), chunk_count); } - - return chunks; } -std::list > makeAutoSplitPacket( - SharedBuffer data, - u32 chunksize_max, - u16 &split_seqnum) +void makeAutoSplitPacket(const SharedBuffer &data, u32 chunksize_max, + u16 &split_seqnum, std::list> *list) { u32 original_header_size = 1; - std::list > list; - if (data.getSize() + original_header_size > chunksize_max) - { - list = makeSplitPacket(data, chunksize_max, split_seqnum); + + if (data.getSize() + original_header_size > chunksize_max) { + makeSplitPacket(data, chunksize_max, split_seqnum, list); split_seqnum++; - return list; - } - else - { - list.push_back(makeOriginalPacket(data)); + return; } - return list; + + list->push_back(makeOriginalPacket(data)); } -SharedBuffer makeReliablePacket( - SharedBuffer data, - u16 seqnum) +SharedBuffer makeReliablePacket(const SharedBuffer &data, u16 seqnum) { u32 header_size = 3; u32 packet_size = data.getSize() + header_size; SharedBuffer b(packet_size); - writeU8(&b[0], TYPE_RELIABLE); + writeU8(&b[0], PACKET_TYPE_RELIABLE); writeU16(&b[1], seqnum); memcpy(&b[header_size], *data, data.getSize()); @@ -208,22 +164,18 @@ SharedBuffer makeReliablePacket( ReliablePacketBuffer */ -ReliablePacketBuffer::ReliablePacketBuffer(): m_list_size(0) {} - void ReliablePacketBuffer::print() { MutexAutoLock listlock(m_list_mutex); LOG(dout_con<<"Dump of ReliablePacketBuffer:" << std::endl); unsigned int index = 0; - for(std::list::iterator i = m_list.begin(); - i != m_list.end(); - ++i) - { - u16 s = readU16(&(i->data[BASE_HEADER_SIZE+1])); + for (BufferedPacket &bufferedPacket : m_list) { + u16 s = readU16(&(bufferedPacket.data[BASE_HEADER_SIZE+1])); LOG(dout_con<data[BASE_HEADER_SIZE+1])); - /*dout_con<<"findPacket(): finding seqnum="<data[BASE_HEADER_SIZE + 1]); } return p; } + BufferedPacket ReliablePacketBuffer::popSeqnum(u16 seqnum) { MutexAutoLock listlock(m_list_mutex); @@ -304,15 +252,17 @@ BufferedPacket ReliablePacketBuffer::popSeqnum(u16 seqnum) } 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]); } + if (m_list.empty()) { + 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,u16 next_expected) + +void ReliablePacketBuffer::insert(BufferedPacket &p, u16 next_expected) { MutexAutoLock listlock(m_list_mutex); if (p.data.getSize() < BASE_HEADER_SIZE + 3) { @@ -321,7 +271,7 @@ void ReliablePacketBuffer::insert(BufferedPacket &p,u16 next_expected) return; } u8 type = readU8(&p.data[BASE_HEADER_SIZE + 0]); - if (type != TYPE_RELIABLE) { + if (type != PACKET_TYPE_RELIABLE) { errorstream << "ReliablePacketBuffer::insert(): type is not reliable" << std::endl; return; @@ -339,8 +289,7 @@ void ReliablePacketBuffer::insert(BufferedPacket &p,u16 next_expected) return; } - ++m_list_size; - sanity_check(m_list_size <= SEQNUM_MAX+1); // FIXME: Handle the error? + sanity_check(m_list.size() <= SEQNUM_MAX); // FIXME: Handle the error? // Find the right place for the packet and insert it there // If list is empty, just add it @@ -377,6 +326,8 @@ void ReliablePacketBuffer::insert(BufferedPacket &p,u16 next_expected) } if (s == seqnum) { + /* nothing to do this seems to be a resent packet */ + /* for paranoia reason data should be compared */ if ( (readU16(&(i->data[BASE_HEADER_SIZE+1])) != seqnum) || (i->data.getSize() != p.data.getSize()) || @@ -395,16 +346,11 @@ void ReliablePacketBuffer::insert(BufferedPacket &p,u16 next_expected) p.address.serializeString().c_str()); throw IncomingDataCorruption("duplicated packet isn't same as original one"); } - - /* 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 { + } else { m_list.push_back(p); } @@ -415,11 +361,9 @@ void ReliablePacketBuffer::insert(BufferedPacket &p,u16 next_expected) void ReliablePacketBuffer::incrementTimeouts(float dtime) { MutexAutoLock listlock(m_list_mutex); - for(std::list::iterator i = m_list.begin(); - i != m_list.end(); ++i) - { - i->time += dtime; - i->totaltime += dtime; + for (BufferedPacket &bufferedPacket : m_list) { + bufferedPacket.time += dtime; + bufferedPacket.totaltime += dtime; } } @@ -428,14 +372,12 @@ std::list ReliablePacketBuffer::getTimedOuts(float timeout, { MutexAutoLock listlock(m_list_mutex); std::list timed_outs; - for(std::list::iterator i = m_list.begin(); - i != m_list.end(); ++i) - { - if (i->time >= timeout) { - timed_outs.push_back(*i); + for (BufferedPacket &bufferedPacket : m_list) { + if (bufferedPacket.time >= timeout) { + timed_outs.push_back(bufferedPacket); //this packet will be sent right afterwards reset timeout here - i->time = 0.0; + bufferedPacket.time = 0.0f; if (timed_outs.size() >= max_packets) break; } @@ -443,6 +385,48 @@ std::list ReliablePacketBuffer::getTimedOuts(float timeout, return timed_outs; } +/* + IncomingSplitPacket +*/ + +bool IncomingSplitPacket::insert(u32 chunk_num, SharedBuffer &chunkdata) +{ + sanity_check(chunk_num < chunk_count); + + // If chunk already exists, ignore it. + // Sometimes two identical packets may arrive when there is network + // lag and the server re-sends stuff. + if (chunks.find(chunk_num) != chunks.end()) + return false; + + // Set chunk data in buffer + chunks[chunk_num] = chunkdata; + + return true; +} + +SharedBuffer IncomingSplitPacket::reassemble() +{ + sanity_check(allReceived()); + + // Calculate total size + u32 totalsize = 0; + for (const auto &chunk : chunks) + totalsize += chunk.second.getSize(); + + SharedBuffer fulldata(totalsize); + + // Copy chunks to data buffer + u32 start = 0; + for (u32 chunk_i = 0; chunk_i < chunk_count; chunk_i++) { + const SharedBuffer &buf = chunks[chunk_i]; + memcpy(&fulldata[start], *buf, buf.getSize()); + start += buf.getSize(); + } + + return fulldata; +} + /* IncomingSplitBuffer */ @@ -450,17 +434,12 @@ std::list ReliablePacketBuffer::getTimedOuts(float timeout, IncomingSplitBuffer::~IncomingSplitBuffer() { MutexAutoLock listlock(m_map_mutex); - for(std::map::iterator i = m_buf.begin(); - i != m_buf.end(); ++i) - { - delete i->second; + for (auto &i : m_buf) { + delete i.second; } } -/* - This will throw a GotSplitPacketException when a full - split packet is constructed. -*/ -SharedBuffer IncomingSplitBuffer::insert(BufferedPacket &p, bool reliable) + +SharedBuffer IncomingSplitBuffer::insert(const BufferedPacket &p, bool reliable) { MutexAutoLock listlock(m_map_mutex); u32 headersize = BASE_HEADER_SIZE + 7; @@ -473,71 +452,50 @@ SharedBuffer IncomingSplitBuffer::insert(BufferedPacket &p, bool reliable) u16 chunk_count = readU16(&p.data[BASE_HEADER_SIZE+3]); u16 chunk_num = readU16(&p.data[BASE_HEADER_SIZE+5]); - if (type != TYPE_SPLIT) { + if (type != PACKET_TYPE_SPLIT) { errorstream << "IncomingSplitBuffer::insert(): type is not split" << std::endl; return SharedBuffer(); } + if (chunk_num >= chunk_count) { + errorstream << "IncomingSplitBuffer::insert(): chunk_num=" << chunk_num + << " >= chunk_count=" << chunk_count << std::endl; + return SharedBuffer(); + } // Add if doesn't exist - if (m_buf.find(seqnum) == m_buf.end()) - { - IncomingSplitPacket *sp = new IncomingSplitPacket(); - sp->chunk_count = chunk_count; - sp->reliable = reliable; + IncomingSplitPacket *sp; + if (m_buf.find(seqnum) == m_buf.end()) { + sp = new IncomingSplitPacket(chunk_count, reliable); m_buf[seqnum] = sp; + } else { + sp = m_buf[seqnum]; } - IncomingSplitPacket *sp = m_buf[seqnum]; - - // TODO: These errors should be thrown or something? Dunno. - if (chunk_count != sp->chunk_count) - LOG(derr_con<<"Connection: WARNING: chunk_count="<chunk_count="<chunk_count - <chunk_count) { + errorstream << "IncomingSplitBuffer::insert(): chunk_count=" + << chunk_count << " != sp->chunk_count=" << sp->chunk_count + << std::endl; + return SharedBuffer(); + } if (reliable != sp->reliable) LOG(derr_con<<"Connection: WARNING: reliable="<reliable="<reliable <chunks.find(chunk_num) != sp->chunks.end()) - return SharedBuffer(); - // Cut chunk data out of packet u32 chunkdatasize = p.data.getSize() - headersize; SharedBuffer chunkdata(chunkdatasize); memcpy(*chunkdata, &(p.data[headersize]), chunkdatasize); - // Set chunk data in buffer - sp->chunks[chunk_num] = chunkdata; + if (!sp->insert(chunk_num, chunkdata)) + return SharedBuffer(); // If not all chunks are received, return empty buffer - if (sp->allReceived() == false) + if (!sp->allReceived()) return SharedBuffer(); - // Calculate total size - u32 totalsize = 0; - for(std::map >::iterator i = sp->chunks.begin(); - i != sp->chunks.end(); ++i) - { - totalsize += i->second.getSize(); - } - - SharedBuffer fulldata(totalsize); - - // Copy chunks to data buffer - u32 start = 0; - for(u32 chunk_i=0; chunk_ichunk_count; - chunk_i++) - { - SharedBuffer buf = sp->chunks[chunk_i]; - u16 chunkdatasize = buf.getSize(); - memcpy(&fulldata[start], *buf, chunkdatasize); - start += chunkdatasize;; - } + SharedBuffer fulldata = sp->reassemble(); // Remove sp from buffer m_buf.erase(seqnum); @@ -545,66 +503,47 @@ SharedBuffer IncomingSplitBuffer::insert(BufferedPacket &p, bool reliable) return fulldata; } + void IncomingSplitBuffer::removeUnreliableTimedOuts(float dtime, float timeout) { - std::list remove_queue; + std::deque remove_queue; { MutexAutoLock listlock(m_map_mutex); - for(std::map::iterator i = m_buf.begin(); - i != m_buf.end(); ++i) - { - IncomingSplitPacket *p = i->second; + for (auto &i : m_buf) { + IncomingSplitPacket *p = i.second; // Reliable ones are not removed by timeout - if (p->reliable == true) + if (p->reliable) continue; p->time += dtime; if (p->time >= timeout) - remove_queue.push_back(i->first); + remove_queue.push_back(i.first); } } - for(std::list::iterator j = remove_queue.begin(); - j != remove_queue.end(); ++j) - { + for (u16 j : remove_queue) { MutexAutoLock listlock(m_map_mutex); LOG(dout_con<<"NOTE: Removing timed out unreliable split packet"<oldForgePacket(); + reliable = reliable_; } -Channel::~Channel() -{ -} +/* + Channel +*/ u16 Channel::readNextIncomingSeqNum() { @@ -691,7 +630,7 @@ void Channel::UpdateBytesSent(unsigned int bytes, unsigned int packets) { MutexAutoLock internal(m_internal_mutex); current_bytes_transfered += bytes; - current_packet_successfull += packets; + current_packet_successful += packets; } void Channel::UpdateBytesReceived(unsigned int bytes) { @@ -718,17 +657,16 @@ void Channel::UpdatePacketTooLateCounter() current_packet_too_late++; } -void Channel::UpdateTimers(float dtime,bool legacy_peer) +void Channel::UpdateTimers(float dtime) { bpm_counter += dtime; packet_loss_counter += dtime; - if (packet_loss_counter > 1.0) - { - packet_loss_counter -= 1.0; + if (packet_loss_counter > 1.0f) { + packet_loss_counter -= 1.0f; unsigned int packet_loss = 11; /* use a neutral value for initialization */ - unsigned int packets_successfull = 0; + unsigned int packets_successful = 0; //unsigned int packet_too_late = 0; bool reasonable_amount_of_data_transmitted = false; @@ -737,94 +675,78 @@ void Channel::UpdateTimers(float dtime,bool legacy_peer) MutexAutoLock internal(m_internal_mutex); packet_loss = current_packet_loss; //packet_too_late = current_packet_too_late; - packets_successfull = current_packet_successfull; + packets_successful = current_packet_successful; - if (current_bytes_transfered > (unsigned int) (window_size*512/2)) - { + if (current_bytes_transfered > (unsigned int) (window_size*512/2)) { reasonable_amount_of_data_transmitted = true; } current_packet_loss = 0; current_packet_too_late = 0; - current_packet_successfull = 0; - } - - /* dynamic window size is only available for non legacy peers */ - if (!legacy_peer) { - 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), + current_packet_successful = 0; + } + + /* dynamic window size */ + float successful_to_lost_ratio = 0.0f; + bool done = false; + + if (packets_successful > 0) { + successful_to_lost_ratio = packet_loss/packets_successful; + } else if (packet_loss > 0) { + window_size = std::max( + (window_size - 10), + MIN_RELIABLE_WINDOW_SIZE); + done = true; + } + + if (!done) { + if ((successful_to_lost_ratio < 0.01f) && + (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 = std::min( + (window_size + 100), + MAX_RELIABLE_WINDOW_SIZE); + } else if ((successful_to_lost_ratio < 0.05f) && + (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 = std::min( + (window_size + 50), + MAX_RELIABLE_WINDOW_SIZE); + } else if (successful_to_lost_ratio > 0.15f) { + window_size = std::max( + (window_size - 100), + MIN_RELIABLE_WINDOW_SIZE); + } else if (successful_to_lost_ratio > 0.1f) { + window_size = std::max( + (window_size - 50), MIN_RELIABLE_WINDOW_SIZE); - done = true; - } - - 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) - { + if (bpm_counter > 10.0f) { { MutexAutoLock internal(m_internal_mutex); cur_kbps = - (((float) current_bytes_transfered)/bpm_counter)/1024.0; + (((float) current_bytes_transfered)/bpm_counter)/1024.0f; current_bytes_transfered = 0; cur_kbps_lost = - (((float) current_bytes_lost)/bpm_counter)/1024.0; + (((float) current_bytes_lost)/bpm_counter)/1024.0f; current_bytes_lost = 0; cur_incoming_kbps = - (((float) current_bytes_received)/bpm_counter)/1024.0; + (((float) current_bytes_received)/bpm_counter)/1024.0f; current_bytes_received = 0; - bpm_counter = 0; + bpm_counter = 0.0f; } - if (cur_kbps > max_kbps) - { + if (cur_kbps > max_kbps) { max_kbps = cur_kbps; } - if (cur_kbps_lost > max_kbps_lost) - { + if (cur_kbps_lost > max_kbps_lost) { max_kbps_lost = cur_kbps_lost; } @@ -848,40 +770,26 @@ void Channel::UpdateTimers(float dtime,bool legacy_peer) Peer */ -PeerHelper::PeerHelper() : - m_peer(0) -{} - PeerHelper::PeerHelper(Peer* peer) : m_peer(peer) { - if (peer != NULL) - { - if (!peer->IncUseCount()) - { - m_peer = 0; - } - } + if (peer && !peer->IncUseCount()) + m_peer = nullptr; } PeerHelper::~PeerHelper() { - if (m_peer != 0) + if (m_peer) m_peer->DecUseCount(); - m_peer = 0; + m_peer = nullptr; } PeerHelper& PeerHelper::operator=(Peer* peer) { m_peer = peer; - if (peer != NULL) - { - if (!peer->IncUseCount()) - { - m_peer = 0; - } - } + if (peer && !peer->IncUseCount()) + m_peer = nullptr; return *this; } @@ -895,7 +803,8 @@ Peer* PeerHelper::operator&() const return m_peer; } -bool PeerHelper::operator!() { +bool PeerHelper::operator!() +{ return ! m_peer; } @@ -908,8 +817,7 @@ bool Peer::IncUseCount() { MutexAutoLock lock(m_exclusive_access_mutex); - if (!m_pending_deletion) - { + if (!m_pending_deletion) { this->m_usage++; return true; } @@ -930,7 +838,7 @@ void Peer::DecUseCount() delete this; } -void Peer::RTTStatistics(float rtt, std::string profiler_id, +void Peer::RTTStatistics(float rtt, const std::string &profiler_id, unsigned int num_samples) { if (m_last_rtt > 0) { @@ -969,10 +877,9 @@ void Peer::RTTStatistics(float rtt, std::string profiler_id, 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); + if (!profiler_id.empty()) { + g_profiler->graphAdd(profiler_id + " RTT [ms]", rtt * 1000.f); + g_profiler->graphAdd(profiler_id + " jitter [ms]", jitter * 1000.f); } } /* save values required for next loop */ @@ -982,7 +889,7 @@ void Peer::RTTStatistics(float rtt, std::string profiler_id, bool Peer::isTimedOut(float timeout) { MutexAutoLock lock(m_exclusive_access_mutex); - u32 current_time = porting::getTimeMs(); + u64 current_time = porting::getTimeMs(); float dtime = CALC_DTIME(m_last_timeout_check,current_time); m_last_timeout_check = current_time; @@ -1014,11 +921,10 @@ void Peer::Drop() } UDPPeer::UDPPeer(u16 a_id, Address a_address, Connection* connection) : - Peer(a_address,a_id,connection), - m_pending_disconnect(false), - resend_timeout(0.5), - m_legacy_peer(true) + Peer(a_address,a_id,connection) { + for (Channel &channel : channels) + channel.setWindowSize(START_RELIABLE_WINDOW_SIZE); } bool UDPPeer::getAddress(MTProtocols type,Address& toset) @@ -1032,15 +938,6 @@ bool UDPPeer::getAddress(MTProtocols type,Address& toset) return false; } -void UDPPeer::setNonLegacyPeer() -{ - m_legacy_peer = false; - for(unsigned int i=0; i< CHANNEL_COUNT; i++) - { - channels->setWindowSize(g_settings->getU16("max_packets_per_iteration")); - } -} - void UDPPeer::reportRTT(float rtt) { if (rtt < 0.0) { @@ -1064,7 +961,7 @@ bool UDPPeer::Ping(float dtime,SharedBuffer& data) if (m_ping_timer >= PING_TIMEOUT) { // Create and send PING packet - writeU8(&data[0], TYPE_CONTROL); + writeU8(&data[0], PACKET_TYPE_CONTROL); writeU8(&data[1], CONTROLTYPE_PING); m_ping_timer = 0.0; return true; @@ -1078,22 +975,29 @@ void UDPPeer::PutReliableSendCommand(ConnectionCommand &c, if (m_pending_disconnect) return; - if ( channels[c.channelnum].queued_commands.empty() && + Channel &chan = channels[c.channelnum]; + + if (chan.queued_commands.empty() && /* don't queue more packets then window size */ - (channels[c.channelnum].queued_reliables.size() - < (channels[c.channelnum].getWindowSize()/2))) { + (chan.queued_reliables.size() < chan.getWindowSize() / 2)) { LOG(dout_con<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); + chan.queued_commands.push_back(c); } } else { LOG(dout_con<getDesc() <<" Queueing reliable command for peer id: " << c.peer_id <<" data size: " << c.data.getSize() <= chan.getWindowSize() / 2) { + LOG(derr_con << m_connection->getDesc() + << "Possible packet stall to peer id: " << c.peer_id + << " queued_commands=" << chan.queued_commands.size() + << std::endl); + } } } @@ -1104,22 +1008,22 @@ bool UDPPeer::processReliableSendCommand( if (m_pending_disconnect) return true; + Channel &chan = channels[c.channelnum]; + u32 chunksize_max = max_packet_size - BASE_HEADER_SIZE - RELIABLE_HEADER_SIZE; sanity_check(c.data.getSize() < MAX_RELIABLE_WINDOW_SIZE*512); - std::list > originals; - u16 split_sequence_number = channels[c.channelnum].readNextSplitSeqNum(); + std::list> originals; + u16 split_sequence_number = chan.readNextSplitSeqNum(); - if (c.raw) - { - originals.push_back(c.data); - } - else { - originals = makeAutoSplitPacket(c.data, chunksize_max,split_sequence_number); - channels[c.channelnum].setNextSplitSeqNum(split_sequence_number); + if (c.raw) { + originals.emplace_back(c.data); + } else { + makeAutoSplitPacket(c.data, chunksize_max,split_sequence_number, &originals); + chan.setNextSplitSeqNum(split_sequence_number); } bool have_sequence_number = true; @@ -1127,10 +1031,8 @@ bool UDPPeer::processReliableSendCommand( std::queue toadd; volatile u16 initial_sequence_number = 0; - for(std::list >::iterator i = originals.begin(); - i != originals.end(); ++i) - { - u16 seqnum = channels[c.channelnum].getOutgoingSequenceNumber(have_sequence_number); + for (SharedBuffer &original : originals) { + u16 seqnum = chan.getOutgoingSequenceNumber(have_sequence_number); /* oops, we don't have enough sequence numbers to send this packet */ if (!have_sequence_number) @@ -1142,7 +1044,7 @@ bool UDPPeer::processReliableSendCommand( have_initial_sequence_number = true; } - SharedBuffer reliable = makeReliablePacket(*i, seqnum); + SharedBuffer reliable = makeReliablePacket(original, seqnum); // Add base headers and make a packet BufferedPacket p = con::makePacket(address, reliable, @@ -1154,7 +1056,7 @@ bool UDPPeer::processReliableSendCommand( if (have_sequence_number) { volatile u16 pcount = 0; - while(toadd.size() > 0) { + while (!toadd.empty()) { BufferedPacket p = toadd.front(); toadd.pop(); // LOG(dout_con<getDesc() @@ -1162,41 +1064,46 @@ bool UDPPeer::processReliableSendCommand( // << " channel: " << (c.channelnum&0xFF) // << " seqnum: " << readU16(&p.data[BASE_HEADER_SIZE+1]) // << std::endl) - channels[c.channelnum].queued_reliables.push(p); + chan.queued_reliables.push(p); pcount++; } - sanity_check(channels[c.channelnum].queued_reliables.size() < 0xFFFF); + sanity_check(chan.queued_reliables.size() < 0xFFFF); return true; } - else { - volatile u16 packets_available = toadd.size(); - /* we didn't get a single sequence number no need to fill queue */ - if (!have_initial_sequence_number) - { - return false; - } - while(toadd.size() > 0) { - /* remove packet */ - toadd.pop(); - bool successfully_put_back_sequence_number - = channels[c.channelnum].putBackSequenceNumber( - (initial_sequence_number+toadd.size() % (SEQNUM_MAX+1))); - - FATAL_ERROR_IF(!successfully_put_back_sequence_number, "error"); - } - LOG(dout_con<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); + volatile u16 packets_available = toadd.size(); + /* we didn't get a single sequence number no need to fill queue */ + if (!have_initial_sequence_number) { return false; } + + while (!toadd.empty()) { + /* remove packet */ + toadd.pop(); + + bool successfully_put_back_sequence_number + = chan.putBackSequenceNumber( + (initial_sequence_number+toadd.size() % (SEQNUM_MAX+1))); + + FATAL_ERROR_IF(!successfully_put_back_sequence_number, "error"); + } + + // DO NOT REMOVE n_queued! It avoids a deadlock of async locked + // 'log_message_mutex' and 'm_list_mutex'. + u32 n_queued = chan.outgoing_reliables_sent.size(); + + LOG(dout_con<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 : " + << n_queued + << std::endl); + + return false; } void UDPPeer::RunCommandQueues( @@ -1205,21 +1112,21 @@ void UDPPeer::RunCommandQueues( unsigned int maxtransfer) { - for (unsigned int i = 0; i < CHANNEL_COUNT; i++) { + for (Channel &channel : channels) { unsigned int commands_processed = 0; - if ((channels[i].queued_commands.size() > 0) && - (channels[i].queued_reliables.size() < maxtransfer) && + if ((!channel.queued_commands.empty()) && + (channel.queued_reliables.size() < maxtransfer) && (commands_processed < maxcommands)) { try { - ConnectionCommand c = channels[i].queued_commands.front(); + ConnectionCommand c = channel.queued_commands.front(); LOG(dout_con << m_connection->getDesc() << " processing queued reliable command " << std::endl); // Packet is processed, remove it from queue if (processReliableSendCommand(c,max_packet_size)) { - channels[i].queued_commands.pop_front(); + channel.queued_commands.pop_front(); } else { LOG(dout_con << m_connection->getDesc() << " Failed to queue packets for peer_id: " << c.peer_id @@ -1246,1570 +1153,124 @@ void UDPPeer::setNextSplitSequenceNumber(u8 channel, u16 seqnum) channels[channel].setNextSplitSeqNum(seqnum); } -SharedBuffer UDPPeer::addSpiltPacket(u8 channel, - BufferedPacket toadd, - bool reliable) +SharedBuffer UDPPeer::addSplitPacket(u8 channel, const BufferedPacket &toadd, + bool reliable) { assert(channel < CHANNEL_COUNT); // Pre-condition - return channels[channel].incoming_splits.insert(toadd,reliable); + return channels[channel].incoming_splits.insert(toadd, reliable); } -/******************************************************************************/ -/* Connection Threads */ -/******************************************************************************/ +/* + Connection +*/ -ConnectionSendThread::ConnectionSendThread(unsigned int max_packet_size, - float timeout) : - Thread("ConnectionSend"), - m_connection(NULL), - 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) -{ -} +Connection::Connection(u32 protocol_id, u32 max_packet_size, float timeout, + bool ipv6, PeerHandler *peerhandler) : + m_udpSocket(ipv6), + m_protocol_id(protocol_id), + m_sendThread(new ConnectionSendThread(max_packet_size, timeout)), + m_receiveThread(new ConnectionReceiveThread(max_packet_size)), + m_bc_peerhandler(peerhandler) -void * ConnectionSendThread::run() { - assert(m_connection); - - LOG(dout_con<getDesc() - <<"ConnectionSend thread started"<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); + /* Amount of time Receive() will wait for data, this is entirely different + * from the connection timeout */ + m_udpSocket.setTimeoutMs(500); - /* remove all triggers */ - while(m_send_sleep_semaphore.wait(0)) {} + m_sendThread->setParent(this); + m_receiveThread->setParent(this); - lasttime = curtime; - curtime = porting::getTimeMs(); - float dtime = CALC_DTIME(lasttime,curtime); + m_sendThread->start(); + m_receiveThread->start(); +} - /* 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); +Connection::~Connection() +{ + m_shutting_down = true; + // request threads to stop + m_sendThread->stop(); + m_receiveThread->stop(); - c = m_connection->m_command_queue.pop_frontNoEx(0); - } + //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); - /* send non reliable packets */ - sendPackets(dtime); + // wait for threads to finish + m_sendThread->wait(); + m_receiveThread->wait(); - END_DEBUG_EXCEPTION_HANDLER + // Delete peers + for (auto &peer : m_peers) { + delete peer.second; } +} - PROFILE(g_profiler->remove(ThreadIdentifier.str())); - return NULL; +/* Internal stuff */ +void Connection::putEvent(ConnectionEvent &e) +{ + assert(e.type != CONNEVENT_NONE); // Pre-condition + m_event_queue.push_back(e); } -void ConnectionSendThread::Trigger() +void Connection::TriggerSend() { - m_send_sleep_semaphore.post(); + m_sendThread->Trigger(); } -bool ConnectionSendThread::packetsQueued() +PeerHelper Connection::getPeerNoEx(session_t peer_id) { - std::list peerIds = m_connection->getPeerIDs(); + MutexAutoLock peerlock(m_peers_mutex); + std::map::iterator node = m_peers.find(peer_id); - if (!m_outgoing_queue.empty() && !peerIds.empty()) - return true; + if (node == m_peers.end()) { + return PeerHelper(NULL); + } - for(std::list::iterator j = peerIds.begin(); - j != peerIds.end(); ++j) - { - PeerHelper peer = m_connection->getPeerNoEx(*j); + // Error checking + FATAL_ERROR_IF(node->second->id != peer_id, "Invalid peer id"); - if (!peer) - continue; + return PeerHelper(node->second); +} - if (dynamic_cast(&peer) == 0) +/* find peer_id for address */ +u16 Connection::lookupPeer(Address& sender) +{ + MutexAutoLock peerlock(m_peers_mutex); + std::map::iterator j; + j = m_peers.begin(); + for(; j != m_peers.end(); ++j) + { + Peer *peer = j->second; + if (peer->isPendingDeletion()) continue; - for(u16 i=0; i < CHANNEL_COUNT; i++) { - Channel *channel = &(dynamic_cast(&peer))->channels[i]; + Address tocheck; - if (channel->queued_commands.size() > 0) { - return true; - } - } - } + if ((peer->getAddress(MTP_MINETEST_RELIABLE_UDP, tocheck)) && (tocheck == sender)) + return peer->id; + if ((peer->getAddress(MTP_UDP, tocheck)) && (tocheck == sender)) + return peer->id; + } - return false; + return PEER_ID_INEXISTENT; } -void ConnectionSendThread::runTimeouts(float dtime) +bool Connection::deletePeer(session_t peer_id, bool timeout) { - std::list timeouted_peers; - std::list peerIds = m_connection->getPeerIDs(); + Peer *peer = 0; - for(std::list::iterator j = peerIds.begin(); - j != peerIds.end(); ++j) - { - PeerHelper peer = m_connection->getPeerNoEx(*j); - - if (!peer) - continue; - - if (dynamic_cast(&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 data(2); // data for sending ping, required here because of goto - - /* - Check peer timeout - */ - if (peer->isTimedOut(m_timeout)) - { - infostream<getDesc() - <<"RunTimeouts(): Peer "<id - <<" has timed out." - <<" (source=peer->timeout_counter)" - <id); - // Don't bother going through the buffers of this one - continue; - } - - float resend_timeout = dynamic_cast(&peer)->getResendTimeout(); - bool retry_count_exceeded = false; - for(u16 i=0; i timed_outs; - Channel *channel = &(dynamic_cast(&peer))->channels[i]; - - if (dynamic_cast(&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_sent.incrementTimeouts(dtime); - - unsigned int numpeers = m_connection->m_peers.size(); - - if (numpeers == 0) - return; - - // Re-send timed out outgoing reliables - timed_outs = channel-> - outgoing_reliables_sent.getTimedOuts(resend_timeout, - (m_max_data_packets_per_iteration/numpeers)); - - channel->UpdatePacketLossCounter(timed_outs.size()); - g_profiler->graphAdd("packets_lost", timed_outs.size()); - - m_iteration_packets_avaialble -= timed_outs.size(); - - for(std::list::iterator k = timed_outs.begin(); - k != timed_outs.end(); ++k) - { - u16 peer_id = readPeerId(*(k->data)); - u8 channelnum = readChannel(*(k->data)); - u16 seqnum = readU16(&(k->data[BASE_HEADER_SIZE+1])); - - channel->UpdateBytesLost(k->data.getSize()); - k->resend_count++; - - if (k-> resend_count > MAX_RELIABLE_RETRY) { - retry_count_exceeded = true; - timeouted_peers.push_back(peer->id); - /* no need to check additional packets if a single one did timeout*/ - break; - } - - LOG(derr_con<getDesc() - <<"RE-SENDING timed-out RELIABLE to " - << k->address.serializeString() - << "(t/o="<(&peer)->id <(&peer)->id, 0, data, true)) - { - //retrigger with reduced ping interval - dynamic_cast(&peer)->Ping(4.0,data); - } - } - - dynamic_cast(&peer)->RunCommandQueues(m_max_packet_size, - m_max_commands_per_iteration, - m_max_packets_requeued); - } - - // Remove timed out peers - for(std::list::iterator i = timeouted_peers.begin(); - i != timeouted_peers.end(); ++i) - { - LOG(derr_con<getDesc() - <<"RunTimeouts(): Removing peer "<<(*i)<deletePeer(*i, true); - } -} - -void ConnectionSendThread::rawSend(const BufferedPacket &packet) -{ - try{ - m_connection->m_udpSocket.Send(packet.address, *packet.data, - packet.data.getSize()); - LOG(dout_con <getDesc() - << " rawSend: " << packet.data.getSize() - << " bytes sent" << std::endl); - } catch(SendFailedException &e) { - LOG(derr_con<getDesc() - <<"Connection::rawSend(): SendFailedException: " - <outgoing_reliables_sent.insert(p, - (channel->readOutgoingSequenceNumber() - MAX_RELIABLE_WINDOW_SIZE) - % (MAX_RELIABLE_WINDOW_SIZE+1)); - } - catch(AlreadyExistsException &e) - { - LOG(derr_con<getDesc() - <<"WARNING: Going to send a reliable packet" - <<" in outgoing buffer" < data, bool reliable) -{ - PeerHelper peer = m_connection->getPeerNoEx(peer_id); - if (!peer) { - LOG(dout_con<getDesc() - <<" INFO: dropped packet for non existent peer_id: " - << peer_id << std::endl); - FATAL_ERROR_IF(!reliable, "Trying to send raw packet reliable but no peer found!"); - return false; - } - Channel *channel = &(dynamic_cast(&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 reliable = makeReliablePacket(data, seqnum); - Address peer_address; - peer->getAddress(MTP_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<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<getDesc() - <<" INFO: queueing reliable packet for peer_id: " << peer_id - <<" channel: " << channelnum - <<" seqnum: " << seqnum << std::endl); - channel->queued_reliables.push(p); - return false; - } - } - else - { - Address peer_address; - - if (peer->getAddress(MTP_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<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); // Pre-condition - - switch(c.type) { - case CONNCMD_NONE: - LOG(dout_con<getDesc() - <<"UDP processing reliable CONNCMD_NONE"<getDesc() - <<"UDP processing reliable CONNCMD_SEND"<getDesc() - <<"UDP processing CONNCMD_SEND_TO_ALL"<getDesc() - <<"UDP processing reliable CONCMD_CREATE_PEER"<getDesc() - <<"UDP processing reliable CONCMD_DISABLE_LEGACY"<getDesc() - <<" Invalid reliable command type: " << c.type <getDesc() - <<" UDP processing CONNCMD_NONE"<getDesc() - <<" UDP processing CONNCMD_SERVE port=" - <getDesc() - <<" UDP processing CONNCMD_CONNECT"<getDesc() - <<" UDP processing CONNCMD_DISCONNECT"<getDesc() - <<" UDP processing CONNCMD_DISCONNECT_PEER"<getDesc() - <<" UDP processing CONNCMD_SEND"<getDesc() - <<" UDP processing CONNCMD_SEND_TO_ALL"<getDesc() - <<" UDP processing CONCMD_ACK"<getDesc() - <<" Invalid command type: " << c.type <getDesc() - <<"UDP serving at port " << bind_address.serializeString() <m_udpSocket.Bind(bind_address); - m_connection->SetPeerID(PEER_ID_SERVER); - } - catch(SocketException &e) { - // Create event - ConnectionEvent ce; - ce.bindFailed(); - m_connection->putEvent(ce); - } -} - -void ConnectionSendThread::connect(Address address) -{ - LOG(dout_con<getDesc()<<" connecting to "<createServerPeer(address); - - // Create event - ConnectionEvent e; - e.peerAdded(peer->id, peer->address); - m_connection->putEvent(e); - - Address bind_addr; - - if (address.isIPv6()) - bind_addr.setAddress((IPv6AddressBytes*) NULL); - else - bind_addr.setAddress(0,0,0,0); - - m_connection->m_udpSocket.Bind(bind_addr); - - // Send a dummy packet to server with peer_id = PEER_ID_INEXISTENT - m_connection->SetPeerID(PEER_ID_INEXISTENT); - NetworkPacket pkt(0,0); - m_connection->Send(PEER_ID_SERVER, 0, &pkt, true); -} - -void ConnectionSendThread::disconnect() -{ - LOG(dout_con<getDesc()<<" disconnecting"< data(2); - writeU8(&data[0], TYPE_CONTROL); - writeU8(&data[1], CONTROLTYPE_DISCO); - - - // Send to all - std::list peerids = m_connection->getPeerIDs(); - - for (std::list::iterator i = peerids.begin(); - i != peerids.end(); - ++i) - { - sendAsPacket(*i, 0,data,false); - } -} - -void ConnectionSendThread::disconnect_peer(u16 peer_id) -{ - LOG(dout_con<getDesc()<<" disconnecting peer"< data(2); - writeU8(&data[0], TYPE_CONTROL); - writeU8(&data[1], CONTROLTYPE_DISCO); - sendAsPacket(peer_id, 0,data,false); - - PeerHelper peer = m_connection->getPeerNoEx(peer_id); - - if (!peer) - return; - - if (dynamic_cast(&peer) == 0) - { - return; - } - - dynamic_cast(&peer)->m_pending_disconnect = true; -} - -void ConnectionSendThread::send(u16 peer_id, u8 channelnum, - SharedBuffer data) -{ - assert(channelnum < CHANNEL_COUNT); // Pre-condition - - PeerHelper peer = m_connection->getPeerNoEx(peer_id); - if (!peer) - { - LOG(dout_con<getDesc()<<" peer: peer_id="<>>NOT<<< found on sending packet" - << ", channel " << (channelnum % 0xFF) - << ", size: " << data.getSize() <getDesc()<<" sending to peer_id="<getNextSplitSequenceNumber(channelnum); - - u32 chunksize_max = m_max_packet_size - BASE_HEADER_SIZE; - std::list > originals; - - originals = makeAutoSplitPacket(data, chunksize_max,split_sequence_number); - - peer->setNextSplitSequenceNumber(channelnum,split_sequence_number); - - for(std::list >::iterator i = originals.begin(); - i != originals.end(); ++i) - { - SharedBuffer original = *i; - sendAsPacket(peer_id, channelnum, original); - } -} - -void ConnectionSendThread::sendReliable(ConnectionCommand &c) -{ - PeerHelper peer = m_connection->getPeerNoEx(c.peer_id); - if (!peer) - return; - - peer->PutReliableSendCommand(c,m_max_packet_size); -} - -void ConnectionSendThread::sendToAll(u8 channelnum, SharedBuffer data) -{ - std::list peerids = m_connection->getPeerIDs(); - - for (std::list::iterator i = peerids.begin(); - i != peerids.end(); - ++i) - { - send(*i, channelnum, data); - } -} - -void ConnectionSendThread::sendToAllReliable(ConnectionCommand &c) -{ - std::list peerids = m_connection->getPeerIDs(); - - for (std::list::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 peerIds = m_connection->getPeerIDs(); - std::list pendingDisconnect; - std::map pending_unreliable; - - for(std::list::iterator - j = peerIds.begin(); - j != peerIds.end(); ++j) - { - PeerHelper peer = m_connection->getPeerNoEx(*j); - //peer may have been removed - if (!peer) { - LOG(dout_con<getDesc()<< " Peer not found: peer_id=" << *j << std::endl); - continue; - } - peer->m_increment_packets_remaining = m_iteration_packets_avaialble/m_connection->m_peers.size(); - - if (dynamic_cast(&peer) == 0) - { - continue; - } - - if (dynamic_cast(&peer)->m_pending_disconnect) - { - pendingDisconnect.push_back(*j); - } - - PROFILE(std::stringstream peerIdentifier); - PROFILE(peerIdentifier << "sendPackets[" << m_connection->getDesc() << ";" << *j << ";RELIABLE]"); - PROFILE(ScopeProfiler peerprofiler(g_profiler, peerIdentifier.str(), SPT_AVG)); - - LOG(dout_con<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(&peer)->channels[i].outgoing_reliables_sent.getFirstSeqnum(next_to_ack); - u16 next_to_receive = 0; - dynamic_cast(&peer)->channels[i].incoming_reliables.getFirstSeqnum(next_to_receive); - - LOG(dout_con<getDesc()<< "\t channel: " - << i << ", peer quota:" - << peer->m_increment_packets_remaining - << std::endl - << "\t\t\treliables on wire: " - << dynamic_cast(&peer)->channels[i].outgoing_reliables_sent.size() - << ", waiting for ack for " << next_to_ack - << std::endl - << "\t\t\tincoming_reliables: " - << dynamic_cast(&peer)->channels[i].incoming_reliables.size() - << ", next reliable packet: " - << dynamic_cast(&peer)->channels[i].readNextIncomingSeqNum() - << ", next queued: " << next_to_receive - << std::endl - << "\t\t\treliables queued : " - << dynamic_cast(&peer)->channels[i].queued_reliables.size() - << std::endl - << "\t\t\tqueued commands : " - << dynamic_cast(&peer)->channels[i].queued_commands.size() - << std::endl); - - while ((dynamic_cast(&peer)->channels[i].queued_reliables.size() > 0) && - (dynamic_cast(&peer)->channels[i].outgoing_reliables_sent.size() - < dynamic_cast(&peer)->channels[i].getWindowSize())&& - (peer->m_increment_packets_remaining > 0)) - { - BufferedPacket p = dynamic_cast(&peer)->channels[i].queued_reliables.front(); - dynamic_cast(&peer)->channels[i].queued_reliables.pop(); - Channel* channel = &(dynamic_cast(&peer)->channels[i]); - LOG(dout_con<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--; - } - } - } - - if (m_outgoing_queue.size()) - { - LOG(dout_con<getDesc() - << " Handle non reliable queue (" - << m_outgoing_queue.size() << " pkts)" << std::endl); - } - - unsigned int initial_queuesize = m_outgoing_queue.size(); - /* send non reliable packets*/ - for(unsigned int i=0;i < initial_queuesize;i++) { - OutgoingPacket packet = m_outgoing_queue.front(); - m_outgoing_queue.pop(); - - if (packet.reliable) - continue; - - PeerHelper peer = m_connection->getPeerNoEx(packet.peer_id); - if (!peer) { - LOG(dout_con<getDesc() - <<" Outgoing queue: peer_id="<>>NOT<<< found on sending packet" - << ", channel " << (packet.channelnum % 0xFF) - << ", size: " << packet.data.getSize() <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(packet); - pending_unreliable[packet.peer_id] = true; - } - } - - for(std::list::iterator - k = pendingDisconnect.begin(); - k != pendingDisconnect.end(); ++k) - { - if (!pending_unreliable[*k]) - { - m_connection->deletePeer(*k,false); - } - } -} - -void ConnectionSendThread::sendAsPacket(u16 peer_id, u8 channelnum, - SharedBuffer data, bool ack) -{ - OutgoingPacket packet(peer_id, channelnum, data, false, ack); - m_outgoing_queue.push(packet); -} - -ConnectionReceiveThread::ConnectionReceiveThread(unsigned int max_packet_size) : - Thread("ConnectionReceive"), - m_connection(NULL) -{ -} - -void * ConnectionReceiveThread::run() -{ - assert(m_connection); - - LOG(dout_con<getDesc() - <<"ConnectionReceive thread started"<getDesc() << "]"); - -#ifdef DEBUG_CONNECTION_KBPS - u32 curtime = porting::getTimeMs(); - u32 lasttime = curtime; - float debug_print_timer = 0.0; -#endif - - while(!stopRequested()) { - BEGIN_DEBUG_EXCEPTION_HANDLER - PROFILE(ScopeProfiler sp(g_profiler, ThreadIdentifier.str(), SPT_AVG)); - -#ifdef DEBUG_CONNECTION_KBPS - lasttime = curtime; - curtime = porting::getTimeMs(); - float dtime = CALC_DTIME(lasttime,curtime); -#endif - - /* receive packets */ - receive(); - -#ifdef DEBUG_CONNECTION_KBPS - debug_print_timer += dtime; - if (debug_print_timer > 20.0) { - debug_print_timer -= 20.0; - - std::list peerids = m_connection->getPeerIDs(); - - for (std::list::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; jchannels[j].getCurrentDownloadRateKB(); - peer_loss += peer->channels[j].getCurrentLossRateKB(); - avg_rate += peer->channels[j].getAvgDownloadRateKB(); - avg_loss += peer->channels[j].getAvgLossRateKB(); - } - - std::stringstream output; - output << std::fixed << std::setprecision(1); - output << "OUT to Peer " << *i << " RATES (good / loss) " << std::endl; - output << "\tcurrent (sum): " << peer_current << "kb/s "<< peer_loss << "kb/s" << std::endl; - output << "\taverage (sum): " << avg_rate << "kb/s "<< avg_loss << "kb/s" << std::endl; - output << std::setfill(' '); - for(u16 j=0; jchannels[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 - } - - PROFILE(g_profiler->remove(ThreadIdentifier.str())); - return NULL; -} - -// Receive packets from the network and buffers and create ConnectionEvents -void ConnectionReceiveThread::receive() -{ - // use IPv6 minimum allowed MTU as receive buffer size as this is - // theoretical reliable upper boundary of a udp packet for all IPv6 enabled - // infrastructure - unsigned int packet_maxsize = 1500; - SharedBuffer 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))) { - loop_count++; - try { - if (packet_queued) { - bool data_left = true; - u16 peer_id; - SharedBuffer resultdata; - while(data_left) { - try { - data_left = getFromBuffers(peer_id, resultdata); - if (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 < BASE_HEADER_SIZE) || - (readU32(&packetdata[0]) != m_connection->GetProtocolID())) - { - LOG(derr_con<getDesc() - <<"Receive(): Invalid incoming packet, " - <<"size: " << received_size - <<", protocol: " - << ((received_size >= 4) ? readU32(&packetdata[0]) : -1) - << std::endl); - continue; - } - - u16 peer_id = readPeerId(*packetdata); - u8 channelnum = readChannel(*packetdata); - - if (channelnum > CHANNEL_COUNT-1) { - LOG(derr_con<getDesc() - <<"Receive(): Invalid channel "<lookupPeer(sender); - // We do not have to remind the peer of its - // peer id as the CONTROLTYPE_SET_PEER_ID - // command was sent reliably. - } - - /* The peer was not found in our lists. Add it. */ - if (peer_id == PEER_ID_INEXISTENT) { - peer_id = m_connection->createPeer(sender, MTP_MINETEST_RELIABLE_UDP, 0); - } - - PeerHelper peer = m_connection->getPeerNoEx(peer_id); - - if (!peer) { - LOG(dout_con<getDesc() - <<" got packet from unknown peer_id: " - <getAddress(MTP_UDP, peer_address)) { - if (peer_address != sender) { - LOG(derr_con<getDesc() - <getDesc() - <<" Peer "<getDesc() - <getDesc() - <<" Peer "<ResetTimeout(); - - Channel *channel = 0; - - if (dynamic_cast(&peer) != 0) - { - channel = &(dynamic_cast(&peer)->channels[channelnum]); - } - - if (channel != 0) { - channel->UpdateBytesReceived(received_size); - } - - // Throw the received packet to channel->processPacket() - - // Make a new SharedBuffer from the data without the base headers - SharedBuffer 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 resultdata = processPacket - (channel, strippeddata, peer_id, channelnum, false); - - LOG(dout_con<getDesc() - <<" ProcessPacket from peer_id: " << peer_id - << ",channel: " << (channelnum & 0xFF) << ", returned " - << resultdata.getSize() << " bytes" <putEvent(e); - } - catch(ProcessedSilentlyException &e) { - } - catch(ProcessedQueued &e) { - packet_queued = true; - } - } - catch(InvalidIncomingDataException &e) { - } - catch(ProcessedSilentlyException &e) { - } - } -} - -bool ConnectionReceiveThread::getFromBuffers(u16 &peer_id, SharedBuffer &dst) -{ - std::list peerids = m_connection->getPeerIDs(); - - for(std::list::iterator j = peerids.begin(); - j != peerids.end(); ++j) - { - PeerHelper peer = m_connection->getPeerNoEx(*j); - if (!peer) - continue; - - if (dynamic_cast(&peer) == 0) - continue; - - for(u16 i=0; i(&peer))->channels[i]; - - if (checkIncomingBuffers(channel, peer_id, dst)) { - return true; - } - } - } - return false; -} - -bool ConnectionReceiveThread::checkIncomingBuffers(Channel *channel, - u16 &peer_id, SharedBuffer &dst) -{ - u16 firstseqnum = 0; - if (channel->incoming_reliables.getFirstSeqnum(firstseqnum)) - { - 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]); - - LOG(dout_con<getDesc() - <<"UNBUFFERING TYPE_RELIABLE" - <<" seqnum="<= MAX_UDP_PEERS) { - std::string errmsg = "Invalid peer_id=" + itos(peer_id); - errorstream << errmsg << std::endl; - throw InvalidIncomingDataException(errmsg.c_str()); - } - - if (type == TYPE_CONTROL) - { - if (packetdata.getSize() < 2) - throw InvalidIncomingDataException("packetdata.getSize() < 2"); - - u8 controltype = readU8(&(packetdata[1])); - - if (controltype == CONTROLTYPE_ACK) - { - assert(channel != NULL); - - if (packetdata.getSize() < 4) { - throw InvalidIncomingDataException( - "packetdata.getSize() < 4 (ACK header size)"); - } - - u16 seqnum = readU16(&packetdata[2]); - LOG(dout_con<getDesc() - <<" [ CONTROLTYPE_ACK: channelnum=" - <<((int)channelnum&0xff)<<", peer_id="<outgoing_reliables_sent.popSeqnum(seqnum); - - // only calculate rtt from straight sent packets - if (p.resend_count == 0) { - // Get round trip time - unsigned int current_time = porting::getTimeMs(); - - // a overflow is quite unlikely but as it'd result in major - // rtt miscalculation we handle it here - if (current_time > p.absolute_send_time) - { - float rtt = (current_time - p.absolute_send_time) / 1000.0; - - // Let peer calculate stuff according to it - // (avg_rtt and resend_timeout) - dynamic_cast(&peer)->reportRTT(rtt); - } - else if (p.totaltime > 0) - { - float rtt = p.totaltime; - - // Let peer calculate stuff according to it - // (avg_rtt and resend_timeout) - dynamic_cast(&peer)->reportRTT(rtt); - } - } - //put bytes for max bandwidth calculation - channel->UpdateBytesSent(p.data.getSize(),1); - if (channel->outgoing_reliables_sent.size() == 0) - { - m_connection->TriggerSend(); - } - } - catch(NotFoundException &e) { - LOG(derr_con<getDesc() - <<"WARNING: ACKed packet not " - "in outgoing queue" - <UpdatePacketTooLateCounter(); - } - throw ProcessedSilentlyException("Got an ACK"); - } - else if (controltype == CONTROLTYPE_SET_PEER_ID) { - // 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]); - LOG(dout_con<getDesc() - <<"Got new peer id: "<GetPeerID() != PEER_ID_INEXISTENT) - { - LOG(derr_con<getDesc() - <<"WARNING: Not changing" - " existing peer id."<getDesc()<<"changing own peer id"<SetPeerID(peer_id_new); - } - - ConnectionCommand cmd; - - SharedBuffer 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) - { - // Just ignore it, the incoming data already reset - // the timeout counter - LOG(dout_con<getDesc()<<"PING"<getDesc() - <<"DISCO: Removing peer "<<(peer_id)<deletePeer(peer_id, false) == false) - { - derr_con<getDesc() - <<"DISCO: Peer not found"<(&peer)->setNonLegacyPeer(); - throw ProcessedSilentlyException("Got non legacy control"); - } - else{ - LOG(derr_con<getDesc() - <<"INVALID TYPE_CONTROL: invalid controltype=" - <<((int)controltype&0xff)<getDesc() - <<"RETURNING TYPE_ORIGINAL to user" - < payload(packetdata.getSize() - ORIGINAL_HEADER_SIZE); - memcpy(*payload, &(packetdata[ORIGINAL_HEADER_SIZE]), payload.getSize()); - return payload; - } - else if (type == TYPE_SPLIT) - { - Address peer_address; - - if (peer->getAddress(MTP_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 data = - peer->addSpiltPacket(channelnum,packet,reliable); - - if (data.getSize() != 0) - { - LOG(dout_con<getDesc() - <<"RETURNING TYPE_SPLIT: Constructed full data, " - <<"size="<getDesc()<<"BUFFERED TYPE_SPLIT"<readNextIncomingSeqNum(),MAX_RELIABLE_WINDOW_SIZE)) - { - 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<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 - // we don't know how long this packet was on wire don't do silly guessing - // dynamic_cast(&peer)->reportRTT(dynamic_cast(&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(MTP_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 = con::makePacket( - peer_address, - packetdata, - m_connection->GetProtocolID(), - peer_id, - channelnum); - try{ - channel->incoming_reliables.insert(packet,channel->readNextIncomingSeqNum()); - - LOG(dout_con<getDesc() - << "BUFFERING, TYPE_RELIABLE peer_id: " << peer_id - << ", channel: " << (channelnum&0xFF) - << ", seqnum: " << seqnum << std::endl;) - - throw ProcessedQueued("Buffered future reliable packet"); - } - catch(AlreadyExistsException &e) - { - } - catch(IncomingDataCorruption &e) - { - ConnectionCommand discon; - discon.disconnect_peer(peer_id); - m_connection->putCommand(discon); - - LOG(derr_con<getDesc() - << "INVALID, TYPE_RELIABLE peer_id: " << peer_id - << ", channel: " << (channelnum&0xFF) - << ", seqnum: " << seqnum - << "DROPPING CLIENT!" << std::endl;) - } - } - - /* we got a packet to process right now */ - LOG(dout_con<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)) - { - if (queued_seqnum == seqnum) - { - BufferedPacket queued_packet = channel->incoming_reliables.popFirst(); - /** TODO find a way to verify the new against the old packet */ - } - } - - channel->incNextIncomingSeqNum(); - - // Get out the inside packet and re-process it - SharedBuffer payload(packetdata.getSize() - RELIABLE_HEADER_SIZE); - memcpy(*payload, &packetdata[RELIABLE_HEADER_SIZE], payload.getSize()); - - return processPacket(channel, payload, peer_id, channelnum, true); - } - else - { - derr_con<getDesc() - <<"Got invalid type="<<((int)type&0xff)<::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); // Pre-condition - m_event_queue.push_back(e); -} - -PeerHelper Connection::getPeer(u16 peer_id) -{ - MutexAutoLock peerlock(m_peers_mutex); - std::map::iterator node = m_peers.find(peer_id); - - if (node == m_peers.end()) { - throw PeerNotFoundException("GetPeer: Peer not found (possible timeout)"); - } - - // Error checking - FATAL_ERROR_IF(node->second->id != peer_id, "Invalid peer id"); - - return PeerHelper(node->second); -} - -PeerHelper Connection::getPeerNoEx(u16 peer_id) -{ - MutexAutoLock peerlock(m_peers_mutex); - std::map::iterator node = m_peers.find(peer_id); - - if (node == m_peers.end()) { - return PeerHelper(NULL); - } - - // Error checking - FATAL_ERROR_IF(node->second->id != peer_id, "Invalid peer id"); - - return PeerHelper(node->second); -} - -/* find peer_id for address */ -u16 Connection::lookupPeer(Address& sender) -{ - MutexAutoLock peerlock(m_peers_mutex); - std::map::iterator j; - j = m_peers.begin(); - for(; j != m_peers.end(); ++j) - { - Peer *peer = j->second; - if (peer->isPendingDeletion()) - continue; - - Address tocheck; - - if ((peer->getAddress(MTP_MINETEST_RELIABLE_UDP, tocheck)) && (tocheck == sender)) - return peer->id; - - if ((peer->getAddress(MTP_UDP, tocheck)) && (tocheck == sender)) - return peer->id; - } - - return PEER_ID_INEXISTENT; -} - -std::list Connection::getPeers() -{ - std::list list; - for(std::map::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) -{ - Peer *peer = 0; - - /* lock list as short as possible */ + /* lock list as short as possible */ { MutexAutoLock 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); - m_peer_ids.remove(peer_id); + auto it = std::find(m_peer_ids.begin(), m_peer_ids.end(), peer_id); + m_peer_ids.erase(it); } Address peer_address; @@ -2842,7 +1303,7 @@ void Connection::putCommand(ConnectionCommand &c) { if (!m_shutting_down) { m_command_queue.push_back(c); - m_sendThread.Trigger(); + m_sendThread->Trigger(); } } @@ -2867,7 +1328,7 @@ bool Connection::Connected() if (m_peers.size() != 1) return false; - std::map::iterator node = m_peers.find(PEER_ID_SERVER); + std::map::iterator node = m_peers.find(PEER_ID_SERVER); if (node == m_peers.end()) return false; @@ -2884,16 +1345,21 @@ void Connection::Disconnect() putCommand(c); } -void Connection::Receive(NetworkPacket* pkt) +bool Connection::Receive(NetworkPacket *pkt, u32 timeout) { + /* + Note that this function can potentially wait infinitely if non-data + events keep happening before the timeout expires. + This is not considered to be a problem (is it?) + */ for(;;) { - ConnectionEvent e = waitEvent(m_bc_receive_timeout); + ConnectionEvent e = waitEvent(timeout); if (e.type != CONNEVENT_NONE) LOG(dout_con << getDesc() << ": Receive: got event: " << e.describe() << std::endl); switch(e.type) { case CONNEVENT_NONE: - throw NoIncomingDataException("No incoming data"); + return false; case CONNEVENT_DATA_RECEIVED: // Data size is lesser than command size, ignoring packet if (e.data.getSize() < 2) { @@ -2901,7 +1367,7 @@ void Connection::Receive(NetworkPacket* pkt) } pkt->putRawPacket(*e.data, e.data.getSize(), e.peer_id); - return; + return true; case CONNEVENT_PEER_ADDED: { UDPPeer tmp(e.peer_id, e.address, this); if (m_bc_peerhandler) @@ -2919,11 +1385,23 @@ void Connection::Receive(NetworkPacket* pkt) "(port already in use?)"); } } - throw NoIncomingDataException("No incoming data"); + return false; +} + +void Connection::Receive(NetworkPacket *pkt) +{ + bool any = Receive(pkt, m_bc_receive_timeout); + if (!any) + throw NoIncomingDataException("No incoming data"); +} + +bool Connection::TryReceive(NetworkPacket *pkt) +{ + return Receive(pkt, 0); } -void Connection::Send(u16 peer_id, u8 channelnum, - NetworkPacket* pkt, bool reliable) +void Connection::Send(session_t peer_id, u8 channelnum, + NetworkPacket *pkt, bool reliable) { assert(channelnum < CHANNEL_COUNT); // Pre-condition @@ -2933,7 +1411,7 @@ void Connection::Send(u16 peer_id, u8 channelnum, putCommand(c); } -Address Connection::GetPeerAddress(u16 peer_id) +Address Connection::GetPeerAddress(session_t peer_id) { PeerHelper peer = getPeerNoEx(peer_id); @@ -2944,7 +1422,7 @@ Address Connection::GetPeerAddress(u16 peer_id) return peer_address; } -float Connection::getPeerStat(u16 peer_id, rtt_stat_type type) +float Connection::getPeerStat(session_t peer_id, rtt_stat_type type) { PeerHelper peer = getPeerNoEx(peer_id); if (!peer) return -1; @@ -2959,25 +1437,25 @@ float Connection::getLocalStat(rate_stat_type type) float retval = 0.0; - for (u16 j=0; j(&peer)->channels) { switch(type) { case CUR_DL_RATE: - retval += dynamic_cast(&peer)->channels[j].getCurrentDownloadRateKB(); + retval += channel.getCurrentDownloadRateKB(); break; case AVG_DL_RATE: - retval += dynamic_cast(&peer)->channels[j].getAvgDownloadRateKB(); + retval += channel.getAvgDownloadRateKB(); break; case CUR_INC_RATE: - retval += dynamic_cast(&peer)->channels[j].getCurrentIncomingRateKB(); + retval += channel.getCurrentIncomingRateKB(); break; case AVG_INC_RATE: - retval += dynamic_cast(&peer)->channels[j].getAvgIncomingRateKB(); + retval += channel.getAvgIncomingRateKB(); break; case AVG_LOSS_RATE: - retval += dynamic_cast(&peer)->channels[j].getAvgLossRateKB(); + retval += channel.getAvgLossRateKB(); break; case CUR_LOSS_RATE: - retval += dynamic_cast(&peer)->channels[j].getCurrentLossRateKB(); + retval += channel.getCurrentLossRateKB(); break; default: FATAL_ERROR("Connection::getLocalStat Invalid stat type"); @@ -2991,7 +1469,7 @@ 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 = m_next_remote_peer_id; + session_t peer_id_new = m_next_remote_peer_id; u16 overflow = MAX_UDP_PEERS; /* @@ -3031,7 +1509,7 @@ u16 Connection::createPeer(Address& sender, MTProtocols protocol, int fd) ConnectionCommand cmd; SharedBuffer reply(4); - writeU8(&reply[0], TYPE_CONTROL); + writeU8(&reply[0], PACKET_TYPE_CONTROL); writeU8(&reply[1], CONTROLTYPE_SET_PEER_ID); writeU16(&reply[2], peer_id_new); cmd.createPeer(peer_id_new,reply); @@ -3053,25 +1531,20 @@ void Connection::PrintInfo(std::ostream &out) m_info_mutex.unlock(); } -void Connection::PrintInfo() -{ - PrintInfo(dout_con); -} - const std::string Connection::getDesc() { return std::string("con(")+ itos(m_udpSocket.GetHandle())+"/"+itos(m_peer_id)+")"; } -void Connection::DisconnectPeer(u16 peer_id) +void Connection::DisconnectPeer(session_t peer_id) { ConnectionCommand discon; discon.disconnect_peer(peer_id); putCommand(discon); } -void Connection::sendAck(u16 peer_id, u8 channelnum, u16 seqnum) +void Connection::sendAck(session_t peer_id, u8 channelnum, u16 seqnum) { assert(channelnum < CHANNEL_COUNT); // Pre-condition @@ -3082,18 +1555,18 @@ void Connection::sendAck(u16 peer_id, u8 channelnum, u16 seqnum) ConnectionCommand c; SharedBuffer ack(4); - writeU8(&ack[0], TYPE_CONTROL); + writeU8(&ack[0], PACKET_TYPE_CONTROL); writeU8(&ack[1], CONTROLTYPE_ACK); writeU16(&ack[2], seqnum); c.ack(peer_id, channelnum, ack); putCommand(c); - m_sendThread.Trigger(); + m_sendThread->Trigger(); } UDPPeer* Connection::createServerPeer(Address& address) { - if (getPeerNoEx(PEER_ID_SERVER) != 0) + if (ConnectedToServer()) { throw ConnectionException("Already connected to a server"); }