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

* ep-engine/watson:
  MB-26979: ChkptProcessorTask should not own dcp stream objs
  MB-24142: Use correct unit for slowTask recording
  [BP] MB-25798: Don't schedule backfill until previous is complete
  [BP] MB-25798: Re-register a dropped cursor if we don't backfill
  [BP] MB-25798: Backfill task leave stream state unchanged
  [BP] MB-25798: Log when DcpProducer::Buffer log is full
  [BP] MB-25798: Log seqno data for when not scheduling backfill
  [BP] MB-25798: Improved logging for handle slow stream & scheduling backfill
  MB-25630: Include read-only KVStore stats in 'cbstats kvtimings'

Author: daverigby

src/checkpoint.cc

@@ -678,9 +678,9 @@ CursorRegResult CheckpointManager::registerCursorBySeqno(
          * number we are looking for is higher than anything currently assigned
          *  and there is already an assert above for this case.
          */
-        LOG(EXTENSION_LOG_WARNING, "Cursor not registered into vb %d "
-            " for stream '%s' because seqno %" PRIu64 " is too high",
-            vbucketId, name.c_str(), startBySeqno);
+        throw std::logic_error(
+                "CheckpointManager::registerCursorBySeqno the sequences number "
+                "is higher than anything currently assigned");
     }
     return result;
 }

src/connmap.cc

@@ -1146,7 +1146,7 @@ void DcpConnMap::closeStreams(CookieToConnectionMap& map) {
         DcpProducer* producer = dynamic_cast<DcpProducer*> (itr.second.get());
         if (producer) {
             producer->closeAllStreams();
-            producer->clearCheckpointProcessorTaskQueues();
+            producer->cancelCheckpointCreatorTask();
             // The producer may be in EWOULDBLOCK (if it's idle), therefore
             // notify him to ensure the front-end connection can close the TCP
             // connection.
@@ -1207,7 +1207,7 @@ void DcpConnMap::disconnect(const void *cookie) {
         DcpProducer* producer = dynamic_cast<DcpProducer*> (conn.get());
         if (producer) {
             producer->closeAllStreams();
-            producer->clearCheckpointProcessorTaskQueues();
+            producer->cancelCheckpointCreatorTask();
         } else {
             // Cancel consumer's processer task before closing all streams
             static_cast<DcpConsumer*>(conn.get())->cancelTask();

src/couch-kvstore/couch-kvstore.cc

@@ -1011,9 +1011,6 @@ void CouchKVStore::addStats(const std::string &prefix,
 
 void CouchKVStore::addTimingStats(const std::string &prefix,
                                   ADD_STAT add_stat, const void *c) {
-    if (isReadOnly()) {
-        return;
-    }
     const char *prefix_str = prefix.c_str();
     addStat(prefix_str, "commit",      st.commitHisto,      add_stat, c);
     addStat(prefix_str, "compact",     st.compactHisto,     add_stat, c);

src/dcp/producer.cc

@@ -81,7 +81,16 @@ bool DcpProducer::BufferLog::pauseIfFull() {
 
 void DcpProducer::BufferLog::unpauseIfSpaceAvailable() {
     ReaderLockHolder rlh(logLock);
-    if (getState_UNLOCKED() != Full) {
+    if (getState_UNLOCKED() == Full) {
+        LOG(EXTENSION_LOG_NOTICE, "%s Unable to notify paused connection "
+                    "because DcpProducer::BufferLog is full; "
+                    "ackedBytes:%" PRIu64 ", bytesSent:%" PRIu64 ", "
+                    "maxBytes:%" PRIu64 ,
+                    producer.logHeader(),
+                    uint64_t(ackedBytes),
+                    uint64_t(bytesSent),
+                    uint64_t(maxBytes));
+    } else {
         producer.notifyPaused(true);
     }
 }
@@ -93,6 +102,14 @@ void DcpProducer::BufferLog::acknowledge(size_t bytes) {
         release_UNLOCKED(bytes);
         ackedBytes += bytes;
         if (state == Full) {
+            LOG(EXTENSION_LOG_NOTICE, "%s Notifying paused connection now that "
+                "DcpProducer::Bufferlog is no longer full; "
+                "ackedBytes:%" PRIu64 ", bytesSent:%" PRIu64 ", "
+                "maxBytes:%" PRIu64 ,
+                producer.logHeader(),
+                uint64_t(ackedBytes),
+                uint64_t(bytesSent),
+                uint64_t(maxBytes));
             producer.notifyPaused(true);
         }
     }
@@ -167,14 +184,18 @@ DcpProducer::DcpProducer(EventuallyPersistentEngine &e, const void *cookie,
 
     backfillMgr.reset(new BackfillManager(&engine_));
 
-    checkpointCreatorTask = new ActiveStreamCheckpointProcessorTask(e);
+    checkpointCreatorTask = new ActiveStreamCheckpointProcessorTask(e, this);
     ExecutorPool::get()->schedule(checkpointCreatorTask, AUXIO_TASK_IDX);
 }
 
 DcpProducer::~DcpProducer() {
     backfillMgr.reset();
     delete rejectResp;
+}
 
+void DcpProducer::cancelCheckpointCreatorTask() {
+    static_cast<ActiveStreamCheckpointProcessorTask*>(
+            checkpointCreatorTask.get())->cancelTask();
     ExecutorPool::get()->cancel(checkpointCreatorTask->getId());
 }
 
@@ -784,18 +805,8 @@ bool DcpProducer::handleSlowStream(uint16_t vbid,
         if (stream) {
             if (stream->getName().compare(name) == 0) {
                 ActiveStream* as = static_cast<ActiveStream*>(stream.get());
-                if (as) {
-                    LOG(EXTENSION_LOG_NOTICE, "%s (vb %" PRIu16 ")  Producer "
-                        "is handling slow stream;"
-                        " state:%s lastReadSeqno:%" PRIu64
-                        " lastSentSeqno:%" PRIu64,
-                        logHeader(), vbid,
-                        Stream::stateName(as->getState()),
-                        as->getLastReadSeqno(),
-                        as->getLastSentSeqno());
-                    as->handleSlowStream();
-                    return true;
-                }
+                as->handleSlowStream();
+                return true;
             }
         }
     }
@@ -1030,8 +1041,3 @@ void DcpProducer::scheduleCheckpointProcessorTask(stream_t s) {
     static_cast<ActiveStreamCheckpointProcessorTask*>(checkpointCreatorTask.get())
         ->schedule(s);
 }
-
-void DcpProducer::clearCheckpointProcessorTaskQueues() {
-    static_cast<ActiveStreamCheckpointProcessorTask*>(checkpointCreatorTask.get())
-        ->clearQueues();
-}

src/dcp/producer.h

@@ -32,7 +32,13 @@ class DcpProducer : public Producer {
     DcpProducer(EventuallyPersistentEngine &e, const void *cookie,
                 const std::string &n, bool notifyOnly);
 
-    ~DcpProducer();
+    virtual ~DcpProducer();
+
+    /**
+     * Clears active stream checkpoint processor task's queue, resets its
+     * shared reference to the producer and cancels the task.
+     */
+    void cancelCheckpointCreatorTask();
 
     ENGINE_ERROR_CODE streamRequest(uint32_t flags, uint32_t opaque,
                                     uint16_t vbucket, uint64_t start_seqno,
@@ -201,10 +207,14 @@ class DcpProducer : public Producer {
     */
     void scheduleCheckpointProcessorTask(stream_t s);
 
-    /*
-        Clears active stream checkpoint processor task's queue.
-    */
-    void clearCheckpointProcessorTaskQueues();
+    /**
+     * Returns a shared reference to the stream associated with the vbucket
+     *
+     * @param vbid Vbucket id
+     *
+     * @return shared reference ptr to the stream associated
+     */
+    stream_t findStreamByVbid(uint16_t vbid);
 
 protected:
 
@@ -218,13 +228,14 @@ class DcpProducer : public Producer {
         Couchbase::RelaxedAtomic<bool> enabled;
     } noopCtx;
 
+    ExTask checkpointCreatorTask;
+
 private:
 
 
     DcpResponse* getNextItem();
 
     size_t getItemsRemaining();
-    stream_t findStreamByVbid(uint16_t vbid);
 
     std::string priority;
 
@@ -256,9 +267,7 @@ class DcpProducer : public Producer {
     AtomicValue<size_t> itemsSent;
     AtomicValue<size_t> totalBytesSent;
 
-    ExTask checkpointCreatorTask;
     static const uint32_t defaultNoopInerval;
-
 };
 
 #endif  // SRC_DCP_PRODUCER_H_