/*
Minetest
Copyright (C) 2013 celeron55, Perttu Ahola <celeron55@gmail.com>
This program is free software; you can redistribute it and/or modify
it under the terms of the GNU Lesser General Public License as published by
the Free Software Foundation; either version 2.1 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Lesser General Public License for more details.
You should have received a copy of the GNU Lesser General Public License along
with this program; if not, write to the Free Software Foundation, Inc.,
51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
*/
#include <iomanip>
#include <errno.h>
#include "connection.h"
#include "serialization.h"
#include "log.h"
#include "porting.h"
#include "network/networkpacket.h"
#include "util/serialize.h"
#include "util/numeric.h"
#include "util/string.h"
#include "settings.h"
#include "profiler.h"
namespace con
{
/******************************************************************************/
/* defines used for debugging and profiling */
/******************************************************************************/
#ifdef NDEBUG
#define LOG(a) a
#define PROFILE(a)
#undef DEBUG_CONNECTION_KBPS
#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
#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)
{
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;
BufferedPacket p(packet_size);
p.address = address;
writeU32(&p.data[0], protocol_id);
writeU16(&p.data[4], sender_peer_id);
writeU8(&p.data[6], channel);
memcpy(&p.data[BASE_HEADER_SIZE], data, datasize);
return p;
}
BufferedPacket makePacket(Address &address, SharedBuffer<u8> &data,
u32 protocol_id, u16 sender_peer_id, u8 channel)
{
return makePacket(address, *data, data.getSize(),
protocol_id, sender_peer_id, channel);
}
SharedBuffer<u8> makeOriginalPacket(
SharedBuffer<u8> data)
{
u32 header_size = 1;
u32 packet_size = data.getSize() + header_size;
SharedBuffer<u8> b(packet_size);
writeU8(&(b[0]), TYPE_ORIGINAL);
if (data.getSize() > 0) {
memcpy(&(b[header_size]), *data, data.getSize());
}
return b;
}
std::list<SharedBuffer<u8> > makeSplitPacket(
SharedBuffer<u8> data,
u32 chunksize_max,
u16 seqnum)
{
// 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;
u32 end = 0;
u32 chunk_num = 0;
u16 chunk_count = 0;
do{
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
writeU16(&chunk[5], chunk_num);
memcpy(&chunk[chunk_header_size], &data[start], payload_size);
chunks.push_back(chunk);
chunk_count++;
start = end + 1;
chunk_num++;
}
while(end != data.getSize() - 1);
for(std::list<SharedBuffer<u8> >::iterator i = chunks.begin();
i != chunks.end(); ++i)
{
// Write chunk_count
writeU16(&((*i)[3]), chunk_count);
}
return chunks;
}
std::list<SharedBuffer<u8> > makeAutoSplitPacket(
SharedBuffer<u8> data,
u32 chunksize_max,
u16 &split_seqnum)
{
u32 original_header_size = 1;
std::list<SharedBuffer<u8> > list;
if (data.getSize() + original_header_size > chunksize_max)
{
list = makeSplitPacket(data, chunksize_max, split_seqnum);
split_seqnum++;
return list;
}
else
{
list.push_back(makeOriginalPacket(data));
}
return list;
}
SharedBuffer<u8> makeReliablePacket(
SharedBuffer<u8> data,
u16 seqnum)
{
u32 header_size = 3;
u32 packet_size = data.getSize() + header_size;
SharedBuffer<u8> b(packet_size);
writeU8(&b[0], TYPE_RELIABLE);
writeU16(&b[1], seqnum);
memcpy(&b[header_size], *data, data.getSize());
return b;
}
/*
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<BufferedPacket>::iterator i = m_list.begin();
i != m_list.end();
++i)
{
u16 s = readU16(&(i->data[BASE_HEADER_SIZE+1]));
LOG(dout_con<<index<< ":" << s << std::endl);
index++;
}
}
bool ReliablePacketBuffer::empty()
{
MutexAutoLock listlock(m_list_mutex);
return m_list.empty();
}
u32 ReliablePacketBuffer::size()
{
return m_list_size;
}
bool ReliablePacketBuffer::containsPacket(u16 seqnum)
{
return !(findPacket(seqnum) == m_list.end());
}
RPBSearchResult ReliablePacketBuffer::findPacket(u16 seqnum)
{
std::list<BufferedPacket>::iterator i = m_list.begin();
for(; i != m_list.end(); ++i)
{
u16 s = readU16(&(i->data[BASE_HEADER_SIZE+1]));
/*dout_con<<"findPacket(): finding seqnum="<<seqnum
<<", comparing to s="<<s<<std::endl;*/
if (s == seqnum)
break;
}
return i;
}
RPBSearchResult ReliablePacketBuffer::notFound()
{
return m_list.end();
}
bool ReliablePacketBuffer::getFirstSeqnum(u16& result)
{
MutexAutoLock listlock(m_list_mutex);
if (m_list.empty())
return false;
BufferedPacket p = *m_list.begin();
result = readU16(&p.data[BASE_HEADER_SIZE+1]);
return true;
}
BufferedPacket ReliablePacketBuffer::popFirst()
{
MutexAutoLock listlock(m_list_mutex);
if (m_list.empty())
throw NotFoundException("Buffer is empty");
BufferedPacket p = *m_list.begin();
m_list.erase(m_list.begin());
--m_list_size;
if (m_list_size == 0) {
m_oldest_non_answered_ack = 0;
} else {
m_oldest_non_answered_ack =
readU16(&(*m_list.begin()).data[BASE_HEADER_SIZE+1]);
}
return p;
}
BufferedPacket ReliablePacketBuffer::popSeqnum(u16 seqnum)
{
MutexAutoLock listlock(m_list_mutex);
RPBSearchResult r = findPacket(seqnum);
if (r == notFound()) {
LOG(dout_con<<"Sequence number: " << seqnum
<< " not found in reliable buffer"<<std::endl);
throw NotFoundException("seqnum not found in buffer");
}
BufferedPacket p = *r;
RPBSearchResult next = r;
++next;
if (next != notFound()) {
u16 s = readU16(&(next->data[BASE_HEADER_SIZE+1]));
m_oldest_non_answered_ack = s;
}
m_list.erase(r);
--m_list_size;
if (m_list_size == 0)
{ m_oldest_non_answered_ack = 0; }
else
{ m_oldest_non_answered_ack = readU16(&(*m_list.begin()).data[BASE_HEADER_SIZE+1]); }
return p;
}
void ReliablePacketBuffer::insert(BufferedPacket &p,u16 next_expected)
{
MutexAutoLock listlock(m_list_mutex);
if (p.data.getSize() < BASE_HEADER_SIZE + 3) {
errorstream << "ReliablePacketBuffer::insert(): Invalid data size for "
"reliable packet" << std::endl;
return;
}
u8 type = readU8(&p.data[BASE_HEADER_SIZE + 0]);
if (type != TYPE_RELIABLE) {
errorstream << "ReliablePacketBuffer::insert(): type is not reliable"
<< std::endl;
return;
}
u16 seqnum = readU16(&p.data[BASE_HEADER_SIZE + 1]);
if (!seqnum_in_window(seqnum, next_expected, MAX_RELIABLE_WINDOW_SIZE)) {
errorstream << "ReliablePacketBuffer::insert(): seqnum is outside of "
"expected window " << std::endl;
return;
}
if (seqnum == next_expected) {
errorstream << "ReliablePacketBuffer::insert(): seqnum is next expected"
<< std::endl;
return;
}
++m_list_size;
sanity_check(m_list_size <= SEQNUM_MAX+1); // FIXME: Handle the error?
// Find the right place for the packet and insert it there
// If list is empty, just add it
if (m_list.empty())
{
m_list.push_back(p);
m_oldest_non_answered_ack = seqnum;
// Done.
return;
}
// Otherwise find the right place
std::list<BufferedPacket>::iterator i = m_list.begin();
// Find the first packet in the list which has a higher seqnum
u16 s = readU16(&(i->data[BASE_HEADER_SIZE+1]));
/* case seqnum is smaller then next_expected seqnum */
/* this is true e.g. on wrap around */
if (seqnum < next_expected) {
while(((s < seqnum) || (s >= next_expected)) && (i != m_list.end())) {
++i;
if (i != m_list.end())
s = readU16(&(i->data[BASE_HEADER_SIZE+1]));
}
}
/* non wrap around case (at least for incoming and next_expected */
else
{
while(((s < seqnum) && (s >= next_expected)) && (i != m_list.end())) {
++i;
if (i != m_list.end())
s = readU16(&(i->data[BASE_HEADER_SIZE+1]));
}
}
if (s == seqnum) {
if (
(readU16(&(i->data[BASE_HEADER_SIZE+1])) != seqnum) ||
(i->data.getSize() != p.data.getSize()) ||
(i->address != p.address)
)
{
/* if this happens your maximum transfer window may be to big */
fprintf(stderr,
"Duplicated seqnum %d non matching packet detected:\n",
seqnum);
fprintf(stderr, "Old: seqnum: %05d size: %04d, address: %s\n",
readU16(&(i->data[BASE_HEADER_SIZE+1])),i->data.getSize(),
i->address.serializeString().c_str());
fprintf(stderr, "New: seqnum: %05d size: %04u, address: %s\n",
readU16(&(p.data[BASE_HEADER_SIZE+1])),p.data.getSize(),
p.address.serializeString().c_str());
throw IncomingDataCorruption("duplicated packet isn't same as original one");
}
/* nothing to do this seems to be a resent packet */
/* for paranoia reason data should be compared */
--m_list_size;
}
/* insert or push back */
else if (i != m_list.end()) {
m_list.insert(i, p);
}
else {
m_list.push_back(p);
}
/* update last packet number */
m_oldest_non_answered_ack = readU16(&(*m_list.begin()).data[BASE_HEADER_SIZE+1]);
}
void ReliablePacketBuffer::incrementTimeouts(float dtime)
{
MutexAutoLock listlock(m_list_mutex);
for(std::list<BufferedPacket>::iterator i = m_list.begin();
i != m_list.end(); ++i)
{
i->time += dtime;
i->totaltime += dtime;
}
}
std::list<BufferedPacket> ReliablePacketBuffer::getTimedOuts(float timeout,
unsigned int max_packets)
{
MutexAutoLock listlock(m_list_mutex);
std::list<BufferedPacket> timed_outs;
for(std::list<BufferedPacket>::iterator i = m_list.begin();
i != m_list.end(); ++i)
{
if (i->time >= timeout) {
timed_outs.push_back(*i);
//this packet will be sent right afterwards reset timeout here
i->time = 0.0;
if (timed_outs.size() >= max_packets)
break;
}
}
return timed_outs;
}
/*
IncomingSplitBuffer
*/
IncomingSplitBuffer::~IncomingSplitBuffer()
{
MutexAutoLock listlock(m_map_mutex);
for(std::map<u16, IncomingSplitPacket*>::iterator i = m_buf.begin();
i != m_buf.end(); ++i)
{
delete i->second;
}
}
/*
This will throw a GotSplitPacketException when a full
split packet is constructed.
*/
SharedBuffer<u8> IncomingSplitBuffer::insert(BufferedPacket &p, bool reliable)
{
MutexAutoLock listlock(m_map_mutex);
u32 headersize = BASE_HEADER_SIZE + 7;
if (p.data.getSize() < headersize) {
errorstream << "Invalid data size for split packet" << std::endl;
return SharedBuffer<u8>();
}
u8 type = readU8(&p.data[BASE_HEADER_SIZE+0]);
u16 seqnum = readU16(&p.data[BASE_HEADER_SIZE+1]);
u16 chunk_count = readU16(&p.data[BASE_HEADER_SIZE+3]);
u16 chunk_num = readU16(&p.data[BASE_HEADER_SIZE+5]);
if (type != TYPE_SPLIT) {
errorstream << "IncomingSplitBuffer::insert(): type is not split"
<< std::endl;
return SharedBuffer<u8>();
}
// 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;
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
<<" != sp->chunk_count="<<sp->chunk_count
<<std::endl);
if (reliable != sp->reliable)
LOG(derr_con<<"Connection: WARNING: reliable="<<reliable
<<" != sp->reliable="<<sp->reliable
<<std::endl);
// If chunk already exists, ignore it.
// Sometimes two identical packets may arrive when there is network
// 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>();
// Calculate total size
u32 totalsize = 0;
for(std::map<u16, SharedBuffer<u8> >::iterator i = sp->chunks.begin();
i != sp->chunks.end(); ++i)
{
totalsize += i->second.getSize();
}
SharedBuffer<u8> fulldata(totalsize);
// Copy chunks to data buffer
u32 start = 0;
for(u32 chunk_i=0; chunk_i<sp->chunk_count;
chunk_i++)
{
SharedBuffer<u8> buf = sp->chunks[chunk_i];
u16 chunkdatasize = buf.getSize();
memcpy(&fulldata[start], *buf, chunkdatasize);
start += chunkdatasize;;
}
// Remove sp from buffer
m_buf.erase(seqnum);
delete sp;
return fulldata;
}
void IncomingSplitBuffer::removeUnreliableTimedOuts(float dtime, float timeout)
{
std::list<u16> remove_queue;
{
MutexAutoLock listlock(m_map_mutex);
for(std::map<u16, IncomingSplitPacket*>::iterator i = m_buf.begin();
i != m_buf.end(); ++i)
{
IncomingSplitPacket *p = i->second;
// Reliable ones are not removed by timeout
if (p->reliable == true)
continue;
p->time += dtime;
if (p->time >= timeout)
remove_queue.push_back(i->first);
}
}
for(std::list<u16>::iterator j = remove_queue.begin();
j != remove_queue.end(); ++j)
{
MutexAutoLock listlock(m_map_mutex);
LOG(dout_con<<"NOTE: Removing timed out unreliable split packet"<<std::endl);
delete m_buf[*j];
m_buf.erase(*j);
}
}
/*
Channel
*/
Channel::Channel() :
window_size(MIN_RELIABLE_WINDOW_SIZE),
next_incoming_seqnum(SEQNUM_INITIAL),
next_outgoing_seqnum(SEQNUM_INITIAL),
next_outgoing_split_seqnum(SEQNUM_INITIAL),
current_packet_loss(0),
current_packet_too_late(0),
current_packet_successfull(0),
packet_loss_counter(0),
current_bytes_transfered(0),
current_bytes_received(0),
|