Fix "disconnected: write(m_post_fd, &buffer, 1): Broken pipe" EventLoop shutdown races.

The EventLoop shutdown sequence has race conditions that could cause it to shut
down right before a removeClient write(m_post_fd, ...) call is about to happen,
if thread run in an unexpected order and cause the write to fail.

Cases where this can happen are described in
bitcoin/bitcoin#31151 (comment) and the
possible causes are that (1) m_mutex is not used to protect some EventLoop
member variables that are accessed from multiple threads, and (2) the
removeClient method can do unnecessary write(m_post_fd, ...) calls before the
loop is supposed to exit because it is not checking m_async_fns.empty(), and
these multiple write calls can make the event loop exit early and cause the
last write() call to fail. PR should fix both these issues.

Fixes bitcoin/bitcoin#31151

Author: ryanofsky

include/mp/proxy-io.h

@@ -165,6 +165,8 @@ class EventLoop
     //! Add/remove remote client reference counts.
     void addClient(std::unique_lock<std::mutex>& lock);
     void removeClient(std::unique_lock<std::mutex>& lock);
+    //! Check if loop should exit.
+    bool done(std::unique_lock<std::mutex>& lock);
 
     Logger log()
     {

src/mp/proxy.cpp

@@ -188,29 +188,27 @@ void EventLoop::loop()
 
     kj::Own<kj::AsyncIoStream> wait_stream{
         m_io_context.lowLevelProvider->wrapSocketFd(m_wait_fd, kj::LowLevelAsyncIoProvider::TAKE_OWNERSHIP)};
+    int post_fd{m_post_fd};
     char buffer = 0;
     for (;;) {
         size_t read_bytes = wait_stream->read(&buffer, 0, 1).wait(m_io_context.waitScope);
-        if (read_bytes == 1) {
-            std::unique_lock<std::mutex> lock(m_mutex);
-            if (m_post_fn) {
-                Unlock(lock, *m_post_fn);
-                m_post_fn = nullptr;
-            }
-        } else {
-            throw std::logic_error("EventLoop wait_stream closed unexpectedly");
-        }
-        m_cv.notify_all();
-        if (m_num_clients == 0 && m_async_fns.empty()) {
-            log() << "EventLoop::loop done, cancelling event listeners.";
-            m_task_set.reset();
-            log() << "EventLoop::loop bye.";
+        if (read_bytes != 1) throw std::logic_error("EventLoop wait_stream closed unexpectedly");
+        std::unique_lock<std::mutex> lock(m_mutex);
+        if (m_post_fn) {
+            Unlock(lock, *m_post_fn);
+            m_post_fn = nullptr;
+            m_cv.notify_all();
+        } else if (done(lock)) {
             break;
         }
     }
+    log() << "EventLoop::loop done, cancelling event listeners.";
+    m_task_set.reset();
+    log() << "EventLoop::loop bye.";
     wait_stream = nullptr;
+    KJ_SYSCALL(::close(post_fd));
+    std::unique_lock<std::mutex> lock(m_mutex);
     m_wait_fd = -1;
-    KJ_SYSCALL(::close(m_post_fd));
     m_post_fd = -1;
 }
 
@@ -222,9 +220,10 @@ void EventLoop::post(const std::function<void()>& fn)
     std::unique_lock<std::mutex> lock(m_mutex);
     m_cv.wait(lock, [this] { return m_post_fn == nullptr; });
     m_post_fn = &fn;
+    int post_fd{m_post_fd};
     Unlock(lock, [&] {
         char buffer = 0;
-        KJ_SYSCALL(write(m_post_fd, &buffer, 1));
+        KJ_SYSCALL(write(post_fd, &buffer, 1));
     });
     m_cv.wait(lock, [this, &fn] { return m_post_fn != &fn; });
 }
@@ -233,13 +232,13 @@ void EventLoop::addClient(std::unique_lock<std::mutex>& lock) { m_num_clients +=
 
 void EventLoop::removeClient(std::unique_lock<std::mutex>& lock)
 {
-    assert(m_num_clients > 0);
     m_num_clients -= 1;
-    if (m_num_clients == 0) {
+    if (done(lock)) {
         m_cv.notify_all();
+        int post_fd{m_post_fd};
         Unlock(lock, [&] {
             char buffer = 0;
-            KJ_SYSCALL(write(m_post_fd, &buffer, 1)); // NOLINT(bugprone-suspicious-semicolon)
+            KJ_SYSCALL(write(post_fd, &buffer, 1)); // NOLINT(bugprone-suspicious-semicolon)
         });
     }
 }
@@ -268,6 +267,12 @@ void EventLoop::startAsyncThread(std::unique_lock<std::mutex>& lock)
     }
 }
 
+bool EventLoop::done(std::unique_lock<std::mutex>& lock)
+{
+    assert(m_num_clients >= 0);
+    return m_num_clients == 0 && m_async_fns.empty();
+}
+
 std::tuple<ConnThread, bool> SetThread(ConnThreads& threads, std::mutex& mutex, Connection* connection, std::function<Thread::Client()> make_thread)
 {
     std::unique_lock<std::mutex> lock(mutex);