Adopt CheckedMutex in DefaultSocketProvider

tgoyne · tgoyne · commit 21ed9fed7c86 · 2024-05-29T11:26:47.000-07:00
diff --git a/src/realm/sync/network/default_socket.cpp b/src/realm/sync/network/default_socket.cpp
@@ -523,20 +523,21 @@ DefaultSocketProvider::~DefaultSocketProvider()
 
 void DefaultSocketProvider::start()
 {
-    std::unique_lock<std::mutex> lock(m_mutex);
+    util::CheckedUniqueLock lock(m_mutex);
     // Has the thread already been started or is running
     if (m_state == State::Starting || m_state == State::Running)
         return; // early return
 
     // If the thread has been previously run, make sure it has been joined first
-    if (m_state == State::Stopping) {
+    if (m_thread.joinable()) {
         state_wait_for(lock, State::Stopped);
+        m_thread.join();
     }
 
     m_logger_ptr->trace("Default event loop: start()");
     REALM_ASSERT(m_state == State::Stopped);
 
-    do_state_update(lock, State::Starting);
+    do_state_update(State::Starting);
     m_thread = std::thread{&DefaultSocketProvider::event_loop, this};
     // Wait for the thread to start before continuing
     state_wait_for(lock, State::Running);
@@ -545,17 +546,17 @@ void DefaultSocketProvider::start()
 void DefaultSocketProvider::OnlyForTesting::run_event_loop_on_current_thread(DefaultSocketProvider* provider)
 {
     {
-        std::unique_lock<std::mutex> lk(provider->m_mutex);
+        util::CheckedLockGuard lk(provider->m_mutex);
         REALM_ASSERT(provider->m_state == State::Stopped);
-        provider->do_state_update(lk, State::Starting);
+        provider->do_state_update(State::Starting);
     }
 
     provider->event_loop();
 }
 
 void DefaultSocketProvider::OnlyForTesting::prep_event_loop_for_restart(DefaultSocketProvider* provider)
 {
-    std::unique_lock<std::mutex> lk(provider->m_mutex);
+    util::CheckedLockGuard lk(provider->m_mutex);
     REALM_ASSERT(provider->m_state == State::Stopped);
     provider->m_service.reset();
 }
@@ -569,21 +570,22 @@ void DefaultSocketProvider::event_loop()
         if (m_observer_ptr)
             m_observer_ptr->will_destroy_thread();
 
-        std::unique_lock<std::mutex> lock(m_mutex);
-        // Did we get here due to an unhandled exception?
-        if (m_state != State::Stopping) {
-            m_logger_ptr->error("Default event loop: thread exited unexpectedly");
+        {
+            util::CheckedLockGuard lock(m_mutex);
+            // Did we get here due to an unhandled exception?
+            if (m_state != State::Stopping) {
+                m_logger_ptr->error("Default event loop: thread exited unexpectedly");
+            }
+            m_state = State::Stopped;
         }
-        m_state = State::Stopped;
-        lock.unlock();
         m_state_cv.notify_all();
     });
 
     if (m_observer_ptr)
         m_observer_ptr->did_create_thread();
 
     {
-        std::lock_guard<std::mutex> lock(m_mutex);
+        util::CheckedLockGuard lock(m_mutex);
         REALM_ASSERT(m_state == State::Starting);
     }
 
@@ -596,7 +598,7 @@ void DefaultSocketProvider::event_loop()
 
         REALM_ASSERT(status.is_ok());
 
-        std::unique_lock<std::mutex> lock(m_mutex);
+        util::CheckedLockGuard lock(m_mutex);
         // This is a callback from a previous generation
         if (m_event_loop_generation != my_generation) {
             return;
@@ -606,7 +608,7 @@ void DefaultSocketProvider::event_loop()
         }
         m_logger_ptr->trace("Default event loop: service run");
         REALM_ASSERT(m_state == State::Starting);
-        do_state_update(lock, State::Running);
+        do_state_update(State::Running);
     });
 
     // If there is no event loop observer or handle_error function registered, then just
@@ -619,11 +621,12 @@ void DefaultSocketProvider::event_loop()
             m_service.run_until_stopped(); // Throws
         }
         catch (const std::exception& e) {
-            REALM_ASSERT(m_observer_ptr); // should not change while event loop is running
-            std::unique_lock<std::mutex> lock(m_mutex);
-            // Service is no longer running, event loop thread is stopping
-            do_state_update(lock, State::Stopping);
-            lock.unlock();
+            {
+                util::CheckedLockGuard lock(m_mutex);
+                // Service is no longer running, event loop thread is stopping
+                m_state = State::Stopping;
+            }
+            m_state_cv.notify_all();
             m_logger_ptr->error("Default event loop exception: ", e.what());
             // If the error was not handled by the thread loop observer, then rethrow
             if (!m_observer_ptr->handle_error(e))
@@ -634,12 +637,12 @@ void DefaultSocketProvider::event_loop()
 
 void DefaultSocketProvider::stop(bool wait_for_stop)
 {
-    std::unique_lock<std::mutex> lock(m_mutex);
+    util::CheckedUniqueLock lock(m_mutex);
 
     // Do nothing if the thread is not started or running or stop has already been called
     if (m_state == State::Starting || m_state == State::Running) {
         m_logger_ptr->trace("Default event loop: stop()");
-        do_state_update(lock, State::Stopping);
+        do_state_update(State::Stopping);
         // Updating state to Stopping will free a start() if it is waiting for the thread to
         // start and may cause the thread to exit early before calling service.run()
         m_service.stop(); // Unblocks m_service.run()
@@ -661,24 +664,16 @@ void DefaultSocketProvider::stop(bool wait_for_stop)
 //                              |           |          ^
 //                              +----------------------+
 
-void DefaultSocketProvider::do_state_update(std::unique_lock<std::mutex>&, State new_state)
+void DefaultSocketProvider::do_state_update(State new_state)
 {
-    // m_state_mutex should already be locked...
     m_state = new_state;
     m_state_cv.notify_all(); // Let any waiters check the state
 }
 
-void DefaultSocketProvider::state_wait_for(std::unique_lock<std::mutex>& lock, State expected_state)
+void DefaultSocketProvider::state_wait_for(util::CheckedUniqueLock& lock, State expected_state)
 {
-    // Check for condition already met or superseded
-    if (m_state >= expected_state)
-        return;
-
-    m_state_cv.wait(lock, [this, expected_state]() {
-        // are we there yet?
-        if (m_state < expected_state)
-            return false;
-        return true;
+    m_state_cv.wait(lock.native_handle(), [this, expected_state]() REQUIRES(m_mutex) {
+        return m_state >= expected_state;
     });
 }
 
diff --git a/src/realm/sync/network/default_socket.hpp b/src/realm/sync/network/default_socket.hpp
@@ -5,6 +5,7 @@
 #include <realm/sync/socket_provider.hpp>
 #include <realm/sync/network/http.hpp>
 #include <realm/sync/network/network.hpp>
+#include <realm/util/checked_mutex.hpp>
 #include <realm/util/future.hpp>
 #include <realm/util/tagged_bool.hpp>
 
@@ -22,7 +23,7 @@ using port_type = sync::port_type;
 
 class DefaultSocketProvider : public SyncSocketProvider {
 public:
-    class Timer : public SyncSocketProvider::Timer {
+    class Timer final : public SyncSocketProvider::Timer {
     public:
         friend class DefaultSocketProvider;
 
@@ -35,33 +36,30 @@ class DefaultSocketProvider : public SyncSocketProvider {
             m_timer.cancel();
         }
 
-    protected:
+    private:
+        network::DeadlineTimer m_timer;
+
         Timer(network::Service& service, std::chrono::milliseconds delay, FunctionHandler&& handler)
             : m_timer{service}
         {
             m_timer.async_wait(delay, std::move(handler));
         }
-
-    private:
-        network::DeadlineTimer m_timer;
     };
 
-    struct AutoStartTag {};
-
-    using AutoStart = util::TaggedBool<AutoStartTag>;
+    using AutoStart = util::TaggedBool<struct AutoStartTag>;
     DefaultSocketProvider(const std::shared_ptr<util::Logger>& logger, const std::string& user_agent,
                           const std::shared_ptr<BindingCallbackThreadObserver>& observer_ptr = nullptr,
                           AutoStart auto_start = AutoStart{true});
 
     ~DefaultSocketProvider();
 
     // Start the event loop if it is not started already. Otherwise, do nothing.
-    void start();
+    void start() REQUIRES(!m_mutex);
 
     /// Temporary workaround until client shutdown has been updated in a separate PR - these functions
     /// will be handled internally when this happens.
     /// Stops the internal event loop (provided by network::Service)
-    void stop(bool wait_for_stop = false) override;
+    void stop(bool wait_for_stop = false) override REQUIRES(!m_mutex);
 
     std::unique_ptr<WebSocketInterface> connect(std::unique_ptr<WebSocketObserver>, WebSocketEndpoint&&) override;
 
@@ -89,23 +87,23 @@ class DefaultSocketProvider : public SyncSocketProvider {
 private:
     enum class State { Starting, Running, Stopping, Stopped };
 
-    /// Block until the state reaches the expected or later state - return true if state matches expected state
-    void state_wait_for(std::unique_lock<std::mutex>& lock, State expected_state);
+    /// Block until the state reaches the expected or later state
+    void state_wait_for(util::CheckedUniqueLock& lock, State expected_state) REQUIRES(m_mutex);
     /// Internal function for updating the state and signaling the wait_for_state condvar
-    void do_state_update(std::unique_lock<std::mutex>&, State new_state);
+    void do_state_update(State new_state) REQUIRES(m_mutex);
     /// The execution code for the event loop thread
-    void event_loop();
+    void event_loop() REQUIRES(!m_mutex);
 
     std::shared_ptr<util::Logger> m_logger_ptr;
-    std::shared_ptr<BindingCallbackThreadObserver> m_observer_ptr;
+    const std::shared_ptr<BindingCallbackThreadObserver> m_observer_ptr;
     network::Service m_service;
     std::mt19937_64 m_random;
     const std::string m_user_agent;
-    std::mutex m_mutex;
+    util::CheckedMutex m_mutex;
     uint64_t m_event_loop_generation = 0;
-    State m_state;                      // protected by m_mutex
-    std::condition_variable m_state_cv; // uses m_mutex
-    std::thread m_thread;               // protected by m_mutex
+    State m_state GUARDED_BY(m_mutex);
+    std::condition_variable m_state_cv;
+    std::thread m_thread GUARDED_BY(m_mutex);
 };
 
 /// Class for the Default Socket Provider websockets that allows a simulated
@@ -115,8 +113,6 @@ class DefaultWebSocket : public WebSocketInterface {
     virtual ~DefaultWebSocket() = default;
 
     virtual void force_handshake_response_for_testing(int status_code, std::string body = "") = 0;
-
-protected:
 };
 
 } // namespace realm::sync::websocket
diff --git a/src/realm/sync/noinst/client_impl_base.hpp b/src/realm/sync/noinst/client_impl_base.hpp
@@ -219,7 +219,8 @@ class ClientImpl {
     // SyncSocketProvider
     void post(SyncSocketProvider::FunctionHandler&& handler) REQUIRES(!m_drain_mutex);
     SyncSocketProvider::SyncTimer create_timer(std::chrono::milliseconds delay,
-                                               SyncSocketProvider::FunctionHandler&& handler) REQUIRES(!m_drain_mutex);
+                                               SyncSocketProvider::FunctionHandler&& handler)
+        REQUIRES(!m_drain_mutex);
     using SyncTrigger = std::unique_ptr<Trigger<ClientImpl>>;
     SyncTrigger create_trigger(SyncSocketProvider::FunctionHandler&& handler);
 
@@ -336,7 +337,8 @@ class ClientImpl {
                                            bool verify_servers_ssl_certificate,
                                            util::Optional<std::string> ssl_trust_certificate_path,
                                            std::function<SyncConfig::SSLVerifyCallback>,
-                                           util::Optional<SyncConfig::ProxyConfig>, bool& was_created) REQUIRES(!m_drain_mutex);
+                                           util::Optional<SyncConfig::ProxyConfig>, bool& was_created)
+        REQUIRES(!m_drain_mutex);
 
     // Destroys the specified connection.
     void remove_connection(ClientImpl::Connection&) noexcept REQUIRES(!m_drain_mutex);
