Merge pull request #3603 from vespa-engine/balder/let-protocol-signal-sequencing-requirements-2

Balder/let protocol signal sequencing requirements 2

7 files changed, 24 insertions, 12 deletions

diff --git a/documentapi/src/vespa/documentapi/messagebus/documentprotocol.h b/documentapi/src/vespa/documentapi/messagebus/documentprotocol.h
index b2d1456fd98..c3417d85197 100644
--- a/documentapi/src/vespa/documentapi/messagebus/documentprotocol.h
+++ b/documentapi/src/vespa/documentapi/messagebus/documentprotocol.h
@@ -24,7 +24,7 @@ class SystemState;
 class IRoutingPolicyFactory;
 class IRoutableFactory;
 
-class DocumentProtocol : public mbus::IProtocol {
+class DocumentProtocol final : public mbus::IProtocol {
 private:
     std::unique_ptr<RoutingPolicyRepository>    _routingPolicyRepository;
     std::unique_ptr<RoutableRepository>         _routableRepository;
@@ -264,8 +264,7 @@ public:
      * @param buf A byte buffer that contains a serialized routable.
      * @return The deserialized routable.
      */
-    mbus::Routable::UP deserialize(uint32_t type,
-                                   document::ByteBuffer &buf) const;
+    mbus::Routable::UP deserialize(uint32_t type, document::ByteBuffer &buf) const;
 
     /**
      * This is a convenient entry to the {@link #merge(RoutingContext,std::set)} method by way of a routing
@@ -307,7 +306,7 @@ public:
     mbus::IRoutingPolicy::UP createPolicy(const mbus::string &name, const mbus::string &param) const override;
     mbus::Blob encode(const vespalib::Version &version, const mbus::Routable &routable) const override;
     mbus::Routable::UP decode(const vespalib::Version &version, mbus::BlobRef data) const override;
+    bool requireSequencing() const override { return false; }
 };
 
 }
-
diff --git a/messagebus/src/tests/protocolrepository/protocolrepository.cpp b/messagebus/src/tests/protocolrepository/protocolrepository.cpp
index b2454eb272a..b6178449918 100644
--- a/messagebus/src/tests/protocolrepository/protocolrepository.cpp
+++ b/messagebus/src/tests/protocolrepository/protocolrepository.cpp
@@ -48,6 +48,7 @@ public:
         (void)data;
         throw std::exception();
     }
+    bool requireSequencing() const override { return false; }
 };
 
 int
diff --git a/messagebus/src/vespa/messagebus/iprotocol.h b/messagebus/src/vespa/messagebus/iprotocol.h
index 40cfc779c36..8a4129d1976 100644
--- a/messagebus/src/vespa/messagebus/iprotocol.h
+++ b/messagebus/src/vespa/messagebus/iprotocol.h
@@ -52,8 +52,7 @@ public:
      * @param param Ppolicy specific parameter.
      * @return A newly created routing policy.
      */
-    virtual IRoutingPolicy::UP createPolicy(const string &name,
-                                            const string &param) const = 0;
+    virtual IRoutingPolicy::UP createPolicy(const string &name, const string &param) const = 0;
 
     /**
      * Encodes the protocol specific data of a routable into a byte array.
@@ -80,6 +79,9 @@ public:
      * @return The decoded routable.
      */
     virtual Routable::UP decode(const vespalib::Version &version, BlobRef data) const = 0; // throw()
+
+
+    virtual bool requireSequencing() const = 0;
 };
 
 } // namespace mbus
diff --git a/messagebus/src/vespa/messagebus/network/rpcsend.cpp b/messagebus/src/vespa/messagebus/network/rpcsend.cpp
index cc6b7086756..e23f4dc29d9 100644
--- a/messagebus/src/vespa/messagebus/network/rpcsend.cpp
+++ b/messagebus/src/vespa/messagebus/network/rpcsend.cpp
@@ -220,11 +220,19 @@ RPCSend::decode(vespalib::stringref protocolName, const vespalib::Version & vers
 void
 RPCSend::handleReply(Reply::UP reply)
 {
-    doHandleReply(std::move(reply));
+    const IProtocol * protocol = _net->getOwner().getProtocol(reply->getProtocol());
+    if (!protocol || protocol->requireSequencing()) {
+        doHandleReply(protocol, std::move(reply));
+    } else {
+        auto rejected = _net->getExecutor().execute(makeLambdaTask([this, protocol, reply = std::move(reply)]() mutable {
+            doHandleReply(protocol, std::move(reply));
+        }));
+        assert (!rejected);
+    }
 }
 
 void
-RPCSend::doHandleReply(Reply::UP reply) {
+RPCSend::doHandleReply(const IProtocol * protocol, Reply::UP reply) {
     ReplyContext::UP ctx(static_cast<ReplyContext*>(reply->getContext().value.PTR));
     FRT_RPCRequest &req = ctx->getRequest();
     string version = ctx->getVersion().toString();
@@ -234,7 +242,7 @@ RPCSend::doHandleReply(Reply::UP reply) {
     }
     Blob payload(0);
     if (reply->getType() != 0) {
-        payload = _net->getOwner().getProtocol(reply->getProtocol())->encode(ctx->getVersion(), *reply);
+        payload = protocol->encode(ctx->getVersion(), *reply);
         if (payload.size() == 0) {
             reply->addError(Error(ErrorCode::ENCODE_ERROR, "An error occured while encoding the reply, see log."));
         }
diff --git a/messagebus/src/vespa/messagebus/network/rpcsend.h b/messagebus/src/vespa/messagebus/network/rpcsend.h
index ec455aea7bd..11a042b91c0 100644
--- a/messagebus/src/vespa/messagebus/network/rpcsend.h
+++ b/messagebus/src/vespa/messagebus/network/rpcsend.h
@@ -19,6 +19,7 @@ class Error;
 class Route;
 class Message;
 class RPCServiceAddress;
+class IProtocol;
 
 class PayLoadFiller
 {
@@ -84,7 +85,7 @@ public:
 private:
     void doRequest(FRT_RPCRequest *req);
     void doRequestDone(FRT_RPCRequest *req);
-    void doHandleReply(std::unique_ptr<Reply> reply);
+    void doHandleReply(const IProtocol * protocol, std::unique_ptr<Reply> reply);
     void attach(RPCNetwork &net) final override;
     void handleDiscard(Context ctx) final override;
     void sendByHandover(RoutingNode &recipient, const vespalib::Version &version,
diff --git a/messagebus/src/vespa/messagebus/testlib/simpleprotocol.h b/messagebus/src/vespa/messagebus/testlib/simpleprotocol.h
index 8931e1b46f9..09e1ee9febe 100644
--- a/messagebus/src/vespa/messagebus/testlib/simpleprotocol.h
+++ b/messagebus/src/vespa/messagebus/testlib/simpleprotocol.h
@@ -72,6 +72,7 @@ public:
     IRoutingPolicy::UP createPolicy(const string &name, const string &param) const override;
     Blob encode(const vespalib::Version &version, const Routable &routable) const override;
     Routable::UP decode(const vespalib::Version &version, BlobRef data) const override;
+    virtual bool requireSequencing() const override { return false; }
 };
 
 } // namespace mbus
diff --git a/storageapi/src/vespa/storageapi/mbusprot/storageprotocol.h b/storageapi/src/vespa/storageapi/mbusprot/storageprotocol.h
index 699f1c4c239..10289adaf1a 100644
--- a/storageapi/src/vespa/storageapi/mbusprot/storageprotocol.h
+++ b/storageapi/src/vespa/storageapi/mbusprot/storageprotocol.h
@@ -6,7 +6,7 @@
 
 namespace storage::mbusprot {
 
-class StorageProtocol : public mbus::IProtocol
+class StorageProtocol final : public mbus::IProtocol
 {
 public:
     typedef std::shared_ptr<StorageProtocol> SP;
@@ -20,7 +20,7 @@ public:
     mbus::IRoutingPolicy::UP createPolicy(const mbus::string& name, const mbus::string& param) const override;
     mbus::Blob encode(const vespalib::Version&, const mbus::Routable&) const override;
     mbus::Routable::UP decode(const vespalib::Version&, mbus::BlobRef) const override;
-
+    virtual bool requireSequencing() const override { return true; }
 private:
     ProtocolSerialization5_0 _serializer5_0;
     ProtocolSerialization5_1 _serializer5_1;