src/network/connection.cpp

   1 /*
   2 Minetest
   3 Copyright (C) 2013 celeron55, Perttu Ahola <celeron55@gmail.com>
   4
   5 This program is free software; you can redistribute it and/or modify
   6 it under the terms of the GNU Lesser General Public License as published by
   7 the Free Software Foundation; either version 2.1 of the License, or
   8 (at your option) any later version.
   9
  10 This program is distributed in the hope that it will be useful,
  11 but WITHOUT ANY WARRANTY; without even the implied warranty of
  12 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13 GNU Lesser General Public License for more details.
  14
  15 You should have received a copy of the GNU Lesser General Public License along
  16 with this program; if not, write to the Free Software Foundation, Inc.,
  17 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  18 */
  19
  20 #include <iomanip>
  21 #include <cerrno>
  22 #include "connection.h"
  23 #include "serialization.h"
  24 #include "log.h"
  25 #include "porting.h"
  26 #include "network/networkpacket.h"
  27 #include "network/peerhandler.h"
  28 #include "util/serialize.h"
  29 #include "util/numeric.h"
  30 #include "util/string.h"
  31 #include "settings.h"
  32 #include "profiler.h"
  33
  34 namespace con
  35 {
  36
  37 /******************************************************************************/
  38 /* defines used for debugging and profiling                                   */
  39 /******************************************************************************/
  40 #ifdef NDEBUG
  41 #define LOG(a) a
  42 #define PROFILE(a)
  43 #undef DEBUG_CONNECTION_KBPS
  44 #else
  45 /* this mutex is used to achieve log message consistency */
  46 std::mutex log_message_mutex;
  47 #define LOG(a)                                                                 \
  48         {                                                                          \
  49         MutexAutoLock loglock(log_message_mutex);                                 \
  50         a;                                                                         \
  51         }
  52 #define PROFILE(a) a
  53 //#define DEBUG_CONNECTION_KBPS
  54 #undef DEBUG_CONNECTION_KBPS
  55 #endif
  56
  57
  58 static inline float CALC_DTIME(u64 lasttime, u64 curtime)
  59 {
  60         float value = ( curtime - lasttime) / 1000.0;
  61         return MYMAX(MYMIN(value,0.1),0.0);
  62 }
  63
  64 #define MAX_UDP_PEERS 65535
  65
  66 #define PING_TIMEOUT 5.0
  67
  68 /* maximum number of retries for reliable packets */
  69 #define MAX_RELIABLE_RETRY 5
  70
  71 static u16 readPeerId(u8 *packetdata)
  72 {
  73         return readU16(&packetdata[4]);
  74 }
  75 static u8 readChannel(u8 *packetdata)
  76 {
  77         return readU8(&packetdata[6]);
  78 }
  79
  80 BufferedPacket makePacket(Address &address, u8 *data, u32 datasize,
  81                 u32 protocol_id, u16 sender_peer_id, u8 channel)
  82 {
  83         u32 packet_size = datasize + BASE_HEADER_SIZE;
  84         BufferedPacket p(packet_size);
  85         p.address = address;
  86
  87         writeU32(&p.data[0], protocol_id);
  88         writeU16(&p.data[4], sender_peer_id);
  89         writeU8(&p.data[6], channel);
  90
  91         memcpy(&p.data[BASE_HEADER_SIZE], data, datasize);
  92
  93         return p;
  94 }
  95
  96 BufferedPacket makePacket(Address &address, SharedBuffer<u8> &data,
  97                 u32 protocol_id, u16 sender_peer_id, u8 channel)
  98 {
  99         return makePacket(address, *data, data.getSize(),
 100                         protocol_id, sender_peer_id, channel);
 101 }
 102
 103 SharedBuffer<u8> makeOriginalPacket(
 104                 SharedBuffer<u8> data)
 105 {
 106         u32 header_size = 1;
 107         u32 packet_size = data.getSize() + header_size;
 108         SharedBuffer<u8> b(packet_size);
 109
 110         writeU8(&(b[0]), TYPE_ORIGINAL);
 111         if (data.getSize() > 0) {
 112                 memcpy(&(b[header_size]), *data, data.getSize());
 113         }
 114         return b;
 115 }
 116
 117 std::list<SharedBuffer<u8> > makeSplitPacket(
 118                 SharedBuffer<u8> data,
 119                 u32 chunksize_max,
 120                 u16 seqnum)
 121 {
 122         // Chunk packets, containing the TYPE_SPLIT header
 123         std::list<SharedBuffer<u8> > chunks;
 124
 125         u32 chunk_header_size = 7;
 126         u32 maximum_data_size = chunksize_max - chunk_header_size;
 127         u32 start = 0;
 128         u32 end = 0;
 129         u32 chunk_num = 0;
 130         u16 chunk_count = 0;
 131         do{
 132                 end = start + maximum_data_size - 1;
 133                 if (end > data.getSize() - 1)
 134                         end = data.getSize() - 1;
 135
 136                 u32 payload_size = end - start + 1;
 137                 u32 packet_size = chunk_header_size + payload_size;
 138
 139                 SharedBuffer<u8> chunk(packet_size);
 140
 141                 writeU8(&chunk[0], TYPE_SPLIT);
 142                 writeU16(&chunk[1], seqnum);
 143                 // [3] u16 chunk_count is written at next stage
 144                 writeU16(&chunk[5], chunk_num);
 145                 memcpy(&chunk[chunk_header_size], &data[start], payload_size);
 146
 147                 chunks.push_back(chunk);
 148                 chunk_count++;
 149
 150                 start = end + 1;
 151                 chunk_num++;
 152         }
 153         while(end != data.getSize() - 1);
 154
 155         for (SharedBuffer<u8> &chunk : chunks) {
 156                 // Write chunk_count
 157                 writeU16(&(chunk[3]), chunk_count);
 158         }
 159
 160         return chunks;
 161 }
 162
 163 std::list<SharedBuffer<u8> > makeAutoSplitPacket(
 164                 SharedBuffer<u8> data,
 165                 u32 chunksize_max,
 166                 u16 &split_seqnum)
 167 {
 168         u32 original_header_size = 1;
 169         std::list<SharedBuffer<u8> > list;
 170         if (data.getSize() + original_header_size > chunksize_max)
 171         {
 172                 list = makeSplitPacket(data, chunksize_max, split_seqnum);
 173                 split_seqnum++;
 174                 return list;
 175         }
 176
 177         list.push_back(makeOriginalPacket(data));
 178         return list;
 179 }
 180
 181 SharedBuffer<u8> makeReliablePacket(
 182                 const SharedBuffer<u8> &data,
 183                 u16 seqnum)
 184 {
 185         u32 header_size = 3;
 186         u32 packet_size = data.getSize() + header_size;
 187         SharedBuffer<u8> b(packet_size);
 188
 189         writeU8(&b[0], TYPE_RELIABLE);
 190         writeU16(&b[1], seqnum);
 191
 192         memcpy(&b[header_size], *data, data.getSize());
 193
 194         return b;
 195 }
 196
 197 /*
 198         ReliablePacketBuffer
 199 */
 200
 201 void ReliablePacketBuffer::print()
 202 {
 203         MutexAutoLock listlock(m_list_mutex);
 204         LOG(dout_con<<"Dump of ReliablePacketBuffer:" << std::endl);
 205         unsigned int index = 0;
 206         for (BufferedPacket &bufferedPacket : m_list) {
 207                 u16 s = readU16(&(bufferedPacket.data[BASE_HEADER_SIZE+1]));
 208                 LOG(dout_con<<index<< ":" << s << std::endl);
 209                 index++;
 210         }
 211 }
 212 bool ReliablePacketBuffer::empty()
 213 {
 214         MutexAutoLock listlock(m_list_mutex);
 215         return m_list.empty();
 216 }
 217
 218 u32 ReliablePacketBuffer::size()
 219 {
 220         return m_list_size;
 221 }
 222
 223 bool ReliablePacketBuffer::containsPacket(u16 seqnum)
 224 {
 225         return !(findPacket(seqnum) == m_list.end());
 226 }
 227
 228 RPBSearchResult ReliablePacketBuffer::findPacket(u16 seqnum)
 229 {
 230         std::list<BufferedPacket>::iterator i = m_list.begin();
 231         for(; i != m_list.end(); ++i)
 232         {
 233                 u16 s = readU16(&(i->data[BASE_HEADER_SIZE+1]));
 234                 /*dout_con<<"findPacket(): finding seqnum="<<seqnum
 235                                 <<", comparing to s="<<s<<std::endl;*/
 236                 if (s == seqnum)
 237                         break;
 238         }
 239         return i;
 240 }
 241 RPBSearchResult ReliablePacketBuffer::notFound()
 242 {
 243         return m_list.end();
 244 }
 245 bool ReliablePacketBuffer::getFirstSeqnum(u16& result)
 246 {
 247         MutexAutoLock listlock(m_list_mutex);
 248         if (m_list.empty())
 249                 return false;
 250         BufferedPacket p = *m_list.begin();
 251         result = readU16(&p.data[BASE_HEADER_SIZE+1]);
 252         return true;
 253 }
 254
 255 BufferedPacket ReliablePacketBuffer::popFirst()
 256 {
 257         MutexAutoLock listlock(m_list_mutex);
 258         if (m_list.empty())
 259                 throw NotFoundException("Buffer is empty");
 260         BufferedPacket p = *m_list.begin();
 261         m_list.erase(m_list.begin());
 262         --m_list_size;
 263
 264         if (m_list_size == 0) {
 265                 m_oldest_non_answered_ack = 0;
 266         } else {
 267                 m_oldest_non_answered_ack =
 268                                 readU16(&(*m_list.begin()).data[BASE_HEADER_SIZE+1]);
 269         }
 270         return p;
 271 }
 272 BufferedPacket ReliablePacketBuffer::popSeqnum(u16 seqnum)
 273 {
 274         MutexAutoLock listlock(m_list_mutex);
 275         RPBSearchResult r = findPacket(seqnum);
 276         if (r == notFound()) {
 277                 LOG(dout_con<<"Sequence number: " << seqnum
 278                                 << " not found in reliable buffer"<<std::endl);
 279                 throw NotFoundException("seqnum not found in buffer");
 280         }
 281         BufferedPacket p = *r;
 282
 283
 284         RPBSearchResult next = r;
 285         ++next;
 286         if (next != notFound()) {
 287                 u16 s = readU16(&(next->data[BASE_HEADER_SIZE+1]));
 288                 m_oldest_non_answered_ack = s;
 289         }
 290
 291         m_list.erase(r);
 292         --m_list_size;
 293
 294         if (m_list_size == 0)
 295         { m_oldest_non_answered_ack = 0; }
 296         else
 297         { m_oldest_non_answered_ack = readU16(&(*m_list.begin()).data[BASE_HEADER_SIZE+1]);     }
 298         return p;
 299 }
 300 void ReliablePacketBuffer::insert(BufferedPacket &p,u16 next_expected)
 301 {
 302         MutexAutoLock listlock(m_list_mutex);
 303         if (p.data.getSize() < BASE_HEADER_SIZE + 3) {
 304                 errorstream << "ReliablePacketBuffer::insert(): Invalid data size for "
 305                         "reliable packet" << std::endl;
 306                 return;
 307         }
 308         u8 type = readU8(&p.data[BASE_HEADER_SIZE + 0]);
 309         if (type != TYPE_RELIABLE) {
 310                 errorstream << "ReliablePacketBuffer::insert(): type is not reliable"
 311                         << std::endl;
 312                 return;
 313         }
 314         u16 seqnum = readU16(&p.data[BASE_HEADER_SIZE + 1]);
 315
 316         if (!seqnum_in_window(seqnum, next_expected, MAX_RELIABLE_WINDOW_SIZE)) {
 317                 errorstream << "ReliablePacketBuffer::insert(): seqnum is outside of "
 318                         "expected window " << std::endl;
 319                 return;
 320         }
 321         if (seqnum == next_expected) {
 322                 errorstream << "ReliablePacketBuffer::insert(): seqnum is next expected"
 323                         << std::endl;
 324                 return;
 325         }
 326
 327         ++m_list_size;
 328         sanity_check(m_list_size <= SEQNUM_MAX+1);      // FIXME: Handle the error?
 329
 330         // Find the right place for the packet and insert it there
 331         // If list is empty, just add it
 332         if (m_list.empty())
 333         {
 334                 m_list.push_back(p);
 335                 m_oldest_non_answered_ack = seqnum;
 336                 // Done.
 337                 return;
 338         }
 339
 340         // Otherwise find the right place
 341         std::list<BufferedPacket>::iterator i = m_list.begin();
 342         // Find the first packet in the list which has a higher seqnum
 343         u16 s = readU16(&(i->data[BASE_HEADER_SIZE+1]));
 344
 345         /* case seqnum is smaller then next_expected seqnum */
 346         /* this is true e.g. on wrap around */
 347         if (seqnum < next_expected) {
 348                 while(((s < seqnum) || (s >= next_expected)) && (i != m_list.end())) {
 349                         ++i;
 350                         if (i != m_list.end())
 351                                 s = readU16(&(i->data[BASE_HEADER_SIZE+1]));
 352                 }
 353         }
 354         /* non wrap around case (at least for incoming and next_expected */
 355         else
 356         {
 357                 while(((s < seqnum) && (s >= next_expected)) && (i != m_list.end())) {
 358                         ++i;
 359                         if (i != m_list.end())
 360                                 s = readU16(&(i->data[BASE_HEADER_SIZE+1]));
 361                 }
 362         }
 363
 364         if (s == seqnum) {
 365                 if (
 366                         (readU16(&(i->data[BASE_HEADER_SIZE+1])) != seqnum) ||
 367                         (i->data.getSize() != p.data.getSize()) ||
 368                         (i->address != p.address)
 369                         )
 370                 {
 371                         /* if this happens your maximum transfer window may be to big */
 372                         fprintf(stderr,
 373                                         "Duplicated seqnum %d non matching packet detected:\n",
 374                                         seqnum);
 375                         fprintf(stderr, "Old: seqnum: %05d size: %04d, address: %s\n",
 376                                         readU16(&(i->data[BASE_HEADER_SIZE+1])),i->data.getSize(),
 377                                         i->address.serializeString().c_str());
 378                         fprintf(stderr, "New: seqnum: %05d size: %04u, address: %s\n",
 379                                         readU16(&(p.data[BASE_HEADER_SIZE+1])),p.data.getSize(),
 380                                         p.address.serializeString().c_str());
 381                         throw IncomingDataCorruption("duplicated packet isn't same as original one");
 382                 }
 383
 384                 /* nothing to do this seems to be a resent packet */
 385                 /* for paranoia reason data should be compared */
 386                 --m_list_size;
 387         }
 388         /* insert or push back */
 389         else if (i != m_list.end()) {
 390                 m_list.insert(i, p);
 391         }
 392         else {
 393                 m_list.push_back(p);
 394         }
 395
 396         /* update last packet number */
 397         m_oldest_non_answered_ack = readU16(&(*m_list.begin()).data[BASE_HEADER_SIZE+1]);
 398 }
 399
 400 void ReliablePacketBuffer::incrementTimeouts(float dtime)
 401 {
 402         MutexAutoLock listlock(m_list_mutex);
 403         for (BufferedPacket &bufferedPacket : m_list) {
 404                 bufferedPacket.time += dtime;
 405                 bufferedPacket.totaltime += dtime;
 406         }
 407 }
 408
 409 std::list<BufferedPacket> ReliablePacketBuffer::getTimedOuts(float timeout,
 410                                                                                                         unsigned int max_packets)
 411 {
 412         MutexAutoLock listlock(m_list_mutex);
 413         std::list<BufferedPacket> timed_outs;
 414         for (BufferedPacket &bufferedPacket : m_list) {
 415                 if (bufferedPacket.time >= timeout) {
 416                         timed_outs.push_back(bufferedPacket);
 417
 418                         //this packet will be sent right afterwards reset timeout here
 419                         bufferedPacket.time = 0.0f;
 420                         if (timed_outs.size() >= max_packets)
 421                                 break;
 422                 }
 423         }
 424         return timed_outs;
 425 }
 426
 427 /*
 428         IncomingSplitBuffer
 429 */
 430
 431 IncomingSplitBuffer::~IncomingSplitBuffer()
 432 {
 433         MutexAutoLock listlock(m_map_mutex);
 434         for (auto &i : m_buf) {
 435                 delete i.second;
 436         }
 437 }
 438 /*
 439         This will throw a GotSplitPacketException when a full
 440         split packet is constructed.
 441 */
 442 SharedBuffer<u8> IncomingSplitBuffer::insert(BufferedPacket &p, bool reliable)
 443 {
 444         MutexAutoLock listlock(m_map_mutex);
 445         u32 headersize = BASE_HEADER_SIZE + 7;
 446         if (p.data.getSize() < headersize) {
 447                 errorstream << "Invalid data size for split packet" << std::endl;
 448                 return SharedBuffer<u8>();
 449         }
 450         u8 type = readU8(&p.data[BASE_HEADER_SIZE+0]);
 451         u16 seqnum = readU16(&p.data[BASE_HEADER_SIZE+1]);
 452         u16 chunk_count = readU16(&p.data[BASE_HEADER_SIZE+3]);
 453         u16 chunk_num = readU16(&p.data[BASE_HEADER_SIZE+5]);
 454
 455         if (type != TYPE_SPLIT) {
 456                 errorstream << "IncomingSplitBuffer::insert(): type is not split"
 457                         << std::endl;
 458                 return SharedBuffer<u8>();
 459         }
 460
 461         // Add if doesn't exist
 462         if (m_buf.find(seqnum) == m_buf.end())
 463         {
 464                 IncomingSplitPacket *sp = new IncomingSplitPacket();
 465                 sp->chunk_count = chunk_count;
 466                 sp->reliable = reliable;
 467                 m_buf[seqnum] = sp;
 468         }
 469
 470         IncomingSplitPacket *sp = m_buf[seqnum];
 471
 472         // TODO: These errors should be thrown or something? Dunno.
 473         if (chunk_count != sp->chunk_count)
 474                 LOG(derr_con<<"Connection: WARNING: chunk_count="<<chunk_count
 475                                 <<" != sp->chunk_count="<<sp->chunk_count
 476                                 <<std::endl);
 477         if (reliable != sp->reliable)
 478                 LOG(derr_con<<"Connection: WARNING: reliable="<<reliable
 479                                 <<" != sp->reliable="<<sp->reliable
 480                                 <<std::endl);
 481
 482         // If chunk already exists, ignore it.
 483         // Sometimes two identical packets may arrive when there is network
 484         // lag and the server re-sends stuff.
 485         if (sp->chunks.find(chunk_num) != sp->chunks.end())
 486                 return SharedBuffer<u8>();
 487
 488         // Cut chunk data out of packet
 489         u32 chunkdatasize = p.data.getSize() - headersize;
 490         SharedBuffer<u8> chunkdata(chunkdatasize);
 491         memcpy(*chunkdata, &(p.data[headersize]), chunkdatasize);
 492
 493         // Set chunk data in buffer
 494         sp->chunks[chunk_num] = chunkdata;
 495
 496         // If not all chunks are received, return empty buffer
 497         if (!sp->allReceived())
 498                 return SharedBuffer<u8>();
 499
 500         // Calculate total size
 501         u32 totalsize = 0;
 502         for (const auto &chunk : sp->chunks) {
 503                 totalsize += chunk.second.getSize();
 504         }
 505
 506         SharedBuffer<u8> fulldata(totalsize);
 507
 508         // Copy chunks to data buffer
 509         u32 start = 0;
 510         for(u32 chunk_i=0; chunk_i<sp->chunk_count;
 511                         chunk_i++)
 512         {
 513                 SharedBuffer<u8> buf = sp->chunks[chunk_i];
 514                 u16 chunkdatasize = buf.getSize();
 515                 memcpy(&fulldata[start], *buf, chunkdatasize);
 516                 start += chunkdatasize;;
 517         }
 518
 519         // Remove sp from buffer
 520         m_buf.erase(seqnum);
 521         delete sp;
 522
 523         return fulldata;
 524 }
 525 void IncomingSplitBuffer::removeUnreliableTimedOuts(float dtime, float timeout)
 526 {
 527         std::list<u16> remove_queue;
 528         {
 529                 MutexAutoLock listlock(m_map_mutex);
 530                 for (auto &i : m_buf) {
 531                         IncomingSplitPacket *p = i.second;
 532                         // Reliable ones are not removed by timeout
 533                         if (p->reliable)
 534                                 continue;
 535                         p->time += dtime;
 536                         if (p->time >= timeout)
 537                                 remove_queue.push_back(i.first);
 538                 }
 539         }
 540         for (u16 j : remove_queue) {
 541                 MutexAutoLock listlock(m_map_mutex);
 542                 LOG(dout_con<<"NOTE: Removing timed out unreliable split packet"<<std::endl);
 543                 delete m_buf[j];
 544                 m_buf.erase(j);
 545         }
 546 }
 547
 548 /*
 549         ConnectionCommand
 550  */
 551
 552 void ConnectionCommand::send(u16 peer_id_, u8 channelnum_, NetworkPacket *pkt,
 553         bool reliable_)
 554 {
 555         type = CONNCMD_SEND;
 556         peer_id = peer_id_;
 557         channelnum = channelnum_;
 558         data = pkt->oldForgePacket();
 559         reliable = reliable_;
 560 }
 561
 562 /*
 563         Channel
 564 */
 565
 566 u16 Channel::readNextIncomingSeqNum()
 567 {
 568         MutexAutoLock internal(m_internal_mutex);
 569         return next_incoming_seqnum;
 570 }
 571
 572 u16 Channel::incNextIncomingSeqNum()
 573 {
 574         MutexAutoLock internal(m_internal_mutex);
 575         u16 retval = next_incoming_seqnum;
 576         next_incoming_seqnum++;
 577         return retval;
 578 }
 579
 580 u16 Channel::readNextSplitSeqNum()
 581 {
 582         MutexAutoLock internal(m_internal_mutex);
 583         return next_outgoing_split_seqnum;
 584 }
 585 void Channel::setNextSplitSeqNum(u16 seqnum)
 586 {
 587         MutexAutoLock internal(m_internal_mutex);
 588         next_outgoing_split_seqnum = seqnum;
 589 }
 590
 591 u16 Channel::getOutgoingSequenceNumber(bool& successful)
 592 {
 593         MutexAutoLock internal(m_internal_mutex);
 594         u16 retval = next_outgoing_seqnum;
 595         u16 lowest_unacked_seqnumber;
 596
 597         /* shortcut if there ain't any packet in outgoing list */
 598         if (outgoing_reliables_sent.empty())
 599         {
 600                 next_outgoing_seqnum++;
 601                 return retval;
 602         }
 603
 604         if (outgoing_reliables_sent.getFirstSeqnum(lowest_unacked_seqnumber))
 605         {
 606                 if (lowest_unacked_seqnumber < next_outgoing_seqnum) {
 607                         // ugly cast but this one is required in order to tell compiler we
 608                         // know about difference of two unsigned may be negative in general
 609                         // but we already made sure it won't happen in this case
 610                         if (((u16)(next_outgoing_seqnum - lowest_unacked_seqnumber)) > window_size) {
 611                                 successful = false;
 612                                 return 0;
 613                         }
 614                 }
 615                 else {
 616                         // ugly cast but this one is required in order to tell compiler we
 617                         // know about difference of two unsigned may be negative in general
 618                         // but we already made sure it won't happen in this case
 619                         if ((next_outgoing_seqnum + (u16)(SEQNUM_MAX - lowest_unacked_seqnumber)) >
 620                                 window_size) {
 621                                 successful = false;
 622                                 return 0;
 623                         }
 624                 }
 625         }
 626
 627         next_outgoing_seqnum++;
 628         return retval;
 629 }
 630
 631 u16 Channel::readOutgoingSequenceNumber()
 632 {
 633         MutexAutoLock internal(m_internal_mutex);
 634         return next_outgoing_seqnum;
 635 }
 636
 637 bool Channel::putBackSequenceNumber(u16 seqnum)
 638 {
 639         if (((seqnum + 1) % (SEQNUM_MAX+1)) == next_outgoing_seqnum) {
 640
 641                 next_outgoing_seqnum = seqnum;
 642                 return true;
 643         }
 644         return false;
 645 }
 646
 647 void Channel::UpdateBytesSent(unsigned int bytes, unsigned int packets)
 648 {
 649         MutexAutoLock internal(m_internal_mutex);
 650         current_bytes_transfered += bytes;
 651         current_packet_successfull += packets;
 652 }
 653
 654 void Channel::UpdateBytesReceived(unsigned int bytes) {
 655         MutexAutoLock internal(m_internal_mutex);
 656         current_bytes_received += bytes;
 657 }
 658
 659 void Channel::UpdateBytesLost(unsigned int bytes)
 660 {
 661         MutexAutoLock internal(m_internal_mutex);
 662         current_bytes_lost += bytes;
 663 }
 664
 665
 666 void Channel::UpdatePacketLossCounter(unsigned int count)
 667 {
 668         MutexAutoLock internal(m_internal_mutex);
 669         current_packet_loss += count;
 670 }
 671
 672 void Channel::UpdatePacketTooLateCounter()
 673 {
 674         MutexAutoLock internal(m_internal_mutex);
 675         current_packet_too_late++;
 676 }
 677
 678 void Channel::UpdateTimers(float dtime,bool legacy_peer)
 679 {
 680         bpm_counter += dtime;
 681         packet_loss_counter += dtime;
 682
 683         if (packet_loss_counter > 1.0)
 684         {
 685                 packet_loss_counter -= 1.0;
 686
 687                 unsigned int packet_loss = 11; /* use a neutral value for initialization */
 688                 unsigned int packets_successfull = 0;
 689                 //unsigned int packet_too_late = 0;
 690
 691                 bool reasonable_amount_of_data_transmitted = false;
 692
 693                 {
 694                         MutexAutoLock internal(m_internal_mutex);
 695                         packet_loss = current_packet_loss;
 696                         //packet_too_late = current_packet_too_late;
 697                         packets_successfull = current_packet_successfull;
 698
 699                         if (current_bytes_transfered > (unsigned int) (window_size*512/2))
 700                         {
 701                                 reasonable_amount_of_data_transmitted = true;
 702                         }
 703                         current_packet_loss = 0;
 704                         current_packet_too_late = 0;
 705                         current_packet_successfull = 0;
 706                 }
 707
 708                 /* dynamic window size is only available for non legacy peers */
 709                 if (!legacy_peer) {
 710                         float successfull_to_lost_ratio = 0.0;
 711                         bool done = false;
 712
 713                         if (packets_successfull > 0) {
 714                                 successfull_to_lost_ratio = packet_loss/packets_successfull;
 715                         }
 716                         else if (packet_loss > 0)
 717                         {
 718                                 window_size = MYMAX(
 719                                                 (window_size - 10),
 720                                                 MIN_RELIABLE_WINDOW_SIZE);
 721                                 done = true;
 722                         }
 723
 724                         if (!done)
 725                         {
 726                                 if ((successfull_to_lost_ratio < 0.01) &&
 727                                         (window_size < MAX_RELIABLE_WINDOW_SIZE))
 728                                 {
 729                                         /* don't even think about increasing if we didn't even
 730                                          * use major parts of our window */
 731                                         if (reasonable_amount_of_data_transmitted)
 732                                                 window_size = MYMIN(
 733                                                                 (window_size + 100),
 734                                                                 MAX_RELIABLE_WINDOW_SIZE);
 735                                 }
 736                                 else if ((successfull_to_lost_ratio < 0.05) &&
 737                                                 (window_size < MAX_RELIABLE_WINDOW_SIZE))
 738                                 {
 739                                         /* don't even think about increasing if we didn't even
 740                                          * use major parts of our window */
 741                                         if (reasonable_amount_of_data_transmitted)
 742                                                 window_size = MYMIN(
 743                                                                 (window_size + 50),
 744                                                                 MAX_RELIABLE_WINDOW_SIZE);
 745                                 }
 746                                 else if (successfull_to_lost_ratio > 0.15)
 747                                 {
 748                                         window_size = MYMAX(
 749                                                         (window_size - 100),
 750                                                         MIN_RELIABLE_WINDOW_SIZE);
 751                                 }
 752                                 else if (successfull_to_lost_ratio > 0.1)
 753                                 {
 754                                         window_size = MYMAX(
 755                                                         (window_size - 50),
 756                                                         MIN_RELIABLE_WINDOW_SIZE);
 757                                 }
 758                         }
 759                 }
 760         }
 761
 762         if (bpm_counter > 10.0)
 763         {
 764                 {
 765                         MutexAutoLock internal(m_internal_mutex);
 766                         cur_kbps                 =
 767                                         (((float) current_bytes_transfered)/bpm_counter)/1024.0;
 768                         current_bytes_transfered = 0;
 769                         cur_kbps_lost            =
 770                                         (((float) current_bytes_lost)/bpm_counter)/1024.0;
 771                         current_bytes_lost       = 0;
 772                         cur_incoming_kbps        =
 773                                         (((float) current_bytes_received)/bpm_counter)/1024.0;
 774                         current_bytes_received   = 0;
 775                         bpm_counter              = 0;
 776                 }
 777
 778                 if (cur_kbps > max_kbps)
 779                 {
 780                         max_kbps = cur_kbps;
 781                 }
 782
 783                 if (cur_kbps_lost > max_kbps_lost)
 784                 {
 785                         max_kbps_lost = cur_kbps_lost;
 786                 }
 787
 788                 if (cur_incoming_kbps > max_incoming_kbps) {
 789                         max_incoming_kbps = cur_incoming_kbps;
 790                 }
 791
 792                 rate_samples       = MYMIN(rate_samples+1,10);
 793                 float old_fraction = ((float) (rate_samples-1) )/( (float) rate_samples);
 794                 avg_kbps           = avg_kbps * old_fraction +
 795                                 cur_kbps * (1.0 - old_fraction);
 796                 avg_kbps_lost      = avg_kbps_lost * old_fraction +
 797                                 cur_kbps_lost * (1.0 - old_fraction);
 798                 avg_incoming_kbps  = avg_incoming_kbps * old_fraction +
 799                                 cur_incoming_kbps * (1.0 - old_fraction);
 800         }
 801 }
 802
 803
 804 /*
 805         Peer
 806 */
 807
 808 PeerHelper::PeerHelper(Peer* peer) :
 809         m_peer(peer)
 810 {
 811         if (peer && !peer->IncUseCount())
 812                 m_peer = nullptr;
 813 }
 814
 815 PeerHelper::~PeerHelper()
 816 {
 817         if (m_peer)
 818                 m_peer->DecUseCount();
 819
 820         m_peer = nullptr;
 821 }
 822
 823 PeerHelper& PeerHelper::operator=(Peer* peer)
 824 {
 825         m_peer = peer;
 826         if (peer && !peer->IncUseCount())
 827                 m_peer = nullptr;
 828         return *this;
 829 }
 830
 831 Peer* PeerHelper::operator->() const
 832 {
 833         return m_peer;
 834 }
 835
 836 Peer* PeerHelper::operator&() const
 837 {
 838         return m_peer;
 839 }
 840
 841 bool PeerHelper::operator!() {
 842         return ! m_peer;
 843 }
 844
 845 bool PeerHelper::operator!=(void* ptr)
 846 {
 847         return ((void*) m_peer != ptr);
 848 }
 849
 850 bool Peer::IncUseCount()
 851 {
 852         MutexAutoLock lock(m_exclusive_access_mutex);
 853
 854         if (!m_pending_deletion) {
 855                 this->m_usage++;
 856                 return true;
 857         }
 858
 859         return false;
 860 }
 861
 862 void Peer::DecUseCount()
 863 {
 864         {
 865                 MutexAutoLock lock(m_exclusive_access_mutex);
 866                 sanity_check(m_usage > 0);
 867                 m_usage--;
 868
 869                 if (!((m_pending_deletion) && (m_usage == 0)))
 870                         return;
 871         }
 872         delete this;
 873 }
 874
 875 void Peer::RTTStatistics(float rtt, const std::string &profiler_id,
 876                 unsigned int num_samples) {
 877
 878         if (m_last_rtt > 0) {
 879                 /* set min max values */
 880                 if (rtt < m_rtt.min_rtt)
 881                         m_rtt.min_rtt = rtt;
 882                 if (rtt >= m_rtt.max_rtt)
 883                         m_rtt.max_rtt = rtt;
 884
 885                 /* do average calculation */
 886                 if (m_rtt.avg_rtt < 0.0)
 887                         m_rtt.avg_rtt  = rtt;
 888                 else
 889                         m_rtt.avg_rtt  = m_rtt.avg_rtt * (num_samples/(num_samples-1)) +
 890                                                                 rtt * (1/num_samples);
 891
 892                 /* do jitter calculation */
 893
 894                 //just use some neutral value at beginning
 895                 float jitter = m_rtt.jitter_min;
 896
 897                 if (rtt > m_last_rtt)
 898                         jitter = rtt-m_last_rtt;
 899
 900                 if (rtt <= m_last_rtt)
 901                         jitter = m_last_rtt - rtt;
 902
 903                 if (jitter < m_rtt.jitter_min)
 904                         m_rtt.jitter_min = jitter;
 905                 if (jitter >= m_rtt.jitter_max)
 906                         m_rtt.jitter_max = jitter;
 907
 908                 if (m_rtt.jitter_avg < 0.0)
 909                         m_rtt.jitter_avg  = jitter;
 910                 else
 911                         m_rtt.jitter_avg  = m_rtt.jitter_avg * (num_samples/(num_samples-1)) +
 912                                                                 jitter * (1/num_samples);
 913
 914                 if (!profiler_id.empty()) {
 915                         g_profiler->graphAdd(profiler_id + "_rtt", rtt);
 916                         g_profiler->graphAdd(profiler_id + "_jitter", jitter);
 917                 }
 918         }
 919         /* save values required for next loop */
 920         m_last_rtt = rtt;
 921 }
 922
 923 bool Peer::isTimedOut(float timeout)
 924 {
 925         MutexAutoLock lock(m_exclusive_access_mutex);
 926         u64 current_time = porting::getTimeMs();
 927
 928         float dtime = CALC_DTIME(m_last_timeout_check,current_time);
 929         m_last_timeout_check = current_time;
 930
 931         m_timeout_counter += dtime;
 932
 933         return m_timeout_counter > timeout;
 934 }
 935
 936 void Peer::Drop()
 937 {
 938         {
 939                 MutexAutoLock usage_lock(m_exclusive_access_mutex);
 940                 m_pending_deletion = true;
 941                 if (m_usage != 0)
 942                         return;
 943         }
 944
 945         PROFILE(std::stringstream peerIdentifier1);
 946         PROFILE(peerIdentifier1 << "runTimeouts[" << m_connection->getDesc()
 947                         << ";" << id << ";RELIABLE]");
 948         PROFILE(g_profiler->remove(peerIdentifier1.str()));
 949         PROFILE(std::stringstream peerIdentifier2);
 950         PROFILE(peerIdentifier2 << "sendPackets[" << m_connection->getDesc()
 951                         << ";" << id << ";RELIABLE]");
 952         PROFILE(ScopeProfiler peerprofiler(g_profiler, peerIdentifier2.str(), SPT_AVG));
 953
 954         delete this;
 955 }
 956
 957 UDPPeer::UDPPeer(u16 a_id, Address a_address, Connection* connection) :
 958         Peer(a_address,a_id,connection)
 959 {
 960 }
 961
 962 bool UDPPeer::getAddress(MTProtocols type,Address& toset)
 963 {
 964         if ((type == MTP_UDP) || (type == MTP_MINETEST_RELIABLE_UDP) || (type == MTP_PRIMARY))
 965         {
 966                 toset = address;
 967                 return true;
 968         }
 969
 970         return false;
 971 }
 972
 973 void UDPPeer::setNonLegacyPeer()
 974 {
 975         m_legacy_peer = false;
 976         for(unsigned int i=0; i< CHANNEL_COUNT; i++)
 977         {
 978                 channels->setWindowSize(g_settings->getU16("max_packets_per_iteration"));
 979         }
 980 }
 981
 982 void UDPPeer::reportRTT(float rtt)
 983 {
 984         if (rtt < 0.0) {
 985                 return;
 986         }
 987         RTTStatistics(rtt,"rudp",MAX_RELIABLE_WINDOW_SIZE*10);
 988
 989         float timeout = getStat(AVG_RTT) * RESEND_TIMEOUT_FACTOR;
 990         if (timeout < RESEND_TIMEOUT_MIN)
 991                 timeout = RESEND_TIMEOUT_MIN;
 992         if (timeout > RESEND_TIMEOUT_MAX)
 993                 timeout = RESEND_TIMEOUT_MAX;
 994
 995         MutexAutoLock usage_lock(m_exclusive_access_mutex);
 996         resend_timeout = timeout;
 997 }
 998
 999 bool UDPPeer::Ping(float dtime,SharedBuffer<u8>& data)
1000 {
1001         m_ping_timer += dtime;
1002         if (m_ping_timer >= PING_TIMEOUT)
1003         {
1004                 // Create and send PING packet
1005                 writeU8(&data[0], TYPE_CONTROL);
1006                 writeU8(&data[1], CONTROLTYPE_PING);
1007                 m_ping_timer = 0.0;
1008                 return true;
1009         }
1010         return false;
1011 }
1012
1013 void UDPPeer::PutReliableSendCommand(ConnectionCommand &c,
1014                 unsigned int max_packet_size)
1015 {
1016         if (m_pending_disconnect)
1017                 return;
1018
1019         if ( channels[c.channelnum].queued_commands.empty() &&
1020                         /* don't queue more packets then window size */
1021                         (channels[c.channelnum].queued_reliables.size()
1022                         < (channels[c.channelnum].getWindowSize()/2))) {
1023                 LOG(dout_con<<m_connection->getDesc()
1024                                 <<" processing reliable command for peer id: " << c.peer_id
1025                                 <<" data size: " << c.data.getSize() << std::endl);
1026                 if (!processReliableSendCommand(c,max_packet_size)) {
1027                         channels[c.channelnum].queued_commands.push_back(c);
1028                 }
1029         }
1030         else {
1031                 LOG(dout_con<<m_connection->getDesc()
1032                                 <<" Queueing reliable command for peer id: " << c.peer_id
1033                                 <<" data size: " << c.data.getSize() <<std::endl);
1034                 channels[c.channelnum].queued_commands.push_back(c);
1035         }
1036 }
1037
1038 bool UDPPeer::processReliableSendCommand(
1039                                 ConnectionCommand &c,
1040                                 unsigned int max_packet_size)
1041 {
1042         if (m_pending_disconnect)
1043                 return true;
1044
1045         u32 chunksize_max = max_packet_size
1046                                                         - BASE_HEADER_SIZE
1047                                                         - RELIABLE_HEADER_SIZE;
1048
1049         sanity_check(c.data.getSize() < MAX_RELIABLE_WINDOW_SIZE*512);
1050
1051         std::list<SharedBuffer<u8> > originals;
1052         u16 split_sequence_number = channels[c.channelnum].readNextSplitSeqNum();
1053
1054         if (c.raw)
1055         {
1056                 originals.emplace_back(c.data);
1057         }
1058         else {
1059                 originals = makeAutoSplitPacket(c.data, chunksize_max,split_sequence_number);
1060                 channels[c.channelnum].setNextSplitSeqNum(split_sequence_number);
1061         }
1062
1063         bool have_sequence_number = true;
1064         bool have_initial_sequence_number = false;
1065         std::queue<BufferedPacket> toadd;
1066         volatile u16 initial_sequence_number = 0;
1067
1068         for (SharedBuffer<u8> &original : originals) {
1069                 u16 seqnum = channels[c.channelnum].getOutgoingSequenceNumber(have_sequence_number);
1070
1071                 /* oops, we don't have enough sequence numbers to send this packet */
1072                 if (!have_sequence_number)
1073                         break;
1074
1075                 if (!have_initial_sequence_number)
1076                 {
1077                         initial_sequence_number = seqnum;
1078                         have_initial_sequence_number = true;
1079                 }
1080
1081                 SharedBuffer<u8> reliable = makeReliablePacket(original, seqnum);
1082
1083                 // Add base headers and make a packet
1084                 BufferedPacket p = con::makePacket(address, reliable,
1085                                 m_connection->GetProtocolID(), m_connection->GetPeerID(),
1086                                 c.channelnum);
1087
1088                 toadd.push(p);
1089         }
1090
1091         if (have_sequence_number) {
1092                 volatile u16 pcount = 0;
1093                 while (!toadd.empty()) {
1094                         BufferedPacket p = toadd.front();
1095                         toadd.pop();
1096 //                      LOG(dout_con<<connection->getDesc()
1097 //                                      << " queuing reliable packet for peer_id: " << c.peer_id
1098 //                                      << " channel: " << (c.channelnum&0xFF)
1099 //                                      << " seqnum: " << readU16(&p.data[BASE_HEADER_SIZE+1])
1100 //                                      << std::endl)
1101                         channels[c.channelnum].queued_reliables.push(p);
1102                         pcount++;
1103                 }
1104                 sanity_check(channels[c.channelnum].queued_reliables.size() < 0xFFFF);
1105                 return true;
1106         }
1107
1108         volatile u16 packets_available = toadd.size();
1109         /* we didn't get a single sequence number no need to fill queue */
1110         if (!have_initial_sequence_number) {
1111                 return false;
1112         }
1113
1114         while (!toadd.empty()) {
1115                 /* remove packet */
1116                 toadd.pop();
1117
1118                 bool successfully_put_back_sequence_number
1119                         = channels[c.channelnum].putBackSequenceNumber(
1120                                 (initial_sequence_number+toadd.size() % (SEQNUM_MAX+1)));
1121
1122                 FATAL_ERROR_IF(!successfully_put_back_sequence_number, "error");
1123         }
1124
1125         LOG(dout_con<<m_connection->getDesc()
1126                         << " Windowsize exceeded on reliable sending "
1127                         << c.data.getSize() << " bytes"
1128                         << std::endl << "\t\tinitial_sequence_number: "
1129                         << initial_sequence_number
1130                         << std::endl << "\t\tgot at most            : "
1131                         << packets_available << " packets"
1132                         << std::endl << "\t\tpackets queued         : "
1133                         << channels[c.channelnum].outgoing_reliables_sent.size()
1134                         << std::endl);
1135
1136         return false;
1137 }
1138
1139 void UDPPeer::RunCommandQueues(
1140                                                         unsigned int max_packet_size,
1141                                                         unsigned int maxcommands,
1142                                                         unsigned int maxtransfer)
1143 {
1144
1145         for (Channel &channel : channels) {
1146                 unsigned int commands_processed = 0;
1147
1148                 if ((!channel.queued_commands.empty()) &&
1149                                 (channel.queued_reliables.size() < maxtransfer) &&
1150                                 (commands_processed < maxcommands)) {
1151                         try {
1152                                 ConnectionCommand c = channel.queued_commands.front();
1153
1154                                 LOG(dout_con << m_connection->getDesc()
1155                                                 << " processing queued reliable command " << std::endl);
1156
1157                                 // Packet is processed, remove it from queue
1158                                 if (processReliableSendCommand(c,max_packet_size)) {
1159                                         channel.queued_commands.pop_front();
1160                                 } else {
1161                                         LOG(dout_con << m_connection->getDesc()
1162                                                         << " Failed to queue packets for peer_id: " << c.peer_id
1163                                                         << ", delaying sending of " << c.data.getSize()
1164                                                         << " bytes" << std::endl);
1165                                 }
1166                         }
1167                         catch (ItemNotFoundException &e) {
1168                                 // intentionally empty
1169                         }
1170                 }
1171         }
1172 }
1173
1174 u16 UDPPeer::getNextSplitSequenceNumber(u8 channel)
1175 {
1176         assert(channel < CHANNEL_COUNT); // Pre-condition
1177         return channels[channel].readNextSplitSeqNum();
1178 }
1179
1180 void UDPPeer::setNextSplitSequenceNumber(u8 channel, u16 seqnum)
1181 {
1182         assert(channel < CHANNEL_COUNT); // Pre-condition
1183         channels[channel].setNextSplitSeqNum(seqnum);
1184 }
1185
1186 SharedBuffer<u8> UDPPeer::addSpiltPacket(u8 channel,
1187                                                                                         BufferedPacket toadd,
1188                                                                                         bool reliable)
1189 {
1190         assert(channel < CHANNEL_COUNT); // Pre-condition
1191         return channels[channel].incoming_splits.insert(toadd,reliable);
1192 }
1193
1194 /******************************************************************************/
1195 /* Connection Threads                                                         */
1196 /******************************************************************************/
1197
1198 ConnectionSendThread::ConnectionSendThread(unsigned int max_packet_size,
1199                 float timeout) :
1200         Thread("ConnectionSend"),
1201         m_max_packet_size(max_packet_size),
1202         m_timeout(timeout),
1203         m_max_data_packets_per_iteration(g_settings->getU16("max_packets_per_iteration"))
1204 {
1205 }
1206
1207 void * ConnectionSendThread::run()
1208 {
1209         assert(m_connection);
1210
1211         LOG(dout_con<<m_connection->getDesc()
1212                         <<"ConnectionSend thread started"<<std::endl);
1213
1214         u64 curtime = porting::getTimeMs();
1215         u64 lasttime = curtime;
1216
1217         PROFILE(std::stringstream ThreadIdentifier);
1218         PROFILE(ThreadIdentifier << "ConnectionSend: [" << m_connection->getDesc() << "]");
1219
1220         /* if stop is requested don't stop immediately but try to send all        */
1221         /* packets first */
1222         while(!stopRequested() || packetsQueued()) {
1223                 BEGIN_DEBUG_EXCEPTION_HANDLER
1224                 PROFILE(ScopeProfiler sp(g_profiler, ThreadIdentifier.str(), SPT_AVG));
1225
1226                 m_iteration_packets_avaialble = m_max_data_packets_per_iteration;
1227
1228                 /* wait for trigger or timeout */
1229                 m_send_sleep_semaphore.wait(50);
1230
1231                 /* remove all triggers */
1232                 while(m_send_sleep_semaphore.wait(0)) {}
1233
1234                 lasttime = curtime;
1235                 curtime = porting::getTimeMs();
1236                 float dtime = CALC_DTIME(lasttime,curtime);
1237
1238                 /* first do all the reliable stuff */
1239                 runTimeouts(dtime);
1240
1241                 /* translate commands to packets */
1242                 ConnectionCommand c = m_connection->m_command_queue.pop_frontNoEx(0);
1243                 while(c.type != CONNCMD_NONE)
1244                                 {
1245                         if (c.reliable)
1246                                 processReliableCommand(c);
1247                         else
1248                                 processNonReliableCommand(c);
1249
1250                         c = m_connection->m_command_queue.pop_frontNoEx(0);
1251                 }
1252
1253                 /* send non reliable packets */
1254                 sendPackets(dtime);
1255
1256                 END_DEBUG_EXCEPTION_HANDLER
1257         }
1258
1259         PROFILE(g_profiler->remove(ThreadIdentifier.str()));
1260         return NULL;
1261 }
1262
1263 void ConnectionSendThread::Trigger()
1264 {
1265         m_send_sleep_semaphore.post();
1266 }
1267
1268 bool ConnectionSendThread::packetsQueued()
1269 {
1270         std::list<u16> peerIds = m_connection->getPeerIDs();
1271
1272         if (!m_outgoing_queue.empty() && !peerIds.empty())
1273                 return true;
1274
1275         for (u16 peerId : peerIds) {
1276                 PeerHelper peer = m_connection->getPeerNoEx(peerId);
1277
1278                 if (!peer)
1279                         continue;
1280
1281                 if (dynamic_cast<UDPPeer*>(&peer) == 0)
1282                         continue;
1283
1284                 for (Channel &channel : (dynamic_cast<UDPPeer *>(&peer))->channels) {
1285                         if (!channel.queued_commands.empty()) {
1286                                 return true;
1287                         }
1288                 }
1289         }
1290
1291
1292         return false;
1293 }
1294
1295 void ConnectionSendThread::runTimeouts(float dtime)
1296 {
1297         std::list<u16> timeouted_peers;
1298         std::list<u16> peerIds = m_connection->getPeerIDs();
1299
1300         for (u16 &peerId : peerIds) {
1301                 PeerHelper peer = m_connection->getPeerNoEx(peerId);
1302
1303                 if (!peer)
1304                         continue;
1305
1306                 if (dynamic_cast<UDPPeer*>(&peer) == 0)
1307                         continue;
1308
1309                 PROFILE(std::stringstream peerIdentifier);
1310                 PROFILE(peerIdentifier << "runTimeouts[" << m_connection->getDesc()
1311                                 << ";" << peerId << ";RELIABLE]");
1312                 PROFILE(ScopeProfiler peerprofiler(g_profiler, peerIdentifier.str(), SPT_AVG));
1313
1314                 SharedBuffer<u8> data(2); // data for sending ping, required here because of goto
1315
1316                 /*
1317                         Check peer timeout
1318                 */
1319                 if (peer->isTimedOut(m_timeout))
1320                 {
1321                         infostream<<m_connection->getDesc()
1322                                         <<"RunTimeouts(): Peer "<<peer->id
1323                                         <<" has timed out."
1324                                         <<" (source=peer->timeout_counter)"
1325                                         <<std::endl;
1326                         // Add peer to the list
1327                         timeouted_peers.push_back(peer->id);
1328                         // Don't bother going through the buffers of this one
1329                         continue;
1330                 }
1331
1332                 float resend_timeout = dynamic_cast<UDPPeer*>(&peer)->getResendTimeout();
1333                 bool retry_count_exceeded = false;
1334                 for (Channel &channel : (dynamic_cast<UDPPeer *>(&peer))->channels) {
1335                         std::list<BufferedPacket> timed_outs;
1336
1337                         if (dynamic_cast<UDPPeer*>(&peer)->getLegacyPeer())
1338                                 channel.setWindowSize(g_settings->getU16("workaround_window_size"));
1339
1340                         // Remove timed out incomplete unreliable split packets
1341                         channel.incoming_splits.removeUnreliableTimedOuts(dtime, m_timeout);
1342
1343                         // Increment reliable packet times
1344                         channel.outgoing_reliables_sent.incrementTimeouts(dtime);
1345
1346                         unsigned int numpeers = m_connection->m_peers.size();
1347
1348                         if (numpeers == 0)
1349                                 return;
1350
1351                         // Re-send timed out outgoing reliables
1352                         timed_outs = channel.outgoing_reliables_sent.getTimedOuts(resend_timeout,
1353                                         (m_max_data_packets_per_iteration/numpeers));
1354
1355                         channel.UpdatePacketLossCounter(timed_outs.size());
1356                         g_profiler->graphAdd("packets_lost", timed_outs.size());
1357
1358                         m_iteration_packets_avaialble -= timed_outs.size();
1359
1360                         for(std::list<BufferedPacket>::iterator k = timed_outs.begin();
1361                                 k != timed_outs.end(); ++k)
1362                         {
1363                                 u16 peer_id = readPeerId(*(k->data));
1364                                 u8 channelnum  = readChannel(*(k->data));
1365                                 u16 seqnum  = readU16(&(k->data[BASE_HEADER_SIZE+1]));
1366
1367                                 channel.UpdateBytesLost(k->data.getSize());
1368                                 k->resend_count++;
1369
1370                                 if (k-> resend_count > MAX_RELIABLE_RETRY) {
1371                                         retry_count_exceeded = true;
1372                                         timeouted_peers.push_back(peer->id);
1373                                         /* no need to check additional packets if a single one did timeout*/
1374                                         break;
1375                                 }
1376
1377                                 LOG(derr_con<<m_connection->getDesc()
1378                                                 <<"RE-SENDING timed-out RELIABLE to "
1379                                                 << k->address.serializeString()
1380                                                 << "(t/o="<<resend_timeout<<"): "
1381                                                 <<"from_peer_id="<<peer_id
1382                                                 <<", channel="<<((int)channelnum&0xff)
1383                                                 <<", seqnum="<<seqnum
1384                                                 <<std::endl);
1385
1386                                 rawSend(*k);
1387
1388                                 // do not handle rtt here as we can't decide if this packet was
1389                                 // lost or really takes more time to transmit
1390                         }
1391
1392                         if (retry_count_exceeded) {
1393                                 break; /* no need to check other channels if we already did timeout */
1394                         }
1395
1396                         channel.UpdateTimers(dtime,dynamic_cast<UDPPeer*>(&peer)->getLegacyPeer());
1397                 }
1398
1399                 /* skip to next peer if we did timeout */
1400                 if (retry_count_exceeded)
1401                         continue;
1402
1403                 /* send ping if necessary */
1404                 if (dynamic_cast<UDPPeer*>(&peer)->Ping(dtime,data)) {
1405                         LOG(dout_con<<m_connection->getDesc()
1406                                         <<"Sending ping for peer_id: "
1407                                         << dynamic_cast<UDPPeer*>(&peer)->id <<std::endl);
1408                         /* this may fail if there ain't a sequence number left */
1409                         if (!rawSendAsPacket(dynamic_cast<UDPPeer*>(&peer)->id, 0, data, true))
1410                         {
1411                                 //retrigger with reduced ping interval
1412                                 dynamic_cast<UDPPeer*>(&peer)->Ping(4.0,data);
1413                         }
1414                 }
1415
1416                 dynamic_cast<UDPPeer*>(&peer)->RunCommandQueues(m_max_packet_size,
1417                                                                 m_max_commands_per_iteration,
1418                                                                 m_max_packets_requeued);
1419         }
1420
1421         // Remove timed out peers
1422         for (u16 timeouted_peer : timeouted_peers) {
1423                 LOG(derr_con << m_connection->getDesc()
1424                                 << "RunTimeouts(): Removing peer "<< timeouted_peer <<std::endl);
1425                 m_connection->deletePeer(timeouted_peer, true);
1426         }
1427 }
1428
1429 void ConnectionSendThread::rawSend(const BufferedPacket &packet)
1430 {
1431         try{
1432                 m_connection->m_udpSocket.Send(packet.address, *packet.data,
1433                                 packet.data.getSize());
1434                 LOG(dout_con <<m_connection->getDesc()
1435                                 << " rawSend: " << packet.data.getSize()
1436                                 << " bytes sent" << std::endl);
1437         } catch(SendFailedException &e) {
1438                 LOG(derr_con<<m_connection->getDesc()
1439                                 <<"Connection::rawSend(): SendFailedException: "
1440                                 <<packet.address.serializeString()<<std::endl);
1441         }
1442 }
1443
1444 void ConnectionSendThread::sendAsPacketReliable(BufferedPacket& p, Channel* channel)
1445 {
1446         try{
1447                 p.absolute_send_time = porting::getTimeMs();
1448                 // Buffer the packet
1449                 channel->outgoing_reliables_sent.insert(p,
1450                         (channel->readOutgoingSequenceNumber() - MAX_RELIABLE_WINDOW_SIZE)
1451                         % (MAX_RELIABLE_WINDOW_SIZE+1));
1452         }
1453         catch(AlreadyExistsException &e)
1454         {
1455                 LOG(derr_con<<m_connection->getDesc()
1456                                 <<"WARNING: Going to send a reliable packet"
1457                                 <<" in outgoing buffer" <<std::endl);
1458         }
1459
1460         // Send the packet
1461         rawSend(p);
1462 }
1463
1464 bool ConnectionSendThread::rawSendAsPacket(u16 peer_id, u8 channelnum,
1465                 SharedBuffer<u8> data, bool reliable)
1466 {
1467         PeerHelper peer = m_connection->getPeerNoEx(peer_id);
1468         if (!peer) {
1469                 LOG(dout_con<<m_connection->getDesc()
1470                                 <<" INFO: dropped packet for non existent peer_id: "
1471                                 << peer_id << std::endl);
1472                 FATAL_ERROR_IF(!reliable, "Trying to send raw packet reliable but no peer found!");
1473                 return false;
1474         }
1475         Channel *channel = &(dynamic_cast<UDPPeer*>(&peer)->channels[channelnum]);
1476
1477         if (reliable)
1478         {
1479                 bool have_sequence_number_for_raw_packet = true;
1480                 u16 seqnum =
1481                                 channel->getOutgoingSequenceNumber(have_sequence_number_for_raw_packet);
1482
1483                 if (!have_sequence_number_for_raw_packet)
1484                         return false;
1485
1486                 SharedBuffer<u8> reliable = makeReliablePacket(data, seqnum);
1487                 Address peer_address;
1488                 peer->getAddress(MTP_MINETEST_RELIABLE_UDP, peer_address);
1489
1490                 // Add base headers and make a packet
1491                 BufferedPacket p = con::makePacket(peer_address, reliable,
1492                                 m_connection->GetProtocolID(), m_connection->GetPeerID(),
1493                                 channelnum);
1494
1495                 // first check if our send window is already maxed out
1496                 if (channel->outgoing_reliables_sent.size()
1497                                 < channel->getWindowSize()) {
1498                         LOG(dout_con<<m_connection->getDesc()
1499                                         <<" INFO: sending a reliable packet to peer_id " << peer_id
1500                                         <<" channel: " << channelnum
1501                                         <<" seqnum: " << seqnum << std::endl);
1502                         sendAsPacketReliable(p,channel);
1503                         return true;
1504                 }
1505
1506                 LOG(dout_con<<m_connection->getDesc()
1507                                 <<" INFO: queueing reliable packet for peer_id: " << peer_id
1508                                 <<" channel: " << channelnum
1509                                 <<" seqnum: " << seqnum << std::endl);
1510                 channel->queued_reliables.push(p);
1511                 return false;
1512         }
1513
1514         Address peer_address;
1515         if (peer->getAddress(MTP_UDP, peer_address)) {
1516                 // Add base headers and make a packet
1517                 BufferedPacket p = con::makePacket(peer_address, data,
1518                                 m_connection->GetProtocolID(), m_connection->GetPeerID(),
1519                                 channelnum);
1520
1521                 // Send the packet
1522                 rawSend(p);
1523                 return true;
1524         }
1525
1526         LOG(dout_con << m_connection->getDesc()
1527                         << " INFO: dropped unreliable packet for peer_id: " << peer_id
1528                         << " because of (yet) missing udp address" << std::endl);
1529         return false;
1530 }
1531
1532 void ConnectionSendThread::processReliableCommand(ConnectionCommand &c)
1533 {
1534         assert(c.reliable);  // Pre-condition
1535
1536         switch(c.type) {
1537         case CONNCMD_NONE:
1538                 LOG(dout_con<<m_connection->getDesc()
1539                                 <<"UDP processing reliable CONNCMD_NONE"<<std::endl);
1540                 return;
1541
1542         case CONNCMD_SEND:
1543                 LOG(dout_con<<m_connection->getDesc()
1544                                 <<"UDP processing reliable CONNCMD_SEND"<<std::endl);
1545                 sendReliable(c);
1546                 return;
1547
1548         case CONNCMD_SEND_TO_ALL:
1549                 LOG(dout_con<<m_connection->getDesc()
1550                                 <<"UDP processing CONNCMD_SEND_TO_ALL"<<std::endl);
1551                 sendToAllReliable(c);
1552                 return;
1553
1554         case CONCMD_CREATE_PEER:
1555                 LOG(dout_con<<m_connection->getDesc()
1556                                 <<"UDP processing reliable CONCMD_CREATE_PEER"<<std::endl);
1557                 if (!rawSendAsPacket(c.peer_id,c.channelnum,c.data,c.reliable))
1558                 {
1559                         /* put to queue if we couldn't send it immediately */
1560                         sendReliable(c);
1561                 }
1562                 return;
1563
1564         case CONCMD_DISABLE_LEGACY:
1565                 LOG(dout_con<<m_connection->getDesc()
1566                                 <<"UDP processing reliable CONCMD_DISABLE_LEGACY"<<std::endl);
1567                 if (!rawSendAsPacket(c.peer_id,c.channelnum,c.data,c.reliable))
1568                 {
1569                         /* put to queue if we couldn't send it immediately */
1570                         sendReliable(c);
1571                 }
1572                 return;
1573
1574         case CONNCMD_SERVE:
1575         case CONNCMD_CONNECT:
1576         case CONNCMD_DISCONNECT:
1577         case CONCMD_ACK:
1578                 FATAL_ERROR("Got command that shouldn't be reliable as reliable command");
1579         default:
1580                 LOG(dout_con<<m_connection->getDesc()
1581                                 <<" Invalid reliable command type: " << c.type <<std::endl);
1582         }
1583 }
1584
1585
1586 void ConnectionSendThread::processNonReliableCommand(ConnectionCommand &c)
1587 {
1588         assert(!c.reliable); // Pre-condition
1589
1590         switch(c.type) {
1591         case CONNCMD_NONE:
1592                 LOG(dout_con<<m_connection->getDesc()
1593                                 <<" UDP processing CONNCMD_NONE"<<std::endl);
1594                 return;
1595         case CONNCMD_SERVE:
1596                 LOG(dout_con<<m_connection->getDesc()
1597                                 <<" UDP processing CONNCMD_SERVE port="
1598                                 <<c.address.serializeString()<<std::endl);
1599                 serve(c.address);
1600                 return;
1601         case CONNCMD_CONNECT:
1602                 LOG(dout_con<<m_connection->getDesc()
1603                                 <<" UDP processing CONNCMD_CONNECT"<<std::endl);
1604                 connect(c.address);
1605                 return;
1606         case CONNCMD_DISCONNECT:
1607                 LOG(dout_con<<m_connection->getDesc()
1608                                 <<" UDP processing CONNCMD_DISCONNECT"<<std::endl);
1609                 disconnect();
1610                 return;
1611         case CONNCMD_DISCONNECT_PEER:
1612                 LOG(dout_con<<m_connection->getDesc()
1613                                 <<" UDP processing CONNCMD_DISCONNECT_PEER"<<std::endl);
1614                 disconnect_peer(c.peer_id);
1615                 return;
1616         case CONNCMD_SEND:
1617                 LOG(dout_con<<m_connection->getDesc()
1618                                 <<" UDP processing CONNCMD_SEND"<<std::endl);
1619                 send(c.peer_id, c.channelnum, c.data);
1620                 return;
1621         case CONNCMD_SEND_TO_ALL:
1622                 LOG(dout_con<<m_connection->getDesc()
1623                                 <<" UDP processing CONNCMD_SEND_TO_ALL"<<std::endl);
1624                 sendToAll(c.channelnum, c.data);
1625                 return;
1626         case CONCMD_ACK:
1627                 LOG(dout_con<<m_connection->getDesc()
1628                                 <<" UDP processing CONCMD_ACK"<<std::endl);
1629                 sendAsPacket(c.peer_id,c.channelnum,c.data,true);
1630                 return;
1631         case CONCMD_CREATE_PEER:
1632                 FATAL_ERROR("Got command that should be reliable as unreliable command");
1633         default:
1634                 LOG(dout_con<<m_connection->getDesc()
1635                                 <<" Invalid command type: " << c.type <<std::endl);
1636         }
1637 }
1638
1639 void ConnectionSendThread::serve(Address bind_address)
1640 {
1641         LOG(dout_con<<m_connection->getDesc()
1642                         <<"UDP serving at port " << bind_address.serializeString() <<std::endl);
1643         try{
1644                 m_connection->m_udpSocket.Bind(bind_address);
1645                 m_connection->SetPeerID(PEER_ID_SERVER);
1646         }
1647         catch(SocketException &e) {
1648                 // Create event
1649                 ConnectionEvent ce;
1650                 ce.bindFailed();
1651                 m_connection->putEvent(ce);
1652         }
1653 }
1654
1655 void ConnectionSendThread::connect(Address address)
1656 {
1657         LOG(dout_con<<m_connection->getDesc()<<" connecting to "<<address.serializeString()
1658                         <<":"<<address.getPort()<<std::endl);
1659
1660         UDPPeer *peer = m_connection->createServerPeer(address);
1661
1662         // Create event
1663         ConnectionEvent e;
1664         e.peerAdded(peer->id, peer->address);
1665         m_connection->putEvent(e);
1666
1667         Address bind_addr;
1668
1669         if (address.isIPv6())
1670                 bind_addr.setAddress((IPv6AddressBytes*) NULL);
1671         else
1672                 bind_addr.setAddress(0,0,0,0);
1673
1674         m_connection->m_udpSocket.Bind(bind_addr);
1675
1676         // Send a dummy packet to server with peer_id = PEER_ID_INEXISTENT
1677         m_connection->SetPeerID(PEER_ID_INEXISTENT);
1678         NetworkPacket pkt(0,0);
1679         m_connection->Send(PEER_ID_SERVER, 0, &pkt, true);
1680 }
1681
1682 void ConnectionSendThread::disconnect()
1683 {
1684         LOG(dout_con<<m_connection->getDesc()<<" disconnecting"<<std::endl);
1685
1686         // Create and send DISCO packet
1687         SharedBuffer<u8> data(2);
1688         writeU8(&data[0], TYPE_CONTROL);
1689         writeU8(&data[1], CONTROLTYPE_DISCO);
1690
1691
1692         // Send to all
1693         std::list<u16> peerids = m_connection->getPeerIDs();
1694
1695         for (u16 peerid : peerids) {
1696                 sendAsPacket(peerid, 0,data,false);
1697         }
1698 }
1699
1700 void ConnectionSendThread::disconnect_peer(u16 peer_id)
1701 {
1702         LOG(dout_con<<m_connection->getDesc()<<" disconnecting peer"<<std::endl);
1703
1704         // Create and send DISCO packet
1705         SharedBuffer<u8> data(2);
1706         writeU8(&data[0], TYPE_CONTROL);
1707         writeU8(&data[1], CONTROLTYPE_DISCO);
1708         sendAsPacket(peer_id, 0,data,false);
1709
1710         PeerHelper peer = m_connection->getPeerNoEx(peer_id);
1711
1712         if (!peer)
1713                 return;
1714
1715         if (dynamic_cast<UDPPeer*>(&peer) == 0)
1716         {
1717                 return;
1718         }
1719
1720         dynamic_cast<UDPPeer*>(&peer)->m_pending_disconnect = true;
1721 }
1722
1723 void ConnectionSendThread::send(u16 peer_id, u8 channelnum,
1724                 SharedBuffer<u8> data)
1725 {
1726         assert(channelnum < CHANNEL_COUNT); // Pre-condition
1727
1728         PeerHelper peer = m_connection->getPeerNoEx(peer_id);
1729         if (!peer)
1730         {
1731                 LOG(dout_con<<m_connection->getDesc()<<" peer: peer_id="<<peer_id
1732                                 << ">>>NOT<<< found on sending packet"
1733                                 << ", channel " << (channelnum % 0xFF)
1734                                 << ", size: " << data.getSize() <<std::endl);
1735                 return;
1736         }
1737
1738         LOG(dout_con<<m_connection->getDesc()<<" sending to peer_id="<<peer_id
1739                         << ", channel " << (channelnum % 0xFF)
1740                         << ", size: " << data.getSize() <<std::endl);
1741
1742         u16 split_sequence_number = peer->getNextSplitSequenceNumber(channelnum);
1743
1744         u32 chunksize_max = m_max_packet_size - BASE_HEADER_SIZE;
1745         std::list<SharedBuffer<u8> > originals;
1746
1747         originals = makeAutoSplitPacket(data, chunksize_max,split_sequence_number);
1748
1749         peer->setNextSplitSequenceNumber(channelnum,split_sequence_number);
1750
1751         for (const SharedBuffer<u8> &original : originals) {
1752                 sendAsPacket(peer_id, channelnum, original);
1753         }
1754 }
1755
1756 void ConnectionSendThread::sendReliable(ConnectionCommand &c)
1757 {
1758         PeerHelper peer = m_connection->getPeerNoEx(c.peer_id);
1759         if (!peer)
1760                 return;
1761
1762         peer->PutReliableSendCommand(c,m_max_packet_size);
1763 }
1764
1765 void ConnectionSendThread::sendToAll(u8 channelnum, SharedBuffer<u8> data)
1766 {
1767         std::list<u16> peerids = m_connection->getPeerIDs();
1768
1769         for (u16 peerid : peerids) {
1770                 send(peerid, channelnum, data);
1771         }
1772 }
1773
1774 void ConnectionSendThread::sendToAllReliable(ConnectionCommand &c)
1775 {
1776         std::list<u16> peerids = m_connection->getPeerIDs();
1777
1778         for (u16 peerid : peerids) {
1779                 PeerHelper peer = m_connection->getPeerNoEx(peerid);
1780
1781                 if (!peer)
1782                         continue;
1783
1784                 peer->PutReliableSendCommand(c,m_max_packet_size);
1785         }
1786 }
1787
1788 void ConnectionSendThread::sendPackets(float dtime)
1789 {
1790         std::list<u16> peerIds = m_connection->getPeerIDs();
1791         std::list<u16> pendingDisconnect;
1792         std::map<u16,bool> pending_unreliable;
1793
1794         for (u16 peerId : peerIds) {
1795                 PeerHelper peer = m_connection->getPeerNoEx(peerId);
1796                 //peer may have been removed
1797                 if (!peer) {
1798                         LOG(dout_con<<m_connection->getDesc()<< " Peer not found: peer_id=" << peerId
1799                                 << std::endl);
1800                         continue;
1801                 }
1802                 peer->m_increment_packets_remaining = m_iteration_packets_avaialble/m_connection->m_peers.size();
1803
1804                 UDPPeer *udpPeer = dynamic_cast<UDPPeer*>(&peer);
1805
1806                 if (!udpPeer) {
1807                         continue;
1808                 }
1809
1810                 if (udpPeer->m_pending_disconnect) {
1811                         pendingDisconnect.push_back(peerId);
1812                 }
1813
1814                 PROFILE(std::stringstream peerIdentifier);
1815                 PROFILE(peerIdentifier << "sendPackets[" << m_connection->getDesc() << ";" << peerId
1816                         << ";RELIABLE]");
1817                 PROFILE(ScopeProfiler peerprofiler(g_profiler, peerIdentifier.str(), SPT_AVG));
1818
1819                 LOG(dout_con<<m_connection->getDesc()
1820                                 << " Handle per peer queues: peer_id=" << peerId
1821                         << " packet quota: " << peer->m_increment_packets_remaining << std::endl);
1822
1823                 // first send queued reliable packets for all peers (if possible)
1824                 for (unsigned int i=0; i < CHANNEL_COUNT; i++) {
1825                         Channel &channel = udpPeer->channels[i];
1826                         u16 next_to_ack = 0;
1827
1828                         channel.outgoing_reliables_sent.getFirstSeqnum(next_to_ack);
1829                         u16 next_to_receive = 0;
1830                         channel.incoming_reliables.getFirstSeqnum(next_to_receive);
1831
1832                         LOG(dout_con<<m_connection->getDesc()<< "\t channel: "
1833                                                 << i << ", peer quota:"
1834                                                 << peer->m_increment_packets_remaining
1835                                                 << std::endl
1836                                         << "\t\t\treliables on wire: "
1837                                                 << channel.outgoing_reliables_sent.size()
1838                                                 << ", waiting for ack for " << next_to_ack
1839                                                 << std::endl
1840                                         << "\t\t\tincoming_reliables: "
1841                                                 << channel.incoming_reliables.size()
1842                                                 << ", next reliable packet: "
1843                                                 << channel.readNextIncomingSeqNum()
1844                                                 << ", next queued: " << next_to_receive
1845                                                 << std::endl
1846                                         << "\t\t\treliables queued : "
1847                                                 << channel.queued_reliables.size()
1848                                                 << std::endl
1849                                         << "\t\t\tqueued commands  : "
1850                                                 << channel.queued_commands.size()
1851                                                 << std::endl);
1852
1853                         while ((!channel.queued_reliables.empty()) &&
1854                                         (channel.outgoing_reliables_sent.size()
1855                                                         < channel.getWindowSize())&&
1856                                                         (peer->m_increment_packets_remaining > 0))
1857                         {
1858                                 BufferedPacket p = channel.queued_reliables.front();
1859                                 channel.queued_reliables.pop();
1860                                 LOG(dout_con<<m_connection->getDesc()
1861                                                 <<" INFO: sending a queued reliable packet "
1862                                                 <<" channel: " << i
1863                                                 <<", seqnum: " << readU16(&p.data[BASE_HEADER_SIZE+1])
1864                                                 << std::endl);
1865                                 sendAsPacketReliable(p, &channel);
1866                                 peer->m_increment_packets_remaining--;
1867                         }
1868                 }
1869         }
1870
1871         if (!m_outgoing_queue.empty()) {
1872                 LOG(dout_con<<m_connection->getDesc()
1873                                 << " Handle non reliable queue ("
1874                                 << m_outgoing_queue.size() << " pkts)" << std::endl);
1875         }
1876
1877         unsigned int initial_queuesize = m_outgoing_queue.size();
1878         /* send non reliable packets*/
1879         for(unsigned int i=0;i < initial_queuesize;i++) {
1880                 OutgoingPacket packet = m_outgoing_queue.front();
1881                 m_outgoing_queue.pop();
1882
1883                 if (packet.reliable)
1884                         continue;
1885
1886                 PeerHelper peer = m_connection->getPeerNoEx(packet.peer_id);
1887                 if (!peer) {
1888                         LOG(dout_con<<m_connection->getDesc()
1889                                                         <<" Outgoing queue: peer_id="<<packet.peer_id
1890                                                         << ">>>NOT<<< found on sending packet"
1891                                                         << ", channel " << (packet.channelnum % 0xFF)
1892                                                         << ", size: " << packet.data.getSize() <<std::endl);
1893                         continue;
1894                 }
1895
1896                 /* send acks immediately */
1897                 if (packet.ack) {
1898                         rawSendAsPacket(packet.peer_id, packet.channelnum,
1899                                                                 packet.data, packet.reliable);
1900                         peer->m_increment_packets_remaining =
1901                                         MYMIN(0,peer->m_increment_packets_remaining--);
1902                 }
1903                 else if (
1904                         ( peer->m_increment_packets_remaining > 0) ||
1905                         (stopRequested())) {
1906                         rawSendAsPacket(packet.peer_id, packet.channelnum,
1907                                         packet.data, packet.reliable);
1908                         peer->m_increment_packets_remaining--;
1909                 }
1910                 else {
1911                         m_outgoing_queue.push(packet);
1912                         pending_unreliable[packet.peer_id] = true;
1913                 }
1914         }
1915
1916         for (u16 peerId : pendingDisconnect) {
1917                 if (!pending_unreliable[peerId])
1918                 {
1919                         m_connection->deletePeer(peerId,false);
1920                 }
1921         }
1922 }
1923
1924 void ConnectionSendThread::sendAsPacket(u16 peer_id, u8 channelnum,
1925                 SharedBuffer<u8> data, bool ack)
1926 {
1927         OutgoingPacket packet(peer_id, channelnum, data, false, ack);
1928         m_outgoing_queue.push(packet);
1929 }
1930
1931 ConnectionReceiveThread::ConnectionReceiveThread(unsigned int max_packet_size) :
1932         Thread("ConnectionReceive")
1933 {
1934 }
1935
1936 void * ConnectionReceiveThread::run()
1937 {
1938         assert(m_connection);
1939
1940         LOG(dout_con<<m_connection->getDesc()
1941                         <<"ConnectionReceive thread started"<<std::endl);
1942
1943         PROFILE(std::stringstream ThreadIdentifier);
1944         PROFILE(ThreadIdentifier << "ConnectionReceive: [" << m_connection->getDesc() << "]");
1945
1946 #ifdef DEBUG_CONNECTION_KBPS
1947         u64 curtime = porting::getTimeMs();
1948         u64 lasttime = curtime;
1949         float debug_print_timer = 0.0;
1950 #endif
1951
1952         while(!stopRequested()) {
1953                 BEGIN_DEBUG_EXCEPTION_HANDLER
1954                 PROFILE(ScopeProfiler sp(g_profiler, ThreadIdentifier.str(), SPT_AVG));
1955
1956 #ifdef DEBUG_CONNECTION_KBPS
1957                 lasttime = curtime;
1958                 curtime = porting::getTimeMs();
1959                 float dtime = CALC_DTIME(lasttime,curtime);
1960 #endif
1961
1962                 /* receive packets */
1963                 receive();
1964
1965 #ifdef DEBUG_CONNECTION_KBPS
1966                 debug_print_timer += dtime;
1967                 if (debug_print_timer > 20.0) {
1968                         debug_print_timer -= 20.0;
1969
1970                         std::list<u16> peerids = m_connection->getPeerIDs();
1971
1972                         for (std::list<u16>::iterator i = peerids.begin();
1973                                         i != peerids.end();
1974                                         i++)
1975                         {
1976                                 PeerHelper peer = m_connection->getPeerNoEx(*i);
1977                                 if (!peer)
1978                                         continue;
1979
1980                                 float peer_current = 0.0;
1981                                 float peer_loss = 0.0;
1982                                 float avg_rate = 0.0;
1983                                 float avg_loss = 0.0;
1984
1985                                 for(u16 j=0; j<CHANNEL_COUNT; j++)
1986                                 {
1987                                         peer_current +=peer->channels[j].getCurrentDownloadRateKB();
1988                                         peer_loss += peer->channels[j].getCurrentLossRateKB();
1989                                         avg_rate += peer->channels[j].getAvgDownloadRateKB();
1990                                         avg_loss += peer->channels[j].getAvgLossRateKB();
1991                                 }
1992
1993                                 std::stringstream output;
1994                                 output << std::fixed << std::setprecision(1);
1995                                 output << "OUT to Peer " << *i << " RATES (good / loss) " << std::endl;
1996                                 output << "\tcurrent (sum): " << peer_current << "kb/s "<< peer_loss << "kb/s" << std::endl;
1997                                 output << "\taverage (sum): " << avg_rate << "kb/s "<< avg_loss << "kb/s" << std::endl;
1998                                 output << std::setfill(' ');
1999                                 for(u16 j=0; j<CHANNEL_COUNT; j++)
2000                                 {
2001                                         output << "\tcha " << j << ":"
2002                                                 << " CUR: " << std::setw(6) << peer->channels[j].getCurrentDownloadRateKB() <<"kb/s"
2003                                                 << " AVG: " << std::setw(6) << peer->channels[j].getAvgDownloadRateKB() <<"kb/s"
2004                                                 << " MAX: " << std::setw(6) << peer->channels[j].getMaxDownloadRateKB() <<"kb/s"
2005                                                 << " /"
2006                                                 << " CUR: " << std::setw(6) << peer->channels[j].getCurrentLossRateKB() <<"kb/s"
2007                                                 << " AVG: " << std::setw(6) << peer->channels[j].getAvgLossRateKB() <<"kb/s"
2008                                                 << " MAX: " << std::setw(6) << peer->channels[j].getMaxLossRateKB() <<"kb/s"
2009                                                 << " / WS: " << peer->channels[j].getWindowSize()
2010                                                 << std::endl;
2011                                 }
2012
2013                                 fprintf(stderr,"%s\n",output.str().c_str());
2014                         }
2015                 }
2016 #endif
2017                 END_DEBUG_EXCEPTION_HANDLER
2018         }
2019
2020         PROFILE(g_profiler->remove(ThreadIdentifier.str()));
2021         return NULL;
2022 }
2023
2024 // Receive packets from the network and buffers and create ConnectionEvents
2025 void ConnectionReceiveThread::receive()
2026 {
2027         // use IPv6 minimum allowed MTU as receive buffer size as this is
2028         // theoretical reliable upper boundary of a udp packet for all IPv6 enabled
2029         // infrastructure
2030         unsigned int packet_maxsize = 1500;
2031         SharedBuffer<u8> packetdata(packet_maxsize);
2032
2033         bool packet_queued = true;
2034
2035         unsigned int loop_count = 0;
2036
2037         /* first of all read packets from socket */
2038         /* check for incoming data available */
2039         while( (loop_count < 10) &&
2040                         (m_connection->m_udpSocket.WaitData(50))) {
2041                 loop_count++;
2042                 try {
2043                         if (packet_queued) {
2044                                 bool data_left = true;
2045                                 u16 peer_id;
2046                                 SharedBuffer<u8> resultdata;
2047                                 while(data_left) {
2048                                         try {
2049                                                 data_left = getFromBuffers(peer_id, resultdata);
2050                                                 if (data_left) {
2051                                                         ConnectionEvent e;
2052                                                         e.dataReceived(peer_id, resultdata);
2053                                                         m_connection->putEvent(e);
2054                                                 }
2055                                         }
2056                                         catch(ProcessedSilentlyException &e) {
2057                                                 /* try reading again */
2058                                         }
2059                                 }
2060                                 packet_queued = false;
2061                         }
2062
2063                         Address sender;
2064                         s32 received_size = m_connection->m_udpSocket.Receive(sender, *packetdata, packet_maxsize);
2065
2066                         if ((received_size < BASE_HEADER_SIZE) ||
2067                                 (readU32(&packetdata[0]) != m_connection->GetProtocolID()))
2068                         {
2069                                 LOG(derr_con<<m_connection->getDesc()
2070                                                 <<"Receive(): Invalid incoming packet, "
2071                                                 <<"size: " << received_size
2072                                                 <<", protocol: "
2073                                                 << ((received_size >= 4) ? readU32(&packetdata[0]) : -1)
2074                                                 << std::endl);
2075                                 continue;
2076                         }
2077
2078                         u16 peer_id          = readPeerId(*packetdata);
2079                         u8 channelnum        = readChannel(*packetdata);
2080
2081                         if (channelnum > CHANNEL_COUNT-1) {
2082                                 LOG(derr_con<<m_connection->getDesc()
2083                                                 <<"Receive(): Invalid channel "<<channelnum<<std::endl);
2084                                 throw InvalidIncomingDataException("Channel doesn't exist");
2085                         }
2086
2087                         /* Try to identify peer by sender address (may happen on join) */
2088                         if (peer_id == PEER_ID_INEXISTENT) {
2089                                 peer_id = m_connection->lookupPeer(sender);
2090                                 // We do not have to remind the peer of its
2091                                 // peer id as the CONTROLTYPE_SET_PEER_ID
2092                                 // command was sent reliably.
2093                         }
2094
2095                         /* The peer was not found in our lists. Add it. */
2096                         if (peer_id == PEER_ID_INEXISTENT) {
2097                                 peer_id = m_connection->createPeer(sender, MTP_MINETEST_RELIABLE_UDP, 0);
2098                         }
2099
2100                         PeerHelper peer = m_connection->getPeerNoEx(peer_id);
2101
2102                         if (!peer) {
2103                                 LOG(dout_con<<m_connection->getDesc()
2104                                                 <<" got packet from unknown peer_id: "
2105                                                 <<peer_id<<" Ignoring."<<std::endl);
2106                                 continue;
2107                         }
2108
2109                         // Validate peer address
2110
2111                         Address peer_address;
2112
2113                         if (peer->getAddress(MTP_UDP, peer_address)) {
2114                                 if (peer_address != sender) {
2115                                         LOG(derr_con<<m_connection->getDesc()
2116                                                         <<m_connection->getDesc()
2117                                                         <<" Peer "<<peer_id<<" sending from different address."
2118                                                         " Ignoring."<<std::endl);
2119                                         continue;
2120                                 }
2121                         }
2122                         else {
2123
2124                                 bool invalid_address = true;
2125                                 if (invalid_address) {
2126                                         LOG(derr_con<<m_connection->getDesc()
2127                                                         <<m_connection->getDesc()
2128                                                         <<" Peer "<<peer_id<<" unknown."
2129                                                         " Ignoring."<<std::endl);
2130                                         continue;
2131                                 }
2132                         }
2133
2134                         peer->ResetTimeout();
2135
2136                         Channel *channel = 0;
2137
2138                         if (dynamic_cast<UDPPeer*>(&peer) != 0)
2139                         {
2140                                 channel = &(dynamic_cast<UDPPeer*>(&peer)->channels[channelnum]);
2141                         }
2142
2143                         if (channel != 0) {
2144                                 channel->UpdateBytesReceived(received_size);
2145                         }
2146
2147                         // Throw the received packet to channel->processPacket()
2148
2149                         // Make a new SharedBuffer from the data without the base headers
2150                         SharedBuffer<u8> strippeddata(received_size - BASE_HEADER_SIZE);
2151                         memcpy(*strippeddata, &packetdata[BASE_HEADER_SIZE],
2152                                         strippeddata.getSize());
2153
2154                         try{
2155                                 // Process it (the result is some data with no headers made by us)
2156                                 SharedBuffer<u8> resultdata = processPacket
2157                                                 (channel, strippeddata, peer_id, channelnum, false);
2158
2159                                 LOG(dout_con<<m_connection->getDesc()
2160                                                 <<" ProcessPacket from peer_id: " << peer_id
2161                                                 << ",channel: " << (channelnum & 0xFF) << ", returned "
2162                                                 << resultdata.getSize() << " bytes" <<std::endl);
2163
2164                                 ConnectionEvent e;
2165                                 e.dataReceived(peer_id, resultdata);
2166                                 m_connection->putEvent(e);
2167                         }
2168                         catch(ProcessedSilentlyException &e) {
2169                         }
2170                         catch(ProcessedQueued &e) {
2171                                 packet_queued = true;
2172                         }
2173                 }
2174                 catch(InvalidIncomingDataException &e) {
2175                 }
2176                 catch(ProcessedSilentlyException &e) {
2177                 }
2178         }
2179 }
2180
2181 bool ConnectionReceiveThread::getFromBuffers(u16 &peer_id, SharedBuffer<u8> &dst)
2182 {
2183         std::list<u16> peerids = m_connection->getPeerIDs();
2184
2185         for (u16 peerid : peerids) {
2186                 PeerHelper peer = m_connection->getPeerNoEx(peerid);
2187                 if (!peer)
2188                         continue;
2189
2190                 if (dynamic_cast<UDPPeer*>(&peer) == 0)
2191                         continue;
2192
2193                 for (Channel &channel : (dynamic_cast<UDPPeer *>(&peer))->channels) {
2194                         if (checkIncomingBuffers(&channel, peer_id, dst)) {
2195                                 return true;
2196                         }
2197                 }
2198         }
2199         return false;
2200 }
2201
2202 bool ConnectionReceiveThread::checkIncomingBuffers(Channel *channel,
2203                 u16 &peer_id, SharedBuffer<u8> &dst)
2204 {
2205         u16 firstseqnum = 0;
2206         if (channel->incoming_reliables.getFirstSeqnum(firstseqnum))
2207         {
2208                 if (firstseqnum == channel->readNextIncomingSeqNum())
2209                 {
2210                         BufferedPacket p = channel->incoming_reliables.popFirst();
2211                         peer_id = readPeerId(*p.data);
2212                         u8 channelnum = readChannel(*p.data);
2213                         u16 seqnum = readU16(&p.data[BASE_HEADER_SIZE+1]);
2214
2215                         LOG(dout_con<<m_connection->getDesc()
2216                                         <<"UNBUFFERING TYPE_RELIABLE"
2217                                         <<" seqnum="<<seqnum
2218                                         <<" peer_id="<<peer_id
2219                                         <<" channel="<<((int)channelnum&0xff)
2220                                         <<std::endl);
2221
2222                         channel->incNextIncomingSeqNum();
2223
2224                         u32 headers_size = BASE_HEADER_SIZE + RELIABLE_HEADER_SIZE;
2225                         // Get out the inside packet and re-process it
2226                         SharedBuffer<u8> payload(p.data.getSize() - headers_size);
2227                         memcpy(*payload, &p.data[headers_size], payload.getSize());
2228
2229                         dst = processPacket(channel, payload, peer_id, channelnum, true);
2230                         return true;
2231                 }
2232         }
2233         return false;
2234 }
2235
2236 SharedBuffer<u8> ConnectionReceiveThread::processPacket(Channel *channel,
2237                 SharedBuffer<u8> packetdata, u16 peer_id, u8 channelnum, bool reliable)
2238 {
2239         PeerHelper peer = m_connection->getPeerNoEx(peer_id);
2240
2241         if (!peer) {
2242                 errorstream << "Peer not found (possible timeout)" << std::endl;
2243                 throw ProcessedSilentlyException("Peer not found (possible timeout)");
2244         }
2245
2246         if (packetdata.getSize() < 1)
2247                 throw InvalidIncomingDataException("packetdata.getSize() < 1");
2248
2249         u8 type = readU8(&(packetdata[0]));
2250
2251         if (MAX_UDP_PEERS <= 65535 && peer_id >= MAX_UDP_PEERS) {
2252                 std::string errmsg = "Invalid peer_id=" + itos(peer_id);
2253                 errorstream << errmsg << std::endl;
2254                 throw InvalidIncomingDataException(errmsg.c_str());
2255         }
2256
2257         if (type == TYPE_CONTROL)
2258         {
2259                 if (packetdata.getSize() < 2)
2260                         throw InvalidIncomingDataException("packetdata.getSize() < 2");
2261
2262                 u8 controltype = readU8(&(packetdata[1]));
2263
2264                 if (controltype == CONTROLTYPE_ACK)
2265                 {
2266                         assert(channel != NULL);
2267
2268                         if (packetdata.getSize() < 4) {
2269                                 throw InvalidIncomingDataException(
2270                                         "packetdata.getSize() < 4 (ACK header size)");
2271                         }
2272
2273                         u16 seqnum = readU16(&packetdata[2]);
2274                         LOG(dout_con<<m_connection->getDesc()
2275                                         <<" [ CONTROLTYPE_ACK: channelnum="
2276                                         <<((int)channelnum&0xff)<<", peer_id="<<peer_id
2277                                         <<", seqnum="<<seqnum<< " ]"<<std::endl);
2278
2279                         try{
2280                                 BufferedPacket p =
2281                                                 channel->outgoing_reliables_sent.popSeqnum(seqnum);
2282
2283                                 // only calculate rtt from straight sent packets
2284                                 if (p.resend_count == 0) {
2285                                         // Get round trip time
2286                                         u64 current_time = porting::getTimeMs();
2287
2288                                         // a overflow is quite unlikely but as it'd result in major
2289                                         // rtt miscalculation we handle it here
2290                                         if (current_time > p.absolute_send_time)
2291                                         {
2292                                                 float rtt = (current_time - p.absolute_send_time) / 1000.0;
2293
2294                                                 // Let peer calculate stuff according to it
2295                                                 // (avg_rtt and resend_timeout)
2296                                                 dynamic_cast<UDPPeer*>(&peer)->reportRTT(rtt);
2297                                         }
2298                                         else if (p.totaltime > 0)
2299                                         {
2300                                                 float rtt = p.totaltime;
2301
2302                                                 // Let peer calculate stuff according to it
2303                                                 // (avg_rtt and resend_timeout)
2304                                                 dynamic_cast<UDPPeer*>(&peer)->reportRTT(rtt);
2305                                         }
2306                                 }
2307                                 //put bytes for max bandwidth calculation
2308                                 channel->UpdateBytesSent(p.data.getSize(),1);
2309                                 if (channel->outgoing_reliables_sent.size() == 0)
2310                                 {
2311                                         m_connection->TriggerSend();
2312                                 }
2313                         }
2314                         catch(NotFoundException &e) {
2315                                 LOG(derr_con<<m_connection->getDesc()
2316                                                 <<"WARNING: ACKed packet not "
2317                                                 "in outgoing queue"
2318                                                 <<std::endl);
2319                                 channel->UpdatePacketTooLateCounter();
2320                         }
2321                         throw ProcessedSilentlyException("Got an ACK");
2322                 }
2323                 else if (controltype == CONTROLTYPE_SET_PEER_ID) {
2324                         // Got a packet to set our peer id
2325                         if (packetdata.getSize() < 4)
2326                                 throw InvalidIncomingDataException
2327                                                 ("packetdata.getSize() < 4 (SET_PEER_ID header size)");
2328                         u16 peer_id_new = readU16(&packetdata[2]);
2329                         LOG(dout_con<<m_connection->getDesc()
2330                                         <<"Got new peer id: "<<peer_id_new<<"... "<<std::endl);
2331
2332                         if (m_connection->GetPeerID() != PEER_ID_INEXISTENT)
2333                         {
2334                                 LOG(derr_con<<m_connection->getDesc()
2335                                                 <<"WARNING: Not changing"
2336                                                 " existing peer id."<<std::endl);
2337                         }
2338                         else
2339                         {
2340                                 LOG(dout_con<<m_connection->getDesc()<<"changing own peer id"<<std::endl);
2341                                 m_connection->SetPeerID(peer_id_new);
2342                         }
2343
2344                         ConnectionCommand cmd;
2345
2346                         SharedBuffer<u8> reply(2);
2347                         writeU8(&reply[0], TYPE_CONTROL);
2348                         writeU8(&reply[1], CONTROLTYPE_ENABLE_BIG_SEND_WINDOW);
2349                         cmd.disableLegacy(PEER_ID_SERVER,reply);
2350                         m_connection->putCommand(cmd);
2351
2352                         throw ProcessedSilentlyException("Got a SET_PEER_ID");
2353                 }
2354                 else if (controltype == CONTROLTYPE_PING)
2355                 {
2356                         // Just ignore it, the incoming data already reset
2357                         // the timeout counter
2358                         LOG(dout_con<<m_connection->getDesc()<<"PING"<<std::endl);
2359                         throw ProcessedSilentlyException("Got a PING");
2360                 }
2361                 else if (controltype == CONTROLTYPE_DISCO)
2362                 {
2363                         // Just ignore it, the incoming data already reset
2364                         // the timeout counter
2365                         LOG(dout_con<<m_connection->getDesc()
2366                                         <<"DISCO: Removing peer "<<(peer_id)<<std::endl);
2367
2368                         if (!m_connection->deletePeer(peer_id, false)) {
2369                                 derr_con<<m_connection->getDesc()
2370                                                 <<"DISCO: Peer not found"<<std::endl;
2371                         }
2372
2373                         throw ProcessedSilentlyException("Got a DISCO");
2374                 }
2375                 else if (controltype == CONTROLTYPE_ENABLE_BIG_SEND_WINDOW)
2376                 {
2377                         dynamic_cast<UDPPeer*>(&peer)->setNonLegacyPeer();
2378                         throw ProcessedSilentlyException("Got non legacy control");
2379                 }
2380                 else{
2381                         LOG(derr_con<<m_connection->getDesc()
2382                                         <<"INVALID TYPE_CONTROL: invalid controltype="
2383                                         <<((int)controltype&0xff)<<std::endl);
2384                         throw InvalidIncomingDataException("Invalid control type");
2385                 }
2386         }
2387         else if (type == TYPE_ORIGINAL)
2388         {
2389                 if (packetdata.getSize() <= ORIGINAL_HEADER_SIZE)
2390                         throw InvalidIncomingDataException
2391                                         ("packetdata.getSize() <= ORIGINAL_HEADER_SIZE");
2392                 LOG(dout_con<<m_connection->getDesc()
2393                                 <<"RETURNING TYPE_ORIGINAL to user"
2394                                 <<std::endl);
2395                 // Get the inside packet out and return it
2396                 SharedBuffer<u8> payload(packetdata.getSize() - ORIGINAL_HEADER_SIZE);
2397                 memcpy(*payload, &(packetdata[ORIGINAL_HEADER_SIZE]), payload.getSize());
2398                 return payload;
2399         }
2400         else if (type == TYPE_SPLIT)
2401         {
2402                 Address peer_address;
2403
2404                 if (peer->getAddress(MTP_UDP, peer_address)) {
2405
2406                         // We have to create a packet again for buffering
2407                         // This isn't actually too bad an idea.
2408                         BufferedPacket packet = makePacket(
2409                                         peer_address,
2410                                         packetdata,
2411                                         m_connection->GetProtocolID(),
2412                                         peer_id,
2413                                         channelnum);
2414
2415                         // Buffer the packet
2416                         SharedBuffer<u8> data =
2417                                         peer->addSpiltPacket(channelnum,packet,reliable);
2418
2419                         if (data.getSize() != 0)
2420                         {
2421                                 LOG(dout_con<<m_connection->getDesc()
2422                                                 <<"RETURNING TYPE_SPLIT: Constructed full data, "
2423                                                 <<"size="<<data.getSize()<<std::endl);
2424                                 return data;
2425                         }
2426                         LOG(dout_con<<m_connection->getDesc()<<"BUFFERED TYPE_SPLIT"<<std::endl);
2427                         throw ProcessedSilentlyException("Buffered a split packet chunk");
2428                 }
2429                 else {
2430                         //TODO throw some error
2431                 }
2432         }
2433         else if (type == TYPE_RELIABLE)
2434         {
2435                 assert(channel != NULL);
2436
2437                 // Recursive reliable packets not allowed
2438                 if (reliable)
2439                         throw InvalidIncomingDataException("Found nested reliable packets");
2440
2441                 if (packetdata.getSize() < RELIABLE_HEADER_SIZE)
2442                         throw InvalidIncomingDataException
2443                                         ("packetdata.getSize() < RELIABLE_HEADER_SIZE");
2444
2445                 u16 seqnum = readU16(&packetdata[1]);
2446                 bool is_future_packet = false;
2447                 bool is_old_packet = false;
2448
2449                 /* packet is within our receive window send ack */
2450                 if (seqnum_in_window(seqnum, channel->readNextIncomingSeqNum(),MAX_RELIABLE_WINDOW_SIZE))
2451                 {
2452                         m_connection->sendAck(peer_id,channelnum,seqnum);
2453                 }
2454                 else {
2455                         is_future_packet = seqnum_higher(seqnum, channel->readNextIncomingSeqNum());
2456                         is_old_packet    = seqnum_higher(channel->readNextIncomingSeqNum(), seqnum);
2457
2458
2459                         /* packet is not within receive window, don't send ack.           *
2460                          * if this was a valid packet it's gonna be retransmitted         */
2461                         if (is_future_packet)
2462                         {
2463                                 throw ProcessedSilentlyException("Received packet newer then expected, not sending ack");
2464                         }
2465
2466                         /* seems like our ack was lost, send another one for a old packet */
2467                         if (is_old_packet)
2468                         {
2469                                 LOG(dout_con<<m_connection->getDesc()
2470                                                 << "RE-SENDING ACK: peer_id: " << peer_id
2471                                                 << ", channel: " << (channelnum&0xFF)
2472                                                 << ", seqnum: " << seqnum << std::endl;)
2473                                 m_connection->sendAck(peer_id,channelnum,seqnum);
2474
2475                                 // we already have this packet so this one was on wire at least
2476                                 // the current timeout
2477                                 // we don't know how long this packet was on wire don't do silly guessing
2478                                 // dynamic_cast<UDPPeer*>(&peer)->reportRTT(dynamic_cast<UDPPeer*>(&peer)->getResendTimeout());
2479
2480                                 throw ProcessedSilentlyException("Retransmitting ack for old packet");
2481                         }
2482                 }
2483
2484                 if (seqnum != channel->readNextIncomingSeqNum())
2485                 {
2486                         Address peer_address;
2487
2488                         // this is a reliable packet so we have a udp address for sure
2489                         peer->getAddress(MTP_MINETEST_RELIABLE_UDP, peer_address);
2490                         // This one comes later, buffer it.
2491                         // Actually we have to make a packet to buffer one.
2492                         // Well, we have all the ingredients, so just do it.
2493                         BufferedPacket packet = con::makePacket(
2494                                         peer_address,
2495                                         packetdata,
2496                                         m_connection->GetProtocolID(),
2497                                         peer_id,
2498                                         channelnum);
2499                         try{
2500                                 channel->incoming_reliables.insert(packet,channel->readNextIncomingSeqNum());
2501
2502                                 LOG(dout_con<<m_connection->getDesc()
2503                                                 << "BUFFERING, TYPE_RELIABLE peer_id: " << peer_id
2504                                                 << ", channel: " << (channelnum&0xFF)
2505                                                 << ", seqnum: " << seqnum << std::endl;)
2506
2507                                 throw ProcessedQueued("Buffered future reliable packet");
2508                         }
2509                         catch(AlreadyExistsException &e)
2510                         {
2511                         }
2512                         catch(IncomingDataCorruption &e)
2513                         {
2514                                 ConnectionCommand discon;
2515                                 discon.disconnect_peer(peer_id);
2516                                 m_connection->putCommand(discon);
2517
2518                                 LOG(derr_con<<m_connection->getDesc()
2519                                                 << "INVALID, TYPE_RELIABLE peer_id: " << peer_id
2520                                                 << ", channel: " << (channelnum&0xFF)
2521                                                 << ", seqnum: " << seqnum
2522                                                 << "DROPPING CLIENT!" << std::endl;)
2523                         }
2524                 }
2525
2526                 /* we got a packet to process right now */
2527                 LOG(dout_con<<m_connection->getDesc()
2528                                 << "RECURSIVE, TYPE_RELIABLE peer_id: " << peer_id
2529                                 << ", channel: " << (channelnum&0xFF)
2530                                 << ", seqnum: " << seqnum << std::endl;)
2531
2532
2533                 /* check for resend case */
2534                 u16 queued_seqnum = 0;
2535                 if (channel->incoming_reliables.getFirstSeqnum(queued_seqnum))
2536                 {
2537                         if (queued_seqnum == seqnum)
2538                         {
2539                                 BufferedPacket queued_packet = channel->incoming_reliables.popFirst();
2540                                 /** TODO find a way to verify the new against the old packet */
2541                         }
2542                 }
2543
2544                 channel->incNextIncomingSeqNum();
2545
2546                 // Get out the inside packet and re-process it
2547                 SharedBuffer<u8> payload(packetdata.getSize() - RELIABLE_HEADER_SIZE);
2548                 memcpy(*payload, &packetdata[RELIABLE_HEADER_SIZE], payload.getSize());
2549
2550                 return processPacket(channel, payload, peer_id, channelnum, true);
2551         }
2552         else
2553         {
2554                 derr_con<<m_connection->getDesc()
2555                                 <<"Got invalid type="<<((int)type&0xff)<<std::endl;
2556                 throw InvalidIncomingDataException("Invalid packet type");
2557         }
2558
2559         // We should never get here.
2560         FATAL_ERROR("Invalid execution point");
2561 }
2562
2563 /*
2564         Connection
2565 */
2566
2567 Connection::Connection(u32 protocol_id, u32 max_packet_size, float timeout,
2568                 bool ipv6, PeerHandler *peerhandler) :
2569         m_udpSocket(ipv6),
2570         m_protocol_id(protocol_id),
2571         m_sendThread(max_packet_size, timeout),
2572         m_receiveThread(max_packet_size),
2573         m_bc_peerhandler(peerhandler)
2574
2575 {
2576         m_udpSocket.setTimeoutMs(5);
2577
2578         m_sendThread.setParent(this);
2579         m_receiveThread.setParent(this);
2580
2581         m_sendThread.start();
2582         m_receiveThread.start();
2583
2584 }
2585
2586
2587 Connection::~Connection()
2588 {
2589         m_shutting_down = true;
2590         // request threads to stop
2591         m_sendThread.stop();
2592         m_receiveThread.stop();
2593
2594         //TODO for some unkonwn reason send/receive threads do not exit as they're
2595         // supposed to be but wait on peer timeout. To speed up shutdown we reduce
2596         // timeout to half a second.
2597         m_sendThread.setPeerTimeout(0.5);
2598
2599         // wait for threads to finish
2600         m_sendThread.wait();
2601         m_receiveThread.wait();
2602
2603         // Delete peers
2604         for (auto &peer : m_peers) {
2605                 delete peer.second;
2606         }
2607 }
2608
2609 /* Internal stuff */
2610 void Connection::putEvent(ConnectionEvent &e)
2611 {
2612         assert(e.type != CONNEVENT_NONE); // Pre-condition
2613         m_event_queue.push_back(e);
2614 }
2615
2616 PeerHelper Connection::getPeer(u16 peer_id)
2617 {
2618         MutexAutoLock peerlock(m_peers_mutex);
2619         std::map<u16, Peer*>::iterator node = m_peers.find(peer_id);
2620
2621         if (node == m_peers.end()) {
2622                 throw PeerNotFoundException("GetPeer: Peer not found (possible timeout)");
2623         }
2624
2625         // Error checking
2626         FATAL_ERROR_IF(node->second->id != peer_id, "Invalid peer id");
2627
2628         return PeerHelper(node->second);
2629 }
2630
2631 PeerHelper Connection::getPeerNoEx(u16 peer_id)
2632 {
2633         MutexAutoLock peerlock(m_peers_mutex);
2634         std::map<u16, Peer*>::iterator node = m_peers.find(peer_id);
2635
2636         if (node == m_peers.end()) {
2637                 return PeerHelper(NULL);
2638         }
2639
2640         // Error checking
2641         FATAL_ERROR_IF(node->second->id != peer_id, "Invalid peer id");
2642
2643         return PeerHelper(node->second);
2644 }
2645
2646 /* find peer_id for address */
2647 u16 Connection::lookupPeer(Address& sender)
2648 {
2649         MutexAutoLock peerlock(m_peers_mutex);
2650         std::map<u16, Peer*>::iterator j;
2651         j = m_peers.begin();
2652         for(; j != m_peers.end(); ++j)
2653         {
2654                 Peer *peer = j->second;
2655                 if (peer->isPendingDeletion())
2656                         continue;
2657
2658                 Address tocheck;
2659
2660                 if ((peer->getAddress(MTP_MINETEST_RELIABLE_UDP, tocheck)) && (tocheck == sender))
2661                         return peer->id;
2662
2663                 if ((peer->getAddress(MTP_UDP, tocheck)) && (tocheck == sender))
2664                         return peer->id;
2665         }
2666
2667         return PEER_ID_INEXISTENT;
2668 }
2669
2670 std::list<Peer*> Connection::getPeers()
2671 {
2672         std::list<Peer*> list;
2673         for (auto &p : m_peers) {
2674                 Peer *peer = p.second;
2675                 list.push_back(peer);
2676         }
2677         return list;
2678 }
2679
2680 bool Connection::deletePeer(u16 peer_id, bool timeout)
2681 {
2682         Peer *peer = 0;
2683
2684         /* lock list as short as possible */
2685         {
2686                 MutexAutoLock peerlock(m_peers_mutex);
2687                 if (m_peers.find(peer_id) == m_peers.end())
2688                         return false;
2689                 peer = m_peers[peer_id];
2690                 m_peers.erase(peer_id);
2691                 m_peer_ids.remove(peer_id);
2692         }
2693
2694         Address peer_address;
2695         //any peer has a primary address this never fails!
2696         peer->getAddress(MTP_PRIMARY, peer_address);
2697         // Create event
2698         ConnectionEvent e;
2699         e.peerRemoved(peer_id, timeout, peer_address);
2700         putEvent(e);
2701
2702
2703         peer->Drop();
2704         return true;
2705 }
2706
2707 /* Interface */
2708
2709 ConnectionEvent Connection::waitEvent(u32 timeout_ms)
2710 {
2711         try {
2712                 return m_event_queue.pop_front(timeout_ms);
2713         } catch(ItemNotFoundException &ex) {
2714                 ConnectionEvent e;
2715                 e.type = CONNEVENT_NONE;
2716                 return e;
2717         }
2718 }
2719
2720 void Connection::putCommand(ConnectionCommand &c)
2721 {
2722         if (!m_shutting_down) {
2723                 m_command_queue.push_back(c);
2724                 m_sendThread.Trigger();
2725         }
2726 }
2727
2728 void Connection::Serve(Address bind_addr)
2729 {
2730         ConnectionCommand c;
2731         c.serve(bind_addr);
2732         putCommand(c);
2733 }
2734
2735 void Connection::Connect(Address address)
2736 {
2737         ConnectionCommand c;
2738         c.connect(address);
2739         putCommand(c);
2740 }
2741
2742 bool Connection::Connected()
2743 {
2744         MutexAutoLock peerlock(m_peers_mutex);
2745
2746         if (m_peers.size() != 1)
2747                 return false;
2748
2749         std::map<u16, Peer*>::iterator node = m_peers.find(PEER_ID_SERVER);
2750         if (node == m_peers.end())
2751                 return false;
2752
2753         if (m_peer_id == PEER_ID_INEXISTENT)
2754                 return false;
2755
2756         return true;
2757 }
2758
2759 void Connection::Disconnect()
2760 {
2761         ConnectionCommand c;
2762         c.disconnect();
2763         putCommand(c);
2764 }
2765
2766 void Connection::Receive(NetworkPacket* pkt)
2767 {
2768         for(;;) {
2769                 ConnectionEvent e = waitEvent(m_bc_receive_timeout);
2770                 if (e.type != CONNEVENT_NONE)
2771                         LOG(dout_con << getDesc() << ": Receive: got event: "
2772                                         << e.describe() << std::endl);
2773                 switch(e.type) {
2774                 case CONNEVENT_NONE:
2775                         throw NoIncomingDataException("No incoming data");
2776                 case CONNEVENT_DATA_RECEIVED:
2777                         // Data size is lesser than command size, ignoring packet
2778                         if (e.data.getSize() < 2) {
2779                                 continue;
2780                         }
2781
2782                         pkt->putRawPacket(*e.data, e.data.getSize(), e.peer_id);
2783                         return;
2784                 case CONNEVENT_PEER_ADDED: {
2785                         UDPPeer tmp(e.peer_id, e.address, this);
2786                         if (m_bc_peerhandler)
2787                                 m_bc_peerhandler->peerAdded(&tmp);
2788                         continue;
2789                 }
2790                 case CONNEVENT_PEER_REMOVED: {
2791                         UDPPeer tmp(e.peer_id, e.address, this);
2792                         if (m_bc_peerhandler)
2793                                 m_bc_peerhandler->deletingPeer(&tmp, e.timeout);
2794                         continue;
2795                 }
2796                 case CONNEVENT_BIND_FAILED:
2797                         throw ConnectionBindFailed("Failed to bind socket "
2798                                         "(port already in use?)");
2799                 }
2800         }
2801         throw NoIncomingDataException("No incoming data");
2802 }
2803
2804 void Connection::Send(u16 peer_id, u8 channelnum,
2805                 NetworkPacket* pkt, bool reliable)
2806 {
2807         assert(channelnum < CHANNEL_COUNT); // Pre-condition
2808
2809         ConnectionCommand c;
2810
2811         c.send(peer_id, channelnum, pkt, reliable);
2812         putCommand(c);
2813 }
2814
2815 Address Connection::GetPeerAddress(u16 peer_id)
2816 {
2817         PeerHelper peer = getPeerNoEx(peer_id);
2818
2819         if (!peer)
2820                 throw PeerNotFoundException("No address for peer found!");
2821         Address peer_address;
2822         peer->getAddress(MTP_PRIMARY, peer_address);
2823         return peer_address;
2824 }
2825
2826 float Connection::getPeerStat(u16 peer_id, rtt_stat_type type)
2827 {
2828         PeerHelper peer = getPeerNoEx(peer_id);
2829         if (!peer) return -1;
2830         return peer->getStat(type);
2831 }
2832
2833 float Connection::getLocalStat(rate_stat_type type)
2834 {
2835         PeerHelper peer = getPeerNoEx(PEER_ID_SERVER);
2836
2837         FATAL_ERROR_IF(!peer, "Connection::getLocalStat we couldn't get our own peer? are you serious???");
2838
2839         float retval = 0.0;
2840
2841         for (Channel &channel : dynamic_cast<UDPPeer *>(&peer)->channels) {
2842                 switch(type) {
2843                         case CUR_DL_RATE:
2844                                 retval += channel.getCurrentDownloadRateKB();
2845                                 break;
2846                         case AVG_DL_RATE:
2847                                 retval += channel.getAvgDownloadRateKB();
2848                                 break;
2849                         case CUR_INC_RATE:
2850                                 retval += channel.getCurrentIncomingRateKB();
2851                                 break;
2852                         case AVG_INC_RATE:
2853                                 retval += channel.getAvgIncomingRateKB();
2854                                 break;
2855                         case AVG_LOSS_RATE:
2856                                 retval += channel.getAvgLossRateKB();
2857                                 break;
2858                         case CUR_LOSS_RATE:
2859                                 retval += channel.getCurrentLossRateKB();
2860                                 break;
2861                 default:
2862                         FATAL_ERROR("Connection::getLocalStat Invalid stat type");
2863                 }
2864         }
2865         return retval;
2866 }
2867
2868 u16 Connection::createPeer(Address& sender, MTProtocols protocol, int fd)
2869 {
2870         // Somebody wants to make a new connection
2871
2872         // Get a unique peer id (2 or higher)
2873         u16 peer_id_new = m_next_remote_peer_id;
2874         u16 overflow =  MAX_UDP_PEERS;
2875
2876         /*
2877                 Find an unused peer id
2878         */
2879         MutexAutoLock lock(m_peers_mutex);
2880         bool out_of_ids = false;
2881         for(;;) {
2882                 // Check if exists
2883                 if (m_peers.find(peer_id_new) == m_peers.end())
2884
2885                         break;
2886                 // Check for overflow
2887                 if (peer_id_new == overflow) {
2888                         out_of_ids = true;
2889                         break;
2890                 }
2891                 peer_id_new++;
2892         }
2893
2894         if (out_of_ids) {
2895                 errorstream << getDesc() << " ran out of peer ids" << std::endl;
2896                 return PEER_ID_INEXISTENT;
2897         }
2898
2899         // Create a peer
2900         Peer *peer = 0;
2901         peer = new UDPPeer(peer_id_new, sender, this);
2902
2903         m_peers[peer->id] = peer;
2904         m_peer_ids.push_back(peer->id);
2905
2906         m_next_remote_peer_id = (peer_id_new +1 ) % MAX_UDP_PEERS;
2907
2908         LOG(dout_con << getDesc()
2909                         << "createPeer(): giving peer_id=" << peer_id_new << std::endl);
2910
2911         ConnectionCommand cmd;
2912         SharedBuffer<u8> reply(4);
2913         writeU8(&reply[0], TYPE_CONTROL);
2914         writeU8(&reply[1], CONTROLTYPE_SET_PEER_ID);
2915         writeU16(&reply[2], peer_id_new);
2916         cmd.createPeer(peer_id_new,reply);
2917         putCommand(cmd);
2918
2919         // Create peer addition event
2920         ConnectionEvent e;
2921         e.peerAdded(peer_id_new, sender);
2922         putEvent(e);
2923
2924         // We're now talking to a valid peer_id
2925         return peer_id_new;
2926 }
2927
2928 void Connection::PrintInfo(std::ostream &out)
2929 {
2930         m_info_mutex.lock();
2931         out<<getDesc()<<": ";
2932         m_info_mutex.unlock();
2933 }
2934
2935 void Connection::PrintInfo()
2936 {
2937         PrintInfo(dout_con);
2938 }
2939
2940 const std::string Connection::getDesc()
2941 {
2942         return std::string("con(")+
2943                         itos(m_udpSocket.GetHandle())+"/"+itos(m_peer_id)+")";
2944 }
2945
2946 void Connection::DisconnectPeer(u16 peer_id)
2947 {
2948         ConnectionCommand discon;
2949         discon.disconnect_peer(peer_id);
2950         putCommand(discon);
2951 }
2952
2953 void Connection::sendAck(u16 peer_id, u8 channelnum, u16 seqnum)
2954 {
2955         assert(channelnum < CHANNEL_COUNT); // Pre-condition
2956
2957         LOG(dout_con<<getDesc()
2958                         <<" Queuing ACK command to peer_id: " << peer_id <<
2959                         " channel: " << (channelnum & 0xFF) <<
2960                         " seqnum: " << seqnum << std::endl);
2961
2962         ConnectionCommand c;
2963         SharedBuffer<u8> ack(4);
2964         writeU8(&ack[0], TYPE_CONTROL);
2965         writeU8(&ack[1], CONTROLTYPE_ACK);
2966         writeU16(&ack[2], seqnum);
2967
2968         c.ack(peer_id, channelnum, ack);
2969         putCommand(c);
2970         m_sendThread.Trigger();
2971 }
2972
2973 UDPPeer* Connection::createServerPeer(Address& address)
2974 {
2975         if (getPeerNoEx(PEER_ID_SERVER) != 0)
2976         {
2977                 throw ConnectionException("Already connected to a server");
2978         }
2979
2980         UDPPeer *peer = new UDPPeer(PEER_ID_SERVER, address, this);
2981
2982         {
2983                 MutexAutoLock lock(m_peers_mutex);
2984                 m_peers[peer->id] = peer;
2985                 m_peer_ids.push_back(peer->id);
2986         }
2987
2988         return peer;
2989 }
2990
2991 } // namespace