Connection::Receive(): receive Network Packet instead of SharedBuffer<u8>.

Because we get a Buffer<u8> from ConnectionEvent, don't convert it to SharedBuffer<u8> and return it to Server/Client::Receive which will convert it to NetworkPacket
Instead, put the Buffer<u8> directly to NetworkPacket and return it to packet processing
This remove a long existing memory copy
Also check the packet size directly into Connection::Receive instead of packet processing
This commit is contained in:
Loic Blot 2015-03-31 10:35:51 +02:00
parent ab77bf98ee
commit 1fe4256462
9 changed files with 104 additions and 98 deletions

View File

@ -834,10 +834,9 @@ void Client::ReceiveAll()
void Client::Receive() void Client::Receive()
{ {
DSTACK(__FUNCTION_NAME); DSTACK(__FUNCTION_NAME);
SharedBuffer<u8> data; NetworkPacket pkt;
u16 sender_peer_id; m_con.Receive(&pkt);
u32 datasize = m_con.Receive(sender_peer_id, data); ProcessData(&pkt);
ProcessData(*data, datasize, sender_peer_id);
} }
inline void Client::handleCommand(NetworkPacket* pkt) inline void Client::handleCommand(NetworkPacket* pkt)
@ -849,19 +848,12 @@ inline void Client::handleCommand(NetworkPacket* pkt)
/* /*
sender_peer_id given to this shall be quaranteed to be a valid peer sender_peer_id given to this shall be quaranteed to be a valid peer
*/ */
void Client::ProcessData(u8 *data, u32 datasize, u16 sender_peer_id) void Client::ProcessData(NetworkPacket *pkt)
{ {
DSTACK(__FUNCTION_NAME); DSTACK(__FUNCTION_NAME);
// Ignore packets that don't even fit a command ToClientCommand command = (ToClientCommand) pkt->getCommand();
if(datasize < 2) { u32 sender_peer_id = pkt->getPeerId();
m_packetcounter.add(60000);
return;
}
NetworkPacket pkt(data, datasize, sender_peer_id);
ToClientCommand command = (ToClientCommand) pkt.getCommand();
//infostream<<"Client: received command="<<command<<std::endl; //infostream<<"Client: received command="<<command<<std::endl;
m_packetcounter.add((u16)command); m_packetcounter.add((u16)command);
@ -889,7 +881,7 @@ void Client::ProcessData(u8 *data, u32 datasize, u16 sender_peer_id)
* as a byte mask * as a byte mask
*/ */
if(toClientCommandTable[command].state == TOCLIENT_STATE_NOT_CONNECTED) { if(toClientCommandTable[command].state == TOCLIENT_STATE_NOT_CONNECTED) {
handleCommand(&pkt); handleCommand(pkt);
return; return;
} }
@ -904,7 +896,7 @@ void Client::ProcessData(u8 *data, u32 datasize, u16 sender_peer_id)
Handle runtime commands Handle runtime commands
*/ */
handleCommand(&pkt); handleCommand(pkt);
} }
void Client::Send(NetworkPacket* pkt) void Client::Send(NetworkPacket* pkt)

View File

@ -392,7 +392,7 @@ class Client : public con::PeerHandler, public InventoryManager, public IGameDef
void handleCommand_LocalPlayerAnimations(NetworkPacket* pkt); void handleCommand_LocalPlayerAnimations(NetworkPacket* pkt);
void handleCommand_EyeOffset(NetworkPacket* pkt); void handleCommand_EyeOffset(NetworkPacket* pkt);
void ProcessData(u8 *data, u32 datasize, u16 sender_peer_id); void ProcessData(NetworkPacket *pkt);
// Returns true if something was received // Returns true if something was received
bool AsyncProcessPacket(); bool AsyncProcessPacket();

View File

@ -24,6 +24,7 @@ with this program; if not, write to the Free Software Foundation, Inc.,
#include "serialization.h" #include "serialization.h"
#include "log.h" #include "log.h"
#include "porting.h" #include "porting.h"
#include "network/networkpacket.h"
#include "util/serialize.h" #include "util/serialize.h"
#include "util/numeric.h" #include "util/numeric.h"
#include "util/string.h" #include "util/string.h"
@ -2884,30 +2885,36 @@ void Connection::Disconnect()
putCommand(c); putCommand(c);
} }
u32 Connection::Receive(u16 &peer_id, SharedBuffer<u8> &data) void Connection::Receive(NetworkPacket* pkt)
{ {
for(;;) { for(;;) {
ConnectionEvent e = waitEvent(m_bc_receive_timeout); ConnectionEvent e = waitEvent(m_bc_receive_timeout);
if (e.type != CONNEVENT_NONE) if (e.type != CONNEVENT_NONE)
LOG(dout_con<<getDesc()<<": Receive: got event: " LOG(dout_con << getDesc() << ": Receive: got event: "
<<e.describe()<<std::endl); << e.describe() << std::endl);
switch(e.type) { switch(e.type) {
case CONNEVENT_NONE: case CONNEVENT_NONE:
throw NoIncomingDataException("No incoming data"); throw NoIncomingDataException("No incoming data");
case CONNEVENT_DATA_RECEIVED: case CONNEVENT_DATA_RECEIVED:
peer_id = e.peer_id; // Data size is lesser than command size, ignoring packet
data = SharedBuffer<u8>(e.data); if (e.data.getSize() < 2) {
return e.data.getSize(); continue;
}
pkt->putRawPacket(*e.data, e.data.getSize(), e.peer_id);
return;
case CONNEVENT_PEER_ADDED: { case CONNEVENT_PEER_ADDED: {
UDPPeer tmp(e.peer_id, e.address, this); UDPPeer tmp(e.peer_id, e.address, this);
if (m_bc_peerhandler) if (m_bc_peerhandler)
m_bc_peerhandler->peerAdded(&tmp); m_bc_peerhandler->peerAdded(&tmp);
continue; } continue;
}
case CONNEVENT_PEER_REMOVED: { case CONNEVENT_PEER_REMOVED: {
UDPPeer tmp(e.peer_id, e.address, this); UDPPeer tmp(e.peer_id, e.address, this);
if (m_bc_peerhandler) if (m_bc_peerhandler)
m_bc_peerhandler->deletingPeer(&tmp, e.timeout); m_bc_peerhandler->deletingPeer(&tmp, e.timeout);
continue; } continue;
}
case CONNEVENT_BIND_FAILED: case CONNEVENT_BIND_FAILED:
throw ConnectionBindFailed("Failed to bind socket " throw ConnectionBindFailed("Failed to bind socket "
"(port already in use?)"); "(port already in use?)");

View File

@ -34,6 +34,8 @@ with this program; if not, write to the Free Software Foundation, Inc.,
#include <list> #include <list>
#include <map> #include <map>
class NetworkPacket;
namespace con namespace con
{ {
@ -1025,7 +1027,7 @@ class Connection
void Connect(Address address); void Connect(Address address);
bool Connected(); bool Connected();
void Disconnect(); void Disconnect();
u32 Receive(u16 &peer_id, SharedBuffer<u8> &data); void Receive(NetworkPacket* pkt);
void Send(u16 peer_id, u8 channelnum, NetworkPacket* pkt, bool reliable); void Send(u16 peer_id, u8 channelnum, NetworkPacket* pkt, bool reliable);
u16 GetPeerID() { return m_peer_id; } u16 GetPeerID() { return m_peer_id; }
Address GetPeerAddress(u16 peer_id); Address GetPeerAddress(u16 peer_id);

View File

@ -22,17 +22,6 @@ with this program; if not, write to the Free Software Foundation, Inc.,
#include "exceptions.h" #include "exceptions.h"
#include "util/serialize.h" #include "util/serialize.h"
NetworkPacket::NetworkPacket(u8 *data, u32 datasize, u16 peer_id):
m_read_offset(0), m_peer_id(peer_id)
{
m_read_offset = 0;
m_datasize = datasize - 2;
// split command and datas
m_command = readU16(&data[0]);
m_data = std::vector<u8>(&data[2], &data[2 + m_datasize]);
}
NetworkPacket::NetworkPacket(u16 command, u32 datasize, u16 peer_id): NetworkPacket::NetworkPacket(u16 command, u32 datasize, u16 peer_id):
m_datasize(datasize), m_read_offset(0), m_command(command), m_peer_id(peer_id) m_datasize(datasize), m_read_offset(0), m_command(command), m_peer_id(peer_id)
{ {
@ -50,6 +39,20 @@ NetworkPacket::~NetworkPacket()
m_data.clear(); m_data.clear();
} }
void NetworkPacket::putRawPacket(u8 *data, u32 datasize, u16 peer_id)
{
// If a m_command is already set, we are rewriting on same packet
// This is not permitted
assert(m_command == 0);
m_datasize = datasize - 2;
m_peer_id = peer_id;
// split command and datas
m_command = readU16(&data[0]);
m_data = std::vector<u8>(&data[2], &data[2 + m_datasize]);
}
char* NetworkPacket::getString(u32 from_offset) char* NetworkPacket::getString(u32 from_offset)
{ {
if (from_offset >= m_datasize) if (from_offset >= m_datasize)

View File

@ -28,11 +28,14 @@ class NetworkPacket
{ {
public: public:
NetworkPacket(u8 *data, u32 datasize, u16 peer_id);
NetworkPacket(u16 command, u32 datasize, u16 peer_id); NetworkPacket(u16 command, u32 datasize, u16 peer_id);
NetworkPacket(u16 command, u32 datasize); NetworkPacket(u16 command, u32 datasize);
NetworkPacket(): m_datasize(0), m_read_offset(0), m_command(0),
m_peer_id(0) {}
~NetworkPacket(); ~NetworkPacket();
void putRawPacket(u8 *data, u32 datasize, u16 peer_id);
// Getters // Getters
u32 getSize() { return m_datasize; } u32 getSize() { return m_datasize; }
u16 getPeerId() { return m_peer_id; } u16 getPeerId() { return m_peer_id; }

View File

@ -1018,10 +1018,11 @@ void Server::Receive()
DSTACK(__FUNCTION_NAME); DSTACK(__FUNCTION_NAME);
SharedBuffer<u8> data; SharedBuffer<u8> data;
u16 peer_id; u16 peer_id;
u32 datasize;
try { try {
datasize = m_con.Receive(peer_id,data); NetworkPacket pkt;
ProcessData(*data, datasize, peer_id); m_con.Receive(&pkt);
peer_id = pkt.getPeerId();
ProcessData(&pkt);
} }
catch(con::InvalidIncomingDataException &e) { catch(con::InvalidIncomingDataException &e) {
infostream<<"Server::Receive(): " infostream<<"Server::Receive(): "
@ -1149,13 +1150,14 @@ inline void Server::handleCommand(NetworkPacket* pkt)
(this->*opHandle.handler)(pkt); (this->*opHandle.handler)(pkt);
} }
void Server::ProcessData(u8 *data, u32 datasize, u16 peer_id) void Server::ProcessData(NetworkPacket *pkt)
{ {
DSTACK(__FUNCTION_NAME); DSTACK(__FUNCTION_NAME);
// Environment is locked first. // Environment is locked first.
JMutexAutoLock envlock(m_env_mutex); JMutexAutoLock envlock(m_env_mutex);
ScopeProfiler sp(g_profiler, "Server::ProcessData"); ScopeProfiler sp(g_profiler, "Server::ProcessData");
u32 peer_id = pkt->getPeerId();
try { try {
Address address = getPeerAddress(peer_id); Address address = getPeerAddress(peer_id);
@ -1179,18 +1181,13 @@ void Server::ProcessData(u8 *data, u32 datasize, u16 peer_id)
* respond for some time, your server was overloaded or * respond for some time, your server was overloaded or
* things like that. * things like that.
*/ */
infostream << "Server::ProcessData(): Cancelling: peer " infostream << "Server::ProcessData(): Canceling: peer "
<< peer_id << " not found" << std::endl; << peer_id << " not found" << std::endl;
return; return;
} }
try { try {
if(datasize < 2) ToServerCommand command = (ToServerCommand) pkt->getCommand();
return;
NetworkPacket pkt(data, datasize, peer_id);
ToServerCommand command = (ToServerCommand) pkt.getCommand();
// Command must be handled into ToServerCommandHandler // Command must be handled into ToServerCommandHandler
if (command >= TOSERVER_NUM_MSG_TYPES) { if (command >= TOSERVER_NUM_MSG_TYPES) {
@ -1199,7 +1196,7 @@ void Server::ProcessData(u8 *data, u32 datasize, u16 peer_id)
} }
if (toServerCommandTable[command].state == TOSERVER_STATE_NOT_CONNECTED) { if (toServerCommandTable[command].state == TOSERVER_STATE_NOT_CONNECTED) {
handleCommand(&pkt); handleCommand(pkt);
return; return;
} }
@ -1214,7 +1211,7 @@ void Server::ProcessData(u8 *data, u32 datasize, u16 peer_id)
/* Handle commands related to client startup */ /* Handle commands related to client startup */
if (toServerCommandTable[command].state == TOSERVER_STATE_STARTUP) { if (toServerCommandTable[command].state == TOSERVER_STATE_STARTUP) {
handleCommand(&pkt); handleCommand(pkt);
return; return;
} }
@ -1227,7 +1224,7 @@ void Server::ProcessData(u8 *data, u32 datasize, u16 peer_id)
return; return;
} }
handleCommand(&pkt); handleCommand(pkt);
} }
catch(SendFailedException &e) { catch(SendFailedException &e) {
errorstream << "Server::ProcessData(): SendFailedException: " errorstream << "Server::ProcessData(): SendFailedException: "

View File

@ -219,7 +219,7 @@ class Server : public con::PeerHandler, public MapEventReceiver,
void handleCommand_NodeMetaFields(NetworkPacket* pkt); void handleCommand_NodeMetaFields(NetworkPacket* pkt);
void handleCommand_InventoryFields(NetworkPacket* pkt); void handleCommand_InventoryFields(NetworkPacket* pkt);
void ProcessData(u8 *data, u32 datasize, u16 peer_id); void ProcessData(NetworkPacket *pkt);
void Send(NetworkPacket* pkt); void Send(NetworkPacket* pkt);

View File

@ -1939,13 +1939,12 @@ struct TestConnection: public TestBase
try try
{ {
u16 peer_id; NetworkPacket pkt;
SharedBuffer<u8> data; infostream << "** running client.Receive()" << std::endl;
infostream<<"** running client.Receive()"<<std::endl; client.Receive(&pkt);
u32 size = client.Receive(peer_id, data); infostream << "** Client received: peer_id=" << pkt.getPeerId()
infostream<<"** Client received: peer_id="<<peer_id << ", size=" << pkt.getSize()
<<", size="<<size << std::endl;
<<std::endl;
} }
catch(con::NoIncomingDataException &e) catch(con::NoIncomingDataException &e)
{ {
@ -1961,13 +1960,12 @@ struct TestConnection: public TestBase
try try
{ {
u16 peer_id; NetworkPacket pkt;
SharedBuffer<u8> data; infostream << "** running server.Receive()" << std::endl;
infostream<<"** running server.Receive()"<<std::endl; server.Receive(&pkt);
u32 size = server.Receive(peer_id, data); infostream<<"** Server received: peer_id=" << pkt.getPeerId()
infostream<<"** Server received: peer_id="<<peer_id << ", size=" << pkt.getSize()
<<", size="<<size << std::endl;
<<std::endl;
} }
catch(con::NoIncomingDataException &e) catch(con::NoIncomingDataException &e)
{ {
@ -1988,13 +1986,12 @@ struct TestConnection: public TestBase
{ {
try try
{ {
u16 peer_id; NetworkPacket pkt;
SharedBuffer<u8> data; infostream << "** running client.Receive()" << std::endl;
infostream<<"** running client.Receive()"<<std::endl; client.Receive(&pkt);
u32 size = client.Receive(peer_id, data); infostream << "** Client received: peer_id=" << pkt.getPeerId()
infostream<<"** Client received: peer_id="<<peer_id << ", size=" << pkt.getSize()
<<", size="<<size << std::endl;
<<std::endl;
} }
catch(con::NoIncomingDataException &e) catch(con::NoIncomingDataException &e)
{ {
@ -2006,13 +2003,12 @@ struct TestConnection: public TestBase
try try
{ {
u16 peer_id; NetworkPacket pkt;
SharedBuffer<u8> data; infostream << "** running server.Receive()" << std::endl;
infostream<<"** running server.Receive()"<<std::endl; server.Receive(&pkt);
u32 size = server.Receive(peer_id, data); infostream << "** Server received: peer_id=" << pkt.getPeerId()
infostream<<"** Server received: peer_id="<<peer_id << ", size=" << pkt.getSize()
<<", size="<<size << std::endl;
<<std::endl;
} }
catch(con::NoIncomingDataException &e) catch(con::NoIncomingDataException &e)
{ {
@ -2022,24 +2018,26 @@ struct TestConnection: public TestBase
Simple send-receive test Simple send-receive test
*/ */
{ {
NetworkPacket pkt((u8*) "Hello World !", 14, 0); NetworkPacket pkt;
pkt.putRawPacket((u8*) "Hello World !", 14, 0);
SharedBuffer<u8> sentdata = pkt.oldForgePacket(); Buffer<u8> sentdata = pkt.oldForgePacket();
infostream<<"** running client.Send()"<<std::endl; infostream<<"** running client.Send()"<<std::endl;
client.Send(PEER_ID_SERVER, 0, &pkt, true); client.Send(PEER_ID_SERVER, 0, &pkt, true);
sleep_ms(50); sleep_ms(50);
u16 peer_id; NetworkPacket recvpacket;
SharedBuffer<u8> recvdata;
infostream << "** running server.Receive()" << std::endl; infostream << "** running server.Receive()" << std::endl;
u32 size = server.Receive(peer_id, recvdata); server.Receive(&recvpacket);
infostream << "** Server received: peer_id=" << peer_id infostream << "** Server received: peer_id=" << pkt.getPeerId()
<< ", size=" << size << ", size=" << pkt.getSize()
<< ", data=" << (const char*)pkt.getU8Ptr(0) << ", data=" << (const char*)pkt.getU8Ptr(0)
<< std::endl; << std::endl;
Buffer<u8> recvdata = pkt.oldForgePacket();
UASSERT(memcmp(*sentdata, *recvdata, recvdata.getSize()) == 0); UASSERT(memcmp(*sentdata, *recvdata, recvdata.getSize()) == 0);
} }
@ -2061,29 +2059,33 @@ struct TestConnection: public TestBase
snprintf(buf, 10, "%.2X", ((int)((const char*)pkt.getU8Ptr(0))[i])&0xff); snprintf(buf, 10, "%.2X", ((int)((const char*)pkt.getU8Ptr(0))[i])&0xff);
infostream<<buf; infostream<<buf;
} }
if(datasize>20) if(datasize > 20)
infostream<<"..."; infostream << "...";
infostream<<std::endl; infostream << std::endl;
SharedBuffer<u8> sentdata = pkt.oldForgePacket(); Buffer<u8> sentdata = pkt.oldForgePacket();
server.Send(peer_id_client, 0, &pkt, true); server.Send(peer_id_client, 0, &pkt, true);
//sleep_ms(3000); //sleep_ms(3000);
SharedBuffer<u8> recvdata; Buffer<u8> recvdata;
infostream<<"** running client.Receive()"<<std::endl; infostream << "** running client.Receive()" << std::endl;
u16 peer_id = 132; u16 peer_id = 132;
u16 size = 0; u16 size = 0;
bool received = false; bool received = false;
u32 timems0 = porting::getTimeMs(); u32 timems0 = porting::getTimeMs();
for(;;){ for(;;) {
if(porting::getTimeMs() - timems0 > 5000 || received) if(porting::getTimeMs() - timems0 > 5000 || received)
break; break;
try{ try {
size = client.Receive(peer_id, recvdata); NetworkPacket pkt;
client.Receive(&pkt);
size = pkt.getSize();
peer_id = pkt.getPeerId();
recvdata = pkt.oldForgePacket();
received = true; received = true;
}catch(con::NoIncomingDataException &e){ } catch(con::NoIncomingDataException &e) {
} }
sleep_ms(10); sleep_ms(10);
} }