Separate support for SyncWrites and SyncReplication

Consumers should be able to stream Prepares, Commits, and Aborts
if they wish without acking.

Change-Id: I7eb3df1a214cef5acccaa6ab5284a9f1d311b23e
Reviewed-on: http://review.couchbase.org/113037
Reviewed-by: Dave Rigby <[email protected]>
Tested-by: Build Bot <[email protected]>

Author: BenHuddleston

engines/ep/src/connhandler.h

@@ -17,6 +17,7 @@
 
 #pragma once
 
+#include "dcp/dcp-types.h"
 #include "utility.h"
 
 #include <memcached/dcp.h>
@@ -266,7 +267,23 @@ class ConnHandler {
         // Empty
     }
 
+    /**
+     * Does the Connection support SyncReplication (Acking prepares)?
+     */
     bool isSyncReplicationEnabled() const {
+        return supportsSyncReplication.load() ==
+               SyncReplication::SyncReplication;
+    }
+
+    /**
+     * Does the Connection support SyncWrites (sending and receiving Prepares,
+     * Commits, and Aborts)?
+     */
+    bool isSyncWritesEnabled() const {
+        return supportsSyncReplication.load() != SyncReplication::No;
+    }
+
+    SyncReplication getSyncReplSupport() const {
         return supportsSyncReplication.load();
     }
 
@@ -321,8 +338,12 @@ class ConnHandler {
     //! The bucketLogger for this connection
     std::shared_ptr<BucketLogger> logger;
 
-    /// Does this DCP Connection support Synchronous Replication
-    std::atomic<bool> supportsSyncReplication{false};
+    /**
+     * Does this DCP Connection support Synchronous Replication (i.e. acking
+     * Prepares). A connection should support SyncWrites to support
+     * SyncReplication.
+     */
+    std::atomic<SyncReplication> supportsSyncReplication{SyncReplication::No};
 
 private:
 

engines/ep/src/dcp/active_stream.cc

@@ -78,8 +78,7 @@ ActiveStream::ActiveStream(EventuallyPersistentEngine* e,
       forceValueCompression(p->isForceValueCompressionEnabled()
                                     ? ForceValueCompression::Yes
                                     : ForceValueCompression::No),
-      syncReplication(p->isSyncReplicationEnabled() ? SyncReplication::Yes
-                                                    : SyncReplication::No),
+      syncReplication(p->getSyncReplSupport()),
       filter(std::move(f)),
       sid(filter.getStreamId()) {
     const char* type = "";
@@ -949,7 +948,7 @@ std::unique_ptr<DcpResponse> ActiveStream::makeResponseFromItem(
             start_seqno_ < static_cast<uint64_t>(item->getPrepareSeqno());
 
     if ((item->getOperation() == queue_op::commit_sync_write) &&
-        (syncReplication == SyncReplication::Yes) &&
+        (supportSyncWrites()) &&
         (!prepareSeqnoGTRequestedStart ||
          sendCommitSyncWriteAs == SendCommitSyncWriteAs::Commit)) {
         return std::make_unique<CommitSyncWrite>(opaque_,
@@ -1101,7 +1100,7 @@ void ActiveStream::processItems(OutstandingItemsResult& outstandingItemsResult,
 }
 
 bool ActiveStream::shouldProcessItem(const Item& item) {
-    if (!item.shouldReplicate(syncReplication == SyncReplication::Yes)) {
+    if (!item.shouldReplicate(supportSyncWrites())) {
         return false;
     }
 

engines/ep/src/dcp/active_stream.h

@@ -282,7 +282,11 @@ class ActiveStream : public Stream,
     void nextCheckpointItemTask(const LockHolder& streamMutex);
 
     bool supportSyncReplication() const {
-        return syncReplication == SyncReplication::Yes;
+        return syncReplication == SyncReplication::SyncReplication;
+    }
+
+    bool supportSyncWrites() const {
+        return syncReplication != SyncReplication::No;
     }
 
     /* Indicates that a backfill has been scheduled and has not yet completed.
@@ -467,7 +471,13 @@ class ActiveStream : public Stream,
     /// Should items be forcefully compressed on this stream?
     const ForceValueCompression forceValueCompression;
 
-    /// Does this stream support synchronous replication?
+    /// Does this stream support synchronous replication (i.e. acking Prepares)?
+    /**
+     * What level of SyncReplication does this stream Support:
+     *  - None
+     *  - SyncWrites: Sending Prepares/Commits/Aborts
+     *  - SyncReplication: SyncWrites + Acking Prepares
+     */
     const SyncReplication syncReplication;
 
     /**

engines/ep/src/dcp/consumer.cc

@@ -975,8 +975,9 @@ bool DcpConsumer::handleResponse(const protocol_binary_response_header* resp) {
         // the Sync Replication negotiation-key.
         if (resp->response.getOpaque() == syncReplNegotiation.opaque) {
             syncReplNegotiation.state = SyncReplNegotiation::State::Completed;
-            supportsSyncReplication.store(resp->response.getStatus() ==
-                                          cb::mcbp::Status::Success);
+            if (resp->response.getStatus() == cb::mcbp::Status::Success) {
+                supportsSyncReplication.store(SyncReplication::SyncReplication);
+            }
         }
         return true;
     } else if (opcode == cb::mcbp::ClientOpcode::GetErrorMap) {
@@ -1501,14 +1502,14 @@ ENGINE_ERROR_CODE DcpConsumer::enableExpiryOpcode(
 
 ENGINE_ERROR_CODE DcpConsumer::enableSynchronousReplication(
         dcp_message_producers* producers) {
-    // enable_synchronous_replication and consumer_name are separated into two
+    // enable_sync_writes and consumer_name are separated into two
     // different variables as in the future non-replication consumers may wish
     // to stream prepares and commits.
     switch (syncReplNegotiation.state) {
     case SyncReplNegotiation::State::PendingRequest: {
         uint32_t opaque = ++opaqueCounter;
-        ENGINE_ERROR_CODE ret = producers->control(
-                opaque, "enable_synchronous_replication", "true");
+        ENGINE_ERROR_CODE ret =
+                producers->control(opaque, "enable_sync_writes", "true");
         syncReplNegotiation.state = SyncReplNegotiation::State::PendingResponse;
         syncReplNegotiation.opaque = opaque;
         return ret;

engines/ep/src/dcp/dcp-types.h

@@ -155,5 +155,8 @@ enum class MultipleStreamRequests : bool {
     No
 };
 
-/// Does the stream support synchronous replication?
-enum class SyncReplication : bool { Yes, No };
+/** Does the stream support synchronous replication (i.e. acking Prepares)?
+ * A Stream may also support just SyncWrites (receiving Prepares and Commits
+ * without acking).
+ */
+enum class SyncReplication : char { No, SyncWrites, SyncReplication };

engines/ep/src/dcp/producer.cc

@@ -979,13 +979,19 @@ ENGINE_ERROR_CODE DcpProducer::control(uint32_t opaque,
         }
         multipleStreamRequests = MultipleStreamRequests::Yes;
         return ENGINE_SUCCESS;
-    } else if (key == "enable_synchronous_replication") {
+    } else if (key == "enable_sync_writes") {
         if (valueStr == "true") {
-            supportsSyncReplication = true;
+            supportsSyncReplication = SyncReplication::SyncWrites;
+            if (!consumerName.empty()) {
+                supportsSyncReplication = SyncReplication::SyncReplication;
+            }
             return ENGINE_SUCCESS;
         }
-    } else if (key == "consumer_name") {
+    } else if (key == "consumer_name" && !valueStr.empty()) {
         consumerName = valueStr;
+        if (supportsSyncReplication == SyncReplication::SyncWrites) {
+            supportsSyncReplication = SyncReplication::SyncReplication;
+        }
         return ENGINE_SUCCESS;
     }
 
@@ -1269,6 +1275,7 @@ void DcpProducer::addStats(const AddStatFn& add_stat, const void* c) {
             add_stat,
             c);
     addStat("synchronous_replication", isSyncReplicationEnabled(), add_stat, c);
+    addStat("synchronous_writes", isSyncWritesEnabled(), add_stat, c);
 
     // Possible that the producer has had its streams closed and hence doesn't
     // have a backfill manager anymore.

engines/ep/tests/ep_testsuite_dcp.cc

@@ -3654,7 +3654,7 @@ static uint32_t add_stream_for_consumer(EngineIface* h,
     dcpStepAndExpectControlMsg("supports_hifi_MFU"s);
     dcpStepAndExpectControlMsg("send_stream_end_on_client_close_stream"s);
     dcpStepAndExpectControlMsg("enable_expiry_opcode"s);
-    dcpStepAndExpectControlMsg("enable_synchronous_replication"s);
+    dcpStepAndExpectControlMsg("enable_sync_writes"s);
     simulateProdRespAtSyncReplNegotiation(h, cookie, producers);
     dcpStepAndExpectControlMsg("consumer_name"s);
 
@@ -4345,7 +4345,7 @@ static void drainDcpControl(EngineIface* engine,
     do {
         dcp_step(engine, cookie, producers);
         // The Sync Repl negotiation introduces a blocking step
-        if (producers.last_key == "enable_synchronous_replication") {
+        if (producers.last_key == "enable_sync_writes") {
             simulateProdRespAtSyncReplNegotiation(engine, cookie, producers);
         }
     } while (producers.last_op == cb::mcbp::ClientOpcode::DcpControl);

engines/ep/tests/mock/mock_dcp_consumer.h

@@ -113,14 +113,14 @@ class MockDcpConsumer: public DcpConsumer {
      * Used for simulating a successful Consumer-Producer Sync Repl handshake.
      */
     void enableSyncReplication() {
-        supportsSyncReplication = true;
+        supportsSyncReplication = SyncReplication::SyncReplication;
     }
 
     /**
      * Disable SyncRepl for testing
      */
     void disableSyncReplication() {
-        supportsSyncReplication = false;
+        supportsSyncReplication = SyncReplication::No;
     }
 
     /**

engines/ep/tests/mock/mock_dcp_producer.h

@@ -84,9 +84,10 @@ class MockDcpProducer : public DcpProducer {
         return enableExpiryOpcode;
     }
 
-    void setSyncReplication(bool value) {
+    void setSyncReplication(SyncReplication value) {
         supportsSyncReplication = value;
     }
+
     /**
      * Create the ActiveStreamCheckpointProcessorTask and assign to
      * checkpointCreator->task

engines/ep/tests/module_tests/dcp_durability_stream_test.cc

@@ -50,7 +50,7 @@ void DurabilityActiveStreamTest::setUp(bool startCheckpointProcessorTask) {
             {{"topology", nlohmann::json::array({{active, replica}})}});
 
     // Enable SyncReplication and flow-control (Producer BufferLog)
-    setupProducer({{"enable_synchronous_replication", "true"},
+    setupProducer({{"enable_sync_writes", "true"},
                    {"connection_buffer_size", "52428800"},
                    {"consumer_name", "test_consumer"}},
                   startCheckpointProcessorTask);