u32 packet_size = data.getSize() + header_size;
SharedBuffer<u8> b(packet_size);
- writeU8(&b[0], TYPE_ORIGINAL);
-
- memcpy(&b[header_size], *data, data.getSize());
-
+ writeU8(&(b[0]), TYPE_ORIGINAL);
+ if (data.getSize() > 0) {
+ memcpy(&(b[header_size]), *data, data.getSize());
+ }
return b;
}
{
// Chunk packets, containing the TYPE_SPLIT header
std::list<SharedBuffer<u8> > chunks;
-
+
u32 chunk_header_size = 7;
u32 maximum_data_size = chunksize_max - chunk_header_size;
u32 start = 0;
end = start + maximum_data_size - 1;
if(end > data.getSize() - 1)
end = data.getSize() - 1;
-
+
u32 payload_size = end - start + 1;
u32 packet_size = chunk_header_size + payload_size;
SharedBuffer<u8> chunk(packet_size);
-
+
writeU8(&chunk[0], TYPE_SPLIT);
writeU16(&chunk[1], seqnum);
// [3] u16 chunk_count is written at next stage
chunks.push_back(chunk);
chunk_count++;
-
+
start = end + 1;
chunk_num++;
}
SharedBuffer<u8> data,
u16 seqnum)
{
- /*dstream<<"BEGIN SharedBuffer<u8> makeReliablePacket()"<<std::endl;
- dstream<<"data.getSize()="<<data.getSize()<<", data[0]="
- <<((unsigned int)data[0]&0xff)<<std::endl;*/
u32 header_size = 3;
u32 packet_size = data.getSize() + header_size;
SharedBuffer<u8> b(packet_size);
memcpy(&b[header_size], *data, data.getSize());
- /*dstream<<"data.getSize()="<<data.getSize()<<", data[0]="
- <<((unsigned int)data[0]&0xff)<<std::endl;*/
- //dstream<<"END SharedBuffer<u8> makeReliablePacket()"<<std::endl;
return b;
}
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]); }
+ 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()){
- LOG(dout_con<<"Sequence number: " << seqnum << " not found in reliable buffer"<<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;
)
{
/* if this happens your maximum transfer window may be to big */
- fprintf(stderr, "Duplicated seqnum %d non matching packet detected:\n",seqnum);
+ 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());
+ 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());
+ 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");
}
sp->reliable = reliable;
m_buf[seqnum] = sp;
}
-
+
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
// lag and the server re-sends stuff.
if(sp->chunks.find(chunk_num) != sp->chunks.end())
return SharedBuffer<u8>();
-
+
// Cut chunk data out of packet
u32 chunkdatasize = p.data.getSize() - headersize;
SharedBuffer<u8> chunkdata(chunkdatasize);
memcpy(*chunkdata, &(p.data[headersize]), chunkdatasize);
-
+
// Set chunk data in buffer
sp->chunks[chunk_num] = chunkdata;
-
+
// If not all chunks are received, return empty buffer
if(sp->allReceived() == false)
return SharedBuffer<u8>();
{
totalsize += i->second.getSize();
}
-
+
SharedBuffer<u8> fulldata(totalsize);
// Copy chunks to data buffer
next_outgoing_split_seqnum(SEQNUM_INITIAL),
current_packet_loss(0),
current_packet_too_late(0),
+ current_packet_successfull(0),
packet_loss_counter(0),
current_bytes_transfered(0),
+ current_bytes_received(0),
current_bytes_lost(0),
max_kbps(0.0),
cur_kbps(0.0),
avg_kbps(0.0),
+ max_incoming_kbps(0.0),
+ cur_incoming_kbps(0.0),
+ avg_incoming_kbps(0.0),
max_kbps_lost(0.0),
cur_kbps_lost(0.0),
avg_kbps_lost(0.0),
- bpm_counter(0.0)
+ bpm_counter(0.0),
+ rate_samples(0)
{
}
current_packet_successfull += packets;
}
+void Channel::UpdateBytesReceived(unsigned int bytes) {
+ JMutexAutoLock internal(m_internal_mutex);
+ current_bytes_received += bytes;
+}
+
void Channel::UpdateBytesLost(unsigned int bytes)
{
JMutexAutoLock internal(m_internal_mutex);
current_packet_too_late++;
}
-void Channel::UpdateTimers(float dtime)
+void Channel::UpdateTimers(float dtime,bool legacy_peer)
{
bpm_counter += dtime;
packet_loss_counter += dtime;
packet_too_late = current_packet_too_late;
packets_successfull = current_packet_successfull;
- if (current_bytes_transfered > (window_size*512/2))
+ if (current_bytes_transfered > (unsigned int) (window_size*512/2))
{
reasonable_amount_of_data_transmitted = true;
}
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;
- }
+ /* dynamic window size is only available for non legacy peers */
+ if (!legacy_peer) {
+ float successfull_to_lost_ratio = 0.0;
+ bool done = false;
- 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);
+ if (packets_successfull > 0) {
+ successfull_to_lost_ratio = packet_loss/packets_successfull;
}
- 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)
+ else if (packet_loss > 0)
{
window_size = MYMAX(
- (window_size - 100),
- MIN_RELIABLE_WINDOW_SIZE);
+ (window_size - 10),
+ MIN_RELIABLE_WINDOW_SIZE);
+ done = true;
}
- else if (successfull_to_lost_ratio > 0.1)
+
+ if (!done)
{
- window_size = MYMAX(
- (window_size - 50),
- MIN_RELIABLE_WINDOW_SIZE);
+ 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);
+ }
}
}
}
{
{
JMutexAutoLock internal(m_internal_mutex);
- cur_kbps = (current_bytes_transfered/bpm_counter)/1024;
+ cur_kbps =
+ (((float) current_bytes_transfered)/bpm_counter)/1024.0;
current_bytes_transfered = 0;
- cur_kbps_lost = (current_bytes_lost/bpm_counter)/1024;
- current_bytes_lost = 0;
- bpm_counter = 0;
+ cur_kbps_lost =
+ (((float) current_bytes_lost)/bpm_counter)/1024.0;
+ current_bytes_lost = 0;
+ cur_incoming_kbps =
+ (((float) current_bytes_received)/bpm_counter)/1024.0;
+ current_bytes_received = 0;
+ bpm_counter = 0;
}
if (cur_kbps > max_kbps)
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;
+ if (cur_incoming_kbps > max_incoming_kbps) {
+ max_incoming_kbps = cur_incoming_kbps;
+ }
+
+ rate_samples = MYMIN(rate_samples+1,10);
+ float old_fraction = ((float) (rate_samples-1) )/( (float) rate_samples);
+ avg_kbps = avg_kbps * old_fraction +
+ cur_kbps * (1.0 - old_fraction);
+ avg_kbps_lost = avg_kbps_lost * old_fraction +
+ cur_kbps_lost * (1.0 - old_fraction);
+ avg_incoming_kbps = avg_incoming_kbps * old_fraction +
+ cur_incoming_kbps * (1.0 - old_fraction);
}
}
PeerHelper::PeerHelper() :
m_peer(0)
-{
-}
+{}
PeerHelper::PeerHelper(Peer* peer) :
m_peer(peer)
delete this;
}
-void Peer::RTTStatistics(float rtt,
- std::string profiler_id,
- unsigned int num_samples) {
+void Peer::RTTStatistics(float rtt, std::string profiler_id,
+ unsigned int num_samples) {
if (m_last_rtt > 0) {
/* set min max values */
}
PROFILE(std::stringstream peerIdentifier1);
- PROFILE(peerIdentifier1 << "runTimeouts[" << m_connection->getDesc() << ";" << id << ";RELIABLE]");
+ 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(peerIdentifier2 << "sendPackets[" << m_connection->getDesc()
+ << ";" << id << ";RELIABLE]");
PROFILE(ScopeProfiler peerprofiler(g_profiler, peerIdentifier2.str(), SPT_AVG));
delete this;
bool UDPPeer::getAddress(MTProtocols type,Address& toset)
{
- if ((type == UDP) || (type == MINETEST_RELIABLE_UDP) || (type == PRIMARY))
+ if ((type == MTP_UDP) || (type == MTP_MINETEST_RELIABLE_UDP) || (type == MTP_PRIMARY))
{
toset = address;
return true;
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()
+ << " 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;
}
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);
+ << ", delaying sending of " << c.data.getSize()
+ << " bytes" << std::endl);
channels[i].queued_commands.push_front(c);
}
}
/* send non reliable packets */
sendPackets(dtime);
- END_DEBUG_EXCEPTION_HANDLER(derr_con);
+ END_DEBUG_EXCEPTION_HANDLER(errorstream);
}
PROFILE(g_profiler->remove(ThreadIdentifier.str()));
continue;
PROFILE(std::stringstream peerIdentifier);
- PROFILE(peerIdentifier << "runTimeouts[" << m_connection->getDesc() << ";" << *j << ";RELIABLE]");
+ 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
*/
if(peer->isTimedOut(m_timeout))
{
- LOG(derr_con<<m_connection->getDesc()
+ infostream<<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
(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<BufferedPacket>::iterator j = timed_outs.begin();
- j != timed_outs.end(); ++j)
+ for(std::list<BufferedPacket>::iterator k = timed_outs.begin();
+ k != timed_outs.end(); ++k)
{
- u16 peer_id = readPeerId(*(j->data));
- u8 channelnum = readChannel(*(j->data));
- u16 seqnum = readU16(&(j->data[BASE_HEADER_SIZE+1]));
+ u16 peer_id = readPeerId(*(k->data));
+ u8 channelnum = readChannel(*(k->data));
+ u16 seqnum = readU16(&(k->data[BASE_HEADER_SIZE+1]));
- channel->UpdateBytesLost(j->data.getSize());
+ channel->UpdateBytesLost(k->data.getSize());
+ k->resend_count++;
LOG(derr_con<<m_connection->getDesc()
<<"RE-SENDING timed-out RELIABLE to "
- << j->address.serializeString()
+ << k->address.serializeString()
<< "(t/o="<<resend_timeout<<"): "
<<"from_peer_id="<<peer_id
<<", channel="<<((int)channelnum&0xff)
<<", seqnum="<<seqnum
<<std::endl);
- rawSend(*j);
+ rawSend(*k);
// 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);
+ channel->UpdateTimers(dtime,dynamic_cast<UDPPeer*>(&peer)->getLegacyPeer());
}
/* send ping if necessary */
void ConnectionSendThread::rawSend(const BufferedPacket &packet)
{
try{
- m_connection->m_udpSocket.Send(packet.address, *packet.data, packet.data.getSize());
+ 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);
+ << " rawSend: " << packet.data.getSize()
+ << " bytes sent" << std::endl);
} catch(SendFailedException &e){
LOG(derr_con<<m_connection->getDesc()
<<"Connection::rawSend(): SendFailedException: "
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));
+ (channel->readOutgoingSequenceNumber() - MAX_RELIABLE_WINDOW_SIZE)
+ % (MAX_RELIABLE_WINDOW_SIZE+1));
}
catch(AlreadyExistsException &e)
{
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);
+ <<" 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;
}
if(reliable)
{
bool have_sequence_number_for_raw_packet = true;
- u16 seqnum = channel->getOutgoingSequenceNumber(have_sequence_number_for_raw_packet);
+ 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);
+ peer->getAddress(MTP_MINETEST_RELIABLE_UDP, peer_address);
// Add base headers and make a packet
BufferedPacket p = con::makePacket(peer_address, reliable,
{
Address peer_address;
- if (peer->getAddress(UDP,peer_address))
+ if (peer->getAddress(MTP_UDP, peer_address))
{
// Add base headers and make a packet
BufferedPacket p = con::makePacket(peer_address, data,
switch(c.type){
case CONNCMD_NONE:
- LOG(dout_con<<m_connection->getDesc()<<"UDP processing reliable CONNCMD_NONE"<<std::endl);
+ 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);
+ 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);
+ 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);
+ 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 */
return;
case CONCMD_DISABLE_LEGACY:
- LOG(dout_con<<m_connection->getDesc()<<"UDP processing reliable CONCMD_DISABLE_LEGACY"<<std::endl);
+ 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 */
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);
+ LOG(dout_con<<m_connection->getDesc()
+ <<" Invalid reliable command type: " << c.type <<std::endl);
}
}
switch(c.type){
case CONNCMD_NONE:
- LOG(dout_con<<m_connection->getDesc()<<" UDP processing CONNCMD_NONE"<<std::endl);
+ 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="
+ LOG(dout_con<<m_connection->getDesc()
+ <<" UDP processing CONNCMD_SERVE port="
<<c.address.serializeString()<<std::endl);
serve(c.address);
return;
case CONNCMD_CONNECT:
- LOG(dout_con<<m_connection->getDesc()<<" UDP processing CONNCMD_CONNECT"<<std::endl);
+ 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);
+ 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);
+ 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);
+ 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);
+ LOG(dout_con<<m_connection->getDesc()
+ <<" UDP processing CONNCMD_SEND_TO_ALL"<<std::endl);
sendToAll(c.channelnum, c.data);
return;
case CONCMD_ACK:
- LOG(dout_con<<m_connection->getDesc()<<" UDP processing CONCMD_ACK"<<std::endl);
+ 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);
+ LOG(dout_con<<m_connection->getDesc()
+ <<" Invalid command type: " << c.type <<std::endl);
}
}
}
ConnectionReceiveThread::ConnectionReceiveThread(Connection* parent,
- unsigned int max_packet_size) :
+ unsigned int max_packet_size) :
m_connection(parent)
{
}
}
}
#endif
- END_DEBUG_EXCEPTION_HANDLER(derr_con);
+ END_DEBUG_EXCEPTION_HANDLER(errorstream);
}
PROFILE(g_profiler->remove(ThreadIdentifier.str()));
return NULL;
// infrastructure
unsigned int packet_maxsize = 1500;
SharedBuffer<u8> packetdata(packet_maxsize);
-
+
bool packet_queued = true;
unsigned int loop_count = 0;
LOG(derr_con<<m_connection->getDesc()
<<"Receive(): Invalid incoming packet, "
<<"size: " << received_size
- <<", protocol: " << readU32(&packetdata[0]) <<std::endl);
+ <<", 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<<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;
/* 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);
+ peer_id = m_connection->createPeer(sender, MTP_MINETEST_RELIABLE_UDP, 0);
}
PeerHelper peer = m_connection->getPeerNoEx(peer_id);
Address peer_address;
- if (peer->getAddress(UDP,peer_address)) {
+ if (peer->getAddress(MTP_UDP, peer_address)) {
if (peer_address != sender) {
LOG(derr_con<<m_connection->getDesc()
<<m_connection->getDesc()
}
}
-
+
/* mark peer as seen with id */
if (!(packet_peer_id == PEER_ID_INEXISTENT))
peer->setSentWithID();
{
channel = &(dynamic_cast<UDPPeer*>(&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<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 & 0xFF) << ", returned "
<< resultdata.getSize() << " bytes" <<std::endl);
-
+
ConnectionEvent e;
e.dataReceived(peer_id, resultdata);
m_connection->putEvent(e);
return false;
}
-bool ConnectionReceiveThread::checkIncomingBuffers(Channel *channel, u16 &peer_id,
- SharedBuffer<u8> &dst)
+bool ConnectionReceiveThread::checkIncomingBuffers(Channel *channel,
+ u16 &peer_id, SharedBuffer<u8> &dst)
{
u16 firstseqnum = 0;
if (channel->incoming_reliables.getFirstSeqnum(firstseqnum))
}
SharedBuffer<u8> ConnectionReceiveThread::processPacket(Channel *channel,
- SharedBuffer<u8> packetdata, u16 peer_id,
- u8 channelnum, bool reliable)
+ SharedBuffer<u8> packetdata, u16 peer_id, u8 channelnum, bool reliable)
{
- PeerHelper peer = m_connection->getPeer(peer_id);
+ PeerHelper peer = m_connection->getPeerNoEx(peer_id);
+
+ if (!peer) {
+ errorstream << "Peer not found (possible timeout)" << std::endl;
+ throw ProcessedSilentlyException("Peer not found (possible timeout)");
+ }
if(packetdata.getSize() < 1)
throw InvalidIncomingDataException("packetdata.getSize() < 1");
- u8 type = readU8(&packetdata[0]);
+ u8 type = readU8(&(packetdata[0]));
if(type == TYPE_CONTROL)
{
if(packetdata.getSize() < 2)
throw InvalidIncomingDataException("packetdata.getSize() < 2");
- u8 controltype = readU8(&packetdata[1]);
+ u8 controltype = readU8(&(packetdata[1]));
if( (controltype == CONTROLTYPE_ACK)
&& (peer_id <= MAX_UDP_PEERS))
try{
BufferedPacket p =
channel->outgoing_reliables_sent.popSeqnum(seqnum);
- // Get round trip time
- unsigned int current_time = porting::getTimeMs();
- if (current_time > p.absolute_send_time)
- {
- float rtt = (current_time - p.absolute_send_time) / 1000.0;
+ // only calculate rtt from straight sent packets
+ if (p.resend_count == 0) {
+ // Get round trip time
+ unsigned int current_time = porting::getTimeMs();
- // 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;
+ // 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<UDPPeer*>(&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<UDPPeer*>(&peer)->reportRTT(rtt);
+ // 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);
}
else if(type == TYPE_ORIGINAL)
{
- if(packetdata.getSize() < ORIGINAL_HEADER_SIZE)
+ if(packetdata.getSize() <= ORIGINAL_HEADER_SIZE)
throw InvalidIncomingDataException
- ("packetdata.getSize() < ORIGINAL_HEADER_SIZE");
+ ("packetdata.getSize() <= ORIGINAL_HEADER_SIZE");
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());
+ memcpy(*payload, &(packetdata[ORIGINAL_HEADER_SIZE]), payload.getSize());
return payload;
}
else if(type == TYPE_SPLIT)
{
Address peer_address;
- if (peer->getAddress(UDP,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.
// 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());
+ // we don't know how long this packet was on wire don't do silly guessing
+ // dynamic_cast<UDPPeer*>(&peer)->reportRTT(dynamic_cast<UDPPeer*>(&peer)->getResendTimeout());
throw ProcessedSilentlyException("Retransmitting ack for old packet");
}
Address peer_address;
// this is a reliable packet so we have a udp address for sure
- peer->getAddress(MINETEST_RELIABLE_UDP,peer_address);
+ 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.
*/
Connection::Connection(u32 protocol_id, u32 max_packet_size, float timeout,
- bool ipv6):
+ bool ipv6) :
m_udpSocket(ipv6),
m_command_queue(),
m_event_queue(),
}
Connection::Connection(u32 protocol_id, u32 max_packet_size, float timeout,
- bool ipv6, PeerHandler *peerhandler):
+ bool ipv6, PeerHandler *peerhandler) :
m_udpSocket(ipv6),
m_command_queue(),
m_event_queue(),
Address tocheck;
- if ((peer->getAddress(MINETEST_RELIABLE_UDP,tocheck)) && (tocheck == sender))
+ if ((peer->getAddress(MTP_MINETEST_RELIABLE_UDP, tocheck)) && (tocheck == sender))
return peer->id;
- if ((peer->getAddress(UDP,tocheck)) && (tocheck == sender))
+ if ((peer->getAddress(MTP_UDP, tocheck)) && (tocheck == sender))
return peer->id;
}
Address peer_address;
//any peer has a primary address this never fails!
- peer->getAddress(PRIMARY,peer_address);
+ peer->getAddress(MTP_PRIMARY, peer_address);
// Create event
ConnectionEvent e;
e.peerRemoved(peer_id, timeout, peer_address);
if(m_peers.size() != 1)
return false;
-
+
std::map<u16, Peer*>::iterator node = m_peers.find(PEER_ID_SERVER);
if(node == m_peers.end())
return false;
-
+
if(m_peer_id == PEER_ID_INEXISTENT)
return false;
if (!peer)
throw PeerNotFoundException("No address for peer found!");
Address peer_address;
- peer->getAddress(PRIMARY,peer_address);
+ peer->getAddress(MTP_PRIMARY, peer_address);
return peer_address;
}
return peer->getStat(type);
}
+float Connection::getLocalStat(rate_stat_type type)
+{
+ PeerHelper peer = getPeerNoEx(PEER_ID_SERVER);
+
+ if (!peer) {
+ assert("Connection::getLocalStat we couldn't get our own peer? are you serious???" == 0);
+ }
+
+ float retval = 0.0;
+
+ for (u16 j=0; j<CHANNEL_COUNT; j++) {
+ switch(type) {
+ case CUR_DL_RATE:
+ retval += dynamic_cast<UDPPeer*>(&peer)->channels[j].getCurrentDownloadRateKB();
+ break;
+ case AVG_DL_RATE:
+ retval += dynamic_cast<UDPPeer*>(&peer)->channels[j].getAvgDownloadRateKB();
+ break;
+ case CUR_INC_RATE:
+ retval += dynamic_cast<UDPPeer*>(&peer)->channels[j].getCurrentIncomingRateKB();
+ break;
+ case AVG_INC_RATE:
+ retval += dynamic_cast<UDPPeer*>(&peer)->channels[j].getAvgIncomingRateKB();
+ break;
+ case AVG_LOSS_RATE:
+ retval += dynamic_cast<UDPPeer*>(&peer)->channels[j].getAvgLossRateKB();
+ break;
+ case CUR_LOSS_RATE:
+ retval += dynamic_cast<UDPPeer*>(&peer)->channels[j].getCurrentLossRateKB();
+ break;
+ default:
+ assert("Connection::getLocalStat Invalid stat type" == 0);
+ }
+ }
+ return retval;
+}
+
u16 Connection::createPeer(Address& sender, MTProtocols protocol, int fd)
{
// Somebody wants to make a new connection
const std::string Connection::getDesc()
{
- return std::string("con(")+itos(m_udpSocket.GetHandle())+"/"+itos(m_peer_id)+")";
+ return std::string("con(")+
+ itos(m_udpSocket.GetHandle())+"/"+itos(m_peer_id)+")";
}
void Connection::DisconnectPeer(u16 peer_id)
putCommand(discon);
}
-void Connection::sendAck(u16 peer_id, u8 channelnum, u16 seqnum) {
-
+void Connection::sendAck(u16 peer_id, u8 channelnum, u16 seqnum)
+{
assert(channelnum < CHANNEL_COUNT);
LOG(dout_con<<getDesc()
}
} // namespace
-
projects / minetest.git / blobdiff