]> git.lizzy.rs Git - minetest.git/blobdiff - src/network/connection.cpp
Clean up threading
[minetest.git] / src / network / connection.cpp
index fb31a07f0d85d3362a1c898bf2796ac542a5d2c4..10f0d55433f5a6e87da4978ccb35f11d364de2e0 100644 (file)
@@ -20,10 +20,10 @@ with this program; if not, write to the Free Software Foundation, Inc.,
 #include <iomanip>
 #include <errno.h>
 #include "connection.h"
-#include "main.h"
 #include "serialization.h"
 #include "log.h"
 #include "porting.h"
+#include "network/networkpacket.h"
 #include "util/serialize.h"
 #include "util/numeric.h"
 #include "util/string.h"
@@ -42,10 +42,10 @@ namespace con
 #undef DEBUG_CONNECTION_KBPS
 #else
 /* this mutex is used to achieve log message consistency */
-JMutex log_message_mutex;
+Mutex log_message_mutex;
 #define LOG(a)                                                                 \
        {                                                                          \
-       JMutexAutoLock loglock(log_message_mutex);                                 \
+       MutexAutoLock loglock(log_message_mutex);                                 \
        a;                                                                         \
        }
 #define PROFILE(a) a
@@ -209,7 +209,7 @@ ReliablePacketBuffer::ReliablePacketBuffer(): m_list_size(0) {}
 
 void ReliablePacketBuffer::print()
 {
-       JMutexAutoLock listlock(m_list_mutex);
+       MutexAutoLock listlock(m_list_mutex);
        LOG(dout_con<<"Dump of ReliablePacketBuffer:" << std::endl);
        unsigned int index = 0;
        for(std::list<BufferedPacket>::iterator i = m_list.begin();
@@ -223,7 +223,7 @@ void ReliablePacketBuffer::print()
 }
 bool ReliablePacketBuffer::empty()
 {
-       JMutexAutoLock listlock(m_list_mutex);
+       MutexAutoLock listlock(m_list_mutex);
        return m_list.empty();
 }
 
@@ -256,7 +256,7 @@ RPBSearchResult ReliablePacketBuffer::notFound()
 }
 bool ReliablePacketBuffer::getFirstSeqnum(u16& result)
 {
-       JMutexAutoLock listlock(m_list_mutex);
+       MutexAutoLock listlock(m_list_mutex);
        if (m_list.empty())
                return false;
        BufferedPacket p = *m_list.begin();
@@ -266,7 +266,7 @@ bool ReliablePacketBuffer::getFirstSeqnum(u16& result)
 
 BufferedPacket ReliablePacketBuffer::popFirst()
 {
-       JMutexAutoLock listlock(m_list_mutex);
+       MutexAutoLock listlock(m_list_mutex);
        if (m_list.empty())
                throw NotFoundException("Buffer is empty");
        BufferedPacket p = *m_list.begin();
@@ -283,7 +283,7 @@ BufferedPacket ReliablePacketBuffer::popFirst()
 }
 BufferedPacket ReliablePacketBuffer::popSeqnum(u16 seqnum)
 {
-       JMutexAutoLock listlock(m_list_mutex);
+       MutexAutoLock listlock(m_list_mutex);
        RPBSearchResult r = findPacket(seqnum);
        if (r == notFound()) {
                LOG(dout_con<<"Sequence number: " << seqnum
@@ -311,14 +311,30 @@ BufferedPacket ReliablePacketBuffer::popSeqnum(u16 seqnum)
 }
 void ReliablePacketBuffer::insert(BufferedPacket &p,u16 next_expected)
 {
-       JMutexAutoLock listlock(m_list_mutex);
-       FATAL_ERROR_IF(p.data.getSize() < BASE_HEADER_SIZE+3, "Invalid data size");
-       u8 type = readU8(&p.data[BASE_HEADER_SIZE+0]);
-       sanity_check(type == TYPE_RELIABLE);
-       u16 seqnum = readU16(&p.data[BASE_HEADER_SIZE+1]);
+       MutexAutoLock listlock(m_list_mutex);
+       if (p.data.getSize() < BASE_HEADER_SIZE + 3) {
+               errorstream << "ReliablePacketBuffer::insert(): Invalid data size for "
+                       "reliable packet" << std::endl;
+               return;
+       }
+       u8 type = readU8(&p.data[BASE_HEADER_SIZE + 0]);
+       if (type != TYPE_RELIABLE) {
+               errorstream << "ReliablePacketBuffer::insert(): type is not reliable"
+                       << std::endl;
+               return;
+       }
+       u16 seqnum = readU16(&p.data[BASE_HEADER_SIZE + 1]);
 
-       sanity_check(seqnum_in_window(seqnum, next_expected, MAX_RELIABLE_WINDOW_SIZE));
-       sanity_check(seqnum != next_expected);
+       if (!seqnum_in_window(seqnum, next_expected, MAX_RELIABLE_WINDOW_SIZE)) {
+               errorstream << "ReliablePacketBuffer::insert(): seqnum is outside of "
+                       "expected window " << std::endl;
+               return;
+       }
+       if (seqnum == next_expected) {
+               errorstream << "ReliablePacketBuffer::insert(): seqnum is next expected"
+                       << std::endl;
+               return;
+       }
 
        ++m_list_size;
        sanity_check(m_list_size <= SEQNUM_MAX+1);      // FIXME: Handle the error?
@@ -377,10 +393,6 @@ void ReliablePacketBuffer::insert(BufferedPacket &p,u16 next_expected)
                        throw IncomingDataCorruption("duplicated packet isn't same as original one");
                }
 
-               sanity_check(readU16(&(i->data[BASE_HEADER_SIZE+1])) == seqnum);
-               sanity_check(i->data.getSize() == p.data.getSize());
-               sanity_check(i->address == p.address);
-
                /* nothing to do this seems to be a resent packet */
                /* for paranoia reason data should be compared */
                --m_list_size;
@@ -399,7 +411,7 @@ void ReliablePacketBuffer::insert(BufferedPacket &p,u16 next_expected)
 
 void ReliablePacketBuffer::incrementTimeouts(float dtime)
 {
-       JMutexAutoLock listlock(m_list_mutex);
+       MutexAutoLock listlock(m_list_mutex);
        for(std::list<BufferedPacket>::iterator i = m_list.begin();
                i != m_list.end(); ++i)
        {
@@ -411,7 +423,7 @@ void ReliablePacketBuffer::incrementTimeouts(float dtime)
 std::list<BufferedPacket> ReliablePacketBuffer::getTimedOuts(float timeout,
                                                                                                        unsigned int max_packets)
 {
-       JMutexAutoLock listlock(m_list_mutex);
+       MutexAutoLock listlock(m_list_mutex);
        std::list<BufferedPacket> timed_outs;
        for(std::list<BufferedPacket>::iterator i = m_list.begin();
                i != m_list.end(); ++i)
@@ -434,7 +446,7 @@ std::list<BufferedPacket> ReliablePacketBuffer::getTimedOuts(float timeout,
 
 IncomingSplitBuffer::~IncomingSplitBuffer()
 {
-       JMutexAutoLock listlock(m_map_mutex);
+       MutexAutoLock listlock(m_map_mutex);
        for(std::map<u16, IncomingSplitPacket*>::iterator i = m_buf.begin();
                i != m_buf.end(); ++i)
        {
@@ -447,15 +459,23 @@ IncomingSplitBuffer::~IncomingSplitBuffer()
 */
 SharedBuffer<u8> IncomingSplitBuffer::insert(BufferedPacket &p, bool reliable)
 {
-       JMutexAutoLock listlock(m_map_mutex);
+       MutexAutoLock listlock(m_map_mutex);
        u32 headersize = BASE_HEADER_SIZE + 7;
-       FATAL_ERROR_IF(p.data.getSize() < headersize, "Invalid data size");
+       if (p.data.getSize() < headersize) {
+               errorstream << "Invalid data size for split packet" << std::endl;
+               return SharedBuffer<u8>();
+       }
        u8 type = readU8(&p.data[BASE_HEADER_SIZE+0]);
-       sanity_check(type == TYPE_SPLIT);
        u16 seqnum = readU16(&p.data[BASE_HEADER_SIZE+1]);
        u16 chunk_count = readU16(&p.data[BASE_HEADER_SIZE+3]);
        u16 chunk_num = readU16(&p.data[BASE_HEADER_SIZE+5]);
 
+       if (type != TYPE_SPLIT) {
+               errorstream << "IncomingSplitBuffer::insert(): type is not split"
+                       << std::endl;
+               return SharedBuffer<u8>();
+       }
+
        // Add if doesn't exist
        if (m_buf.find(seqnum) == m_buf.end())
        {
@@ -526,7 +546,7 @@ void IncomingSplitBuffer::removeUnreliableTimedOuts(float dtime, float timeout)
 {
        std::list<u16> remove_queue;
        {
-               JMutexAutoLock listlock(m_map_mutex);
+               MutexAutoLock listlock(m_map_mutex);
                for(std::map<u16, IncomingSplitPacket*>::iterator i = m_buf.begin();
                        i != m_buf.end(); ++i)
                {
@@ -542,7 +562,7 @@ void IncomingSplitBuffer::removeUnreliableTimedOuts(float dtime, float timeout)
        for(std::list<u16>::iterator j = remove_queue.begin();
                j != remove_queue.end(); ++j)
        {
-               JMutexAutoLock listlock(m_map_mutex);
+               MutexAutoLock listlock(m_map_mutex);
                LOG(dout_con<<"NOTE: Removing timed out unreliable split packet"<<std::endl);
                delete m_buf[*j];
                m_buf.erase(*j);
@@ -585,13 +605,13 @@ Channel::~Channel()
 
 u16 Channel::readNextIncomingSeqNum()
 {
-       JMutexAutoLock internal(m_internal_mutex);
+       MutexAutoLock internal(m_internal_mutex);
        return next_incoming_seqnum;
 }
 
 u16 Channel::incNextIncomingSeqNum()
 {
-       JMutexAutoLock internal(m_internal_mutex);
+       MutexAutoLock internal(m_internal_mutex);
        u16 retval = next_incoming_seqnum;
        next_incoming_seqnum++;
        return retval;
@@ -599,18 +619,18 @@ u16 Channel::incNextIncomingSeqNum()
 
 u16 Channel::readNextSplitSeqNum()
 {
-       JMutexAutoLock internal(m_internal_mutex);
+       MutexAutoLock internal(m_internal_mutex);
        return next_outgoing_split_seqnum;
 }
 void Channel::setNextSplitSeqNum(u16 seqnum)
 {
-       JMutexAutoLock internal(m_internal_mutex);
+       MutexAutoLock internal(m_internal_mutex);
        next_outgoing_split_seqnum = seqnum;
 }
 
-u16 Channel::getOutgoingSequenceNumber(bool& successfull)
+u16 Channel::getOutgoingSequenceNumber(bool& successful)
 {
-       JMutexAutoLock internal(m_internal_mutex);
+       MutexAutoLock internal(m_internal_mutex);
        u16 retval = next_outgoing_seqnum;
        u16 lowest_unacked_seqnumber;
 
@@ -628,7 +648,7 @@ u16 Channel::getOutgoingSequenceNumber(bool& successfull)
                        // know about difference of two unsigned may be negative in general
                        // but we already made sure it won't happen in this case
                        if (((u16)(next_outgoing_seqnum - lowest_unacked_seqnumber)) > window_size) {
-                               successfull = false;
+                               successful = false;
                                return 0;
                        }
                }
@@ -638,7 +658,7 @@ u16 Channel::getOutgoingSequenceNumber(bool& successfull)
                        // but we already made sure it won't happen in this case
                        if ((next_outgoing_seqnum + (u16)(SEQNUM_MAX - lowest_unacked_seqnumber)) >
                                window_size) {
-                               successfull = false;
+                               successful = false;
                                return 0;
                        }
                }
@@ -650,7 +670,7 @@ u16 Channel::getOutgoingSequenceNumber(bool& successfull)
 
 u16 Channel::readOutgoingSequenceNumber()
 {
-       JMutexAutoLock internal(m_internal_mutex);
+       MutexAutoLock internal(m_internal_mutex);
        return next_outgoing_seqnum;
 }
 
@@ -666,32 +686,32 @@ bool Channel::putBackSequenceNumber(u16 seqnum)
 
 void Channel::UpdateBytesSent(unsigned int bytes, unsigned int packets)
 {
-       JMutexAutoLock internal(m_internal_mutex);
+       MutexAutoLock internal(m_internal_mutex);
        current_bytes_transfered += bytes;
        current_packet_successfull += packets;
 }
 
 void Channel::UpdateBytesReceived(unsigned int bytes) {
-       JMutexAutoLock internal(m_internal_mutex);
+       MutexAutoLock internal(m_internal_mutex);
        current_bytes_received += bytes;
 }
 
 void Channel::UpdateBytesLost(unsigned int bytes)
 {
-       JMutexAutoLock internal(m_internal_mutex);
+       MutexAutoLock internal(m_internal_mutex);
        current_bytes_lost += bytes;
 }
 
 
 void Channel::UpdatePacketLossCounter(unsigned int count)
 {
-       JMutexAutoLock internal(m_internal_mutex);
+       MutexAutoLock internal(m_internal_mutex);
        current_packet_loss += count;
 }
 
 void Channel::UpdatePacketTooLateCounter()
 {
-       JMutexAutoLock internal(m_internal_mutex);
+       MutexAutoLock internal(m_internal_mutex);
        current_packet_too_late++;
 }
 
@@ -711,7 +731,7 @@ void Channel::UpdateTimers(float dtime,bool legacy_peer)
                bool reasonable_amount_of_data_transmitted = false;
 
                {
-                       JMutexAutoLock internal(m_internal_mutex);
+                       MutexAutoLock internal(m_internal_mutex);
                        packet_loss = current_packet_loss;
                        //packet_too_late = current_packet_too_late;
                        packets_successfull = current_packet_successfull;
@@ -782,7 +802,7 @@ void Channel::UpdateTimers(float dtime,bool legacy_peer)
        if (bpm_counter > 10.0)
        {
                {
-                       JMutexAutoLock internal(m_internal_mutex);
+                       MutexAutoLock internal(m_internal_mutex);
                        cur_kbps                 =
                                        (((float) current_bytes_transfered)/bpm_counter)/1024.0;
                        current_bytes_transfered = 0;
@@ -883,7 +903,7 @@ bool PeerHelper::operator!=(void* ptr)
 
 bool Peer::IncUseCount()
 {
-       JMutexAutoLock lock(m_exclusive_access_mutex);
+       MutexAutoLock lock(m_exclusive_access_mutex);
 
        if (!m_pending_deletion)
        {
@@ -897,7 +917,7 @@ bool Peer::IncUseCount()
 void Peer::DecUseCount()
 {
        {
-               JMutexAutoLock lock(m_exclusive_access_mutex);
+               MutexAutoLock lock(m_exclusive_access_mutex);
                sanity_check(m_usage > 0);
                m_usage--;
 
@@ -958,7 +978,7 @@ void Peer::RTTStatistics(float rtt, std::string profiler_id,
 
 bool Peer::isTimedOut(float timeout)
 {
-       JMutexAutoLock lock(m_exclusive_access_mutex);
+       MutexAutoLock lock(m_exclusive_access_mutex);
        u32 current_time = porting::getTimeMs();
 
        float dtime = CALC_DTIME(m_last_timeout_check,current_time);
@@ -972,7 +992,7 @@ bool Peer::isTimedOut(float timeout)
 void Peer::Drop()
 {
        {
-               JMutexAutoLock usage_lock(m_exclusive_access_mutex);
+               MutexAutoLock usage_lock(m_exclusive_access_mutex);
                m_pending_deletion = true;
                if (m_usage != 0)
                        return;
@@ -1031,7 +1051,7 @@ void UDPPeer::reportRTT(float rtt)
        if (timeout > RESEND_TIMEOUT_MAX)
                timeout = RESEND_TIMEOUT_MAX;
 
-       JMutexAutoLock usage_lock(m_exclusive_access_mutex);
+       MutexAutoLock usage_lock(m_exclusive_access_mutex);
        resend_timeout = timeout;
 }
 
@@ -1063,14 +1083,14 @@ void UDPPeer::PutReliableSendCommand(ConnectionCommand &c,
                                <<" processing reliable command for peer id: " << c.peer_id
                                <<" data size: " << c.data.getSize() << std::endl);
                if (!processReliableSendCommand(c,max_packet_size)) {
-                       channels[c.channelnum].queued_commands.push(c);
+                       channels[c.channelnum].queued_commands.push_back(c);
                }
        }
        else {
                LOG(dout_con<<m_connection->getDesc()
                                <<" Queueing reliable command for peer id: " << c.peer_id
                                <<" data size: " << c.data.getSize() <<std::endl);
-               channels[c.channelnum].queued_commands.push(c);
+               channels[c.channelnum].queued_commands.push_back(c);
        }
 }
 
@@ -1182,25 +1202,26 @@ void UDPPeer::RunCommandQueues(
                                                        unsigned int maxtransfer)
 {
 
-       for (unsigned int i = 0; i < CHANNEL_COUNT; i++)
-       {
+       for (unsigned int i = 0; i < CHANNEL_COUNT; i++) {
                unsigned int commands_processed = 0;
 
                if ((channels[i].queued_commands.size() > 0) &&
                                (channels[i].queued_reliables.size() < maxtransfer) &&
-                               (commands_processed < maxcommands))
-               {
+                               (commands_processed < maxcommands)) {
                        try {
                                ConnectionCommand c = channels[i].queued_commands.front();
-                               channels[i].queued_commands.pop();
-                               LOG(dout_con<<m_connection->getDesc()
-                                               <<" processing queued reliable command "<<std::endl);
-                               if (!processReliableSendCommand(c,max_packet_size)) {
-                                       LOG(dout_con<<m_connection->getDesc()
+
+                               LOG(dout_con << m_connection->getDesc()
+                                               << " processing queued reliable command " << std::endl);
+
+                               // Packet is processed, remove it from queue
+                               if (processReliableSendCommand(c,max_packet_size)) {
+                                       channels[i].queued_commands.pop_front();
+                               } else {
+                                       LOG(dout_con << m_connection->getDesc()
                                                        << " Failed to queue packets for peer_id: " << c.peer_id
                                                        << ", delaying sending of " << c.data.getSize()
                                                        << " bytes" << std::endl);
-                                       channels[i].queued_commands.push(c);
                                }
                        }
                        catch (ItemNotFoundException &e) {
@@ -1234,8 +1255,9 @@ SharedBuffer<u8> UDPPeer::addSpiltPacket(u8 channel,
 /* Connection Threads                                                         */
 /******************************************************************************/
 
-ConnectionSendThread::ConnectionSendThread( unsigned int max_packet_size,
-                                                                                       float timeout) :
+ConnectionSendThread::ConnectionSendThread(unsigned int max_packet_size,
+               float timeout) :
+       Thread("ConnectionSend"),
        m_connection(NULL),
        m_max_packet_size(max_packet_size),
        m_timeout(timeout),
@@ -1245,11 +1267,9 @@ ConnectionSendThread::ConnectionSendThread( unsigned int max_packet_size,
 {
 }
 
-void * ConnectionSendThread::Thread()
+void * ConnectionSendThread::run()
 {
-       assert(m_connection != NULL);
-       ThreadStarted();
-       log_register_thread("ConnectionSend");
+       assert(m_connection);
 
        LOG(dout_con<<m_connection->getDesc()
                        <<"ConnectionSend thread started"<<std::endl);
@@ -1260,21 +1280,19 @@ void * ConnectionSendThread::Thread()
        PROFILE(std::stringstream ThreadIdentifier);
        PROFILE(ThreadIdentifier << "ConnectionSend: [" << m_connection->getDesc() << "]");
 
-       porting::setThreadName("ConnectionSend");
-
        /* if stop is requested don't stop immediately but try to send all        */
        /* packets first */
-       while(!StopRequested() || packetsQueued()) {
+       while(!stopRequested() || packetsQueued()) {
                BEGIN_DEBUG_EXCEPTION_HANDLER
                PROFILE(ScopeProfiler sp(g_profiler, ThreadIdentifier.str(), SPT_AVG));
 
                m_iteration_packets_avaialble = m_max_data_packets_per_iteration;
 
                /* wait for trigger or timeout */
-               m_send_sleep_semaphore.Wait(50);
+               m_send_sleep_semaphore.wait(50);
 
                /* remove all triggers */
-               while(m_send_sleep_semaphore.Wait(0)) {}
+               while(m_send_sleep_semaphore.wait(0)) {}
 
                lasttime = curtime;
                curtime = porting::getTimeMs();
@@ -1307,7 +1325,7 @@ void * ConnectionSendThread::Thread()
 
 void ConnectionSendThread::Trigger()
 {
-       m_send_sleep_semaphore.Post();
+       m_send_sleep_semaphore.post();
 }
 
 bool ConnectionSendThread::packetsQueued()
@@ -1328,12 +1346,10 @@ bool ConnectionSendThread::packetsQueued()
                if (dynamic_cast<UDPPeer*>(&peer) == 0)
                        continue;
 
-               for(u16 i=0; i<CHANNEL_COUNT; i++)
-               {
+               for(u16 i=0; i < CHANNEL_COUNT; i++) {
                        Channel *channel = &(dynamic_cast<UDPPeer*>(&peer))->channels[i];
 
-                       if (channel->queued_commands.size() > 0)
-                       {
+                       if (channel->queued_commands.size() > 0) {
                                return true;
                        }
                }
@@ -1725,9 +1741,8 @@ void ConnectionSendThread::connect(Address address)
 
        // Send a dummy packet to server with peer_id = PEER_ID_INEXISTENT
        m_connection->SetPeerID(PEER_ID_INEXISTENT);
-       NetworkPacket* pkt = new NetworkPacket(0,0);
-       m_connection->Send(PEER_ID_SERVER, 0, pkt, true);
-       delete pkt;
+       NetworkPacket pkt(0,0);
+       m_connection->Send(PEER_ID_SERVER, 0, &pkt, true);
 }
 
 void ConnectionSendThread::disconnect()
@@ -1966,7 +1981,7 @@ void ConnectionSendThread::sendPackets(float dtime)
                }
                else if (
                        ( peer->m_increment_packets_remaining > 0) ||
-                       (StopRequested())) {
+                       (stopRequested())) {
                        rawSendAsPacket(packet.peer_id, packet.channelnum,
                                        packet.data, packet.reliable);
                        peer->m_increment_packets_remaining--;
@@ -1996,15 +2011,14 @@ void ConnectionSendThread::sendAsPacket(u16 peer_id, u8 channelnum,
 }
 
 ConnectionReceiveThread::ConnectionReceiveThread(unsigned int max_packet_size) :
+       Thread("ConnectionReceive"),
        m_connection(NULL)
 {
 }
 
-void * ConnectionReceiveThread::Thread()
+void * ConnectionReceiveThread::run()
 {
-       assert(m_connection != NULL);
-       ThreadStarted();
-       log_register_thread("ConnectionReceive");
+       assert(m_connection);
 
        LOG(dout_con<<m_connection->getDesc()
                        <<"ConnectionReceive thread started"<<std::endl);
@@ -2012,15 +2026,13 @@ void * ConnectionReceiveThread::Thread()
        PROFILE(std::stringstream ThreadIdentifier);
        PROFILE(ThreadIdentifier << "ConnectionReceive: [" << m_connection->getDesc() << "]");
 
-       porting::setThreadName("ConnectionReceive");
-
 #ifdef DEBUG_CONNECTION_KBPS
        u32 curtime = porting::getTimeMs();
        u32 lasttime = curtime;
        float debug_print_timer = 0.0;
 #endif
 
-       while(!StopRequested()) {
+       while(!stopRequested()) {
                BEGIN_DEBUG_EXCEPTION_HANDLER
                PROFILE(ScopeProfiler sp(g_profiler, ThreadIdentifier.str(), SPT_AVG));
 
@@ -2329,8 +2341,9 @@ SharedBuffer<u8> ConnectionReceiveThread::processPacket(Channel *channel,
        u8 type = readU8(&(packetdata[0]));
 
        if (MAX_UDP_PEERS <= 65535 && peer_id >= MAX_UDP_PEERS) {
-               errorstream << "Something is wrong with peer_id" << std::endl;
-               FATAL_ERROR("");
+               std::string errmsg = "Invalid peer_id=" + itos(peer_id);
+               errorstream << errmsg << std::endl;
+               throw InvalidIncomingDataException(errmsg.c_str());
        }
 
        if (type == TYPE_CONTROL)
@@ -2342,10 +2355,12 @@ SharedBuffer<u8> ConnectionReceiveThread::processPacket(Channel *channel,
 
                if (controltype == CONTROLTYPE_ACK)
                {
-                       FATAL_ERROR_IF(channel == 0, "Invalid channel (0)");
-                       if (packetdata.getSize() < 4)
-                               throw InvalidIncomingDataException
-                                               ("packetdata.getSize() < 4 (ACK header size)");
+                       assert(channel != NULL);
+
+                       if (packetdata.getSize() < 4) {
+                               throw InvalidIncomingDataException(
+                                       "packetdata.getSize() < 4 (ACK header size)");
+                       }
 
                        u16 seqnum = readU16(&packetdata[2]);
                        LOG(dout_con<<m_connection->getDesc()
@@ -2510,7 +2525,8 @@ SharedBuffer<u8> ConnectionReceiveThread::processPacket(Channel *channel,
        }
        else if (type == TYPE_RELIABLE)
        {
-               FATAL_ERROR_IF(channel == 0, "Invalid channel (0)");
+               assert(channel != NULL);
+
                // Recursive reliable packets not allowed
                if (reliable)
                        throw InvalidIncomingDataException("Found nested reliable packets");
@@ -2641,30 +2657,6 @@ SharedBuffer<u8> ConnectionReceiveThread::processPacket(Channel *channel,
        Connection
 */
 
-Connection::Connection(u32 protocol_id, u32 max_packet_size, float timeout,
-               bool ipv6) :
-       m_udpSocket(ipv6),
-       m_command_queue(),
-       m_event_queue(),
-       m_peer_id(0),
-       m_protocol_id(protocol_id),
-       m_sendThread(max_packet_size, timeout),
-       m_receiveThread(max_packet_size),
-       m_info_mutex(),
-       m_bc_peerhandler(0),
-       m_bc_receive_timeout(0),
-       m_shutting_down(false),
-       m_next_remote_peer_id(2)
-{
-       m_udpSocket.setTimeoutMs(5);
-
-       m_sendThread.setParent(this);
-       m_receiveThread.setParent(this);
-
-       m_sendThread.Start();
-       m_receiveThread.Start();
-}
-
 Connection::Connection(u32 protocol_id, u32 max_packet_size, float timeout,
                bool ipv6, PeerHandler *peerhandler) :
        m_udpSocket(ipv6),
@@ -2686,8 +2678,8 @@ Connection::Connection(u32 protocol_id, u32 max_packet_size, float timeout,
        m_sendThread.setParent(this);
        m_receiveThread.setParent(this);
 
-       m_sendThread.Start();
-       m_receiveThread.Start();
+       m_sendThread.start();
+       m_receiveThread.start();
 
 }
 
@@ -2696,8 +2688,8 @@ Connection::~Connection()
 {
        m_shutting_down = true;
        // request threads to stop
-       m_sendThread.Stop();
-       m_receiveThread.Stop();
+       m_sendThread.stop();
+       m_receiveThread.stop();
 
        //TODO for some unkonwn reason send/receive threads do not exit as they're
        // supposed to be but wait on peer timeout. To speed up shutdown we reduce
@@ -2705,8 +2697,8 @@ Connection::~Connection()
        m_sendThread.setPeerTimeout(0.5);
 
        // wait for threads to finish
-       m_sendThread.Wait();
-       m_receiveThread.Wait();
+       m_sendThread.wait();
+       m_receiveThread.wait();
 
        // Delete peers
        for(std::map<u16, Peer*>::iterator
@@ -2726,7 +2718,7 @@ void Connection::putEvent(ConnectionEvent &e)
 
 PeerHelper Connection::getPeer(u16 peer_id)
 {
-       JMutexAutoLock peerlock(m_peers_mutex);
+       MutexAutoLock peerlock(m_peers_mutex);
        std::map<u16, Peer*>::iterator node = m_peers.find(peer_id);
 
        if (node == m_peers.end()) {
@@ -2741,7 +2733,7 @@ PeerHelper Connection::getPeer(u16 peer_id)
 
 PeerHelper Connection::getPeerNoEx(u16 peer_id)
 {
-       JMutexAutoLock peerlock(m_peers_mutex);
+       MutexAutoLock peerlock(m_peers_mutex);
        std::map<u16, Peer*>::iterator node = m_peers.find(peer_id);
 
        if (node == m_peers.end()) {
@@ -2757,7 +2749,7 @@ PeerHelper Connection::getPeerNoEx(u16 peer_id)
 /* find peer_id for address */
 u16 Connection::lookupPeer(Address& sender)
 {
-       JMutexAutoLock peerlock(m_peers_mutex);
+       MutexAutoLock peerlock(m_peers_mutex);
        std::map<u16, Peer*>::iterator j;
        j = m_peers.begin();
        for(; j != m_peers.end(); ++j)
@@ -2796,7 +2788,7 @@ bool Connection::deletePeer(u16 peer_id, bool timeout)
 
        /* lock list as short as possible */
        {
-               JMutexAutoLock peerlock(m_peers_mutex);
+               MutexAutoLock peerlock(m_peers_mutex);
                if (m_peers.find(peer_id) == m_peers.end())
                        return false;
                peer = m_peers[peer_id];
@@ -2819,16 +2811,6 @@ bool Connection::deletePeer(u16 peer_id, bool timeout)
 
 /* Interface */
 
-ConnectionEvent Connection::getEvent()
-{
-       if (m_event_queue.empty()) {
-               ConnectionEvent e;
-               e.type = CONNEVENT_NONE;
-               return e;
-       }
-       return m_event_queue.pop_frontNoEx();
-}
-
 ConnectionEvent Connection::waitEvent(u32 timeout_ms)
 {
        try {
@@ -2864,7 +2846,7 @@ void Connection::Connect(Address address)
 
 bool Connection::Connected()
 {
-       JMutexAutoLock peerlock(m_peers_mutex);
+       MutexAutoLock peerlock(m_peers_mutex);
 
        if (m_peers.size() != 1)
                return false;
@@ -2886,30 +2868,36 @@ void Connection::Disconnect()
        putCommand(c);
 }
 
-u32 Connection::Receive(u16 &peer_id, SharedBuffer<u8> &data)
+void Connection::Receive(NetworkPacket* pkt)
 {
        for(;;) {
                ConnectionEvent e = waitEvent(m_bc_receive_timeout);
                if (e.type != CONNEVENT_NONE)
-                       LOG(dout_con<<getDesc()<<": Receive: got event: "
-                                       <<e.describe()<<std::endl);
+                       LOG(dout_con << getDesc() << ": Receive: got event: "
+                                       << e.describe() << std::endl);
                switch(e.type) {
                case CONNEVENT_NONE:
                        throw NoIncomingDataException("No incoming data");
                case CONNEVENT_DATA_RECEIVED:
-                       peer_id = e.peer_id;
-                       data = SharedBuffer<u8>(e.data);
-                       return e.data.getSize();
+                       // Data size is lesser than command size, ignoring packet
+                       if (e.data.getSize() < 2) {
+                               continue;
+                       }
+
+                       pkt->putRawPacket(*e.data, e.data.getSize(), e.peer_id);
+                       return;
                case CONNEVENT_PEER_ADDED: {
                        UDPPeer tmp(e.peer_id, e.address, this);
                        if (m_bc_peerhandler)
                                m_bc_peerhandler->peerAdded(&tmp);
-                       continue; }
+                       continue;
+               }
                case CONNEVENT_PEER_REMOVED: {
                        UDPPeer tmp(e.peer_id, e.address, this);
                        if (m_bc_peerhandler)
                                m_bc_peerhandler->deletingPeer(&tmp, e.timeout);
-                       continue; }
+                       continue;
+               }
                case CONNEVENT_BIND_FAILED:
                        throw ConnectionBindFailed("Failed to bind socket "
                                        "(port already in use?)");
@@ -2925,7 +2913,7 @@ void Connection::Send(u16 peer_id, u8 channelnum,
 
        ConnectionCommand c;
 
-       c.send(peer_id, channelnum, pkt->oldForgePacket(), reliable);
+       c.send(peer_id, channelnum, pkt, reliable);
        putCommand(c);
 }
 
@@ -2993,7 +2981,7 @@ u16 Connection::createPeer(Address& sender, MTProtocols protocol, int fd)
        /*
                Find an unused peer id
        */
-       JMutexAutoLock lock(m_peers_mutex);
+       MutexAutoLock lock(m_peers_mutex);
        bool out_of_ids = false;
        for(;;) {
                // Check if exists
@@ -3044,9 +3032,9 @@ u16 Connection::createPeer(Address& sender, MTProtocols protocol, int fd)
 
 void Connection::PrintInfo(std::ostream &out)
 {
-       m_info_mutex.Lock();
+       m_info_mutex.lock();
        out<<getDesc()<<": ";
-       m_info_mutex.Unlock();
+       m_info_mutex.unlock();
 }
 
 void Connection::PrintInfo()
@@ -3097,7 +3085,7 @@ UDPPeer* Connection::createServerPeer(Address& address)
        UDPPeer *peer = new UDPPeer(PEER_ID_SERVER, address, this);
 
        {
-               JMutexAutoLock lock(m_peers_mutex);
+               MutexAutoLock lock(m_peers_mutex);
                m_peers[peer->id] = peer;
                m_peer_ids.push_back(peer->id);
        }