Reduce code visibility and use rwlock.

11 files changed, 39 insertions, 42 deletions

diff --git a/searchcore/src/vespa/searchcore/proton/common/feedtoken.cpp b/searchcore/src/vespa/searchcore/proton/common/feedtoken.cpp
index d61eaa2c6ab..251989cd263 100644
--- a/searchcore/src/vespa/searchcore/proton/common/feedtoken.cpp
+++ b/searchcore/src/vespa/searchcore/proton/common/feedtoken.cpp
@@ -1,6 +1,7 @@
 // Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 
 #include "feedtoken.h"
+#include <vespa/persistence/spi/result.h>
 
 namespace proton::feedtoken {
 
@@ -27,6 +28,12 @@ State::ack()
 }
 
 void
+State::setResult(ResultUP result, bool documentWasFound) {
+    _documentWasFound = documentWasFound;
+    _result = std::move(result);
+}
+
+void
 State::fail()
 {
     bool alreadySent = _alreadySent.exchange(true);
diff --git a/searchcore/src/vespa/searchcore/proton/common/feedtoken.h b/searchcore/src/vespa/searchcore/proton/common/feedtoken.h
index cf6c1f5a7e9..411dd256ccd 100644
--- a/searchcore/src/vespa/searchcore/proton/common/feedtoken.h
+++ b/searchcore/src/vespa/searchcore/proton/common/feedtoken.h
@@ -1,10 +1,10 @@
 // Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 #pragma once
 
-#include <vespa/persistence/spi/persistenceprovider.h>
 #include <vespa/searchlib/common/idestructorcallback.h>
 #include <atomic>
 
+namespace storage::spi { class Result; }
 namespace proton {
 
 typedef std::unique_ptr<storage::spi::Result> ResultUP;
@@ -33,10 +33,7 @@ public:
     State(ITransport & transport);
     ~State() override;
     void fail();
-    void setResult(ResultUP result, bool documentWasFound) {
-        _documentWasFound = documentWasFound;
-        _result = std::move(result);
-    }
+    void setResult(ResultUP result, bool documentWasFound);
     const storage::spi::Result &getResult() { return *_result; }
 protected:
     void ack();
diff --git a/searchcore/src/vespa/searchcore/proton/persistenceengine/resulthandler.h b/searchcore/src/vespa/searchcore/proton/persistenceengine/resulthandler.h
index 60ff696cb03..6712a35bd65 100644
--- a/searchcore/src/vespa/searchcore/proton/persistenceengine/resulthandler.h
+++ b/searchcore/src/vespa/searchcore/proton/persistenceengine/resulthandler.h
@@ -8,7 +8,7 @@ namespace proton {
 template <typename ResultType>
 class IResultHandler {
 public:
-    virtual ~IResultHandler() { }
+    virtual ~IResultHandler() = default;
     virtual void handle(const ResultType &result) = 0;
 };
 
diff --git a/searchcore/src/vespa/searchcore/proton/persistenceengine/transport_latch.h b/searchcore/src/vespa/searchcore/proton/persistenceengine/transport_latch.h
index da4e19d3584..0e6b5d23053 100644
--- a/searchcore/src/vespa/searchcore/proton/persistenceengine/transport_latch.h
+++ b/searchcore/src/vespa/searchcore/proton/persistenceengine/transport_latch.h
@@ -2,6 +2,7 @@
 #pragma once
 
 #include <vespa/persistence/spi/result.h>
+#include <vespa/persistence/spi/operationcomplete.h>
 #include <vespa/searchcore/proton/common/feedtoken.h>
 #include <vespa/vespalib/util/sequence.h>
 #include <vespa/vespalib/util/count_down_latch.h>
diff --git a/searchcore/src/vespa/searchcore/proton/server/combiningfeedview.h b/searchcore/src/vespa/searchcore/proton/server/combiningfeedview.h
index 5625d41ccdc..079b8699ce4 100644
--- a/searchcore/src/vespa/searchcore/proton/server/combiningfeedview.h
+++ b/searchcore/src/vespa/searchcore/proton/server/combiningfeedview.h
@@ -6,6 +6,8 @@
 #include "replaypacketdispatcher.h"
 #include "ibucketstatecalculator.h"
 #include <vespa/searchcore/proton/common/feedtoken.h>
+#include <vespa/document/bucket/bucketspace.h>
+#include <vespa/document/base/globalid.h>
 #include <vespa/searchlib/common/serialnum.h>
 
 namespace proton {
diff --git a/searchcore/src/vespa/searchcore/proton/server/feedhandler.cpp b/searchcore/src/vespa/searchcore/proton/server/feedhandler.cpp
index 684bf0125ac..275aa8f14fb 100644
--- a/searchcore/src/vespa/searchcore/proton/server/feedhandler.cpp
+++ b/searchcore/src/vespa/searchcore/proton/server/feedhandler.cpp
@@ -11,6 +11,7 @@
 #include <vespa/document/datatype/documenttype.h>
 #include <vespa/document/repo/documenttyperepo.h>
 #include <vespa/document/update/documentupdate.h>
+#include <vespa/document/fieldvalue/document.h>
 #include <vespa/searchcore/proton/bucketdb/ibucketdbhandler.h>
 #include <vespa/searchcore/proton/persistenceengine/i_resource_write_filter.h>
 #include <vespa/searchcore/proton/persistenceengine/transport_latch.h>
@@ -84,7 +85,7 @@ void
 FeedHandler::doHandleOperation(FeedToken token, FeedOperation::UP op)
 {
     assert(_writeService.master().isCurrentThread());
-    std::lock_guard<std::mutex> guard(_feedLock);
+    ReadGuard guard(_feedLock);
     _feedState->handleOperation(std::move(token), std::move(op));
 }
 
@@ -296,34 +297,22 @@ FeedHandler::considerDelayedPrune()
 }
 
 
-FeedState::SP
+std::shared_ptr<FeedState>
 FeedHandler::getFeedState() const
 {
-    FeedState::SP state;
-    {
-        std::lock_guard<std::mutex> guard(_feedLock);
-        state = _feedState;
-    }
-    return state;
-}
-
-
-void
-FeedHandler::changeFeedState(FeedState::SP newState)
-{
-    std::lock_guard<std::mutex> guard(_feedLock);
-    changeFeedState(std::move(newState), guard);
+    ReadGuard guard(_feedLock);
+    return _feedState;
 }
 
 
 void
-FeedHandler::changeFeedState(FeedState::SP newState, const std::lock_guard<std::mutex> &)
+FeedHandler::changeFeedState(FeedStateSP newState)
 {
+    WriteGuard guard(_feedLock);
     LOG(debug, "Change feed state from '%s' -> '%s'", _feedState->getName().c_str(), newState->getName().c_str());
     _feedState = std::move(newState);
 }
 
-
 FeedHandler::FeedHandler(IThreadingService &writeService,
                          const vespalib::string &tlsSpec,
                          const DocTypeName &docTypeName,
@@ -630,7 +619,7 @@ FeedHandler::receive(const Packet &packet)
     // Called directly when replaying transaction log
     // (by fnet thread).  Called via DocumentDB::recoverPacket() when
     // recovering from another node.
-    FeedState::SP state = getFeedState();
+    FeedStateSP state = getFeedState();
     auto wrap = make_shared<PacketWrapper>(packet, _tlsReplayProgress.get());
     state->receive(wrap, _writeService.master());
     wrap->gate.await();
diff --git a/searchcore/src/vespa/searchcore/proton/server/feedhandler.h b/searchcore/src/vespa/searchcore/proton/server/feedhandler.h
index 2e6b2616118..e9c54a121a7 100644
--- a/searchcore/src/vespa/searchcore/proton/server/feedhandler.h
+++ b/searchcore/src/vespa/searchcore/proton/server/feedhandler.h
@@ -13,7 +13,7 @@
 #include <vespa/searchcore/proton/common/doctypename.h>
 #include <vespa/searchcore/proton/common/feedtoken.h>
 #include <vespa/searchlib/transactionlog/translogclient.h>
-#include <mutex>
+#include <shared_mutex>
 
 namespace searchcorespi { namespace index { struct IThreadingService; } }
 
@@ -56,6 +56,10 @@ private:
     typedef document::BucketId              BucketId;
     using FeedStateSP = std::shared_ptr<FeedState>;
     using FeedOperationUP = std::unique_ptr<FeedOperation>;
+    using ReadGuard = std::shared_lock<std::shared_mutex>;
+    using WriteGuard = std::unique_lock<std::shared_mutex>;
+    using IThreadingService = searchcorespi::index::IThreadingService;
+
 
     class TlsMgrWriter : public TlsWriter {
         TransactionLogManager &_tls_mgr;
@@ -70,7 +74,6 @@ private:
         bool erase(SerialNum oldest_to_keep) override;
         SerialNum sync(SerialNum syncTo) override;
     };
-    typedef searchcorespi::index::IThreadingService IThreadingService;
 
     IThreadingService                     &_writeService;
     DocTypeName                            _docTypeName;
@@ -85,7 +88,7 @@ private:
     SerialNum                              _serialNum;
     SerialNum                              _prunedSerialNum;
     bool                                   _delayedPrune;
-    mutable std::mutex                     _feedLock;
+    mutable std::shared_mutex              _feedLock;
     FeedStateSP                            _feedState;
     // used by master write thread tasks
     IFeedView                             *_activeFeedView;
@@ -132,7 +135,6 @@ private:
 
     FeedStateSP getFeedState() const;
     void changeFeedState(FeedStateSP newState);
-    void changeFeedState(FeedStateSP newState, const std::lock_guard<std::mutex> &feedGuard);
 public:
     FeedHandler(const FeedHandler &) = delete;
     FeedHandler & operator = (const FeedHandler &) = delete;
diff --git a/searchcore/src/vespa/searchcore/proton/server/feedstate.cpp b/searchcore/src/vespa/searchcore/proton/server/feedstate.cpp
index 3628505ed66..be09816b9a1 100644
--- a/searchcore/src/vespa/searchcore/proton/server/feedstate.cpp
+++ b/searchcore/src/vespa/searchcore/proton/server/feedstate.cpp
@@ -4,7 +4,6 @@
 #include <vespa/searchcore/proton/feedoperation/feedoperation.h>
 #include <vespa/vespalib/util/exceptions.h>
 
-using document::BucketId;
 using vespalib::IllegalStateException;
 using vespalib::make_string;
 
diff --git a/searchcore/src/vespa/searchcore/proton/server/feedstate.h b/searchcore/src/vespa/searchcore/proton/server/feedstate.h
index fa0a1702499..6de1d7a4322 100644
--- a/searchcore/src/vespa/searchcore/proton/server/feedstate.h
+++ b/searchcore/src/vespa/searchcore/proton/server/feedstate.h
@@ -2,17 +2,16 @@
 
 #pragma once
 
-#include "packetwrapper.h"
-#include <vespa/document/bucket/bucketid.h>
-#include <vespa/document/fieldvalue/document.h>
 #include <vespa/searchcore/proton/common/feedtoken.h>
-#include <vespa/searchcore/proton/persistenceengine/resulthandler.h>
 #include <vespa/vespalib/stllike/string.h>
-#include <vespa/vespalib/util/executor.h>
 
+namespace vespalib {
+    class Executor;
+}
 namespace proton {
 
 class FeedOperation;
+class PacketWrapper;
 
 /**
  * Class representing the current state of a feed handler.
@@ -26,21 +25,20 @@ private:
 
 protected:
     using FeedOperationUP = std::unique_ptr<FeedOperation>;
+    using PacketWrapperSP = std::shared_ptr<PacketWrapper>;
     void throwExceptionInReceive(const vespalib::string &docType, uint64_t serialRangeFrom,
                                  uint64_t serialRangeTo, size_t packetSize);
     void throwExceptionInHandleOperation(const vespalib::string &docType, const FeedOperation &op);
 
 public:
-    typedef std::shared_ptr<FeedState> SP;
-
     FeedState(Type type) : _type(type) {}
-    virtual ~FeedState() {}
+    virtual ~FeedState() = default;
 
     Type getType() const { return _type; }
     vespalib::string getName() const;
 
     virtual void handleOperation(FeedToken token, FeedOperationUP op) = 0;
-    virtual void receive(const PacketWrapper::SP &wrap, vespalib::Executor &executor) = 0;
+    virtual void receive(const PacketWrapperSP &wrap, vespalib::Executor &executor) = 0;
 };
 
 }  // namespace proton
diff --git a/searchcore/src/vespa/searchcore/proton/server/feedstates.h b/searchcore/src/vespa/searchcore/proton/server/feedstates.h
index 2cf0ee1a4dd..0c2e9109cce 100644
--- a/searchcore/src/vespa/searchcore/proton/server/feedstates.h
+++ b/searchcore/src/vespa/searchcore/proton/server/feedstates.h
@@ -4,6 +4,7 @@
 
 #include "feedhandler.h"
 #include "feedstate.h"
+#include "packetwrapper.h"
 #include "ireplaypackethandler.h"
 #include <vespa/searchcore/proton/common/commit_time_tracker.h>
 
@@ -33,7 +34,7 @@ public:
         throwExceptionInHandleOperation(_doc_type_name, *op);
     }
 
-    void receive(const PacketWrapper::SP &wrap, vespalib::Executor &) override {
+    void receive(const PacketWrapperSP &wrap, vespalib::Executor &) override {
         throwExceptionInReceive(_doc_type_name.c_str(), wrap->packet.range().from(),
                                 wrap->packet.range().to(), wrap->packet.size());
     }
@@ -60,7 +61,7 @@ public:
         throwExceptionInHandleOperation(_doc_type_name, *op);
     }
 
-    void receive(const PacketWrapper::SP &wrap, vespalib::Executor &executor) override;
+    void receive(const PacketWrapperSP &wrap, vespalib::Executor &executor) override;
 };
 
 
@@ -81,7 +82,7 @@ public:
         _handler.performOperation(std::move(token), std::move(op));
     }
 
-    void receive(const PacketWrapper::SP &wrap, vespalib::Executor &) override {
+    void receive(const PacketWrapperSP &wrap, vespalib::Executor &) override {
         throwExceptionInReceive(_handler.getDocTypeName().c_str(), wrap->packet.range().from(),
                                 wrap->packet.range().to(), wrap->packet.size());
     }
diff --git a/searchcore/src/vespa/searchcore/proton/server/visibilityhandler.h b/searchcore/src/vespa/searchcore/proton/server/visibilityhandler.h
index 560b6e75423..c22128685c1 100644
--- a/searchcore/src/vespa/searchcore/proton/server/visibilityhandler.h
+++ b/searchcore/src/vespa/searchcore/proton/server/visibilityhandler.h
@@ -7,6 +7,7 @@
 #include <vespa/searchcore/proton/common/icommitable.h>
 #include <vespa/searchcorespi/index/ithreadingservice.h>
 #include <vespa/vespalib/util/varholder.h>
+#include <vespa/vespalib/util/time.h>
 #include <mutex>
 
 namespace proton {