MB-19732: Record time for all DCP consumer messages

The DCP documentation states that the consumer should see
some sort of message or a No-Op message in a period
equal to twice the noop interval otherwise it should close
its connection.  See documentation/commands/no-op.md in
https://github.com/couchbaselabs/dcp-documentation

This patch changes from checking only the receival of a
no-op message to check for recieving the following messages
- add stream
- close stream
- deletion
- expiration
- flush
- mutation
- set VBucket state
- snapshot Marker
- stream end

Change-Id: Ib2268dba339cbf3701f3c7782ee8256bddc79ba3
Reviewed-on: http://review.couchbase.org/64727
Well-Formed: buildbot <build@couchbase.com>
Tested-by: buildbot <build@couchbase.com>
Reviewed-by: Dave Rigby <daver@couchbase.com>

diff --git a/src/dcp-consumer.cc b/src/dcp-consumer.cc
index e8d7282..472c7a3 100644 (file)
--- a/src/dcp-consumer.cc
+++ b/src/dcp-consumer.cc
@@ -71,7 +71,7 @@ std::string Processer::getDescription() {
 DcpConsumer::DcpConsumer(EventuallyPersistentEngine &engine, const void *cookie,
                          const std::string &name)
     : Consumer(engine, cookie, name), opaqueCounter(0), processTaskId(0),
-          itemsToProcess(false), lastNoopTime(ep_current_time()), backoffs(0) {
+          itemsToProcess(false), lastMessageTime(ep_current_time()), backoffs(0) {
     Configuration& config = engine.getConfiguration();
     streams = new passive_stream_t[config.getMaxVbuckets()];
     setSupportAck(false);
@@ -96,6 +96,7 @@ DcpConsumer::~DcpConsumer() {
 
 ENGINE_ERROR_CODE DcpConsumer::addStream(uint32_t opaque, uint16_t vbucket,
                                          uint32_t flags) {
+    lastMessageTime = ep_current_time();
     LockHolder lh(readyMutex);
     if (doDisconnect()) {
         return ENGINE_DISCONNECT;
@@ -143,6 +144,7 @@ ENGINE_ERROR_CODE DcpConsumer::addStream(uint32_t opaque, uint16_t vbucket,
 }
 
 ENGINE_ERROR_CODE DcpConsumer::closeStream(uint32_t opaque, uint16_t vbucket) {
+    lastMessageTime = ep_current_time();
     if (doDisconnect()) {
         streams[vbucket].reset();
         return ENGINE_DISCONNECT;
@@ -168,6 +170,7 @@ ENGINE_ERROR_CODE DcpConsumer::closeStream(uint32_t opaque, uint16_t vbucket) {
 
 ENGINE_ERROR_CODE DcpConsumer::streamEnd(uint32_t opaque, uint16_t vbucket,
                                          uint32_t flags) {
+    lastMessageTime = ep_current_time();
     if (doDisconnect()) {
         return ENGINE_DISCONNECT;
     }
@@ -205,6 +208,7 @@ ENGINE_ERROR_CODE DcpConsumer::mutation(uint32_t opaque, const void* key,
                                         uint64_t bySeqno, uint64_t revSeqno,
                                         uint32_t exptime, uint8_t nru,
                                         const void* meta, uint16_t nmeta) {
+    lastMessageTime = ep_current_time();
     if (doDisconnect()) {
         return ENGINE_DISCONNECT;
     }
@@ -247,6 +251,7 @@ ENGINE_ERROR_CODE DcpConsumer::deletion(uint32_t opaque, const void* key,
                                         uint16_t vbucket, uint64_t bySeqno,
                                         uint64_t revSeqno, const void* meta,
                                         uint16_t nmeta) {
+    lastMessageTime = ep_current_time();
     if (doDisconnect()) {
         return ENGINE_DISCONNECT;
     }
@@ -286,6 +291,7 @@ ENGINE_ERROR_CODE DcpConsumer::expiration(uint32_t opaque, const void* key,
                                           uint16_t vbucket, uint64_t bySeqno,
                                           uint64_t revSeqno, const void* meta,
                                           uint16_t nmeta) {
+    // lastMessageTime is set in deletion function
     return deletion(opaque, key, nkey, cas, vbucket, bySeqno, revSeqno, meta,
                     nmeta);
 }
@@ -295,6 +301,7 @@ ENGINE_ERROR_CODE DcpConsumer::snapshotMarker(uint32_t opaque,
                                               uint64_t start_seqno,
                                               uint64_t end_seqno,
                                               uint32_t flags) {
+    lastMessageTime = ep_current_time();
     if (doDisconnect()) {
         return ENGINE_DISCONNECT;
     }
@@ -329,11 +336,12 @@ ENGINE_ERROR_CODE DcpConsumer::snapshotMarker(uint32_t opaque,
 }
 
 ENGINE_ERROR_CODE DcpConsumer::noop(uint32_t opaque) {
-    lastNoopTime = ep_current_time();
+    lastMessageTime = ep_current_time();
     return ENGINE_SUCCESS;
 }
 
 ENGINE_ERROR_CODE DcpConsumer::flush(uint32_t opaque, uint16_t vbucket) {
+    lastMessageTime = ep_current_time();
     if (doDisconnect()) {
         return ENGINE_DISCONNECT;
     }
@@ -344,6 +352,7 @@ ENGINE_ERROR_CODE DcpConsumer::flush(uint32_t opaque, uint16_t vbucket) {
 ENGINE_ERROR_CODE DcpConsumer::setVBucketState(uint32_t opaque,
                                                uint16_t vbucket,
                                                vbucket_state_t state) {
+    lastMessageTime = ep_current_time();
     if (doDisconnect()) {
         return ENGINE_DISCONNECT;
     }
@@ -741,9 +750,10 @@ ENGINE_ERROR_CODE DcpConsumer::handleNoop(struct dcp_message_producers* producer
         return ret;
     }
 
-    if ((ep_current_time() - lastNoopTime) > (noopInterval * 2)) {
-        LOG(EXTENSION_LOG_WARNING, "%s Disconnecting because noop message has "
-            "no been received for %u seconds", logHeader(), (noopInterval * 2));
+    if ((ep_current_time() - lastMessageTime) > (noopInterval * 2)) {
+        LOG(EXTENSION_LOG_WARNING, "%s Disconnecting because a message has "
+            "not been received for %u seconds. lastMessageTime was %u seconds ago.",
+            logHeader(), (noopInterval * 2), (ep_current_time() - lastMessageTime));
         return ENGINE_DISCONNECT;
     }
 

diff --git a/src/dcp-consumer.h b/src/dcp-consumer.h
index 700d405..0f4d4eb 100644 (file)
--- a/src/dcp-consumer.h
+++ b/src/dcp-consumer.h
@@ -128,7 +128,7 @@ private:
     passive_stream_t* streams;
     opaque_map opaqueMap_;
 
-    rel_time_t lastNoopTime;
+    rel_time_t lastMessageTime;
     Couchbase::RelaxedAtomic<uint32_t> backoffs;
     uint32_t noopInterval;
     bool enableNoop;