pa/eqemu-server
2
0
Fork 0
mirror of https://github.com/EQEmu/Server.git synced 2025-12-13 10:31:29 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Refactoring streams, will be an ongoing process.

Browse Source
This commit is contained in:
KimLS 2025-01-28 23:39:51 -08:00
parent f2f05479dc
commit 29bdc86d12
2 changed files with 56 additions and 56 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

37
common/net/daybreak_connection.cpp
View File

@ -337,20 +337,10 @@ EQ::Net::DaybreakConnection::DaybreakConnection(DaybreakConnectionManager *owner
m_combined[1] = OP_Combined; m_combined[1] = OP_Combined;
m_last_session_stats = Clock::now(); m_last_session_stats = Clock::now();
m_outgoing_budget = owner->m_options.outgoing_data_rate; m_outgoing_budget = owner->m_options.outgoing_data_rate;
m_streams = new DaybreakStream*[4];
m_streams[0] = nullptr;
m_streams[1] = nullptr;
m_streams[2] = nullptr;
m_streams[3] = nullptr;
} }
EQ::Net::DaybreakConnection::~DaybreakConnection() EQ::Net::DaybreakConnection::~DaybreakConnection()
{ {
for (int i = 0; i < 4; ++i) {
if (m_streams[i] != nullptr) {
delete m_streams[i];
}
}
} }
void EQ::Net::DaybreakConnection::Close() void EQ::Net::DaybreakConnection::Close()
@ -505,7 +495,7 @@ void EQ::Net::DaybreakConnection::ProcessPacket(Packet &p)
void EQ::Net::DaybreakConnection::ProcessQueue() void EQ::Net::DaybreakConnection::ProcessQueue()
{ {
for (int i = 0; i < 4; ++i) { for (int i = 0; i < 4; ++i) {
auto stream = m_streams[i]; auto &stream = m_streams[i];
for (;;) { for (;;) {
if (stream == nullptr) if (stream == nullptr)
break; break;
@ -525,7 +515,7 @@ void EQ::Net::DaybreakConnection::ProcessQueue()
void EQ::Net::DaybreakConnection::RemoveFromQueue(int stream, uint16_t seq) void EQ::Net::DaybreakConnection::RemoveFromQueue(int stream, uint16_t seq)
{ {
auto s = m_streams[stream]; auto &s = m_streams[stream];
if (s == nullptr) { if (s == nullptr) {
return; return;
} }
@ -540,7 +530,7 @@ void EQ::Net::DaybreakConnection::RemoveFromQueue(int stream, uint16_t seq)
void EQ::Net::DaybreakConnection::AddToQueue(int stream, uint16_t seq, const Packet &p) void EQ::Net::DaybreakConnection::AddToQueue(int stream, uint16_t seq, const Packet &p)
{ {
auto s = m_streams[stream]; auto &s = m_streams[stream];
if (s == nullptr) { if (s == nullptr) {
return; return;
} }
@ -699,10 +689,9 @@ void EQ::Net::DaybreakConnection::ProcessDecodedPacket(const Packet &p)
auto header = p.GetSerialize<DaybreakReliableHeader>(0); auto header = p.GetSerialize<DaybreakReliableHeader>(0);
auto sequence = NetworkToHost(header.sequence); auto sequence = NetworkToHost(header.sequence);
auto stream_id = header.opcode - OP_Packet; auto stream_id = header.opcode - OP_Packet;
auto stream = m_streams[stream_id]; auto &stream = m_streams[stream_id];
if (stream == nullptr) { if (stream == nullptr) {
stream = new DaybreakStream(); stream.reset(new DaybreakReliableStream());
m_streams[stream_id] = stream;
} }
auto order = CompareSequence(stream->sequence_in, sequence); auto order = CompareSequence(stream->sequence_in, sequence);
@ -732,10 +721,9 @@ void EQ::Net::DaybreakConnection::ProcessDecodedPacket(const Packet &p)
auto header = p.GetSerialize<DaybreakReliableHeader>(0); auto header = p.GetSerialize<DaybreakReliableHeader>(0);
auto sequence = NetworkToHost(header.sequence); auto sequence = NetworkToHost(header.sequence);
auto stream_id = header.opcode - OP_Fragment; auto stream_id = header.opcode - OP_Fragment;
auto stream = m_streams[stream_id]; auto &stream = m_streams[stream_id];
if (stream == nullptr) { if (stream == nullptr) {
stream = new DaybreakStream(); stream.reset(new DaybreakReliableStream());
m_streams[stream_id] = stream;
} }
auto order = CompareSequence(stream->sequence_in, sequence); auto order = CompareSequence(stream->sequence_in, sequence);
@ -1130,7 +1118,7 @@ void EQ::Net::DaybreakConnection::ProcessResend(int stream)
auto resends = 0; auto resends = 0;
auto now = Clock::now(); auto now = Clock::now();
auto s = m_streams[stream]; auto &s = m_streams[stream];
if (s == nullptr) { if (s == nullptr) {
return; return;
} }
@ -1196,7 +1184,7 @@ void EQ::Net::DaybreakConnection::Ack(int stream, uint16_t seq)
{ {
auto now = Clock::now(); auto now = Clock::now();
auto s = m_streams[stream]; auto &s = m_streams[stream];
if (s == nullptr) { if (s == nullptr) {
return; return;
} }
@ -1224,7 +1212,7 @@ void EQ::Net::DaybreakConnection::Ack(int stream, uint16_t seq)
void EQ::Net::DaybreakConnection::OutOfOrderAck(int stream, uint16_t seq) void EQ::Net::DaybreakConnection::OutOfOrderAck(int stream, uint16_t seq)
{ {
auto now = Clock::now(); auto now = Clock::now();
auto s = m_streams[stream]; auto &s = m_streams[stream];
if (s == nullptr) { if (s == nullptr) {
return; return;
} }
@ -1444,10 +1432,9 @@ void EQ::Net::DaybreakConnection::InternalQueuePacket(Packet &p, int stream_id,
return; return;
} }
auto stream = m_streams[stream_id]; auto &stream = m_streams[stream_id];
if (stream == nullptr) { if (stream == nullptr) {
stream = new DaybreakStream(); stream.reset(new DaybreakReliableStream());
m_streams[stream_id] = stream;
} }
auto max_raw_size = m_max_packet_size - m_crc_bytes - DaybreakReliableHeader::size() - 1; // -1 for compress flag auto max_raw_size = m_max_packet_size - m_crc_bytes - DaybreakReliableHeader::size() - 1; // -1 for compress flag

75
common/net/daybreak_connection.h
View File

@ -133,6 +133,38 @@ namespace EQ
uint64_t bytes_before_encode; uint64_t bytes_before_encode;
}; };
//Refactoring this stuff
struct DaybreakSentPacket
{
DynamicPacket packet;
Timestamp last_sent;
Timestamp first_sent;
size_t times_resent;
size_t resend_delay;
};
class DaybreakReliableStream
{
public:
DaybreakReliableStream() {
sequence_in = 0;
sequence_out = 0;
fragment_current_bytes = 0;
fragment_total_bytes = 0;
}
//private:
uint16_t sequence_in;
uint16_t sequence_out;
std::map<uint16_t, Packet*> packet_queue;
DynamicPacket fragment_packet;
uint32_t fragment_current_bytes;
uint32_t fragment_total_bytes;
std::map<uint16_t, DaybreakSentPacket> sent_packets;
};
class DaybreakConnectionManager; class DaybreakConnectionManager;
class DaybreakConnection; class DaybreakConnection;
class DaybreakConnection class DaybreakConnection
@ -181,37 +213,7 @@ namespace EQ
Timestamp m_close_time; Timestamp m_close_time;
double m_outgoing_budget; double m_outgoing_budget;
struct DaybreakSentPacket std::unique_ptr<DaybreakReliableStream> m_streams[4];
{
DynamicPacket packet;
Timestamp last_sent;
Timestamp first_sent;
size_t times_resent;
size_t resend_delay;
};
class DaybreakStream
{
public:
DaybreakStream() {
sequence_in = 0;
sequence_out = 0;
fragment_current_bytes = 0;
fragment_total_bytes = 0;
}
//private:
uint16_t sequence_in;
uint16_t sequence_out;
std::map<uint16_t, Packet*> packet_queue;
DynamicPacket fragment_packet;
uint32_t fragment_current_bytes;
uint32_t fragment_total_bytes;
std::map<uint16_t, DaybreakSentPacket> sent_packets;
};
DaybreakStream** m_streams;
std::weak_ptr<DaybreakConnection> m_self; std::weak_ptr<DaybreakConnection> m_self;
void Process(); void Process();
@ -273,6 +275,14 @@ namespace EQ
resend_timeout = 30000; resend_timeout = 30000;
connection_close_time = 2000; connection_close_time = 2000;
outgoing_data_rate = 0.0; outgoing_data_rate = 0.0;
//this is emperically based on what the client seems to set for theirs
//this is the max number of packets that can be sent/recv per channel before acks come in
//we'll use this to preallocate some buffers
max_outgoing_packets_per_channel = 400;
max_incoming_packets_per_channel = 400;
//This is the max size of a packet that can be sent or received
max_total_packet_size = 1024 * 1024 * 16;
} }
size_t max_packet_size; size_t max_packet_size;
@ -296,6 +306,9 @@ namespace EQ
DaybreakEncodeType encode_passes[2]; DaybreakEncodeType encode_passes[2];
int port; int port;
double outgoing_data_rate; double outgoing_data_rate;
int max_outgoing_packets_per_channel;
int max_incoming_packets_per_channel;
int max_total_packet_size;
}; };
class DaybreakConnectionManager class DaybreakConnectionManager