In commit "proxy: add local connection limit to ListenConnections" (8d09ad37dbf4fdcbc00fb6b6357fc968825db9f4)
I don't think this extra tracking of listeners should be necessary and looks like this implementation could be simplified significantly. Would suggest:
<details><summary>diff</summary>
<p>
--- a/include/mp/proxy-io.h
+++ b/include/mp/proxy-io.h
@@ -294,9 +294,6 @@ public:
//! Check if loop should exit.
bool done() const MP_REQUIRES(m_mutex);
- //! Stop accepting new incoming connections.
- void closeListeners();
-
//! Process name included in thread names so combined debug output from
//! multiple processes is easier to understand.
const char* m_exe_name;
@@ -341,9 +338,6 @@ public:
//! List of connections.
std::list<Connection> m_incoming_connections;
- //! List of socket listeners.
- std::list<std::shared_ptr<Listener>> m_listeners;
-
//! Logging options
LogOptions m_log_opts;
@@ -872,17 +866,6 @@ struct Listener
return m_max_connections && m_active_connections >= *m_max_connections;
}
- void holdAcceptRef(EventLoop& loop)
- {
- if (m_max_connections && *m_max_connections > 0) m_accept_ref.emplace(loop);
- }
-
- void close()
- {
- m_closed = true;
- m_accept_ref.reset();
- }
-
//! Handle incoming connections by calling _Serve, to create ProxyServer
//! objects and forward requests to the init object.
template <typename InitInterface, typename InitImpl>
@@ -890,31 +873,23 @@ struct Listener
kj::Own<kj::ConnectionReceiver> m_receiver;
std::optional<size_t> m_max_connections;
- std::optional<EventLoopRef> m_accept_ref;
size_t m_active_connections{0};
- bool m_closed{false};
};
template <typename InitInterface, typename InitImpl>
void Listener::listen(EventLoop& loop, InitImpl& init, const std::shared_ptr<Listener>& self)
{
- if (m_closed || atCapacity()) return;
+ if (atCapacity()) return;
auto* receiver = m_receiver.get();
- // Capped listeners need to keep the event loop alive while below capacity
- // and waiting for another connection. Store the ref on the Listener so
- // closeListeners() can release it during process shutdown.
- holdAcceptRef(loop);
loop.m_task_set->add(receiver->accept().then(
[&loop, &init, self](kj::Own<kj::AsyncIoStream>&& stream) {
- self->m_accept_ref.reset();
- if (self->m_closed) return;
++self->m_active_connections;
_Serve<InitInterface>(loop, kj::mv(stream), init, [&loop, &init, self] {
const bool resume_accept{self->atCapacity()};
assert(self->m_active_connections > 0);
--self->m_active_connections;
- if (resume_accept && !self->m_closed) self->listen<InitInterface>(loop, init, self);
+ if (resume_accept) self->listen<InitInterface>(loop, init, self);
});
self->listen<InitInterface>(loop, init, self);
}));
@@ -941,7 +916,6 @@ void ListenConnections(EventLoop& loop, int fd, InitImpl& init, std::optional<si
auto listener{std::make_shared<Listener>(
loop.m_io_context.lowLevelProvider->wrapListenSocketFd(fd, kj::LowLevelAsyncIoProvider::TAKE_OWNERSHIP),
max_connections)};
- loop.m_listeners.push_back(listener);
listener->listen<InitInterface>(loop, init, listener);
});
}
--- a/src/mp/proxy.cpp
+++ b/src/mp/proxy.cpp
@@ -324,14 +324,6 @@ bool EventLoop::done() const
return m_num_clients == 0 && m_async_fns->empty();
}
-void EventLoop::closeListeners()
-{
- assert(std::this_thread::get_id() == m_thread_id);
- for (auto& listener : m_listeners) {
- listener->close();
- }
-}
-
std::tuple<ConnThread, bool> SetThread(GuardedRef<ConnThreads> threads, Connection* connection, const std::function<Thread::Client()>& make_thread)
{
assert(std::this_thread::get_id() == connection->m_loop->m_thread_id);
--- a/test/mp/test/listen_tests.cpp
+++ b/test/mp/test/listen_tests.cpp
@@ -121,7 +121,7 @@ class ListenSetup
{
public:
explicit ListenSetup(std::optional<size_t> max_connections = std::nullopt)
- : capped_listener(max_connections.has_value()), thread([this, max_connections] {
+ : thread([this, max_connections] {
EventLoop loop("mptest-server", [this](mp::LogMessage log) {
KJ_LOG(INFO, log.level, log.message);
if (log.level == mp::Log::Raise) throw std::runtime_error(log.message);
@@ -136,10 +136,7 @@ public:
++connected_count;
counter_cv.notify_all();
};
- {
- std::lock_guard<std::mutex> lock(counter_mutex);
- event_loop = &loop;
- }
+ m_loop_ref.emplace(loop);
FooImplementation foo;
ListenConnections<messages::FooInterface>(loop, listener.release(), foo, max_connections);
ready_promise.set_value();
@@ -151,14 +148,7 @@ public:
~ListenSetup()
{
- if (capped_listener) {
- EventLoop* loop;
- {
- std::lock_guard<std::mutex> lock(counter_mutex);
- loop = event_loop;
- }
- if (loop) loop->sync([&] { loop->closeListeners(); });
- }
+ m_loop_ref.reset();
thread.join();
}
@@ -190,10 +180,9 @@ public:
UnixListener listener;
std::promise<void> ready_promise;
- bool capped_listener{false};
+ std::optional<EventLoopRef> m_loop_ref;
std::mutex counter_mutex;
std::condition_variable counter_cv;
- EventLoop* event_loop{nullptr};
size_t connected_count{0};
size_t disconnected_count{0};
//! Thread variable should be after other struct members so the thread does
</p>
</details>
However, one thing I am just realizing about the ListenConnections API is there is no way to stop listening after you have started listening without shutting down the entire event loop. This is ok for bitcoin core right now, since it only needs to start listening on startup and stop on shutdown. But for other applications it could make sense to be able to stop listening for new connections without shutting down existing ones. This could be implemented by making ListenConnections return shared_ptr<Listener> instead of void and giving Listener a close method. But that feature wouldn't be worth extra complexity in this commit, and also might make more sense as a separate PR.
