Skip to content
Permalink
Browse files

Rework packet receiving in ServerThread

Notably it tries to receive all queued packets
between server steps, not just one.
  • Loading branch information
sfan5 committed Nov 14, 2019
1 parent 9776460 commit c10952b5742ba4364ac7e6fc9f9f2b44c73357b5
Showing with 86 additions and 24 deletions.
  1. +22 −5 src/network/connection.cpp
  2. +3 −0 src/network/connection.h
  3. +9 −0 src/network/networkpacket.cpp
  4. +1 −0 src/network/networkpacket.h
  5. +51 −19 src/server.cpp
@@ -1323,24 +1323,29 @@ void Connection::Disconnect()
putCommand(c);
}

void Connection::Receive(NetworkPacket* pkt)
bool Connection::Receive(NetworkPacket *pkt, u32 timeout)
{
/*
Note that this function can potentially wait infinitely if non-data
events keep happening before the timeout expires.
This is not considered to be a problem (is it?)
*/
for(;;) {
ConnectionEvent e = waitEvent(m_bc_receive_timeout);
ConnectionEvent e = waitEvent(timeout);
if (e.type != CONNEVENT_NONE)
LOG(dout_con << getDesc() << ": Receive: got event: "
<< e.describe() << std::endl);
switch(e.type) {
case CONNEVENT_NONE:
throw NoIncomingDataException("No incoming data");
return false;
case CONNEVENT_DATA_RECEIVED:
// 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;
return true;
case CONNEVENT_PEER_ADDED: {
UDPPeer tmp(e.peer_id, e.address, this);
if (m_bc_peerhandler)
@@ -1358,7 +1363,19 @@ void Connection::Receive(NetworkPacket* pkt)
"(port already in use?)");
}
}
throw NoIncomingDataException("No incoming data");
return false;
}

void Connection::Receive(NetworkPacket *pkt)
{
bool any = Receive(pkt, m_bc_receive_timeout);
if (!any)
throw NoIncomingDataException("No incoming data");
}

bool Connection::TryReceive(NetworkPacket *pkt)
{
return Receive(pkt, 0);
}

void Connection::Send(session_t peer_id, u8 channelnum,
@@ -771,6 +771,7 @@ class Connection
bool Connected();
void Disconnect();
void Receive(NetworkPacket* pkt);
bool TryReceive(NetworkPacket *pkt);
void Send(session_t peer_id, u8 channelnum, NetworkPacket *pkt, bool reliable);
session_t GetPeerID() const { return m_peer_id; }
Address GetPeerAddress(session_t peer_id);
@@ -803,6 +804,8 @@ class Connection
UDPSocket m_udpSocket;
MutexedQueue<ConnectionCommand> m_command_queue;

bool Receive(NetworkPacket *pkt, u32 timeout);

void putEvent(ConnectionEvent &e);

void TriggerSend();
@@ -66,6 +66,15 @@ void NetworkPacket::putRawPacket(u8 *data, u32 datasize, session_t peer_id)
memcpy(m_data.data(), &data[2], m_datasize);
}

void NetworkPacket::clear()
{
m_data.clear();
m_datasize = 0;
m_read_offset = 0;
m_command = 0;
m_peer_id = 0;
}

const char* NetworkPacket::getString(u32 from_offset)
{
checkReadOffset(from_offset, 0);
@@ -35,6 +35,7 @@ class NetworkPacket
~NetworkPacket();

void putRawPacket(u8 *data, u32 datasize, session_t peer_id);
void clear();

// Getters
u32 getSize() const { return m_datasize; }
@@ -93,6 +93,15 @@ void *ServerThread::run()
{
BEGIN_DEBUG_EXCEPTION_HANDLER

/*
* The real business of the server happens on the ServerThread.
* How this works:
* AsyncRunStep() runs an actual server step as soon as enough time has
* passed (dedicated_server_loop keeps track of that).
* Receive() blocks at least(!) 30ms waiting for a packet (so this loop
* doesn't busy wait) and will process any remaining packets.
*/

m_server->AsyncRunStep(true);

while (!stopRequested()) {
@@ -101,7 +110,6 @@ void *ServerThread::run()

m_server->Receive();

} catch (con::NoIncomingDataException &e) {
} catch (con::PeerNotFoundException &e) {
infostream<<"Server: PeerNotFoundException"<<std::endl;
} catch (ClientNotFoundException &e) {
@@ -911,24 +919,43 @@ void Server::AsyncRunStep(bool initial_step)

void Server::Receive()
{
session_t peer_id = 0;
try {
NetworkPacket pkt;
m_con->Receive(&pkt);
peer_id = pkt.getPeerId();
ProcessData(&pkt);
} catch (const con::InvalidIncomingDataException &e) {
infostream << "Server::Receive(): InvalidIncomingDataException: what()="
<< e.what() << std::endl;
} catch (const SerializationError &e) {
infostream << "Server::Receive(): SerializationError: what()="
<< e.what() << std::endl;
} catch (const ClientStateError &e) {
errorstream << "ProcessData: peer=" << peer_id << e.what() << std::endl;
DenyAccess_Legacy(peer_id, L"Your client sent something server didn't expect."
L"Try reconnecting or updating your client");
} catch (const con::PeerNotFoundException &e) {
// Do nothing
NetworkPacket pkt;
session_t peer_id;
bool first = true;
for (;;) {
pkt.clear();
peer_id = 0;
try {
/*
In the first iteration *wait* for a packet, afterwards process
all packets that are immediately available (no waiting).
*/
if (first) {
m_con->Receive(&pkt);
first = false;
} else {
if (!m_con->TryReceive(&pkt))
return;
}

peer_id = pkt.getPeerId();
ProcessData(&pkt);
} catch (const con::InvalidIncomingDataException &e) {
infostream << "Server::Receive(): InvalidIncomingDataException: what()="
<< e.what() << std::endl;
} catch (const SerializationError &e) {
infostream << "Server::Receive(): SerializationError: what()="
<< e.what() << std::endl;
} catch (const ClientStateError &e) {
errorstream << "ProcessData: peer=" << peer_id << " what()="
<< e.what() << std::endl;
DenyAccess_Legacy(peer_id, L"Your client sent something server didn't expect."
L"Try reconnecting or updating your client");
} catch (const con::PeerNotFoundException &e) {
// Do nothing
} catch (const con::NoIncomingDataException &e) {
return;
}
}
}

@@ -3728,6 +3755,11 @@ void dedicated_server_loop(Server &server, bool &kill)
static thread_local const float profiler_print_interval =
g_settings->getFloat("profiler_print_interval");

/*
* The dedicated server loop only does time-keeping (in Server::step) and
* provides a way to main.cpp to kill the server externally (bool &kill).
*/

for(;;) {
// This is kind of a hack but can be done like this
// because server.step() is very light

0 comments on commit c10952b

Please sign in to comment.
You can’t perform that action at this time.