Ensure that only a single prepare restart happens at the same time in proton.

5 files changed, 116 insertions, 31 deletions

diff --git a/searchcore/src/vespa/searchcore/proton/server/CMakeLists.txt b/searchcore/src/vespa/searchcore/proton/server/CMakeLists.txt
index 8ee6a4f73ae..b371439dd0e 100644
--- a/searchcore/src/vespa/searchcore/proton/server/CMakeLists.txt
+++ b/searchcore/src/vespa/searchcore/proton/server/CMakeLists.txt
@@ -68,6 +68,7 @@ vespa_add_library(searchcore_server STATIC
     operationdonecontext.cpp
     pendinglidtracker.cpp
     persistencehandlerproxy.cpp
+    prepare_restart_handler.cpp
     proton.cpp
     proton_config_fetcher.cpp
     proton_config_snapshot.cpp
diff --git a/searchcore/src/vespa/searchcore/proton/server/prepare_restart_handler.cpp b/searchcore/src/vespa/searchcore/proton/server/prepare_restart_handler.cpp
new file mode 100644
index 00000000000..aef49c51f7a
--- /dev/null
+++ b/searchcore/src/vespa/searchcore/proton/server/prepare_restart_handler.cpp
@@ -0,0 +1,60 @@
+// Copyright 2018 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include "prepare_restart_handler.h"
+#include <vespa/searchcore/proton/flushengine/flushengine.h>
+#include <vespa/searchcore/proton/flushengine/prepare_restart_flush_strategy.h>
+
+#include <vespa/log/log.h>
+LOG_SETUP(".proton.server.prepare_restart_handler");
+
+namespace proton {
+
+PrepareRestartHandler::PrepareRestartHandler(FlushEngine &flushEngine)
+    : _flushEngine(flushEngine),
+      _mutex(),
+      _cond(),
+      _running(false)
+{
+}
+
+bool
+PrepareRestartHandler::prepareRestart(const ProtonConfig &protonCfg)
+{
+    std::unique_lock lock(_mutex);
+    if (!_flushEngine.HasThread()) {
+        return false;
+    }
+    if (!_running) {
+        performPrepareRestart(protonCfg, lock);
+    } else {
+        _cond.wait(lock, [this] { return !_running; });
+        LOG(info, "prepareRestart(): Waited for another thread performing prepareRestart()");
+    }
+    return true;
+}
+
+namespace {
+
+PrepareRestartFlushStrategy::Config
+createPrepareRestartConfig(const PrepareRestartHandler::ProtonConfig &protonCfg)
+{
+    return PrepareRestartFlushStrategy::Config(protonCfg.flush.preparerestart.replaycost,
+                                               protonCfg.flush.preparerestart.replayoperationcost,
+                                               protonCfg.flush.preparerestart.writecost);
+}
+
+}
+
+void
+PrepareRestartHandler::performPrepareRestart(const ProtonConfig &protonCfg, std::unique_lock<std::mutex> &lock)
+{
+    _running = true;
+    lock.unlock();
+    auto strategy = std::make_shared<PrepareRestartFlushStrategy>(createPrepareRestartConfig(protonCfg));
+    _flushEngine.setStrategy(strategy);
+    lock.lock();
+    _running = false;
+    _cond.notify_all();
+}
+
+}
diff --git a/searchcore/src/vespa/searchcore/proton/server/prepare_restart_handler.h b/searchcore/src/vespa/searchcore/proton/server/prepare_restart_handler.h
new file mode 100644
index 00000000000..2adf01d36ee
--- /dev/null
+++ b/searchcore/src/vespa/searchcore/proton/server/prepare_restart_handler.h
@@ -0,0 +1,37 @@
+// Copyright 2018 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#pragma once
+
+#include <vespa/searchcore/config/config-proton.h>
+#include <condition_variable>
+#include <memory>
+#include <mutex>
+
+namespace proton {
+
+class FlushEngine;
+
+/**
+ * Class ensuring that only a single prepare restart happens at the same time.
+ *
+ * If another thread tries to start a new prepare restart while one is running, this thread waits
+ * until the ongoing operation is done and returns successfully. No extra work is done.
+ */
+class PrepareRestartHandler {
+public:
+    using ProtonConfig = vespa::config::search::core::internal::InternalProtonType;
+
+private:
+    FlushEngine &_flushEngine;
+    std::mutex _mutex;
+    std::condition_variable _cond;
+    bool _running;
+
+    void performPrepareRestart(const ProtonConfig &protonCfg, std::unique_lock<std::mutex> &lock);
+
+public:
+    PrepareRestartHandler(FlushEngine &flushEngine);
+    bool prepareRestart(const ProtonConfig &protonCfg);
+};
+
+}
diff --git a/searchcore/src/vespa/searchcore/proton/server/proton.cpp b/searchcore/src/vespa/searchcore/proton/server/proton.cpp
index d152fb1e2e1..492618b0472 100644
--- a/searchcore/src/vespa/searchcore/proton/server/proton.cpp
+++ b/searchcore/src/vespa/searchcore/proton/server/proton.cpp
@@ -2,26 +2,27 @@
 
 #include "disk_mem_usage_sampler.h"
 #include "document_db_explorer.h"
+#include "fileconfigmanager.h"
 #include "flushhandlerproxy.h"
 #include "memoryflush.h"
 #include "persistencehandlerproxy.h"
+#include "prepare_restart_handler.h"
 #include "proton.h"
+#include "proton_config_snapshot.h"
 #include "resource_usage_explorer.h"
 #include "searchhandlerproxy.h"
 #include "simpleflush.h"
-#include "proton_config_snapshot.h"
-#include "fileconfigmanager.h"
 
-#include <vespa/searchcore/proton/reference/document_db_reference_registry.h>
+#include <vespa/document/base/exceptions.h>
+#include <vespa/document/datatype/documenttype.h>
+#include <vespa/document/repo/documenttyperepo.h>
+#include <vespa/searchcommon/common/schemaconfigurer.h>
 #include <vespa/searchcore/proton/flushengine/flush_engine_explorer.h>
 #include <vespa/searchcore/proton/flushengine/prepare_restart_flush_strategy.h>
 #include <vespa/searchcore/proton/flushengine/tls_stats_factory.h>
+#include <vespa/searchcore/proton/reference/document_db_reference_registry.h>
 #include <vespa/searchlib/transactionlog/trans_log_server_explorer.h>
 #include <vespa/searchlib/util/fileheadertk.h>
-#include <vespa/searchcommon/common/schemaconfigurer.h>
-#include <vespa/document/base/exceptions.h>
-#include <vespa/document/datatype/documenttype.h>
-#include <vespa/document/repo/documenttyperepo.h>
 #include <vespa/vespalib/io/fileutil.h>
 #include <vespa/vespalib/util/closuretask.h>
 #include <vespa/vespalib/util/host_name.h>
@@ -177,6 +178,7 @@ Proton::Proton(const config::ConfigUri & configUri,
       _docsumBySlime(),
       _memoryFlushConfigUpdater(),
       _flushEngine(),
+      _prepareRestartHandler(),
       _rpcHooks(),
       _healthAdapter(*this),
       _genericStateHandler(CUSTOM_COMPONENT_API_PATH, *this),
@@ -295,6 +297,7 @@ Proton::init(const BootstrapConfig::SP & configSnapshot)
     _protonConfigurer.applyInitialConfig(initializeThreads);
     initializeThreads.reset();
 
+    _prepareRestartHandler = std::make_unique<PrepareRestartHandler>(*_flushEngine);
     RPCHooks::Params rpcParams(*this, protonConfig.rpcport, _configUri.getConfigId());
     rpcParams.slobrok_config = _configUri.createWithNewId(protonConfig.slobrokconfigid);
     _rpcHooks.reset(new RPCHooks(rpcParams));
@@ -634,29 +637,11 @@ Proton::triggerFlush()
     return true;
 }
 
-namespace {
-
-PrepareRestartFlushStrategy::Config
-createPrepareRestartConfig(const ProtonConfig &protonConfig)
-{
-    return PrepareRestartFlushStrategy::Config(protonConfig.flush.preparerestart.replaycost,
-                                               protonConfig.flush.preparerestart.replayoperationcost,
-                                               protonConfig.flush.preparerestart.writecost);
-}
-
-}
-
 bool
 Proton::prepareRestart()
 {
-    if (!_flushEngine || ! _flushEngine->HasThread()) {
-        return false;
-    }
     BootstrapConfig::SP configSnapshot = getActiveConfigSnapshot();
-    auto strategy = std::make_shared<PrepareRestartFlushStrategy>(
-            createPrepareRestartConfig(configSnapshot->getProtonConfig()));
-    _flushEngine->setStrategy(strategy);
-    return true;
+    return _prepareRestartHandler->prepareRestart(configSnapshot->getProtonConfig());
 }
 
 namespace {
diff --git a/searchcore/src/vespa/searchcore/proton/server/proton.h b/searchcore/src/vespa/searchcore/proton/server/proton.h
index e7e7153be8b..6dfc6c429a3 100644
--- a/searchcore/src/vespa/searchcore/proton/server/proton.h
+++ b/searchcore/src/vespa/searchcore/proton/server/proton.h
@@ -2,16 +2,16 @@
 
 #pragma once
 
-#include "health_adapter.h"
-#include "idocumentdbowner.h"
+#include "bootstrapconfig.h"
 #include "bootstrapconfigmanager.h"
 #include "documentdb.h"
+#include "health_adapter.h"
+#include "i_proton_configurer_owner.h"
+#include "idocumentdbowner.h"
 #include "memory_flush_config_updater.h"
 #include "proton_config_fetcher.h"
-#include "i_proton_configurer_owner.h"
 #include "proton_configurer.h"
 #include "rpc_hooks.h"
-#include "bootstrapconfig.h"
 #include <vespa/searchcore/proton/flushengine/flushengine.h>
 #include <vespa/searchcore/proton/matchengine/matchengine.h>
 #include <vespa/searchcore/proton/matching/querylimiter.h>
@@ -19,8 +19,8 @@
 #include <vespa/searchcore/proton/persistenceengine/i_resource_write_filter.h>
 #include <vespa/searchcore/proton/persistenceengine/ipersistenceengineowner.h>
 #include <vespa/searchcore/proton/persistenceengine/persistenceengine.h>
-#include <vespa/searchcore/proton/summaryengine/summaryengine.h>
 #include <vespa/searchcore/proton/summaryengine/docsum_by_slime.h>
+#include <vespa/searchcore/proton/summaryengine/summaryengine.h>
 #include <vespa/searchlib/common/fileheadercontext.h>
 #include <vespa/searchlib/engine/monitorapi.h>
 #include <vespa/searchlib/engine/transportserver.h>
@@ -38,6 +38,7 @@ namespace proton {
 
 class DiskMemUsageSampler;
 class IDocumentDBReferenceRegistry;
+class PrepareRestartHandler;
 
 class Proton : public IProtonConfigurerOwner,
                public search::engine::MonitorServer,
@@ -98,6 +99,7 @@ private:
     DocsumBySlime::UP               _docsumBySlime;
     MemoryFlushConfigUpdater::UP    _memoryFlushConfigUpdater;
     FlushEngine::UP                 _flushEngine;
+    std::unique_ptr<PrepareRestartHandler> _prepareRestartHandler;
     RPCHooks::UP                    _rpcHooks;
     HealthAdapter                   _healthAdapter;
     vespalib::GenericStateHandler   _genericStateHandler;