#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"
void ReliablePacketBuffer::insert(BufferedPacket &p,u16 next_expected)
{
JMutexAutoLock listlock(m_list_mutex);
- assert(p.data.getSize() >= BASE_HEADER_SIZE+3);
+ FATAL_ERROR_IF(p.data.getSize() < BASE_HEADER_SIZE+3, "Invalid data size");
u8 type = readU8(&p.data[BASE_HEADER_SIZE+0]);
- assert(type == TYPE_RELIABLE);
+ sanity_check(type == TYPE_RELIABLE);
u16 seqnum = readU16(&p.data[BASE_HEADER_SIZE+1]);
- assert(seqnum_in_window(seqnum,next_expected,MAX_RELIABLE_WINDOW_SIZE));
- assert(seqnum != next_expected);
+ sanity_check(seqnum_in_window(seqnum, next_expected, MAX_RELIABLE_WINDOW_SIZE));
+ sanity_check(seqnum != next_expected);
++m_list_size;
- assert(m_list_size <= SEQNUM_MAX+1);
+ sanity_check(m_list_size <= SEQNUM_MAX+1); // FIXME: Handle the error?
// Find the right place for the packet and insert it there
// If list is empty, just add it
throw IncomingDataCorruption("duplicated packet isn't same as original one");
}
- assert(readU16(&(i->data[BASE_HEADER_SIZE+1])) == seqnum);
- assert(i->data.getSize() == p.data.getSize());
- assert(i->address == p.address);
+ 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 */
{
JMutexAutoLock listlock(m_map_mutex);
u32 headersize = BASE_HEADER_SIZE + 7;
- assert(p.data.getSize() >= headersize);
+ FATAL_ERROR_IF(p.data.getSize() < headersize, "Invalid data size");
u8 type = readU8(&p.data[BASE_HEADER_SIZE+0]);
- assert(type == TYPE_SPLIT);
+ 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]);
next_outgoing_split_seqnum = seqnum;
}
-u16 Channel::getOutgoingSequenceNumber(bool& successfull)
+u16 Channel::getOutgoingSequenceNumber(bool& successful)
{
JMutexAutoLock internal(m_internal_mutex);
u16 retval = next_outgoing_seqnum;
// 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;
}
}
// 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;
}
}
{
{
JMutexAutoLock lock(m_exclusive_access_mutex);
- assert(m_usage > 0);
+ sanity_check(m_usage > 0);
m_usage--;
if (!((m_pending_deletion) && (m_usage == 0)))
if ( channels[c.channelnum].queued_commands.empty() &&
/* don't queue more packets then window size */
(channels[c.channelnum].queued_reliables.size()
- < (channels[c.channelnum].getWindowSize()/2)))
- {
+ < (channels[c.channelnum].getWindowSize()/2))) {
LOG(dout_con<<m_connection->getDesc()
<<" processing reliable command for peer id: " << c.peer_id
<<" data size: " << c.data.getSize() << std::endl);
- if (!processReliableSendCommand(c,max_packet_size))
- {
+ if (!processReliableSendCommand(c,max_packet_size)) {
channels[c.channelnum].queued_commands.push_back(c);
}
}
- else
- {
+ else {
LOG(dout_con<<m_connection->getDesc()
<<" Queueing reliable command for peer id: " << c.peer_id
<<" data size: " << c.data.getSize() <<std::endl);
- BASE_HEADER_SIZE
- RELIABLE_HEADER_SIZE;
- assert(c.data.getSize() < MAX_RELIABLE_WINDOW_SIZE*512);
+ sanity_check(c.data.getSize() < MAX_RELIABLE_WINDOW_SIZE*512);
std::list<SharedBuffer<u8> > originals;
u16 split_sequence_number = channels[c.channelnum].readNextSplitSeqNum();
bool have_sequence_number = true;
bool have_initial_sequence_number = false;
- Queue<BufferedPacket> toadd;
+ std::queue<BufferedPacket> toadd;
volatile u16 initial_sequence_number = 0;
for(std::list<SharedBuffer<u8> >::iterator i = originals.begin();
m_connection->GetProtocolID(), m_connection->GetPeerID(),
c.channelnum);
- toadd.push_back(p);
+ toadd.push(p);
}
if (have_sequence_number) {
volatile u16 pcount = 0;
while(toadd.size() > 0) {
- BufferedPacket p = toadd.pop_front();
+ BufferedPacket p = toadd.front();
+ toadd.pop();
// LOG(dout_con<<connection->getDesc()
// << " queuing reliable packet for peer_id: " << c.peer_id
// << " channel: " << (c.channelnum&0xFF)
// << " seqnum: " << readU16(&p.data[BASE_HEADER_SIZE+1])
// << std::endl)
- channels[c.channelnum].queued_reliables.push_back(p);
+ channels[c.channelnum].queued_reliables.push(p);
pcount++;
}
- assert(channels[c.channelnum].queued_reliables.size() < 0xFFFF);
+ sanity_check(channels[c.channelnum].queued_reliables.size() < 0xFFFF);
return true;
}
else {
}
while(toadd.size() > 0) {
/* remove packet */
- toadd.pop_front();
+ toadd.pop();
bool successfully_put_back_sequence_number
= channels[c.channelnum].putBackSequenceNumber(
(initial_sequence_number+toadd.size() % (SEQNUM_MAX+1)));
- assert(successfully_put_back_sequence_number);
+ FATAL_ERROR_IF(!successfully_put_back_sequence_number, "error");
}
LOG(dout_con<<m_connection->getDesc()
<< " Windowsize exceeded on reliable sending "
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.pop_front();
- LOG(dout_con<<m_connection->getDesc()
- <<" processing queued reliable command "<<std::endl);
- if (!processReliableSendCommand(c,max_packet_size)) {
- LOG(dout_con<<m_connection->getDesc()
+ ConnectionCommand c = channels[i].queued_commands.front();
+
+ LOG(dout_con << m_connection->getDesc()
+ << " processing queued reliable command " << std::endl);
+
+ // Packet is processed, remove it from queue
+ if (processReliableSendCommand(c,max_packet_size)) {
+ channels[i].queued_commands.pop_front();
+ } 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_front(c);
}
}
catch (ItemNotFoundException &e) {
u16 UDPPeer::getNextSplitSequenceNumber(u8 channel)
{
- assert(channel < CHANNEL_COUNT);
+ assert(channel < CHANNEL_COUNT); // Pre-condition
return channels[channel].readNextIncomingSeqNum();
}
void UDPPeer::setNextSplitSequenceNumber(u8 channel, u16 seqnum)
{
- assert(channel < CHANNEL_COUNT);
+ assert(channel < CHANNEL_COUNT); // Pre-condition
channels[channel].setNextSplitSeqNum(seqnum);
}
BufferedPacket toadd,
bool reliable)
{
- assert(channel < CHANNEL_COUNT);
+ assert(channel < CHANNEL_COUNT); // Pre-condition
return channels[channel].incoming_splits.insert(toadd,reliable);
}
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;
}
}
LOG(derr_con<<m_connection->getDesc()
<<"WARNING: Going to send a reliable packet"
<<" in outgoing buffer" <<std::endl);
- //assert(0);
}
// Send the packet
LOG(dout_con<<m_connection->getDesc()
<<" INFO: dropped packet for non existent peer_id: "
<< peer_id << std::endl);
- assert(reliable && "trying to send raw packet reliable but no peer found!");
+ FATAL_ERROR_IF(!reliable, "Trying to send raw packet reliable but no peer found!");
return false;
}
Channel *channel = &(dynamic_cast<UDPPeer*>(&peer)->channels[channelnum]);
<<" INFO: queueing reliable packet for peer_id: " << peer_id
<<" channel: " << channelnum
<<" seqnum: " << seqnum << std::endl);
- channel->queued_reliables.push_back(p);
+ channel->queued_reliables.push(p);
return false;
}
}
void ConnectionSendThread::processReliableCommand(ConnectionCommand &c)
{
- assert(c.reliable);
+ assert(c.reliable); // Pre-condition
switch(c.type) {
case CONNCMD_NONE:
case CONNCMD_CONNECT:
case CONNCMD_DISCONNECT:
case CONCMD_ACK:
- assert("Got command that shouldn't be reliable as reliable command" == 0);
+ FATAL_ERROR("Got command that shouldn't be reliable as reliable command");
default:
LOG(dout_con<<m_connection->getDesc()
<<" Invalid reliable command type: " << c.type <<std::endl);
void ConnectionSendThread::processNonReliableCommand(ConnectionCommand &c)
{
- assert(!c.reliable);
+ assert(!c.reliable); // Pre-condition
switch(c.type) {
case CONNCMD_NONE:
sendAsPacket(c.peer_id,c.channelnum,c.data,true);
return;
case CONCMD_CREATE_PEER:
- assert("Got command that should be reliable as unreliable command" == 0);
+ FATAL_ERROR("Got command that should be reliable as unreliable command");
default:
LOG(dout_con<<m_connection->getDesc()
<<" Invalid command type: " << c.type <<std::endl);
// 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()
void ConnectionSendThread::send(u16 peer_id, u8 channelnum,
SharedBuffer<u8> data)
{
- assert(channelnum < CHANNEL_COUNT);
+ assert(channelnum < CHANNEL_COUNT); // Pre-condition
PeerHelper peer = m_connection->getPeerNoEx(peer_id);
if (!peer)
< dynamic_cast<UDPPeer*>(&peer)->channels[i].getWindowSize())&&
(peer->m_increment_packets_remaining > 0))
{
- BufferedPacket p = dynamic_cast<UDPPeer*>(&peer)->channels[i].queued_reliables.pop_front();
+ BufferedPacket p = dynamic_cast<UDPPeer*>(&peer)->channels[i].queued_reliables.front();
+ dynamic_cast<UDPPeer*>(&peer)->channels[i].queued_reliables.pop();
Channel* channel = &(dynamic_cast<UDPPeer*>(&peer)->channels[i]);
LOG(dout_con<<m_connection->getDesc()
<<" INFO: sending a queued reliable packet "
unsigned int initial_queuesize = m_outgoing_queue.size();
/* send non reliable packets*/
for(unsigned int i=0;i < initial_queuesize;i++) {
- OutgoingPacket packet = m_outgoing_queue.pop_front();
+ OutgoingPacket packet = m_outgoing_queue.front();
+ m_outgoing_queue.pop();
- assert(!packet.reliable &&
- "reliable packets are not allowed in outgoing queue!");
+ if (packet.reliable)
+ continue;
PeerHelper peer = m_connection->getPeerNoEx(packet.peer_id);
if (!peer) {
peer->m_increment_packets_remaining--;
}
else {
- m_outgoing_queue.push_back(packet);
+ m_outgoing_queue.push(packet);
pending_unreliable[packet.peer_id] = true;
}
}
SharedBuffer<u8> data, bool ack)
{
OutgoingPacket packet(peer_id, channelnum, data, false, ack);
- m_outgoing_queue.push_back(packet);
+ m_outgoing_queue.push(packet);
}
ConnectionReceiveThread::ConnectionReceiveThread(unsigned int max_packet_size) :
if (MAX_UDP_PEERS <= 65535 && peer_id >= MAX_UDP_PEERS) {
errorstream << "Something is wrong with peer_id" << std::endl;
- assert(0);
+ FATAL_ERROR("");
}
if (type == TYPE_CONTROL)
if (controltype == CONTROLTYPE_ACK)
{
- assert(channel != 0);
+ FATAL_ERROR_IF(channel == 0, "Invalid channel (0)");
if (packetdata.getSize() < 4)
throw InvalidIncomingDataException
("packetdata.getSize() < 4 (ACK header size)");
}
else if (type == TYPE_RELIABLE)
{
- assert(channel != 0);
+ FATAL_ERROR_IF(channel == 0, "Invalid channel (0)");
// Recursive reliable packets not allowed
if (reliable)
throw InvalidIncomingDataException("Found nested reliable packets");
}
// We should never get here.
- // If you get here, add an exception or a return to some of the
- // above conditionals.
- assert(0);
- throw BaseException("Error in Channel::ProcessPacket()");
+ FATAL_ERROR("Invalid execution point");
}
/*
/* Internal stuff */
void Connection::putEvent(ConnectionEvent &e)
{
- assert(e.type != CONNEVENT_NONE);
+ assert(e.type != CONNEVENT_NONE); // Pre-condition
m_event_queue.push_back(e);
}
}
// Error checking
- assert(node->second->id == peer_id);
+ FATAL_ERROR_IF(node->second->id != peer_id, "Invalid peer id");
return PeerHelper(node->second);
}
}
// Error checking
- assert(node->second->id == peer_id);
+ FATAL_ERROR_IF(node->second->id != peer_id, "Invalid peer id");
return PeerHelper(node->second);
}
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?)");
void Connection::Send(u16 peer_id, u8 channelnum,
NetworkPacket* pkt, bool reliable)
{
- assert(channelnum < CHANNEL_COUNT);
+ assert(channelnum < CHANNEL_COUNT); // Pre-condition
ConnectionCommand c;
- c.send(peer_id, channelnum, pkt->oldForgePacket(), reliable);
+ c.send(peer_id, channelnum, pkt, reliable);
putCommand(c);
}
{
PeerHelper peer = getPeerNoEx(PEER_ID_SERVER);
- if (!peer) {
- assert("Connection::getLocalStat we couldn't get our own peer? are you serious???" == 0);
- }
+ FATAL_ERROR_IF(!peer, "Connection::getLocalStat we couldn't get our own peer? are you serious???");
float retval = 0.0;
retval += dynamic_cast<UDPPeer*>(&peer)->channels[j].getCurrentLossRateKB();
break;
default:
- assert("Connection::getLocalStat Invalid stat type" == 0);
+ FATAL_ERROR("Connection::getLocalStat Invalid stat type");
}
}
return retval;
void Connection::sendAck(u16 peer_id, u8 channelnum, u16 seqnum)
{
- assert(channelnum < CHANNEL_COUNT);
+ assert(channelnum < CHANNEL_COUNT); // Pre-condition
LOG(dout_con<<getDesc()
<<" Queuing ACK command to peer_id: " << peer_id <<
projects / oweals / minetest.git / blobdiff