text: /**
   *    Copyright (C) 2013 10gen Inc.
   *
   *    This program is free software: you can redistribute it and/or  modify
   *    it under the terms of the GNU Affero General Public License, version 3,
   *    as published by the Free Software Foundation.
   *
   *    This program is distributed in the hope that it will be useful,
   *    but WITHOUT ANY WARRANTY; without even the implied warranty of
   *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   *    GNU Affero General Public License for more details.
   *
   *    You should have received a copy of the GNU Affero General Public License
   *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
   *
   *    As a special exception, the copyright holders give permission to link the
   *    code of portions of this program with the OpenSSL library under certain
   *    conditions as described in each individual source file and distribute
   *    linked combinations including the program with the OpenSSL library. You
   *    must comply with the GNU Affero General Public License in all respects for
   *    all of the code used other than as permitted herein. If you modify file(s)
   *    with this exception, you may extend this exception to your version of the
   *    file(s), but you are not obligated to do so. If you do not wish to do so,
   *    delete this exception statement from your version. If you delete this
   *    exception statement from all source files in the program, then also delete
   *    it in the license file.
   */
  
  #define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kSharding
  
  #include "mongo/platform/basic.h"
  
  #include "mongo/db/range_deleter.h"
  
  #include <boost/date_time/posix_time/posix_time_duration.hpp>
  #include <memory>
  
  #include "mongo/db/client.h"
  #include "mongo/db/repl/replication_coordinator_global.h"
  #include "mongo/db/service_context.h"
  #include "mongo/db/write_concern_options.h"
  #include "mongo/util/concurrency/idle_thread_block.h"
  #include "mongo/util/exit.h"
  #include "mongo/util/log.h"
  #include "mongo/util/mongoutils/str.h"
  #include "mongo/util/time_support.h"
  #include "mongo/util/timer.h"
  
  using std::unique_ptr;
  using std::endl;
  using std::set;
  using std::pair;
  using std::string;
  
  namespace {
  const long int kNotEmptyTimeoutMillis = 200;
  const long long int kMaxCursorCheckIntervalMillis = 500;
  const size_t kDeleteJobsHistory = 10;  // entries
  
  /**
   * Removes an element from the container that holds a pointer type, and deletes the
   * pointer as well. Returns true if the element was found.
   */
  template <typename ContainerType, typename ContainerElementType>
  bool deletePtrElement(ContainerType* container, ContainerElementType elem) {
      typename ContainerType::iterator iter = container->find(elem);
  
      if (iter == container->end()) {
          return false;
      }
  
      delete *iter;
      container->erase(iter);
      return true;
  }
  }
  
  namespace mongo {
  
  namespace duration = boost::posix_time;
  
  static void logCursorsWaiting(RangeDeleteEntry* entry) {
      // We always log the first cursors waiting message (so we have cursor ids in the logs).
      // After 15 minutes (the cursor timeout period), we start logging additional messages at
      // a 1 minute interval.
      static const auto kLogCursorsThreshold = Minutes{15};
      static const auto kLogCursorsInterval = Minutes{1};
  
      Date_t currentTime = jsTime();
      Milliseconds elapsedMillisSinceQueued{0};
  
      // We always log the first message when lastLoggedTime == 0
      if (entry->lastLoggedTS != Date_t()) {
          if (currentTime > entry->stats.queueStartTS)
              elapsedMillisSinceQueued = currentTime - entry->stats.queueStartTS;
  
          // Not logging, threshold not passed
          if (elapsedMillisSinceQueued < kLogCursorsThreshold)
              return;
  
          Milliseconds elapsedMillisSinceLog{0};
          if (currentTime > entry->lastLoggedTS)
              elapsedMillisSinceLog = currentTime - entry->lastLoggedTS;
  
          // Not logging, logged a short time ago
          if (elapsedMillisSinceLog < kLogCursorsInterval)
              return;
      }
  
      str::stream cursorList;
      for (std::set<CursorId>::const_iterator it = entry->cursorsToWait.begin();
           it != entry->cursorsToWait.end();
           ++it) {
          if (it != entry->cursorsToWait.begin())
              cursorList << ", ";
          cursorList << *it;
      }
  
      log() << "waiting for open cursors before removing range "
            << "[" << redact(entry->options.range.minKey) << ", "
            << redact(entry->options.range.maxKey) << ") "
            << "in " << entry->options.range.ns
            << (entry->lastLoggedTS == Date_t()
                    ? string("")
                    : string(str::stream() << ", elapsed secs: "
                                           << durationCount<Seconds>(elapsedMillisSinceQueued)))
            << ", cursor ids: [" << string(cursorList) << "]";
  
      entry->lastLoggedTS = currentTime;
  }
  
  struct RangeDeleter::NSMinMax {
      NSMinMax(std::string ns, const BSONObj min, const BSONObj max) : ns(ns), min(min), max(max) {}
  
      std::string ns;
  
      // Inclusive lower range.
      BSONObj min;
  
      // Exclusive upper range.
      BSONObj max;
  };
  
  bool RangeDeleter::NSMinMaxCmp::operator()(const NSMinMax* lhs, const NSMinMax* rhs) const {
      const int nsComp = lhs->ns.compare(rhs->ns);
  
      if (nsComp < 0) {
          return true;
      }
  
      if (nsComp > 0) {
          return false;
      }
  
      return compareRanges(lhs->min, lhs->max, rhs->min, rhs->max) < 0;
  }
  
  RangeDeleter::RangeDeleter(RangeDeleterEnv* env)
      : _env(env),  // ownership xfer
        _stopRequested(false),
        _deletesInProgress(0) {}
  
  RangeDeleter::~RangeDeleter() {
      for (TaskList::iterator it = _notReadyQueue.begin(); it != _notReadyQueue.end(); ++it) {
          delete (*it);
      }
  
      for (TaskList::iterator it = _taskQueue.begin(); it != _taskQueue.end(); ++it) {
          delete (*it);
      }
  
      for (NSMinMaxSet::iterator it = _deleteSet.begin(); it != _deleteSet.end(); ++it) {
          delete (*it);
      }
  
      for (std::deque<DeleteJobStats*>::iterator it = _statsHistory.begin();
           it != _statsHistory.end();
           ++it) {
          delete (*it);
      }
  }
  
  void RangeDeleter::startWorkers() {
      if (!_worker) {
          _worker.reset(new stdx::thread(stdx::bind(&RangeDeleter::doWork, this)));
      }
  }
  
  void RangeDeleter::stopWorkers() {
      {
          stdx::lock_guard<stdx::mutex> sl(_stopMutex);
          _stopRequested = true;
      }
  
      if (_worker) {
          _worker->join();
      }
  
      stdx::unique_lock<stdx::mutex> sl(_queueMutex);
      while (_deletesInProgress > 0) {
          _nothingInProgressCV.wait(sl);
      }
  }
  
  bool RangeDeleter::queueDelete(OperationContext* txn,
                                 const RangeDeleterOptions& options,
                                 Notification<void>* doneSignal,
                                 std::string* errMsg) {
      string dummy;
      if (errMsg == NULL)
          errMsg = &dummy;
  
      const string& ns(options.range.ns);
      const BSONObj& min(options.range.minKey);
      const BSONObj& max(options.range.maxKey);
  
      unique_ptr<RangeDeleteEntry> toDelete(new RangeDeleteEntry(options));
      toDelete->doneSignal = doneSignal;
  
      {
          stdx::lock_guard<stdx::mutex> sl(_queueMutex);
          if (_stopRequested) {
              *errMsg = "deleter is already stopped.";
              return false;
          }
  
          if (!canEnqueue_inlock(ns, min, max, errMsg)) {
              return false;
          }
  
          _deleteSet.insert(new NSMinMax(ns, min.getOwned(), max.getOwned()));
      }
  
      if (options.waitForOpenCursors) {
          _env->getCursorIds(txn, ns, &toDelete->cursorsToWait);
      }
  
      toDelete->stats.queueStartTS = jsTime();
  
      if (!toDelete->cursorsToWait.empty())
          logCursorsWaiting(toDelete.get());
  
      {
          stdx::lock_guard<stdx::mutex> sl(_queueMutex);
  
          if (toDelete->cursorsToWait.empty()) {
              toDelete->stats.queueEndTS = jsTime();
              _taskQueue.push_back(toDelete.release());
              _taskQueueNotEmptyCV.notify_one();
          } else {
              _notReadyQueue.push_back(toDelete.release());
          }
      }
  
      return true;
  }
  
  namespace {
  const int kWTimeoutMillis = 60 * 60 * 1000;
  
  bool _waitForMajority(OperationContext* txn, std::string* errMsg) {
      const WriteConcernOptions writeConcern(
          WriteConcernOptions::kMajority, WriteConcernOptions::SyncMode::UNSET, kWTimeoutMillis);
  
      repl::ReplicationCoordinator::StatusAndDuration replStatus =
          repl::getGlobalReplicationCoordinator()->awaitReplicationOfLastOpForClient(txn,
                                                                                     writeConcern);
      if (!replStatus.status.isOK()) {
          *errMsg = str::stream() << "rangeDeleter failed while waiting for replication after "
                                  << durationCount<Seconds>(replStatus.duration) << " seconds due to "
                                  << replStatus.status.toString();
          log() << *errMsg;
      }
  
      return replStatus.status.isOK();
  }
  }  // namespace
  
  bool RangeDeleter::deleteNow(OperationContext* txn,
                               const RangeDeleterOptions& options,
                               string* errMsg) {
      if (stopRequested()) {
          *errMsg = "deleter is already stopped.";
          return false;
      }
  
      string dummy;
      if (errMsg == NULL)
          errMsg = &dummy;
  
      const string& ns(options.range.ns);
      const BSONObj& min(options.range.minKey);
      const BSONObj& max(options.range.maxKey);
  
      NSMinMax deleteRange(ns, min, max);
      {
          stdx::lock_guard<stdx::mutex> sl(_queueMutex);
          if (!canEnqueue_inlock(ns, min, max, errMsg)) {
              return false;
          }
  
          _deleteSet.insert(&deleteRange);
  
          // Note: count for pending deletes is an integral part of the shutdown story.
          // Therefore, to simplify things, there is no "pending" state for deletes in
          // deleteNow, the state transition is simply inProgress -> done.
          _deletesInProgress++;
      }
  
      RangeDeleteEntry taskDetails(options);
      if (options.waitForOpenCursors) {
          _env->getCursorIds(txn, ns, &taskDetails.cursorsToWait);
      }
  
      long long checkIntervalMillis = 5;
  
  
      taskDetails.stats.queueStartTS = jsTime();
  
      for (; !taskDetails.cursorsToWait.empty(); sleepmillis(checkIntervalMillis)) {
          logCursorsWaiting(&taskDetails);
  
          set<CursorId> cursorsNow;
          _env->getCursorIds(txn, ns, &cursorsNow);
  
          set<CursorId> cursorsLeft;
          std::set_intersection(taskDetails.cursorsToWait.begin(),
                                taskDetails.cursorsToWait.end(),
                                cursorsNow.begin(),
                                cursorsNow.end(),
                                std::inserter(cursorsLeft, cursorsLeft.end()));
  
          taskDetails.cursorsToWait.swap(cursorsLeft);
  
          if (stopRequested()) {
              *errMsg = "deleter was stopped.";
  
              stdx::lock_guard<stdx::mutex> sl(_queueMutex);
              _deleteSet.erase(&deleteRange);
  
              _deletesInProgress--;
  
              if (_deletesInProgress == 0) {
                  _nothingInProgressCV.notify_one();
              }
  
              return false;
          }
  
          if (checkIntervalMillis < kMaxCursorCheckIntervalMillis) {
              checkIntervalMillis *= 2;
          }
      }
      taskDetails.stats.queueEndTS = jsTime();
  
      taskDetails.stats.deleteStartTS = jsTime();
      bool result = _env->deleteRange(txn, taskDetails, &taskDetails.stats.deletedDocCount, errMsg);
  
      taskDetails.stats.deleteEndTS = jsTime();
  
      if (result) {
          taskDetails.stats.waitForReplStartTS = jsTime();
          result = _waitForMajority(txn, errMsg);
          taskDetails.stats.waitForReplEndTS = jsTime();
      }
  
      {
          stdx::lock_guard<stdx::mutex> sl(_queueMutex);
          _deleteSet.erase(&deleteRange);
  
          _deletesInProgress--;
  
          if (_deletesInProgress == 0) {
              _nothingInProgressCV.notify_one();
          }
      }
  
      recordDelStats(new DeleteJobStats(taskDetails.stats));
      return result;
  }
  
  void RangeDeleter::getStatsHistory(std::vector<DeleteJobStats*>* stats) const {
      stats->clear();
      stats->reserve(kDeleteJobsHistory);
  
      stdx::lock_guard<stdx::mutex> sl(_statsHistoryMutex);
      for (std::deque<DeleteJobStats*>::const_iterator it = _statsHistory.begin();
           it != _statsHistory.end();
           ++it) {
          stats->push_back(new DeleteJobStats(**it));
      }
  }
  
  BSONObj RangeDeleter::toBSON() const {
      stdx::lock_guard<stdx::mutex> sl(_queueMutex);
  
      BSONObjBuilder builder;
  
      BSONArrayBuilder notReadyBuilder(builder.subarrayStart("notReady"));
      for (TaskList::const_iterator iter = _notReadyQueue.begin(); iter != _notReadyQueue.end();
           ++iter) {
          notReadyBuilder.append((*iter)->toBSON());
      }
      notReadyBuilder.doneFast();
  
      BSONArrayBuilder readyBuilder(builder.subarrayStart("ready"));
      for (TaskList::const_iterator iter = _taskQueue.begin(); iter != _taskQueue.end(); ++iter) {
          readyBuilder.append((*iter)->toBSON());
      }
      readyBuilder.doneFast();
  
      return builder.obj();
  }
  
  void RangeDeleter::doWork() {
      Client::initThreadIfNotAlready("RangeDeleter");
      Client* client = &cc();
  
      while (!inShutdown() && !stopRequested()) {
          string errMsg;
  
          RangeDeleteEntry* nextTask = NULL;
  
          {
              stdx::unique_lock<stdx::mutex> sl(_queueMutex);
              while (_taskQueue.empty()) {
                  {
                      MONGO_IDLE_THREAD_BLOCK;
                      _taskQueueNotEmptyCV.wait_for(
                          sl, Milliseconds(kNotEmptyTimeoutMillis).toSystemDuration());
                  }
  
                  if (stopRequested()) {
                      log() << "stopping range deleter worker" << endl;
                      return;
                  }
  
                  if (_taskQueue.empty()) {
                      // Try to check if some deletes are ready and move them to the
                      // ready queue.
  
                      TaskList::iterator iter = _notReadyQueue.begin();
                      while (iter != _notReadyQueue.end()) {
                          RangeDeleteEntry* entry = *iter;
  
                          set<CursorId> cursorsNow;
                          if (entry->options.waitForOpenCursors) {
                              auto txn = client->makeOperationContext();
                              _env->getCursorIds(txn.get(), entry->options.range.ns, &cursorsNow);
                          }
  
                          set<CursorId> cursorsLeft;
                          std::set_intersection(entry->cursorsToWait.begin(),
                                                entry->cursorsToWait.end(),
                                                cursorsNow.begin(),
                                                cursorsNow.end(),
                                                std::inserter(cursorsLeft, cursorsLeft.end()));
  
                          entry->cursorsToWait.swap(cursorsLeft);
  
                          if (entry->cursorsToWait.empty()) {
                              (*iter)->stats.queueEndTS = jsTime();
                              _taskQueue.push_back(*iter);
                              _taskQueueNotEmptyCV.notify_one();
                              iter = _notReadyQueue.erase(iter);
                          } else {
                              logCursorsWaiting(entry);
                              ++iter;
                          }
                      }
                  }
              }
  
              if (stopRequested()) {
                  log() << "stopping range deleter worker" << endl;
                  return;
              }
  
              nextTask = _taskQueue.front();
              _taskQueue.pop_front();
  
              _deletesInProgress++;
          }
  
          {
              auto txn = client->makeOperationContext();
              nextTask->stats.deleteStartTS = jsTime();
              bool delResult =
                  _env->deleteRange(txn.get(), *nextTask, &nextTask->stats.deletedDocCount, &errMsg);
              nextTask->stats.deleteEndTS = jsTime();
  
              if (delResult) {
                  nextTask->stats.waitForReplStartTS = jsTime();
  
                  if (!_waitForMajority(txn.get(), &errMsg)) {
                      warning() << "Error encountered while waiting for replication: " << errMsg;
                  }
  
                  nextTask->stats.waitForReplEndTS = jsTime();
              } else {
                  warning() << "Error encountered while trying to delete range: " << redact(errMsg);
              }
          }
  
          {
              stdx::lock_guard<stdx::mutex> sl(_queueMutex);
  
              NSMinMax setEntry(nextTask->options.range.ns,
                                nextTask->options.range.minKey,
                                nextTask->options.range.maxKey);
              deletePtrElement(&_deleteSet, &setEntry);
              _deletesInProgress--;
  
              if (nextTask->doneSignal) {
                  nextTask->doneSignal->set();
              }
          }
  
          recordDelStats(new DeleteJobStats(nextTask->stats));
          delete nextTask;
          nextTask = NULL;
      }
  }
  
  bool RangeDeleter::canEnqueue_inlock(StringData ns,
                                       const BSONObj& min,
                                       const BSONObj& max,
                                       string* errMsg) const {
      NSMinMax toDelete(ns.toString(), min, max);
      if (_deleteSet.count(&toDelete) > 0) {
          *errMsg = str::stream() << "ns: " << ns << ", min: " << min << ", max: " << max
                                  << " is already being processed for deletion.";
          return false;
      }
  
      return true;
  }
  
  bool RangeDeleter::stopRequested() const {
      stdx::lock_guard<stdx::mutex> sl(_stopMutex);
      return _stopRequested;
  }
  
  size_t RangeDeleter::getTotalDeletes() const {
      stdx::lock_guard<stdx::mutex> sl(_queueMutex);
      return _deleteSet.size();
  }
  
  size_t RangeDeleter::getPendingDeletes() const {
      stdx::lock_guard<stdx::mutex> sl(_queueMutex);
      return _notReadyQueue.size() + _taskQueue.size();
  }
  
  size_t RangeDeleter::getDeletesInProgress() const {
      stdx::lock_guard<stdx::mutex> sl(_queueMutex);
      return _deletesInProgress;
  }
  
  void RangeDeleter::recordDelStats(DeleteJobStats* newStat) {
      stdx::lock_guard<stdx::mutex> sl(_statsHistoryMutex);
      if (_statsHistory.size() == kDeleteJobsHistory) {
          delete _statsHistory.front();
          _statsHistory.pop_front();
      }
  
      _statsHistory.push_back(newStat);
  }
  
  RangeDeleteEntry::RangeDeleteEntry(const RangeDeleterOptions& options)
      : options(options), doneSignal(nullptr) {}
  
  BSONObj RangeDeleteEntry::toBSON() const {
      BSONObjBuilder builder;
      builder.append("ns", options.range.ns);
      builder.append("min", options.range.minKey);
      builder.append("max", options.range.maxKey);
      BSONArrayBuilder cursorBuilder(builder.subarrayStart("cursors"));
  
      for (std::set<CursorId>::const_iterator it = cursorsToWait.begin(); it != cursorsToWait.end();
           ++it) {
          cursorBuilder.append((long long)*it);
      }
      cursorBuilder.doneFast();
  
      return builder.done().copy();
  }
  
  RangeDeleterOptions::RangeDeleterOptions(const KeyRange& range)
      : range(range), fromMigrate(false), onlyRemoveOrphanedDocs(false), waitForOpenCursors(false) {}
  }
  