MB-47318: [BP] Add PassiveStream always buffered mode

Add a DcpControl that can force consumer to always process
messages on the AUXIO task.

Change-Id: Ie96930982efc43ee1ec56e997ca29c658ae0b3e6
Reviewed-on: https://review.couchbase.org/c/kv_engine/+/165016
Tested-by: Build Bot <[email protected]>
Reviewed-by: Dave Rigby <[email protected]>
Well-Formed: Restriction Checker

Author: jimwwalker

engines/ep/src/dcp/consumer.cc

@@ -1907,6 +1907,15 @@ ENGINE_ERROR_CODE DcpConsumer::control(uint32_t opaque,
                 setFlowControlBufSize(result);
                 return ENGINE_SUCCESS;
             }
+        } else if (key == "always_buffer_operations") {
+            alwaysBufferOperations = value == "true";
+            // Warn about this because only tests should be here
+            logger->warn(
+                    "always_buffer_operations:{} results in "
+                    "alwaysBufferOperations:{}",
+                    value,
+                    alwaysBufferOperations ? "true" : "false");
+            return ENGINE_SUCCESS;
         }
 
         logger->warn("Invalid ctrl parameter {} {}", key, value);
@@ -1915,3 +1924,7 @@ ENGINE_ERROR_CODE DcpConsumer::control(uint32_t opaque,
 
     return ConnHandler::control(opaque, key, value);
 }
+
+bool DcpConsumer::shouldBufferOperations() const {
+    return alwaysBufferOperations;
+}

engines/ep/src/dcp/consumer.h

@@ -340,6 +340,12 @@ class DcpConsumer : public ConnHandler,
         return allowSanitizeValueInDeletion.load();
     }
 
+    /**
+     * Force streams to buffer?
+     * @return true if streams should always buffer operations
+     */
+    bool shouldBufferOperations() const;
+
 protected:
     /**
      * Records when the consumer last received a message from producer.
@@ -651,6 +657,8 @@ class DcpConsumer : public ConnHandler,
      */
     std::atomic_bool allowSanitizeValueInDeletion;
 
+    bool alwaysBufferOperations{false};
+
     static const std::string noopCtrlMsg;
     static const std::string noopIntervalCtrlMsg;
     static const std::string connBufferCtrlMsg;

engines/ep/src/dcp/passive_stream.cc

@@ -64,7 +64,8 @@ PassiveStream::PassiveStream(EventuallyPersistentEngine* e,
       cur_snapshot_type(Snapshot::None),
       cur_snapshot_ack(false),
       cur_snapshot_prepare(false),
-      vb_manifest_uid(vb_manifest_uid) {
+      vb_manifest_uid(vb_manifest_uid),
+      alwaysBufferOperations(c->shouldBufferOperations()) {
     LockHolder lh(streamMutex);
     streamRequest_UNLOCKED(vb_uuid);
     itemsReady.store(true);
@@ -327,7 +328,7 @@ ENGINE_ERROR_CODE PassiveStream::messageReceived(
             vb_);
         return ENGINE_DISCONNECT;
     case ReplicationThrottle::Status::Process:
-        if (buffer.empty()) {
+        if (buffer.empty() && !alwaysBufferOperations) {
             /* Process the response here itself rather than buffering it */
             ENGINE_ERROR_CODE ret = ENGINE_SUCCESS;
             switch (dcpResponse->getEvent()) {

engines/ep/src/dcp/passive_stream.h

@@ -339,4 +339,6 @@ class PassiveStream : public Stream {
     // Flag indicating if the most recent call to processMessage
     // backed off due to ENOMEM. Only used for limiting logging
     std::atomic<bool> isNoMemory{false};
+
+    bool alwaysBufferOperations{false};
 };