Merge remote-tracking branch 'ep-engine/watson' into kv_engine/watson_ep

* ep-engine/watson:
  MB-29483: Disable DCP cursor dropping
  MB-29287: Add dcp_reflection_test
  MB-29287: Move SynchronousEPEngine building to seperate method
  MB-29287: Add string / streaming methods for DCP responses

Change-Id: Ic4d36779ef5281cdc4cf2a3bc4275e291a5ad2c5

Author: daverigby

CMakeLists.txt

@@ -189,6 +189,7 @@ ADD_EXECUTABLE(ep-engine_ep_unit_tests
   tests/module_tests/checkpoint_test.cc
   tests/module_tests/defragmenter_test.cc
   tests/module_tests/ep_unit_tests_main.cc
+  tests/module_tests/dcp_reflection_test.cc
   tests/module_tests/dcp_test.cc
   tests/module_tests/evp_engine_test.cc
   tests/module_tests/evp_store_rollback_test.cc

src/dcp/consumer.cc

@@ -123,7 +123,8 @@ DcpConsumer::DcpConsumer(EventuallyPersistentEngine &engine, const void *cookie,
     pendingSetPriority = true;
     pendingEnableExtMetaData = true;
     pendingEnableValueCompression = config.isDcpValueCompressionEnabled();
-    pendingSupportCursorDropping = true;
+    // MB-29369: Don't request cursor dropping.
+    pendingSupportCursorDropping = false;
 
     ExTask task = new Processor(&engine, this, 1);
     processorTaskId = ExecutorPool::get()->schedule(task, NONIO_TASK_IDX);

src/dcp/producer.cc

@@ -172,15 +172,10 @@ DcpProducer::DcpProducer(EventuallyPersistentEngine &e, const void *cookie,
     enableExtMetaData = false;
     enableValueCompression = false;
 
-    // Cursor dropping is disabled for replication connections by default,
-    // but will be enabled through a control message to support backward
-    // compatibility. For all other type of DCP connections, cursor dropping
-    // will be enabled by default.
-    if (name.find("replication") < name.length()) {
-        supportsCursorDropping = false;
-    } else {
-        supportsCursorDropping = true;
-    }
+    // MB-29369: Cursor dropping is currently disabled for all
+    // connections due to race condition which can result in skipping
+    // mutations.
+    supportsCursorDropping = false;
 
     backfillMgr.reset(new BackfillManager(&engine_));
 
@@ -584,11 +579,8 @@ ENGINE_ERROR_CODE DcpProducer::control(uint32_t opaque, const void* key,
         }
         return ENGINE_SUCCESS;
     } else if (strncmp(param, "supports_cursor_dropping", nkey) == 0) {
-        if (valueStr == "true") {
-            supportsCursorDropping = true;
-        } else {
-            supportsCursorDropping = false;
-        }
+        // MB-29369: Cursor dropping currently disabled. Ignore requests
+        // to enable.
         return ENGINE_SUCCESS;
     } else if (strncmp(param, "set_noop_interval", nkey) == 0) {
         if (parseUint32(valueStr.c_str(), &noopCtx.noopInterval)) {

src/dcp/response.cc

@@ -35,3 +35,34 @@ const uint32_t SetVBucketState::baseMsgBytes = 25;
 const uint32_t SnapshotMarker::baseMsgBytes = 44;
 const uint32_t MutationResponse::mutationBaseMsgBytes = 55;
 const uint32_t MutationResponse::deletionBaseMsgBytes = 42;
+
+
+std::string to_string(dcp_event_t event) {
+    switch (event) {
+        case DCP_MUTATION:
+            return "DCP_MUTATION";
+        case DCP_DELETION:
+            return "DCP_DELETION";
+        case DCP_EXPIRATION:
+            return "DCP_EXPIRATION";
+        case DCP_FLUSH:
+            return "DCP_FLUSH";
+        case DCP_SET_VBUCKET:
+            return "DCP_SET_VBUCKET";
+        case DCP_STREAM_REQ:
+            return "DCP_STREAM_REQ";
+        case DCP_STREAM_END:
+            return "DCP_STREAM_END";
+        case DCP_SNAPSHOT_MARKER:
+            return "DCP_SNAPSHOT_MARKER";
+        case DCP_ADD_STREAM:
+            return "DCP_ADD_STREAM";
+    }
+    return "<invalid_dcp_event_t>(" + std::to_string(int(event)) + ")";
+}
+
+std::ostream& operator<<(std::ostream& os, const DcpResponse& r) {
+    os << "DcpResponse[" << &r << "] with"
+       << " event:" << to_string(r.getEvent());
+    return os;
+}

src/dcp/response.h

@@ -35,6 +35,7 @@ enum dcp_event_t {
     DCP_ADD_STREAM
 };
 
+std::string to_string(dcp_event_t event);
 
 enum dcp_marker_flag_t {
     MARKER_FLAG_MEMORY = 0x01,
@@ -59,7 +60,7 @@ class DcpResponse {
         return opaque_;
     }
 
-    dcp_event_t getEvent() {
+    dcp_event_t getEvent() const {
         return event_;
     }
 
@@ -95,6 +96,8 @@ class DcpResponse {
     dcp_event_t event_;
 };
 
+std::ostream& operator<<(std::ostream& os, const DcpResponse& r);
+
 class StreamRequest : public DcpResponse {
 public:
     StreamRequest(uint16_t vbucket, uint32_t opaque, uint32_t flags,

src/dcp/stream.cc

@@ -70,6 +70,10 @@ bool Stream::isInMemory() const {
     return state_.load() == STREAM_IN_MEMORY;
 }
 
+bool Stream::isInTakeoverSend() const {
+    return state_.load() == STREAM_TAKEOVER_SEND;
+}
+
 void Stream::clear_UNLOCKED() {
     while (!readyQ.empty()) {
         DcpResponse* resp = readyQ.front();

src/dcp/stream.h

@@ -139,6 +139,8 @@ class Stream : public RCValue {
     /// @Returns true if state_ is InMemory
     bool isInMemory() const;
 
+    bool isInTakeoverSend() const;
+
     void clear() {
         LockHolder lh(streamMutex);
         clear_UNLOCKED();

src/ep_engine.h

@@ -1025,6 +1025,7 @@ class EventuallyPersistentEngine : public ENGINE_HANDLE_V1 {
                               int& keyOffset);
 
     SERVER_HANDLE_V1 *serverApi;
+    // epstore is an owning pointer.
     EventuallyPersistentStore *epstore;
     WorkLoadPolicy *workload;
     bucket_priority_t workloadPriority;

tests/ep_testsuite_dcp.cc

@@ -2321,6 +2321,9 @@ static test_result test_dcp_agg_stats(ENGINE_HANDLE *h, ENGINE_HANDLE_V1 *h1) {
 
 static test_result test_dcp_cursor_dropping(ENGINE_HANDLE *h,
                                             ENGINE_HANDLE_V1 *h1) {
+    // MB-29369: Cursor dropping currently disabled
+    return SKIPPED;
+
     /* Initially write a few items */
     int num_items = 25;
     const int cursor_dropping_mem_thres_perc = 90;
@@ -2407,6 +2410,9 @@ static test_result test_dcp_cursor_dropping(ENGINE_HANDLE *h,
 
 static test_result test_dcp_cursor_dropping_backfill(ENGINE_HANDLE *h,
                                                      ENGINE_HANDLE_V1 *h1) {
+    // MB-29369: Cursor dropping currently disabled
+    return SKIPPED;
+
     /* Initially write a few items */
     int num_items = 50;
     const int cursor_dropping_mem_thres_perc = 90;
@@ -2907,10 +2913,13 @@ static uint32_t add_stream_for_consumer(ENGINE_HANDLE *h, ENGINE_HANDLE_V1 *h1,
         cb_assert(dcp_last_opaque != opaque);
     }
 
+    // MB-29369: Cursor dropping currently disabled.
+#if 0
     dcp_step(h, h1, cookie);
     cb_assert(dcp_last_op == PROTOCOL_BINARY_CMD_DCP_CONTROL);
     cb_assert(dcp_last_key.compare("supports_cursor_dropping") == 0);
     cb_assert(dcp_last_opaque != opaque);
+#endif
 
     checkeq(ENGINE_SUCCESS,
             h1->dcp.add_stream(h, cookie, opaque, vbucket, flags),