SERVER-35219 Change the sleep on the destination side into a cond var wait on the donor side of session migration.

1 files changed, 36 insertions, 2 deletions

diff --git a/src/mongo/db/s/session_catalog_migration_source.h b/src/mongo/db/s/session_catalog_migration_source.h
index 04e1fc00132..b866e89e84a 100644
--- a/src/mongo/db/s/session_catalog_migration_source.h
+++ b/src/mongo/db/s/session_catalog_migration_source.h
@@ -39,6 +39,7 @@
 #include "mongo/db/session_txn_record_gen.h"
 #include "mongo/db/transaction_history_iterator.h"
 #include "mongo/stdx/mutex.h"
+#include "mongo/util/concurrency/notification.h"
 #include "mongo/util/concurrency/with_lock.h"
 
 namespace mongo {
@@ -100,6 +101,17 @@ public:
     bool fetchNextOplog(OperationContext* opCtx);
 
     /**
+     * Returns a notification that can be used to wait for new oplog entries to fetch. Note
+     * that this should only be called if hasMoreOplog/fetchNextOplog returned false at
+     * least once.
+     *
+     * If the notification is set to true, then that means that there is no longer a need to
+     * fetch more oplog because the data migration has entered the critical section and
+     * the buffer for oplog to fetch is empty or the data migration has aborted.
+     */
+    std::shared_ptr<Notification<bool>> getNotificationForNewOplog();
+
+    /**
      * Returns the oplog document that was last fetched by the fetchNextOplog call.
      * Returns an empty object if there are no oplog to fetch.
      */
@@ -117,6 +129,18 @@ public:
         return _rollbackIdAtInit;
     }
 
+    /**
+     * Inform this session migration machinery that the data migration just entered the critical
+     * section.
+     */
+    void onCommitCloneStarted();
+
+    /**
+     * Inform this session migration machinery that the data migration just terminated and
+     * entering the cleanup phase (can be aborted or committed).
+     */
+    void onCloneCleanup();
+
 private:
     /**
      * An iterator for extracting session write oplogs that need to be cloned during migration.
@@ -148,6 +172,8 @@ private:
         std::unique_ptr<TransactionHistoryIterator> _writeHistoryIterator;
     };
 
+    enum class State { kActive, kCommitStarted, kCleanup };
+
     ///////////////////////////////////////////////////////////////////////////
     // Methods for extracting the oplog entries from session information.
 
@@ -178,7 +204,7 @@ private:
     /**
      * Returns true if there are oplog generated by new writes that needs to be fetched.
      */
-    bool _hasNewWrites();
+    bool _hasNewWrites(WithLock);
 
     /**
      * Attempts to fetch the next oplog entry from the new writes that was saved by saveNewWriteTS.
@@ -211,7 +237,7 @@ private:
     // Used to store the last fetched oplog. This enables calling get multiple times.
     boost::optional<repl::OplogEntry> _lastFetchedOplog;
 
-    // Protects _newWriteTsList, _lastFetchedNewWriteOplog
+    // Protects _newWriteTsList, _lastFetchedNewWriteOplog, _state, _newOplogNotification
     stdx::mutex _newOplogMutex;
 
     // Stores oplog opTime of new writes that are coming in.
@@ -219,6 +245,14 @@ private:
 
     // Used to store the last fetched oplog from _newWriteTsList.
     boost::optional<repl::OplogEntry> _lastFetchedNewWriteOplog;
+
+    // Stores the current state.
+    State _state{State::kActive};
+
+    // Holds the latest request for notification of new oplog entries that needs to be fetched.
+    // Sets to true if there is no need to fetch an oplog anymore (for example, because migration
+    // aborted).
+    std::shared_ptr<Notification<bool>> _newOplogNotification;
 };
 
 }  // namespace mongo