Network: Use unique_ptr for packets, use deque for packet queues
This commit is contained in:
@@ -26,7 +26,6 @@ Packet::Packet(NetworkSocketHandler *cs)
|
||||
assert(cs != nullptr);
|
||||
|
||||
this->cs = cs;
|
||||
this->next = nullptr;
|
||||
this->pos = 0; // We start reading from here
|
||||
this->size = 0;
|
||||
this->buffer = MallocT<byte>(SHRT_MAX);
|
||||
@@ -53,7 +52,6 @@ Packet::~Packet()
|
||||
void Packet::ResetState(PacketType type)
|
||||
{
|
||||
this->cs = nullptr;
|
||||
this->next = nullptr;
|
||||
|
||||
/* Skip the size so we can write that in before sending the packet */
|
||||
this->pos = 0;
|
||||
@@ -66,7 +64,7 @@ void Packet::ResetState(PacketType type)
|
||||
*/
|
||||
void Packet::PrepareToSend()
|
||||
{
|
||||
assert(this->cs == nullptr && this->next == nullptr);
|
||||
assert(this->cs == nullptr);
|
||||
|
||||
this->buffer[0] = GB(this->size, 0, 8);
|
||||
this->buffer[1] = GB(this->size, 8, 8);
|
||||
@@ -204,7 +202,7 @@ bool Packet::CanReadFromPacket(uint bytes_to_read, bool non_fatal)
|
||||
*/
|
||||
void Packet::ReadRawPacketSize()
|
||||
{
|
||||
assert(this->cs != nullptr && this->next == nullptr);
|
||||
assert(this->cs != nullptr);
|
||||
this->size = (PacketSize)this->buffer[0];
|
||||
this->size += (PacketSize)this->buffer[1] << 8;
|
||||
}
|
||||
|
@@ -39,8 +39,6 @@ typedef uint8 PacketType; ///< Identifier for the packet
|
||||
* (year % 4 == 0) and ((year % 100 != 0) or (year % 400 == 0))
|
||||
*/
|
||||
struct Packet {
|
||||
/** The next packet. Used for queueing packets before sending. */
|
||||
Packet *next;
|
||||
/**
|
||||
* The size of the whole packet for received packets. For packets
|
||||
* that will be sent, the value is filled in just before the
|
||||
|
@@ -22,7 +22,6 @@
|
||||
*/
|
||||
NetworkTCPSocketHandler::NetworkTCPSocketHandler(SOCKET s) :
|
||||
NetworkSocketHandler(),
|
||||
packet_queue(nullptr), packet_recv(nullptr),
|
||||
sock(s), writable(false)
|
||||
{
|
||||
}
|
||||
@@ -41,13 +40,8 @@ NetworkRecvStatus NetworkTCPSocketHandler::CloseConnection(bool error)
|
||||
NetworkSocketHandler::CloseConnection(error);
|
||||
|
||||
/* Free all pending and partially received packets */
|
||||
while (this->packet_queue != nullptr) {
|
||||
Packet *p = this->packet_queue->next;
|
||||
delete this->packet_queue;
|
||||
this->packet_queue = p;
|
||||
}
|
||||
delete this->packet_recv;
|
||||
this->packet_recv = nullptr;
|
||||
this->packet_queue.clear();
|
||||
this->packet_recv.reset();
|
||||
|
||||
return NETWORK_RECV_STATUS_OKAY;
|
||||
}
|
||||
@@ -58,9 +52,8 @@ NetworkRecvStatus NetworkTCPSocketHandler::CloseConnection(bool error)
|
||||
* if the OS-network-buffer is full)
|
||||
* @param packet the packet to send
|
||||
*/
|
||||
void NetworkTCPSocketHandler::SendPacket(Packet *packet)
|
||||
void NetworkTCPSocketHandler::SendPacket(std::unique_ptr<Packet> packet)
|
||||
{
|
||||
Packet *p;
|
||||
assert(packet != nullptr);
|
||||
|
||||
packet->PrepareToSend();
|
||||
@@ -70,16 +63,7 @@ void NetworkTCPSocketHandler::SendPacket(Packet *packet)
|
||||
* to do a denial of service attack! */
|
||||
if (packet->size < ((SHRT_MAX * 2) / 3)) packet->buffer = ReallocT(packet->buffer, packet->size);
|
||||
|
||||
/* Locate last packet buffered for the client */
|
||||
p = this->packet_queue;
|
||||
if (p == nullptr) {
|
||||
/* No packets yet */
|
||||
this->packet_queue = packet;
|
||||
} else {
|
||||
/* Skip to the last packet */
|
||||
while (p->next != nullptr) p = p->next;
|
||||
p->next = packet;
|
||||
}
|
||||
this->packet_queue.push_back(std::move(packet));
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -95,14 +79,13 @@ void NetworkTCPSocketHandler::SendPacket(Packet *packet)
|
||||
SendPacketsState NetworkTCPSocketHandler::SendPackets(bool closing_down)
|
||||
{
|
||||
ssize_t res;
|
||||
Packet *p;
|
||||
|
||||
/* We can not write to this socket!! */
|
||||
if (!this->writable) return SPS_NONE_SENT;
|
||||
if (!this->IsConnected()) return SPS_CLOSED;
|
||||
|
||||
p = this->packet_queue;
|
||||
while (p != nullptr) {
|
||||
while (!this->packet_queue.empty()) {
|
||||
Packet *p = this->packet_queue.front().get();
|
||||
res = send(this->sock, (const char*)p->buffer + p->pos, p->size - p->pos, 0);
|
||||
if (res == -1) {
|
||||
int err = GET_LAST_ERROR();
|
||||
@@ -127,9 +110,7 @@ SendPacketsState NetworkTCPSocketHandler::SendPackets(bool closing_down)
|
||||
/* Is this packet sent? */
|
||||
if (p->pos == p->size) {
|
||||
/* Go to the next packet */
|
||||
this->packet_queue = p->next;
|
||||
delete p;
|
||||
p = this->packet_queue;
|
||||
this->packet_queue.pop_front();
|
||||
} else {
|
||||
return SPS_PARTLY_SENT;
|
||||
}
|
||||
@@ -142,17 +123,17 @@ SendPacketsState NetworkTCPSocketHandler::SendPackets(bool closing_down)
|
||||
* Receives a packet for the given client
|
||||
* @return The received packet (or nullptr when it didn't receive one)
|
||||
*/
|
||||
Packet *NetworkTCPSocketHandler::ReceivePacket()
|
||||
std::unique_ptr<Packet> NetworkTCPSocketHandler::ReceivePacket()
|
||||
{
|
||||
ssize_t res;
|
||||
|
||||
if (!this->IsConnected()) return nullptr;
|
||||
|
||||
if (this->packet_recv == nullptr) {
|
||||
this->packet_recv = new Packet(this);
|
||||
this->packet_recv.reset(new Packet(this));
|
||||
}
|
||||
|
||||
Packet *p = this->packet_recv;
|
||||
Packet *p = this->packet_recv.get();
|
||||
|
||||
/* Read packet size */
|
||||
if (p->pos < sizeof(PacketSize)) {
|
||||
@@ -205,11 +186,11 @@ Packet *NetworkTCPSocketHandler::ReceivePacket()
|
||||
p->pos += res;
|
||||
}
|
||||
|
||||
/* Prepare for receiving a new packet */
|
||||
this->packet_recv = nullptr;
|
||||
|
||||
p->PrepareToRead();
|
||||
return p;
|
||||
|
||||
/* Prepare for receiving a new packet */
|
||||
return std::move(this->packet_recv);
|
||||
}
|
||||
|
||||
/**
|
||||
|
@@ -15,6 +15,9 @@
|
||||
#include "address.h"
|
||||
#include "packet.h"
|
||||
|
||||
#include <deque>
|
||||
#include <memory>
|
||||
|
||||
/** The states of sending the packets. */
|
||||
enum SendPacketsState {
|
||||
SPS_CLOSED, ///< The connection got closed.
|
||||
@@ -26,8 +29,8 @@ enum SendPacketsState {
|
||||
/** Base socket handler for all TCP sockets */
|
||||
class NetworkTCPSocketHandler : public NetworkSocketHandler {
|
||||
private:
|
||||
Packet *packet_queue; ///< Packets that are awaiting delivery
|
||||
Packet *packet_recv; ///< Partially received packet
|
||||
std::deque<std::unique_ptr<Packet>> packet_queue; ///< Packets that are awaiting delivery
|
||||
std::unique_ptr<Packet> packet_recv; ///< Partially received packet
|
||||
public:
|
||||
SOCKET sock; ///< The socket currently connected to
|
||||
bool writable; ///< Can we write to this socket?
|
||||
@@ -39,10 +42,16 @@ public:
|
||||
bool IsConnected() const { return this->sock != INVALID_SOCKET; }
|
||||
|
||||
NetworkRecvStatus CloseConnection(bool error = true) override;
|
||||
virtual void SendPacket(Packet *packet);
|
||||
virtual void SendPacket(std::unique_ptr<Packet> packet);
|
||||
|
||||
void SendPacket(Packet *packet)
|
||||
{
|
||||
this->SendPacket(std::unique_ptr<Packet>(packet));
|
||||
}
|
||||
|
||||
SendPacketsState SendPackets(bool closing_down = false);
|
||||
|
||||
virtual Packet *ReceivePacket();
|
||||
virtual std::unique_ptr<Packet> ReceivePacket();
|
||||
|
||||
bool CanSendReceive();
|
||||
|
||||
@@ -50,7 +59,7 @@ public:
|
||||
* Whether there is something pending in the send queue.
|
||||
* @return true when something is pending in the send queue.
|
||||
*/
|
||||
bool HasSendQueue() { return this->packet_queue != nullptr; }
|
||||
bool HasSendQueue() { return !this->packet_queue.empty(); }
|
||||
|
||||
NetworkTCPSocketHandler(SOCKET s = INVALID_SOCKET);
|
||||
~NetworkTCPSocketHandler();
|
||||
|
@@ -112,9 +112,9 @@ NetworkRecvStatus NetworkAdminSocketHandler::HandlePacket(Packet *p)
|
||||
*/
|
||||
NetworkRecvStatus NetworkAdminSocketHandler::ReceivePackets()
|
||||
{
|
||||
Packet *p;
|
||||
std::unique_ptr<Packet> p;
|
||||
while ((p = this->ReceivePacket()) != nullptr) {
|
||||
NetworkRecvStatus res = this->HandlePacket(p);
|
||||
NetworkRecvStatus res = this->HandlePacket(p.get());
|
||||
if (res != NETWORK_RECV_STATUS_OKAY) return res;
|
||||
}
|
||||
|
||||
|
@@ -204,12 +204,11 @@ bool NetworkContentSocketHandler::ReceivePackets()
|
||||
*
|
||||
* What arbitrary number to choose is the ultimate question though.
|
||||
*/
|
||||
Packet *p;
|
||||
std::unique_ptr<Packet> p;
|
||||
static const int MAX_PACKETS_TO_RECEIVE = 42;
|
||||
int i = MAX_PACKETS_TO_RECEIVE;
|
||||
while (--i != 0 && (p = this->ReceivePacket()) != nullptr) {
|
||||
bool cont = this->HandlePacket(p);
|
||||
delete p;
|
||||
bool cont = this->HandlePacket(p.get());
|
||||
if (!cont) return true;
|
||||
}
|
||||
|
||||
|
@@ -137,10 +137,9 @@ NetworkRecvStatus NetworkGameSocketHandler::HandlePacket(Packet *p)
|
||||
*/
|
||||
NetworkRecvStatus NetworkGameSocketHandler::ReceivePackets()
|
||||
{
|
||||
Packet *p;
|
||||
std::unique_ptr<Packet> p;
|
||||
while ((p = this->ReceivePacket()) != nullptr) {
|
||||
NetworkRecvStatus res = HandlePacket(p);
|
||||
delete p;
|
||||
NetworkRecvStatus res = HandlePacket(p.get());
|
||||
if (res != NETWORK_RECV_STATUS_OKAY) return res;
|
||||
}
|
||||
|
||||
|
@@ -60,9 +60,9 @@ template SocketList TCPListenHandler<ServerNetworkGameSocketHandler, PACKET_SERV
|
||||
/** Writing a savegame directly to a number of packets. */
|
||||
struct PacketWriter : SaveFilter {
|
||||
ServerNetworkGameSocketHandler *cs; ///< Socket we are associated with.
|
||||
Packet *current; ///< The packet we're currently writing to.
|
||||
std::unique_ptr<Packet> current; ///< The packet we're currently writing to.
|
||||
size_t total_size; ///< Total size of the compressed savegame.
|
||||
Packet *packets; ///< Packet queue of the savegame; send these "slowly" to the client.
|
||||
std::deque<std::unique_ptr<Packet>> packets; ///< Packet queue of the savegame; send these "slowly" to the client.
|
||||
std::mutex mutex; ///< Mutex for making threaded saving safe.
|
||||
std::condition_variable exit_sig; ///< Signal for threaded destruction of this packet writer.
|
||||
|
||||
@@ -70,7 +70,7 @@ struct PacketWriter : SaveFilter {
|
||||
* Create the packet writer.
|
||||
* @param cs The socket handler we're making the packets for.
|
||||
*/
|
||||
PacketWriter(ServerNetworkGameSocketHandler *cs) : SaveFilter(nullptr), cs(cs), current(nullptr), total_size(0), packets(nullptr)
|
||||
PacketWriter(ServerNetworkGameSocketHandler *cs) : SaveFilter(nullptr), cs(cs), total_size(0)
|
||||
{
|
||||
}
|
||||
|
||||
@@ -83,13 +83,8 @@ struct PacketWriter : SaveFilter {
|
||||
|
||||
/* This must all wait until the Destroy function is called. */
|
||||
|
||||
while (this->packets != nullptr) {
|
||||
Packet *p = this->packets->next;
|
||||
delete this->packets;
|
||||
this->packets = p;
|
||||
}
|
||||
|
||||
delete this->current;
|
||||
this->packets.clear();
|
||||
this->current.reset();
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -129,21 +124,19 @@ struct PacketWriter : SaveFilter {
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(this->mutex);
|
||||
|
||||
return this->packets != nullptr;
|
||||
return !this->packets.empty();
|
||||
}
|
||||
|
||||
/**
|
||||
* Pop a single created packet from the queue with packets.
|
||||
*/
|
||||
Packet *PopPacket()
|
||||
std::unique_ptr<Packet> PopPacket()
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(this->mutex);
|
||||
|
||||
Packet *p = this->packets;
|
||||
if (p == nullptr) return nullptr;
|
||||
this->packets = p->next;
|
||||
p->next = nullptr;
|
||||
|
||||
if (this->packets.empty()) return nullptr;
|
||||
std::unique_ptr<Packet> p = std::move(this->packets.front());
|
||||
this->packets.pop_front();
|
||||
return p;
|
||||
}
|
||||
|
||||
@@ -152,13 +145,14 @@ struct PacketWriter : SaveFilter {
|
||||
{
|
||||
if (this->current == nullptr) return;
|
||||
|
||||
Packet **p = &this->packets;
|
||||
while (*p != nullptr) {
|
||||
p = &(*p)->next;
|
||||
this->packets.push_back(std::move(this->current));
|
||||
}
|
||||
*p = this->current;
|
||||
|
||||
this->current = nullptr;
|
||||
void PrependQueue(std::unique_ptr<Packet> p)
|
||||
{
|
||||
if (p == nullptr) return;
|
||||
|
||||
this->packets.push_front(std::move(p));
|
||||
}
|
||||
|
||||
void Write(byte *buf, size_t size) override
|
||||
@@ -166,7 +160,7 @@ struct PacketWriter : SaveFilter {
|
||||
/* We want to abort the saving when the socket is closed. */
|
||||
if (this->cs == nullptr) SlError(STR_NETWORK_ERROR_LOSTCONNECTION);
|
||||
|
||||
if (this->current == nullptr) this->current = new Packet(PACKET_SERVER_MAP_DATA);
|
||||
if (this->current == nullptr) this->current.reset(new Packet(PACKET_SERVER_MAP_DATA));
|
||||
|
||||
std::lock_guard<std::mutex> lock(this->mutex);
|
||||
|
||||
@@ -179,7 +173,7 @@ struct PacketWriter : SaveFilter {
|
||||
|
||||
if (this->current->size == SHRT_MAX) {
|
||||
this->AppendQueue();
|
||||
if (buf != bufe) this->current = new Packet(PACKET_SERVER_MAP_DATA);
|
||||
if (buf != bufe) this->current.reset(new Packet(PACKET_SERVER_MAP_DATA));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -197,13 +191,13 @@ struct PacketWriter : SaveFilter {
|
||||
this->AppendQueue();
|
||||
|
||||
/* Add a packet stating that this is the end to the queue. */
|
||||
this->current = new Packet(PACKET_SERVER_MAP_DONE);
|
||||
this->current.reset(new Packet(PACKET_SERVER_MAP_DONE));
|
||||
this->AppendQueue();
|
||||
|
||||
/* Fast-track the size to the client. */
|
||||
Packet *p = new Packet(PACKET_SERVER_MAP_SIZE);
|
||||
std::unique_ptr<Packet> p(new Packet(PACKET_SERVER_MAP_SIZE));
|
||||
p->Send_uint32((uint32)this->total_size);
|
||||
this->cs->NetworkTCPSocketHandler::SendPacket(p);
|
||||
this->PrependQueue(std::move(p));
|
||||
}
|
||||
};
|
||||
|
||||
@@ -241,7 +235,7 @@ ServerNetworkGameSocketHandler::~ServerNetworkGameSocketHandler()
|
||||
}
|
||||
}
|
||||
|
||||
Packet *ServerNetworkGameSocketHandler::ReceivePacket()
|
||||
std::unique_ptr<Packet> ServerNetworkGameSocketHandler::ReceivePacket()
|
||||
{
|
||||
/* Only allow receiving when we have some buffer free; this value
|
||||
* can go negative, but eventually it will become positive again. */
|
||||
@@ -249,7 +243,7 @@ Packet *ServerNetworkGameSocketHandler::ReceivePacket()
|
||||
|
||||
/* We can receive a packet, so try that and if needed account for
|
||||
* the amount of received data. */
|
||||
Packet *p = this->NetworkTCPSocketHandler::ReceivePacket();
|
||||
std::unique_ptr<Packet> p = this->NetworkTCPSocketHandler::ReceivePacket();
|
||||
if (p != nullptr) this->receive_limit -= p->size;
|
||||
return p;
|
||||
}
|
||||
@@ -627,14 +621,14 @@ NetworkRecvStatus ServerNetworkGameSocketHandler::SendMap()
|
||||
bool has_packets = true;
|
||||
|
||||
for (uint i = 0; i < sent_packets; i++) {
|
||||
Packet *p = this->savegame->PopPacket();
|
||||
std::unique_ptr<Packet> p = this->savegame->PopPacket();
|
||||
if (p == nullptr) {
|
||||
has_packets = false;
|
||||
break;
|
||||
}
|
||||
last_packet = p->buffer[2] == PACKET_SERVER_MAP_DONE;
|
||||
|
||||
this->SendPacket(p);
|
||||
this->SendPacket(std::move(p));
|
||||
|
||||
if (last_packet) {
|
||||
/* There is no more data, so break the for */
|
||||
|
@@ -85,7 +85,7 @@ public:
|
||||
ServerNetworkGameSocketHandler(SOCKET s);
|
||||
~ServerNetworkGameSocketHandler();
|
||||
|
||||
virtual Packet *ReceivePacket() override;
|
||||
virtual std::unique_ptr<Packet> ReceivePacket() override;
|
||||
NetworkRecvStatus CloseConnection(NetworkRecvStatus status) override;
|
||||
void GetClientName(char *client_name, const char *last) const;
|
||||
|
||||
|
Reference in New Issue
Block a user