gitea-mirror/monero
1
0
Fork 0
mirror of https://github.com/monero-project/monero.git synced 2026-07-30 15:40:24 -07:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #10298

Browse Source 
ee9e4a4 p2p: connection patches (j-berman)
This commit is contained in:
tobtoht
2026-02-03 19:52:22 +00:00
parent 027bbf9e2b ee9e4a49ba
commit 7cc1142889
3 changed files with 173 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
contrib/epee/include/net/abstract_tcp_server2.h
+3
View File
@@ -128,6 +128,7 @@ namespace net_utils
void start_handshake();
void start_read();
void finish_read(size_t bytes_transferred);
void start_write();
void start_shutdown();
void cancel_socket();
@@ -139,6 +140,7 @@ namespace net_utils
void terminate();
void on_terminating();
void terminate_async();
bool send(epee::byte_slice message);
bool start_internal(
@@ -192,6 +194,7 @@ namespace net_utils
bool wait_read;
bool handle_read;
bool cancel_read;
bool shutdown_read;
bool wait_write;
bool handle_write;
contrib/epee/include/net/abstract_tcp_server2.inl
+102 -62
View File
@@ -171,7 +171,7 @@ namespace net_utils
return;
m_state.timers.general.wait_expire = true;
auto self = connection<T>::shared_from_this();
m_timers.general.async_wait([this, self](const ec_t & ec){
auto on_wait = [this, self] {
std::lock_guard<std::mutex> guard(m_state.lock);
m_state.timers.general.wait_expire = false;
if (m_state.timers.general.cancel_expire) {
@@ -189,6 +189,9 @@ namespace net_utils
interrupt();
else if (m_state.status == status_t::INTERRUPTED)
terminate();
};
m_timers.general.async_wait([this, self, on_wait](const ec_t & ec){
boost::asio::post(m_strand, on_wait);
});
}
@@ -242,27 +245,7 @@ namespace net_utils
)
) {
m_state.ssl.enabled = false;
m_state.socket.handle_read = true;
boost::asio::post(
connection_basic::strand_,
[this, self, bytes_transferred]{
bool success = m_handler.handle_recv(
reinterpret_cast<char *>(m_state.data.read.buffer.data()),
bytes_transferred
);
std::lock_guard<std::mutex> guard(m_state.lock);
m_state.socket.handle_read = false;
if (m_state.status == status_t::INTERRUPTED)
on_interrupted();
else if (m_state.status == status_t::TERMINATING)
on_terminating();
else if (!success)
interrupt();
else {
start_read();
}
}
);
finish_read(bytes_transferred);
}
else {
m_state.ssl.detected = true;
@@ -322,7 +305,7 @@ namespace net_utils
void connection<T>::start_read()
{
if (m_state.timers.throttle.in.wait_expire || m_state.socket.wait_read ||
m_state.socket.handle_read
m_state.socket.handle_read || m_state.socket.shutdown_read
) {
return;
}
@@ -346,7 +329,7 @@ namespace net_utils
if (duration > duration_t{}) {
m_timers.throttle.in.expires_after(duration);
m_state.timers.throttle.in.wait_expire = true;
m_timers.throttle.in.async_wait([this, self](const ec_t &ec){
auto on_wait = [this, self](const ec_t &ec){
std::lock_guard<std::mutex> guard(m_state.lock);
m_state.timers.throttle.in.wait_expire = false;
if (m_state.timers.throttle.in.cancel_expire) {
@@ -355,8 +338,16 @@ namespace net_utils
}
else if (ec.value())
interrupt();
else
};
m_timers.throttle.in.async_wait([this, self, on_wait](const ec_t &ec){
std::lock_guard<std::mutex> guard(m_state.lock);
const bool error_status = m_state.timers.throttle.in.cancel_expire || ec.value();
if (error_status)
boost::asio::post(m_strand, std::bind(on_wait, ec));
else {
m_state.timers.throttle.in.wait_expire = false;
start_read();
}
});
return;
}
@@ -392,33 +383,7 @@ namespace net_utils
m_conn_context.m_recv_cnt += bytes_transferred;
start_timer(get_timeout_from_bytes_read(bytes_transferred), true);
}
// Post handle_recv to a separate `strand_`, distinct from `m_strand`
// which is listening for reads/writes. This avoids a circular dep.
// handle_recv can queue many writes, and `m_strand` will process those
// writes until the connection terminates without deadlocking waiting
// for handle_recv.
m_state.socket.handle_read = true;
boost::asio::post(
connection_basic::strand_,
[this, self, bytes_transferred]{
bool success = m_handler.handle_recv(
reinterpret_cast<char *>(m_state.data.read.buffer.data()),
bytes_transferred
);
std::lock_guard<std::mutex> guard(m_state.lock);
m_state.socket.handle_read = false;
if (m_state.status == status_t::INTERRUPTED)
on_interrupted();
else if (m_state.status == status_t::TERMINATING)
on_terminating();
else if (!success)
interrupt();
else {
start_read();
}
}
);
finish_read(bytes_transferred);
}
};
if (!m_state.ssl.enabled)
@@ -444,6 +409,62 @@ namespace net_utils
);
}
template<typename T>
void connection<T>::finish_read(size_t bytes_transferred)
{
// Post handle_recv to a separate `strand_`, distinct from `m_strand`
// which is listening for reads/writes. This avoids a circular dep.
// handle_recv can queue many writes, and `m_strand` will process those
// writes until the connection terminates without deadlocking waiting
// for handle_recv.
m_state.socket.handle_read = true;
auto self = connection<T>::shared_from_this();
boost::asio::post(
connection_basic::strand_,
[this, self, bytes_transferred]{
bool success = m_handler.handle_recv(
reinterpret_cast<char *>(m_state.data.read.buffer.data()),
bytes_transferred
);
std::lock_guard<std::mutex> guard(m_state.lock);
const bool error_status = m_state.status == status_t::INTERRUPTED
|| m_state.status == status_t::TERMINATING
|| !success;
if (!error_status) {
m_state.socket.handle_read = false;
start_read();
return;
}
boost::asio::post(
m_strand,
[this, self, success]{
// expect error_status == true
std::lock_guard<std::mutex> guard(m_state.lock);
m_state.socket.handle_read = false;
if (m_state.status == status_t::INTERRUPTED)
on_interrupted();
else if (m_state.status == status_t::TERMINATING)
on_terminating();
else if (!success) {
ec_t ec;
if (m_state.socket.wait_write) {
// Allow the already queued writes time to finish, but no more new reads
connection_basic::socket_.next_layer().shutdown(
socket_t::shutdown_receive,
ec
);
m_state.socket.shutdown_read = true;
}
if (!m_state.socket.wait_write || ec.value()) {
interrupt();
}
}
}
);
}
);
}
template<typename T>
void connection<T>::start_write()
{
@@ -475,7 +496,7 @@ namespace net_utils
if (duration > duration_t{}) {
m_timers.throttle.out.expires_after(duration);
m_state.timers.throttle.out.wait_expire = true;
m_timers.throttle.out.async_wait([this, self](const ec_t &ec){
auto on_wait = [this, self](const ec_t &ec){
std::lock_guard<std::mutex> guard(m_state.lock);
m_state.timers.throttle.out.wait_expire = false;
if (m_state.timers.throttle.out.cancel_expire) {
@@ -484,8 +505,16 @@ namespace net_utils
}
else if (ec.value())
interrupt();
else
};
m_timers.throttle.out.async_wait([this, self, on_wait](const ec_t &ec){
std::lock_guard<std::mutex> guard(m_state.lock);
const bool error_status = m_state.timers.throttle.out.cancel_expire || ec.value();
if (error_status)
boost::asio::post(m_strand, std::bind(on_wait, ec));
else {
m_state.timers.throttle.out.wait_expire = false;
start_write();
}
});
}
}
@@ -533,7 +562,12 @@ namespace net_utils
m_state.data.write.total_bytes -=
std::min(m_state.data.write.total_bytes, byte_count);
m_state.condition.notify_all();
start_write();
if (m_state.data.write.queue.empty() && m_state.socket.shutdown_read) {
// All writes have been sent and reads shutdown already, connection can be closed
interrupt();
} else {
start_write();
}
}
};
if (!m_state.ssl.enabled)
@@ -762,6 +796,17 @@ namespace net_utils
m_state.status = status_t::WASTED;
}
template<typename T>
void connection<T>::terminate_async()
{
// synchronize with intermediate writes on `m_strand`
auto self = connection<T>::shared_from_this();
boost::asio::post(m_strand, [this, self] {
std::lock_guard<std::mutex> guard(m_state.lock);
terminate();
});
}
template<typename T>
bool connection<T>::send(epee::byte_slice message)
{
@@ -814,12 +859,7 @@ namespace net_utils
);
m_state.data.write.wait_consume = false;
if (!success) {
// synchronize with intermediate writes on `m_strand`
auto self = connection<T>::shared_from_this();
boost::asio::post(m_strand, [this, self] {
std::lock_guard<std::mutex> guard(m_state.lock);
terminate();
});
terminate_async();
return false;
}
else
@@ -1093,7 +1133,7 @@ namespace net_utils
std::lock_guard<std::mutex> guard(m_state.lock);
if (m_state.status != status_t::RUNNING)
return false;
terminate();
terminate_async();
return true;
}