Use standard locking in StorageBucketDBInitializer.

4 files changed, 33 insertions, 13 deletions

diff --git a/storage/src/vespa/storage/bucketdb/storagebucketdbinitializer.cpp b/storage/src/vespa/storage/bucketdb/storagebucketdbinitializer.cpp
index fc2c2066b6f..1bc473308a4 100644
--- a/storage/src/vespa/storage/bucketdb/storagebucketdbinitializer.cpp
+++ b/storage/src/vespa/storage/bucketdb/storagebucketdbinitializer.cpp
@@ -13,11 +13,13 @@
 #include <vespa/vespalib/stllike/hash_map.hpp>
 #include <vespa/config/helper/configgetter.hpp>
 #include <iomanip>
+#include <chrono>
 
 #include <vespa/log/bufferedlogger.h>
 LOG_SETUP(".storage.bucketdb.initializer");
 
 using document::BucketSpace;
+using namespace std::chrono_literals;
 
 namespace storage {
 
@@ -117,10 +119,11 @@ StorageBucketDBInitializer::Metrics::Metrics(framework::Component& component)
 StorageBucketDBInitializer::Metrics::~Metrics() {}
 
 StorageBucketDBInitializer::GlobalState::GlobalState()
-        : _insertedCount(0), _infoReadCount(0),
-          _infoSetByLoad(0), _dirsListed(0), _dirsToList(0),
-          _gottenInitProgress(false), _doneListing(false),
-          _doneInitializing(false)
+    : _lists(), _joins(), _infoRequests(), _replies(),
+      _insertedCount(0), _infoReadCount(0),
+      _infoSetByLoad(0), _dirsListed(0), _dirsToList(0),
+      _gottenInitProgress(false), _doneListing(false),
+      _doneInitializing(false), _workerLock(), _workerCond(), _replyLock()
 { }
 StorageBucketDBInitializer::GlobalState::~GlobalState() { }
 
@@ -183,7 +186,7 @@ void
 StorageBucketDBInitializer::onClose()
 {
     if (_system._thread.get() != 0) {
-        _system._thread->interruptAndJoin(&_state._workerMonitor);
+        _system._thread->interruptAndJoin(_state._workerLock, _state._workerCond);
         _system._thread.reset(0);
     }
 }
@@ -191,11 +194,11 @@ StorageBucketDBInitializer::onClose()
 void
 StorageBucketDBInitializer::run(framework::ThreadHandle& thread)
 {
-    vespalib::MonitorGuard monitor(_state._workerMonitor);
+    std::unique_lock<std::mutex> guard(_state._workerLock);
     while (!thread.interrupted() && !_state._doneInitializing) {
         std::list<api::StorageMessage::SP> replies;
         {
-            vespalib::LockGuard lock(_state._replyLock);
+            std::lock_guard<std::mutex> replyGuard(_state._replyLock);
             _state._replies.swap(replies);
         }
         for (std::list<api::StorageMessage::SP>::iterator it = replies.begin();
@@ -218,7 +221,7 @@ StorageBucketDBInitializer::run(framework::ThreadHandle& thread)
             updateInitProgress();
         }
         if (replies.empty()) {
-            monitor.wait(10);
+            _state._workerCond.wait_for(guard, 10ms);
             thread.registerTick(framework::WAIT_CYCLE);
         } else {
             thread.registerTick(framework::PROCESS_CYCLE);
@@ -258,7 +261,7 @@ void
 StorageBucketDBInitializer::reportHtmlStatus(
         std::ostream& out, const framework::HttpUrlPath&) const
 {
-    vespalib::Monitor monitor(_state._workerMonitor);
+    std::lock_guard<std::mutex> guard(_state._workerLock);
     out << "\n  <h2>Config</h2>\n"
         << "    <table>\n"
         << "      <tr><td>Max pending info reads per disk</td><td>"
@@ -583,7 +586,7 @@ StorageBucketDBInitializer::onInternalReply(
         case ReadBucketInfoReply::ID:
         case InternalBucketJoinReply::ID:
         {
-            vespalib::LockGuard lock(_state._replyLock);
+            std::lock_guard<std::mutex> guard(_state._replyLock);
             _state._replies.push_back(reply);
             return true;
         }
diff --git a/storage/src/vespa/storage/bucketdb/storagebucketdbinitializer.h b/storage/src/vespa/storage/bucketdb/storagebucketdbinitializer.h
index 57b95e14f48..642fe43ad8a 100644
--- a/storage/src/vespa/storage/bucketdb/storagebucketdbinitializer.h
+++ b/storage/src/vespa/storage/bucketdb/storagebucketdbinitializer.h
@@ -47,11 +47,12 @@
 #include <vespa/storageframework/generic/status/htmlstatusreporter.h>
 #include <vespa/storageframework/generic/clock/timer.h>
 #include <vespa/vespalib/stllike/hash_map.h>
-#include <vespa/vespalib/util/sync.h>
 #include <vespa/vdslib/state/nodestate.h>
 #include <vespa/config/subscription/configuri.h>
 #include <list>
 #include <unordered_map>
+#include <mutex>
+#include <condition_variable>
 
 namespace storage {
 
@@ -120,9 +121,10 @@ class StorageBucketDBInitializer : public StorageLink,
             // This lock is held while the worker thread is working, such that
             // status retrieval can lock it. Listing part only grabs it when
             // needed to supporting listing in multiple threads
-        vespalib::Monitor _workerMonitor;
+        mutable std::mutex       _workerLock;
+        std::condition_variable  _workerCond;
             // This lock protects the reply list.
-        vespalib::Monitor _replyLock;
+        std::mutex               _replyLock;
 
         GlobalState();
         ~GlobalState();
diff --git a/storageframework/src/vespa/storageframework/generic/thread/thread.cpp b/storageframework/src/vespa/storageframework/generic/thread/thread.cpp
index 1a88b4d2044..5ed3f7dc5e6 100644
--- a/storageframework/src/vespa/storageframework/generic/thread/thread.cpp
+++ b/storageframework/src/vespa/storageframework/generic/thread/thread.cpp
@@ -17,5 +17,16 @@ Thread::interruptAndJoin(vespalib::Monitor* m)
     join();
 }
 
+void
+Thread::interruptAndJoin(std::mutex &m, std::condition_variable &cv)
+{
+    interrupt();
+    {
+        std::lock_guard<std::mutex> guard(m);
+        cv.notify_all();
+    }
+    join();
+}
+
 } // framework
 } // storage
diff --git a/storageframework/src/vespa/storageframework/generic/thread/thread.h b/storageframework/src/vespa/storageframework/generic/thread/thread.h
index ceeba79ebe2..72054ff725a 100644
--- a/storageframework/src/vespa/storageframework/generic/thread/thread.h
+++ b/storageframework/src/vespa/storageframework/generic/thread/thread.h
@@ -14,6 +14,8 @@
 
 #include "runnable.h"
 #include <vespa/vespalib/stllike/string.h>
+#include <mutex>
+#include <condition_variable>
 
 namespace vespalib {
     class Monitor;
@@ -58,6 +60,8 @@ public:
      * through a monitor after the signalling face.
      */
     void interruptAndJoin(vespalib::Monitor* m);
+
+    void interruptAndJoin(std::mutex &m, std::condition_variable &cv);
 };
 
 }