SERVER-30096 Add stepdown/stepup logic to ChunkSplitter

6 files changed, 120 insertions, 83 deletions

diff --git a/src/mongo/db/s/SConscript b/src/mongo/db/s/SConscript
index dfbd5aeea04..26d13d499e9 100644
--- a/src/mongo/db/s/SConscript
+++ b/src/mongo/db/s/SConscript
@@ -62,6 +62,7 @@ env.Library(
     source=[
         'active_migrations_registry.cpp',
         'chunk_move_write_concern_options.cpp',
+        'chunk_splitter.cpp',
         'collection_range_deleter.cpp',
         'collection_sharding_state.cpp',
         'metadata_manager.cpp',
@@ -304,17 +305,6 @@ env.CppUnitTest(
 )
 
 env.Library(
-    target='chunk_splitter',
-    source=[
-        'chunk_splitter.cpp',
-    ],
-    LIBDEPS=[
-        '$BUILD_DIR/mongo/util/concurrency/thread_pool',
-        '$BUILD_DIR/mongo/db/service_context',
-    ]
-)
-
-env.Library(
     target='split_vector',
     source=[
     'split_vector.cpp',
@@ -335,17 +325,6 @@ env.Library(
 )
 
 env.CppUnitTest(
-    target='chunk_splitter_test',
-    source=[
-        'chunk_splitter_test.cpp'
-    ],
-    LIBDEPS=[
-        'chunk_splitter',
-        '$BUILD_DIR/mongo/s/shard_server_test_fixture'
-    ]
-)
-
-env.CppUnitTest(
     target='split_vector_test',
     source=[
         'split_vector_test.cpp',
diff --git a/src/mongo/db/s/chunk_splitter.cpp b/src/mongo/db/s/chunk_splitter.cpp
index 1f0575ce278..3088a74933c 100644
--- a/src/mongo/db/s/chunk_splitter.cpp
+++ b/src/mongo/db/s/chunk_splitter.cpp
@@ -26,14 +26,15 @@
  *    it in the license file.
  */
 
+#define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kSharding
+
 #include "mongo/platform/basic.h"
 
 #include "mongo/db/s/chunk_splitter.h"
 
 #include "mongo/db/client.h"
-#include "mongo/db/namespace_string.h"
-#include "mongo/db/operation_context.h"
-#include "mongo/s/catalog/type_chunk.h"
+#include "mongo/util/assert_util.h"
+#include "mongo/util/log.h"
 
 namespace mongo {
 namespace {
@@ -56,7 +57,7 @@ ThreadPool::Options makeDefaultThreadPoolOptions() {
 
 }  // namespace
 
-ChunkSplitter::ChunkSplitter() : _threadPool(makeDefaultThreadPoolOptions()) {
+ChunkSplitter::ChunkSplitter() : _isPrimary(false), _threadPool(makeDefaultThreadPoolOptions()) {
     _threadPool.startup();
 }
 
@@ -65,10 +66,56 @@ ChunkSplitter::~ChunkSplitter() {
     _threadPool.join();
 }
 
-bool ChunkSplitter::trySplitting(OperationContext* opCtx,
-                                 const NamespaceString& nss,
-                                 const ChunkRange& chunkRange) {
-    return false;
+void ChunkSplitter::setReplicaSetMode(bool isPrimary) {
+    stdx::lock_guard<stdx::mutex> scopedLock(_mutex);
+    _isPrimary = isPrimary;
+}
+
+void ChunkSplitter::initiateChunkSplitter() {
+    stdx::lock_guard<stdx::mutex> scopedLock(_mutex);
+    if (_isPrimary) {
+        return;
+    }
+    _isPrimary = true;
+
+    log() << "The ChunkSplitter has started and will accept autosplit tasks. Any tasks that did not"
+          << " have time to drain the last time this node was a primary shall be run.";
+}
+
+void ChunkSplitter::interruptChunkSplitter() {
+    stdx::lock_guard<stdx::mutex> scopedLock(_mutex);
+    if (!_isPrimary) {
+        return;
+    }
+    _isPrimary = false;
+
+    log() << "The ChunkSplitter has stopped and will no longer run autosplit tasks. Any autosplit "
+          << "tasks that have already started will be allowed to finish.";
+}
+
+void ChunkSplitter::trySplitting(const NamespaceString& nss,
+                                 const BSONObj& min,
+                                 const BSONObj& max) {
+    if (!_isPrimary) {
+        return;
+    }
+
+    uassertStatusOK(
+        _threadPool.schedule([ this, nss, min, max ]() noexcept { _runAutosplit(nss, min, max); }));
+}
+
+void ChunkSplitter::_runAutosplit(const NamespaceString& nss,
+                                  const BSONObj& min,
+                                  const BSONObj& max) {
+    if (!_isPrimary) {
+        return;
+    }
+
+    try {
+        // TODO SERVER-30020
+    } catch (const std::exception& e) {
+        log() << "caught exception while splitting chunk: " << redact(e.what());
+    }
 }
 
 }  // namespace mongo
diff --git a/src/mongo/db/s/chunk_splitter.h b/src/mongo/db/s/chunk_splitter.h
index 5f6b62affc3..ee86b4bc4ad 100644
--- a/src/mongo/db/s/chunk_splitter.h
+++ b/src/mongo/db/s/chunk_splitter.h
@@ -32,27 +32,55 @@
 
 namespace mongo {
 
-class ChunkRange;
 class NamespaceString;
-class OperationContext;
-class ThreadPool;
 
 /**
  * Handles asynchronous auto-splitting of chunks.
  */
 class ChunkSplitter {
+    MONGO_DISALLOW_COPYING(ChunkSplitter);
+
 public:
     ChunkSplitter();
     ~ChunkSplitter();
 
     /**
-     * Schedules an autosplit task. Returns whether or not the task was successfully scheduled.
+     * Sets the mode of the ChunkSplitter to either primary or secondary.
+     * The ChunkSplitter is only active when primary.
+     */
+    void setReplicaSetMode(bool isPrimary);
+
+    /**
+     * Invoked when the shard server primary enters the 'PRIMARY' state to set up the ChunkSplitter
+     * to begin accepting split requests.
      */
-    bool trySplitting(OperationContext* opCtx,
-                      const NamespaceString& nss,
-                      const ChunkRange& chunkRange);
+    void initiateChunkSplitter();
+
+    /**
+     * Invoked when this node which is currently serving as a 'PRIMARY' steps down.
+     *
+     * This method might be called multiple times in succession, which is what happens as a result
+     * of incomplete transition to primary so it is resilient to that.
+     */
+    void interruptChunkSplitter();
+
+    /**
+     * Schedules an autosplit task. This function throws on scheduling failure.
+     */
+    void trySplitting(const NamespaceString& nss, const BSONObj& min, const BSONObj& max);
 
 private:
+    /**
+     * Determines if the specified chunk should be split and then performs any necessary split.
+     */
+    void _runAutosplit(const NamespaceString& nss, const BSONObj& min, const BSONObj& max);
+
+    // Protects the state below.
+    stdx::mutex _mutex;
+
+    // The ChunkSplitter is only active on a primary node.
+    bool _isPrimary;
+
     // Thread pool for parallelizing splits.
     ThreadPool _threadPool;
 };
diff --git a/src/mongo/db/s/chunk_splitter_test.cpp b/src/mongo/db/s/chunk_splitter_test.cpp
deleted file mode 100644
index 4668fbd9743..00000000000
--- a/src/mongo/db/s/chunk_splitter_test.cpp
+++ /dev/null
@@ -1,44 +0,0 @@
-/**
- *    Copyright (C) 2017 MongoDB Inc.
- *
- *    This program is free software: you can redistribute it and/or  modify
- *    it under the terms of the GNU Affero General Public License, version 3,
- *    as published by the Free Software Foundation.
- *
- *    This program is distributed in the hope that it will be useful,
- *    but WITHOUT ANY WARRANTY; without even the implied warranty of
- *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- *    GNU Affero General Public License for more details.
- *
- *    You should have received a copy of the GNU Affero General Public License
- *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
- *
- *    As a special exception, the copyright holders give permission to link the
- *    code of portions of this program with the OpenSSL library under certain
- *    conditions as described in each individual source file and distribute
- *    linked combinations including the program with the OpenSSL library. You
- *    must comply with the GNU Affero General Public License in all respects for
- *    all of the code used other than as permitted herein. If you modify file(s)
- *    with this exception, you may extend this exception to your version of the
- *    file(s), but you are not obligated to do so. If you do not wish to do so,
- *    delete this exception statement from your version. If you delete this
- *    exception statement from all source files in the program, then also delete
- *    it in the license file.
- */
-
-#include "mongo/platform/basic.h"
-
-#include "mongo/db/s/chunk_splitter.h"
-
-#include "mongo/s/shard_server_test_fixture.h"
-
-namespace mongo {
-namespace {
-
-class ChunkSplitterTest : public ShardServerTestFixture {};
-
-TEST_F(ChunkSplitterTest, SplitTest) {}
-TEST_F(ChunkSplitterTest, DontSplitTest) {}
-
-}  // namespace
-}  // namespace mongo
diff --git a/src/mongo/db/s/sharding_state.cpp b/src/mongo/db/s/sharding_state.cpp
index 693b9085684..fd868900bdd 100644
--- a/src/mongo/db/s/sharding_state.cpp
+++ b/src/mongo/db/s/sharding_state.cpp
@@ -113,7 +113,8 @@ void updateShardIdentityConfigStringCB(const string& setName, const string& newC
 }  // namespace
 
 ShardingState::ShardingState()
-    : _initializationState(static_cast<uint32_t>(InitializationState::kNew)),
+    : _chunkSplitter(stdx::make_unique<ChunkSplitter>()),
+      _initializationState(static_cast<uint32_t>(InitializationState::kNew)),
       _initializationStatus(Status(ErrorCodes::InternalError, "Uninitialized value")),
       _globalInit(&initializeGlobalShardingStateForMongod) {}
 
@@ -204,6 +205,14 @@ CollectionShardingState* ShardingState::getNS(const std::string& ns, OperationCo
     return it->second.get();
 }
 
+void ShardingState::initiateChunkSplitter() {
+    _chunkSplitter->initiateChunkSplitter();
+}
+
+void ShardingState::interruptChunkSplitter() {
+    _chunkSplitter->interruptChunkSplitter();
+}
+
 void ShardingState::markCollectionsNotShardedAtStepdown() {
     stdx::lock_guard<stdx::mutex> lk(_mutex);
     for (auto& coll : _collections) {
@@ -324,7 +333,8 @@ Status ShardingState::initializeFromShardIdentity(OperationContext* opCtx,
                 &ShardRegistry::replicaSetChangeShardRegistryUpdateHook);
             ReplicaSetMonitor::setAsynchronousConfigChangeHook(&updateShardIdentityConfigStringCB);
 
-            // Determine primary/secondary/standalone state in order to set it on the CatalogCache.
+            // Determine primary/secondary/standalone state in order to properly initialize sharding
+            // components.
             auto replCoord = repl::ReplicationCoordinator::get(opCtx);
             bool isReplSet =
                 replCoord->getReplicationMode() == repl::ReplicationCoordinator::modeReplSet;
@@ -333,6 +343,7 @@ Status ShardingState::initializeFromShardIdentity(OperationContext* opCtx,
                                repl::MemberState::RS_PRIMARY);
 
             Grid::get(opCtx)->catalogCache()->initializeReplicaSetRole(isStandaloneOrPrimary);
+            _chunkSplitter->setReplicaSetMode(isStandaloneOrPrimary);
 
             log() << "initialized sharding components for "
                   << (isStandaloneOrPrimary ? "primary" : "secondary") << " node.";
diff --git a/src/mongo/db/s/sharding_state.h b/src/mongo/db/s/sharding_state.h
index d5f53976b7f..51654ee0a9a 100644
--- a/src/mongo/db/s/sharding_state.h
+++ b/src/mongo/db/s/sharding_state.h
@@ -35,6 +35,7 @@
 #include "mongo/bson/oid.h"
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/s/active_migrations_registry.h"
+#include "mongo/db/s/chunk_splitter.h"
 #include "mongo/db/s/collection_range_deleter.h"
 #include "mongo/db/s/migration_destination_manager.h"
 #include "mongo/executor/task_executor.h"
@@ -143,6 +144,18 @@ public:
     CollectionShardingState* getNS(const std::string& ns, OperationContext* opCtx);
 
     /**
+     * Should be invoked when the shard server primary enters the 'PRIMARY' state.
+     * Sets up the ChunkSplitter to begin accepting split requests.
+     */
+    void initiateChunkSplitter();
+
+    /**
+     * Should be invoked when this node which is currently serving as a 'PRIMARY' steps down.
+     * Sets the state of the ChunkSplitter so that it will no longer accept split requests.
+     */
+    void interruptChunkSplitter();
+
+    /**
      * Iterates through all known sharded collections and marks them (in memory only) as not sharded
      * so that no filtering will be happening for slaveOk queries.
      */
@@ -311,6 +324,9 @@ private:
     // Tracks the active move chunk operations running on this shard
     ActiveMigrationsRegistry _activeMigrationsRegistry;
 
+    // Handles asynchronous auto-splitting of chunks
+    std::unique_ptr<ChunkSplitter> _chunkSplitter;
+
     // Protects state below
     stdx::mutex _mutex;