Prevent EventLoop async cleanup thread early exit during shutdown

Antoine Poinsot <[email protected]> reported a bug with details in
#182 (comment)
where an IPC client rapidly connecting and disconnecting to the server in a
loop could cause problems.

One problem, fixed by this commit, was that if a server process is shutting
down, the async cleanup thread in `EventLoop::startAsyncThread` responsible for
destroying unused server object on unclean disconnects could detect the done()
condition and decide to exit right before a new incoming connection is
processed, and exit prematurely when there might be more cleanup work for it
do. If this happens, process shutdown could hang waiting for cleanup work that
will never be completed.

This commit fixes that problem by changing the `EventLoop::startAsyncThread()`
while condition to check whether the `EventLoop::loop() method has _exited_,
instead of checking whether it is about to exit. Specifically the change makes
the `m_async_fns` list into a `std::optional` value and sets it to `nullopt`
when the `loop()` method exits.

Author: ryanofsky

include/mp/proxy-io.h

@@ -218,7 +218,7 @@ class EventLoop
     kj::Function<void()>* m_post_fn MP_GUARDED_BY(m_mutex) = nullptr;
 
     //! Callback functions to run on async thread.
-    CleanupList m_async_fns MP_GUARDED_BY(m_mutex);
+    std::optional<CleanupList> m_async_fns MP_GUARDED_BY(m_mutex);
 
     //! Pipe read handle used to wake up the event loop thread.
     int m_wait_fd = -1;

include/mp/proxy.h

@@ -62,7 +62,7 @@ class EventLoopRef
     ~EventLoopRef() { reset(); }
     EventLoop& operator*() const { assert(m_loop); return *m_loop; }
     EventLoop* operator->() const { assert(m_loop); return m_loop; }
-    bool reset();
+    void reset(bool relock=false);
 
     EventLoop* m_loop{nullptr};
     Lock* m_lock{nullptr};

include/mp/util.h

@@ -161,6 +161,7 @@ struct PtrOrValue {
 #define MP_RELEASE(...)         MP_TSA(release_capability(__VA_ARGS__))
 #define MP_ASSERT_CAPABILITY(x) MP_TSA(assert_capability(x))
 #define MP_GUARDED_BY(x)        MP_TSA(guarded_by(x))
+#define MP_NO_TSA               MP_TSA(no_thread_safety_analysis)
 
 class MP_CAPABILITY("mutex") Mutex {
 public:

src/mp/proxy.cpp

@@ -56,27 +56,29 @@ EventLoopRef::EventLoopRef(EventLoop& loop, Lock* lock) : m_loop(&loop), m_lock(
     m_loop->m_num_clients += 1;
 }
 
-bool EventLoopRef::reset()
+// Due to the conditionals in this function, MP_NO_TSA is required to avoid
+// error "error: mutex 'loop_lock' is not held on every path through here
+// [-Wthread-safety-analysis]"
+void EventLoopRef::reset(bool relock) MP_NO_TSA
 {
-    bool done = false;
     if (auto* loop{m_loop}) {
         m_loop = nullptr;
         auto loop_lock{PtrOrValue{m_lock, loop->m_mutex}};
         loop_lock->assert_locked(loop->m_mutex);
         assert(loop->m_num_clients > 0);
         loop->m_num_clients -= 1;
         if (loop->done()) {
-            done = true;
             loop->m_cv.notify_all();
             int post_fd{loop->m_post_fd};
             loop_lock->unlock();
             char buffer = 0;
             KJ_SYSCALL(write(post_fd, &buffer, 1)); // NOLINT(bugprone-suspicious-semicolon)
-            // Do not try to relock `loop_lock` after writing, because the event loop
-            // could wake up and destroy itself and the mutex might no longer exist.
+            // By default, do not try to relock `loop_lock` after writing,
+            // because the event loop could wake up and destroy itself and the
+            // mutex might no longer exist.
+            if (relock) loop_lock->lock();
         }
     }
-    return done;
 }
 
 ProxyContext::ProxyContext(Connection* connection) : connection(connection), loop{*connection->m_loop} {}
@@ -137,7 +139,7 @@ Connection::~Connection()
     }
     while (!m_async_cleanup_fns.empty()) {
         const Lock lock(m_loop->m_mutex);
-        m_loop->m_async_fns.emplace_back(std::move(m_async_cleanup_fns.front()));
+        m_loop->m_async_fns->emplace_back(std::move(m_async_cleanup_fns.front()));
         m_async_cleanup_fns.pop_front();
     }
     Lock lock(m_loop->m_mutex);
@@ -204,7 +206,7 @@ EventLoop::~EventLoop()
     if (m_async_thread.joinable()) m_async_thread.join();
     const Lock lock(m_mutex);
     KJ_ASSERT(m_post_fn == nullptr);
-    KJ_ASSERT(m_async_fns.empty());
+    KJ_ASSERT(!m_async_fns);
     KJ_ASSERT(m_wait_fd == -1);
     KJ_ASSERT(m_post_fd == -1);
     KJ_ASSERT(m_num_clients == 0);
@@ -220,6 +222,12 @@ void EventLoop::loop()
     g_thread_context.loop_thread = true;
     KJ_DEFER(g_thread_context.loop_thread = false);
 
+    {
+        const Lock lock(m_mutex);
+        assert(!m_async_fns);
+        m_async_fns.emplace();
+    }
+
     kj::Own<kj::AsyncIoStream> wait_stream{
         m_io_context.lowLevelProvider->wrapSocketFd(m_wait_fd, kj::LowLevelAsyncIoProvider::TAKE_OWNERSHIP)};
     int post_fd{m_post_fd};
@@ -248,6 +256,8 @@ void EventLoop::loop()
     const Lock lock(m_mutex);
     m_wait_fd = -1;
     m_post_fd = -1;
+    m_async_fns.reset();
+    m_cv.notify_all();
 }
 
 void EventLoop::post(kj::Function<void()> fn)
@@ -270,24 +280,21 @@ void EventLoop::post(kj::Function<void()> fn)
 
 void EventLoop::startAsyncThread()
 {
+    assert (std::this_thread::get_id() == m_thread_id);
     if (m_async_thread.joinable()) {
         // Notify to wake up the async thread if it is already running.
         m_cv.notify_all();
-    } else if (!m_async_fns.empty()) {
+    } else if (!m_async_fns->empty()) {
         m_async_thread = std::thread([this] {
             Lock lock(m_mutex);
-            while (!done()) {
-                if (!m_async_fns.empty()) {
+            while (m_async_fns) {
+                if (!m_async_fns->empty()) {
                     EventLoopRef ref{*this, &lock};
-                    const std::function<void()> fn = std::move(m_async_fns.front());
-                    m_async_fns.pop_front();
+                    const std::function<void()> fn = std::move(m_async_fns->front());
+                    m_async_fns->pop_front();
                     Unlock(lock, fn);
-                    // Reset ref and break if that returns true instead of
-                    // passively letting ref go out of scope. This is important
-                    // because the ref destructor would leave m_mutex unlocked
-                    // when done() returns true, causing undefined behavior if
-                    // the loop continued to execute.
-                    if (ref.reset()) break;
+                    // Important to relock because of the wait() call below.
+                    ref.reset(/*relock=*/true);
                     // Continue without waiting in case there are more async_fns
                     continue;
                 }
@@ -300,7 +307,7 @@ void EventLoop::startAsyncThread()
 bool EventLoop::done() const
 {
     assert(m_num_clients >= 0);
-    return m_num_clients == 0 && m_async_fns.empty();
+    return m_num_clients == 0 && m_async_fns->empty();
 }
 
 std::tuple<ConnThread, bool> SetThread(ConnThreads& threads, std::mutex& mutex, Connection* connection, const std::function<Thread::Client()>& make_thread)