refactor: Add clang thread safety annotations to EventLoop

Add basic thread safety annotations to EventLoop. Use could be expanded to
other functions. Use can be expanded and deepened but this puts basic
functionality in place.

Use of annotations was discussed in
#129 (comment)

Author: ryanofsky

include/mp/proxy-io.h

@@ -186,10 +186,10 @@ class EventLoop
     //! is important that ProxyServer::m_impl destructors do not run on the
     //! eventloop thread because they may need it to do I/O if they perform
     //! other IPC calls.
-    void startAsyncThread(std::unique_lock<std::mutex>& lock);
+    void startAsyncThread() MP_REQUIRES(m_mutex);
 
     //! Check if loop should exit.
-    bool done(std::unique_lock<std::mutex>& lock);
+    bool done() MP_REQUIRES(m_mutex);
 
     Logger log()
     {
@@ -212,10 +212,10 @@ class EventLoop
     std::thread m_async_thread;
 
     //! Callback function to run on event loop thread during post() or sync() call.
-    kj::Function<void()>* m_post_fn = nullptr;
+    kj::Function<void()>* m_post_fn MP_GUARDED_BY(m_mutex) = nullptr;
 
     //! Callback functions to run on async thread.
-    CleanupList m_async_fns;
+    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;
@@ -225,11 +225,11 @@ class EventLoop
 
     //! Number of clients holding references to ProxyServerBase objects that
     //! reference this event loop.
-    int m_num_clients = 0;
+    int m_num_clients MP_GUARDED_BY(m_mutex) = 0;
 
     //! Mutex and condition variable used to post tasks to event loop and async
     //! thread.
-    std::mutex m_mutex;
+    Mutex m_mutex;
     std::condition_variable m_cv;
 
     //! Capnp IO context.

include/mp/proxy.h

@@ -54,18 +54,18 @@ inline void CleanupRun(CleanupList& fns) {
 class EventLoopRef
 {
 public:
-    explicit EventLoopRef(EventLoop& loop, std::unique_lock<std::mutex>* lock = nullptr);
+    explicit EventLoopRef(EventLoop& loop, Lock* lock = nullptr);
     EventLoopRef(EventLoopRef&& other) noexcept : m_loop(other.m_loop) { other.m_loop = nullptr; }
     EventLoopRef(const EventLoopRef&) = delete;
     EventLoopRef& operator=(const EventLoopRef&) = delete;
     EventLoopRef& operator=(EventLoopRef&&) = delete;
     ~EventLoopRef() { reset(); }
     EventLoop& operator*() const { assert(m_loop); return *m_loop; }
     EventLoop* operator->() const { assert(m_loop); return m_loop; }
-    bool reset(std::unique_lock<std::mutex>* lock = nullptr);
+    bool reset(Lock* lock = nullptr);
 
     EventLoop* m_loop{nullptr};
-    std::unique_lock<std::mutex>* m_lock{nullptr};
+    Lock* m_lock{nullptr};
 };
 
 //! Context data associated with proxy client and server classes.

include/mp/util.h

@@ -6,13 +6,15 @@
 #define MP_UTIL_H
 
 #include <capnp/schema.h>
+#include <cassert>
 #include <cstddef>
 #include <functional>
 #include <future>
 #include <kj/common.h>
 #include <kj/exception.h>
 #include <kj/string-tree.h>
 #include <memory>
+#include <mutex>
 #include <string.h>
 #include <string>
 #include <tuple>
@@ -145,6 +147,44 @@ struct PtrOrValue {
     T* operator->() const { return &**this; }
 };
 
+// Annotated mutex and lock class (https://clang.llvm.org/docs/ThreadSafetyAnalysis.html)
+#if defined(__clang__) && (!defined(SWIG))
+#define MP_TSA(x)   __attribute__((x))
+#else
+#define MP_TSA(x)   // no-op
+#endif
+
+#define MP_CAPABILITY(x)        MP_TSA(capability(x))
+#define MP_SCOPED_CAPABILITY    MP_TSA(scoped_lockable)
+#define MP_REQUIRES(x)          MP_TSA(requires_capability(x))
+#define MP_ACQUIRE(...)         MP_TSA(acquire_capability(__VA_ARGS__))
+#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))
+
+class MP_CAPABILITY("mutex") Mutex {
+public:
+    void lock() MP_ACQUIRE() { m_mutex.lock(); }
+    void unlock() MP_RELEASE() { m_mutex.unlock(); }
+
+    std::mutex m_mutex;
+};
+
+class MP_SCOPED_CAPABILITY Lock {
+public:
+    explicit Lock(Mutex& m) MP_ACQUIRE(m) : m_lock(m.m_mutex) {}
+    ~Lock() MP_RELEASE() {}
+    void unlock() MP_RELEASE() { m_lock.unlock(); }
+    void lock() MP_ACQUIRE() { m_lock.lock(); }
+    void assert_locked(Mutex& mutex) MP_ASSERT_CAPABILITY() MP_ASSERT_CAPABILITY(mutex)
+    {
+        assert(m_lock.mutex() == &mutex.m_mutex);
+        assert(m_lock);
+    }
+
+    std::unique_lock<std::mutex> m_lock;
+};
+
 //! Analog to std::lock_guard that unlocks instead of locks.
 template <typename Lock>
 struct UnlockGuard

src/mp/proxy.cpp

@@ -49,20 +49,21 @@ void LoggingErrorHandler::taskFailed(kj::Exception&& exception)
     m_loop.log() << "Uncaught exception in daemonized task.";
 }
 
-EventLoopRef::EventLoopRef(EventLoop& loop, std::unique_lock<std::mutex>* lock) : m_loop(&loop), m_lock(lock)
+EventLoopRef::EventLoopRef(EventLoop& loop, Lock* lock) : m_loop(&loop), m_lock(lock)
 {
     auto loop_lock{PtrOrValue{m_lock, m_loop->m_mutex}};
     m_loop->m_num_clients += 1;
 }
 
-bool EventLoopRef::reset(std::unique_lock<std::mutex>* lock)
+bool EventLoopRef::reset(Lock* lock)
 {
     bool done = false;
     if (m_loop) {
         auto loop_lock{PtrOrValue{lock ? lock : m_lock, m_loop->m_mutex}};
+        loop_lock->assert_locked(m_loop->m_mutex);
         assert(m_loop->m_num_clients > 0);
         m_loop->m_num_clients -= 1;
-        if (m_loop->done(*loop_lock)) {
+        if (m_loop->done()) {
             done = true;
             m_loop->m_cv.notify_all();
             int post_fd{m_loop->m_post_fd};
@@ -133,18 +134,18 @@ Connection::~Connection()
         m_sync_cleanup_fns.pop_front();
     }
     while (!m_async_cleanup_fns.empty()) {
-        const std::unique_lock<std::mutex> lock(m_loop->m_mutex);
+        const Lock lock(m_loop->m_mutex);
         m_loop->m_async_fns.emplace_back(std::move(m_async_cleanup_fns.front()));
         m_async_cleanup_fns.pop_front();
     }
-    std::unique_lock<std::mutex> lock(m_loop->m_mutex);
-    m_loop->startAsyncThread(lock);
+    Lock lock(m_loop->m_mutex);
+    m_loop->startAsyncThread();
     m_loop.reset(&lock);
 }
 
 CleanupIt Connection::addSyncCleanup(std::function<void()> fn)
 {
-    const std::unique_lock<std::mutex> lock(m_loop->m_mutex);
+    const Lock lock(m_loop->m_mutex);
     // Add cleanup callbacks to the front of list, so sync cleanup functions run
     // in LIFO order. This is a good approach because sync cleanup functions are
     // added as client objects are created, and it is natural to clean up
@@ -158,13 +159,13 @@ CleanupIt Connection::addSyncCleanup(std::function<void()> fn)
 
 void Connection::removeSyncCleanup(CleanupIt it)
 {
-    const std::unique_lock<std::mutex> lock(m_loop->m_mutex);
+    const Lock lock(m_loop->m_mutex);
     m_sync_cleanup_fns.erase(it);
 }
 
 void Connection::addAsyncCleanup(std::function<void()> fn)
 {
-    const std::unique_lock<std::mutex> lock(m_loop->m_mutex);
+    const Lock lock(m_loop->m_mutex);
     // Add async cleanup callbacks to the back of the list. Unlike the sync
     // cleanup list, this list order is more significant because it determines
     // the order server objects are destroyed when there is a sudden disconnect,
@@ -200,7 +201,7 @@ EventLoop::EventLoop(const char* exe_name, LogFn log_fn, void* context)
 EventLoop::~EventLoop()
 {
     if (m_async_thread.joinable()) m_async_thread.join();
-    const std::lock_guard<std::mutex> lock(m_mutex);
+    const Lock lock(m_mutex);
     KJ_ASSERT(m_post_fn == nullptr);
     KJ_ASSERT(m_async_fns.empty());
     KJ_ASSERT(m_wait_fd == -1);
@@ -225,12 +226,12 @@ void EventLoop::loop()
     for (;;) {
         const size_t read_bytes = wait_stream->read(&buffer, 0, 1).wait(m_io_context.waitScope);
         if (read_bytes != 1) throw std::logic_error("EventLoop wait_stream closed unexpectedly");
-        std::unique_lock<std::mutex> lock(m_mutex);
+        Lock lock(m_mutex);
         if (m_post_fn) {
             Unlock(lock, *m_post_fn);
             m_post_fn = nullptr;
             m_cv.notify_all();
-        } else if (done(lock)) {
+        } else if (done()) {
             // Intentionally do not break if m_post_fn was set, even if done()
             // would return true, to ensure that the EventLoopRef write(post_fd)
             // call always succeeds and the loop does not exit between the time
@@ -243,7 +244,7 @@ void EventLoop::loop()
     log() << "EventLoop::loop bye.";
     wait_stream = nullptr;
     KJ_SYSCALL(::close(post_fd));
-    const std::unique_lock<std::mutex> lock(m_mutex);
+    const Lock lock(m_mutex);
     m_wait_fd = -1;
     m_post_fd = -1;
 }
@@ -254,26 +255,26 @@ void EventLoop::post(kj::Function<void()> fn)
         fn();
         return;
     }
-    std::unique_lock<std::mutex> lock(m_mutex);
+    Lock lock(m_mutex);
     EventLoopRef ref(*this, &lock);
-    m_cv.wait(lock, [this] { return m_post_fn == nullptr; });
+    m_cv.wait(lock.m_lock, [this]() MP_REQUIRES(m_mutex) { return m_post_fn == nullptr; });
     m_post_fn = &fn;
     int post_fd{m_post_fd};
     Unlock(lock, [&] {
         char buffer = 0;
         KJ_SYSCALL(write(post_fd, &buffer, 1));
     });
-    m_cv.wait(lock, [this, &fn] { return m_post_fn != &fn; });
+    m_cv.wait(lock.m_lock, [this, &fn]() MP_REQUIRES(m_mutex) { return m_post_fn != &fn; });
 }
 
-void EventLoop::startAsyncThread(std::unique_lock<std::mutex>& lock)
+void EventLoop::startAsyncThread()
 {
     if (m_async_thread.joinable()) {
         m_cv.notify_all();
     } else if (!m_async_fns.empty()) {
         m_async_thread = std::thread([this] {
-            std::unique_lock<std::mutex> lock(m_mutex);
-            while (!done(lock)) {
+            Lock lock(m_mutex);
+            while (!done()) {
                 if (!m_async_fns.empty()) {
                     EventLoopRef ref{*this, &lock};
                     const std::function<void()> fn = std::move(m_async_fns.front());
@@ -290,17 +291,15 @@ void EventLoop::startAsyncThread(std::unique_lock<std::mutex>& lock)
                     // Continue without waiting in case there are more async_fns
                     continue;
                 }
-                m_cv.wait(lock);
+                m_cv.wait(lock.m_lock);
             }
         });
     }
 }
 
-bool EventLoop::done(std::unique_lock<std::mutex>& lock)
+bool EventLoop::done()
 {
     assert(m_num_clients >= 0);
-    assert(lock.owns_lock());
-    assert(lock.mutex() == &m_mutex);
     return m_num_clients == 0 && m_async_fns.empty();
 }