Make wait_for_(upload|download)_complete_or_client_stopped() thin wrappers around async completion

Author: tgoyne

src/realm/sync/client.cpp

@@ -228,13 +228,6 @@ class SessionWrapper final : public util::AtomicRefCountBase, DB::CommitListener
     std::vector<WaitOperCompletionHandler> m_download_completion_handlers;
     std::vector<WaitOperCompletionHandler> m_sync_completion_handlers;
 
-    // `m_target_*load_mark` and `m_reached_*load_mark` are protected by
-    // `m_client.m_mutex`. `m_staged_*load_mark` must only be accessed by the
-    // event loop thread.
-    std::int_fast64_t m_target_upload_mark = 0, m_target_download_mark = 0;
-    std::int_fast64_t m_staged_upload_mark = 0, m_staged_download_mark = 0;
-    std::int_fast64_t m_reached_upload_mark = 0, m_reached_download_mark = 0;
-
     void on_upload_completion();
     void on_download_completion();
     void on_suspended(const SessionErrorInfo& error_info);
@@ -1338,8 +1331,12 @@ void SessionWrapper::async_wait_for(bool upload_completion, bool download_comple
     REALM_ASSERT(upload_completion || download_completion);
 
     m_client.post([self = util::bind_ptr{this}, handler = std::move(handler), upload_completion,
-                   download_completion]() mutable {
+                   download_completion](Status status) mutable {
         REALM_ASSERT(self->m_actualized);
+        if (!status.is_ok()) {
+            handler(status); // Throws
+            return;
+        }
         if (REALM_UNLIKELY(!self->m_sess)) {
             // Already finalized
             handler({ErrorCodes::OperationAborted, "Session finalized before callback could run"}); // Throws
@@ -1373,36 +1370,11 @@ bool SessionWrapper::wait_for_upload_complete_or_client_stopped()
     // Thread safety required
     REALM_ASSERT(!m_abandoned);
 
-    std::int_fast64_t target_mark;
-    {
-        util::CheckedLockGuard lock{m_client.m_mutex};
-        target_mark = ++m_target_upload_mark;
-    }
-
-    m_client.post([self = util::bind_ptr{this}, target_mark] {
-        REALM_ASSERT(self->m_actualized);
-        // The session wrapper may already have been finalized. This can only
-        // happen if it was abandoned, but in that case, the call of
-        // wait_for_upload_complete_or_client_stopped() must have returned
-        // already.
-        if (REALM_UNLIKELY(!self->m_sess))
-            return;
-        if (target_mark > self->m_staged_upload_mark) {
-            self->m_staged_upload_mark = target_mark;
-            SessionImpl& sess = *self->m_sess;
-            sess.request_upload_completion_notification(); // Throws
-        }
-    }); // Throws
-
-    bool completion_condition_was_satisfied;
-    {
-        util::CheckedUniqueLock lock{m_client.m_mutex};
-        m_client.m_wait_or_client_stopped_cond.wait(lock.native_handle(), [&]() REQUIRES(m_client.m_mutex) {
-            return m_reached_upload_mark >= target_mark || m_client.m_stopped;
-        });
-        completion_condition_was_satisfied = !m_client.m_stopped;
-    }
-    return completion_condition_was_satisfied;
+    auto pf = util::make_promise_future<bool>();
+    async_wait_for(true, false, [promise = std::move(pf.promise)](Status status) mutable {
+        promise.emplace_value(status.is_ok());
+    });
+    return pf.future.get();
 }
 
 
@@ -1411,36 +1383,11 @@ bool SessionWrapper::wait_for_download_complete_or_client_stopped()
     // Thread safety required
     REALM_ASSERT(!m_abandoned);
 
-    std::int_fast64_t target_mark;
-    {
-        util::CheckedLockGuard lock{m_client.m_mutex};
-        target_mark = ++m_target_download_mark;
-    }
-
-    m_client.post([self = util::bind_ptr{this}, target_mark] {
-        REALM_ASSERT(self->m_actualized);
-        // The session wrapper may already have been finalized. This can only
-        // happen if it was abandoned, but in that case, the call of
-        // wait_for_download_complete_or_client_stopped() must have returned
-        // already.
-        if (REALM_UNLIKELY(!self->m_sess))
-            return;
-        if (target_mark > self->m_staged_download_mark) {
-            self->m_staged_download_mark = target_mark;
-            SessionImpl& sess = *self->m_sess;
-            sess.request_download_completion_notification(); // Throws
-        }
-    }); // Throws
-
-    bool completion_condition_was_satisfied;
-    {
-        util::CheckedUniqueLock lock{m_client.m_mutex};
-        m_client.m_wait_or_client_stopped_cond.wait(lock.native_handle(), [&]() REQUIRES(m_client.m_mutex) {
-            return m_reached_download_mark >= target_mark || m_client.m_stopped;
-        });
-        completion_condition_was_satisfied = !m_client.m_stopped;
-    }
-    return completion_condition_was_satisfied;
+    auto pf = util::make_promise_future<bool>();
+    async_wait_for(false, true, [promise = std::move(pf.promise)](Status status) mutable {
+        promise.emplace_value(status.is_ok());
+    });
+    return pf.future.get();
 }
 
 
@@ -1564,6 +1511,24 @@ void SessionWrapper::force_close()
     m_sess = nullptr;
     // Everything is being torn down, no need to report connection state anymore
     m_connection_state_change_listener = {};
+
+    // All outstanding wait operations must be canceled
+    while (!m_upload_completion_handlers.empty()) {
+        auto handler = std::move(m_upload_completion_handlers.back());
+        m_upload_completion_handlers.pop_back();
+        handler({ErrorCodes::OperationAborted, "Sync session is being closed before upload was complete"}); // Throws
+    }
+    while (!m_download_completion_handlers.empty()) {
+        auto handler = std::move(m_download_completion_handlers.back());
+        m_download_completion_handlers.pop_back();
+        handler(
+            {ErrorCodes::OperationAborted, "Sync session is being closed before download was complete"}); // Throws
+    }
+    while (!m_sync_completion_handlers.empty()) {
+        auto handler = std::move(m_sync_completion_handlers.back());
+        m_sync_completion_handlers.pop_back();
+        handler({ErrorCodes::OperationAborted, "Sync session is being closed before sync was complete"}); // Throws
+    }
 }
 
 // Must be called from event loop thread
@@ -1586,25 +1551,6 @@ void SessionWrapper::finalize()
     // deactivation.
     m_db->release_sync_agent();
     m_db = nullptr;
-
-    // All outstanding wait operations must be canceled
-    while (!m_upload_completion_handlers.empty()) {
-        auto handler = std::move(m_upload_completion_handlers.back());
-        m_upload_completion_handlers.pop_back();
-        handler(
-            {ErrorCodes::OperationAborted, "Sync session is being finalized before upload was complete"}); // Throws
-    }
-    while (!m_download_completion_handlers.empty()) {
-        auto handler = std::move(m_download_completion_handlers.back());
-        m_download_completion_handlers.pop_back();
-        handler(
-            {ErrorCodes::OperationAborted, "Sync session is being finalized before download was complete"}); // Throws
-    }
-    while (!m_sync_completion_handlers.empty()) {
-        auto handler = std::move(m_sync_completion_handlers.back());
-        m_sync_completion_handlers.pop_back();
-        handler({ErrorCodes::OperationAborted, "Sync session is being finalized before sync was complete"}); // Throws
-    }
 }
 
 
@@ -1636,11 +1582,6 @@ void SessionWrapper::on_upload_completion()
         m_download_completion_handlers.push_back(std::move(handler)); // Throws
         m_sync_completion_handlers.pop_back();
     }
-    util::CheckedLockGuard lock{m_client.m_mutex};
-    if (m_staged_upload_mark > m_reached_upload_mark) {
-        m_reached_upload_mark = m_staged_upload_mark;
-        m_client.m_wait_or_client_stopped_cond.notify_all();
-    }
 }
 
 
@@ -1663,12 +1604,6 @@ void SessionWrapper::on_download_completion()
         m_flx_subscription_store->update_state(m_flx_pending_mark_version, SubscriptionSet::State::Complete);
         m_flx_pending_mark_version = SubscriptionSet::EmptyVersion;
     }
-
-    util::CheckedLockGuard lock{m_client.m_mutex};
-    if (m_staged_download_mark > m_reached_download_mark) {
-        m_reached_download_mark = m_staged_download_mark;
-        m_client.m_wait_or_client_stopped_cond.notify_all();
-    }
 }