text: 
  /* -*- Mode: C++; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */
  /*
   *     Copyright 2010 Couchbase, Inc
   *
   *   Licensed under the Apache License, Version 2.0 (the "License");
   *   you may not use this file except in compliance with the License.
   *   You may obtain a copy of the License at
   *
   *       http://www.apache.org/licenses/LICENSE-2.0
   *
   *   Unless required by applicable law or agreed to in writing, software
   *   distributed under the License is distributed on an "AS IS" BASIS,
   *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   *   See the License for the specific language governing permissions and
   *   limitations under the License.
   */
  
  #include "config.h"
  
  #include "ep_engine.h"
  
  #include "backfill.h"
  #include "common.h"
  #include "connmap.h"
  #include "dcp/consumer.h"
  #include "dcp/dcpconnmap.h"
  #include "dcp/flow-control-manager.h"
  #include "dcp/producer.h"
  #include "ep_bucket.h"
  #include "ephemeral_bucket.h"
  #include "failover-table.h"
  #include "flusher.h"
  #include "htresizer.h"
  #include "logger.h"
  #include "memory_tracker.h"
  #include "replicationthrottle.h"
  #include "stats-info.h"
  #define STATWRITER_NAMESPACE core_engine
  #include "statwriter.h"
  #undef STATWRITER_NAMESPACE
  #include "string_utils.h"
  #include "warmup.h"
  
  #include <JSON_checker.h>
  #include <cJSON_utils.h>
  #include <memcached/engine.h>
  #include <memcached/extension.h>
  #include <memcached/protocol_binary.h>
  #include <memcached/server_api.h>
  #include <memcached/util.h>
  #include <platform/cb_malloc.h>
  #include <platform/checked_snprintf.h>
  #include <platform/make_unique.h>
  #include <platform/platform.h>
  #include <platform/processclock.h>
  
  #include <cstdio>
  #include <cstring>
  #include <fcntl.h>
  #include <fstream>
  #include <iostream>
  #include <limits>
  #include <mutex>
  #include <stdarg.h>
  #include <string>
  #include <vector>
  
  static size_t percentOf(size_t val, double percent) {
      return static_cast<size_t>(static_cast<double>(val) * percent);
  }
  
  struct EPHandleReleaser {
      void operator()(EventuallyPersistentEngine*) {
          ObjectRegistry::onSwitchThread(nullptr);
      }
  };
  
  using EPHandle = std::unique_ptr<EventuallyPersistentEngine, EPHandleReleaser>;
  
  /**
   * Helper function to acquire a handle to the engine which allows access to
   * the engine while the handle is in scope.
   * @param handle pointer to the engine
   * @return EPHandle which is a unique_ptr to an EventuallyPersistentEngine
   * with a custom deleter (EPHandleReleaser) which performs the required
   * ObjectRegistry release.
   */
  
  static inline EPHandle acquireEngine(ENGINE_HANDLE* handle) {
      auto ret = reinterpret_cast<EventuallyPersistentEngine*>(handle);
      ObjectRegistry::onSwitchThread(ret);
  
      return EPHandle(ret);
  }
  
  /**
   * Call the response callback and return the appropriate value so that
   * the core knows what to do..
   */
  static ENGINE_ERROR_CODE sendResponse(ADD_RESPONSE response, const void *key,
                                        uint16_t keylen,
                                        const void *ext, uint8_t extlen,
                                        const void *body, uint32_t bodylen,
                                        uint8_t datatype, uint16_t status,
                                        uint64_t cas, const void *cookie)
  {
      ENGINE_ERROR_CODE rv = ENGINE_FAILED;
      EventuallyPersistentEngine *e = ObjectRegistry::onSwitchThread(NULL, true);
      if (response(key, keylen, ext, extlen, body, bodylen, datatype,
                   status, cas, cookie)) {
          rv = ENGINE_SUCCESS;
      }
      ObjectRegistry::onSwitchThread(e);
      return rv;
  }
  
  template <typename T>
  static void validate(T v, T l, T h) {
      if (v < l || v > h) {
          throw std::runtime_error("Value out of range.");
      }
  }
  
  
  static void checkNumeric(const char* str) {
      int i = 0;
      if (str[0] == '-') {
          i++;
      }
      for (; str[i]; i++) {
          using namespace std;
          if (!isdigit(str[i])) {
              throw std::runtime_error("Value is not numeric");
          }
      }
  }
  
  static const engine_info* EvpGetInfo(ENGINE_HANDLE* handle) {
      return acquireEngine(handle)->getInfo();
  }
  
  static ENGINE_ERROR_CODE EvpInitialize(ENGINE_HANDLE* handle,
                                         const char* config_str) {
      return acquireEngine(handle)->initialize(config_str);
  }
  
  static void EvpDestroy(ENGINE_HANDLE* handle, const bool force) {
      auto eng = acquireEngine(handle);
      eng->destroy(force);
      delete eng.get();
  }
  
  static ENGINE_ERROR_CODE EvpItemAllocate(ENGINE_HANDLE* handle,
                                           const void* cookie,
                                           item** itm,
                                           const DocKey& key,
                                           const size_t nbytes,
                                           const int flags,
                                           const rel_time_t exptime,
                                           uint8_t datatype,
                                           uint16_t vbucket) {
      if (!mcbp::datatype::is_valid(datatype)) {
          LOG(EXTENSION_LOG_WARNING, "Invalid value for datatype "
              " (ItemAllocate)");
          return ENGINE_EINVAL;
      }
  
      return acquireEngine(handle)->itemAllocate(itm,
                                                 key,
                                                 nbytes,
                                                 0, // No privileged bytes
                                                 flags,
                                                 exptime,
                                                 datatype,
                                                 vbucket);
  }
  
  static bool EvpGetItemInfo(ENGINE_HANDLE *handle, const void *,
                             const item* itm, item_info *itm_info);
  static void EvpItemRelease(ENGINE_HANDLE* handle, const void *cookie,
                             item* itm);
  
  
  static std::pair<cb::unique_item_ptr, item_info> EvpItemAllocateEx(ENGINE_HANDLE* handle,
                                                                     const void* cookie,
                                                                     const DocKey& key,
                                                                     const size_t nbytes,
                                                                     const size_t priv_nbytes,
                                                                     const int flags,
                                                                     const rel_time_t exptime,
                                                                     uint8_t datatype,
                                                                     uint16_t vbucket) {
  
      item* it = nullptr;
      auto err = acquireEngine(handle)->itemAllocate(
              &it, key, nbytes, priv_nbytes, flags, exptime, datatype, vbucket);
  
      if (err != ENGINE_SUCCESS) {
          throw cb::engine_error(cb::engine_errc(err),
                                 "EvpItemAllocateEx: failed to allocate memory");
      }
  
      item_info info;
      if (!EvpGetItemInfo(handle, cookie, it, &info)) {
          EvpItemRelease(handle, cookie, it);
          throw cb::engine_error(cb::engine_errc::failed,
                                 "EvpItemAllocateEx: EvpGetItemInfo failed");
      }
  
      return std::make_pair(cb::unique_item_ptr{it, cb::ItemDeleter{handle}},
                            info);
  }
  
  static ENGINE_ERROR_CODE EvpItemDelete(ENGINE_HANDLE* handle,
                                         const void* cookie,
                                         const DocKey& key,
                                         uint64_t* cas,
                                         uint16_t vbucket,
                                         mutation_descr_t* mut_info) {
      if (!cas) {
          LOG(EXTENSION_LOG_WARNING,
              "EvpItemDelete(): cas ptr passed is null for vb: %" PRIu16,
              vbucket);
          return ENGINE_EINVAL;
      }
      return acquireEngine(handle)->itemDelete(
              cookie, key, *cas, vbucket, nullptr, nullptr, mut_info);
  }
  
  static void EvpItemRelease(ENGINE_HANDLE* handle,
                             const void* cookie,
                             item* itm) {
      acquireEngine(handle)->itemRelease(cookie, itm);
  }
  
  static ENGINE_ERROR_CODE EvpGet(ENGINE_HANDLE* handle,
                                  const void* cookie,
                                  item** itm,
                                  const DocKey& key,
                                  uint16_t vbucket,
                                  DocumentState document_state) {
      get_options_t options = static_cast<get_options_t>(QUEUE_BG_FETCH |
                                                         HONOR_STATES |
                                                         TRACK_REFERENCE |
                                                         DELETE_TEMP |
                                                         HIDE_LOCKED_CAS |
                                                         TRACK_STATISTICS);
  
      if (document_state == DocumentState::Deleted) {
          options = static_cast<get_options_t>(options | GET_DELETED_VALUE);
      }
  
      return acquireEngine(handle)->get(cookie, itm, key, vbucket, options);
  }
  
  static ENGINE_ERROR_CODE EvpGetLocked(ENGINE_HANDLE* handle,
                                        const void* cookie,
                                        item** itm,
                                        const DocKey& key,
                                        uint16_t vbucket,
                                        uint32_t lock_timeout) {
      return acquireEngine(handle)->get_locked(
              cookie, itm, key, vbucket, lock_timeout);
  }
  
  static ENGINE_ERROR_CODE EvpUnlock(ENGINE_HANDLE* handle,
                                     const void* cookie,
                                     const DocKey& key,
                                     uint16_t vbucket,
                                     uint64_t cas) {
      return acquireEngine(handle)->unlock(cookie, key, vbucket, cas);
  }
  
  static ENGINE_ERROR_CODE EvpGetStats(ENGINE_HANDLE* handle,
                                       const void* cookie,
                                       const char* stat_key,
                                       int nkey,
                                       ADD_STAT add_stat) {
      return acquireEngine(handle)->getStats(cookie, stat_key, nkey, add_stat);
  }
  
  static ENGINE_ERROR_CODE EvpStore(ENGINE_HANDLE* handle,
                                    const void* cookie,
                                    item* itm,
                                    uint64_t* cas,
                                    ENGINE_STORE_OPERATION operation,
                                    DocumentState document_state) {
      ENGINE_ERROR_CODE err_code = ENGINE_SUCCESS;
      auto engine = acquireEngine(handle);
      switch (document_state) {
      case DocumentState::Alive: {
          err_code = engine->store(cookie, itm, cas, operation);
          break;
      }
      case DocumentState::Deleted: {
          Item* item = static_cast<Item*>(itm);
          ItemMetaData itm_meta;
          mutation_descr_t mut_info;
  
          if (!cas) {
              LOG(EXTENSION_LOG_WARNING,
                  "EvpStore(): cas ptr passed is null for vb: %" PRIu16,
                  item->getVBucketId());
              return ENGINE_EINVAL;
          }
  
          /* Set the item as deleted */
          item->setDeleted();
          err_code = engine->itemDelete(cookie,
                                        item->getKey(),
                                        *cas,
                                        item->getVBucketId(),
                                        item,
                                        &itm_meta,
                                        &mut_info);
          if (err_code == ENGINE_SUCCESS) {
              item->setBySeqno(mut_info.seqno);
              item->setCas(itm_meta.cas);
              item->setFlags(itm_meta.flags);
              item->setExpTime(itm_meta.exptime);
          }
          break;
      }
      default:
          return ENGINE_ENOTSUP;
      }
      return err_code;
  }
  
  static ENGINE_ERROR_CODE EvpFlush(ENGINE_HANDLE* handle,
                                    const void* cookie) {
      return acquireEngine(handle)->flush(cookie);
  }
  
  static void EvpResetStats(ENGINE_HANDLE* handle, const void*) {
      acquireEngine(handle)->resetStats();
  }
  
  static protocol_binary_response_status setTapParam(
      EventuallyPersistentEngine* e,
      const char* keyz,
      const char* valz,
      std::string& msg) {
      protocol_binary_response_status rv = PROTOCOL_BINARY_RESPONSE_SUCCESS;
  
      try {
          if (strcmp(keyz, "tap_keepalive") == 0) {
              int v = std::stoi(valz);
              validate(v, 0, MAX_TAP_KEEP_ALIVE);
              e->setTapKeepAlive(static_cast<uint32_t>(v));
          } else if (strcmp(keyz, "replication_throttle_threshold") == 0) {
              e->getConfiguration().setReplicationThrottleThreshold(
                  std::stoull(valz));
          } else if (strcmp(keyz, "replication_throttle_queue_cap") == 0) {
              e->getConfiguration().setReplicationThrottleQueueCap(
                  std::stoll(valz));
          } else if (strcmp(keyz, "replication_throttle_cap_pcnt") == 0) {
              e->getConfiguration().setReplicationThrottleCapPcnt(
                  std::stoull(valz));
          } else {
              msg = "Unknown config param";
              rv = PROTOCOL_BINARY_RESPONSE_KEY_ENOENT;
          }
          // Handles exceptions thrown by the standard
          // library stoi/stoul style functions when not numeric
      } catch (std::invalid_argument&) {
          msg = "Argument was not numeric";
          rv = PROTOCOL_BINARY_RESPONSE_EINVAL;
  
          // Handles exceptions thrown by the standard library stoi/stoul
          // style functions when the conversion does not fit in the datatype
      } catch (std::out_of_range&) {
          msg = "Argument was out of range";
          rv = PROTOCOL_BINARY_RESPONSE_EINVAL;
  
          // Handles any miscellaenous exceptions in addition to the range_error
          // exceptions thrown by the configuration::set<param>() methods
      } catch (std::exception& error) {
          msg = error.what();
          rv = PROTOCOL_BINARY_RESPONSE_EINVAL;
      }
  
      return rv;
  }
  
  static protocol_binary_response_status setCheckpointParam(
      EventuallyPersistentEngine* e,
      const char* keyz,
      const char* valz,
      std::string& msg) {
      protocol_binary_response_status rv = PROTOCOL_BINARY_RESPONSE_SUCCESS;
  
      try {
          if (strcmp(keyz, "chk_max_items") == 0) {
              size_t v = std::stoull(valz);
              validate(v, size_t(MIN_CHECKPOINT_ITEMS),
                       size_t(MAX_CHECKPOINT_ITEMS));
              e->getConfiguration().setChkMaxItems(v);
          } else if (strcmp(keyz, "chk_period") == 0) {
              size_t v = std::stoull(valz);
              validate(v, size_t(MIN_CHECKPOINT_PERIOD),
                       size_t(MAX_CHECKPOINT_PERIOD));
              e->getConfiguration().setChkPeriod(v);
          } else if (strcmp(keyz, "max_checkpoints") == 0) {
              size_t v = std::stoull(valz);
              validate(v, size_t(DEFAULT_MAX_CHECKPOINTS),
                       size_t(MAX_CHECKPOINTS_UPPER_BOUND));
              e->getConfiguration().setMaxCheckpoints(v);
          } else if (strcmp(keyz, "item_num_based_new_chk") == 0) {
              e->getConfiguration().setItemNumBasedNewChk(cb_stob(valz));
          } else if (strcmp(keyz, "keep_closed_chks") == 0) {
              e->getConfiguration().setKeepClosedChks(cb_stob(valz));
          } else if (strcmp(keyz, "enable_chk_merge") == 0) {
              e->getConfiguration().setEnableChkMerge(cb_stob(valz));
          } else {
              msg = "Unknown config param";
              rv = PROTOCOL_BINARY_RESPONSE_KEY_ENOENT;
          }
  
          // Handles exceptions thrown by the cb_stob function
      } catch (invalid_argument_bool& error) {
          msg = error.what();
          rv = PROTOCOL_BINARY_RESPONSE_EINVAL;
  
          // Handles exceptions thrown by the standard
          // library stoi/stoul style functions when not numeric
      } catch (std::invalid_argument&) {
          msg = "Argument was not numeric";
          rv = PROTOCOL_BINARY_RESPONSE_EINVAL;
  
          // Handles exceptions thrown by the standard library stoi/stoul
          // style functions when the conversion does not fit in the datatype
      } catch (std::out_of_range&) {
          msg = "Argument was out of range";
          rv = PROTOCOL_BINARY_RESPONSE_EINVAL;
  
          // Handles any miscellaenous exceptions in addition to the range_error
          // exceptions thrown by the configuration::set<param>() methods
      } catch (std::exception& error) {
          msg = error.what();
          rv = PROTOCOL_BINARY_RESPONSE_EINVAL;
      }
  
      return rv;
  }
  
  static protocol_binary_response_status setFlushParam(
      EventuallyPersistentEngine* e,
      const char* keyz,
      const char* valz,
      std::string& msg) {
      protocol_binary_response_status rv = PROTOCOL_BINARY_RESPONSE_SUCCESS;
  
      // Handle the actual mutation.
      try {
          if (strcmp(keyz, "bg_fetch_delay") == 0) {
              e->getConfiguration().setBgFetchDelay(std::stoull(valz));
          } else if (strcmp(keyz, "flushall_enabled") == 0) {
              e->getConfiguration().setFlushallEnabled(cb_stob(valz));
          } else if (strcmp(keyz, "max_size") == 0) {
              size_t vsize = std::stoull(valz);
  
              e->getConfiguration().setMaxSize(vsize);
              EPStats& st = e->getEpStats();
              e->getConfiguration().setMemLowWat(percentOf(vsize,
                                                           st.mem_low_wat_percent));
              e->getConfiguration().setMemHighWat(percentOf(vsize,
                                                            st.mem_high_wat_percent));
          } else if (strcmp(keyz, "mem_low_wat") == 0) {
              e->getConfiguration().setMemLowWat(std::stoull(valz));
          } else if (strcmp(keyz, "mem_high_wat") == 0) {
              e->getConfiguration().setMemHighWat(std::stoull(valz));
          } else if (strcmp(keyz, "backfill_mem_threshold") == 0) {
              e->getConfiguration().setBackfillMemThreshold(
                  std::stoull(valz));
          } else if (strcmp(keyz, "compaction_exp_mem_threshold") == 0) {
              e->getConfiguration().setCompactionExpMemThreshold(
                  std::stoull(valz));
          } else if (strcmp(keyz, "mutation_mem_threshold") == 0) {
              e->getConfiguration().setMutationMemThreshold(
                  std::stoull(valz));
          } else if (strcmp(keyz, "timing_log") == 0) {
              EPStats& stats = e->getEpStats();
              std::ostream* old = stats.timingLog;
              stats.timingLog = NULL;
              delete old;
              if (strcmp(valz, "off") == 0) {
                  LOG(EXTENSION_LOG_INFO, "Disabled timing log.");
              } else {
                  std::ofstream* tmp(new std::ofstream(valz));
                  if (tmp->good()) {
                      LOG(EXTENSION_LOG_INFO,
                          "Logging detailed timings to ``%s''.", valz);
                      stats.timingLog = tmp;
                  } else {
                      LOG(EXTENSION_LOG_WARNING,
                          "Error setting detailed timing log to ``%s'':  %s",
                          valz, strerror(errno));
                      delete tmp;
                  }
              }
          } else if (strcmp(keyz, "exp_pager_enabled") == 0) {
              e->getConfiguration().setExpPagerEnabled(cb_stob(valz));
          } else if (strcmp(keyz, "exp_pager_stime") == 0) {
              e->getConfiguration().setExpPagerStime(std::stoull(valz));
          } else if (strcmp(keyz, "exp_pager_initial_run_time") == 0) {
              e->getConfiguration().setExpPagerInitialRunTime(
                  std::stoll(valz));
          } else if (strcmp(keyz, "access_scanner_enabled") == 0) {
              e->getConfiguration().setAccessScannerEnabled(cb_stob(valz));
          } else if (strcmp(keyz, "alog_sleep_time") == 0) {
              e->getConfiguration().setAlogSleepTime(std::stoull(valz));
          } else if (strcmp(keyz, "alog_task_time") == 0) {
              e->getConfiguration().setAlogTaskTime(std::stoull(valz));
          } else if (strcmp(keyz, "pager_active_vb_pcnt") == 0) {
              e->getConfiguration().setPagerActiveVbPcnt(std::stoull(valz));
          } else if (strcmp(keyz, "warmup_min_memory_threshold") == 0) {
              e->getConfiguration().setWarmupMinMemoryThreshold(
                  std::stoull(valz));
          } else if (strcmp(keyz, "warmup_min_items_threshold") == 0) {
              e->getConfiguration().setWarmupMinItemsThreshold(
                  std::stoull(valz));
          } else if (strcmp(keyz, "max_num_readers") == 0) {
              size_t value = std::stoull(valz);
              e->getConfiguration().setMaxNumReaders(value);
              ExecutorPool::get()->setMaxReaders(value);
          } else if (strcmp(keyz, "max_num_writers") == 0) {
              size_t value = std::stoull(valz);
              e->getConfiguration().setMaxNumWriters(value);
              ExecutorPool::get()->setMaxWriters(value);
          } else if (strcmp(keyz, "max_num_auxio") == 0) {
              size_t value = std::stoull(valz);
              e->getConfiguration().setMaxNumAuxio(value);
              ExecutorPool::get()->setMaxAuxIO(value);
          } else if (strcmp(keyz, "max_num_nonio") == 0) {
              size_t value = std::stoull(valz);
              e->getConfiguration().setMaxNumNonio(value);
              ExecutorPool::get()->setMaxNonIO(value);
          } else if (strcmp(keyz, "bfilter_enabled") == 0) {
              e->getConfiguration().setBfilterEnabled(cb_stob(valz));
          } else if (strcmp(keyz, "bfilter_residency_threshold") == 0) {
              e->getConfiguration().setBfilterResidencyThreshold(
                  std::stof(valz));
          } else if (strcmp(keyz, "defragmenter_enabled") == 0) {
              e->getConfiguration().setDefragmenterEnabled(cb_stob(valz));
          } else if (strcmp(keyz, "defragmenter_interval") == 0) {
              size_t v = std::stoull(valz);
              // Adding separate validation as external limit is minimum 1
              // to prevent setting defragmenter to constantly run
              validate(v, size_t(1), std::numeric_limits<size_t>::max());
              e->getConfiguration().setDefragmenterInterval(v);
          } else if (strcmp(keyz, "defragmenter_age_threshold") == 0) {
              e->getConfiguration().setDefragmenterAgeThreshold(
                  std::stoull(valz));
          } else if (strcmp(keyz, "defragmenter_chunk_duration") == 0) {
              e->getConfiguration().setDefragmenterChunkDuration(
                  std::stoull(valz));
          } else if (strcmp(keyz, "defragmenter_run") == 0) {
              e->runDefragmenterTask();
          } else if (strcmp(keyz, "compaction_write_queue_cap") == 0) {
              e->getConfiguration().setCompactionWriteQueueCap(
                  std::stoull(valz));
          } else if (strcmp(keyz, "dcp_min_compression_ratio") == 0) {
              e->getConfiguration().setDcpMinCompressionRatio(
                  std::stof(valz));
          } else if (strcmp(keyz, "access_scanner_run") == 0) {
              if (!(e->runAccessScannerTask())) {
                  rv = PROTOCOL_BINARY_RESPONSE_ETMPFAIL;
              }
          } else if (strcmp(keyz, "vb_state_persist_run") == 0) {
              e->runVbStatePersistTask(std::stoi(valz));
          } else {
              msg = "Unknown config param";
              rv = PROTOCOL_BINARY_RESPONSE_KEY_ENOENT;
          }
          // Handles exceptions thrown by the cb_stob function
      } catch (invalid_argument_bool& error) {
          msg = error.what();
          rv = PROTOCOL_BINARY_RESPONSE_EINVAL;
  
          // Handles exceptions thrown by the standard
          // library stoi/stoul style functions when not numeric
      } catch (std::invalid_argument&) {
          msg = "Argument was not numeric";
          rv = PROTOCOL_BINARY_RESPONSE_EINVAL;
  
          // Handles exceptions thrown by the standard library stoi/stoul
          // style functions when the conversion does not fit in the datatype
      } catch (std::out_of_range&) {
          msg = "Argument was out of range";
          rv = PROTOCOL_BINARY_RESPONSE_EINVAL;
  
          // Handles any miscellaneous exceptions in addition to the range_error
          // exceptions thrown by the configuration::set<param>() methods
      } catch (std::exception& error) {
          msg = error.what();
          rv = PROTOCOL_BINARY_RESPONSE_EINVAL;
      }
  
      return rv;
  }
  
  static protocol_binary_response_status setDcpParam(
      EventuallyPersistentEngine* e,
      const char* keyz,
      const char* valz,
      std::string& msg) {
      protocol_binary_response_status rv = PROTOCOL_BINARY_RESPONSE_SUCCESS;
      try {
  
          if (strcmp(keyz,
                     "dcp_consumer_process_buffered_messages_yield_limit") == 0) {
              size_t v = atoi(valz);
              checkNumeric(valz);
              validate(v, size_t(1), std::numeric_limits<size_t>::max());
              e->getConfiguration().setDcpConsumerProcessBufferedMessagesYieldLimit(
                  v);
          } else if (
              strcmp(keyz, "dcp_consumer_process_buffered_messages_batch_size") ==
              0) {
              size_t v = atoi(valz);
              checkNumeric(valz);
              validate(v, size_t(1), std::numeric_limits<size_t>::max());
              e->getConfiguration().setDcpConsumerProcessBufferedMessagesBatchSize(
                  v);
          } else {
              msg = "Unknown config param";
              rv = PROTOCOL_BINARY_RESPONSE_KEY_ENOENT;
          }
      } catch (std::runtime_error& ex) {
          msg = "Value out of range.";
          rv = PROTOCOL_BINARY_RESPONSE_EINVAL;
      }
  
      return rv;
  }
  
  static protocol_binary_response_status setVbucketParam(
      EventuallyPersistentEngine* e,
      uint16_t vbucket,
      const char* keyz,
      const char* valz,
      std::string& msg) {
      protocol_binary_response_status rv = PROTOCOL_BINARY_RESPONSE_SUCCESS;
      try {
          if (strcmp(keyz, "hlc_drift_ahead_threshold_us") == 0) {
              uint64_t v = std::strtoull(valz, nullptr, 10);
              checkNumeric(valz);
              e->getConfiguration().setHlcDriftAheadThresholdUs(v);
          } else if (strcmp(keyz, "hlc_drift_behind_threshold_us") == 0) {
              uint64_t v = std::strtoull(valz, nullptr, 10);
              checkNumeric(valz);
              e->getConfiguration().setHlcDriftBehindThresholdUs(v);
          } else if (strcmp(keyz, "max_cas") == 0) {
              uint64_t v = std::strtoull(valz, nullptr, 10);
              checkNumeric(valz);
              LOG(EXTENSION_LOG_WARNING, "setVbucketParam: max_cas:%" PRIu64 " "
                  "vb:%" PRIu16 "\n", v, vbucket);
              if (e->getKVBucket()->forceMaxCas(vbucket, v) != ENGINE_SUCCESS) {
                  rv = PROTOCOL_BINARY_RESPONSE_NOT_MY_VBUCKET;
                  msg = "Not my vbucket";
              }
          } else {
              msg = "Unknown config param";
              rv = PROTOCOL_BINARY_RESPONSE_KEY_ENOENT;
          }
      } catch (std::runtime_error& ex) {
          msg = "Value out of range.";
          rv = PROTOCOL_BINARY_RESPONSE_EINVAL;
      }
      return rv;
  }
  
  static protocol_binary_response_status evictKey(
      EventuallyPersistentEngine* e,
      protocol_binary_request_header
      * request,
      const char** msg,
      size_t* msg_size,
      DocNamespace docNamespace) {
      protocol_binary_request_no_extras* req =
          (protocol_binary_request_no_extras*)request;
  
      const uint8_t* keyPtr = reinterpret_cast<const uint8_t*>(request) +
                              sizeof(*request);
      size_t keylen = ntohs(req->message.header.request.keylen);
      uint16_t vbucket = ntohs(request->request.vbucket);
  
      LOG(EXTENSION_LOG_DEBUG, "Manually evicting object with key{%.*s}\n",
          int(keylen), keyPtr);
      auto rv = e->evictKey(DocKey(keyPtr, keylen, docNamespace), vbucket,
                            msg, msg_size);
      if (rv == PROTOCOL_BINARY_RESPONSE_NOT_MY_VBUCKET ||
          rv == PROTOCOL_BINARY_RESPONSE_KEY_ENOENT) {
          if (e->isDegradedMode()) {
              return PROTOCOL_BINARY_RESPONSE_ETMPFAIL;
          }
      }
      return rv;
  }
  
  static protocol_binary_response_status setParam(
      EventuallyPersistentEngine* e,
      protocol_binary_request_set_param
      * req,
      std::string& msg) {
  
      size_t keylen = ntohs(req->message.header.request.keylen);
      uint8_t extlen = req->message.header.request.extlen;
      size_t vallen = ntohl(req->message.header.request.bodylen);
      uint16_t vbucket = ntohs(req->message.header.request.vbucket);
      protocol_binary_engine_param_t paramtype =
          static_cast<protocol_binary_engine_param_t>(ntohl(
              req->message.body.param_type));
  
      if (keylen == 0 || (vallen - keylen - extlen) == 0) {
          return PROTOCOL_BINARY_RESPONSE_EINVAL;
      }
  
      const char* keyp = reinterpret_cast<const char*>(req->bytes)
                         + sizeof(req->bytes);
      const char* valuep = keyp + keylen;
      vallen -= (keylen + extlen);
  
      char keyz[128];
      char valz[512];
  
      // Read the key.
      if (keylen >= sizeof(keyz)) {
          msg = "Key is too large.";
          return PROTOCOL_BINARY_RESPONSE_EINVAL;
      }
      memcpy(keyz, keyp, keylen);
      keyz[keylen] = 0x00;
  
      // Read the value.
      if (vallen >= sizeof(valz)) {
          msg = "Value is too large.";
          return PROTOCOL_BINARY_RESPONSE_EINVAL;
      }
      memcpy(valz, valuep, vallen);
      valz[vallen] = 0x00;
  
      protocol_binary_response_status rv;
  
      switch (paramtype) {
      case protocol_binary_engine_param_flush:
          rv = setFlushParam(e, keyz, valz, msg);
          break;
      case protocol_binary_engine_param_tap:
          rv = setTapParam(e, keyz, valz, msg);
          break;
      case protocol_binary_engine_param_checkpoint:
          rv = setCheckpointParam(e, keyz, valz, msg);
          break;
      case protocol_binary_engine_param_dcp:
          rv = setDcpParam(e, keyz, valz, msg);
          break;
      case protocol_binary_engine_param_vbucket:
          rv = setVbucketParam(e, vbucket, keyz, valz, msg);
          break;
      default:
          rv = PROTOCOL_BINARY_RESPONSE_UNKNOWN_COMMAND;
      }
  
      return rv;
  }
  
  static ENGINE_ERROR_CODE getVBucket(EventuallyPersistentEngine* e,
                                      const void* cookie,
                                      protocol_binary_request_header* request,
                                      ADD_RESPONSE response) {
      protocol_binary_request_get_vbucket* req =
          reinterpret_cast<protocol_binary_request_get_vbucket*>(request);
      if (req == nullptr) {
          throw std::invalid_argument("getVBucket: Unable to convert req"
                                          " to protocol_binary_request_get_vbucket");
      }
  
      uint16_t vbucket = ntohs(req->message.header.request.vbucket);
      RCPtr<VBucket> vb = e->getVBucket(vbucket);
      if (!vb) {
          return e->sendNotMyVBucketResponse(response, cookie, 0);
      } else {
          vbucket_state_t state = (vbucket_state_t)ntohl(vb->getState());
          return sendResponse(response, NULL, 0, NULL, 0, &state,
                              sizeof(state),
                              PROTOCOL_BINARY_RAW_BYTES,
                              PROTOCOL_BINARY_RESPONSE_SUCCESS, 0, cookie);
      }
  }
  
  static ENGINE_ERROR_CODE setVBucket(EventuallyPersistentEngine* e,
                                      const void* cookie,
                                      protocol_binary_request_header* request,
                                      ADD_RESPONSE response) {
  
      protocol_binary_request_set_vbucket* req =
          reinterpret_cast<protocol_binary_request_set_vbucket*>(request);
  
      uint64_t cas = ntohll(req->message.header.request.cas);
  
      size_t bodylen = ntohl(req->message.header.request.bodylen)
                       - ntohs(req->message.header.request.keylen);
      if (bodylen != sizeof(vbucket_state_t)) {
          const std::string msg("Incorrect packet format");
          return sendResponse(response, NULL, 0, NULL, 0, msg.c_str(),
                              msg.length(), PROTOCOL_BINARY_RAW_BYTES,
                              PROTOCOL_BINARY_RESPONSE_EINVAL,
                              cas, cookie);
      }
  
      vbucket_state_t state;
      memcpy(&state, &req->message.body.state, sizeof(state));
      state = static_cast<vbucket_state_t>(ntohl(state));
  
      if (!is_valid_vbucket_state_t(state)) {
          const std::string msg("Invalid vbucket state");
          return sendResponse(response, NULL, 0, NULL, 0, msg.c_str(),
                              msg.length(), PROTOCOL_BINARY_RAW_BYTES,
                              PROTOCOL_BINARY_RESPONSE_EINVAL,
                              cas, cookie);
      }
  
      uint16_t vb = ntohs(req->message.header.request.vbucket);
      if (e->setVBucketState(vb, state, false) == ENGINE_ERANGE) {
          const std::string msg("VBucket number too big");
          return sendResponse(response, NULL, 0, NULL, 0, msg.c_str(),
                              msg.length(), PROTOCOL_BINARY_RAW_BYTES,
                              PROTOCOL_BINARY_RESPONSE_ERANGE,
                              cas, cookie);
      }
      return sendResponse(response, NULL, 0, NULL, 0, NULL, 0,
                          PROTOCOL_BINARY_RAW_BYTES,
                          PROTOCOL_BINARY_RESPONSE_SUCCESS,
                          cas, cookie);
  }
  
  static ENGINE_ERROR_CODE delVBucket(EventuallyPersistentEngine* e,
                                      const void* cookie,
                                      protocol_binary_request_header* req,
                                      ADD_RESPONSE response) {
  
      uint64_t cas = ntohll(req->request.cas);
  
      protocol_binary_response_status res = PROTOCOL_BINARY_RESPONSE_SUCCESS;
      uint16_t vbucket = ntohs(req->request.vbucket);
  
      std::string msg = "";
      if (ntohs(req->request.keylen) > 0 || req->request.extlen > 0) {
          msg = "Incorrect packet format.";
          return sendResponse(response, NULL, 0, NULL, 0, msg.c_str(),
                              msg.length(),
                              PROTOCOL_BINARY_RAW_BYTES,
                              PROTOCOL_BINARY_RESPONSE_EINVAL, cas, cookie);
      }
  
      bool sync = false;
      uint32_t bodylen = ntohl(req->request.bodylen);
      if (bodylen > 0) {
          const char* ptr = reinterpret_cast<const char*>(req->bytes) +
                            sizeof(req->bytes);
          if (bodylen == 7 && strncmp(ptr, "async=0", bodylen) == 0) {
              sync = true;
          }
      }
  
      ENGINE_ERROR_CODE err;
      void* es = e->getEngineSpecific(cookie);
      if (sync) {
          if (es == NULL) {
              err = e->deleteVBucket(vbucket, cookie);
              e->storeEngineSpecific(cookie, e);
          } else {
              e->storeEngineSpecific(cookie, NULL);
              LOG(EXTENSION_LOG_INFO,
                  "Completed sync deletion of vbucket %u",
                  (unsigned)vbucket);
              err = ENGINE_SUCCESS;
          }
      } else {
          err = e->deleteVBucket(vbucket);
      }
      switch (err) {
      case ENGINE_SUCCESS:
          LOG(EXTENSION_LOG_NOTICE,
              "Deletion of vbucket %d was completed.", vbucket);
          break;
      case ENGINE_NOT_MY_VBUCKET:
          LOG(EXTENSION_LOG_WARNING, "Deletion of vbucket %d failed "
              "because the vbucket doesn't exist!!!", vbucket);
          res = PROTOCOL_BINARY_RESPONSE_NOT_MY_VBUCKET;
          break;
      case ENGINE_EINVAL:
          LOG(EXTENSION_LOG_WARNING, "Deletion of vbucket %d failed "
              "because the vbucket is not in a dead state\n", vbucket);
          msg = "Failed to delete vbucket.  Must be in the dead state.";
          res = PROTOCOL_BINARY_RESPONSE_EINVAL;
          break;
      case ENGINE_EWOULDBLOCK:
          LOG(EXTENSION_LOG_NOTICE, "Request for vbucket %d deletion is in"
                  " EWOULDBLOCK until the database file is removed from disk",
              vbucket);
          e->storeEngineSpecific(cookie, req);
          return ENGINE_EWOULDBLOCK;
      default:
          LOG(EXTENSION_LOG_WARNING, "Deletion of vbucket %d failed "
              "because of unknown reasons\n", vbucket);
          msg = "Failed to delete vbucket.  Unknown reason.";
          res = PROTOCOL_BINARY_RESPONSE_EINTERNAL;
      }
  
      if (err != ENGINE_NOT_MY_VBUCKET) {
          return sendResponse(response, NULL, 0, NULL, 0, msg.c_str(),
                              msg.length(), PROTOCOL_BINARY_RAW_BYTES,
                              res, cas, cookie);
      } else {
          return e->sendNotMyVBucketResponse(response, cookie, cas);
      }
  }
  
  static ENGINE_ERROR_CODE getReplicaCmd(EventuallyPersistentEngine* e,
                                         protocol_binary_request_header* request,
                                         const void* cookie,
                                         Item** it,
                                         const char** msg,
                                         protocol_binary_response_status* res,
                                         DocNamespace docNamespace) {
      KVBucketIface* kvb = e->getKVBucket();
      protocol_binary_request_no_extras* req =
          (protocol_binary_request_no_extras*)request;
      int keylen = ntohs(req->message.header.request.keylen);
      uint16_t vbucket = ntohs(req->message.header.request.vbucket);
      ENGINE_ERROR_CODE error_code;
      DocKey key(reinterpret_cast<const uint8_t*>(request) + sizeof(*request),
                 keylen, docNamespace);
  
      GetValue rv(kvb->getReplica(key, vbucket, cookie));
  
      if ((error_code = rv.getStatus()) != ENGINE_SUCCESS) {
          if (error_code == ENGINE_NOT_MY_VBUCKET) {
              *res = PROTOCOL_BINARY_RESPONSE_NOT_MY_VBUCKET;
              return error_code;
          } else if (error_code == ENGINE_TMPFAIL) {
              *msg = "NOT_FOUND";
              *res = PROTOCOL_BINARY_RESPONSE_KEY_ENOENT;
          } else {
              return error_code;
          }
      } else {
          *it = rv.getValue();
          *res = PROTOCOL_BINARY_RESPONSE_SUCCESS;
      }
      ++(e->getEpStats().numOpsGet);
      return ENGINE_SUCCESS;
  }
  
  static ENGINE_ERROR_CODE compactDB(EventuallyPersistentEngine* e,
                                     const void* cookie,
                                     protocol_binary_request_compact_db* req,
                                     ADD_RESPONSE response) {
  
      std::string msg = "";
      protocol_binary_response_status res = PROTOCOL_BINARY_RESPONSE_SUCCESS;
      compaction_ctx compactreq;
      uint64_t cas = ntohll(req->message.header.request.cas);
  
      if (ntohs(req->message.header.request.keylen) > 0 ||
          req->message.header.request.extlen != 24) {
          LOG(EXTENSION_LOG_WARNING,
              "Compaction received bad ext/key len %d/%d.",
              req->message.header.request.extlen,
              ntohs(req->message.header.request.keylen));
          msg = "Incorrect packet format.";
          return sendResponse(response, NULL, 0, NULL, 0, msg.c_str(),
                              msg.length(),
                              PROTOCOL_BINARY_RAW_BYTES,
                              PROTOCOL_BINARY_RESPONSE_EINVAL, cas, cookie);
      }
      EPStats& stats = e->getEpStats();
      compactreq.purge_before_ts = ntohll(req->message.body.purge_before_ts);
      compactreq.purge_before_seq =
          ntohll(req->message.body.purge_before_seq);
      compactreq.drop_deletes = req->message.body.drop_deletes;
      compactreq.db_file_id = e->getKVBucket()->getDBFileId(*req);
      uint16_t vbid = ntohs(req->message.header.request.vbucket);
  
      ENGINE_ERROR_CODE err;
      void* es = e->getEngineSpecific(cookie);
      if (es == NULL) {
          ++stats.pendingCompactions;
          e->storeEngineSpecific(cookie, e);
          err = e->compactDB(vbid, compactreq, cookie);
      } else {
          e->storeEngineSpecific(cookie, NULL);
          err = ENGINE_SUCCESS;
      }
  
      switch (err) {
      case ENGINE_SUCCESS:
          LOG(EXTENSION_LOG_NOTICE,
              "Compaction of db file id: %d completed.", compactreq.db_file_id);
          break;
      case ENGINE_NOT_MY_VBUCKET:
          --stats.pendingCompactions;
          LOG(EXTENSION_LOG_WARNING, "Compaction of db file id: %d failed "
              "because the db file doesn't exist!!!", compactreq.db_file_id);
          res = PROTOCOL_BINARY_RESPONSE_NOT_MY_VBUCKET;
          break;
      case ENGINE_EINVAL:
          --stats.pendingCompactions;
          LOG(EXTENSION_LOG_WARNING, "Compaction of db file id: %d failed "
              "because of an invalid argument", compactreq.db_file_id);
          res = PROTOCOL_BINARY_RESPONSE_EINVAL;
          break;
      case ENGINE_EWOULDBLOCK:
          LOG(EXTENSION_LOG_NOTICE,
              "Compaction of db file id: %d scheduled "
                  "(awaiting completion).", compactreq.db_file_id);
          e->storeEngineSpecific(cookie, req);
          return ENGINE_EWOULDBLOCK;
      case ENGINE_TMPFAIL:
          LOG(EXTENSION_LOG_WARNING, "Request to compact db file id: %d hit"
                  " a temporary failure and may need to be retried",
              compactreq.db_file_id);
          msg = "Temporary failure in compacting db file.";
          res = PROTOCOL_BINARY_RESPONSE_ETMPFAIL;
          break;
      default:
          --stats.pendingCompactions;
          LOG(EXTENSION_LOG_WARNING, "Compaction of db file id: %d failed "
              "because of unknown reasons\n", compactreq.db_file_id);
          msg = "Failed to compact db file.  Unknown reason.";
          res = PROTOCOL_BINARY_RESPONSE_EINTERNAL;
          break;
      }
  
      if (err != ENGINE_NOT_MY_VBUCKET) {
          return sendResponse(response, NULL, 0, NULL, 0, msg.c_str(),
                              msg.length(), PROTOCOL_BINARY_RAW_BYTES,
                              res, cas, cookie);
      } else {
          return e->sendNotMyVBucketResponse(response, cookie, cas);
      }
  }
  
  static ENGINE_ERROR_CODE processUnknownCommand(
      EventuallyPersistentEngine* h,
      const void* cookie,
      protocol_binary_request_header* request,
      ADD_RESPONSE response,
      DocNamespace docNamespace) {
      protocol_binary_response_status res =
          PROTOCOL_BINARY_RESPONSE_UNKNOWN_COMMAND;
      std::string dynamic_msg;
      const char* msg = NULL;
      size_t msg_size = 0;
      Item* itm = NULL;
  
      EPStats& stats = h->getEpStats();
      ENGINE_ERROR_CODE rv = ENGINE_SUCCESS;
  
      /**
       * Session validation
       * (For ns_server commands only)
       */
      switch (request->request.opcode) {
      case PROTOCOL_BINARY_CMD_SET_PARAM:
      case PROTOCOL_BINARY_CMD_SET_VBUCKET:
      case PROTOCOL_BINARY_CMD_DEL_VBUCKET:
      case PROTOCOL_BINARY_CMD_DEREGISTER_TAP_CLIENT:
      case PROTOCOL_BINARY_CMD_CHANGE_VB_FILTER:
      case PROTOCOL_BINARY_CMD_SET_CLUSTER_CONFIG:
      case PROTOCOL_BINARY_CMD_COMPACT_DB: {
          if (h->getEngineSpecific(cookie) == NULL) {
              uint64_t cas = ntohll(request->request.cas);
              if (!h->validateSessionCas(cas)) {
                  const std::string message("Invalid session token");
                  return sendResponse(response, NULL, 0, NULL, 0,
                                      message.c_str(), message.length(),
                                      PROTOCOL_BINARY_RAW_BYTES,
                                      PROTOCOL_BINARY_RESPONSE_KEY_EEXISTS,
                                      cas, cookie);
              }
          }
          break;
      }
      default:
          break;
      }
  
      switch (request->request.opcode) {
      case PROTOCOL_BINARY_CMD_GET_ALL_VB_SEQNOS:
          return h->getAllVBucketSequenceNumbers(cookie, request, response);
  
      case PROTOCOL_BINARY_CMD_GET_VBUCKET: {
          BlockTimer timer(&stats.getVbucketCmdHisto);
          rv = getVBucket(h, cookie, request, response);
          return rv;
      }
      case PROTOCOL_BINARY_CMD_DEL_VBUCKET: {
          BlockTimer timer(&stats.delVbucketCmdHisto);
          rv = delVBucket(h, cookie, request, response);
          if (rv != ENGINE_EWOULDBLOCK) {
              h->decrementSessionCtr();
              h->storeEngineSpecific(cookie, NULL);
          }
          return rv;
      }
      case PROTOCOL_BINARY_CMD_SET_VBUCKET: {
          BlockTimer timer(&stats.setVbucketCmdHisto);
          rv = setVBucket(h, cookie, request, response);
          h->decrementSessionCtr();
          return rv;
      }
      case PROTOCOL_BINARY_CMD_TOUCH:
      case PROTOCOL_BINARY_CMD_GAT:
      case PROTOCOL_BINARY_CMD_GATQ: {
          rv = h->touch(cookie, request, response, docNamespace);
          return rv;
      }
      case PROTOCOL_BINARY_CMD_STOP_PERSISTENCE:
          res = h->stopFlusher(&msg, &msg_size);
          break;
      case PROTOCOL_BINARY_CMD_START_PERSISTENCE:
          res = h->startFlusher(&msg, &msg_size);
          break;
      case PROTOCOL_BINARY_CMD_SET_PARAM:
          res = setParam(h,
                         reinterpret_cast<protocol_binary_request_set_param*>(request),
                         dynamic_msg);
          msg = dynamic_msg.c_str();
          msg_size = dynamic_msg.length();
          h->decrementSessionCtr();
          break;
      case PROTOCOL_BINARY_CMD_EVICT_KEY:
          res = evictKey(h, request, &msg, &msg_size, docNamespace);
          break;
      case PROTOCOL_BINARY_CMD_OBSERVE:
          return h->observe(cookie, request, response, docNamespace);
      case PROTOCOL_BINARY_CMD_OBSERVE_SEQNO:
          return h->observe_seqno(cookie, request, response);
      case PROTOCOL_BINARY_CMD_DEREGISTER_TAP_CLIENT: {
          rv = h->deregisterTapClient(cookie, request, response);
          h->decrementSessionCtr();
          return rv;
      }
      case PROTOCOL_BINARY_CMD_RESET_REPLICATION_CHAIN: {
          rv = h->resetReplicationChain(cookie, request, response);
          return rv;
      }
      case PROTOCOL_BINARY_CMD_CHANGE_VB_FILTER: {
          rv = h->changeTapVBFilter(cookie, request, response);
          h->decrementSessionCtr();
          return rv;
      }
      case PROTOCOL_BINARY_CMD_LAST_CLOSED_CHECKPOINT:
      case PROTOCOL_BINARY_CMD_CREATE_CHECKPOINT:
      case PROTOCOL_BINARY_CMD_CHECKPOINT_PERSISTENCE: {
          rv = h->handleCheckpointCmds(cookie, request, response);
          return rv;
      }
      case PROTOCOL_BINARY_CMD_SEQNO_PERSISTENCE: {
          rv = h->handleSeqnoCmds(cookie, request, response);
          return rv;
      }
      case PROTOCOL_BINARY_CMD_GET_META:
      case PROTOCOL_BINARY_CMD_GETQ_META: {
          rv = h->getMeta(cookie,
                          reinterpret_cast<protocol_binary_request_get_meta*>
                          (request), response,
                          docNamespace);
          return rv;
      }
      case PROTOCOL_BINARY_CMD_SET_WITH_META:
      case PROTOCOL_BINARY_CMD_SETQ_WITH_META:
      case PROTOCOL_BINARY_CMD_ADD_WITH_META:
      case PROTOCOL_BINARY_CMD_ADDQ_WITH_META: {
          rv = h->setWithMeta(cookie,
                              reinterpret_cast<protocol_binary_request_set_with_meta*>
                              (request), response,
                              docNamespace);
          return rv;
      }
      case PROTOCOL_BINARY_CMD_DEL_WITH_META:
      case PROTOCOL_BINARY_CMD_DELQ_WITH_META: {
          rv = h->deleteWithMeta(cookie,
                                 reinterpret_cast<protocol_binary_request_delete_with_meta*>
                                 (request), response,
                                 docNamespace);
          return rv;
      }
      case PROTOCOL_BINARY_CMD_RETURN_META: {
          return h->returnMeta(cookie,
                               reinterpret_cast<protocol_binary_request_return_meta*>
                               (request), response,
                               docNamespace);
      }
      case PROTOCOL_BINARY_CMD_GET_REPLICA:
          rv = getReplicaCmd(h, request, cookie, &itm, &msg, &res, docNamespace);
          if (rv != ENGINE_SUCCESS && rv != ENGINE_NOT_MY_VBUCKET) {
              return rv;
          }
          break;
      case PROTOCOL_BINARY_CMD_ENABLE_TRAFFIC:
      case PROTOCOL_BINARY_CMD_DISABLE_TRAFFIC: {
          rv = h->handleTrafficControlCmd(cookie, request, response);
          return rv;
      }
      case PROTOCOL_BINARY_CMD_SET_CLUSTER_CONFIG: {
          rv = h->setClusterConfig(cookie,
                                   reinterpret_cast<protocol_binary_request_set_cluster_config*>
                                   (request), response);
          h->decrementSessionCtr();
          return rv;
      }
      case PROTOCOL_BINARY_CMD_GET_CLUSTER_CONFIG:
          return h->getClusterConfig(cookie,
                                     reinterpret_cast<protocol_binary_request_get_cluster_config*>
                                     (request), response);
      case PROTOCOL_BINARY_CMD_COMPACT_DB: {
          rv = compactDB(h, cookie,
                         (protocol_binary_request_compact_db*)(request),
                         response);
          if (rv != ENGINE_EWOULDBLOCK) {
              h->decrementSessionCtr();
              h->storeEngineSpecific(cookie, NULL);
          }
          return rv;
      }
      case PROTOCOL_BINARY_CMD_GET_RANDOM_KEY: {
          if (request->request.extlen != 0 ||
              request->request.keylen != 0 ||
              request->request.bodylen != 0) {
              return ENGINE_EINVAL;
          }
          return h->getRandomKey(cookie, response);
      }
      case PROTOCOL_BINARY_CMD_GET_KEYS: {
          return h->getAllKeys(cookie,
                               reinterpret_cast<protocol_binary_request_get_keys*>
                               (request), response,
                               docNamespace);
      }
          // MB-21143: Remove adjusted time/drift API, but return NOT_SUPPORTED
      case PROTOCOL_BINARY_CMD_GET_ADJUSTED_TIME:
      case PROTOCOL_BINARY_CMD_SET_DRIFT_COUNTER_STATE: {
          return sendResponse(response, NULL, 0, NULL, 0, NULL, 0,
                              PROTOCOL_BINARY_RAW_BYTES,
                              PROTOCOL_BINARY_RESPONSE_NOT_SUPPORTED, 0,
                              cookie);
      }
      }
  
      if (itm) {
          uint32_t flags = itm->getFlags();
          rv = sendResponse(response,
                            static_cast<const void*>(itm->getKey().data()),
                            itm->getKey().size(),
                            (const void*)&flags, sizeof(uint32_t),
                            static_cast<const void*>(itm->getData()),
                            itm->getNBytes(), itm->getDataType(),
                            static_cast<uint16_t>(res), itm->getCas(),
                            cookie);
          delete itm;
      } else if (rv == ENGINE_NOT_MY_VBUCKET) {
          return h->sendNotMyVBucketResponse(response, cookie, 0);
      } else {
          msg_size = (msg_size > 0 || msg == NULL) ? msg_size : strlen(msg);
          rv = sendResponse(response, NULL, 0, NULL, 0,
                            msg, static_cast<uint16_t>(msg_size),
                            PROTOCOL_BINARY_RAW_BYTES,
                            static_cast<uint16_t>(res), 0, cookie);
  
      }
      return rv;
  }
  
  static ENGINE_ERROR_CODE EvpUnknownCommand(ENGINE_HANDLE* handle,
                                             const void* cookie,
                                             protocol_binary_request_header
                                             * request,
                                             ADD_RESPONSE response,
                                             DocNamespace doc_namespace) {
      auto engine = acquireEngine(handle);
      auto ret = processUnknownCommand(
              engine.get(), cookie, request, response, doc_namespace);
      return ret;
  }
  
  static void EvpItemSetCas(ENGINE_HANDLE*, const void*,
                            item* itm, uint64_t cas) {
      static_cast<Item*>(itm)->setCas(cas);
  }
  
  static ENGINE_ERROR_CODE EvpTapNotify(ENGINE_HANDLE* handle,
                                        const void* cookie,
                                        void* engine_specific,
                                        uint16_t nengine,
                                        uint8_t ttl,
                                        uint16_t tap_flags,
                                        tap_event_t tap_event,
                                        uint32_t tap_seqno,
                                        const void* key,
                                        size_t nkey,
                                        uint32_t flags,
                                        uint32_t exptime,
                                        uint64_t cas,
                                        uint8_t datatype,
                                        const void* data,
                                        size_t ndata,
                                        uint16_t vbucket) {
      if (!mcbp::datatype::is_valid(datatype)) {
          LOG(EXTENSION_LOG_WARNING, "Invalid value for datatype "
              " (TapNotify)");
          return ENGINE_EINVAL;
      }
  
      return acquireEngine(handle)->tapNotify(cookie,
                                              engine_specific,
                                              nengine,
                                              ttl,
                                              tap_flags,
                                              (uint16_t)tap_event,
                                              tap_seqno,
                                              key,
                                              nkey,
                                              flags,
                                              exptime,
                                              cas,
                                              datatype,
                                              data,
                                              ndata,
                                              vbucket);
  }
  
  static tap_event_t EvpTapIterator(ENGINE_HANDLE* handle,
                                    const void* cookie, item** itm,
                                    void** es, uint16_t* nes, uint8_t* ttl,
                                    uint16_t* flags, uint32_t* seqno,
                                    uint16_t* vbucket) {
      uint16_t tap_event = acquireEngine(handle)->walkTapQueue(
              cookie, itm, es, nes, ttl, flags, seqno, vbucket);
      return static_cast<tap_event_t>(tap_event);
  }
  
  static TAP_ITERATOR EvpGetTapIterator(ENGINE_HANDLE* handle,
                                        const void* cookie,
                                        const void* client,
                                        size_t nclient,
                                        uint32_t flags,
                                        const void* userdata,
                                        size_t nuserdata) {
      auto engine = acquireEngine(handle);
      TAP_ITERATOR iterator = NULL;
      {
          std::string c(static_cast<const char*>(client), nclient);
          // Figure out what we want from the userdata before adding it to
          // the API to the handle
          if (engine->createTapQueue(cookie, c, flags, userdata, nuserdata)) {
              iterator = EvpTapIterator;
          }
      }
      return iterator;
  }
  
  
  static ENGINE_ERROR_CODE EvpDcpStep(ENGINE_HANDLE* handle,
                                      const void* cookie,
                                      struct dcp_message_producers* producers) {
      auto engine = acquireEngine(handle);
      ConnHandler* conn = engine->getConnHandler(cookie);
      if (conn) {
          return conn->step(producers);
      }
      return ENGINE_DISCONNECT;
  }
  
  
  static ENGINE_ERROR_CODE EvpDcpOpen(ENGINE_HANDLE* handle,
                                      const void* cookie,
                                      uint32_t opaque,
                                      uint32_t seqno,
                                      uint32_t flags,
                                      void* name,
                                      uint16_t nname) {
      return acquireEngine(handle)->dcpOpen(
              cookie, opaque, seqno, flags, name, nname);
  }
  
  static ENGINE_ERROR_CODE EvpDcpAddStream(ENGINE_HANDLE* handle,
                                           const void* cookie,
                                           uint32_t opaque,
                                           uint16_t vbucket,
                                           uint32_t flags) {
      return acquireEngine(handle)->dcpAddStream(cookie, opaque, vbucket, flags);
  }
  
  static ENGINE_ERROR_CODE EvpDcpCloseStream(ENGINE_HANDLE* handle,
                                             const void* cookie,
                                             uint32_t opaque,
                                             uint16_t vbucket) {
      auto engine = acquireEngine(handle);
      ConnHandler* conn = engine->getConnHandler(cookie);
      if (conn) {
          return conn->closeStream(opaque, vbucket);
      }
      return ENGINE_DISCONNECT;
  }
  
  
  static ENGINE_ERROR_CODE EvpDcpStreamReq(ENGINE_HANDLE* handle,
                                           const void* cookie,
                                           uint32_t flags,
                                           uint32_t opaque,
                                           uint16_t vbucket,
                                           uint64_t startSeqno,
                                           uint64_t endSeqno,
                                           uint64_t vbucketUuid,
                                           uint64_t snapStartSeqno,
                                           uint64_t snapEndSeqno,
                                           uint64_t* rollbackSeqno,
                                           dcp_add_failover_log callback) {
      auto engine = acquireEngine(handle);
      ConnHandler* conn = engine->getConnHandler(cookie);
      if (conn) {
          return conn->streamRequest(flags,
                                     opaque,
                                     vbucket,
                                     startSeqno,
                                     endSeqno,
                                     vbucketUuid,
                                     snapStartSeqno,
                                     snapEndSeqno,
                                     rollbackSeqno,
                                     callback);
      }
      return ENGINE_DISCONNECT;
  }
  
  static ENGINE_ERROR_CODE EvpDcpGetFailoverLog(ENGINE_HANDLE* handle,
                                                const void* cookie,
                                                uint32_t opaque,
                                                uint16_t vbucket,
                                                dcp_add_failover_log callback) {
      auto engine = acquireEngine(handle);
      ConnHandler* conn = engine->getConnHandler(cookie);
      if (conn) {
          return conn->getFailoverLog(opaque, vbucket, callback);
      }
      return ENGINE_DISCONNECT;
  }
  
  
  static ENGINE_ERROR_CODE EvpDcpStreamEnd(ENGINE_HANDLE* handle,
                                           const void* cookie,
                                           uint32_t opaque,
                                           uint16_t vbucket,
                                           uint32_t flags) {
      auto engine = acquireEngine(handle);
      ConnHandler* conn = engine->getConnHandler(cookie);
      if (conn) {
          return conn->streamEnd(opaque, vbucket, flags);
      }
      return ENGINE_DISCONNECT;
  }
  
  
  static ENGINE_ERROR_CODE EvpDcpSnapshotMarker(ENGINE_HANDLE* handle,
                                                const void* cookie,
                                                uint32_t opaque,
                                                uint16_t vbucket,
                                                uint64_t start_seqno,
                                                uint64_t end_seqno,
                                                uint32_t flags) {
      auto engine = acquireEngine(handle);
      ConnHandler* conn = engine->getConnHandler(cookie);
      if (conn) {
          return conn->snapshotMarker(
                  opaque, vbucket, start_seqno, end_seqno, flags);
      }
      return ENGINE_DISCONNECT;
  }
  
  static ENGINE_ERROR_CODE EvpDcpMutation(ENGINE_HANDLE* handle,
                                          const void* cookie,
                                          uint32_t opaque,
                                          const void* key,
                                          uint16_t nkey,
                                          const void* value,
                                          uint32_t nvalue,
                                          uint64_t cas,
                                          uint16_t vbucket,
                                          uint32_t flags,
                                          uint8_t datatype,
                                          uint64_t bySeqno,
                                          uint64_t revSeqno,
                                          uint32_t expiration,
                                          uint32_t lockTime,
                                          const void* meta,
                                          uint16_t nmeta,
                                          uint8_t nru) {
      if (!mcbp::datatype::is_valid(datatype)) {
          LOG(EXTENSION_LOG_WARNING, "Invalid value for datatype "
              " (DCPMutation)");
          return ENGINE_EINVAL;
      }
      auto engine = acquireEngine(handle);
      ConnHandler* conn = engine->getConnHandler(cookie);
      if (conn) {
          return conn->mutation(opaque,
                                key,
                                nkey,
                                value,
                                nvalue,
                                cas,
                                vbucket,
                                flags,
                                datatype,
                                lockTime,
                                bySeqno,
                                revSeqno,
                                expiration,
                                nru,
                                meta,
                                nmeta);
      }
      return ENGINE_DISCONNECT;
  }
  
  static ENGINE_ERROR_CODE EvpDcpDeletion(ENGINE_HANDLE* handle,
                                          const void* cookie,
                                          uint32_t opaque,
                                          const void* key,
                                          uint16_t nkey,
                                          uint64_t cas,
                                          uint16_t vbucket,
                                          uint64_t bySeqno,
                                          uint64_t revSeqno,
                                          const void* meta,
                                          uint16_t nmeta) {
      auto engine = acquireEngine(handle);
      ConnHandler* conn = engine->getConnHandler(cookie);
      if (conn) {
          return conn->deletion(opaque,
                                key,
                                nkey,
                                cas,
                                vbucket,
                                bySeqno,
                                revSeqno,
                                meta,
                                nmeta);
      }
      return ENGINE_DISCONNECT;
  }
  
  static ENGINE_ERROR_CODE EvpDcpExpiration(ENGINE_HANDLE* handle,
                                            const void* cookie,
                                            uint32_t opaque,
                                            const void* key,
                                            uint16_t nkey,
                                            uint64_t cas,
                                            uint16_t vbucket,
                                            uint64_t bySeqno,
                                            uint64_t revSeqno,
                                            const void* meta,
                                            uint16_t nmeta) {
      auto engine = acquireEngine(handle);
      ConnHandler* conn = engine->getConnHandler(cookie);
      if (conn) {
          return conn->expiration(opaque,
                                  key,
                                  nkey,
                                  cas,
                                  vbucket,
                                  bySeqno,
                                  revSeqno,
                                  meta,
                                  nmeta);
      }
      return ENGINE_DISCONNECT;
  }
  
  static ENGINE_ERROR_CODE EvpDcpFlush(ENGINE_HANDLE* handle,
                                       const void* cookie,
                                       uint32_t opaque,
                                       uint16_t vbucket) {
      auto engine = acquireEngine(handle);
      ConnHandler* conn = engine->getConnHandler(cookie);
      if (conn) {
          return conn->flushall(opaque, vbucket);
      }
      return ENGINE_DISCONNECT;
  }
  
  static ENGINE_ERROR_CODE EvpDcpSetVbucketState(ENGINE_HANDLE* handle,
                                                 const void* cookie,
                                                 uint32_t opaque,
                                                 uint16_t vbucket,
                                                 vbucket_state_t state) {
      auto engine = acquireEngine(handle);
      ConnHandler* conn = engine->getConnHandler(cookie);
      if (conn) {
          return conn->setVBucketState(opaque, vbucket, state);
      }
      return ENGINE_DISCONNECT;
  }
  
  static ENGINE_ERROR_CODE EvpDcpNoop(ENGINE_HANDLE* handle,
                                      const void* cookie,
                                      uint32_t opaque) {
      auto engine = acquireEngine(handle);
      ConnHandler* conn = engine->getConnHandler(cookie);
      if (conn) {
          return conn->noop(opaque);
      }
      return ENGINE_DISCONNECT;
  }
  
  static ENGINE_ERROR_CODE EvpDcpBufferAcknowledgement(ENGINE_HANDLE* handle,
                                                       const void* cookie,
                                                       uint32_t opaque,
                                                       uint16_t vbucket,
                                                       uint32_t buffer_bytes) {
      auto engine = acquireEngine(handle);
      ConnHandler* conn = engine->getConnHandler(cookie);
      if (conn) {
          return conn->bufferAcknowledgement(opaque, vbucket, buffer_bytes);
      }
      return ENGINE_DISCONNECT;
  }
  
  static ENGINE_ERROR_CODE EvpDcpControl(ENGINE_HANDLE* handle,
                                         const void* cookie,
                                         uint32_t opaque,
                                         const void* key,
                                         uint16_t nkey,
                                         const void* value,
                                         uint32_t nvalue) {
      auto engine = acquireEngine(handle);
      ConnHandler* conn = engine->getConnHandler(cookie);
      if (conn) {
          return conn->control(opaque, key, nkey, value, nvalue);
      }
      return ENGINE_DISCONNECT;
  }
  
  static ENGINE_ERROR_CODE EvpDcpResponseHandler(ENGINE_HANDLE* handle,
                                                 const void* cookie,
                                                 protocol_binary_response_header* response) {
      auto engine = acquireEngine(handle);
      ConnHandler* conn = engine->getConnHandler(cookie);
      if (conn) {
          return conn->handleResponse(response);
      }
      return ENGINE_DISCONNECT;
  }
  
  static void EvpHandleDisconnect(const void* cookie,
                                  ENGINE_EVENT_TYPE type,
                                  const void* event_data,
                                  const void* cb_data) {
      if (type != ON_DISCONNECT) {
          throw std::invalid_argument("EvpHandleDisconnect: type "
                                          "(which is" + std::to_string(type) +
                                      ") is not ON_DISCONNECT");
      }
      if (event_data != nullptr) {
          throw std::invalid_argument("EvpHandleDisconnect: event_data "
                                          "is not NULL");
      }
      void* c = const_cast<void*>(cb_data);
      acquireEngine(static_cast<ENGINE_HANDLE*>(c))->handleDisconnect(cookie);
  }
  
  static void EvpHandleDeleteBucket(const void* cookie,
                                    ENGINE_EVENT_TYPE type,
                                    const void* event_data,
                                    const void* cb_data) {
      if (type != ON_DELETE_BUCKET) {
          throw std::invalid_argument("EvpHandleDeleteBucket: type "
                                          "(which is" + std::to_string(type) +
                                      ") is not ON_DELETE_BUCKET");
      }
      if (event_data != nullptr) {
          throw std::invalid_argument("EvpHandleDeleteBucket: event_data "
                                          "is not NULL");
      }
      void* c = const_cast<void*>(cb_data);
      acquireEngine(static_cast<ENGINE_HANDLE*>(c))->handleDeleteBucket(cookie);
  }
  
  void EvpSetLogLevel(ENGINE_HANDLE* handle, EXTENSION_LOG_LEVEL level) {
      Logger::setGlobalLogLevel(level);
  }
  
  /**
   * The only public interface to the eventually persistent engine.
   * Allocate a new instance and initialize it
   * @param interface the highest interface the server supports (we only
   *                  support interface 1)
   * @param get_server_api callback function to get the server exported API
   *                  functions
   * @param handle Where to return the new instance
   * @return ENGINE_SUCCESS on success
   */
  ENGINE_ERROR_CODE create_instance(uint64_t interface,
                                    GET_SERVER_API get_server_api,
                                    ENGINE_HANDLE** handle) {
      SERVER_HANDLE_V1* api = get_server_api();
      if (interface != 1 || api == NULL) {
          return ENGINE_ENOTSUP;
      }
  
      Logger::setLoggerAPI(api->log);
  
      MemoryTracker::getInstance(*api->alloc_hooks);
      ObjectRegistry::initialize(api->alloc_hooks->get_allocation_size);
  
      std::atomic<size_t>* inital_tracking = new std::atomic<size_t>();
  
      ObjectRegistry::setStats(inital_tracking);
      EventuallyPersistentEngine* engine;
      engine = new EventuallyPersistentEngine(get_server_api);
      ObjectRegistry::setStats(NULL);
  
      if (engine == NULL) {
          return ENGINE_ENOMEM;
      }
  
      if (MemoryTracker::trackingMemoryAllocations()) {
          engine->getEpStats().memoryTrackerEnabled.store(true);
          engine->getEpStats().totalMemory->store(inital_tracking->load());
      }
      delete inital_tracking;
  
      initialize_time_functions(api->core);
  
      *handle = reinterpret_cast<ENGINE_HANDLE*> (engine);
  
      return ENGINE_SUCCESS;
  }
  
  /*
      This method is called prior to unloading of the shared-object.
      Global clean-up should be performed from this method.
  */
  void destroy_engine() {
      ExecutorPool::shutdown();
      // A single MemoryTracker exists for *all* buckets
      // and must be destroyed before unloading the shared object.
      MemoryTracker::destroyInstance();
  }
  
  static bool EvpGetItemInfo(ENGINE_HANDLE* handle, const void*,
                             const item* itm, item_info* itm_info) {
      const Item* it = reinterpret_cast<const Item*>(itm);
      auto engine = acquireEngine(handle);
      itm_info->cas = it->getCas();
  
      if (engine) {
          RCPtr<VBucket> vb = engine->getKVBucket()->getVBucket(
              it->getVBucketId());
  
          if (vb) {
              itm_info->vbucket_uuid = vb->failovers->getLatestUUID();
          } else {
              itm_info->vbucket_uuid = 0;
          }
  
      } else {
          itm_info->vbucket_uuid = 0;
      }
  
      itm_info->seqno = it->getBySeqno();
      itm_info->exptime = it->getExptime();
      itm_info->nbytes = it->getNBytes();
      itm_info->datatype = it->getDataType();
      itm_info->flags = it->getFlags();
      itm_info->nkey = static_cast<uint16_t>(it->getKey().size());
      itm_info->key = it->getKey().data();
      itm_info->value[0].iov_base = const_cast<char*>(it->getData());
      itm_info->value[0].iov_len = it->getNBytes();
  
      if (it->isDeleted()) {
          itm_info->document_state = DocumentState::Deleted;
      } else {
          itm_info->document_state = DocumentState::Alive;
      }
      return true;
  }
  
  static bool EvpSetItemInfo(ENGINE_HANDLE* handle, const void* cookie,
                             item* itm, const item_info* itm_info) {
      Item* it = reinterpret_cast<Item*>(itm);
      if (!it) {
          return false;
      }
      it->setDataType(itm_info->datatype);
      return true;
  }
  
  static ENGINE_ERROR_CODE EvpGetClusterConfig(ENGINE_HANDLE* handle,
                                               const void* cookie,
                                               engine_get_vb_map_cb callback) {
      auto engine = acquireEngine(handle);
      LockHolder lh(engine->clusterConfig.lock);
      const char* config = engine->clusterConfig.config.data();
      uint32_t len = engine->clusterConfig.config.size();
      engine.reset(); // Want to release the engine before the callback
      return callback(cookie, config, len);
  }
  
  void LOG(EXTENSION_LOG_LEVEL severity, const char *fmt, ...) {
      va_list va;
      va_start(va, fmt);
      global_logger.vlog(severity, fmt, va);
      va_end(va);
  }
  
  
  EventuallyPersistentEngine::EventuallyPersistentEngine(
                                      GET_SERVER_API get_server_api) :
      clusterConfig(), kvBucket(nullptr), workload(NULL),
      workloadPriority(NO_BUCKET_PRIORITY),
      replicationThrottle(NULL), getServerApiFunc(get_server_api),
      dcpConnMap_(NULL),
      dcpFlowControlManager_(NULL),
      tapConnMap(NULL) ,
      tapConfig(NULL), checkpointConfig(NULL),
      trafficEnabled(false), flushAllEnabled(false), startupTime(0),
      taskable(this)
  {
      interface.interface = 1;
      ENGINE_HANDLE_V1::get_info = EvpGetInfo;
      ENGINE_HANDLE_V1::initialize = EvpInitialize;
      ENGINE_HANDLE_V1::destroy = EvpDestroy;
      ENGINE_HANDLE_V1::allocate = EvpItemAllocate;
      ENGINE_HANDLE_V1::allocate_ex = EvpItemAllocateEx;
      ENGINE_HANDLE_V1::remove = EvpItemDelete;
      ENGINE_HANDLE_V1::release = EvpItemRelease;
      ENGINE_HANDLE_V1::get = EvpGet;
      ENGINE_HANDLE_V1::get_locked = EvpGetLocked;
      ENGINE_HANDLE_V1::unlock = EvpUnlock;
      ENGINE_HANDLE_V1::get_stats = EvpGetStats;
      ENGINE_HANDLE_V1::reset_stats = EvpResetStats;
      ENGINE_HANDLE_V1::store = EvpStore;
      ENGINE_HANDLE_V1::flush = EvpFlush;
      ENGINE_HANDLE_V1::unknown_command = EvpUnknownCommand;
      ENGINE_HANDLE_V1::get_tap_iterator = EvpGetTapIterator;
      ENGINE_HANDLE_V1::tap_notify = EvpTapNotify;
      ENGINE_HANDLE_V1::item_set_cas = EvpItemSetCas;
      ENGINE_HANDLE_V1::get_item_info = EvpGetItemInfo;
      ENGINE_HANDLE_V1::set_item_info = EvpSetItemInfo;
      ENGINE_HANDLE_V1::get_engine_vb_map = EvpGetClusterConfig;
  
      ENGINE_HANDLE_V1::dcp.step = EvpDcpStep;
      ENGINE_HANDLE_V1::dcp.open = EvpDcpOpen;
      ENGINE_HANDLE_V1::dcp.add_stream = EvpDcpAddStream;
      ENGINE_HANDLE_V1::dcp.close_stream = EvpDcpCloseStream;
      ENGINE_HANDLE_V1::dcp.get_failover_log = EvpDcpGetFailoverLog;
      ENGINE_HANDLE_V1::dcp.stream_req = EvpDcpStreamReq;
      ENGINE_HANDLE_V1::dcp.stream_end = EvpDcpStreamEnd;
      ENGINE_HANDLE_V1::dcp.snapshot_marker = EvpDcpSnapshotMarker;
      ENGINE_HANDLE_V1::dcp.mutation = EvpDcpMutation;
      ENGINE_HANDLE_V1::dcp.deletion = EvpDcpDeletion;
      ENGINE_HANDLE_V1::dcp.expiration = EvpDcpExpiration;
      ENGINE_HANDLE_V1::dcp.flush = EvpDcpFlush;
      ENGINE_HANDLE_V1::dcp.set_vbucket_state = EvpDcpSetVbucketState;
      ENGINE_HANDLE_V1::dcp.noop = EvpDcpNoop;
      ENGINE_HANDLE_V1::dcp.buffer_acknowledgement = EvpDcpBufferAcknowledgement;
      ENGINE_HANDLE_V1::dcp.control = EvpDcpControl;
      ENGINE_HANDLE_V1::dcp.response_handler = EvpDcpResponseHandler;
      ENGINE_HANDLE_V1::set_log_level = EvpSetLogLevel;
  
      serverApi = getServerApiFunc();
      memset(&info, 0, sizeof(info));
      info.info.description = "EP engine v" VERSION;
      info.info.features[info.info.num_features++].feature = ENGINE_FEATURE_CAS;
      info.info.features[info.info.num_features++].feature =
                                               ENGINE_FEATURE_PERSISTENT_STORAGE;
      info.info.features[info.info.num_features++].feature = ENGINE_FEATURE_LRU;
      info.info.features[info.info.num_features++].feature = ENGINE_FEATURE_DATATYPE;
  }
  
  ENGINE_ERROR_CODE EventuallyPersistentEngine::reserveCookie(const void *cookie)
  {
      EventuallyPersistentEngine *epe =
                                      ObjectRegistry::onSwitchThread(NULL, true);
      ENGINE_ERROR_CODE rv = serverApi->cookie->reserve(cookie);
      ObjectRegistry::onSwitchThread(epe);
      return rv;
  }
  
  ENGINE_ERROR_CODE EventuallyPersistentEngine::releaseCookie(const void *cookie)
  {
      EventuallyPersistentEngine *epe =
                                      ObjectRegistry::onSwitchThread(NULL, true);
      ENGINE_ERROR_CODE rv = serverApi->cookie->release(cookie);
      ObjectRegistry::onSwitchThread(epe);
      return rv;
  }
  
  void EventuallyPersistentEngine::registerEngineCallback(ENGINE_EVENT_TYPE type,
                                                          EVENT_CALLBACK cb,
                                                          const void *cb_data) {
      EventuallyPersistentEngine *epe =
                                      ObjectRegistry::onSwitchThread(NULL, true);
      SERVER_CALLBACK_API *sapi = getServerApi()->callback;
      sapi->register_callback(reinterpret_cast<ENGINE_HANDLE*>(this),
                              type, cb, cb_data);
      ObjectRegistry::onSwitchThread(epe);
  }
  
  /**
   * A configuration value changed listener that responds to ep-engine
   * parameter changes by invoking engine-specific methods on
   * configuration change events.
   */
  class EpEngineValueChangeListener : public ValueChangedListener {
  public:
      EpEngineValueChangeListener(EventuallyPersistentEngine &e) : engine(e) {
          // EMPTY
      }
  
      virtual void sizeValueChanged(const std::string &key, size_t value) {
          if (key.compare("getl_max_timeout") == 0) {
              engine.setGetlMaxTimeout(value);
          } else if (key.compare("getl_default_timeout") == 0) {
              engine.setGetlDefaultTimeout(value);
          } else if (key.compare("max_item_size") == 0) {
              engine.setMaxItemSize(value);
          }
      }
  
      virtual void booleanValueChanged(const std::string &key, bool value) {
          if (key.compare("flushall_enabled") == 0) {
              engine.setFlushAll(value);
          }
      }
  private:
      EventuallyPersistentEngine &engine;
  };
  
  
  
  ENGINE_ERROR_CODE EventuallyPersistentEngine::initialize(const char* config) {
      resetStats();
      if (config != NULL) {
          if (!configuration.parseConfiguration(config, serverApi)) {
              LOG(EXTENSION_LOG_NOTICE, "Failed to parse the configuration config "
                  "during bucket initialization.  config=%s", config);
              return ENGINE_FAILED;
          }
      }
  
      name = configuration.getCouchBucket();
      maxFailoverEntries = configuration.getMaxFailoverEntries();
  
      // Start updating the variables from the config!
      HashTable::setDefaultNumBuckets(configuration.getHtSize());
      HashTable::setDefaultNumLocks(configuration.getHtLocks());
      StoredValue::setMutationMemoryThreshold(
                                        configuration.getMutationMemThreshold());
  
      if (configuration.getMaxSize() == 0) {
          configuration.setMaxSize(std::numeric_limits<size_t>::max());
      }
  
      if (configuration.getMemLowWat() == std::numeric_limits<size_t>::max()) {
          stats.mem_low_wat_percent.store(0.75);
          configuration.setMemLowWat(percentOf(
                  configuration.getMaxSize(), stats.mem_low_wat_percent.load()));
      }
  
      if (configuration.getMemHighWat() == std::numeric_limits<size_t>::max()) {
          stats.mem_high_wat_percent.store(0.85);
          configuration.setMemHighWat(percentOf(
                  configuration.getMaxSize(), stats.mem_high_wat_percent.load()));
      }
  
      maxItemSize = configuration.getMaxItemSize();
      configuration.addValueChangedListener("max_item_size",
                                         new EpEngineValueChangeListener(*this));
  
      getlDefaultTimeout = configuration.getGetlDefaultTimeout();
      configuration.addValueChangedListener("getl_default_timeout",
                                         new EpEngineValueChangeListener(*this));
      getlMaxTimeout = configuration.getGetlMaxTimeout();
      configuration.addValueChangedListener("getl_max_timeout",
                                         new EpEngineValueChangeListener(*this));
  
      flushAllEnabled = configuration.isFlushallEnabled();
      configuration.addValueChangedListener("flushall_enabled",
                                         new EpEngineValueChangeListener(*this));
  
      workload = new WorkLoadPolicy(configuration.getMaxNumWorkers(),
                                    configuration.getMaxNumShards());
      if ((unsigned int)workload->getNumShards() >
                                                configuration.getMaxVbuckets()) {
          LOG(EXTENSION_LOG_WARNING, "Invalid configuration: Shards must be "
              "equal or less than max number of vbuckets");
          return ENGINE_FAILED;
      }
  
      dcpConnMap_ = new DcpConnMap(*this);
  
      /* Get the flow control policy */
      std::string flowCtlPolicy = configuration.getDcpFlowControlPolicy();
  
      if (!flowCtlPolicy.compare("static")) {
          dcpFlowControlManager_ = new DcpFlowControlManagerStatic(*this);
      } else if (!flowCtlPolicy.compare("dynamic")) {
          dcpFlowControlManager_ = new DcpFlowControlManagerDynamic(*this);
      } else if (!flowCtlPolicy.compare("aggressive")) {
          dcpFlowControlManager_ = new DcpFlowControlManagerAggressive(*this);
      } else {
          /* Flow control is not enabled */
          dcpFlowControlManager_ = new DcpFlowControlManager(*this);
      }
  
      tapConnMap = new TapConnMap(*this);
      tapConfig = new TapConfig(*this);
      replicationThrottle = new ReplicationThrottle(configuration, stats);
      TapConfig::addConfigChangeListener(*this);
  
      checkpointConfig = new CheckpointConfig(*this);
      CheckpointConfig::addConfigChangeListener(*this);
  
      const std::string bucketType = configuration.getBucketType();
      if (bucketType == "persistent") {
          kvBucket = new EPBucket(*this);
      } else if (bucketType == "ephemeral") {
          // Disable warmup - it is not applicable to Ephemeral buckets.
          configuration.setWarmup(false);
          // Disable TAP - not supported for Ephemeral.
          configuration.setTap(false);
  
          kvBucket = new EphemeralBucket(*this);
      } else {
          throw std::invalid_argument(bucketType + " is not a recognized bucket "
                                      "type");
      }
  
      initializeEngineCallbacks();
  
      // Complete the initialization of the ep-store
      if (!kvBucket->initialize()) {
          return ENGINE_FAILED;
      }
  
      if(configuration.isDataTrafficEnabled()) {
          enableTraffic(true);
      }
  
      tapConnMap->initialize(TAP_CONN_NOTIFIER);
      dcpConnMap_->initialize(DCP_CONN_NOTIFIER);
  
      // record engine initialization time
      startupTime.store(ep_real_time());
  
      LOG(EXTENSION_LOG_NOTICE,
          "EP Engine: Initialization of %s bucket complete",
          bucketType.c_str());
  
      return ENGINE_SUCCESS;
  }
  
  void EventuallyPersistentEngine::destroy(bool force) {
      stats.forceShutdown = force;
      stats.isShutdown = true;
  
      if (kvBucket) {
          kvBucket->snapshotStats();
      }
      if (tapConnMap) {
          tapConnMap->shutdownAllConnections();
      }
      if (dcpConnMap_) {
          dcpConnMap_->shutdownAllConnections();
      }
  }
  
  ENGINE_ERROR_CODE EventuallyPersistentEngine::flush(const void *cookie){
      if (!flushAllEnabled) {
          return ENGINE_ENOTSUP;
      }
  
      if (!isDegradedMode()) {
          return ENGINE_TMPFAIL;
      }
  
      /*
       * Supporting only a SYNC operation for bucket flush
       */
  
      void* es = getEngineSpecific(cookie);
      if (es == NULL) {
  
          // Check if diskFlushAll was false and set it to true
          // if yes, if the atomic variable weren't false, then
          // we will assume that a flushAll has been scheduled
          // already and return TMPFAIL.
          if (kvBucket->scheduleFlushAllTask(cookie)) {
              storeEngineSpecific(cookie, this);
              return ENGINE_EWOULDBLOCK;
          } else {
              LOG(EXTENSION_LOG_INFO, "Tried to trigger a bucket flush, but"
                      "there seems to be a task running already!");
              return ENGINE_TMPFAIL;
          }
  
      } else {
          storeEngineSpecific(cookie, NULL);
          LOG(EXTENSION_LOG_NOTICE, "Completed bucket flush operation");
          return ENGINE_SUCCESS;
      }
  }
  
  ENGINE_ERROR_CODE EventuallyPersistentEngine::get_locked(const void* cookie,
                                                           item** itm,
                                                           const DocKey& key,
                                                           uint16_t vbucket,
                                                           uint32_t lock_timeout) {
  
      auto default_timeout = static_cast<uint32_t>(getGetlDefaultTimeout());
  
      if (lock_timeout == 0) {
          lock_timeout = default_timeout;
      } else if (lock_timeout > static_cast<uint32_t>(getGetlMaxTimeout())) {
          LOG(EXTENSION_LOG_WARNING,
              "EventuallyPersistentEngine::get_locked: "
              "Illegal value for lock timeout specified %u. "
              "Using default value: %u", lock_timeout, default_timeout);
          lock_timeout = default_timeout;
      }
  
      auto result = kvBucket->getLocked(key, vbucket, ep_current_time(),
                                        lock_timeout, cookie);
  
      if (result.getStatus() == ENGINE_SUCCESS) {
          ++stats.numOpsGet;
          *itm = result.getValue();
      }
  
      return result.getStatus();
  }
  
  ENGINE_ERROR_CODE EventuallyPersistentEngine::unlock(const void* cookie,
                                                       const DocKey& key,
                                                       uint16_t vbucket,
                                                       uint64_t cas) {
      return kvBucket->unlockKey(key, vbucket, cas, ep_current_time());
  }
  
  
  ENGINE_ERROR_CODE EventuallyPersistentEngine::store(const void *cookie,
                                                      item* itm,
                                                      uint64_t *cas,
                                                      ENGINE_STORE_OPERATION
                                                                       operation) {
      BlockTimer timer(&stats.storeCmdHisto);
      ENGINE_ERROR_CODE ret;
      Item *it = static_cast<Item*>(itm);
  
      switch (operation) {
      case OPERATION_CAS:
          if (it->getCas() == 0) {
              // Using a cas command with a cas wildcard doesn't make sense
              ret = ENGINE_NOT_STORED;
              break;
          }
          // FALLTHROUGH
      case OPERATION_SET:
          if (isDegradedMode()) {
              return ENGINE_TMPFAIL;
          }
          ret = kvBucket->set(*it, cookie);
          if (ret == ENGINE_SUCCESS) {
              *cas = it->getCas();
          }
  
          break;
  
      case OPERATION_ADD:
          if (isDegradedMode()) {
              return ENGINE_TMPFAIL;
          }
  
          if (it->getCas() != 0) {
              // Adding an item with a cas value doesn't really make sense...
              return ENGINE_KEY_EEXISTS;
          }
  
          ret = kvBucket->add(*it, cookie);
          if (ret == ENGINE_SUCCESS) {
              *cas = it->getCas();
          }
          break;
  
      case OPERATION_REPLACE:
          ret = kvBucket->replace(*it, cookie);
          if (ret == ENGINE_SUCCESS) {
              *cas = it->getCas();
          }
          break;
      default:
          ret = ENGINE_ENOTSUP;
      }
  
      switch (ret) {
      case ENGINE_SUCCESS:
          ++stats.numOpsStore;
          break;
      case ENGINE_ENOMEM:
          ret = memoryCondition();
          break;
      case ENGINE_NOT_STORED:
      case ENGINE_NOT_MY_VBUCKET:
          if (isDegradedMode()) {
              return ENGINE_TMPFAIL;
          }
          break;
      default:
          break;
      }
  
      return ret;
  }
  
  inline uint16_t EventuallyPersistentEngine::doWalkTapQueue(const void *cookie,
                                                             item **itm,
                                                             void **es,
                                                             uint16_t *nes,
                                                             uint8_t *ttl,
                                                             uint16_t *flags,
                                                             uint32_t *seqno,
                                                             uint16_t *vbucket,
                                                             TapProducer
                                                                     *connection,
                                                             bool &retry) {
      *es = NULL;
      *nes = 0;
      *ttl = (uint8_t)-1;
      *seqno = 0;
      *flags = 0;
      *vbucket = 0;
  
      retry = false;
  
      if (connection->shouldFlush()) {
          return TAP_FLUSH;
      }
  
      if (connection->isTimeForNoop()) {
          LOG(EXTENSION_LOG_INFO, "%s Sending a NOOP message.\n",
              connection->logHeader());
          return TAP_NOOP;
      }
  
      if (connection->isSuspended() || connection->windowIsFull()) {
          LOG(EXTENSION_LOG_INFO, "%s Connection in pause state because it is in"
              " suspended state or its ack windows is full.\n",
              connection->logHeader());
          return TAP_PAUSE;
      }
  
      uint16_t ret = TAP_PAUSE;
      VBucketEvent ev = connection->nextVBucketHighPriority();
      if (ev.event != TAP_PAUSE) {
          switch (ev.event) {
          case TAP_VBUCKET_SET:
              LOG(EXTENSION_LOG_NOTICE,
                 "%s Sending TAP_VBUCKET_SET with vbucket %d and state \"%s\"\n",
                  connection->logHeader(), ev.vbucket,
                  VBucket::toString(ev.state));
              connection->encodeVBucketStateTransition(ev, es, nes, vbucket);
              break;
          case TAP_OPAQUE:
              LOG(EXTENSION_LOG_NOTICE,
                  "%s Sending TAP_OPAQUE with command \"%s\" and vbucket %d\n",
                  connection->logHeader(),
                  TapProducer::opaqueCmdToString(ntohl((uint32_t) ev.state)),
                  ev.vbucket);
              connection->opaqueCommandCode = (uint32_t) ev.state;
              *vbucket = ev.vbucket;
              *es = &connection->opaqueCommandCode;
              *nes = sizeof(connection->opaqueCommandCode);
              break;
  
          default:
              throw std::logic_error("EventuallyPersistentEngine::doWalkTapQueue:"
                      " Unknown VBucketEvent message type:" +
                      std::to_string(ev.event) + " for connection:" +
                      connection->logHeader());
          }
          return ev.event;
      }
  
      if (connection->waitForOpaqueMsgAck()) {
          return TAP_PAUSE;
      }
  
      VBucketFilter backFillVBFilter;
      if (connection->runBackfill(backFillVBFilter)) {
          queueBackfill(backFillVBFilter, connection);
      }
  
      uint8_t nru = INITIAL_NRU_VALUE;
      Item *it = connection->getNextItem(cookie, vbucket, ret, nru);
      switch (ret) {
      case TAP_CHECKPOINT_START:
      case TAP_CHECKPOINT_END:
      case TAP_MUTATION:
      case TAP_DELETION:
          *itm = it;
          if (ret == TAP_MUTATION) {
              *nes = TapEngineSpecific::packSpecificData(ret, connection,
                                                         it->getRevSeqno(), nru);
              *es = connection->specificData;
          } else if (ret == TAP_DELETION) {
              *nes = TapEngineSpecific::packSpecificData(ret, connection,
                                                         it->getRevSeqno());
              *es = connection->specificData;
          } else if (ret == TAP_CHECKPOINT_START) {
              // Send the current value of the max deleted seqno
              RCPtr<VBucket> vb = getVBucket(*vbucket);
              if (!vb) {
                  retry = true;
                  return TAP_NOOP;
              }
              *nes = TapEngineSpecific::packSpecificData(ret, connection,
                                                 vb->ht.getMaxDeletedRevSeqno());
              *es = connection->specificData;
          }
          break;
      case TAP_NOOP:
          retry = true;
          break;
      default:
          break;
      }
  
      if (ret == TAP_PAUSE && (connection->dumpQueue || connection->doTakeOver)){
          VBucketEvent vbev = connection->checkDumpOrTakeOverCompletion();
          if (vbev.event == TAP_VBUCKET_SET) {
              LOG(EXTENSION_LOG_NOTICE,
                 "%s Sending TAP_VBUCKET_SET with vbucket %d and state \"%s\"\n",
                  connection->logHeader(), vbev.vbucket,
                  VBucket::toString(vbev.state));
              connection->encodeVBucketStateTransition(vbev, es, nes, vbucket);
          }
          ret = vbev.event;
      }
  
      return ret;
  }
  
  uint16_t EventuallyPersistentEngine::walkTapQueue(const void *cookie,
                                                    item **itm,
                                                    void **es,
                                                    uint16_t *nes,
                                                    uint8_t *ttl,
                                                    uint16_t *flags,
                                                    uint32_t *seqno,
                                                    uint16_t *vbucket) {
      TapProducer *connection = getTapProducer(cookie);
      if (!connection) {
          LOG(EXTENSION_LOG_WARNING,
              "Failed to lookup TAP connection.. Disconnecting\n");
          return TAP_DISCONNECT;
      }
  
      connection->setPaused(false);
  
      bool retry = false;
      uint16_t ret;
  
      connection->setLastWalkTime();
      do {
          ret = doWalkTapQueue(cookie, itm, es, nes, ttl, flags,
                               seqno, vbucket, connection, retry);
      } while (retry);
  
      if (ret != TAP_PAUSE && ret != TAP_DISCONNECT) {
          connection->lastMsgTime = ep_current_time();
          if (ret == TAP_NOOP) {
              *seqno = 0;
          } else {
              ++stats.numTapFetched;
              *seqno = connection->getSeqno();
              if (connection->requestAck(ret, *vbucket)) {
                  *flags = TAP_FLAG_ACK;
                  connection->seqnoAckRequested = *seqno;
              }
  
              if (ret == TAP_MUTATION) {
                  if (connection->haveFlagByteorderSupport()) {
                      *flags |= TAP_FLAG_NETWORK_BYTE_ORDER;
                  }
              }
          }
      } else {
          connection->setPaused(true);
          connection->setNotifySent(false);
      }
  
      return ret;
  }
  
  bool EventuallyPersistentEngine::createTapQueue(const void *cookie,
                                                  std::string &client,
                                                  uint32_t flags,
                                                  const void *userdata,
                                                  size_t nuserdata) {
      if (reserveCookie(cookie) != ENGINE_SUCCESS) {
          return false;
      }
  
      std::string tapName = "eq_tapq:";
      if (client.length() == 0) {
          tapName.assign(ConnHandler::getAnonName());
      } else {
          tapName.append(client);
      }
  
      // Decoding the userdata section of the packet and update the filters
      const char *ptr = static_cast<const char*>(userdata);
      uint64_t backfillAge = 0;
      std::vector<uint16_t> vbuckets;
      std::map<uint16_t, uint64_t> lastCheckpointIds;
  
      if (flags & TAP_CONNECT_FLAG_BACKFILL) { /* */
          if (nuserdata < sizeof(backfillAge)) {
              LOG(EXTENSION_LOG_WARNING,
                  "Backfill age is missing. Reject connection request from %s\n",
                  tapName.c_str());
              return false;
          }
          // use memcpy to avoid alignemt issues
          memcpy(&backfillAge, ptr, sizeof(backfillAge));
          backfillAge = ntohll(backfillAge);
          nuserdata -= sizeof(backfillAge);
          ptr += sizeof(backfillAge);
      }
  
      if (flags & TAP_CONNECT_FLAG_LIST_VBUCKETS) {
          uint16_t nvbuckets;
          if (nuserdata < sizeof(nvbuckets)) {
              LOG(EXTENSION_LOG_WARNING,
              "Number of vbuckets is missing. Reject connection request from %s"
              "\n", tapName.c_str());
              return false;
          }
          memcpy(&nvbuckets, ptr, sizeof(nvbuckets));
          nuserdata -= sizeof(nvbuckets);
          ptr += sizeof(nvbuckets);
          nvbuckets = ntohs(nvbuckets);
          if (nvbuckets > 0) {
              if (nuserdata < (sizeof(uint16_t) * nvbuckets)) {
                  LOG(EXTENSION_LOG_WARNING,
                  "# of vbuckets not matched. Reject connection request from %s"
                  "\n", tapName.c_str());
                  return false;
              }
              for (uint16_t ii = 0; ii < nvbuckets; ++ii) {
                  uint16_t val;
                  memcpy(&val, ptr, sizeof(nvbuckets));
                  ptr += sizeof(uint16_t);
                  vbuckets.push_back(ntohs(val));
              }
              nuserdata -= (sizeof(uint16_t) * nvbuckets);
          }
      }
  
      if (flags & TAP_CONNECT_CHECKPOINT) {
          uint16_t nCheckpoints = 0;
          if (nuserdata >= sizeof(nCheckpoints)) {
              memcpy(&nCheckpoints, ptr, sizeof(nCheckpoints));
              nuserdata -= sizeof(nCheckpoints);
              ptr += sizeof(nCheckpoints);
              nCheckpoints = ntohs(nCheckpoints);
          }
          if (nCheckpoints > 0) {
              if (nuserdata <
                  ((sizeof(uint16_t) + sizeof(uint64_t)) * nCheckpoints)) {
                  LOG(EXTENSION_LOG_WARNING, "# of checkpoint Ids not matched. "
                      "Reject connection request from %s\n", tapName.c_str());
                  return false;
              }
              for (uint16_t j = 0; j < nCheckpoints; ++j) {
                  uint16_t vbid;
                  uint64_t checkpointId;
                  memcpy(&vbid, ptr, sizeof(vbid));
                  ptr += sizeof(uint16_t);
                  memcpy(&checkpointId, ptr, sizeof(checkpointId));
                  ptr += sizeof(uint64_t);
                  lastCheckpointIds[ntohs(vbid)] = ntohll(checkpointId);
              }
          }
      }
  
      TapProducer *tp = tapConnMap->newProducer(cookie, tapName, flags,
                                   backfillAge,
                                   static_cast<int>(
                                   configuration.getTapKeepalive()),
                                   vbuckets,
                                   lastCheckpointIds);
  
      tapConnMap->notifyPausedConnection(tp, true);
      return true;
  }
  
  ENGINE_ERROR_CODE EventuallyPersistentEngine::tapNotify(const void *cookie,
                                                          void *engine_specific,
                                                          uint16_t nengine,
                                                          uint8_t ttl,
                                                          uint16_t tap_flags,
                                                          uint16_t tap_event,
                                                          uint32_t tap_seqno,
                                                          const void *key,
                                                          size_t nkey,
                                                          uint32_t flags,
                                                          uint32_t exptime,
                                                          uint64_t cas,
                                                          uint8_t datatype,
                                                          const void *data,
                                                          size_t ndata,
                                                          uint16_t vbucket)
  {
      (void) ttl;
      void *specific = getEngineSpecific(cookie);
      ConnHandler *connection = NULL;
      if (specific == NULL) {
          if (tap_event == TAP_ACK) {
              LOG(EXTENSION_LOG_WARNING, "Tap producer with cookie %s does not "
                  "exist. Force disconnect...\n", (char *) cookie);
              // tap producer is no longer connected..
              return ENGINE_DISCONNECT;
          } else {
              connection = tapConnMap->newConsumer(cookie);
              if (connection == NULL) {
                  LOG(EXTENSION_LOG_WARNING, "Failed to create new tap consumer."
                      " Force disconnect\n");
                  return ENGINE_DISCONNECT;
              }
              storeEngineSpecific(cookie, connection);
          }
      } else {
          connection = reinterpret_cast<ConnHandler *>(specific);
      }
  
  
      ENGINE_ERROR_CODE ret = ENGINE_SUCCESS;
  
      if (tap_event == TAP_MUTATION || tap_event == TAP_DELETION) {
          if (!replicationThrottle->shouldProcess()) {
              ++stats.replicationThrottled;
              if (connection->supportsAck()) {
                  ret = ENGINE_TMPFAIL;
              } else {
                  ret = ENGINE_DISCONNECT;
                  LOG(EXTENSION_LOG_WARNING, "%s Can't throttle streams without "
                      "ack support. Force disconnect...\n",
                      connection->logHeader());
              }
              return ret;
          }
      }
  
      switch (tap_event) {
      case TAP_ACK:
          // TAP works only with the DefaultCollection
          ret = processTapAck(cookie, tap_seqno, tap_flags,
                              DocKey(static_cast<const uint8_t*>(key), nkey,
                                     DocNamespace::DefaultCollection));
          break;
      case TAP_FLUSH:
          ret = flush(cookie);
          LOG(EXTENSION_LOG_NOTICE, "%s Received flush.\n",
              connection->logHeader());
          break;
      case TAP_DELETION:
          {
              uint64_t revSeqno;
              TapEngineSpecific::readSpecificData(tap_event, engine_specific,
                                                  nengine, &revSeqno);
  
              ret = connection->deletion(0, key, nkey, cas, vbucket, 0, revSeqno,
                                         NULL, 0);
          }
          break;
  
      case TAP_CHECKPOINT_START:
      case TAP_CHECKPOINT_END:
          {
              TapConsumer *tc = dynamic_cast<TapConsumer*>(connection);
              if (tc) {
                  if (tap_event == TAP_CHECKPOINT_START &&
                      nengine == TapEngineSpecific::sizeRevSeqno) {
                      // Set the current value for the max deleted seqno
                      RCPtr<VBucket> vb = getVBucket(vbucket);
                      if (!vb) {
                          return ENGINE_TMPFAIL;
                      }
                      uint64_t seqnum;
                      TapEngineSpecific::readSpecificData(tap_event,
                                                          engine_specific,
                                                          nengine,
                                                          &seqnum);
                      vb->ht.setMaxDeletedRevSeqno(seqnum);
                  }
  
                  if (data) {
                      uint64_t checkpointId;
                      memcpy(&checkpointId, data, sizeof(checkpointId));
                      checkpointId = ntohll(checkpointId);
                      ConnHandlerCheckPoint(tc, tap_event, vbucket,
                                            checkpointId);
                  }
                  else {
                      ret = ENGINE_DISCONNECT;
                      LOG(EXTENSION_LOG_WARNING,
                          "%s Checkpoint Id is missing in "
                          "CHECKPOINT messages. Force disconnect...\n",
                          connection->logHeader());
                  }
              }
              else {
                  ret = ENGINE_DISCONNECT;
                  LOG(EXTENSION_LOG_WARNING,
                      "%s not a consumer! Force disconnect\n",
                      connection->logHeader());
              }
          }
  
          break;
  
      case TAP_MUTATION:
          {
              uint8_t nru = INITIAL_NRU_VALUE;
              uint64_t revSeqno = 0;
              TapEngineSpecific::readSpecificData(tap_event, engine_specific,
                                                  nengine, &revSeqno, &nru);
  
              if (!isDatatypeSupported(cookie)) {
                  datatype = PROTOCOL_BINARY_RAW_BYTES;
                  const unsigned char *dat = (const unsigned char*)data;
                  const int datlen = ndata;
                  if (checkUTF8JSON(dat, datlen)) {
                      datatype = PROTOCOL_BINARY_DATATYPE_JSON;
                  }
              }
              ret = connection->mutation(0, key, nkey, data, ndata, cas, vbucket,
                                         flags, datatype, 0, 0, revSeqno, exptime,
                                         nru, NULL, 0);
          }
  
          break;
  
      case TAP_OPAQUE:
          if (nengine == sizeof(uint32_t)) {
              uint32_t cc;
              memcpy(&cc, engine_specific, sizeof(cc));
              cc = ntohl(cc);
  
              switch (cc) {
              case TAP_OPAQUE_ENABLE_AUTO_NACK:
                  // @todo: the memcached core will _ALWAYS_ send nack
                  //        if it encounter an error. This should be
                  // set as the default when we move to .next after 2.0
                  // (currently we need to allow the message for
                  // backwards compatibility)
                  LOG(EXTENSION_LOG_INFO, "%s Enable auto nack mode\n",
                      connection->logHeader());
                  break;
              case TAP_OPAQUE_ENABLE_CHECKPOINT_SYNC:
                  connection->setSupportCheckpointSync(true);
                  LOG(EXTENSION_LOG_INFO,
                      "%s Enable checkpoint synchronization\n",
                      connection->logHeader());
                  break;
              case TAP_OPAQUE_OPEN_CHECKPOINT:
                  /**
                   * This event is only received by the TAP client that wants to
                   * get mutations from closed checkpoints only. At this time,
                   * only incremental backup client receives this event so that
                   * it can close the connection and reconnect later.
                   */
                  LOG(EXTENSION_LOG_INFO, "%s Beginning of checkpoint.\n",
                      connection->logHeader());
                  break;
              case TAP_OPAQUE_INITIAL_VBUCKET_STREAM:
                  {
                      LOG(EXTENSION_LOG_INFO,
                          "%s Backfill started for vbucket %d.\n",
                          connection->logHeader(), vbucket);
                      BlockTimer timer(&stats.tapVbucketResetHisto);
                      ret = resetVBucket(vbucket) ? ENGINE_SUCCESS :
                                                    ENGINE_DISCONNECT;
                      if (ret == ENGINE_DISCONNECT) {
                          LOG(EXTENSION_LOG_WARNING,
                           "%s Failed to reset a vbucket %d. Force disconnect\n",
                              connection->logHeader(), vbucket);
                      } else {
                          LOG(EXTENSION_LOG_NOTICE,
                           "%s Reset vbucket %d was completed succecssfully.\n",
                              connection->logHeader(), vbucket);
                      }
  
                      TapConsumer *tc = dynamic_cast<TapConsumer*>(connection);
                      if (tc) {
                          tc->setBackfillPhase(true, vbucket);
                      } else {
                          ret = ENGINE_DISCONNECT;
                          LOG(EXTENSION_LOG_WARNING,
                              "TAP consumer doesn't exists. Force disconnect\n");
                      }
                  }
                  break;
              case TAP_OPAQUE_CLOSE_BACKFILL:
                  {
                      LOG(EXTENSION_LOG_INFO, "%s Backfill finished.\n",
                          connection->logHeader());
                      TapConsumer *tc = dynamic_cast<TapConsumer*>(connection);
                      if (tc) {
                          tc->setBackfillPhase(false, vbucket);
                      } else {
                          ret = ENGINE_DISCONNECT;
                          LOG(EXTENSION_LOG_WARNING,
                              "%s not a consumer! Force disconnect\n",
                              connection->logHeader());
                      }
                  }
                  break;
              case TAP_OPAQUE_CLOSE_TAP_STREAM:
                  /**
                   * This event is sent by the eVBucketMigrator to notify that
                   * the source node closes the tap replication stream and
                   * switches to TAKEOVER_VBUCKETS phase.
                   * This is just an informative message and doesn't require any
                   * action.
                   */
                  LOG(EXTENSION_LOG_INFO,
                  "%s Received close tap stream. Switching to takeover phase.\n",
                      connection->logHeader());
                  break;
              case TAP_OPAQUE_COMPLETE_VB_FILTER_CHANGE:
                  /**
                   * This opaque message is just for notifying that the source
                   * node receives change_vbucket_filter request and processes
                   * it successfully.
                   */
                  LOG(EXTENSION_LOG_INFO,
                  "%s Notified that the source node changed a vbucket filter.\n",
                      connection->logHeader());
                  break;
              default:
                  LOG(EXTENSION_LOG_WARNING,
                      "%s Received an unknown opaque command\n",
                      connection->logHeader());
              }
          } else {
              LOG(EXTENSION_LOG_WARNING,
                  "%s Received tap opaque with unknown size %d\n",
                  connection->logHeader(), nengine);
          }
          break;
  
      case TAP_VBUCKET_SET:
          {
              BlockTimer timer(&stats.tapVbucketSetHisto);
  
              if (nengine != sizeof(vbucket_state_t)) {
                  // illegal datasize
                  LOG(EXTENSION_LOG_WARNING,
                      "%s Received TAP_VBUCKET_SET with illegal size."
                      " Force disconnect\n", connection->logHeader());
                  ret = ENGINE_DISCONNECT;
                  break;
              }
  
              vbucket_state_t state;
              memcpy(&state, engine_specific, nengine);
              state = (vbucket_state_t)ntohl(state);
  
              ret = connection->setVBucketState(0, vbucket, state);
          }
          break;
  
      default:
          // Unknown command
          LOG(EXTENSION_LOG_WARNING,
              "%s Recieved bad opcode, ignoring message\n",
              connection->logHeader());
      }
  
      connection->processedEvent(tap_event, ret);
      return ret;
  }
  
  ENGINE_ERROR_CODE EventuallyPersistentEngine::ConnHandlerCheckPoint(
                                                        TapConsumer *consumer,
                                                        uint8_t event,
                                                        uint16_t vbucket,
                                                        uint64_t checkpointId) {
      ENGINE_ERROR_CODE ret = ENGINE_SUCCESS;
  
      if (consumer->processCheckpointCommand(event, vbucket, checkpointId)) {
          getKVBucket()->wakeUpFlusher();
          ret = ENGINE_SUCCESS;
      }
      else {
          ret = ENGINE_DISCONNECT;
          LOG(EXTENSION_LOG_WARNING, "%s Error processing "
              "checkpoint %" PRIu64 ". Force disconnect\n",
              consumer->logHeader(), checkpointId);
      }
  
      return ret;
  }
  
  TapProducer* EventuallyPersistentEngine::getTapProducer(const void *cookie) {
      TapProducer *rv =
          reinterpret_cast<TapProducer*>(getEngineSpecific(cookie));
      if (!(rv && rv->isConnected())) {
          LOG(EXTENSION_LOG_WARNING,
              "Walking a non-existent tap queue, disconnecting\n");
          return NULL;
      }
  
      if (rv->doDisconnect()) {
          LOG(EXTENSION_LOG_WARNING,
              "%s Disconnecting pending connection\n", rv->logHeader());
          return NULL;
      }
      return rv;
  }
  
  void EventuallyPersistentEngine::initializeEngineCallbacks() {
      // Register the ON_DISCONNECT callback
      registerEngineCallback(ON_DISCONNECT, EvpHandleDisconnect, this);
      // Register the ON_DELETE_BUCKET callback
      registerEngineCallback(ON_DELETE_BUCKET, EvpHandleDeleteBucket, this);
  }
  
  ENGINE_ERROR_CODE EventuallyPersistentEngine::processTapAck(const void *cookie,
                                                              uint32_t seqno,
                                                              uint16_t status,
                                                              const DocKey& key)
  {
      TapProducer *connection = getTapProducer(cookie);
      if (!connection) {
          LOG(EXTENSION_LOG_WARNING,
              "Unable to process tap ack. No producer found\n");
          return ENGINE_DISCONNECT;
      }
  
      return connection->processAck(seqno, status, key);
  }
  
  void EventuallyPersistentEngine::queueBackfill(const VBucketFilter
                                                               &backfillVBFilter,
                                                 Producer *tc)
  {
      auto bfv = std::make_unique<BackFillVisitor>(
              this, *tapConnMap, tc, backfillVBFilter);
      getKVBucket()->visit(std::move(bfv),
                           "Backfill task",
                           NONIO_TASK_IDX,
                           TaskId::BackfillVisitorTask,
                           1);
  }
  
  void VBucketCountVisitor::visitBucket(RCPtr<VBucket> &vb) {
      ++numVbucket;
      item_eviction_policy_t policy = engine.getKVBucket()->
                                                         getItemEvictionPolicy();
      numItems += vb->getNumItems(policy);
      numTempItems += vb->getNumTempItems();
      nonResident += vb->getNumNonResidentItems(policy);
  
      if (vb->getHighPriorityChkSize() > 0) {
          chkPersistRemaining++;
      }
  
      if (desired_state != vbucket_state_dead) {
          htMemory += vb->ht.memorySize();
          htItemMemory += vb->ht.getItemMemory();
          htCacheSize += vb->ht.cacheSize;
          numEjects += vb->ht.getNumEjects();
          numExpiredItems += vb->numExpiredItems;
          metaDataMemory += vb->ht.metaDataMemory;
          metaDataDisk += vb->metaDataDisk;
          opsCreate += vb->opsCreate;
          opsUpdate += vb->opsUpdate;
          opsDelete += vb->opsDelete;
          opsReject += vb->opsReject;
  
          queueSize += vb->dirtyQueueSize;
          queueMemory += vb->dirtyQueueMem;
          queueFill += vb->dirtyQueueFill;
          queueDrain += vb->dirtyQueueDrain;
          queueAge += vb->getQueueAge();
          pendingWrites += vb->dirtyQueuePendingWrites;
          rollbackItemCount += vb->getRollbackItemCount();
  
          /*
           * The bucket stat reports the total drift of the vbuckets.
           */
          auto absHLCDrift = vb->getHLCDriftStats();
          totalAbsHLCDrift.total += absHLCDrift.total;
          totalAbsHLCDrift.updates += absHLCDrift.updates;
  
          /*
           * Total up the exceptions
           */
          auto driftExceptionCounters = vb->getHLCDriftExceptionCounters();
          totalHLCDriftExceptionCounters.ahead += driftExceptionCounters.ahead;
          totalHLCDriftExceptionCounters.behind += driftExceptionCounters.behind;
      }
  }
  
  void VBucketCountAggregator::visitBucket(RCPtr<VBucket> &vb) {
      std::map<vbucket_state_t, VBucketCountVisitor*>::iterator it;
      it = visitorMap.find(vb->getState());
      if ( it != visitorMap.end() ) {
          it->second->visitBucket(vb);
      }
  }
  
  void VBucketCountAggregator::addVisitor(VBucketCountVisitor* visitor) {
      visitorMap[visitor->getVBucketState()] = visitor;
  }
  
  ENGINE_ERROR_CODE EventuallyPersistentEngine::doEngineStats(const void *cookie,
                                                             ADD_STAT add_stat) {
      VBucketCountAggregator aggregator;
  
      VBucketCountVisitor activeCountVisitor(*this, vbucket_state_active);
      aggregator.addVisitor(&activeCountVisitor);
  
      VBucketCountVisitor replicaCountVisitor(*this, vbucket_state_replica);
      aggregator.addVisitor(&replicaCountVisitor);
  
      VBucketCountVisitor pendingCountVisitor(*this, vbucket_state_pending);
      aggregator.addVisitor(&pendingCountVisitor);
  
      VBucketCountVisitor deadCountVisitor(*this, vbucket_state_dead);
      aggregator.addVisitor(&deadCountVisitor);
  
      kvBucket->visit(aggregator);
  
      kvBucket->updateCachedResidentRatio(
                                      activeCountVisitor.getMemResidentPer(),
                                      replicaCountVisitor.getMemResidentPer());
      replicationThrottle->adjustWriteQueueCap(activeCountVisitor.getNumItems() +
                                       replicaCountVisitor.getNumItems() +
                                       pendingCountVisitor.getNumItems());
  
      configuration.addStats(add_stat, cookie);
  
      EPStats &epstats = getEpStats();
      add_casted_stat("ep_version", VERSION, add_stat, cookie);
      add_casted_stat("ep_storage_age",
                      epstats.dirtyAge, add_stat, cookie);
      add_casted_stat("ep_storage_age_highwat",
                      epstats.dirtyAgeHighWat, add_stat, cookie);
      add_casted_stat("ep_num_workers", ExecutorPool::get()->getNumWorkersStat(),
                      add_stat, cookie);
  
      if (getWorkloadPriority() == HIGH_BUCKET_PRIORITY) {
          add_casted_stat("ep_bucket_priority", "HIGH", add_stat, cookie);
      } else if (getWorkloadPriority() == LOW_BUCKET_PRIORITY) {
          add_casted_stat("ep_bucket_priority", "LOW", add_stat, cookie);
      }
  
      add_casted_stat("ep_total_enqueued",
                      epstats.totalEnqueued, add_stat, cookie);
      add_casted_stat("ep_expired_access", epstats.expired_access,
                      add_stat, cookie);
      add_casted_stat("ep_expired_compactor", epstats.expired_compactor,
                      add_stat, cookie);
      add_casted_stat("ep_expired_pager", epstats.expired_pager,
                      add_stat, cookie);
      add_casted_stat("ep_queue_size",
                      epstats.diskQueueSize, add_stat, cookie);
      add_casted_stat("ep_diskqueue_items",
                      epstats.diskQueueSize, add_stat, cookie);
      auto* flusher = kvBucket->getFlusher(EP_PRIMARY_SHARD);
      if (flusher) {
          add_casted_stat("ep_commit_num", epstats.flusherCommits,
                          add_stat, cookie);
          add_casted_stat("ep_commit_time",
                          epstats.commit_time, add_stat, cookie);
          add_casted_stat("ep_commit_time_total",
                          epstats.cumulativeCommitTime, add_stat, cookie);
          add_casted_stat("ep_item_begin_failed",
                          epstats.beginFailed, add_stat, cookie);
          add_casted_stat("ep_item_commit_failed",
                          epstats.commitFailed, add_stat, cookie);
          add_casted_stat("ep_item_flush_expired",
                          epstats.flushExpired, add_stat, cookie);
          add_casted_stat("ep_item_flush_failed",
                          epstats.flushFailed, add_stat, cookie);
          add_casted_stat("ep_flusher_state",
                          flusher->stateName(), add_stat, cookie);
          add_casted_stat("ep_flusher_todo",
                          epstats.flusher_todo, add_stat, cookie);
          add_casted_stat("ep_total_persisted",
                          epstats.totalPersisted, add_stat, cookie);
          add_casted_stat("ep_uncommitted_items",
                          epstats.flusher_todo, add_stat, cookie);
      }
      add_casted_stat("ep_vbucket_del",
                      epstats.vbucketDeletions, add_stat, cookie);
      add_casted_stat("ep_vbucket_del_fail",
                      epstats.vbucketDeletionFail, add_stat, cookie);
      add_casted_stat("ep_flush_duration_total",
                      epstats.cumulativeFlushTime, add_stat, cookie);
      add_casted_stat("ep_flush_all", kvBucket->isFlushAllScheduled(), add_stat,
                      cookie);
      add_casted_stat("curr_items", activeCountVisitor.getNumItems(), add_stat,
                      cookie);
      add_casted_stat("curr_temp_items", activeCountVisitor.getNumTempItems(),
                      add_stat, cookie);
      add_casted_stat("curr_items_tot",
                      activeCountVisitor.getNumItems() +
                      replicaCountVisitor.getNumItems() +
                      pendingCountVisitor.getNumItems(),
                      add_stat, cookie);
      add_casted_stat("vb_active_num", activeCountVisitor.getVBucketNumber(),
                      add_stat, cookie);
      add_casted_stat("vb_active_curr_items", activeCountVisitor.getNumItems(),
                      add_stat, cookie);
      add_casted_stat("vb_active_num_non_resident",
                      activeCountVisitor.getNonResident(),
                      add_stat, cookie);
      add_casted_stat("vb_active_perc_mem_resident",
                      activeCountVisitor.getMemResidentPer(),
                      add_stat, cookie);
      add_casted_stat("vb_active_eject", activeCountVisitor.getEjects(),
                      add_stat, cookie);
      add_casted_stat("vb_active_expired", activeCountVisitor.getExpired(),
                      add_stat, cookie);
      add_casted_stat("vb_active_meta_data_memory",
                      activeCountVisitor.getMetaDataMemory(),
                      add_stat, cookie);
      add_casted_stat("vb_active_meta_data_disk",
                      activeCountVisitor.getMetaDataDisk(),
                      add_stat, cookie);
      add_casted_stat("vb_active_ht_memory",
                      activeCountVisitor.getHashtableMemory(),
                      add_stat, cookie);
      add_casted_stat("vb_active_itm_memory", activeCountVisitor.getItemMemory(),
                      add_stat, cookie);
      add_casted_stat("vb_active_ops_create", activeCountVisitor.getOpsCreate(),
                      add_stat, cookie);
      add_casted_stat("vb_active_ops_update", activeCountVisitor.getOpsUpdate(),
                      add_stat, cookie);
      add_casted_stat("vb_active_ops_delete", activeCountVisitor.getOpsDelete(),
                      add_stat, cookie);
      add_casted_stat("vb_active_ops_reject", activeCountVisitor.getOpsReject(),
                      add_stat, cookie);
      add_casted_stat("vb_active_queue_size", activeCountVisitor.getQueueSize(),
                      add_stat, cookie);
      add_casted_stat("vb_active_queue_memory",
                      activeCountVisitor.getQueueMemory(), add_stat, cookie);
      add_casted_stat("vb_active_queue_age", activeCountVisitor.getAge(),
                      add_stat, cookie);
      add_casted_stat("vb_active_queue_pending",
                      activeCountVisitor.getPendingWrites(), add_stat, cookie);
      add_casted_stat("vb_active_queue_fill", activeCountVisitor.getQueueFill(),
                      add_stat, cookie);
      add_casted_stat("vb_active_queue_drain",
                      activeCountVisitor.getQueueDrain(), add_stat, cookie);
      add_casted_stat("vb_active_rollback_item_count",
                      activeCountVisitor.getRollbackItemCount(),
                      add_stat, cookie);
  
      add_casted_stat("vb_replica_num", replicaCountVisitor.getVBucketNumber(),
                      add_stat, cookie);
      add_casted_stat("vb_replica_curr_items", replicaCountVisitor.getNumItems(),
                      add_stat, cookie);
      add_casted_stat("vb_replica_num_non_resident",
                      replicaCountVisitor.getNonResident(), add_stat, cookie);
      add_casted_stat("vb_replica_perc_mem_resident",
                      replicaCountVisitor.getMemResidentPer(),
                      add_stat, cookie);
      add_casted_stat("vb_replica_eject", replicaCountVisitor.getEjects(),
                      add_stat, cookie);
      add_casted_stat("vb_replica_expired", replicaCountVisitor.getExpired(),
                      add_stat, cookie);
      add_casted_stat("vb_replica_meta_data_memory",
                      replicaCountVisitor.getMetaDataMemory(), add_stat, cookie);
      add_casted_stat("vb_replica_meta_data_disk",
                      replicaCountVisitor.getMetaDataDisk(), add_stat, cookie);
      add_casted_stat("vb_replica_ht_memory",
                      replicaCountVisitor.getHashtableMemory(),
                      add_stat, cookie);
      add_casted_stat("vb_replica_itm_memory",
                      replicaCountVisitor.getItemMemory(), add_stat, cookie);
      add_casted_stat("vb_replica_ops_create",
                      replicaCountVisitor.getOpsCreate(), add_stat, cookie);
      add_casted_stat("vb_replica_ops_update",
                      replicaCountVisitor.getOpsUpdate(), add_stat, cookie);
      add_casted_stat("vb_replica_ops_delete",
                      replicaCountVisitor.getOpsDelete(), add_stat, cookie);
      add_casted_stat("vb_replica_ops_reject",
                      replicaCountVisitor.getOpsReject(), add_stat, cookie);
      add_casted_stat("vb_replica_queue_size",
                      replicaCountVisitor.getQueueSize(), add_stat, cookie);
      add_casted_stat("vb_replica_queue_memory",
                      replicaCountVisitor.getQueueMemory(),
                      add_stat, cookie);
      add_casted_stat("vb_replica_queue_age",
                      replicaCountVisitor.getAge(), add_stat, cookie);
      add_casted_stat("vb_replica_queue_pending",
                      replicaCountVisitor.getPendingWrites(),
                      add_stat, cookie);
      add_casted_stat("vb_replica_queue_fill",
                      replicaCountVisitor.getQueueFill(), add_stat, cookie);
      add_casted_stat("vb_replica_queue_drain",
                      replicaCountVisitor.getQueueDrain(), add_stat, cookie);
      add_casted_stat("vb_replica_rollback_item_count",
                      replicaCountVisitor.getRollbackItemCount(),
                      add_stat, cookie);
  
      add_casted_stat("vb_pending_num",
                      pendingCountVisitor.getVBucketNumber(), add_stat, cookie);
      add_casted_stat("vb_pending_curr_items",
                      pendingCountVisitor.getNumItems(), add_stat, cookie);
      add_casted_stat("vb_pending_num_non_resident",
                      pendingCountVisitor.getNonResident(),
                      add_stat, cookie);
      add_casted_stat("vb_pending_perc_mem_resident",
                      pendingCountVisitor.getMemResidentPer(), add_stat, cookie);
      add_casted_stat("vb_pending_eject", pendingCountVisitor.getEjects(),
                      add_stat, cookie);
      add_casted_stat("vb_pending_expired", pendingCountVisitor.getExpired(),
                      add_stat, cookie);
      add_casted_stat("vb_pending_meta_data_memory",
                      pendingCountVisitor.getMetaDataMemory(),
                      add_stat, cookie);
      add_casted_stat("vb_pending_meta_data_disk",
                      pendingCountVisitor.getMetaDataDisk(),
                      add_stat, cookie);
      add_casted_stat("vb_pending_ht_memory",
                      pendingCountVisitor.getHashtableMemory(),
                      add_stat, cookie);
      add_casted_stat("vb_pending_itm_memory",
                      pendingCountVisitor.getItemMemory(), add_stat, cookie);
      add_casted_stat("vb_pending_ops_create",
                      pendingCountVisitor.getOpsCreate(), add_stat, cookie);
      add_casted_stat("vb_pending_ops_update",
                      pendingCountVisitor.getOpsUpdate(), add_stat, cookie);
      add_casted_stat("vb_pending_ops_delete",
                      pendingCountVisitor.getOpsDelete(), add_stat, cookie);
      add_casted_stat("vb_pending_ops_reject",
                      pendingCountVisitor.getOpsReject(), add_stat, cookie);
      add_casted_stat("vb_pending_queue_size",
                      pendingCountVisitor.getQueueSize(), add_stat, cookie);
      add_casted_stat("vb_pending_queue_memory",
                      pendingCountVisitor.getQueueMemory(),
                      add_stat, cookie);
      add_casted_stat("vb_pending_queue_age", pendingCountVisitor.getAge(),
                      add_stat, cookie);
      add_casted_stat("vb_pending_queue_pending",
                      pendingCountVisitor.getPendingWrites(),
                      add_stat, cookie);
      add_casted_stat("vb_pending_queue_fill",
                      pendingCountVisitor.getQueueFill(), add_stat, cookie);
      add_casted_stat("vb_pending_queue_drain",
                      pendingCountVisitor.getQueueDrain(), add_stat, cookie);
      add_casted_stat("vb_pending_rollback_item_count",
                      pendingCountVisitor.getRollbackItemCount(),
                      add_stat, cookie);
  
      add_casted_stat("vb_dead_num", deadCountVisitor.getVBucketNumber(),
                      add_stat, cookie);
  
      kvBucket->getFileStats(cookie, add_stat);
  
      add_casted_stat("ep_persist_vbstate_total",
                      epstats.totalPersistVBState, add_stat, cookie);
  
      add_casted_stat("ep_vb_total",
                      activeCountVisitor.getVBucketNumber() +
                      replicaCountVisitor.getVBucketNumber() +
                      pendingCountVisitor.getVBucketNumber() +
                      deadCountVisitor.getVBucketNumber(),
                      add_stat, cookie);
  
      add_casted_stat("ep_total_new_items",
                      activeCountVisitor.getOpsCreate() +
                      replicaCountVisitor.getOpsCreate() +
                      pendingCountVisitor.getOpsCreate(),
                      add_stat, cookie);
      add_casted_stat("ep_total_del_items",
                      activeCountVisitor.getOpsDelete() +
                      replicaCountVisitor.getOpsDelete() +
                      pendingCountVisitor.getOpsDelete(),
                      add_stat, cookie);
      add_casted_stat("ep_diskqueue_memory",
                      activeCountVisitor.getQueueMemory() +
                      replicaCountVisitor.getQueueMemory() +
                      pendingCountVisitor.getQueueMemory(),
                      add_stat, cookie);
      add_casted_stat("ep_diskqueue_fill",
                      activeCountVisitor.getQueueFill() +
                      replicaCountVisitor.getQueueFill() +
                      pendingCountVisitor.getQueueFill(),
                      add_stat, cookie);
      add_casted_stat("ep_diskqueue_drain",
                      activeCountVisitor.getQueueDrain() +
                      replicaCountVisitor.getQueueDrain() +
                      pendingCountVisitor.getQueueDrain(),
                      add_stat, cookie);
      add_casted_stat("ep_diskqueue_pending",
                      activeCountVisitor.getPendingWrites() +
                      replicaCountVisitor.getPendingWrites() +
                      pendingCountVisitor.getPendingWrites(),
                      add_stat, cookie);
      add_casted_stat("ep_meta_data_memory",
                      activeCountVisitor.getMetaDataMemory() +
                      replicaCountVisitor.getMetaDataMemory() +
                      pendingCountVisitor.getMetaDataMemory(),
                      add_stat, cookie);
      add_casted_stat("ep_meta_data_disk",
                      activeCountVisitor.getMetaDataDisk() +
                      replicaCountVisitor.getMetaDataDisk() +
                      pendingCountVisitor.getMetaDataDisk(),
                      add_stat, cookie);
  
      size_t memUsed =  stats.getTotalMemoryUsed();
      add_casted_stat("mem_used", memUsed, add_stat, cookie);
      add_casted_stat("ep_mem_low_wat_percent", stats.mem_low_wat_percent,
                      add_stat, cookie);
      add_casted_stat("ep_mem_high_wat_percent", stats.mem_high_wat_percent,
                      add_stat, cookie);
      add_casted_stat("bytes", memUsed, add_stat, cookie);
      add_casted_stat("ep_kv_size", stats.currentSize, add_stat, cookie);
      add_casted_stat("ep_blob_num", stats.numBlob, add_stat, cookie);
  #if defined(HAVE_JEMALLOC) || defined(HAVE_TCMALLOC)
      add_casted_stat("ep_blob_overhead", stats.blobOverhead, add_stat, cookie);
  #else
      add_casted_stat("ep_blob_overhead", "unknown", add_stat, cookie);
  #endif
      add_casted_stat("ep_value_size", stats.totalValueSize, add_stat, cookie);
      add_casted_stat("ep_storedval_size", stats.totalStoredValSize,
                      add_stat, cookie);
  #if defined(HAVE_JEMALLOC) || defined(HAVE_TCMALLOC)
      add_casted_stat("ep_storedval_overhead", stats.blobOverhead, add_stat, cookie);
  #else
      add_casted_stat("ep_storedval_overhead", "unknown", add_stat, cookie);
  #endif
      add_casted_stat("ep_storedval_num", stats.numStoredVal, add_stat, cookie);
      add_casted_stat("ep_overhead", stats.memOverhead, add_stat, cookie);
      add_casted_stat("ep_item_num", stats.numItem, add_stat, cookie);
      add_casted_stat("ep_total_cache_size",
                      activeCountVisitor.getCacheSize() +
                      replicaCountVisitor.getCacheSize() +
                      pendingCountVisitor.getCacheSize(),
                      add_stat, cookie);
  
      add_casted_stat("rollback_item_count",
                      activeCountVisitor.getRollbackItemCount() +
                      replicaCountVisitor.getRollbackItemCount() +
                      pendingCountVisitor.getRollbackItemCount(),
                      add_stat, cookie);
  
      add_casted_stat("ep_oom_errors", stats.oom_errors, add_stat, cookie);
      add_casted_stat("ep_tmp_oom_errors", stats.tmp_oom_errors,
                      add_stat, cookie);
      add_casted_stat("ep_mem_tracker_enabled", stats.memoryTrackerEnabled,
                      add_stat, cookie);
      add_casted_stat("ep_bg_fetched", epstats.bg_fetched,
                      add_stat, cookie);
      add_casted_stat("ep_bg_meta_fetched", epstats.bg_meta_fetched,
                      add_stat, cookie);
      add_casted_stat("ep_bg_remaining_items", epstats.numRemainingBgItems,
                      add_stat, cookie);
      add_casted_stat("ep_bg_remaining_jobs", epstats.numRemainingBgJobs,
                      add_stat, cookie);
      add_casted_stat("ep_max_bg_remaining_jobs", epstats.maxRemainingBgJobs,
                      add_stat, cookie);
      add_casted_stat("ep_tap_bg_fetched", stats.numTapBGFetched,
                      add_stat, cookie);
      add_casted_stat("ep_tap_bg_fetch_requeued", stats.numTapBGFetchRequeued,
                      add_stat, cookie);
      add_casted_stat("ep_num_pager_runs", epstats.pagerRuns,
                      add_stat, cookie);
      add_casted_stat("ep_num_expiry_pager_runs", epstats.expiryPagerRuns,
                      add_stat, cookie);
      add_casted_stat("ep_items_rm_from_checkpoints",
                      epstats.itemsRemovedFromCheckpoints,
                      add_stat, cookie);
      add_casted_stat("ep_num_value_ejects", epstats.numValueEjects,
                      add_stat, cookie);
      add_casted_stat("ep_num_eject_failures", epstats.numFailedEjects,
                      add_stat, cookie);
      add_casted_stat("ep_num_not_my_vbuckets", epstats.numNotMyVBuckets,
                      add_stat, cookie);
  
      add_casted_stat("ep_pending_ops", epstats.pendingOps, add_stat, cookie);
      add_casted_stat("ep_pending_ops_total", epstats.pendingOpsTotal,
                      add_stat, cookie);
      add_casted_stat("ep_pending_ops_max", epstats.pendingOpsMax,
                      add_stat, cookie);
      add_casted_stat("ep_pending_ops_max_duration",
                      epstats.pendingOpsMaxDuration,
                      add_stat, cookie);
  
      add_casted_stat("ep_pending_compactions", epstats.pendingCompactions,
                      add_stat, cookie);
      add_casted_stat("ep_rollback_count", epstats.rollbackCount,
                      add_stat, cookie);
  
      size_t vbDeletions = epstats.vbucketDeletions.load();
      if (vbDeletions > 0) {
          add_casted_stat("ep_vbucket_del_max_walltime",
                          epstats.vbucketDelMaxWalltime,
                          add_stat, cookie);
          add_casted_stat("ep_vbucket_del_avg_walltime",
                          epstats.vbucketDelTotWalltime / vbDeletions,
                          add_stat, cookie);
      }
  
      size_t numBgOps = epstats.bgNumOperations.load();
      if (numBgOps > 0) {
          add_casted_stat("ep_bg_num_samples", epstats.bgNumOperations,
                          add_stat, cookie);
          add_casted_stat("ep_bg_min_wait",
                          epstats.bgMinWait,
                          add_stat, cookie);
          add_casted_stat("ep_bg_max_wait",
                          epstats.bgMaxWait,
                          add_stat, cookie);
          add_casted_stat("ep_bg_wait_avg",
                          epstats.bgWait / numBgOps,
                          add_stat, cookie);
          add_casted_stat("ep_bg_min_load",
                          epstats.bgMinLoad,
                          add_stat, cookie);
          add_casted_stat("ep_bg_max_load",
                          epstats.bgMaxLoad,
                          add_stat, cookie);
          add_casted_stat("ep_bg_load_avg",
                          epstats.bgLoad / numBgOps,
                          add_stat, cookie);
          add_casted_stat("ep_bg_wait",
                          epstats.bgWait,
                          add_stat, cookie);
          add_casted_stat("ep_bg_load",
                          epstats.bgLoad,
                          add_stat, cookie);
      }
  
      add_casted_stat("ep_num_non_resident",
                      activeCountVisitor.getNonResident() +
                      pendingCountVisitor.getNonResident() +
                      replicaCountVisitor.getNonResident(),
                      add_stat, cookie);
  
      add_casted_stat("ep_degraded_mode", isDegradedMode(), add_stat, cookie);
  
      add_casted_stat("ep_mlog_compactor_runs", epstats.mlogCompactorRuns,
                      add_stat, cookie);
      add_casted_stat("ep_num_access_scanner_runs", epstats.alogRuns,
                      add_stat, cookie);
      add_casted_stat("ep_num_access_scanner_skips",
                      epstats.accessScannerSkips, add_stat, cookie);
      add_casted_stat("ep_access_scanner_last_runtime", epstats.alogRuntime,
                      add_stat, cookie);
      add_casted_stat("ep_access_scanner_num_items", epstats.alogNumItems,
                      add_stat, cookie);
  
      if (kvBucket->isAccessScannerEnabled() && epstats.alogTime.load() != 0)
      {
          char timestr[20];
          struct tm alogTim;
          hrtime_t alogTime = epstats.alogTime.load();
          if (cb_gmtime_r((time_t *)&alogTime, &alogTim) == -1) {
              add_casted_stat("ep_access_scanner_task_time", "UNKNOWN", add_stat,
                              cookie);
          } else {
              strftime(timestr, 20, "%Y-%m-%d %H:%M:%S", &alogTim);
              add_casted_stat("ep_access_scanner_task_time", timestr, add_stat,
                              cookie);
          }
      } else {
          add_casted_stat("ep_access_scanner_task_time", "NOT_SCHEDULED",
                          add_stat, cookie);
      }
  
      if (kvBucket->isExpPagerEnabled()) {
          char timestr[20];
          struct tm expPagerTim;
          hrtime_t expPagerTime = epstats.expPagerTime.load();
          if (cb_gmtime_r((time_t *)&expPagerTime, &expPagerTim) == -1) {
              add_casted_stat("ep_expiry_pager_task_time", "UNKNOWN", add_stat,
                              cookie);
          } else {
              strftime(timestr, 20, "%Y-%m-%d %H:%M:%S", &expPagerTim);
              add_casted_stat("ep_expiry_pager_task_time", timestr, add_stat,
                              cookie);
          }
      } else {
          add_casted_stat("ep_expiry_pager_task_time", "NOT_SCHEDULED",
                          add_stat, cookie);
      }
  
      add_casted_stat("ep_startup_time", startupTime.load(), add_stat, cookie);
  
      if (getConfiguration().isWarmup()) {
          Warmup *wp = kvBucket->getWarmup();
          if (wp == nullptr) {
              throw std::logic_error("EPEngine::doEngineStats: warmup is NULL");
          }
          if (!kvBucket->isWarmingUp()) {
              add_casted_stat("ep_warmup_thread", "complete", add_stat, cookie);
          } else {
              add_casted_stat("ep_warmup_thread", "running", add_stat, cookie);
          }
          if (wp->getTime() > 0) {
              add_casted_stat("ep_warmup_time", wp->getTime() / 1000,
                              add_stat, cookie);
          }
          add_casted_stat("ep_warmup_oom", epstats.warmOOM, add_stat, cookie);
          add_casted_stat("ep_warmup_dups", epstats.warmDups, add_stat, cookie);
      }
  
      add_casted_stat("ep_num_ops_get_meta", epstats.numOpsGetMeta,
                      add_stat, cookie);
      add_casted_stat("ep_num_ops_set_meta", epstats.numOpsSetMeta,
                      add_stat, cookie);
      add_casted_stat("ep_num_ops_del_meta", epstats.numOpsDelMeta,
                      add_stat, cookie);
      add_casted_stat("ep_num_ops_set_meta_res_fail",
                      epstats.numOpsSetMetaResolutionFailed, add_stat, cookie);
      add_casted_stat("ep_num_ops_del_meta_res_fail",
                      epstats.numOpsDelMetaResolutionFailed, add_stat, cookie);
      add_casted_stat("ep_num_ops_set_ret_meta", epstats.numOpsSetRetMeta,
                      add_stat, cookie);
      add_casted_stat("ep_num_ops_del_ret_meta", epstats.numOpsDelRetMeta,
                      add_stat, cookie);
      add_casted_stat("ep_num_ops_get_meta_on_set_meta",
                      epstats.numOpsGetMetaOnSetWithMeta, add_stat, cookie);
      add_casted_stat("ep_chk_persistence_timeout",
                      VBucket::getCheckpointFlushTimeout(),
                      add_stat, cookie);
      add_casted_stat("ep_chk_persistence_remains",
                      activeCountVisitor.getChkPersistRemaining() +
                      pendingCountVisitor.getChkPersistRemaining() +
                      replicaCountVisitor.getChkPersistRemaining(),
                      add_stat, cookie);
      add_casted_stat("ep_workload_pattern",
                      workload->stringOfWorkLoadPattern(),
                      add_stat, cookie);
  
      add_casted_stat("ep_defragmenter_num_visited", epstats.defragNumVisited,
                      add_stat, cookie);
      add_casted_stat("ep_defragmenter_num_moved", epstats.defragNumMoved,
                      add_stat, cookie);
  
      add_casted_stat("ep_cursor_dropping_lower_threshold",
                      epstats.cursorDroppingLThreshold, add_stat, cookie);
      add_casted_stat("ep_cursor_dropping_upper_threshold",
                      epstats.cursorDroppingUThreshold, add_stat, cookie);
      add_casted_stat("ep_cursors_dropped",
                      epstats.cursorsDropped, add_stat, cookie);
  
  
      // Note: These are also reported per-shard in 'kvstore' stats, however
      // we want to be able to graph these over time, and hence need to expose
      // to ns_sever at the top-level.
      size_t value = 0;
      if (kvBucket->getKVStoreStat("io_total_read_bytes", value,
                                   KVBucketIface::KVSOption::BOTH)) {
          add_casted_stat("ep_io_total_read_bytes",  value, add_stat, cookie);
      }
      if (kvBucket->getKVStoreStat("io_total_write_bytes", value,
                                   KVBucketIface::KVSOption::BOTH)) {
          add_casted_stat("ep_io_total_write_bytes",  value, add_stat, cookie);
      }
      if (kvBucket->getKVStoreStat("io_compaction_read_bytes", value,
                                   KVBucketIface::KVSOption::BOTH)) {
          add_casted_stat("ep_io_compaction_read_bytes",  value, add_stat, cookie);
      }
      if (kvBucket->getKVStoreStat("io_compaction_write_bytes", value,
                                   KVBucketIface::KVSOption::BOTH)) {
          add_casted_stat("ep_io_compaction_write_bytes",  value, add_stat, cookie);
      }
      if (kvBucket->getKVStoreStat("Block_cache_hits", value,
                                   KVBucketIface::KVSOption::RW)) {
          add_casted_stat("ep_block_cache_hits", value, add_stat, cookie);
      }
      if (kvBucket->getKVStoreStat("Block_cache_misses", value,
                                   KVBucketIface::KVSOption::RW)) {
          add_casted_stat("ep_block_cache_misses", value, add_stat, cookie);
      }
  
      // Add stats for tracking HLC drift
      add_casted_stat("ep_active_hlc_drift",
          activeCountVisitor.getTotalAbsHLCDrift().total, add_stat, cookie);
      add_casted_stat("ep_active_hlc_drift_count",
          activeCountVisitor.getTotalAbsHLCDrift().updates, add_stat, cookie);
      add_casted_stat("ep_replica_hlc_drift",
          replicaCountVisitor.getTotalAbsHLCDrift().total, add_stat, cookie);
      add_casted_stat("ep_replica_hlc_drift_count",
          replicaCountVisitor.getTotalAbsHLCDrift().updates, add_stat, cookie);
  
      add_casted_stat("ep_active_ahead_exceptions",
          activeCountVisitor.getTotalHLCDriftExceptionCounters().ahead,
          add_stat, cookie);
      add_casted_stat("ep_active_behind_exceptions",
          activeCountVisitor.getTotalHLCDriftExceptionCounters().behind,
          add_stat, cookie);
      add_casted_stat("ep_replica_ahead_exceptions",
          replicaCountVisitor.getTotalHLCDriftExceptionCounters().ahead,
          add_stat, cookie);
      add_casted_stat("ep_replica_behind_exceptions",
          replicaCountVisitor.getTotalHLCDriftExceptionCounters().behind,
          add_stat, cookie);
  
      // A single total for ahead exceptions accross all active/replicas
      add_casted_stat("ep_clock_cas_drift_threshold_exceeded",
          activeCountVisitor.getTotalHLCDriftExceptionCounters().ahead +
          replicaCountVisitor.getTotalHLCDriftExceptionCounters().ahead,
          add_stat, cookie);
  
      return ENGINE_SUCCESS;
  }
  
  ENGINE_ERROR_CODE EventuallyPersistentEngine::doMemoryStats(const void *cookie,
                                                             ADD_STAT add_stat) {
      add_casted_stat("bytes", stats.getTotalMemoryUsed(), add_stat, cookie);
      add_casted_stat("mem_used", stats.getTotalMemoryUsed(), add_stat, cookie);
      add_casted_stat("ep_kv_size", stats.currentSize, add_stat, cookie);
      add_casted_stat("ep_value_size", stats.totalValueSize, add_stat, cookie);
      add_casted_stat("ep_overhead", stats.memOverhead, add_stat, cookie);
      add_casted_stat("ep_max_size", stats.getMaxDataSize(), add_stat, cookie);
      add_casted_stat("ep_mem_low_wat", stats.mem_low_wat, add_stat, cookie);
      add_casted_stat("ep_mem_low_wat_percent", stats.mem_low_wat_percent,
                      add_stat, cookie);
      add_casted_stat("ep_mem_high_wat", stats.mem_high_wat, add_stat, cookie);
      add_casted_stat("ep_mem_high_wat_percent", stats.mem_high_wat_percent,
                      add_stat, cookie);
      add_casted_stat("ep_oom_errors", stats.oom_errors, add_stat, cookie);
      add_casted_stat("ep_tmp_oom_errors", stats.tmp_oom_errors,
                      add_stat, cookie);
  
      add_casted_stat("ep_blob_num", stats.numBlob, add_stat, cookie);
  #if defined(HAVE_JEMALLOC) || defined(HAVE_TCMALLOC)
      add_casted_stat("ep_blob_overhead", stats.blobOverhead, add_stat, cookie);
  #else
      add_casted_stat("ep_blob_overhead", "unknown", add_stat, cookie);
  #endif
      add_casted_stat("ep_storedval_size", stats.totalStoredValSize,
                      add_stat, cookie);
  #if defined(HAVE_JEMALLOC) || defined(HAVE_TCMALLOC)
      add_casted_stat("ep_storedval_overhead", stats.blobOverhead, add_stat, cookie);
  #else
      add_casted_stat("ep_storedval_overhead", "unknown", add_stat, cookie);
  #endif
      add_casted_stat("ep_storedval_num", stats.numStoredVal, add_stat, cookie);
      add_casted_stat("ep_item_num", stats.numItem, add_stat, cookie);
  
      std::map<std::string, size_t> alloc_stats;
      MemoryTracker::getInstance(*getServerApiFunc()->alloc_hooks)->
          getAllocatorStats(alloc_stats);
  
      for (const auto& it : alloc_stats) {
          add_casted_stat(it.first.c_str(), it.second, add_stat, cookie);
      }
  
      return ENGINE_SUCCESS;
  }
  
  ENGINE_ERROR_CODE EventuallyPersistentEngine::doVBucketStats(
                                                         const void *cookie,
                                                         ADD_STAT add_stat,
                                                         const char* stat_key,
                                                         int nkey,
                                                         bool prevStateRequested,
                                                         bool details) {
      class StatVBucketVisitor : public VBucketVisitor {
      public:
          StatVBucketVisitor(KVBucketIface* store,
                             const void *c, ADD_STAT a,
                             bool isPrevStateRequested, bool detailsRequested) :
              eps(store), cookie(c), add_stat(a),
              isPrevState(isPrevStateRequested),
              isDetailsRequested(detailsRequested) {}
  
          void visitBucket(RCPtr<VBucket> &vb) override {
              addVBStats(cookie, add_stat, vb, eps, isPrevState,
                         isDetailsRequested);
          }
  
          static void addVBStats(const void *cookie, ADD_STAT add_stat,
                                 RCPtr<VBucket> &vb,
                                 KVBucketIface* store,
                                 bool isPrevStateRequested,
                                 bool detailsRequested) {
              if (!vb) {
                  return;
              }
  
              if (isPrevStateRequested) {
                  try {
                      char buf[16];
                      checked_snprintf(buf, sizeof(buf), "vb_%d", vb->getId());
                      add_casted_stat(buf,
                                      VBucket::toString(vb->getInitialState()),
                                      add_stat, cookie);
                  } catch (std::exception& error) {
                      LOG(EXTENSION_LOG_WARNING,
                          "addVBStats: Failed building stats: %s", error.what());
                  }
              } else {
                  vb->addStats(detailsRequested, add_stat, cookie,
                               store->getItemEvictionPolicy());
              }
          }
  
      private:
          KVBucketIface* eps;
          const void *cookie;
          ADD_STAT add_stat;
          bool isPrevState;
          bool isDetailsRequested;
      };
  
      if (nkey > 16 && strncmp(stat_key, "vbucket-details", 15) == 0) {
          std::string vbid(&stat_key[16], nkey - 16);
          uint16_t vbucket_id(0);
          if (!parseUint16(vbid.c_str(), &vbucket_id)) {
              return ENGINE_EINVAL;
          }
          RCPtr<VBucket> vb = getVBucket(vbucket_id);
          if (!vb) {
              return ENGINE_NOT_MY_VBUCKET;
          }
  
          StatVBucketVisitor::addVBStats(cookie, add_stat, vb, kvBucket,
                                         prevStateRequested, details);
      }
      else {
          StatVBucketVisitor svbv(kvBucket, cookie, add_stat,
                                  prevStateRequested, details);
          kvBucket->visit(svbv);
      }
      return ENGINE_SUCCESS;
  }
  
  ENGINE_ERROR_CODE EventuallyPersistentEngine::doHashStats(const void *cookie,
                                                            ADD_STAT add_stat) {
  
      class StatVBucketVisitor : public VBucketVisitor {
      public:
          StatVBucketVisitor(const void *c, ADD_STAT a) : cookie(c),
                                                          add_stat(a) {}
  
          void visitBucket(RCPtr<VBucket> &vb) override {
              uint16_t vbid = vb->getId();
              char buf[32];
              try {
                  checked_snprintf(buf, sizeof(buf), "vb_%d:state", vbid);
                  add_casted_stat(buf, VBucket::toString(vb->getState()),
                                  add_stat, cookie);
              } catch (std::exception& error) {
                  LOG(EXTENSION_LOG_WARNING,
                      "StatVBucketVisitor::visitBucket: Failed to build stat: %s",
                      error.what());
              }
  
              HashTableDepthStatVisitor depthVisitor;
              vb->ht.visitDepth(depthVisitor);
  
              try {
                  checked_snprintf(buf, sizeof(buf), "vb_%d:size", vbid);
                  add_casted_stat(buf, vb->ht.getSize(), add_stat, cookie);
                  checked_snprintf(buf, sizeof(buf), "vb_%d:locks", vbid);
                  add_casted_stat(buf, vb->ht.getNumLocks(), add_stat, cookie);
                  checked_snprintf(buf, sizeof(buf), "vb_%d:min_depth", vbid);
                  add_casted_stat(buf,
                                  depthVisitor.min == -1 ? 0 : depthVisitor.min,
                                  add_stat, cookie);
                  checked_snprintf(buf, sizeof(buf), "vb_%d:max_depth", vbid);
                  add_casted_stat(buf, depthVisitor.max, add_stat, cookie);
                  checked_snprintf(buf, sizeof(buf), "vb_%d:histo", vbid);
                  add_casted_stat(buf, depthVisitor.depthHisto, add_stat, cookie);
                  checked_snprintf(buf, sizeof(buf), "vb_%d:reported", vbid);
                  add_casted_stat(buf, vb->ht.getNumInMemoryItems(), add_stat,
                                  cookie);
                  checked_snprintf(buf, sizeof(buf), "vb_%d:counted", vbid);
                  add_casted_stat(buf, depthVisitor.size, add_stat, cookie);
                  checked_snprintf(buf, sizeof(buf), "vb_%d:resized", vbid);
                  add_casted_stat(buf, vb->ht.getNumResizes(), add_stat, cookie);
                  checked_snprintf(buf, sizeof(buf), "vb_%d:mem_size", vbid);
                  add_casted_stat(buf, vb->ht.memSize, add_stat, cookie);
                  checked_snprintf(buf, sizeof(buf), "vb_%d:mem_size_counted",
                                   vbid);
                  add_casted_stat(buf, depthVisitor.memUsed, add_stat, cookie);
              } catch (std::exception& error) {
                  LOG(EXTENSION_LOG_WARNING,
                      "StatVBucketVisitor::visitBucket: Failed to build stat: %s",
                      error.what());
              }
          }
  
          const void *cookie;
          ADD_STAT add_stat;
      };
  
      StatVBucketVisitor svbv(cookie, add_stat);
      kvBucket->visit(svbv);
  
      return ENGINE_SUCCESS;
  }
  
  class StatCheckpointVisitor : public VBucketVisitor {
  public:
      StatCheckpointVisitor(KVBucketIface* kvs, const void *c,
                            ADD_STAT a) : kvBucket(kvs), cookie(c),
                                          add_stat(a) {}
  
      void visitBucket(RCPtr<VBucket> &vb) override {
          addCheckpointStat(cookie, add_stat, kvBucket, vb);
      }
  
      static void addCheckpointStat(const void *cookie, ADD_STAT add_stat,
                                    KVBucketIface* eps,
                                    RCPtr<VBucket> &vb) {
          if (!vb) {
              return;
          }
  
          uint16_t vbid = vb->getId();
          char buf[256];
          try {
              checked_snprintf(buf, sizeof(buf), "vb_%d:state", vbid);
              add_casted_stat(buf, VBucket::toString(vb->getState()),
                              add_stat, cookie);
              vb->checkpointManager.addStats(add_stat, cookie);
              checked_snprintf(buf, sizeof(buf), "vb_%d:persisted_checkpoint_id",
                               vbid);
              add_casted_stat(buf, eps->getLastPersistedCheckpointId(vbid),
                              add_stat, cookie);
          } catch (std::exception& error) {
              LOG(EXTENSION_LOG_WARNING,
                  "StatCheckpointVisitor::addCheckpointStat: error building stats: %s",
                  error.what());
          }
      }
  
      KVBucketIface* kvBucket;
      const void *cookie;
      ADD_STAT add_stat;
  };
  
  
  class StatCheckpointTask : public GlobalTask {
  public:
      StatCheckpointTask(EventuallyPersistentEngine *e, const void *c,
              ADD_STAT a) : GlobalTask(e, TaskId::StatCheckpointTask,
                                       0, false),
                            ep(e), cookie(c), add_stat(a) { }
      bool run(void) {
          TRACE_EVENT0("ep-engine/task", "StatsCheckpointTask");
          StatCheckpointVisitor scv(ep->getKVBucket(), cookie, add_stat);
          ep->getKVBucket()->visit(scv);
          ep->notifyIOComplete(cookie, ENGINE_SUCCESS);
          return false;
      }
  
      std::string getDescription() {
          return "checkpoint stats for all vbuckets";
      }
  
  private:
      EventuallyPersistentEngine *ep;
      const void *cookie;
      ADD_STAT add_stat;
  };
  /// @endcond
  
  ENGINE_ERROR_CODE EventuallyPersistentEngine::doCheckpointStats(
                                                            const void *cookie,
                                                            ADD_STAT add_stat,
                                                            const char* stat_key,
                                                            int nkey) {
  
      if (nkey == 10) {
          void* es = getEngineSpecific(cookie);
          if (es == NULL) {
              ExTask task = new StatCheckpointTask(this, cookie, add_stat);
              ExecutorPool::get()->schedule(task, NONIO_TASK_IDX);
              storeEngineSpecific(cookie, this);
              return ENGINE_EWOULDBLOCK;
          } else {
              storeEngineSpecific(cookie, NULL);
          }
      } else if (nkey > 11) {
          std::string vbid(&stat_key[11], nkey - 11);
          uint16_t vbucket_id(0);
          if (!parseUint16(vbid.c_str(), &vbucket_id)) {
              return ENGINE_EINVAL;
          }
          RCPtr<VBucket> vb = getVBucket(vbucket_id);
  
          StatCheckpointVisitor::addCheckpointStat(cookie, add_stat,
                                                   kvBucket, vb);
      }
  
      return ENGINE_SUCCESS;
  }
  
  /**
   * Function object to send stats for a single tap or dcp connection.
   */
  struct ConnStatBuilder {
      ConnStatBuilder(const void *c, ADD_STAT as, ConnCounter& tc)
          : cookie(c), add_stat(as), aggregator(tc) {}
  
      void operator() (connection_t &tc) {
          ++aggregator.totalConns;
          tc->addStats(add_stat, cookie);
  
          Producer *tp = dynamic_cast<Producer*>(tc.get());
          if (tp) {
              ++aggregator.totalProducers;
              tp->aggregateQueueStats(aggregator);
          }
      }
  
      const void *cookie;
      ADD_STAT    add_stat;
      ConnCounter& aggregator;
  };
  
  struct ConnAggStatBuilder {
      ConnAggStatBuilder(std::map<std::string, ConnCounter*> *m,
                        const char *s, size_t sl)
          : counters(m), sep(s), sep_len(sl) {}
  
      ConnCounter *getTarget(connection_t& tc) {
          ConnCounter *rv = NULL;
  
          if (tc.get()) {
              const std::string name(tc->getName());
              size_t pos1 = name.find(':');
              if (pos1 == name.npos) {
                  throw std::invalid_argument("ConnAggStatBuilder::getTarget: "
                          "connection tc (which has name '" + tc->getName() +
                          "' does not include a colon (:)");
              }
              size_t pos2 = name.find(sep, pos1+1, sep_len);
              if (pos2 != name.npos) {
                  std::string prefix(name.substr(pos1+1, pos2 - pos1 - 1));
                  rv = (*counters)[prefix];
                  if (rv == NULL) {
                      rv = new ConnCounter;
                      (*counters)[prefix] = rv;
                  }
              }
          }
          return rv;
      }
  
      void aggregate(connection_t& c, ConnCounter *tc){
          ConnCounter counter;
  
          ++counter.totalConns;
          if (dynamic_cast<Producer*>(c.get())) {
              ++counter.totalProducers;
          }
  
          c->aggregateQueueStats(counter);
  
          ConnCounter* total = getTotalCounter();
          *total += counter;
  
          if (tc) {
              *tc += counter;
          }
      }
  
      ConnCounter *getTotalCounter() {
          ConnCounter *rv = NULL;
          std::string sepr(sep);
          std::string total(sepr + "total");
          rv = (*counters)[total];
          if(rv == NULL) {
              rv = new ConnCounter;
              (*counters)[total] = rv;
          }
          return rv;
      }
  
      void operator() (connection_t& tc) {
          if (tc.get() && tc->isConnected()) {
              ConnCounter *aggregator = getTarget(tc);
              aggregate(tc, aggregator);
          }
      }
  
      std::map<std::string, ConnCounter*> *counters;
      const char *sep;
      size_t sep_len;
  };
  
  /// @endcond
  
  static void showConnAggStat(const std::string &prefix,
                              ConnCounter *counter,
                              const void *cookie,
                              ADD_STAT add_stat,
                              conn_type_t conn_type) {
  
      try {
          char statname[80] = {0};
          const size_t sl(sizeof(statname));
          checked_snprintf(statname, sl, "%s:count", prefix.c_str());
          add_casted_stat(statname, counter->totalConns, add_stat, cookie);
  
          checked_snprintf(statname, sl, "%s:total_backlog_size", prefix.c_str());
          add_casted_stat(statname, counter->conn_totalBacklogSize,
                          add_stat, cookie);
  
          checked_snprintf(statname, sl, "%s:backoff", prefix.c_str());
          add_casted_stat(statname, counter->conn_queueBackoff,
                          add_stat, cookie);
  
          if (conn_type == TAP_CONN) {
              checked_snprintf(statname, sl, "%s:qlen", prefix.c_str());
              add_casted_stat(statname, counter->conn_queue, add_stat, cookie);
  
              checked_snprintf(statname, sl, "%s:fill", prefix.c_str());
              add_casted_stat(statname, counter->conn_queueFill,
                              add_stat, cookie);
  
              checked_snprintf(statname, sl, "%s:drain", prefix.c_str());
              add_casted_stat(statname, counter->conn_queueDrain,
                              add_stat, cookie);
  
              checked_snprintf(statname, sl, "%s:backfill_remaining",
                               prefix.c_str());
              add_casted_stat(statname, counter->conn_queueBackfillRemaining,
                              add_stat, cookie);
  
              checked_snprintf(statname, sl, "%s:itemondisk", prefix.c_str());
              add_casted_stat(statname, counter->conn_queueItemOnDisk,
                              add_stat, cookie);
          }
  
          if (conn_type == DCP_CONN) {
              checked_snprintf(statname, sl, "%s:producer_count", prefix.c_str());
              add_casted_stat(statname, counter->totalProducers, add_stat,
                              cookie);
  
              checked_snprintf(statname, sl, "%s:items_sent", prefix.c_str());
              add_casted_stat(statname, counter->conn_queueDrain,
                              add_stat, cookie);
  
              checked_snprintf(statname, sl, "%s:items_remaining",
                               prefix.c_str());
              add_casted_stat(statname, counter->conn_queueRemaining,
                              add_stat, cookie);
  
              checked_snprintf(statname, sl, "%s:total_bytes", prefix.c_str());
              add_casted_stat(statname, counter->conn_totalBytes,
                              add_stat, cookie);
          }
      } catch (std::exception& error) {
          LOG(EXTENSION_LOG_WARNING,
              "showConnAggStat: Failed to build stats: %s", error.what());
      }
  }
  
  ENGINE_ERROR_CODE EventuallyPersistentEngine::doConnAggStats(
                                                          const void *cookie,
                                                          ADD_STAT add_stat,
                                                          const char *sepPtr,
                                                          size_t sep_len,
                                                          conn_type_t connType) {
      // In practice, this will be 1, but C++ doesn't let me use dynamic
      // array sizes.
      const size_t max_sep_len(8);
      sep_len = std::min(sep_len, max_sep_len);
  
      char sep[max_sep_len + 1];
      memcpy(sep, sepPtr, sep_len);
      sep[sep_len] = 0x00;
  
      std::map<std::string, ConnCounter*> counters;
      ConnAggStatBuilder visitor(&counters, sep, sep_len);
      if (connType == TAP_CONN) {
          tapConnMap->each(visitor);
      } else {
          dcpConnMap_->each(visitor);
      }
  
      std::map<std::string, ConnCounter*>::iterator it;
      for (it = counters.begin(); it != counters.end(); ++it) {
          showConnAggStat(it->first, it->second, cookie, add_stat, connType);
          delete it->second;
      }
  
      return ENGINE_SUCCESS;
  }
  
  ENGINE_ERROR_CODE EventuallyPersistentEngine::doTapStats(const void *cookie,
                                                           ADD_STAT add_stat) {
      ConnCounter aggregator;
      ConnStatBuilder tapVisitor(cookie, add_stat, aggregator);
      tapConnMap->each(tapVisitor);
  
      add_casted_stat("ep_tap_total_fetched", stats.numTapFetched,
                      add_stat, cookie);
      add_casted_stat("ep_tap_bg_max_pending", tapConfig->getBgMaxPending(),
                      add_stat, cookie);
      add_casted_stat("ep_tap_bg_fetched", stats.numTapBGFetched,
                      add_stat, cookie);
      add_casted_stat("ep_tap_bg_fetch_requeued", stats.numTapBGFetchRequeued,
                      add_stat, cookie);
      add_casted_stat("ep_tap_fg_fetched", stats.numTapFGFetched,
                      add_stat, cookie);
      add_casted_stat("ep_tap_deletes", stats.numTapDeletes, add_stat, cookie);
      add_casted_stat("ep_replication_throttled", stats.replicationThrottled, add_stat, cookie);
      add_casted_stat("ep_tap_noop_interval", tapConnMap->getNoopInterval(),
                      add_stat, cookie);
      add_casted_stat("ep_tap_count", aggregator.totalConns, add_stat, cookie);
      add_casted_stat("ep_tap_total_queue", aggregator.conn_queue,
                      add_stat, cookie);
      add_casted_stat("ep_tap_queue_fill", aggregator.conn_queueFill,
                      add_stat, cookie);
      add_casted_stat("ep_tap_queue_drain", aggregator.conn_queueDrain,
                      add_stat, cookie);
      add_casted_stat("ep_tap_queue_backoff", aggregator.conn_queueBackoff,
                      add_stat, cookie);
      add_casted_stat("ep_tap_queue_backfillremaining",
                      aggregator.conn_queueBackfillRemaining, add_stat, cookie);
      add_casted_stat("ep_tap_queue_itemondisk", aggregator.conn_queueItemOnDisk,
                      add_stat, cookie);
      add_casted_stat("ep_tap_total_backlog_size",
                      aggregator.conn_totalBacklogSize, add_stat, cookie);
      add_casted_stat("ep_tap_ack_window_size", tapConfig->getAckWindowSize(),
                      add_stat, cookie);
      add_casted_stat("ep_tap_ack_interval", tapConfig->getAckInterval(),
                      add_stat, cookie);
      add_casted_stat("ep_tap_ack_grace_period", tapConfig->getAckGracePeriod(),
                      add_stat, cookie);
      add_casted_stat("ep_tap_backoff_period",
                      tapConfig->getBackoffSleepTime(),
                      add_stat, cookie);
      add_casted_stat("ep_replication_throttle_threshold",
                      stats.replicationThrottleThreshold * 100.0,
                      add_stat, cookie);
      add_casted_stat("ep_replication_throttle_queue_cap",
                      stats.replicationThrottleWriteQueueCap, add_stat, cookie);
  
      if (stats.tapBgNumOperations > 0) {
          add_casted_stat("ep_tap_bg_num_samples", stats.tapBgNumOperations,
                          add_stat, cookie);
          add_casted_stat("ep_tap_bg_min_wait",
                          stats.tapBgMinWait,
                          add_stat, cookie);
          add_casted_stat("ep_tap_bg_max_wait",
                          stats.tapBgMaxWait,
                          add_stat, cookie);
          add_casted_stat("ep_tap_bg_wait_avg",
                          stats.tapBgWait / stats.tapBgNumOperations,
                          add_stat, cookie);
          add_casted_stat("ep_tap_bg_min_load",
                          stats.tapBgMinLoad,
                          add_stat, cookie);
          add_casted_stat("ep_tap_bg_max_load",
                          stats.tapBgMaxLoad,
                          add_stat, cookie);
          add_casted_stat("ep_tap_bg_load_avg",
                          stats.tapBgLoad / stats.tapBgNumOperations,
                          add_stat, cookie);
      }
  
      return ENGINE_SUCCESS;
  }
  
  ENGINE_ERROR_CODE EventuallyPersistentEngine::doDcpStats(const void *cookie,
                                                           ADD_STAT add_stat) {
      ConnCounter aggregator;
      ConnStatBuilder dcpVisitor(cookie, add_stat, aggregator);
      dcpConnMap_->each(dcpVisitor);
  
      add_casted_stat("ep_dcp_count", aggregator.totalConns, add_stat, cookie);
      add_casted_stat("ep_dcp_producer_count", aggregator.totalProducers, add_stat, cookie);
      add_casted_stat("ep_dcp_total_bytes", aggregator.conn_totalBytes, add_stat, cookie);
      add_casted_stat("ep_dcp_total_queue", aggregator.conn_queue,
                      add_stat, cookie);
      add_casted_stat("ep_dcp_queue_fill", aggregator.conn_queueFill,
                      add_stat, cookie);
      add_casted_stat("ep_dcp_items_sent", aggregator.conn_queueDrain,
                      add_stat, cookie);
      add_casted_stat("ep_dcp_items_remaining", aggregator.conn_queueRemaining,
                      add_stat, cookie);
      add_casted_stat("ep_dcp_queue_backfillremaining",
                      aggregator.conn_queueBackfillRemaining, add_stat, cookie);
      add_casted_stat("ep_dcp_num_running_backfills",
                      dcpConnMap_->getNumActiveSnoozingBackfills(), add_stat, cookie);
      add_casted_stat("ep_dcp_max_running_backfills",
                      dcpConnMap_->getMaxActiveSnoozingBackfills(), add_stat, cookie);
  
      dcpConnMap_->addStats(add_stat, cookie);
      return ENGINE_SUCCESS;
  }
  
  ENGINE_ERROR_CODE EventuallyPersistentEngine::doKeyStats(const void *cookie,
                                                           ADD_STAT add_stat,
                                                           uint16_t vbid,
                                                           const DocKey& key,
                                                           bool validate) {
      ENGINE_ERROR_CODE rv = ENGINE_FAILED;
  
      Item *it = NULL;
      std::shared_ptr<Item> diskItem;
      struct key_stats kstats;
  
      if (fetchLookupResult(cookie, &it)) {
          diskItem.reset(it); // Will be null if the key was not found
          if (!validate) {
              LOG(EXTENSION_LOG_DEBUG,
                  "Found lookup results for non-validating key "
                  "stat call. Would have leaked\n");
              diskItem.reset();
          }
      } else if (validate) {
          rv = kvBucket->statsVKey(key, vbid, cookie);
          if (rv == ENGINE_NOT_MY_VBUCKET || rv == ENGINE_KEY_ENOENT) {
              if (isDegradedMode()) {
                  return ENGINE_TMPFAIL;
              }
          }
          return rv;
      }
  
      rv = kvBucket->getKeyStats(key, vbid, cookie, kstats, false);
      if (rv == ENGINE_SUCCESS) {
          std::string valid("this_is_a_bug");
          if (validate) {
              if (kstats.dirty) {
                  valid.assign("dirty");
              } else if (diskItem.get()) {
                  valid.assign(kvBucket->validateKey(key, vbid, *diskItem));
              } else {
                  valid.assign("ram_but_not_disk");
              }
              LOG(EXTENSION_LOG_DEBUG, "doKeyStats key{%.*s} is %s\n",
                  int(key.size()), key.data(), valid.c_str());
          }
          add_casted_stat("key_is_dirty", kstats.dirty, add_stat, cookie);
          add_casted_stat("key_exptime", kstats.exptime, add_stat, cookie);
          add_casted_stat("key_flags", kstats.flags, add_stat, cookie);
          add_casted_stat("key_cas", kstats.cas, add_stat, cookie);
          add_casted_stat("key_vb_state", VBucket::toString(kstats.vb_state),
                          add_stat,
                          cookie);
          if (validate) {
              add_casted_stat("key_valid", valid.c_str(), add_stat, cookie);
          }
      }
      return rv;
  }
  
  ENGINE_ERROR_CODE EventuallyPersistentEngine::doVbIdFailoverLogStats(
                                                              const void *cookie,
                                                              ADD_STAT add_stat,
                                                              uint16_t vbid) {
      RCPtr<VBucket> vb = getVBucket(vbid);
      if(!vb) {
          return ENGINE_NOT_MY_VBUCKET;
      }
      vb->failovers->addStats(cookie, vb->getId(), add_stat);
      return ENGINE_SUCCESS;
  }
  
  
  ENGINE_ERROR_CODE EventuallyPersistentEngine::doAllFailoverLogStats(
                                                             const void *cookie,
                                                             ADD_STAT add_stat) {
      ENGINE_ERROR_CODE rv = ENGINE_SUCCESS;
      class StatVBucketVisitor : public VBucketVisitor {
      public:
          StatVBucketVisitor(const void *c, ADD_STAT a) :
              cookie(c), add_stat(a) {}
  
          void visitBucket(RCPtr<VBucket> &vb) override {
              vb->failovers->addStats(cookie, vb->getId(), add_stat);
          }
  
      private:
          const void *cookie;
          ADD_STAT add_stat;
      };
  
      StatVBucketVisitor svbv(cookie, add_stat);
      kvBucket->visit(svbv);
  
      return rv;
  }
  
  
  
  ENGINE_ERROR_CODE EventuallyPersistentEngine::doTimingStats(const void *cookie,
                                                             ADD_STAT add_stat) {
      add_casted_stat("bg_wait", stats.bgWaitHisto, add_stat, cookie);
      add_casted_stat("bg_load", stats.bgLoadHisto, add_stat, cookie);
      add_casted_stat("set_with_meta", stats.setWithMetaHisto, add_stat, cookie);
      add_casted_stat("bg_tap_wait", stats.tapBgWaitHisto, add_stat, cookie);
      add_casted_stat("bg_tap_load", stats.tapBgLoadHisto, add_stat, cookie);
      add_casted_stat("pending_ops", stats.pendingOpsHisto, add_stat, cookie);
  
      // Vbucket visitors
      add_casted_stat("access_scanner", stats.accessScannerHisto, add_stat, cookie);
      add_casted_stat("checkpoint_remover", stats.checkpointRemoverHisto, add_stat, cookie);
      add_casted_stat("item_pager", stats.itemPagerHisto, add_stat, cookie);
      add_casted_stat("expiry_pager", stats.expiryPagerHisto, add_stat, cookie);
  
      add_casted_stat("storage_age", stats.dirtyAgeHisto, add_stat, cookie);
  
      // Regular commands
      add_casted_stat("get_cmd", stats.getCmdHisto, add_stat, cookie);
      add_casted_stat("store_cmd", stats.storeCmdHisto, add_stat, cookie);
      add_casted_stat("arith_cmd", stats.arithCmdHisto, add_stat, cookie);
      add_casted_stat("get_stats_cmd", stats.getStatsCmdHisto, add_stat, cookie);
      // Admin commands
      add_casted_stat("get_vb_cmd", stats.getVbucketCmdHisto, add_stat, cookie);
      add_casted_stat("set_vb_cmd", stats.setVbucketCmdHisto, add_stat, cookie);
      add_casted_stat("del_vb_cmd", stats.delVbucketCmdHisto, add_stat, cookie);
      add_casted_stat("chk_persistence_cmd", stats.chkPersistenceHisto,
                      add_stat, cookie);
      // Tap commands
      add_casted_stat("tap_vb_set", stats.tapVbucketSetHisto, add_stat, cookie);
      add_casted_stat("tap_vb_reset", stats.tapVbucketResetHisto,
                      add_stat, cookie);
      add_casted_stat("tap_mutation", stats.tapMutationHisto, add_stat, cookie);
      // Misc
      add_casted_stat("notify_io", stats.notifyIOHisto, add_stat, cookie);
      add_casted_stat("batch_read", stats.getMultiHisto, add_stat, cookie);
  
      // Disk stats
      add_casted_stat("disk_insert", stats.diskInsertHisto, add_stat, cookie);
      add_casted_stat("disk_update", stats.diskUpdateHisto, add_stat, cookie);
      add_casted_stat("disk_del", stats.diskDelHisto, add_stat, cookie);
      add_casted_stat("disk_vb_del", stats.diskVBDelHisto, add_stat, cookie);
      add_casted_stat("disk_commit", stats.diskCommitHisto, add_stat, cookie);
  
      add_casted_stat("item_alloc_sizes", stats.itemAllocSizeHisto,
                      add_stat, cookie);
  
      // Checkpoint cursor stats
      add_casted_stat("persistence_cursor_get_all_items",
                      stats.persistenceCursorGetItemsHisto,
                      add_stat, cookie);
      add_casted_stat("dcp_cursors_get_all_items",
                      stats.dcpCursorsGetItemsHisto,
                      add_stat, cookie);
  
      return ENGINE_SUCCESS;
  }
  
  ENGINE_ERROR_CODE EventuallyPersistentEngine::doSchedulerStats(const void
                                                                  *cookie,
                                                                  ADD_STAT
                                                                  add_stat) {
      for (TaskId id : GlobalTask::allTaskIds) {
          add_casted_stat(GlobalTask::getTaskName(id),
                          stats.schedulingHisto[static_cast<int>(id)],
                          add_stat, cookie);
      }
  
      return ENGINE_SUCCESS;
  }
  
  ENGINE_ERROR_CODE EventuallyPersistentEngine::doRunTimeStats(const void
                                                                  *cookie,
                                                                  ADD_STAT
                                                                  add_stat) {
      for (TaskId id : GlobalTask::allTaskIds) {
          add_casted_stat(GlobalTask::getTaskName(id),
                          stats.taskRuntimeHisto[static_cast<int>(id)],
                          add_stat, cookie);
      }
  
      return ENGINE_SUCCESS;
  }
  
  ENGINE_ERROR_CODE EventuallyPersistentEngine::doDispatcherStats(const void
                                                                  *cookie,
                                                                  ADD_STAT
                                                                  add_stat) {
      ExecutorPool::get()->doWorkerStat(ObjectRegistry::getCurrentEngine(),
                                        cookie, add_stat);
      return ENGINE_SUCCESS;
  }
  
  ENGINE_ERROR_CODE EventuallyPersistentEngine::doWorkloadStats(const void
                                                                *cookie,
                                                                ADD_STAT
                                                                add_stat) {
      try {
          char statname[80] = {0};
          ExecutorPool* expool = ExecutorPool::get();
  
          int readers = expool->getNumReaders();
          checked_snprintf(statname, sizeof(statname), "ep_workload:num_readers");
          add_casted_stat(statname, readers, add_stat, cookie);
  
          int writers = expool->getNumWriters();
          checked_snprintf(statname, sizeof(statname), "ep_workload:num_writers");
          add_casted_stat(statname, writers, add_stat, cookie);
  
          int auxio = expool->getNumAuxIO();
          checked_snprintf(statname, sizeof(statname), "ep_workload:num_auxio");
          add_casted_stat(statname, auxio, add_stat, cookie);
  
          int nonio = expool->getNumNonIO();
          checked_snprintf(statname, sizeof(statname), "ep_workload:num_nonio");
          add_casted_stat(statname, nonio, add_stat, cookie);
  
          int max_readers = expool->getMaxReaders();
          checked_snprintf(statname, sizeof(statname), "ep_workload:max_readers");
          add_casted_stat(statname, max_readers, add_stat, cookie);
  
          int max_writers = expool->getMaxWriters();
          checked_snprintf(statname, sizeof(statname), "ep_workload:max_writers");
          add_casted_stat(statname, max_writers, add_stat, cookie);
  
          int max_auxio = expool->getMaxAuxIO();
          checked_snprintf(statname, sizeof(statname), "ep_workload:max_auxio");
          add_casted_stat(statname, max_auxio, add_stat, cookie);
  
          int max_nonio = expool->getMaxNonIO();
          checked_snprintf(statname, sizeof(statname), "ep_workload:max_nonio");
          add_casted_stat(statname, max_nonio, add_stat, cookie);
  
          int shards = workload->getNumShards();
          checked_snprintf(statname, sizeof(statname), "ep_workload:num_shards");
          add_casted_stat(statname, shards, add_stat, cookie);
  
          int numReadyTasks = expool->getNumReadyTasks();
          checked_snprintf(statname, sizeof(statname), "ep_workload:ready_tasks");
          add_casted_stat(statname, numReadyTasks, add_stat, cookie);
  
          int numSleepers = expool->getNumSleepers();
          checked_snprintf(statname, sizeof(statname),
                           "ep_workload:num_sleepers");
          add_casted_stat(statname, numSleepers, add_stat, cookie);
  
          expool->doTaskQStat(ObjectRegistry::getCurrentEngine(),
                              cookie, add_stat);
  
      } catch (std::exception& error) {
          LOG(EXTENSION_LOG_WARNING,
              "doWorkloadStats: Error building stats: %s", error.what());
      }
  
      return ENGINE_SUCCESS;
  }
  
  void EventuallyPersistentEngine::addSeqnoVbStats(const void *cookie,
                                                   ADD_STAT add_stat,
                                                   const RCPtr<VBucket> &vb) {
      // MB-19359: An atomic read of vbucket state without acquiring the
      // reader lock for state should suffice here.
      uint64_t relHighSeqno = vb->getHighSeqno();
      if (vb->getState() != vbucket_state_active) {
          snapshot_info_t info = vb->checkpointManager.getSnapshotInfo();
          relHighSeqno = info.range.end;
      }
  
      try {
          char buffer[64];
          failover_entry_t entry = vb->failovers->getLatestEntry();
          checked_snprintf(buffer, sizeof(buffer), "vb_%d:high_seqno",
                           vb->getId());
          add_casted_stat(buffer, relHighSeqno, add_stat, cookie);
          checked_snprintf(buffer, sizeof(buffer), "vb_%d:abs_high_seqno",
                           vb->getId());
          add_casted_stat(buffer, vb->getHighSeqno(), add_stat, cookie);
          checked_snprintf(buffer, sizeof(buffer), "vb_%d:last_persisted_seqno",
                           vb->getId());
          add_casted_stat(buffer, vb->getPersistenceSeqno(), add_stat, cookie);
          checked_snprintf(buffer, sizeof(buffer), "vb_%d:uuid", vb->getId());
          add_casted_stat(buffer, entry.vb_uuid, add_stat, cookie);
          checked_snprintf(buffer, sizeof(buffer), "vb_%d:purge_seqno",
                           vb->getId());
          add_casted_stat(buffer, vb->getPurgeSeqno(), add_stat, cookie);
          const snapshot_range_t range = vb->getPersistedSnapshot();
          checked_snprintf(buffer, sizeof(buffer),
                           "vb_%d:last_persisted_snap_start",
                           vb->getId());
          add_casted_stat(buffer, range.start, add_stat, cookie);
          checked_snprintf(buffer, sizeof(buffer),
                           "vb_%d:last_persisted_snap_end",
                           vb->getId());
          add_casted_stat(buffer, range.end, add_stat, cookie);
      } catch (std::exception& error) {
          LOG(EXTENSION_LOG_WARNING,
              "addSeqnoVbStats: error building stats: %s", error.what());
      }
  }
  
  ENGINE_ERROR_CODE EventuallyPersistentEngine::doSeqnoStats(const void *cookie,
                                                            ADD_STAT add_stat,
                                                            const char* stat_key,
                                                            int nkey) {
      if (nkey > 14) {
          std::string value(stat_key + 14, nkey - 14);
  
          try {
              checkNumeric(value.c_str());
          } catch(std::runtime_error &) {
              return ENGINE_EINVAL;
          }
  
          int vbucket = atoi(value.c_str());
          RCPtr<VBucket> vb = getVBucket(vbucket);
          if (!vb || vb->getState() == vbucket_state_dead) {
              return ENGINE_NOT_MY_VBUCKET;
          }
  
          addSeqnoVbStats(cookie, add_stat, vb);
  
          return ENGINE_SUCCESS;
      }
  
      auto vbuckets = kvBucket->getVBuckets().getBuckets();
      for (auto vbid : vbuckets) {
          RCPtr<VBucket> vb = getVBucket(vbid);
          if (vb) {
              addSeqnoVbStats(cookie, add_stat, vb);
          }
      }
      return ENGINE_SUCCESS;
  }
  
  void EventuallyPersistentEngine::addLookupAllKeys(const void *cookie,
                                                    ENGINE_ERROR_CODE err) {
      LockHolder lh(lookupMutex);
      allKeysLookups[cookie] = err;
  }
  
  void EventuallyPersistentEngine::runDefragmenterTask(void) {
      kvBucket->runDefragmenterTask();
  }
  
  bool EventuallyPersistentEngine::runAccessScannerTask(void) {
      return kvBucket->runAccessScannerTask();
  }
  
  void EventuallyPersistentEngine::runVbStatePersistTask(int vbid) {
      kvBucket->runVbStatePersistTask(vbid);
  }
  
  ENGINE_ERROR_CODE EventuallyPersistentEngine::getStats(const void* cookie,
                                                         const char* stat_key,
                                                         int nkey,
                                                         ADD_STAT add_stat) {
      BlockTimer timer(&stats.getStatsCmdHisto);
      if (stat_key != NULL) {
          LOG(EXTENSION_LOG_DEBUG, "stats %.*s", nkey, stat_key);
      } else {
          LOG(EXTENSION_LOG_DEBUG, "stats engine");
      }
  
      const std::string statKey(stat_key, nkey);
  
      ENGINE_ERROR_CODE rv = ENGINE_KEY_ENOENT;
      if (stat_key == NULL) {
          rv = doEngineStats(cookie, add_stat);
      } else if (nkey > 7 && cb_isPrefix(statKey, "tapagg ")) {
          if (configuration.isTap()) {
              rv = doConnAggStats(
                      cookie, add_stat, stat_key + 7, nkey - 7, TAP_CONN);
          }
      } else if (nkey > 7 && cb_isPrefix(statKey, "dcpagg ")) {
          rv = doConnAggStats(cookie, add_stat, stat_key + 7, nkey - 7, DCP_CONN);
      } else if (statKey == "tap") {
          if (configuration.isTap()) {
              rv = doTapStats(cookie, add_stat);
          }
      } else if (statKey == "dcp") {
          rv = doDcpStats(cookie, add_stat);
      } else if (statKey == "hash") {
          rv = doHashStats(cookie, add_stat);
      } else if (statKey == "vbucket") {
          rv = doVBucketStats(cookie, add_stat, stat_key, nkey, false, false);
      } else if (cb_isPrefix(statKey, "vbucket-details")) {
          rv = doVBucketStats(cookie, add_stat, stat_key, nkey, false, true);
      } else if (cb_isPrefix(statKey, "vbucket-seqno")) {
          rv = doSeqnoStats(cookie, add_stat, stat_key, nkey);
      } else if (statKey == "prev-vbucket") {
          rv = doVBucketStats(cookie, add_stat, stat_key, nkey, true, false);
      } else if (cb_isPrefix(statKey, "checkpoint")) {
          rv = doCheckpointStats(cookie, add_stat, stat_key, nkey);
      } else if (statKey == "timings") {
          rv = doTimingStats(cookie, add_stat);
      } else if (statKey == "dispatcher") {
          rv = doDispatcherStats(cookie, add_stat);
      } else if (statKey == "scheduler") {
          rv = doSchedulerStats(cookie, add_stat);
      } else if (statKey == "runtimes") {
          rv = doRunTimeStats(cookie, add_stat);
      } else if (statKey == "memory") {
          rv = doMemoryStats(cookie, add_stat);
      } else if (statKey == "uuid") {
          add_casted_stat("uuid", configuration.getUuid(), add_stat, cookie);
          rv = ENGINE_SUCCESS;
      } else if (nkey > 4 && cb_isPrefix(statKey, "key ")) {
          std::string key;
          std::string vbid;
          std::string s_key(statKey.substr(4, nkey - 4));
          std::stringstream ss(s_key);
          ss >> key;
          ss >> vbid;
          uint16_t vbucket_id(0);
          parseUint16(vbid.c_str(), &vbucket_id);
          // Non-validating, non-blocking version
          // TODO: Collection - getStats needs DocNamespace
          rv = doKeyStats(cookie, add_stat, vbucket_id,
                          DocKey(key, DocNamespace::DefaultCollection), false);
      } else if (nkey > 5 && cb_isPrefix(statKey, "vkey ")) {
          std::string key;
          std::string vbid;
          std::string s_key(statKey.substr(5, nkey - 5));
          std::stringstream ss(s_key);
          ss >> key;
          ss >> vbid;
          uint16_t vbucket_id(0);
          parseUint16(vbid.c_str(), &vbucket_id);
          // Validating version; blocks
          // TODO: Collection - getStats needs DocNamespace
          rv = doKeyStats(cookie, add_stat, vbucket_id,
                          DocKey(key, DocNamespace::DefaultCollection), true);
      } else if (statKey == "kvtimings") {
          getKVBucket()->addKVStoreTimingStats(add_stat, cookie);
          rv = ENGINE_SUCCESS;
      } else if (statKey == "kvstore") {
          getKVBucket()->addKVStoreStats(add_stat, cookie);
          rv = ENGINE_SUCCESS;
      } else if (statKey == "warmup") {
          const auto* warmup = getKVBucket()->getWarmup();
          if (warmup != nullptr) {
              warmup->addStats(add_stat, cookie);
              rv = ENGINE_SUCCESS;
          }
  
      } else if (statKey == "info") {
          add_casted_stat("info", get_stats_info(), add_stat, cookie);
          rv = ENGINE_SUCCESS;
      } else if (statKey == "allocator") {
          char buffer[64 * 1024];
          MemoryTracker::getInstance(*getServerApiFunc()->alloc_hooks)->
                  getDetailedStats(buffer, sizeof(buffer));
          add_casted_stat("detailed", buffer, add_stat, cookie);
          rv = ENGINE_SUCCESS;
      } else if (statKey == "config") {
          configuration.addStats(add_stat, cookie);
          rv = ENGINE_SUCCESS;
      } else if (nkey > 15 && cb_isPrefix(statKey, "tap-vbtakeover")) {
          std::string tStream;
          std::string vbid;
          std::string buffer(statKey.substr(15, nkey - 15));
          std::stringstream ss(buffer);
          ss >> vbid;
          ss >> tStream;
          uint16_t vbucket_id(0);
          parseUint16(vbid.c_str(), &vbucket_id);
          rv = doTapVbTakeoverStats(cookie, add_stat, tStream, vbucket_id);
      } else if (nkey > 15 && cb_isPrefix(statKey, "dcp-vbtakeover")) {
          std::string tStream;
          std::string vbid;
          std::string buffer(statKey.substr(15, nkey - 15));
          std::stringstream ss(buffer);
          ss >> vbid;
          ss >> tStream;
          uint16_t vbucket_id(0);
          parseUint16(vbid.c_str(), &vbucket_id);
          rv = doDcpVbTakeoverStats(cookie, add_stat, tStream, vbucket_id);
      } else if (statKey == "workload") {
          return doWorkloadStats(cookie, add_stat);
      } else if (cb_isPrefix(statKey, "failovers")) {
          if (nkey == 9) {
              rv = doAllFailoverLogStats(cookie, add_stat);
          } else if (statKey.compare(std::string("failovers").length(),
                                     std::string(" ").length(),
                                     " ") == 0) {
              std::string vbid;
              std::string s_key(statKey.substr(10, nkey - 10));
              std::stringstream ss(s_key);
              ss >> vbid;
              uint16_t vbucket_id(0);
              parseUint16(vbid.c_str(), &vbucket_id);
              rv = doVbIdFailoverLogStats(cookie, add_stat, vbucket_id);
          }
      } else if (cb_isPrefix(statKey, "diskinfo")) {
          if (nkey == 8) {
              return kvBucket->getFileStats(cookie, add_stat);
          } else if ((nkey == 15) &&
                  (statKey.compare(std::string("diskinfo").length() + 1,
                                   std::string("detail").length(),
                                   "detail") == 0)) {
              return kvBucket->getPerVBucketDiskStats(cookie, add_stat);
          } else {
              return ENGINE_EINVAL;
          }
      }
  
      return rv;
  }
  
  ENGINE_ERROR_CODE EventuallyPersistentEngine::observe(
                                         const void* cookie,
                                         protocol_binary_request_header *request,
                                         ADD_RESPONSE response,
                                         DocNamespace docNamespace) {
      protocol_binary_request_no_extras *req =
          (protocol_binary_request_no_extras*)request;
  
      size_t offset = 0;
      const uint8_t* data = req->bytes + sizeof(req->bytes);
      uint32_t data_len = ntohl(req->message.header.request.bodylen);
      std::stringstream result;
  
      while (offset < data_len) {
          uint16_t vb_id;
          uint16_t keylen;
  
          // Parse a key
          if (data_len - offset < 4) {
              std::string msg("Invalid packet structure");
              return sendResponse(response, NULL, 0, 0, 0, msg.c_str(),
                                  msg.length(),
                                  PROTOCOL_BINARY_RAW_BYTES,
                                  PROTOCOL_BINARY_RESPONSE_EINVAL, 0,
                                  cookie);
          }
  
          memcpy(&vb_id, data + offset, sizeof(uint16_t));
          vb_id = ntohs(vb_id);
          offset += sizeof(uint16_t);
  
          memcpy(&keylen, data + offset, sizeof(uint16_t));
          keylen = ntohs(keylen);
          offset += sizeof(uint16_t);
  
          if (data_len - offset < keylen) {
              std::string msg("Invalid packet structure");
              return sendResponse(response, NULL, 0, 0, 0, msg.c_str(),
                                  msg.length(),
                                  PROTOCOL_BINARY_RAW_BYTES,
                                  PROTOCOL_BINARY_RESPONSE_EINVAL, 0,
                                  cookie);
          }
  
          DocKey key(data + offset, keylen, docNamespace);
          offset += keylen;
          LOG(EXTENSION_LOG_DEBUG, "Observing key{%.*s} in vb:%" PRIu16,
              int(key.size()), key.data(), vb_id);
  
          // Get key stats
          uint16_t keystatus = 0;
          struct key_stats kstats;
          memset(&kstats, 0, sizeof(key_stats));
          ENGINE_ERROR_CODE rv = kvBucket->getKeyStats(key, vb_id, cookie, kstats,
                                                       true);
          if (rv == ENGINE_SUCCESS) {
              if (kstats.logically_deleted) {
                  keystatus = OBS_STATE_LOGICAL_DEL;
              } else if (!kstats.dirty) {
                  keystatus = OBS_STATE_PERSISTED;
              } else {
                  keystatus = OBS_STATE_NOT_PERSISTED;
              }
          } else if (rv == ENGINE_KEY_ENOENT) {
              keystatus = OBS_STATE_NOT_FOUND;
          } else if (rv == ENGINE_NOT_MY_VBUCKET) {
              return sendNotMyVBucketResponse(response, cookie, 0);
          } else if (rv == ENGINE_EWOULDBLOCK) {
              return rv;
          } else {
              std::string msg("Internal error");
              return sendResponse(response, NULL, 0, 0, 0, msg.c_str(),
                                  msg.length(),
                                  PROTOCOL_BINARY_RAW_BYTES,
                                  PROTOCOL_BINARY_RESPONSE_EINTERNAL, 0,
                                  cookie);
          }
  
          // Put the result into a response buffer
          vb_id = htons(vb_id);
          keylen = htons(keylen);
          uint64_t cas = htonll(kstats.cas);
          result.write((char*) &vb_id, sizeof(uint16_t));
          result.write((char*) &keylen, sizeof(uint16_t));
          result.write(reinterpret_cast<const char*>(key.data()), key.size());
          result.write((char*) &keystatus, sizeof(uint8_t));
          result.write((char*) &cas, sizeof(uint64_t));
      }
  
      uint64_t persist_time = 0;
      double queue_size = static_cast<double>(stats.diskQueueSize);
      double item_trans_time = kvBucket->getTransactionTimePerItem();
  
      if (item_trans_time > 0 && queue_size > 0) {
          persist_time = static_cast<uint32_t>(queue_size * item_trans_time);
      }
      persist_time = persist_time << 32;
  
      return sendResponse(response, NULL, 0, 0, 0, result.str().data(),
                          result.str().length(),
                          PROTOCOL_BINARY_RAW_BYTES,
                          PROTOCOL_BINARY_RESPONSE_SUCCESS, persist_time,
                          cookie);
  }
  
  ENGINE_ERROR_CODE EventuallyPersistentEngine::observe_seqno(
                                         const void* cookie,
                                         protocol_binary_request_header *request,
                                         ADD_RESPONSE response) {
      protocol_binary_request_no_extras *req =
                            (protocol_binary_request_no_extras*)request;
      const char* data = reinterpret_cast<const char*>(req->bytes) +
                                                     sizeof(req->bytes);
      uint16_t vb_id;
      uint64_t vb_uuid;
      uint8_t  format_type;
      uint64_t last_persisted_seqno;
      uint64_t current_seqno;
  
      std::stringstream result;
  
      vb_id = ntohs(req->message.header.request.vbucket);
      memcpy(&vb_uuid, data, sizeof(uint64_t));
      vb_uuid = ntohll(vb_uuid);
  
      LOG(EXTENSION_LOG_DEBUG, "Observing vbucket: %d with uuid: %" PRIu64,
                               vb_id, vb_uuid);
  
      RCPtr<VBucket> vb = kvBucket->getVBucket(vb_id);
  
      if (!vb) {
          return sendNotMyVBucketResponse(response, cookie, 0);
      }
  
      ReaderLockHolder rlh(vb->getStateLock());
      if (vb->getState() == vbucket_state_dead) {
          return sendNotMyVBucketResponse(response, cookie, 0);
      }
  
      //Check if the vb uuid matches with the latest entry
      failover_entry_t entry = vb->failovers->getLatestEntry();
  
      if (vb_uuid != entry.vb_uuid) {
         uint64_t failover_highseqno = 0;
         uint64_t latest_uuid;
         bool found = vb->failovers->getLastSeqnoForUUID(vb_uuid, &failover_highseqno);
         if (!found) {
             return sendResponse(response, NULL, 0, 0, 0, 0, 0,
                                 PROTOCOL_BINARY_RAW_BYTES,
                                 PROTOCOL_BINARY_RESPONSE_KEY_ENOENT, 0,
                                 cookie);
         }
  
         format_type = 1;
         last_persisted_seqno = htonll(vb->getPersistenceSeqno());
         current_seqno = htonll(vb->getHighSeqno());
         latest_uuid = htonll(entry.vb_uuid);
         vb_id = htons(vb_id);
         vb_uuid = htonll(vb_uuid);
         failover_highseqno = htonll(failover_highseqno);
  
         result.write((char*) &format_type, sizeof(uint8_t));
         result.write((char*) &vb_id, sizeof(uint16_t));
         result.write((char*) &latest_uuid, sizeof(uint64_t));
         result.write((char*) &last_persisted_seqno, sizeof(uint64_t));
         result.write((char*) &current_seqno, sizeof(uint64_t));
         result.write((char*) &vb_uuid, sizeof(uint64_t));
         result.write((char*) &failover_highseqno, sizeof(uint64_t));
      } else {
          format_type = 0;
          last_persisted_seqno = htonll(vb->getPersistenceSeqno());
          current_seqno = htonll(vb->getHighSeqno());
          vb_id   =  htons(vb_id);
          vb_uuid =  htonll(vb_uuid);
  
          result.write((char*) &format_type, sizeof(uint8_t));
          result.write((char*) &vb_id, sizeof(uint16_t));
          result.write((char*) &vb_uuid, sizeof(uint64_t));
          result.write((char*) &last_persisted_seqno, sizeof(uint64_t));
          result.write((char*) &current_seqno, sizeof(uint64_t));
      }
  
      return sendResponse(response, NULL, 0, 0, 0, result.str().data(),
                          result.str().length(),
                          PROTOCOL_BINARY_RAW_BYTES,
                          PROTOCOL_BINARY_RESPONSE_SUCCESS, 0,
                          cookie);
  }
  
  ENGINE_ERROR_CODE EventuallyPersistentEngine::touch(const void *cookie,
                                         protocol_binary_request_header *request,
                                         ADD_RESPONSE response,
                                         DocNamespace docNamespace)
  {
      if (request->request.extlen != 4 || request->request.keylen == 0) {
          return sendResponse(response, NULL, 0, NULL, 0, NULL, 0,
                              PROTOCOL_BINARY_RAW_BYTES,
                              PROTOCOL_BINARY_RESPONSE_EINVAL, 0, cookie);
      }
  
      protocol_binary_request_touch *t =
                       reinterpret_cast<protocol_binary_request_touch*>(request);
      const uint8_t* key = t->bytes + sizeof(t->bytes);
      uint32_t exptime = ntohl(t->message.body.expiration);
      uint16_t nkey = ntohs(request->request.keylen);
      uint16_t vbucket = ntohs(request->request.vbucket);
  
      // try to get the object
      DocKey k(key, nkey, docNamespace);
  
      if (exptime != 0) {
          exptime = serverApi->core->abstime(serverApi->core->realtime(exptime));
      }
      GetValue gv(kvBucket->getAndUpdateTtl(k, vbucket, cookie, (time_t)exptime));
      ENGINE_ERROR_CODE rv = gv.getStatus();
      if (rv == ENGINE_SUCCESS) {
          Item *it = gv.getValue();
          if (request->request.opcode == PROTOCOL_BINARY_CMD_TOUCH) {
              rv = sendResponse(response, NULL, 0, NULL, 0, NULL, 0,
                                PROTOCOL_BINARY_RAW_BYTES,
                                PROTOCOL_BINARY_RESPONSE_SUCCESS, it->getCas(),
                                cookie);
          } else { // GET and TOUCH
              uint32_t flags = it->getFlags();
              rv = sendResponse(response, NULL, 0, &flags, sizeof(flags),
                                it->getData(), it->getNBytes(),
                                it->getDataType(),
                                PROTOCOL_BINARY_RESPONSE_SUCCESS, it->getCas(),
                                cookie);
              ++stats.numOpsGet;
          }
          ++stats.numOpsStore;
          delete it;
      } else if (rv == ENGINE_KEY_EEXISTS) {
          if (isDegradedMode()) {
              std::string msg("Temporary Failure");
              rv = sendResponse(response, NULL, 0, NULL, 0, msg.c_str(),
                                msg.length(), PROTOCOL_BINARY_RAW_BYTES,
                                PROTOCOL_BINARY_RESPONSE_ETMPFAIL, 0, cookie);
          } else {
              std::string msg("Lock Error");
              rv = sendResponse(response, NULL, 0, NULL, 0, msg.c_str(),
                                msg.length(), PROTOCOL_BINARY_RAW_BYTES,
                                PROTOCOL_BINARY_RESPONSE_ETMPFAIL, 0, cookie);
          }
      } else if (rv == ENGINE_KEY_ENOENT) {
          if (isDegradedMode()) {
              std::string msg("Temporary Failure");
              rv = sendResponse(response, NULL, 0, NULL, 0, msg.c_str(),
                                msg.length(), PROTOCOL_BINARY_RAW_BYTES,
                                PROTOCOL_BINARY_RESPONSE_ETMPFAIL, 0, cookie);
          } else if (request->request.opcode == PROTOCOL_BINARY_CMD_GATQ) {
              // GATQ should not return response upon cache miss
              rv = ENGINE_SUCCESS;
          } else {
              std::string msg("Not Found");
              rv = sendResponse(response, NULL, 0, NULL, 0, msg.c_str(),
                                msg.length(), PROTOCOL_BINARY_RAW_BYTES,
                                PROTOCOL_BINARY_RESPONSE_KEY_ENOENT, 0, cookie);
          }
      } else if (rv == ENGINE_NOT_MY_VBUCKET) {
          if (isDegradedMode()) {
              std::string msg("Temporary Failure");
              rv = sendResponse(response, NULL, 0, NULL, 0, msg.c_str(),
                                msg.length(), PROTOCOL_BINARY_RAW_BYTES,
                                PROTOCOL_BINARY_RESPONSE_ETMPFAIL, 0, cookie);
          } else {
              rv = sendNotMyVBucketResponse(response, cookie, 0);
          }
      }
  
      return rv;
  }
  
  ENGINE_ERROR_CODE EventuallyPersistentEngine::deregisterTapClient(
                                         const void *cookie,
                                         protocol_binary_request_header *request,
                                         ADD_RESPONSE response)
  {
      uint64_t cas = ntohll(request->request.cas);
  
      std::string tap_name = "eq_tapq:";
      std::string cName((const char*)request->bytes + sizeof(request->bytes) +
                        request->request.extlen, ntohs(request->request.keylen));
      tap_name.append(cName);
  
      // Close the tap connection for the registered TAP client and remove
      // its checkpoint cursors.
      bool rv = tapConnMap->closeConnectionByName(tap_name);
      if (!rv) {
          // If the tap connection is not found, we still need to remove
          /// its checkpoint cursors.
          const VBucketMap &vbuckets = getKVBucket()->getVBuckets();
          for (VBucketMap::id_type vbid = 0; vbid < vbuckets.getSize(); ++vbid) {
              RCPtr<VBucket> vb = vbuckets.getBucket(vbid);
              if (!vb) {
                  continue;
              }
              vb->checkpointManager.removeCursor(tap_name);
          }
      }
  
      return sendResponse(response, NULL, 0, NULL, 0, NULL, 0,
                          PROTOCOL_BINARY_RAW_BYTES,
                          PROTOCOL_BINARY_RESPONSE_SUCCESS, cas, cookie);
  }
  
  ENGINE_ERROR_CODE
  EventuallyPersistentEngine::handleCheckpointCmds(const void *cookie,
                                             protocol_binary_request_header *req,
                                             ADD_RESPONSE response)
  {
      std::stringstream msg;
      uint16_t vbucket = ntohs(req->request.vbucket);
      RCPtr<VBucket> vb = getVBucket(vbucket);
  
      if (!vb) {
          return sendNotMyVBucketResponse(response, cookie, 0);
      }
  
      int16_t status = PROTOCOL_BINARY_RESPONSE_SUCCESS;
  
      switch (req->request.opcode) {
      case PROTOCOL_BINARY_CMD_LAST_CLOSED_CHECKPOINT:
          {
              uint64_t checkpointId = vb->checkpointManager.
                                      getLastClosedCheckpointId();
              checkpointId = htonll(checkpointId);
              return sendResponse(response, NULL, 0, NULL, 0,
                                  &checkpointId, sizeof(checkpointId),
                                  PROTOCOL_BINARY_RAW_BYTES,
                                  status, 0, cookie);
          }
          break;
      case PROTOCOL_BINARY_CMD_CREATE_CHECKPOINT:
          if (vb->getState() != vbucket_state_active) {
              return sendNotMyVBucketResponse(response, cookie, 0);
  
          } else {
              uint64_t checkpointId = htonll(vb->checkpointManager.
                                             createNewCheckpoint());
              getKVBucket()->wakeUpFlusher();
  
              uint64_t persistedChkId = htonll(kvBucket->
                                     getLastPersistedCheckpointId(vb->getId()));
              char val[128];
              memcpy(val, &checkpointId, sizeof(uint64_t));
              memcpy(val + sizeof(uint64_t), &persistedChkId, sizeof(uint64_t));
              return sendResponse(response, NULL, 0, NULL, 0,
                                  val, sizeof(uint64_t) * 2,
                                  PROTOCOL_BINARY_RAW_BYTES,
                                  status, 0, cookie);
          }
          break;
      case PROTOCOL_BINARY_CMD_CHECKPOINT_PERSISTENCE:
          {
              uint16_t keylen = ntohs(req->request.keylen);
              uint32_t bodylen = ntohl(req->request.bodylen);
              if ((bodylen - keylen) == 0) {
                  msg <<
                  "No checkpoint id is given for CMD_CHECKPOINT_PERSISTENCE!!!";
                  status = PROTOCOL_BINARY_RESPONSE_EINVAL;
              } else {
                  uint64_t chk_id;
                  memcpy(&chk_id, req->bytes + sizeof(req->bytes) + keylen,
                         bodylen - keylen);
                  chk_id = ntohll(chk_id);
                  void *es = getEngineSpecific(cookie);
                  if (!es) {
                      vb->addHighPriorityVBEntry(chk_id, cookie, false);
                      storeEngineSpecific(cookie, this);
                      // Wake up the flusher if it is idle.
                      getKVBucket()->wakeUpFlusher();
                      return ENGINE_EWOULDBLOCK;
                  } else {
                      storeEngineSpecific(cookie, NULL);
                      LOG(EXTENSION_LOG_INFO,
                          "Checkpoint %" PRIu64 " persisted for vbucket %d.",
                          chk_id, vbucket);
                  }
              }
          }
          break;
      default:
          {
              msg << "Unknown checkpoint command opcode: " <<
                     req->request.opcode;
              status = PROTOCOL_BINARY_RESPONSE_UNKNOWN_COMMAND;
          }
      }
  
      return sendResponse(response, NULL, 0, NULL, 0,
                          msg.str().c_str(), msg.str().length(),
                          PROTOCOL_BINARY_RAW_BYTES,
                          status, 0, cookie);
  }
  
  ENGINE_ERROR_CODE
  EventuallyPersistentEngine::handleSeqnoCmds(const void *cookie,
                                             protocol_binary_request_header *req,
                                             ADD_RESPONSE response)
  {
      std::stringstream msg;
      uint16_t vbucket = ntohs(req->request.vbucket);
      RCPtr<VBucket> vb = getVBucket(vbucket);
  
      if (!vb) {
          return sendNotMyVBucketResponse(response, cookie, 0);
      }
  
      int16_t status = PROTOCOL_BINARY_RESPONSE_SUCCESS;
  
      uint16_t extlen = req->request.extlen;
      uint32_t bodylen = ntohl(req->request.bodylen);
      if ((bodylen - extlen) != 0) {
          msg << "Invalid packet format";
          status = PROTOCOL_BINARY_RESPONSE_EINVAL;
      } else {
          uint64_t seqno;
          memcpy(&seqno, req->bytes + sizeof(req->bytes), 8);
          seqno = ntohll(seqno);
          void *es = getEngineSpecific(cookie);
          if (!es) {
              uint16_t persisted_seqno = vb->getPersistenceSeqno();
              if (seqno > persisted_seqno) {
                  vb->addHighPriorityVBEntry(seqno, cookie, true);
                  storeEngineSpecific(cookie, this);
                  return ENGINE_EWOULDBLOCK;
              }
          } else {
              storeEngineSpecific(cookie, NULL);
              LOG(EXTENSION_LOG_INFO,
                  "Sequence number %" PRIu64 " persisted for vbucket %d.",
                  seqno, vbucket);
          }
      }
  
      return sendResponse(response, NULL, 0, NULL, 0,
                          msg.str().c_str(), msg.str().length(),
                          PROTOCOL_BINARY_RAW_BYTES,
                          status, 0, cookie);
  }
  
  ENGINE_ERROR_CODE
  EventuallyPersistentEngine::resetReplicationChain(const void *cookie,
                                             protocol_binary_request_header *req,
                                             ADD_RESPONSE response) {
      (void) req;
      tapConnMap->resetReplicaChain();
      return sendResponse(response, NULL, 0, NULL, 0, NULL, 0,
                          PROTOCOL_BINARY_RAW_BYTES,
                          PROTOCOL_BINARY_RESPONSE_SUCCESS, 0, cookie);
  }
  
  ENGINE_ERROR_CODE EventuallyPersistentEngine::getMeta(const void* cookie,
                                       protocol_binary_request_get_meta *request,
                                       ADD_RESPONSE response,
                                       DocNamespace docNamespace)
  {
      if (request->message.header.request.keylen == 0) {
          return sendResponse(response, NULL, 0, NULL, 0, NULL, 0,
                              PROTOCOL_BINARY_RAW_BYTES,
                              PROTOCOL_BINARY_RESPONSE_EINVAL, 0, cookie);
      }
  
      uint8_t extlen = request->message.header.request.extlen;
      DocKey key((request->bytes + sizeof(request->bytes) + extlen),
                 (size_t)ntohs(request->message.header.request.keylen),
                  docNamespace);
      uint16_t vbucket = ntohs(request->message.header.request.vbucket);
  
      ItemMetaData metadata;
      uint32_t deleted;
  
      ENGINE_ERROR_CODE rv = kvBucket->getMetaData(key, vbucket, cookie, metadata,
                                                   deleted);
  
      if (rv == ENGINE_SUCCESS) {
  
          const uint8_t metalen = 20;
          uint8_t meta[metalen];
          deleted = htonl(deleted);
          uint32_t flags = metadata.flags;
          uint32_t exp = htonl(metadata.exptime);
          uint64_t seqno = htonll(metadata.revSeqno);
  
          memcpy(meta, &deleted, 4);
          memcpy(meta + 4, &flags, 4);
          memcpy(meta + 8, &exp, 4);
          memcpy(meta + 12, &seqno, 8);
  
          rv = sendResponse(response, NULL, 0, (const void *)meta,
                            metalen, NULL, 0,
                            PROTOCOL_BINARY_RAW_BYTES,
                            PROTOCOL_BINARY_RESPONSE_SUCCESS,
                            metadata.cas, cookie);
      } else if (rv == ENGINE_NOT_MY_VBUCKET) {
          rv = sendNotMyVBucketResponse(response, cookie, 0);
      } else if (rv != ENGINE_EWOULDBLOCK) {
          if (rv == ENGINE_KEY_ENOENT &&
              request->message.header.request.opcode == PROTOCOL_BINARY_CMD_GETQ_META) {
              rv = ENGINE_SUCCESS;
          } else {
              rv = sendResponse(response, NULL, 0, NULL, 0, NULL, 0,
                                PROTOCOL_BINARY_RAW_BYTES,
                                serverApi->cookie->engine_error2mcbp(cookie, rv),
                                metadata.cas, cookie);
          }
      }
  
      return rv;
  }
  
  protocol_binary_response_status EventuallyPersistentEngine::decodeWithMetaOptions(
                                protocol_binary_request_delete_with_meta* request,
                                GenerateCas& generateCas,
                                bool& skipConflictResolution,
                                int& keyOffset) {
      uint8_t extlen = request->message.header.request.extlen;
      keyOffset = 0;
      bool forceFlag = false;
      if (extlen == 28 || extlen == 30) {
          uint32_t options;
          memcpy(&options, request->bytes + sizeof(request->bytes),
                 sizeof(options));
          options = ntohl(options);
          keyOffset = 4; // 4 bytes for options
  
          if (options & SKIP_CONFLICT_RESOLUTION_FLAG) {
              skipConflictResolution = true;
          }
  
          if (options & FORCE_ACCEPT_WITH_META_OPS) {
              forceFlag = true;
          }
  
          if (options & REGENERATE_CAS) {
              generateCas = GenerateCas::Yes;
          }
      }
  
      // Validate options
      // If regenerate CAS then skip conflict resolution must be set
      // If LWW force must be set
      // If !LWW then force cannot be set).
      if ((generateCas == GenerateCas::Yes && !skipConflictResolution) ||
          (configuration.getConflictResolutionType() == "lww" && !forceFlag &&
           generateCas == GenerateCas::No) ||
          (forceFlag && (configuration.getConflictResolutionType() != "lww"))) {
          return PROTOCOL_BINARY_RESPONSE_EINVAL;
      }
  
  
      return PROTOCOL_BINARY_RESPONSE_SUCCESS;
  }
  
  ENGINE_ERROR_CODE EventuallyPersistentEngine::setWithMeta(const void* cookie,
                                  protocol_binary_request_set_with_meta *request,
                                  ADD_RESPONSE response,
                                  DocNamespace docNamespace)
  {
      // revid_nbytes, flags and exptime is mandatory fields.. and we need a key
      uint8_t extlen = request->message.header.request.extlen;
      uint16_t keylen = ntohs(request->message.header.request.keylen);
      // extlen, the size dicates what is encoded.
      // 24 = no nmeta and no options
      // 26 = nmeta
      // 28 = options (4-byte field)
      // 30 = options and nmeta (options followed by nmeta)
      // so 27, 25 etc... are illegal
      if ((extlen != 24 && extlen != 26 && extlen != 28  && extlen != 30)
          || keylen == 0) {
          return sendResponse(response, NULL, 0, NULL, 0, NULL, 0,
                              PROTOCOL_BINARY_RAW_BYTES,
                              PROTOCOL_BINARY_RESPONSE_EINVAL, 0, cookie);
      }
  
      if (isDegradedMode()) {
          return sendResponse(response, NULL, 0, NULL, 0, NULL, 0,
                              PROTOCOL_BINARY_RAW_BYTES,
                              PROTOCOL_BINARY_RESPONSE_ETMPFAIL,
                              0, cookie);
      }
  
      uint8_t opcode = request->message.header.request.opcode;
      uint8_t* key = request->bytes + sizeof(request->bytes);
      uint16_t vbucket = ntohs(request->message.header.request.vbucket);
      uint32_t bodylen = ntohl(request->message.header.request.bodylen);
      uint8_t datatype = request->message.header.request.datatype;
      size_t vallen = bodylen - keylen - extlen;
  
      uint32_t flags = request->message.body.flags;
      uint32_t expiration = ntohl(request->message.body.expiration);
      uint64_t seqno = ntohll(request->message.body.seqno);
      uint64_t cas = ntohll(request->message.body.cas);
  
      bool skipConflictResolution = false;
      GenerateCas generateCas = GenerateCas::No;
      std::unique_ptr<ExtendedMetaData> emd;
      int keyOffset = 0;
      protocol_binary_response_status error = PROTOCOL_BINARY_RESPONSE_SUCCESS;
      if ((error = decodeWithMetaOptions(request, generateCas,
                                         skipConflictResolution, keyOffset)) !=
          PROTOCOL_BINARY_RESPONSE_SUCCESS) {
          return sendResponse(response, NULL, 0, NULL, 0, NULL, 0,
                              PROTOCOL_BINARY_RAW_BYTES, error, 0, cookie);
      }
  
      if (extlen == 26 || extlen == 30) {
          uint16_t nmeta = 0;
          memcpy(&nmeta, key + keyOffset, sizeof(nmeta));
          keyOffset += 2; // 2 bytes for nmeta
          nmeta = ntohs(nmeta);
          if (nmeta > 0) {
              // Correct the vallen
              vallen -= nmeta;
              try {
                  emd.reset(new ExtendedMetaData(key + keylen + keyOffset + vallen,
                                                 nmeta));
              } catch (const std::bad_alloc&) {
                  return sendResponse(response, NULL, 0, NULL, 0, NULL, 0,
                              PROTOCOL_BINARY_RAW_BYTES,
                              PROTOCOL_BINARY_RESPONSE_ENOMEM, 0, cookie);
              }
  
              if (emd->getStatus() == ENGINE_EINVAL) {
                  return sendResponse(response, NULL, 0, NULL, 0, NULL, 0,
                              PROTOCOL_BINARY_RAW_BYTES,
                              PROTOCOL_BINARY_RESPONSE_EINVAL, 0, cookie);
              }
          }
      }
  
      if (vallen > maxItemSize) {
          LOG(EXTENSION_LOG_WARNING,
              "Item value size %ld for setWithMeta is bigger "
              "than the max size %ld allowed!!!\n", vallen, maxItemSize);
          return sendResponse(response, NULL, 0, NULL, 0, NULL, 0,
                              PROTOCOL_BINARY_RAW_BYTES,
                              PROTOCOL_BINARY_RESPONSE_E2BIG, 0, cookie);
      }
  
      uint8_t* dta = key + keyOffset + keylen;
  
      if (!isDatatypeSupported(cookie)) {
          const int len = vallen;
          const unsigned char *data = (const unsigned char*) dta;
          if (checkUTF8JSON(data, len)) {
              datatype = PROTOCOL_BINARY_DATATYPE_JSON;
          }
      }
  
      uint8_t ext_meta[1];
      uint8_t ext_len = EXT_META_LEN;
      *(ext_meta) = datatype;
      Item *itm = new Item(DocKey(key + keyOffset, keylen, docNamespace),
                           flags, expiration, dta, vallen,
                           ext_meta, ext_len, cas, -1, vbucket);
  
      if (itm == NULL) {
          return sendResponse(response, NULL, 0, NULL, 0, NULL, 0,
                              PROTOCOL_BINARY_RAW_BYTES,
                              PROTOCOL_BINARY_RESPONSE_ENOMEM, 0, cookie);
      }
  
      void *startTimeC = getEngineSpecific(cookie);
      hrtime_t startTime;
      if (startTimeC) {
          startTime = *(static_cast<hrtime_t *> (startTimeC));
      } else {
          startTime = gethrtime();
      }
  
      itm->setRevSeqno(seqno);
  
      bool allowExisting = (opcode == PROTOCOL_BINARY_CMD_SET_WITH_META ||
                            opcode == PROTOCOL_BINARY_CMD_SETQ_WITH_META);
  
      uint8_t meta[16];
      uint64_t by_seqno = 0;
      uint64_t vb_uuid = 0;
  
      ENGINE_ERROR_CODE ret = kvBucket->setWithMeta(*itm,
                                                    ntohll(request->
                                                      message.header.request.cas),
                                                    &by_seqno, cookie,
                                                    skipConflictResolution,
                                                    allowExisting,
                                                    GenerateBySeqno::Yes,
                                                    generateCas,
                                                    emd.get(), false);
  
      if (ret == ENGINE_SUCCESS) {
          ++stats.numOpsSetMeta;
          hrtime_t endTime(gethrtime());
          hrtime_t elapsed = (endTime - startTime) / 1000;
          stats.setWithMetaHisto.add(elapsed);
      } else if (ret == ENGINE_ENOMEM) {
          ret = memoryCondition();
      } else if (ret == ENGINE_EWOULDBLOCK) {
          delete itm;
          ++stats.numOpsGetMetaOnSetWithMeta;
          if (!startTimeC) {
              startTimeC = cb_malloc(sizeof(hrtime_t));
              memcpy(startTimeC, &startTime, sizeof(hrtime_t));
              storeEngineSpecific(cookie, startTimeC);
          }
          return ret;
      }
  
      auto rc = serverApi->cookie->engine_error2mcbp(cookie, ret);
  
      if (ret == ENGINE_SUCCESS) {
          cas = itm->getCas();
      } else {
          cas = 0;
      }
  
      if (startTimeC) {
          cb_free(startTimeC);
          startTimeC = NULL;
          storeEngineSpecific(cookie, startTimeC);
      }
      delete itm;
  
      if ((opcode == PROTOCOL_BINARY_CMD_SETQ_WITH_META || opcode == PROTOCOL_BINARY_CMD_ADDQ_WITH_META) &&
          rc == PROTOCOL_BINARY_RESPONSE_SUCCESS) {
          return ENGINE_SUCCESS;
      }
  
      if (ret == ENGINE_NOT_MY_VBUCKET) {
          return sendNotMyVBucketResponse(response, cookie, cas);
      }
  
      if (ret == ENGINE_SUCCESS && isMutationExtrasSupported(cookie)) {
          RCPtr<VBucket> vb = kvBucket->getVBucket(vbucket);
          vb_uuid = htonll(vb->failovers->getLatestUUID());
          by_seqno = htonll(by_seqno);
          memcpy(meta, &vb_uuid, sizeof(vb_uuid));
          memcpy(meta + sizeof(vb_uuid), &by_seqno, sizeof(by_seqno));
          return sendResponse(response, NULL, 0, (const void *)meta, sizeof(meta),
                              NULL, 0, PROTOCOL_BINARY_RAW_BYTES, rc, cas, cookie);
      }
  
      return sendResponse(response, NULL, 0, NULL, 0, NULL, 0, PROTOCOL_BINARY_RAW_BYTES,
                          rc, cas, cookie);
  }
  
  ENGINE_ERROR_CODE EventuallyPersistentEngine::deleteWithMeta(
                               const void* cookie,
                               protocol_binary_request_delete_with_meta *request,
                               ADD_RESPONSE response,
                               DocNamespace docNamespace) {
      // revid_nbytes, flags and exptime is mandatory fields.. and we need a key
      uint16_t nkey = ntohs(request->message.header.request.keylen);
      uint8_t extlen = request->message.header.request.extlen;
      // extlen, the size dicates what is encoded.
      // 24 = no nmeta and no options
      // 26 = nmeta
      // 28 = options (4-byte field)
      // 30 = options and nmeta (options followed by nmeta)
      // so 27, 25 etc... are illegal
      if ((extlen != 24 && extlen != 26 && extlen != 28  && extlen != 30)
          || nkey == 0) {
          return sendResponse(response, NULL, 0, NULL, 0, NULL, 0,
                              PROTOCOL_BINARY_RAW_BYTES,
                              PROTOCOL_BINARY_RESPONSE_EINVAL, 0, cookie);
      }
  
      if (isDegradedMode()) {
          return sendResponse(response, NULL, 0, NULL, 0, NULL, 0,
                              PROTOCOL_BINARY_RAW_BYTES,
                              PROTOCOL_BINARY_RESPONSE_ETMPFAIL,
                              0, cookie);
      }
  
      uint8_t opcode = request->message.header.request.opcode;
      uint16_t vbucket = ntohs(request->message.header.request.vbucket);
      uint64_t cas = ntohll(request->message.header.request.cas);
  
      uint32_t flags = request->message.body.flags;
      uint32_t expiration = ntohl(request->message.body.expiration);
      uint64_t seqno = ntohll(request->message.body.seqno);
      uint64_t metacas = ntohll(request->message.body.cas);
  
      bool skipConflictResolution = false;
      GenerateCas generateCas = GenerateCas::No;
      std::unique_ptr<ExtendedMetaData> emd;
      int keyOffset = 0;
      protocol_binary_response_status error = PROTOCOL_BINARY_RESPONSE_SUCCESS;
      if ((error = decodeWithMetaOptions(request, generateCas,
                                        skipConflictResolution, keyOffset)) !=
          PROTOCOL_BINARY_RESPONSE_SUCCESS) {
          return sendResponse(response, NULL, 0, NULL, 0, NULL, 0,
                              PROTOCOL_BINARY_RAW_BYTES, error, 0, cookie);
      }
  
      if (extlen == 26 || extlen == 30) {
          uint16_t nmeta = 0;
          memcpy(&nmeta, request->bytes + sizeof(request->bytes),
                 sizeof(nmeta));
          keyOffset += 2; // 2 bytes for nmeta
          nmeta = ntohs(nmeta);
          if (nmeta > 0) {
              try {
                  emd.reset(new ExtendedMetaData(request->bytes +
                                                 nkey +
                                                 keyOffset, nmeta));
              } catch (const std::bad_alloc&) {
                  return sendResponse(response, NULL, 0, NULL, 0, NULL, 0,
                              PROTOCOL_BINARY_RAW_BYTES,
                              PROTOCOL_BINARY_RESPONSE_ENOMEM, 0, cookie);
              }
  
              if (emd->getStatus() == ENGINE_EINVAL) {
                  return sendResponse(response, NULL, 0, NULL, 0, NULL, 0,
                              PROTOCOL_BINARY_RAW_BYTES,
                              PROTOCOL_BINARY_RESPONSE_EINVAL, 0, cookie);
              }
          }
      }
  
      const uint8_t *keyPtr = request->bytes + keyOffset + sizeof(request->bytes);
      DocKey key(keyPtr, nkey, docNamespace);
  
      ItemMetaData itm_meta(metacas, seqno, flags, expiration);
  
      uint8_t meta[16];
      uint64_t by_seqno = 0;
      uint64_t vb_uuid = 0;
  
      ENGINE_ERROR_CODE ret = kvBucket->deleteWithMeta(key,
                                                       cas,
                                                       &by_seqno,
                                                       vbucket,
                                                       cookie,
                                                       skipConflictResolution,
                                                       itm_meta,
                                                       false,
                                                       GenerateBySeqno::Yes,
                                                       generateCas,
                                                       0,
                                                       emd.get(),
                                                       false);
  
      if (ret == ENGINE_SUCCESS) {
          stats.numOpsDelMeta++;
      } else if (ret == ENGINE_ENOMEM) {
          ret = memoryCondition();
      } else if (ret == ENGINE_EWOULDBLOCK) {
          return ENGINE_EWOULDBLOCK;
      }
  
      auto rc = serverApi->cookie->engine_error2mcbp(cookie, ret);
  
      if (opcode == PROTOCOL_BINARY_CMD_DELQ_WITH_META &&
          rc == PROTOCOL_BINARY_RESPONSE_SUCCESS) {
          return ENGINE_SUCCESS;
      }
  
      if (ret == ENGINE_NOT_MY_VBUCKET) {
          return sendNotMyVBucketResponse(response, cookie, cas);
      }
  
      if (ret == ENGINE_SUCCESS && isMutationExtrasSupported(cookie)) {
          RCPtr<VBucket> vb = kvBucket->getVBucket(vbucket);
          vb_uuid = htonll(vb->failovers->getLatestUUID());
          by_seqno = htonll(by_seqno);
          memcpy(meta, &vb_uuid, 8);
          memcpy(meta + 8, &by_seqno, 8);
          return sendResponse(response, NULL, 0, (const void *)meta, sizeof(meta),
                              NULL, 0, PROTOCOL_BINARY_RAW_BYTES, rc, cas, cookie);
      }
  
      return sendResponse(response, NULL, 0, NULL, 0, NULL, 0,
                          PROTOCOL_BINARY_RAW_BYTES, rc, cas, cookie);
  }
  
  ENGINE_ERROR_CODE
  EventuallyPersistentEngine::changeTapVBFilter(const void *cookie,
                                         protocol_binary_request_header *request,
                                                ADD_RESPONSE response) {
      protocol_binary_request_no_extras *req =
                                     (protocol_binary_request_no_extras*)request;
  
      uint64_t cas = ntohll(req->message.header.request.cas);
      uint16_t keylen = ntohs(req->message.header.request.keylen);
      const char *ptr = ((char*)req) + sizeof(req->message.header);
      std::string tap_name = "eq_tapq:";
      tap_name.append(std::string(ptr, keylen));
  
      const char *msg = NULL;
      protocol_binary_response_status rv = PROTOCOL_BINARY_RESPONSE_SUCCESS;
      size_t nuserdata = ntohl(req->message.header.request.bodylen) - keylen;
      uint16_t nvbuckets = 0;
  
      if (nuserdata < sizeof(nvbuckets)) {
          msg = "Number of vbuckets is missing";
          rv = PROTOCOL_BINARY_RESPONSE_EINVAL;
      } else {
          ptr += keylen;
          memcpy(&nvbuckets, ptr, sizeof(nvbuckets));
          nuserdata -= sizeof(nvbuckets);
          ptr += sizeof(nvbuckets);
          nvbuckets = ntohs(nvbuckets);
          if (nvbuckets > 0) {
              if (nuserdata <
                  ((sizeof(uint16_t) + sizeof(uint64_t)) * nvbuckets)) {
                  msg =
                     "Number of (vbucket id, checkpoint id) pair is not matched";
                  rv = PROTOCOL_BINARY_RESPONSE_EINVAL;
              } else {
                  std::vector<uint16_t> vbuckets;
                  std::map<uint16_t, uint64_t> checkpointIds;
                  for (uint16_t i = 0; i < nvbuckets; ++i) {
                      uint16_t vbid;
                      uint64_t chkid;
                      memcpy(&vbid, ptr, sizeof(vbid));
                      ptr += sizeof(uint16_t);
                      memcpy(&chkid, ptr, sizeof(chkid));
                      ptr += sizeof(uint64_t);
                      vbuckets.push_back(ntohs(vbid));
                      checkpointIds[ntohs(vbid)] = ntohll(chkid);
                  }
                  if (!tapConnMap->changeVBucketFilter(tap_name, vbuckets,
                                                       checkpointIds)) {
                      msg = "TAP producer not exist!!!";
                      rv = PROTOCOL_BINARY_RESPONSE_EINVAL;
                  }
              }
          } else {
              msg = "Number of vbuckets should be greater than 0";
              rv = PROTOCOL_BINARY_RESPONSE_EINVAL;
          }
      }
  
      return sendResponse(response, NULL, 0, NULL, 0,
                          msg, msg ? static_cast<uint16_t>(strlen(msg)) : 0,
                          PROTOCOL_BINARY_RAW_BYTES,
                          static_cast<uint16_t>(rv),
                          cas, cookie);
  }
  
  ENGINE_ERROR_CODE
  EventuallyPersistentEngine::handleTrafficControlCmd(const void *cookie,
                                         protocol_binary_request_header *request,
                                         ADD_RESPONSE response)
  {
      std::stringstream msg;
      int16_t status = PROTOCOL_BINARY_RESPONSE_SUCCESS;
  
      switch (request->request.opcode) {
      case PROTOCOL_BINARY_CMD_ENABLE_TRAFFIC:
          if (kvBucket->isWarmingUp()) {
              // engine is still warming up, do not turn on data traffic yet
              msg << "Persistent engine is still warming up!";
              status = PROTOCOL_BINARY_RESPONSE_ETMPFAIL;
          } else if (configuration.isFailpartialwarmup() &&
                     kvBucket->isWarmupOOMFailure()) {
              // engine has completed warm up, but data traffic cannot be
              // turned on due to an OOM failure
              msg << "Data traffic to persistent engine cannot be enabled"
                     " due to out of memory failures during warmup";
              status = PROTOCOL_BINARY_RESPONSE_ENOMEM;
          } else {
              if (enableTraffic(true)) {
                  msg << "Data traffic to persistence engine is enabled";
              } else {
                  msg <<
                        "Data traffic to persistence engine was already enabled";
              }
          }
          break;
      case PROTOCOL_BINARY_CMD_DISABLE_TRAFFIC:
          if (enableTraffic(false)) {
              msg << "Data traffic to persistence engine is disabled";
          } else {
              msg << "Data traffic to persistence engine was already disabled";
          }
          break;
      default:
          msg << "Unknown traffic control opcode: " << request->request.opcode;
          status = PROTOCOL_BINARY_RESPONSE_UNKNOWN_COMMAND;
      }
  
      return sendResponse(response, NULL, 0, NULL, 0,
                          msg.str().c_str(), msg.str().length(),
                          PROTOCOL_BINARY_RAW_BYTES,
                          status, 0, cookie);
  }
  
  ENGINE_ERROR_CODE
  EventuallyPersistentEngine::doDcpVbTakeoverStats(const void *cookie,
                                                   ADD_STAT add_stat,
                                                   std::string &key,
                                                   uint16_t vbid) {
      RCPtr<VBucket> vb = getVBucket(vbid);
      if (!vb) {
          return ENGINE_NOT_MY_VBUCKET;
      }
  
      std::string dcpName("eq_dcpq:");
      dcpName.append(key);
  
      const connection_t &conn = dcpConnMap_->findByName(dcpName);
      if (!conn) {
          /*
           * @TODO: make the following log message INFO level.  However whilst the
           * root cause of MB-22082 is unresolved it is useful to log the event at
           * NOTICE level so its appears in the logs.
           */
          LOG(EXTENSION_LOG_NOTICE,"doDcpVbTakeoverStats - cannot find "
              "connection %s for vb:%" PRIu16, dcpName.c_str(), vbid);
          size_t vb_items = vb->getNumItems(kvBucket->getItemEvictionPolicy());
  
          size_t del_items = 0;
          try {
              del_items = kvBucket->getRWUnderlying(vbid)->
                                                  getNumPersistedDeletes(vbid);
          } catch (std::runtime_error& e) {
              LOG(EXTENSION_LOG_WARNING,
                  "doDcpVbTakeoverStats: exception while getting num "
                  "persisted deletes for vbucket:%" PRIu16 " - treating as 0 "
                  "deletes. Details: %s", vbid, e.what());
          }
          size_t chk_items = vb_items > 0 ?
                             vb->checkpointManager.getNumOpenChkItems() : 0;
          add_casted_stat("status", "does_not_exist", add_stat, cookie);
          add_casted_stat("on_disk_deletes", del_items, add_stat, cookie);
          add_casted_stat("vb_items", vb_items, add_stat, cookie);
          add_casted_stat("chk_items", chk_items, add_stat, cookie);
          add_casted_stat("estimate", vb_items + del_items, add_stat, cookie);
          return ENGINE_SUCCESS;
      }
  
      DcpProducer* producer = dynamic_cast<DcpProducer*>(conn.get());
      if (!producer) {
          return ENGINE_KEY_ENOENT;
      }
      producer->addTakeoverStats(add_stat, cookie, vbid);
  
      return ENGINE_SUCCESS;
  }
  
  ENGINE_ERROR_CODE
  EventuallyPersistentEngine::doTapVbTakeoverStats(const void *cookie,
                                                   ADD_STAT add_stat,
                                                   std::string &key,
                                                   uint16_t vbid) {
      RCPtr<VBucket> vb = getVBucket(vbid);
      if (!vb) {
          return ENGINE_NOT_MY_VBUCKET;
      }
      std::string tapName("eq_tapq:");
      tapName.append(key);
      size_t vb_items = vb->getNumItems(kvBucket->getItemEvictionPolicy());
  
      size_t del_items = 0;
      try {
          del_items = kvBucket->getRWUnderlying(vbid)->
                                                  getNumPersistedDeletes(vbid);
      } catch (std::runtime_error& e) {
          LOG(EXTENSION_LOG_WARNING,
              "doTapVbTakeoverStats: exception while getting num persisted "
              "deletes for vbucket:%" PRIu16 " - treating as 0 deletes. "
              "Details: %s", vbid, e.what());
      }
  
      add_casted_stat("name", tapName, add_stat, cookie);
  
      uint64_t total;
      uint64_t chk_items;
      if (key.length() == 0 || !tapConnMap->findByName(tapName)) {
          chk_items = vb_items > 0 ?
                      vb->checkpointManager.getNumOpenChkItems() : 0;
          total = vb_items + del_items + chk_items;
          add_casted_stat("status", "does_not_exist", add_stat, cookie);
      } else {
          if (tapConnMap->isBackfillCompleted(tapName)) {
              chk_items = vb_items > 0 ?
                  vb->checkpointManager.getNumItemsForCursor(tapName) : 0;
              total = chk_items;
              add_casted_stat("status", "backfill completed", add_stat, cookie);
          } else {
              chk_items = vb_items > 0 ?
                          vb->checkpointManager.getNumOpenChkItems() : 0;
              total = vb_items + del_items + chk_items;
              add_casted_stat("status", "backfilling", add_stat, cookie);
          }
      }
  
      add_casted_stat("estimate", total, add_stat, cookie);
      add_casted_stat("on_disk_deletes", del_items, add_stat, cookie);
      add_casted_stat("chk_items", chk_items, add_stat, cookie);
      add_casted_stat("vb_items", vb_items, add_stat, cookie);
  
      return ENGINE_SUCCESS;
  }
  
  ENGINE_ERROR_CODE
  EventuallyPersistentEngine::returnMeta(const void* cookie,
                                    protocol_binary_request_return_meta *request,
                                    ADD_RESPONSE response,
                                    DocNamespace docNamespace) {
      uint8_t extlen = request->message.header.request.extlen;
      uint16_t keylen = ntohs(request->message.header.request.keylen);
      if (extlen != 12 || request->message.header.request.keylen == 0) {
          return sendResponse(response, NULL, 0, NULL, 0, NULL, 0,
                              PROTOCOL_BINARY_RAW_BYTES,
                              PROTOCOL_BINARY_RESPONSE_EINVAL, 0, cookie);
      }
  
      if (isDegradedMode()) {
          return sendResponse(response, NULL, 0, NULL, 0, NULL, 0,
                              PROTOCOL_BINARY_RAW_BYTES,
                              PROTOCOL_BINARY_RESPONSE_ETMPFAIL,
                              0, cookie);
      }
  
      uint8_t* keyPtr = request->bytes + sizeof(request->bytes);
      uint16_t vbucket = ntohs(request->message.header.request.vbucket);
      uint32_t bodylen = ntohl(request->message.header.request.bodylen);
      uint64_t cas = ntohll(request->message.header.request.cas);
      uint8_t datatype = request->message.header.request.datatype;
      uint32_t mutate_type = ntohl(request->message.body.mutation_type);
      uint32_t flags = ntohl(request->message.body.flags);
      uint32_t exp = ntohl(request->message.body.expiration);
      exp = exp == 0 ? 0 : ep_abs_time(ep_reltime(exp));
      size_t vallen = bodylen - keylen - extlen;
      uint64_t seqno;
  
      ENGINE_ERROR_CODE ret = ENGINE_EINVAL;
      if (mutate_type == SET_RET_META || mutate_type == ADD_RET_META) {
          uint8_t *dta = keyPtr + keylen;
  
          if (!isDatatypeSupported(cookie)) {
              const int len = vallen;
              const unsigned char *data = (const unsigned char*) dta;
              if (checkUTF8JSON(data, len)) {
                  datatype = PROTOCOL_BINARY_DATATYPE_JSON;
              }
          }
  
          uint8_t ext_meta[1];
          uint8_t ext_len = EXT_META_LEN;
          *(ext_meta) = datatype;
          Item *itm = new Item(DocKey(keyPtr, keylen, docNamespace),
                               flags, exp, dta, vallen, ext_meta,
                               ext_len, cas, -1, vbucket);
  
          if (!itm) {
              return sendResponse(response, NULL, 0, NULL, 0, NULL, 0,
                                  PROTOCOL_BINARY_RAW_BYTES,
                                  PROTOCOL_BINARY_RESPONSE_ENOMEM, 0, cookie);
          }
  
          if (mutate_type == SET_RET_META) {
              ret = kvBucket->set(*itm, cookie);
          } else {
              ret = kvBucket->add(*itm, cookie);
          }
          if (ret == ENGINE_SUCCESS) {
              ++stats.numOpsSetRetMeta;
          }
          cas = itm->getCas();
          seqno = htonll(itm->getRevSeqno());
          delete itm;
      } else if (mutate_type == DEL_RET_META) {
          ItemMetaData itm_meta;
          DocKey key(keyPtr, keylen, docNamespace);
          ret = kvBucket->deleteItem(
                  key, cas, vbucket, cookie, false, nullptr, &itm_meta, nullptr);
          if (ret == ENGINE_SUCCESS) {
              ++stats.numOpsDelRetMeta;
          }
          flags = itm_meta.flags;
          exp = itm_meta.exptime;
          cas = itm_meta.cas;
          seqno = htonll(itm_meta.revSeqno);
      } else {
          return sendResponse(response, NULL, 0, NULL, 0, NULL, 0,
                              PROTOCOL_BINARY_RAW_BYTES,
                              PROTOCOL_BINARY_RESPONSE_EINVAL, 0, cookie);
      }
  
      if (ret == ENGINE_NOT_MY_VBUCKET) {
          return sendNotMyVBucketResponse(response, cookie, cas);
      } else if (ret == ENGINE_EWOULDBLOCK) {
          return ret;
      } else if (ret != ENGINE_SUCCESS) {
          auto rc = serverApi->cookie->engine_error2mcbp(cookie, ret);
          return sendResponse(response, NULL, 0, NULL, 0, NULL, 0,
                              PROTOCOL_BINARY_RAW_BYTES, rc, 0, cookie);
      }
  
      uint8_t meta[16];
      exp = htonl(exp);
      memcpy(meta, &flags, 4);
      memcpy(meta + 4, &exp, 4);
      memcpy(meta + 8, &seqno, 8);
  
      return sendResponse(response, NULL, 0, (const void *)meta, 16, NULL, 0,
                          datatype, PROTOCOL_BINARY_RESPONSE_SUCCESS, cas,
                          cookie);
  }
  
  ENGINE_ERROR_CODE
  EventuallyPersistentEngine::setClusterConfig(const void* cookie,
                             protocol_binary_request_set_cluster_config *request,
                             ADD_RESPONSE response) {
  
      uint64_t cas = ntohll(request->message.header.request.cas);
      uint32_t bodylen = ntohl(request->message.header.request.bodylen);
      {
          try {
              LockHolder lh(clusterConfig.lock);
              const uint8_t* body = request->bytes + sizeof(request->message.header);
              clusterConfig.config.assign(body, body + bodylen);
          } catch (std::bad_alloc& e) {
              LOG(EXTENSION_LOG_WARNING,
                  "Failed to update cluster configuration - %s", e.what());
              return sendResponse(response, nullptr, 0, nullptr, 0, nullptr, 0,
                                  PROTOCOL_BINARY_RAW_BYTES,
                                  PROTOCOL_BINARY_RESPONSE_ENOMEM, cas, cookie);
          }
      }
  
      // clusterConfig is opaque to ep-engine, but it /should/ be in JSON, as
      // that's what the clients expect. Attempt to parse it and log the revision.
      bool found_rev = false;
      unique_cJSON_ptr json(cJSON_Parse(clusterConfig.config.c_str()));
      if (json) {
          cJSON* rev = cJSON_GetObjectItem(json.get(), "rev");
          if (rev) {
              found_rev = true;
              std::string rev_string = to_string(rev, false);
              LOG(EXTENSION_LOG_NOTICE,
                  "Updated cluster configuration. New revision: %s",
                  rev_string.c_str());
          }
      }
  
      if (!found_rev) {
          // Failed to parse. Hail Mary time, let's just print the first 100
          // bytes which hopefully includes helpful identifying information.
          const int CONFIG_LIMIT = 100;
          if (clusterConfig.config.size() > CONFIG_LIMIT) {
              LOG(EXTENSION_LOG_WARNING, "Updated cluster configuration. "
                  "Failed to parse JSON config - first %d bytes: '%.*s'...\n",
                  CONFIG_LIMIT, CONFIG_LIMIT, clusterConfig.config.data());
          }
      }
  
      return sendResponse(response, NULL, 0, NULL, 0, NULL, 0,
                          PROTOCOL_BINARY_RAW_BYTES,
                          PROTOCOL_BINARY_RESPONSE_SUCCESS, cas, cookie);
  }
  
  ENGINE_ERROR_CODE
  EventuallyPersistentEngine::getClusterConfig(const void* cookie,
                              protocol_binary_request_get_cluster_config*,
                              ADD_RESPONSE response) {
      LockHolder lh(clusterConfig.lock);
      return sendResponse(response, NULL, 0, NULL, 0, clusterConfig.config.data(),
                          clusterConfig.config.size(), PROTOCOL_BINARY_RAW_BYTES,
                          PROTOCOL_BINARY_RESPONSE_SUCCESS, 0, cookie);
  }
  
  /**
   * Callback class used by AllKeysAPI, for caching fetched keys
   *
   * As by default (or in most cases), number of keys is 1000,
   * and an average key could be 32B in length, initialize buffersize of
   * allKeys to 34000 (1000 * 32 + 1000 * 2), the additional 2 bytes per
   * key is for the keylength.
   *
   * This initially allocated buffersize is doubled whenever the length
   * of the buffer holding all the keys, crosses the buffersize.
   */
  class AllKeysCallback : public Callback<const DocKey&> {
  public:
      AllKeysCallback() {
          buffer.reserve((avgKeySize + sizeof(uint16_t)) * expNumKeys);
      }
  
      void callback(const DocKey& key) {
          if (buffer.size() + key.size() + sizeof(uint16_t) >
              buffer.size()) {
              // Reserve the 2x space for the copy-to buffer.
              buffer.reserve(buffer.size()*2);
          }
          uint16_t outlen = htons(key.size());
          // insert 1 x u16
          const auto* outlenPtr = reinterpret_cast<const char*>(&outlen);
          buffer.insert(buffer.end(), outlenPtr, outlenPtr + sizeof(uint16_t));
          // insert the char buffer
          buffer.insert(buffer.end(), key.data(), key.data()+key.size());
      }
  
      char* getAllKeysPtr() { return buffer.data(); }
      uint64_t getAllKeysLen() { return buffer.size(); }
  
  private:
      std::vector<char> buffer;
  
      static const int avgKeySize = 32;
      static const int expNumKeys = 1000;
  
  };
  
  /*
   * Task that fetches all_docs and returns response,
   * runs in background.
   */
  class FetchAllKeysTask : public GlobalTask {
  public:
      FetchAllKeysTask(EventuallyPersistentEngine *e, const void *c,
                       ADD_RESPONSE resp, const DocKey start_key_,
                       uint16_t vbucket, uint32_t count_) :
          GlobalTask(e, TaskId::FetchAllKeysTask, 0, false), engine(e), cookie(c),
          response(resp), start_key(start_key_), vbid(vbucket),
          count(count_) { }
  
      std::string getDescription() {
          return std::string("Running the ALL_DOCS api on vbucket: %d", vbid);
      }
  
      bool run() {
          TRACE_EVENT0("ep-engine/task", "FetchAllKeysTask");
          ENGINE_ERROR_CODE err;
          if (engine->getKVBucket()->getVBuckets().
                  getBucket(vbid)->isBucketCreation()) {
              // Returning an empty packet with a SUCCESS response as
              // there aren't any keys during the vbucket file creation.
              err = sendResponse(response, NULL, 0, NULL, 0, NULL, 0,
                                 PROTOCOL_BINARY_RAW_BYTES,
                                 PROTOCOL_BINARY_RESPONSE_SUCCESS, 0,
                                 cookie);
          } else {
              auto cb = std::make_shared<AllKeysCallback>();
              err = engine->getKVBucket()->getROUnderlying(vbid)->getAllKeys(
                                                      vbid, start_key, count, cb);
              if (err == ENGINE_SUCCESS) {
                  err =  sendResponse(response, NULL, 0, NULL, 0,
                                      ((AllKeysCallback*)cb.get())->getAllKeysPtr(),
                                      ((AllKeysCallback*)cb.get())->getAllKeysLen(),
                                      PROTOCOL_BINARY_RAW_BYTES,
                                      PROTOCOL_BINARY_RESPONSE_SUCCESS, 0,
                                      cookie);
              }
          }
          engine->addLookupAllKeys(cookie, err);
          engine->notifyIOComplete(cookie, err);
          return false;
      }
  
  private:
      EventuallyPersistentEngine *engine;
      const void *cookie;
      ADD_RESPONSE response;
      StoredDocKey start_key;
      uint16_t vbid;
      uint32_t count;
  };
  
  ENGINE_ERROR_CODE
  EventuallyPersistentEngine::getAllKeys(const void* cookie,
                                  protocol_binary_request_get_keys *request,
                                  ADD_RESPONSE response,
                                  DocNamespace docNamespace) {
      {
          LockHolder lh(lookupMutex);
          auto it = allKeysLookups.find(cookie);
          if (it != allKeysLookups.end()) {
              ENGINE_ERROR_CODE err = it->second;
              allKeysLookups.erase(it);
              return err;
          }
      }
  
      uint16_t vbucket = ntohs(request->message.header.request.vbucket);
      RCPtr<VBucket> vb = getVBucket(vbucket);
  
      if (!vb) {
          return ENGINE_NOT_MY_VBUCKET;
      }
  
      ReaderLockHolder rlh(vb->getStateLock());
      if (vb->getState() != vbucket_state_active) {
          return ENGINE_NOT_MY_VBUCKET;
      }
      //key: key, ext: no. of keys to fetch, sorting-order
      uint16_t keylen = ntohs(request->message.header.request.keylen);
      uint8_t extlen = request->message.header.request.extlen;
  
      uint32_t count = 1000;
  
      if (extlen > 0) {
          if (extlen != sizeof(uint32_t)) {
              return ENGINE_EINVAL;
          }
          memcpy(&count, request->bytes + sizeof(request->bytes),
                 sizeof(uint32_t));
          count = ntohl(count);
      }
  
      if (keylen == 0) {
          LOG(EXTENSION_LOG_WARNING, "No key passed as argument for getAllKeys");
          return ENGINE_EINVAL;
      }
      const uint8_t* keyPtr = (request->bytes + sizeof(request->bytes) + extlen);
      DocKey start_key(keyPtr, keylen, docNamespace);
  
      ExTask task = new FetchAllKeysTask(this, cookie, response, start_key,
                                         vbucket, count);
      ExecutorPool::get()->schedule(task, READER_TASK_IDX);
      return ENGINE_EWOULDBLOCK;
  }
  
  ENGINE_ERROR_CODE EventuallyPersistentEngine::getRandomKey(const void *cookie,
                                                         ADD_RESPONSE response) {
      GetValue gv(kvBucket->getRandomKey());
      ENGINE_ERROR_CODE ret = gv.getStatus();
  
      if (ret == ENGINE_SUCCESS) {
          Item *it = gv.getValue();
          uint32_t flags = it->getFlags();
          ret = sendResponse(response, static_cast<const void *>(it->getKey().data()),
                             it->getKey().size(),
                             (const void *)&flags, sizeof(uint32_t),
                             static_cast<const void *>(it->getData()),
                             it->getNBytes(), it->getDataType(),
                             PROTOCOL_BINARY_RESPONSE_SUCCESS, it->getCas(),
                             cookie);
          delete it;
      }
  
      return ret;
  }
  
  ENGINE_ERROR_CODE EventuallyPersistentEngine::dcpOpen(const void* cookie,
                                                         uint32_t opaque,
                                                         uint32_t seqno,
                                                         uint32_t flags,
                                                         const void *stream_name,
                                                         uint16_t nname)
  {
      (void) opaque;
      (void) seqno;
      std::string connName(static_cast<const char*>(stream_name), nname);
  
      if (reserveCookie(cookie) != ENGINE_SUCCESS) {
          LOG(EXTENSION_LOG_WARNING, "Cannot create DCP connection because cookie"
              "cannot be reserved");
          return ENGINE_DISCONNECT;
      }
  
      if (getEngineSpecific(cookie) != NULL) {
          LOG(EXTENSION_LOG_WARNING, "Cannot open DCP connection as another"
              " connection exists on the same socket");
          return ENGINE_DISCONNECT;
      }
  
      ConnHandler *handler = NULL;
      if (flags & DCP_OPEN_PRODUCER) {
          handler = dcpConnMap_->newProducer(cookie, connName, false);
      } else if (flags & DCP_OPEN_NOTIFIER) {
          handler = dcpConnMap_->newProducer(cookie, connName, true);
      } else {
          handler = dcpConnMap_->newConsumer(cookie, connName);
      }
  
      if (handler == nullptr) {
          LOG(EXTENSION_LOG_WARNING, "EPEngine::dcpOpen: failed to create a handler");
          releaseCookie(cookie);
          return ENGINE_DISCONNECT;
      } else {
          storeEngineSpecific(cookie, handler);
      }
      return ENGINE_SUCCESS;
  }
  
  ENGINE_ERROR_CODE EventuallyPersistentEngine::dcpAddStream(const void* cookie,
                                                             uint32_t opaque,
                                                             uint16_t vbucket,
                                                             uint32_t flags)
  {
      ENGINE_ERROR_CODE errCode = ENGINE_DISCONNECT;
      ConnHandler* conn = getConnHandler(cookie);
      if (conn) {
          errCode = dcpConnMap_->addPassiveStream(*conn, opaque, vbucket, flags);
      }
      return errCode;
  }
  
  ConnHandler* EventuallyPersistentEngine::getConnHandler(const void *cookie) {
      void* specific = getEngineSpecific(cookie);
      ConnHandler* handler = reinterpret_cast<ConnHandler*>(specific);
      if (!handler) {
          LOG(EXTENSION_LOG_WARNING, "Invalid streaming connection");
      }
      return handler;
  }
  
  void EventuallyPersistentEngine::handleDisconnect(const void *cookie) {
      tapConnMap->disconnect(cookie);
      dcpConnMap_->disconnect(cookie);
      /**
       * Decrement session_cas's counter, if the connection closes
       * before a control command (that returned ENGINE_EWOULDBLOCK
       * the first time) makes another attempt.
       *
       * Commands to be considered: DEL_VBUCKET, COMPACT_DB
       */
      if (getEngineSpecific(cookie) != NULL) {
          uint8_t opcode = getOpcodeIfEwouldblockSet(cookie);
          switch(opcode) {
              case PROTOCOL_BINARY_CMD_DEL_VBUCKET:
              case PROTOCOL_BINARY_CMD_COMPACT_DB:
                  {
                      decrementSessionCtr();
                      storeEngineSpecific(cookie, NULL);
                      break;
                  }
              default:
                  break;
          }
      }
  }
  
  void EventuallyPersistentEngine::handleDeleteBucket(const void *cookie) {
      LOG(EXTENSION_LOG_NOTICE, "Shutting down all TAP & DCP connections in "
              "preparation for bucket deletion.");
      tapConnMap->shutdownAllConnections();
      dcpConnMap_->shutdownAllConnections();
  }
  
  ENGINE_ERROR_CODE EventuallyPersistentEngine::getAllVBucketSequenceNumbers(
                                      const void *cookie,
                                      protocol_binary_request_header *request,
                                      ADD_RESPONSE response) {
      protocol_binary_request_get_all_vb_seqnos *req =
          reinterpret_cast<protocol_binary_request_get_all_vb_seqnos*>(request);
  
      // if extlen (hence bodylen) is non-zero, it limits the result to only
      // include the vbuckets in the specified vbucket state.
      size_t bodylen = ntohl(req->message.header.request.bodylen);
  
      vbucket_state_t reqState = static_cast<vbucket_state_t>(0);;
      if (bodylen != 0) {
          memcpy(&reqState, &req->message.body.state, sizeof(reqState));
          reqState = static_cast<vbucket_state_t>(ntohl(reqState));
      }
  
      std::vector<uint8_t> payload;
      auto vbuckets = kvBucket->getVBuckets().getBuckets();
  
      /* Reserve a buffer that's big enough to hold all of them (we might
       * not use all of them. Each entry in the array occupies 10 bytes
       * (two bytes vbucket id followed by 8 bytes sequence number)
       */
      try {
          payload.reserve(vbuckets.size() * (sizeof(uint16_t) + sizeof(uint64_t)));
      } catch (std::bad_alloc) {
          return sendResponse(response, 0, 0, 0, 0, 0, 0,
                              PROTOCOL_BINARY_RAW_BYTES,
                              PROTOCOL_BINARY_RESPONSE_ENOMEM, 0,
                              cookie);
      }
  
      for (auto id : vbuckets) {
          RCPtr<VBucket> vb = getVBucket(id);
          if (vb) {
              auto state = vb->getState();
              bool getSeqnoForThisVb = false;
              if (reqState) {
                  getSeqnoForThisVb = (reqState == state);
              } else {
                  getSeqnoForThisVb = (state == vbucket_state_active) ||
                                      (state == vbucket_state_replica) ||
                                      (state == vbucket_state_pending);
              }
              if (getSeqnoForThisVb) {
                  uint16_t vbid = htons(static_cast<uint16_t>(id));
                  uint64_t highSeqno;
                  if (vb->getState() == vbucket_state_active) {
                      highSeqno = htonll(vb->getHighSeqno());
                  } else {
                      snapshot_info_t info =
                                          vb->checkpointManager.getSnapshotInfo();
                      highSeqno = htonll(info.range.end);
                  }
                  auto offset = payload.size();
                  payload.resize(offset + sizeof(vbid) + sizeof(highSeqno));
                  memcpy(payload.data() + offset, &vbid, sizeof(vbid));
                  memcpy(payload.data() + offset + sizeof(vbid), &highSeqno,
                         sizeof(highSeqno));
              }
          }
      }
  
      return sendResponse(response,
                          0, 0, /* key */
                          0, 0, /* ext field */
                          payload.data(), payload.size(), /* value */
                          PROTOCOL_BINARY_RAW_BYTES,
                          PROTOCOL_BINARY_RESPONSE_SUCCESS, 0,
                          cookie);
  }
  
  void EventuallyPersistentEngine::updateDcpMinCompressionRatio(float value) {
      if (dcpConnMap_) {
          dcpConnMap_->updateMinCompressionRatioForProducers(value);
      }
  }
  
  ENGINE_ERROR_CODE EventuallyPersistentEngine::sendNotMyVBucketResponse(
          ADD_RESPONSE response, const void* cookie, uint64_t cas) {
      LockHolder lh(clusterConfig.lock);
      return sendResponse(response, nullptr, 0, nullptr, 0,
                          clusterConfig.config.data(),
                          clusterConfig.config.size(), PROTOCOL_BINARY_RAW_BYTES,
                          PROTOCOL_BINARY_RESPONSE_NOT_MY_VBUCKET, cas, cookie);
  }
  
  EventuallyPersistentEngine::~EventuallyPersistentEngine() {
      kvBucket->deinitialize();
      LOG(EXTENSION_LOG_NOTICE, "~EPEngine: Completed deinitialize.");
      delete kvBucket;
      LOG(EXTENSION_LOG_NOTICE, "~EPEngine: Deleted KvBucket.");
      delete workload;
      delete dcpConnMap_;
      LOG(EXTENSION_LOG_NOTICE, "~EPEngine: Deleted dcpConnMap_.");
      delete dcpFlowControlManager_;
      delete tapConnMap;
      LOG(EXTENSION_LOG_NOTICE, "~EPEngine: Deleted tapConnMap_.");
      delete tapConfig;
      delete checkpointConfig;
      delete replicationThrottle;
  }
  
  const std::string& EpEngineTaskable::getName() const {
      return myEngine->getName();
  }
  
  task_gid_t EpEngineTaskable::getGID() const {
      return reinterpret_cast<task_gid_t>(myEngine);
  }
  
  bucket_priority_t EpEngineTaskable::getWorkloadPriority() const {
      return myEngine->getWorkloadPriority();
  }
  
  void  EpEngineTaskable::setWorkloadPriority(bucket_priority_t prio) {
      myEngine->setWorkloadPriority(prio);
  }
  
  WorkLoadPolicy&  EpEngineTaskable::getWorkLoadPolicy(void) {
      return myEngine->getWorkLoadPolicy();
  }
  
  void EpEngineTaskable::logQTime(TaskId id,
                                  const ProcessClock::duration enqTime) {
      myEngine->getKVBucket()->logQTime(id, enqTime);
  }
  
  void EpEngineTaskable::logRunTime(TaskId id,
                                    const ProcessClock::duration runTime) {
      myEngine->getKVBucket()->logRunTime(id, runTime);
  }
  