SERVER-51806 Enable lock yielding during the bulk load phase of index builds

(cherry picked from commit 935bd19)

SERVER-59190 IndexAccessMethod can be destructed during index build bulk load yield

(cherry picked from commit 50cc968)

5 files changed, 119 insertions, 3 deletions

diff --git a/jstests/noPassthrough/index_build_yield_bulk_load.js b/jstests/noPassthrough/index_build_yield_bulk_load.js
new file mode 100644
index 00000000000..e9c01c78ff4
--- /dev/null
+++ b/jstests/noPassthrough/index_build_yield_bulk_load.js
@@ -0,0 +1,61 @@
+/*
+ * This test ensures an index build can yield during bulk load phase.
+ */
+(function() {
+
+"use strict";
+
+load("jstests/noPassthrough/libs/index_build.js");
+load("jstests/libs/fail_point_util.js");
+
+const mongodOptions = {};
+const conn = MongoRunner.runMongod(mongodOptions);
+
+TestData.dbName = jsTestName();
+TestData.collName = "coll";
+
+const testDB = conn.getDB(TestData.dbName);
+testDB.getCollection(TestData.collName).drop();
+
+assert.commandWorked(testDB.createCollection(TestData.collName));
+const coll = testDB.getCollection(TestData.collName);
+
+for (let i = 0; i < 3; i++) {
+    assert.commandWorked(coll.insert({_id: i, x: i}));
+}
+
+// Make the index build bulk load yield often.
+assert.commandWorked(
+    conn.adminCommand({setParameter: 1, internalIndexBuildBulkLoadYieldIterations: 1}));
+
+jsTestLog("Enable hangDuringIndexBuildBulkLoadYield fail point");
+let failpoint = configureFailPoint(
+    testDB, "hangDuringIndexBuildBulkLoadYield", {namespace: coll.getFullName()});
+
+jsTestLog("Create index");
+const awaitIndex = IndexBuildTest.startIndexBuild(
+    testDB.getMongo(), coll.getFullName(), {x: 1}, {}, [ErrorCodes.IndexBuildAborted]);
+
+// Wait until index build (bulk load phase) yields.
+jsTestLog("Wait for the index build to yield and hang");
+failpoint.wait();
+
+jsTestLog("Drop the collection");
+const awaitDrop = startParallelShell(() => {
+    const testDB = db.getSiblingDB(TestData.dbName);
+    assert.commandWorked(testDB.runCommand({drop: TestData.collName}));
+}, conn.port);
+
+// Wait until the index build starts aborting to make sure the drop happens before the index build
+// finishes.
+checkLog.containsJson(testDB, 465611);
+failpoint.off();
+
+// "Index build thread exited".
+checkLog.containsJson(testDB, 20655);
+
+awaitIndex();
+awaitDrop();
+
+MongoRunner.stopMongod(conn);
+})();
\ No newline at end of file
diff --git a/src/mongo/db/catalog/multi_index_block.cpp b/src/mongo/db/catalog/multi_index_block.cpp
index d3f9b16afbc..8112fd2514f 100644
--- a/src/mongo/db/catalog/multi_index_block.cpp
+++ b/src/mongo/db/catalog/multi_index_block.cpp
@@ -653,6 +653,11 @@ Status MultiIndexBlock::dumpInsertsFromBulk(
     opCtx->checkForInterrupt();
     invariant(!_buildIsCleanedUp);
     invariant(opCtx->lockState()->isNoop() || !opCtx->lockState()->inAWriteUnitOfWork());
+
+    // Doesn't allow yielding when in a foreground index build.
+    const int32_t kYieldIterations =
+        isBackgroundBuilding() ? internalIndexBuildBulkLoadYieldIterations.load() : 0;
+
     for (size_t i = 0; i < _indexes.size(); i++) {
         if (_indexes[i].bulk == nullptr)
             continue;
@@ -676,6 +681,7 @@ Status MultiIndexBlock::dumpInsertsFromBulk(
                 opCtx,
                 _indexes[i].bulk.get(),
                 dupsAllowed,
+                kYieldIterations,
                 [=](const KeyString::Value& duplicateKey) -> Status {
                     // Do not record duplicates when explicitly ignored. This may be the case on
                     // secondaries.
diff --git a/src/mongo/db/catalog/multi_index_block.idl b/src/mongo/db/catalog/multi_index_block.idl
index 311a7f701f1..fd50131a42f 100644
--- a/src/mongo/db/catalog/multi_index_block.idl
+++ b/src/mongo/db/catalog/multi_index_block.idl
@@ -61,3 +61,14 @@ server_parameters:
     default: 200
     validator:
       gte: 50
+
+  internalIndexBuildBulkLoadYieldIterations:
+    description: "The number of keys bulk-loaded before yielding."
+    set_at:
+      - runtime
+      - startup
+    cpp_varname: internalIndexBuildBulkLoadYieldIterations
+    cpp_vartype: AtomicWord<int>
+    default: 1000
+    validator:
+      gte: 1
diff --git a/src/mongo/db/index/index_access_method.cpp b/src/mongo/db/index/index_access_method.cpp
index 030e3600bd3..b77a25f19e0 100644
--- a/src/mongo/db/index/index_access_method.cpp
+++ b/src/mongo/db/index/index_access_method.cpp
@@ -66,6 +66,8 @@ using std::vector;
 
 using IndexVersion = IndexDescriptor::IndexVersion;
 
+MONGO_FAIL_POINT_DEFINE(hangDuringIndexBuildBulkLoadYield);
+
 namespace {
 
 // Reserved RecordId against which multikey metadata keys are indexed.
@@ -603,13 +605,39 @@ int64_t AbstractIndexAccessMethod::BulkBuilderImpl::getKeysInserted() const {
     return _keysInserted;
 }
 
+void AbstractIndexAccessMethod::_yieldBulkLoad(OperationContext* opCtx,
+                                               const NamespaceString& ns) const {
+    // Releasing locks means a new snapshot should be acquired when restored.
+    opCtx->recoveryUnit()->abandonSnapshot();
+
+    auto locker = opCtx->lockState();
+    Locker::LockSnapshot snapshot;
+    if (locker->saveLockStateAndUnlock(&snapshot)) {
+
+        // Track the number of yields in CurOp.
+        CurOp::get(opCtx)->yielded();
+
+        hangDuringIndexBuildBulkLoadYield.executeIf(
+            [&](auto&&) {
+                LOGV2(5180600, "Hanging index build during bulk load yield");
+                hangDuringIndexBuildBulkLoadYield.pauseWhileSet();
+            },
+            [&](auto&& config) { return config.getStringField("namespace") == ns.ns(); });
+
+        locker->restoreLockState(opCtx, snapshot);
+    }
+}
+
 Status AbstractIndexAccessMethod::commitBulk(OperationContext* opCtx,
                                              BulkBuilder* bulk,
                                              bool dupsAllowed,
+                                             int32_t yieldIterations,
                                              const KeyHandlerFn& onDuplicateKeyInserted,
                                              const RecordIdHandlerFn& onDuplicateRecord) {
     Timer timer;
 
+    auto ns = _indexCatalogEntry->ns();
+
     std::unique_ptr<BulkBuilder::Sorter::Iterator> it(bulk->done());
 
     static constexpr char message[] = "Index Build: inserting keys from external sorter into index";
@@ -625,7 +653,7 @@ Status AbstractIndexAccessMethod::commitBulk(OperationContext* opCtx,
 
     KeyString::Value previousKey;
 
-    while (it->more()) {
+    for (int64_t i = 1; it->more(); ++i) {
         opCtx->checkForInterrupt();
 
         // Get the next datum and add it to the builder.
@@ -674,6 +702,11 @@ Status AbstractIndexAccessMethod::commitBulk(OperationContext* opCtx,
                 return status;
         }
 
+        // Starts yielding locks after the first non-zero 'yieldIterations' inserts.
+        if (yieldIterations && i % yieldIterations == 0) {
+            _yieldBulkLoad(opCtx, ns);
+        }
+
         // If we're here either it's a dup and we're cool with it or the addKey went just fine.
         pm.hit();
     }
@@ -763,7 +796,7 @@ void AbstractIndexAccessMethod::getKeys(const BSONObj& obj,
         }
 
         // If the document applies to the filter (which means that it should have never been
-        // indexed), do not supress the error.
+        // indexed), do not suppress the error.
         const MatchExpression* filter = _indexCatalogEntry->getFilterExpression();
         if (mode == GetKeysMode::kRelaxConstraintsUnfiltered && filter &&
             filter->matchesBSON(obj)) {
diff --git a/src/mongo/db/index/index_access_method.h b/src/mongo/db/index/index_access_method.h
index 6e40b9153b6..e2a0f5b6ef2 100644
--- a/src/mongo/db/index/index_access_method.h
+++ b/src/mongo/db/index/index_access_method.h
@@ -266,9 +266,10 @@ public:
      * Call this when you are ready to finish your bulk work.
      * Pass in the BulkBuilder returned from initiateBulk.
      * @param bulk - Something created from initiateBulk
-     * @param mayInterrupt - Is this commit interruptible (will cancel)
      * @param dupsAllowed - If false and 'dupRecords' is not null, append with the RecordIds of
      *                      the uninserted duplicates.
+     * @param yieldIterations - The number of iterations run before each yielding. Will not yield if
+     * zero.
      * @param onDuplicateKeyInserted - Will be called for each duplicate key inserted into the
      * index.
      * @param onDuplicateRecord - If not nullptr, will be called for each RecordId of uninserted
@@ -277,6 +278,7 @@ public:
     virtual Status commitBulk(OperationContext* opCtx,
                               BulkBuilder* bulk,
                               bool dupsAllowed,
+                              int32_t yieldIterations,
                               const KeyHandlerFn& onDuplicateKeyInserted,
                               const RecordIdHandlerFn& onDuplicateRecord) = 0;
 
@@ -529,6 +531,7 @@ public:
     Status commitBulk(OperationContext* opCtx,
                       BulkBuilder* bulk,
                       bool dupsAllowed,
+                      int32_t yieldIterations,
                       const KeyHandlerFn& onDuplicateKeyInserted,
                       const RecordIdHandlerFn& onDuplicateRecord) final;
 
@@ -601,6 +604,8 @@ private:
                                const KeyString::Value& dataKey,
                                const RecordIdHandlerFn& onDuplicateRecord);
 
+    void _yieldBulkLoad(OperationContext* opCtx, const NamespaceString& ns) const;
+
     const std::unique_ptr<SortedDataInterface> _newInterface;
 };