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

Resend changes

Browse Source
This commit is contained in:
Akkadius 2025-01-28 00:13:43 -06:00
parent 6b548576da
commit d69225d1eb
2 changed files with 110 additions and 99 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

206
common/net/daybreak_connection.cpp
View File

@ -42,11 +42,19 @@ void EQ::Net::DaybreakConnectionManager::Attach(uv_loop_t *loop)
auto update_rate = (uint64_t)(1000.0 / m_options.tic_rate_hertz); auto update_rate = (uint64_t)(1000.0 / m_options.tic_rate_hertz);
uv_timer_start(&m_timer, [](uv_timer_t *handle) { uv_timer_start(&m_timer, [](uv_timer_t *handle) {
DaybreakConnectionManager *c = (DaybreakConnectionManager*)handle->data; DaybreakConnectionManager *c = (DaybreakConnectionManager*)handle->data;
c->UpdateDataBudget(); c->UpdateDataBudget();
c->Process(); c->Process();
c->ProcessResend();
auto now = Clock::now();
if (std::chrono::duration_cast<std::chrono::milliseconds>(now - c->m_last_resend_check).count() > 500) {
c->ProcessResend();
c->m_last_resend_check = Clock::now();
return;
}
}, update_rate, update_rate); }, update_rate, update_rate);
uv_udp_init(loop, &m_socket); uv_udp_init(loop, &m_socket);
@ -60,31 +68,35 @@ void EQ::Net::DaybreakConnectionManager::Attach(uv_loop_t *loop)
static std::atomic<int> buffer_index(0); // Atomic to track the current buffer index static std::atomic<int> buffer_index(0); // Atomic to track the current buffer index
// Select the next static buffer in a round-robin manner // Select the next static buffer in a round-robin manner
rc = uv_udp_recv_start(
rc = uv_udp_recv_start(&m_socket, &m_socket,
[](uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) { [](uv_handle_t *handle, size_t suggested_size, uv_buf_t *buf) {
int current_buffer_index = buffer_index.fetch_add(1) % NUM_STATIC_BUFFERS; int current_buffer_index = buffer_index.fetch_add(1) % NUM_STATIC_BUFFERS;
// std::cout << "recv current_buffer_index: " << current_buffer_index << " current_thread_id: " << std::this_thread::get_id() << std::endl; char *buffer = static_buffers[current_buffer_index];
char* buffer = static_buffers[current_buffer_index];
// clear the buffer // clear the buffer
memset(buffer, 0, 512); memset(buffer, 0, 512);
buf->base = buffer; buf->base = buffer;
buf->len = suggested_size; buf->len = suggested_size;
}, },
[](uv_udp_t* handle, ssize_t nread, const uv_buf_t* buf, const struct sockaddr* addr, unsigned flags) { [](
DaybreakConnectionManager *c = (DaybreakConnectionManager*)handle->data; uv_udp_t *handle,
if (nread < 0 || addr == nullptr) { ssize_t nread,
return; const uv_buf_t *buf,
const struct sockaddr *addr,
unsigned flags
) {
DaybreakConnectionManager *c = (DaybreakConnectionManager *) handle->data;
if (nread < 0 || addr == nullptr) {
return;
}
char endpoint[16];
uv_ip4_name((const sockaddr_in *) addr, endpoint, 16);
auto port = ntohs(((const sockaddr_in *) addr)->sin_port);
c->ProcessPacket(endpoint, port, buf->base, nread);
} }
);
// std::cout << "recv nread: " << nread << " current_thread_id: " << std::this_thread::get_id() << std::endl;
char endpoint[16];
uv_ip4_name((const sockaddr_in*)addr, endpoint, 16);
auto port = ntohs(((const sockaddr_in*)addr)->sin_port);
c->ProcessPacket(endpoint, port, buf->base, nread);
});
m_attached = loop; m_attached = loop;
} }
@ -201,9 +213,7 @@ void EQ::Net::DaybreakConnectionManager::ProcessResend()
auto iter = m_connections.begin(); auto iter = m_connections.begin();
while (iter != m_connections.end()) { while (iter != m_connections.end()) {
auto &connection = iter->second; auto &connection = iter->second;
auto status = connection->m_status; switch (connection->m_status)
switch (status)
{ {
case StatusConnected: case StatusConnected:
case StatusDisconnecting: case StatusDisconnecting:
@ -292,11 +302,13 @@ void EQ::Net::DaybreakConnectionManager::SendDisconnect(const std::string &addr,
memcpy(data, out.Data(), out.Length()); memcpy(data, out.Data(), out.Length());
send_buffers[0] = uv_buf_init(data, out.Length()); send_buffers[0] = uv_buf_init(data, out.Length());
send_req->data = send_buffers[0].base; send_req->data = send_buffers[0].base;
int ret = uv_udp_send(send_req, &m_socket, send_buffers, 1, (sockaddr*)&send_addr, int ret = uv_udp_send(
[](uv_udp_send_t* req, int status) { send_req, &m_socket, send_buffers, 1, (sockaddr *) &send_addr,
delete[](char*)req->data; [](uv_udp_send_t *req, int status) {
delete req; delete[](char *) req->data;
}); delete req;
}
);
} }
//new connection made as server //new connection made as server
@ -538,7 +550,6 @@ void EQ::Net::DaybreakConnection::AddToQueue(int stream, uint16_t seq, const Pac
auto iter = s->packet_queue.find(seq); auto iter = s->packet_queue.find(seq);
if (s->packet_queue.bucket_count() < seq) { if (s->packet_queue.bucket_count() < seq) {
// std::cout << "Resizing packet queue to " << seq << std::endl;
s->packet_queue.reserve(seq); s->packet_queue.reserve(seq);
} }
@ -608,14 +619,14 @@ void EQ::Net::DaybreakConnection::ProcessDecodedPacket(const Packet &p)
(*(current + 4) << 16) | (*(current + 4) << 16) |
(*(current + 5) << 8) | (*(current + 5) << 8) |
(*(current + 6)) (*(current + 6))
); );
current += 7; current += 7;
} }
else { else {
subpacket_length = (uint32_t)( subpacket_length = (uint32_t)(
(*(current + 1) << 8) | (*(current + 1) << 8) |
(*(current + 2)) (*(current + 2))
); );
current += 3; current += 3;
} }
} }
@ -1116,63 +1127,56 @@ void EQ::Net::DaybreakConnection::ProcessResend(int stream)
return; return;
} }
auto resends = 0; if (m_streams[stream].sent_packets.empty()) {
auto now = Clock::now(); return;
auto s = &m_streams[stream]; }
for (auto &entry : s->sent_packets) {
auto time_since_last_send = std::chrono::duration_cast<std::chrono::milliseconds>(now - entry.second.last_sent);
if (entry.second.times_resent == 0) {
if ((size_t)time_since_last_send.count() > entry.second.resend_delay) {
auto &p = entry.second.packet;
if (p.Length() >= DaybreakHeader::size()) {
if (p.GetInt8(0) == 0 && p.GetInt8(1) >= OP_Fragment && p.GetInt8(1) <= OP_Fragment4) {
m_stats.resent_fragments++;
}
else {
m_stats.resent_full++;
}
}
else {
m_stats.resent_full++;
}
m_stats.resent_packets++;
InternalBufferedSend(p); std::cout << "Processing resend for stream " << stream << std::endl;
entry.second.last_sent = now;
entry.second.times_resent++; auto now = Clock::now(); // Current time
entry.second.resend_delay = EQ::Clamp(entry.second.resend_delay * 2, m_owner->m_options.resend_delay_min, m_owner->m_options.resend_delay_max); auto s = &m_streams[stream];
resends++;
} // Get a reference resend delay (assume first packet represents the typical case)
} size_t reference_resend_delay = 0;
else { if (!s->sent_packets.empty()) {
auto time_since_first_sent = std::chrono::duration_cast<std::chrono::milliseconds>(now - entry.second.first_sent); reference_resend_delay = s->sent_packets.begin()->second.resend_delay;
if (time_since_first_sent.count() >= m_owner->m_options.resend_timeout) { }
std::cout << "Resending packet count " << s->sent_packets.size() << " reference_resend_delay " << reference_resend_delay << std::endl;
for (auto& entry : s->sent_packets) {
auto& packet_info = entry.second;
// Handle timeout only if the packet has been resent
if (packet_info.times_resent > 0) {
auto time_since_first_sent = std::chrono::duration_cast<std::chrono::milliseconds>(now - packet_info.first_sent).count();
if (time_since_first_sent >= m_owner->m_options.resend_timeout) {
Close(); Close();
return; return;
} }
if ((size_t)time_since_last_send.count() > entry.second.resend_delay) {
auto &p = entry.second.packet;
if (p.Length() >= DaybreakHeader::size()) {
if (p.GetInt8(0) == 0 && p.GetInt8(1) >= OP_Fragment && p.GetInt8(1) <= OP_Fragment4) {
m_stats.resent_fragments++;
}
else {
m_stats.resent_full++;
}
}
else {
m_stats.resent_full++;
}
m_stats.resent_packets++;
InternalBufferedSend(p);
entry.second.last_sent = now;
entry.second.times_resent++;
entry.second.resend_delay = EQ::Clamp(entry.second.resend_delay * 2, m_owner->m_options.resend_delay_min, m_owner->m_options.resend_delay_max);
resends++;
}
} }
// Increment resend statistics
auto& p = packet_info.packet;
if (p.Length() >= DaybreakHeader::size()) {
if (p.GetInt8(0) == 0 && p.GetInt8(1) >= OP_Fragment && p.GetInt8(1) <= OP_Fragment4) {
m_stats.resent_fragments++;
} else {
m_stats.resent_full++;
}
} else {
m_stats.resent_full++;
}
m_stats.resent_packets++;
// std::cout << "Resending packet " << p.Length() << " on stream " << stream << " times_resent "
// << packet_info.times_resent << " resend_delay " << packet_info.resend_delay << std::endl;
// Resend the packet
InternalBufferedSend(p);
packet_info.last_sent = now;
packet_info.times_resent++;
packet_info.resend_delay = EQ::Clamp(packet_info.resend_delay * 2, m_owner->m_options.resend_delay_min, m_owner->m_options.resend_delay_max);
} }
} }
@ -1350,22 +1354,26 @@ void EQ::Net::DaybreakConnection::InternalSend(Packet &p)
DynamicPacket out; DynamicPacket out;
out.PutPacket(0, p); out.PutPacket(0, p);
for (int i = 0; i < 2; ++i) { for (auto &m_encode_passe: m_encode_passes) {
switch (m_encode_passes[i]) { switch (m_encode_passe) {
case EncodeCompression: case EncodeCompression:
if (out.GetInt8(0) == 0) if (out.GetInt8(0) == 0) {
Compress(out, DaybreakHeader::size(), out.Length() - DaybreakHeader::size()); Compress(out, DaybreakHeader::size(), out.Length() - DaybreakHeader::size());
else }
Compress(out, 1, out.Length() - 1); else {
break; Compress(out, 1, out.Length() - 1);
case EncodeXOR: }
if (out.GetInt8(0) == 0) break;
Encode(out, DaybreakHeader::size(), out.Length() - DaybreakHeader::size()); case EncodeXOR:
else if (out.GetInt8(0) == 0) {
Encode(out, 1, out.Length() - 1); Encode(out, DaybreakHeader::size(), out.Length() - DaybreakHeader::size());
break; }
default: else {
break; Encode(out, 1, out.Length() - 1);
}
break;
default:
break;
} }
} }

3
common/net/daybreak_connection.h
View File

@ -10,6 +10,7 @@
#include <map> #include <map>
#include <queue> #include <queue>
#include <list> #include <list>
#include <set>
namespace EQ namespace EQ
{ {
@ -318,6 +319,8 @@ namespace EQ
void Attach(uv_loop_t *loop); void Attach(uv_loop_t *loop);
void Detach(); void Detach();
Timestamp m_last_resend_check;
EQ::Random m_rand; EQ::Random m_rand;
uv_timer_t m_timer; uv_timer_t m_timer;
uv_udp_t m_socket; uv_udp_t m_socket;