text: /**
   *    Copyright (C) 2008-2015 MongoDB 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 <algorithm>
  #include <map>
  #include <string>
  #include <vector>
  
  #include "mongo/client/connpool.h"
  #include "mongo/db/auth/action_set.h"
  #include "mongo/db/auth/action_type.h"
  #include "mongo/db/auth/authorization_manager.h"
  #include "mongo/db/auth/authorization_manager_global.h"
  #include "mongo/db/auth/authorization_session.h"
  #include "mongo/db/auth/privilege.h"
  #include "mongo/db/catalog/document_validation.h"
  #include "mongo/db/catalog/index_create.h"
  #include "mongo/db/clientcursor.h"
  #include "mongo/db/commands.h"
  #include "mongo/db/concurrency/lock_state.h"
  #include "mongo/db/db_raii.h"
  #include "mongo/db/dbhelpers.h"
  #include "mongo/db/exec/plan_stage.h"
  #include "mongo/db/field_parser.h"
  #include "mongo/db/hasher.h"
  #include "mongo/db/jsobj.h"
  #include "mongo/db/namespace_string.h"
  #include "mongo/db/op_observer.h"
  #include "mongo/db/operation_context_impl.h"
  #include "mongo/db/ops/delete.h"
  #include "mongo/db/query/internal_plans.h"
  #include "mongo/db/query/query_knobs.h"
  #include "mongo/db/range_deleter_service.h"
  #include "mongo/db/repl/repl_client_info.h"
  #include "mongo/db/repl/replication_coordinator_global.h"
  #include "mongo/db/service_context.h"
  #include "mongo/db/storage/mmap_v1/dur.h"
  #include "mongo/db/s/collection_metadata.h"
  #include "mongo/db/s/sharded_connection_info.h"
  #include "mongo/db/s/sharding_state.h"
  #include "mongo/db/write_concern.h"
  #include "mongo/logger/ramlog.h"
  #include "mongo/s/catalog/catalog_manager.h"
  #include "mongo/s/catalog/dist_lock_manager.h"
  #include "mongo/s/catalog/type_chunk.h"
  #include "mongo/s/chunk.h"
  #include "mongo/s/chunk_version.h"
  #include "mongo/s/client/shard.h"
  #include "mongo/s/client/shard_registry.h"
  #include "mongo/s/config.h"
  #include "mongo/s/d_state.h"
  #include "mongo/s/grid.h"
  #include "mongo/s/shard_key_pattern.h"
  #include "mongo/stdx/chrono.h"
  #include "mongo/stdx/condition_variable.h"
  #include "mongo/stdx/memory.h"
  #include "mongo/stdx/mutex.h"
  #include "mongo/stdx/thread.h"
  #include "mongo/util/assert_util.h"
  #include "mongo/util/elapsed_tracker.h"
  #include "mongo/util/exit.h"
  #include "mongo/util/fail_point_service.h"
  #include "mongo/util/log.h"
  #include "mongo/util/processinfo.h"
  #include "mongo/util/startup_test.h"
  
  // Pause while a fail point is enabled.
  #define MONGO_FP_PAUSE_WHILE(symbol)   \
      while (MONGO_FAIL_POINT(symbol)) { \
          sleepmillis(100);              \
      }
  
  namespace mongo {
  
  using namespace stdx::chrono;
  using std::list;
  using std::set;
  using std::string;
  using std::unique_ptr;
  using std::vector;
  
  namespace {
  
  const int kDefaultWTimeoutMs = 60 * 1000;
  const WriteConcernOptions DefaultWriteConcern(2, WriteConcernOptions::NONE, kDefaultWTimeoutMs);
  
  /**
   * Returns the default write concern for migration cleanup (at donor shard) and
   * cloning documents (at recipient shard).
   */
  WriteConcernOptions getDefaultWriteConcern() {
      repl::ReplicationCoordinator* replCoordinator = repl::getGlobalReplicationCoordinator();
      if (replCoordinator->getReplicationMode() == mongo::repl::ReplicationCoordinator::modeReplSet) {
          Status status = replCoordinator->checkIfWriteConcernCanBeSatisfied(DefaultWriteConcern);
          if (status.isOK()) {
              return DefaultWriteConcern;
          }
      }
  
      return WriteConcernOptions(1, WriteConcernOptions::NONE, 0);
  }
  
  }  // namespace
  
  MONGO_FP_DECLARE(failMigrationCommit);
  MONGO_FP_DECLARE(failMigrationConfigWritePrepare);
  MONGO_FP_DECLARE(failMigrationApplyOps);
  
  Tee* migrateLog = RamLog::get("migrate");
  
  class MoveTimingHelper {
  public:
      MoveTimingHelper(OperationContext* txn,
                       const string& where,
                       const string& ns,
                       BSONObj min,
                       BSONObj max,
                       int total,
                       string* cmdErrmsg,
                       string toShard,
                       string fromShard)
          : _txn(txn),
            _where(where),
            _ns(ns),
            _to(toShard),
            _from(fromShard),
            _next(0),
            _total(total),
            _cmdErrmsg(cmdErrmsg) {
          _b.append("min", min);
          _b.append("max", max);
      }
  
      ~MoveTimingHelper() {
          // even if logChange doesn't throw, bson does
          // sigh
          try {
              if (!_to.empty()) {
                  _b.append("to", _to);
              }
              if (!_from.empty()) {
                  _b.append("from", _from);
              }
              if (_next != _total) {
                  _b.append("note", "aborted");
              } else {
                  _b.append("note", "success");
              }
              if (!_cmdErrmsg->empty()) {
                  _b.append("errmsg", *_cmdErrmsg);
              }
  
              grid.catalogManager()->logChange(_txn->getClient()->clientAddress(true),
                                               (string) "moveChunk." + _where,
                                               _ns,
                                               _b.obj());
          } catch (const std::exception& e) {
              warning() << "couldn't record timing for moveChunk '" << _where << "': " << e.what()
                        << migrateLog;
          }
      }
  
      void done(int step) {
          invariant(step == ++_next);
          invariant(step <= _total);
  
          const string s = str::stream() << "step " << step << " of " << _total;
  
          CurOp* op = CurOp::get(_txn);
          {
              stdx::lock_guard<Client> lk(*_txn->getClient());
              op->setMessage_inlock(s.c_str());
          }
  
          _b.appendNumber(s, _t.millis());
          _t.reset();
      }
  
  private:
      OperationContext* const _txn;
      Timer _t;
  
      string _where;
      string _ns;
      string _to;
      string _from;
  
      int _next;
      int _total;  // expected # of steps
  
      const string* _cmdErrmsg;
  
      BSONObjBuilder _b;
  };
  
  class ChunkCommandHelper : public Command {
  public:
      ChunkCommandHelper(const char* name) : Command(name) {}
  
      virtual void help(std::stringstream& help) const {
          help << "internal - should not be called directly";
      }
      virtual bool slaveOk() const {
          return false;
      }
      virtual bool adminOnly() const {
          return true;
      }
      virtual bool isWriteCommandForConfigServer() const {
          return false;
      }
  };
  
  bool isInRange(const BSONObj& obj,
                 const BSONObj& min,
                 const BSONObj& max,
                 const BSONObj& shardKeyPattern) {
      ShardKeyPattern shardKey(shardKeyPattern);
      BSONObj k = shardKey.extractShardKeyFromDoc(obj);
      return k.woCompare(min) >= 0 && k.woCompare(max) < 0;
  }
  
  class MigrateFromStatus {
  public:
      MigrateFromStatus() : _inCriticalSection(false), _memoryUsed(0), _active(false) {}
  
      /**
       * @return false if cannot start. One of the reason for not being able to
       *     start is there is already an existing migration in progress.
       */
      bool start(OperationContext* txn,
                 const std::string& ns,
                 const BSONObj& min,
                 const BSONObj& max,
                 const BSONObj& shardKeyPattern) {
          verify(!min.isEmpty());
          verify(!max.isEmpty());
          verify(!ns.empty());
  
          // Get global shared to synchronize with logOp. Also see comments in the class
          // members declaration for more details.
          Lock::GlobalRead globalShared(txn->lockState());
          stdx::lock_guard<stdx::mutex> lk(_mutex);
  
          if (_active) {
              return false;
          }
  
          _ns = ns;
          _min = min;
          _max = max;
          _shardKeyPattern = shardKeyPattern;
  
          verify(_deleted.size() == 0);
          verify(_reload.size() == 0);
          verify(_memoryUsed == 0);
  
          _active = true;
  
          stdx::lock_guard<stdx::mutex> tLock(_cloneLocsMutex);
          verify(_cloneLocs.size() == 0);
  
          return true;
      }
  
      void done(OperationContext* txn) {
          log() << "MigrateFromStatus::done About to acquire global lock to exit critical "
                   "section";
  
          // Get global shared to synchronize with logOp. Also see comments in the class
          // members declaration for more details.
          Lock::GlobalRead globalShared(txn->lockState());
          stdx::lock_guard<stdx::mutex> lk(_mutex);
  
          _active = false;
          _deleteNotifyExec.reset(NULL);
          _inCriticalSection = false;
          _inCriticalSectionCV.notify_all();
  
          _deleted.clear();
          _reload.clear();
          _memoryUsed = 0;
  
          stdx::lock_guard<stdx::mutex> cloneLock(_cloneLocsMutex);
          _cloneLocs.clear();
      }
  
      void logOp(OperationContext* txn,
                 const char* opstr,
                 const char* ns,
                 const BSONObj& obj,
                 BSONObj* patt,
                 bool notInActiveChunk) {
          ensureShardVersionOKOrThrow(txn->getClient(), ns);
  
          const char op = opstr[0];
  
          if (notInActiveChunk) {
              // Ignore writes that came from the migration process like cleanup so they
              // won't be transferred to the recipient shard. Also ignore ops from
              // _migrateClone and _transferMods since it is impossible to move a chunk
              // to self.
              return;
          }
  
          dassert(txn->lockState()->isWriteLocked());  // Must have Global IX.
  
          if (!_active)
              return;
  
          if (_ns != ns)
              return;
  
          // no need to log if this is not an insertion, an update, or an actual deletion
          // note: opstr 'db' isn't a deletion but a mention that a database exists
          // (for replication machinery mostly).
          if (op == 'n' || op == 'c' || (op == 'd' && opstr[1] == 'b'))
              return;
  
          BSONElement ide;
          if (patt)
              ide = patt->getField("_id");
          else
              ide = obj["_id"];
  
          if (ide.eoo()) {
              warning() << "logOpForSharding got mod with no _id, ignoring  obj: " << obj
                        << migrateLog;
              return;
          }
  
          if (op == 'i' && (!isInRange(obj, _min, _max, _shardKeyPattern))) {
              return;
          }
  
          BSONObj idObj(ide.wrap());
  
          if (op == 'u') {
              BSONObj fullDoc;
              OldClientContext ctx(txn, _ns, false);
              if (!Helpers::findById(txn, ctx.db(), _ns.c_str(), idObj, fullDoc)) {
                  warning() << "logOpForSharding couldn't find: " << idObj
                            << " even though should have" << migrateLog;
                  dassert(false);  // TODO: Abort the migration.
                  return;
              }
  
              if (!isInRange(fullDoc, _min, _max, _shardKeyPattern)) {
                  return;
              }
          }
  
          // Note: can't check if delete is in active chunk since the document is gone!
  
          txn->recoveryUnit()->registerChange(new LogOpForShardingHandler(this, idObj, op));
      }
  
      /**
       * Insert items from docIdList to a new array with the given fieldName in the given
       * builder. If explode is true, the inserted object will be the full version of the
       * document. Note that the whenever an item from the docList is inserted to the array,
       * it will also be removed from docList.
       *
       * Should be holding the collection lock for ns if explode is true.
       */
      void xfer(OperationContext* txn,
                const string& ns,
                Database* db,
                list<BSONObj>* docIdList,
                BSONObjBuilder& builder,
                const char* fieldName,
                long long& size,
                bool explode) {
          const long long maxSize = 1024 * 1024;
  
          if (docIdList->size() == 0 || size > maxSize)
              return;
  
          BSONArrayBuilder arr(builder.subarrayStart(fieldName));
  
          list<BSONObj>::iterator docIdIter = docIdList->begin();
          while (docIdIter != docIdList->end() && size < maxSize) {
              BSONObj idDoc = *docIdIter;
              if (explode) {
                  BSONObj fullDoc;
                  if (Helpers::findById(txn, db, ns.c_str(), idDoc, fullDoc)) {
                      arr.append(fullDoc);
                      size += fullDoc.objsize();
                  }
              } else {
                  arr.append(idDoc);
                  size += idDoc.objsize();
              }
  
              docIdIter = docIdList->erase(docIdIter);
          }
  
          arr.done();
      }
  
      /**
       * called from the dest of a migrate
       * transfers mods from src to dest
       */
      bool transferMods(OperationContext* txn, string& errmsg, BSONObjBuilder& b) {
          long long size = 0;
  
          {
              AutoGetCollectionForRead ctx(txn, getNS());
  
              stdx::lock_guard<stdx::mutex> sl(_mutex);
              if (!_active) {
                  errmsg = "no active migration!";
                  return false;
              }
  
              // TODO: fix SERVER-16540 race
              xfer(txn, _ns, ctx.getDb(), &_deleted, b, "deleted", size, false);
              xfer(txn, _ns, ctx.getDb(), &_reload, b, "reload", size, true);
          }
  
          b.append("size", size);
  
          return true;
      }
  
      /**
       * Get the disklocs that belong to the chunk migrated and sort them in _cloneLocs
       * (to avoid seeking disk later).
       *
       * @param maxChunkSize number of bytes beyond which a chunk's base data (no indices)
       *                     is considered too large to move.
       * @param errmsg filled with textual description of error if this call return false.
       * @return false if approximate chunk size is too big to move or true otherwise.
       */
      bool storeCurrentLocs(OperationContext* txn,
                            long long maxChunkSize,
                            string& errmsg,
                            BSONObjBuilder& result) {
          AutoGetCollectionForRead ctx(txn, getNS());
          Collection* collection = ctx.getCollection();
          if (!collection) {
              errmsg = "ns not found, should be impossible";
              return false;
          }
  
          // Allow multiKey based on the invariant that shard keys must be single-valued.
          // Therefore, any multi-key index prefixed by shard key cannot be multikey over
          // the shard key fields.
          IndexDescriptor* idx =
              collection->getIndexCatalog()->findShardKeyPrefixedIndex(txn,
                                                                       _shardKeyPattern,
                                                                       false);  // requireSingleKey
  
          if (idx == NULL) {
              errmsg = str::stream() << "can't find index with prefix " << _shardKeyPattern
                                     << " in storeCurrentLocs for " << _ns;
              return false;
          }
  
          // Assume both min and max non-empty, append MinKey's to make them fit chosen index
          BSONObj min;
          BSONObj max;
          KeyPattern kp(idx->keyPattern());
  
          {
              // It's alright not to lock _mutex all the way through based on the assumption
              // that this is only called by the main thread that drives the migration and
              // only it can start and stop the current migration.
              stdx::lock_guard<stdx::mutex> sl(_mutex);
  
              invariant(_deleteNotifyExec.get() == NULL);
              unique_ptr<WorkingSet> ws = stdx::make_unique<WorkingSet>();
              unique_ptr<DeleteNotificationStage> dns = stdx::make_unique<DeleteNotificationStage>();
              // Takes ownership of 'ws' and 'dns'.
              auto statusWithPlanExecutor = PlanExecutor::make(
                  txn, std::move(ws), std::move(dns), collection, PlanExecutor::YIELD_MANUAL);
              invariant(statusWithPlanExecutor.isOK());
              _deleteNotifyExec = std::move(statusWithPlanExecutor.getValue());
              _deleteNotifyExec->registerExec();
  
              min = Helpers::toKeyFormat(kp.extendRangeBound(_min, false));
              max = Helpers::toKeyFormat(kp.extendRangeBound(_max, false));
          }
  
          unique_ptr<PlanExecutor> exec(
              InternalPlanner::indexScan(txn, collection, idx, min, max, false));
          // We can afford to yield here because any change to the base data that we might
          // miss is already being queued and will migrate in the 'transferMods' stage.
          exec->setYieldPolicy(PlanExecutor::YIELD_AUTO);
  
          // use the average object size to estimate how many objects a full chunk would carry
          // do that while traversing the chunk's range using the sharding index, below
          // there's a fair amount of slack before we determine a chunk is too large because object
          // sizes will vary
          unsigned long long maxRecsWhenFull;
          long long avgRecSize;
          const long long totalRecs = collection->numRecords(txn);
          if (totalRecs > 0) {
              avgRecSize = collection->dataSize(txn) / totalRecs;
              maxRecsWhenFull = maxChunkSize / avgRecSize;
              maxRecsWhenFull = std::min((unsigned long long)(Chunk::MaxObjectPerChunk + 1),
                                         130 * maxRecsWhenFull / 100 /* slack */);
          } else {
              avgRecSize = 0;
              maxRecsWhenFull = Chunk::MaxObjectPerChunk + 1;
          }
  
          // do a full traversal of the chunk and don't stop even if we think it is a large chunk
          // we want the number of records to better report, in that case
          bool isLargeChunk = false;
          unsigned long long recCount = 0;
          ;
          RecordId dl;
          while (PlanExecutor::ADVANCED == exec->getNext(NULL, &dl)) {
              if (!isLargeChunk) {
                  stdx::lock_guard<stdx::mutex> lk(_cloneLocsMutex);
                  _cloneLocs.insert(dl);
              }
  
              if (++recCount > maxRecsWhenFull) {
                  isLargeChunk = true;
                  // continue on despite knowing that it will fail,
                  // just to get the correct value for recCount.
              }
          }
          exec.reset();
  
          if (isLargeChunk) {
              stdx::lock_guard<stdx::mutex> sl(_mutex);
              warning() << "cannot move chunk: the maximum number of documents for a chunk is "
                        << maxRecsWhenFull << " , the maximum chunk size is " << maxChunkSize
                        << " , average document size is " << avgRecSize << ". Found " << recCount
                        << " documents in chunk "
                        << " ns: " << _ns << " " << _min << " -> " << _max << migrateLog;
  
              result.appendBool("chunkTooBig", true);
              result.appendNumber("estimatedChunkSize", (long long)(recCount * avgRecSize));
              errmsg = "chunk too big to move";
              return false;
          }
  
          log() << "moveChunk number of documents: " << cloneLocsRemaining() << migrateLog;
  
          txn->recoveryUnit()->abandonSnapshot();
          return true;
      }
  
      bool clone(OperationContext* txn, string& errmsg, BSONObjBuilder& result) {
          ElapsedTracker tracker(internalQueryExecYieldIterations, internalQueryExecYieldPeriodMS);
  
          int allocSize = 0;
          {
              AutoGetCollectionForRead ctx(txn, getNS());
  
              stdx::lock_guard<stdx::mutex> sl(_mutex);
              if (!_active) {
                  errmsg = "not active";
                  return false;
              }
  
              Collection* collection = ctx.getCollection();
              if (!collection) {
                  errmsg = str::stream() << "collection " << _ns << " does not exist";
                  return false;
              }
  
              allocSize = std::min(
                  BSONObjMaxUserSize,
                  static_cast<int>((12 + collection->averageObjectSize(txn)) * cloneLocsRemaining()));
          }
  
          bool isBufferFilled = false;
          BSONArrayBuilder clonedDocsArrayBuilder(allocSize);
          while (!isBufferFilled) {
              AutoGetCollectionForRead ctx(txn, getNS());
  
              stdx::lock_guard<stdx::mutex> sl(_mutex);
              if (!_active) {
                  errmsg = "not active";
                  return false;
              }
  
              // TODO: fix SERVER-16540 race
  
              Collection* collection = ctx.getCollection();
  
              if (!collection) {
                  errmsg = str::stream() << "collection " << _ns << " does not exist";
                  return false;
              }
  
              stdx::lock_guard<stdx::mutex> lk(_cloneLocsMutex);
              set<RecordId>::iterator cloneLocsIter = _cloneLocs.begin();
              for (; cloneLocsIter != _cloneLocs.end(); ++cloneLocsIter) {
                  if (tracker.intervalHasElapsed())  // should I yield?
                      break;
  
                  RecordId dl = *cloneLocsIter;
                  Snapshotted<BSONObj> doc;
                  if (!collection->findDoc(txn, dl, &doc)) {
                      // doc was deleted
                      continue;
                  }
  
                  // Use the builder size instead of accumulating 'doc's size so that we take
                  // into consideration the overhead of BSONArray indices, and *always*
                  // append one doc.
                  if (clonedDocsArrayBuilder.arrSize() != 0 &&
                      (clonedDocsArrayBuilder.len() + doc.value().objsize() + 1024) >
                          BSONObjMaxUserSize) {
                      isBufferFilled = true;  // break out of outer while loop
                      break;
                  }
  
                  clonedDocsArrayBuilder.append(doc.value());
              }
  
              _cloneLocs.erase(_cloneLocs.begin(), cloneLocsIter);
  
              // Note: must be holding _cloneLocsMutex, don't move this inside while condition!
              if (_cloneLocs.empty()) {
                  break;
              }
          }
  
          result.appendArray("objects", clonedDocsArrayBuilder.arr());
          return true;
      }
  
      void aboutToDelete(const RecordId& dl) {
          // Even though above we call findDoc to check for existance
          // that check only works for non-mmapv1 engines, and this is needed
          // for mmapv1.
  
          stdx::lock_guard<stdx::mutex> lk(_cloneLocsMutex);
          _cloneLocs.erase(dl);
      }
  
      std::size_t cloneLocsRemaining() {
          stdx::lock_guard<stdx::mutex> lk(_cloneLocsMutex);
          return _cloneLocs.size();
      }
  
      long long mbUsed() const {
          stdx::lock_guard<stdx::mutex> lk(_mutex);
          return _memoryUsed / (1024 * 1024);
      }
  
      bool getInCriticalSection() const {
          stdx::lock_guard<stdx::mutex> lk(_mutex);
          return _inCriticalSection;
      }
  
      void setInCriticalSection(bool b) {
          stdx::lock_guard<stdx::mutex> lk(_mutex);
          _inCriticalSection = b;
          _inCriticalSectionCV.notify_all();
      }
  
      std::string getNS() const {
          stdx::lock_guard<stdx::mutex> sl(_mutex);
          return _ns;
      }
  
      /**
       * @return true if we are NOT in the critical section
       */
      bool waitTillNotInCriticalSection(int maxSecondsToWait) {
          const auto deadline = system_clock::now() + seconds(maxSecondsToWait);
          stdx::unique_lock<stdx::mutex> lk(_mutex);
          while (_inCriticalSection) {
              if (stdx::cv_status::timeout == _inCriticalSectionCV.wait_until(lk, deadline))
                  return false;
          }
  
          return true;
      }
  
      bool isActive() const {
          return _getActive();
      }
  
  private:
      bool _getActive() const {
          stdx::lock_guard<stdx::mutex> lk(_mutex);
          return _active;
      }
      void _setActive(bool b) {
          stdx::lock_guard<stdx::mutex> lk(_mutex);
          _active = b;
      }
  
      /**
       * Used to commit work for LogOpForSharding. Used to keep track of changes in documents
       * that are part of a chunk being migrated.
       */
      class LogOpForShardingHandler : public RecoveryUnit::Change {
      public:
          /**
           * Invariant: idObj should belong to a document that is part of the active chunk
           * being migrated.
           */
          LogOpForShardingHandler(MigrateFromStatus* migrateFromStatus,
                                  const BSONObj& idObj,
                                  const char op)
              : _migrateFromStatus(migrateFromStatus), _idObj(idObj.getOwned()), _op(op) {}
  
          virtual void commit() {
              switch (_op) {
                  case 'd': {
                      stdx::lock_guard<stdx::mutex> sl(_migrateFromStatus->_mutex);
                      _migrateFromStatus->_deleted.push_back(_idObj);
                      _migrateFromStatus->_memoryUsed += _idObj.firstElement().size() + 5;
                      break;
                  }
  
                  case 'i':
                  case 'u': {
                      stdx::lock_guard<stdx::mutex> sl(_migrateFromStatus->_mutex);
                      _migrateFromStatus->_reload.push_back(_idObj);
                      _migrateFromStatus->_memoryUsed += _idObj.firstElement().size() + 5;
                      break;
                  }
  
                  default:
                      invariant(false);
              }
          }
  
          virtual void rollback() {}
  
      private:
          MigrateFromStatus* _migrateFromStatus;
          const BSONObj _idObj;
          const char _op;
      };
  
      /**
       * Used to receive invalidation notifications.
       *
       * XXX: move to the exec/ directory.
       */
      class DeleteNotificationStage : public PlanStage {
      public:
          virtual void invalidate(OperationContext* txn, const RecordId& dl, InvalidationType type);
  
          virtual StageState work(WorkingSetID* out) {
              invariant(false);
          }
          virtual bool isEOF() {
              invariant(false);
              return false;
          }
          virtual void kill() {}
          virtual void saveState() {
              invariant(false);
          }
          virtual void restoreState(OperationContext* opCtx) {
              invariant(false);
          }
          virtual unique_ptr<PlanStageStats> getStats() {
              invariant(false);
              return NULL;
          }
          virtual CommonStats* getCommonStats() const {
              invariant(false);
              return NULL;
          }
          virtual SpecificStats* getSpecificStats() const {
              invariant(false);
              return NULL;
          }
          virtual std::vector<PlanStage*> getChildren() const {
              vector<PlanStage*> empty;
              return empty;
          }
          virtual StageType stageType() const {
              return STAGE_NOTIFY_DELETE;
          }
      };
  
      //
      // All member variables are labeled with one of the following codes indicating the
      // synchronization rules for accessing them.
      //
      // (M)  Must hold _mutex for access.
      // (MG) For reads, _mutex *OR* Global IX Lock must be held.
      //      For writes, the _mutex *AND* (Global Shared or Exclusive Lock) must be held.
      // (C)  Must hold _cloneLocsMutex for access.
      //
      // Locking order:
      //
      // Global Lock -> _mutex -> _cloneLocsMutex
  
      mutable stdx::mutex _mutex;
  
      stdx::condition_variable _inCriticalSectionCV;  // (M)
  
      // Is migration currently in critical section. This can be used to block new writes.
      bool _inCriticalSection;  // (M)
  
      unique_ptr<PlanExecutor> _deleteNotifyExec;  // (M)
  
      // List of _id of documents that were modified that must be re-cloned.
      list<BSONObj> _reload;  // (M)
  
      // List of _id of documents that were deleted during clone that should be deleted later.
      list<BSONObj> _deleted;  // (M)
  
      // bytes in _reload + _deleted
      long long _memoryUsed;  // (M)
  
      // If a migration is currently active.
      bool _active;  // (MG)
  
      string _ns;                // (MG)
      BSONObj _min;              // (MG)
      BSONObj _max;              // (MG)
      BSONObj _shardKeyPattern;  // (MG)
  
      mutable stdx::mutex _cloneLocsMutex;
  
      // List of record id that needs to be transferred from here to the other side.
      set<RecordId> _cloneLocs;  // (C)
  
  } migrateFromStatus;
  
  void MigrateFromStatus::DeleteNotificationStage::invalidate(OperationContext* txn,
                                                              const RecordId& dl,
                                                              InvalidationType type) {
      if (type == INVALIDATION_DELETION) {
          migrateFromStatus.aboutToDelete(dl);
      }
  }
  
  struct MigrateStatusHolder {
      MigrateStatusHolder(OperationContext* txn,
                          const std::string& ns,
                          const BSONObj& min,
                          const BSONObj& max,
                          const BSONObj& shardKeyPattern)
          : _txn(txn) {
          _isAnotherMigrationActive = !migrateFromStatus.start(txn, ns, min, max, shardKeyPattern);
      }
      ~MigrateStatusHolder() {
          if (!_isAnotherMigrationActive) {
              migrateFromStatus.done(_txn);
          }
      }
  
      bool isAnotherMigrationActive() const {
          return _isAnotherMigrationActive;
      }
  
  private:
      OperationContext* _txn;
      bool _isAnotherMigrationActive;
  };
  
  void logOpForSharding(OperationContext* txn,
                        const char* opstr,
                        const char* ns,
                        const BSONObj& obj,
                        BSONObj* patt,
                        bool notInActiveChunk) {
      migrateFromStatus.logOp(txn, opstr, ns, obj, patt, notInActiveChunk);
  }
  
  class TransferModsCommand : public ChunkCommandHelper {
  public:
      void help(std::stringstream& h) const {
          h << "internal";
      }
      TransferModsCommand() : ChunkCommandHelper("_transferMods") {}
      virtual void addRequiredPrivileges(const std::string& dbname,
                                         const BSONObj& cmdObj,
                                         std::vector<Privilege>* out) {
          ActionSet actions;
          actions.addAction(ActionType::internal);
          out->push_back(Privilege(ResourcePattern::forClusterResource(), actions));
      }
      bool run(OperationContext* txn,
               const string&,
               BSONObj& cmdObj,
               int,
               string& errmsg,
               BSONObjBuilder& result) {
          return migrateFromStatus.transferMods(txn, errmsg, result);
      }
  } transferModsCommand;
  
  
  class InitialCloneCommand : public ChunkCommandHelper {
  public:
      void help(std::stringstream& h) const {
          h << "internal";
      }
      InitialCloneCommand() : ChunkCommandHelper("_migrateClone") {}
      virtual void addRequiredPrivileges(const std::string& dbname,
                                         const BSONObj& cmdObj,
                                         std::vector<Privilege>* out) {
          ActionSet actions;
          actions.addAction(ActionType::internal);
          out->push_back(Privilege(ResourcePattern::forClusterResource(), actions));
      }
      bool run(OperationContext* txn,
               const string&,
               BSONObj& cmdObj,
               int,
               string& errmsg,
               BSONObjBuilder& result) {
          return migrateFromStatus.clone(txn, errmsg, result);
      }
  } initialCloneCommand;
  
  // Tests can pause / resume moveChunk's progress at each step by enabling / disabling each fail
  // point.
  MONGO_FP_DECLARE(moveChunkHangAtStep1);
  MONGO_FP_DECLARE(moveChunkHangAtStep2);
  MONGO_FP_DECLARE(moveChunkHangAtStep3);
  MONGO_FP_DECLARE(moveChunkHangAtStep4);
  MONGO_FP_DECLARE(moveChunkHangAtStep5);
  MONGO_FP_DECLARE(moveChunkHangAtStep6);
  
  /**
   * this is the main entry for moveChunk
   * called to initial a move
   * usually by a mongos
   * this is called on the "from" side
   *
   * Format:
   * {
   *   moveChunk: "namespace",
   *   from: "hostAndPort",
   *   fromShard: "shardName",
   *   to: "hostAndPort",
   *   toShard: "shardName",
   *   min: {},
   *   max: {},
   *   maxChunkBytes: numeric,
   *   configdb: "hostAndPort",
   *
   *   // optional
   *   secondaryThrottle: bool, //defaults to true.
   *   writeConcern: {} // applies to individual writes.
   * }
   */
  class MoveChunkCommand : public Command {
  public:
      MoveChunkCommand() : Command("moveChunk") {}
      virtual void help(std::stringstream& help) const {
          help << "should not be calling this directly";
      }
  
      virtual bool slaveOk() const {
          return false;
      }
      virtual bool adminOnly() const {
          return true;
      }
      virtual bool isWriteCommandForConfigServer() const {
          return false;
      }
      virtual Status checkAuthForCommand(ClientBasic* client,
                                         const std::string& dbname,
                                         const BSONObj& cmdObj) {
          if (!AuthorizationSession::get(client)->isAuthorizedForActionsOnResource(
                  ResourcePattern::forExactNamespace(NamespaceString(parseNs(dbname, cmdObj))),
                  ActionType::moveChunk)) {
              return Status(ErrorCodes::Unauthorized, "Unauthorized");
          }
          return Status::OK();
      }
      virtual std::string parseNs(const std::string& dbname, const BSONObj& cmdObj) const {
          return parseNsFullyQualified(dbname, cmdObj);
      }
  
      bool run(OperationContext* txn,
               const string& dbname,
               BSONObj& cmdObj,
               int,
               string& errmsg,
               BSONObjBuilder& result) {
          // 1. Parse options
          // 2. Make sure my view is complete and lock the distributed lock to ensure shard
          //    metadata stability.
          // 3. Migration
          //    Retrieve all RecordIds, which need to be migrated in order to do as little seeking
          //    as possible during transfer. Retrieval of the RecordIds happens under a collection
          //    lock, but then the collection lock is dropped. This opens up an opportunity for
          //    repair or compact to invalidate these RecordIds, because these commands do not
          //    synchronized with migration. Note that data modifications are not a problem,
          //    because we are registered for change notifications.
          //
          // 4. pause till migrate caught up
          // 5. LOCK
          //    a) update my config, essentially locking
          //    b) finish migrate
          //    c) update config server
          //    d) logChange to config server
          // 6. wait for all current cursors to expire
          // 7. remove data locally
  
          // -------------------------------
  
          // 1.
          string ns = parseNs(dbname, cmdObj);
  
          // The shard addresses, redundant, but allows for validation
          string toShardHost = cmdObj["to"].str();
          string fromShardHost = cmdObj["from"].str();
  
          // The shard names
          string toShardName = cmdObj["toShard"].str();
          string fromShardName = cmdObj["fromShard"].str();
  
          // Process secondary throttle settings and assign defaults if necessary.
          BSONObj secThrottleObj;
          WriteConcernOptions writeConcern;
          Status status = writeConcern.parseSecondaryThrottle(cmdObj, &secThrottleObj);
  
          if (!status.isOK()) {
              if (status.code() != ErrorCodes::WriteConcernNotDefined) {
                  warning() << status.toString();
                  return appendCommandStatus(result, status);
              }
  
              writeConcern = getDefaultWriteConcern();
          } else {
              repl::ReplicationCoordinator* replCoordinator = repl::getGlobalReplicationCoordinator();
  
              if (replCoordinator->getReplicationMode() ==
                      repl::ReplicationCoordinator::modeMasterSlave &&
                  writeConcern.shouldWaitForOtherNodes()) {
                  warning() << "moveChunk cannot check if secondary throttle setting "
                            << writeConcern.toBSON()
                            << " can be enforced in a master slave configuration";
              }
  
              Status status = replCoordinator->checkIfWriteConcernCanBeSatisfied(writeConcern);
              if (!status.isOK() && status != ErrorCodes::NoReplicationEnabled) {
                  warning() << status.toString();
                  return appendCommandStatus(result, status);
              }
          }
  
          if (writeConcern.shouldWaitForOtherNodes() &&
              writeConcern.wTimeout == WriteConcernOptions::kNoTimeout) {
              // Don't allow no timeout.
              writeConcern.wTimeout = kDefaultWTimeoutMs;
          }
  
          // Do inline deletion
          bool waitForDelete = cmdObj["waitForDelete"].trueValue();
          if (waitForDelete) {
              log() << "moveChunk waiting for full cleanup after move";
          }
  
          BSONObj min = cmdObj["min"].Obj();
          BSONObj max = cmdObj["max"].Obj();
          BSONElement maxSizeElem = cmdObj["maxChunkSizeBytes"];
  
          if (ns.empty()) {
              errmsg = "need to specify namespace in command";
              return false;
          }
  
          if (toShardName.empty()) {
              errmsg = "need to specify shard to move chunk to";
              return false;
          }
          if (fromShardName.empty()) {
              errmsg = "need to specify shard to move chunk from";
              return false;
          }
  
          if (min.isEmpty()) {
              errmsg = "need to specify a min";
              return false;
          }
  
          if (max.isEmpty()) {
              errmsg = "need to specify a max";
              return false;
          }
  
          if (maxSizeElem.eoo() || !maxSizeElem.isNumber()) {
              errmsg = "need to specify maxChunkSizeBytes";
              return false;
          }
          const long long maxChunkSize = maxSizeElem.numberLong();  // in bytes
  
          // This could be the first call that enables sharding - make sure we initialize the
          // sharding state for this shard.
          if (!shardingState.enabled()) {
              if (cmdObj["configdb"].type() != String) {
                  errmsg = "sharding not enabled";
                  warning() << errmsg;
                  return false;
              }
  
              const string configdb = cmdObj["configdb"].String();
              shardingState.initialize(configdb);
          }
  
          // Initialize our current shard name in the shard state if needed
          shardingState.gotShardName(fromShardName);
  
          // Make sure we're as up-to-date as possible with shard information
          // This catches the case where we had to previously changed a shard's host by
          // removing/adding a shard with the same name
          Shard::reloadShardInfo();
  
          ConnectionString configLoc =
              ConnectionString::parse(shardingState.getConfigServer(), errmsg);
          if (!configLoc.isValid()) {
              warning() << errmsg;
              return false;
          }
  
          MoveTimingHelper timing(
              txn, "from", ns, min, max, 6 /* steps */, &errmsg, toShardName, fromShardName);
  
          log() << "received moveChunk request: " << cmdObj << migrateLog;
  
          timing.done(1);
          MONGO_FP_PAUSE_WHILE(moveChunkHangAtStep1);
  
          // 2.
  
          if (migrateFromStatus.isActive()) {
              errmsg = "migration already in progress";
              warning() << errmsg;
              return false;
          }
  
          //
          // Get the distributed lock
          //
  
          string whyMessage(str::stream() << "migrating chunk [" << minKey << ", " << maxKey
                                          << ") in " << ns);
          auto scopedDistLock = grid.catalogManager()->getDistLockManager()->lock(ns, whyMessage);
  
          if (!scopedDistLock.isOK()) {
              errmsg = stream() << "could not acquire collection lock for " << ns
                                << " to migrate chunk [" << minKey << "," << maxKey << ")"
                                << causedBy(scopedDistLock.getStatus());
  
              warning() << errmsg;
              return false;
          }
  
          BSONObj chunkInfo =
              BSON("min" << min << "max" << max << "from" << fromShardName << "to" << toShardName);
  
          grid.catalogManager()->logChange(
              txn->getClient()->clientAddress(true), "moveChunk.start", ns, chunkInfo);
  
          // Always refresh our metadata remotely
          ChunkVersion origShardVersion;
          Status refreshStatus = shardingState.refreshMetadataNow(txn, ns, &origShardVersion);
  
          if (!refreshStatus.isOK()) {
              errmsg = str::stream() << "moveChunk cannot start migrate of chunk "
                                     << "[" << minKey << "," << maxKey << ")"
                                     << causedBy(refreshStatus.reason());
  
              warning() << errmsg;
              return false;
          }
  
          if (origShardVersion.majorVersion() == 0) {
              // It makes no sense to migrate if our version is zero and we have no chunks
              errmsg = str::stream() << "moveChunk cannot start migrate of chunk "
                                     << "[" << minKey << "," << maxKey << ")"
                                     << " with zero shard version";
  
              warning() << errmsg;
              return false;
          }
  
          // From mongos >= v3.0.
          BSONElement epochElem(cmdObj["epoch"]);
          if (epochElem.type() == jstOID) {
              OID cmdEpoch = epochElem.OID();
  
              if (cmdEpoch != origShardVersion.epoch()) {
                  errmsg = str::stream() << "moveChunk cannot move chunk "
                                         << "[" << minKey << "," << maxKey << "), "
                                         << "collection may have been dropped. "
                                         << "current epoch: " << origShardVersion.epoch()
                                         << ", cmd epoch: " << cmdEpoch;
                  warning() << errmsg;
                  return false;
              }
          }
  
          // Get collection metadata
          const std::shared_ptr<CollectionMetadata> origCollMetadata(
              shardingState.getCollectionMetadata(ns));
  
          // With nonzero shard version, we must have metadata
          invariant(NULL != origCollMetadata);
  
          ChunkVersion origCollVersion = origCollMetadata->getCollVersion();
          BSONObj shardKeyPattern = origCollMetadata->getKeyPattern();
  
          // With nonzero shard version, we must have a coll version >= our shard version
          invariant(origCollVersion >= origShardVersion);
  
          // With nonzero shard version, we must have a shard key
          invariant(!shardKeyPattern.isEmpty());
  
          ChunkType origChunk;
          if (!origCollMetadata->getNextChunk(min, &origChunk) || origChunk.getMin().woCompare(min) ||
              origChunk.getMax().woCompare(max)) {
              // Our boundaries are different from those passed in
              errmsg = str::stream() << "moveChunk cannot find chunk "
                                     << "[" << minKey << "," << maxKey << ")"
                                     << " to migrate, the chunk boundaries may be stale";
  
              warning() << errmsg;
              return false;
          }
  
          log() << "moveChunk request accepted at version " << origShardVersion;
  
          timing.done(2);
          MONGO_FP_PAUSE_WHILE(moveChunkHangAtStep2);
  
          // 3.
          MigrateStatusHolder statusHolder(txn, ns, min, max, shardKeyPattern);
  
          if (statusHolder.isAnotherMigrationActive()) {
              errmsg = "moveChunk is already in progress from this shard";
              warning() << errmsg;
              return false;
          }
  
          ConnectionString fromShardCS;
          ConnectionString toShardCS;
  
          // Resolve the shard connection strings.
          {
              std::shared_ptr<Shard> fromShard = grid.shardRegistry()->getShard(fromShardName);
              uassert(28674,
                      str::stream() << "Source shard " << fromShardName
                                    << " is missing. This indicates metadata corruption.",
                      fromShard);
  
              fromShardCS = fromShard->getConnString();
  
              std::shared_ptr<Shard> toShard = grid.shardRegistry()->getShard(toShardName);
              uassert(28675,
                      str::stream() << "Destination shard " << toShardName
                                    << " is missing. This indicates metadata corruption.",
                      toShard);
  
              toShardCS = toShard->getConnString();
          }
  
          {
              // See comment at the top of the function for more information on what
              // synchronization is used here.
              if (!migrateFromStatus.storeCurrentLocs(txn, maxChunkSize, errmsg, result)) {
                  warning() << errmsg;
                  return false;
              }
  
              ScopedDbConnection connTo(toShardCS);
              BSONObj res;
              bool ok;
  
              const bool isSecondaryThrottle(writeConcern.shouldWaitForOtherNodes());
  
              BSONObjBuilder recvChunkStartBuilder;
              recvChunkStartBuilder.append("_recvChunkStart", ns);
              recvChunkStartBuilder.append("from", fromShardCS.toString());
              recvChunkStartBuilder.append("fromShardName", fromShardName);
              recvChunkStartBuilder.append("toShardName", toShardName);
              recvChunkStartBuilder.append("min", min);
              recvChunkStartBuilder.append("max", max);
              recvChunkStartBuilder.append("shardKeyPattern", shardKeyPattern);
              recvChunkStartBuilder.append("configServer", shardingState.getConfigServer());
              recvChunkStartBuilder.append("secondaryThrottle", isSecondaryThrottle);
  
              // Follow the same convention in moveChunk.
              if (isSecondaryThrottle && !secThrottleObj.isEmpty()) {
                  recvChunkStartBuilder.append("writeConcern", secThrottleObj);
              }
  
              try {
                  ok = connTo->runCommand("admin", recvChunkStartBuilder.done(), res);
              } catch (DBException& e) {
                  errmsg = str::stream() << "moveChunk could not contact to: shard " << toShardName
                                         << " to start transfer" << causedBy(e);
                  warning() << errmsg;
                  return false;
              }
  
              connTo.done();
  
              if (!ok) {
                  errmsg = "moveChunk failed to engage TO-shard in the data transfer: ";
                  verify(res["errmsg"].type());
                  errmsg += res["errmsg"].String();
                  result.append("cause", res);
                  warning() << errmsg;
                  return false;
              }
          }
          timing.done(3);
          MONGO_FP_PAUSE_WHILE(moveChunkHangAtStep3);
  
          // 4.
  
          // Track last result from TO shard for sanity check
          BSONObj res;
          for (int i = 0; i < 86400; i++) {  // don't want a single chunk move to take more than a day
              invariant(!txn->lockState()->isLocked());
  
              // Exponential sleep backoff, up to 1024ms. Don't sleep much on the first few
              // iterations, since we want empty chunk migrations to be fast.
              sleepmillis(1 << std::min(i, 10));
  
              ScopedDbConnection conn(toShardCS);
              bool ok;
              res = BSONObj();
              try {
                  ok = conn->runCommand("admin", BSON("_recvChunkStatus" << 1), res);
                  res = res.getOwned();
              } catch (DBException& e) {
                  errmsg = str::stream() << "moveChunk could not contact to: shard " << toShardName
                                         << " to monitor transfer" << causedBy(e);
                  warning() << errmsg;
                  return false;
              }
  
              conn.done();
  
              if (res["ns"].str() != ns || res["from"].str() != fromShardCS.toString() ||
                  !res["min"].isABSONObj() || res["min"].Obj().woCompare(min) != 0 ||
                  !res["max"].isABSONObj() || res["max"].Obj().woCompare(max) != 0) {
                  // This can happen when the destination aborted the migration and
                  // received another recvChunk before this thread sees the transition
                  // to the abort state. This is currently possible only if multiple migrations
                  // are happening at once. This is an unfortunate consequence of the shards not
                  // being able to keep track of multiple incoming and outgoing migrations.
                  errmsg = str::stream() << "Destination shard aborted migration, "
                                            "now running a new one: " << res;
                  warning() << errmsg;
                  return false;
              }
  
              LOG(0) << "moveChunk data transfer progress: " << res
                     << " my mem used: " << migrateFromStatus.mbUsed() << migrateLog;
  
              if (!ok || res["state"].String() == "fail") {
                  warning() << "moveChunk error transferring data caused migration abort: " << res
                            << migrateLog;
                  errmsg = "data transfer error";
                  result.append("cause", res);
                  return false;
              }
  
              if (res["state"].String() == "steady")
                  break;
  
              if (migrateFromStatus.mbUsed() > (500 * 1024 * 1024)) {
                  // This is too much memory for us to use for this so we're going to abort
                  // the migrate
                  ScopedDbConnection conn(toShardCS);
  
                  BSONObj res;
                  if (!conn->runCommand("admin", BSON("_recvChunkAbort" << 1), res)) {
                      warning() << "Error encountered while trying to abort migration on "
                                << "destination shard" << toShardCS;
                  }
  
                  res = res.getOwned();
                  conn.done();
                  error() << "aborting migrate because too much memory used res: " << res
                          << migrateLog;
                  errmsg = "aborting migrate because too much memory used";
                  result.appendBool("split", true);
                  return false;
              }
  
              txn->checkForInterrupt();
          }
  
          timing.done(4);
          MONGO_FP_PAUSE_WHILE(moveChunkHangAtStep4);
  
          // 5.
  
          // Before we get into the critical section of the migration, let's double check
          // that the docs have been cloned, the config servers are reachable,
          // and the lock is in place.
          log() << "About to check if it is safe to enter critical section";
  
          // Ensure all cloned docs have actually been transferred
          std::size_t locsRemaining = migrateFromStatus.cloneLocsRemaining();
          if (locsRemaining != 0) {
              errmsg = str::stream() << "moveChunk cannot enter critical section before all data is"
                                     << " cloned, " << locsRemaining << " locs were not transferred"
                                     << " but to-shard reported " << res;
  
              // Should never happen, but safe to abort before critical section
              error() << errmsg << migrateLog;
              dassert(false);
              return false;
          }
  
          // Ensure distributed lock still held
          Status lockStatus = scopedDistLock.getValue().checkStatus();
          if (!lockStatus.isOK()) {
              errmsg = str::stream() << "not entering migrate critical section because "
                                     << lockStatus.toString();
              warning() << errmsg;
              return false;
          }
  
          log() << "About to enter migrate critical section";
  
          {
              // 5.a
              // we're under the collection lock here, so no other migrate can change maxVersion
              // or CollectionMetadata state
              migrateFromStatus.setInCriticalSection(true);
              ChunkVersion myVersion = origCollVersion;
              myVersion.incMajor();
  
              {
                  ScopedTransaction transaction(txn, MODE_IX);
                  Lock::DBLock lk(txn->lockState(), nsToDatabaseSubstring(ns), MODE_IX);
                  Lock::CollectionLock collLock(txn->lockState(), ns, MODE_X);
                  verify(myVersion > shardingState.getVersion(ns));
  
                  // bump the metadata's version up and "forget" about the chunk being moved
                  // this is not the commit point but in practice the state in this shard won't
                  // until the commit it done
                  shardingState.donateChunk(txn, ns, min, max, myVersion);
              }
  
              log() << "moveChunk setting version to: " << myVersion << migrateLog;
  
              // 5.b
              // we're under the collection lock here, too, so we can undo the chunk donation because
              // no other state change could be ongoing
  
              BSONObj res;
              bool ok;
  
              try {
                  ScopedDbConnection connTo(toShardCS, 35.0);
                  ok = connTo->runCommand("admin", BSON("_recvChunkCommit" << 1), res);
                  connTo.done();
              } catch (DBException& e) {
                  errmsg = str::stream() << "moveChunk could not contact to: shard "
                                         << toShardCS.toString() << " to commit transfer"
                                         << causedBy(e);
                  warning() << errmsg;
                  ok = false;
              }
  
              if (!ok || MONGO_FAIL_POINT(failMigrationCommit)) {
                  log() << "moveChunk migrate commit not accepted by TO-shard: " << res
                        << " resetting shard version to: " << origShardVersion << migrateLog;
                  {
                      ScopedTransaction transaction(txn, MODE_IX);
                      Lock::DBLock dbLock(txn->lockState(), nsToDatabaseSubstring(ns), MODE_IX);
                      Lock::CollectionLock collLock(txn->lockState(), ns, MODE_X);
  
                      log() << "moveChunk collection lock acquired to reset shard version from "
                               "failed migration";
  
                      // revert the chunk manager back to the state before "forgetting" about the
                      // chunk
                      shardingState.undoDonateChunk(txn, ns, origCollMetadata);
                  }
                  log() << "Shard version successfully reset to clean up failed migration";
  
                  errmsg = "_recvChunkCommit failed!";
                  result.append("cause", res);
                  return false;
              }
  
              log() << "moveChunk migrate commit accepted by TO-shard: " << res << migrateLog;
  
              // 5.c
  
              // version at which the next highest lastmod will be set
              // if the chunk being moved is the last in the shard, nextVersion is that chunk's
              // lastmod otherwise the highest version is from the chunk being bumped on the
              // FROM-shard
              ChunkVersion nextVersion;
  
              // we want to go only once to the configDB but perhaps change two chunks, the one being
              // migrated and another local one (so to bump version for the entire shard)
              // we use the 'applyOps' mechanism to group the two updates and make them safer
              // TODO pull config update code to a module
  
              BSONArrayBuilder updates;
              {
                  // update for the chunk being moved
                  BSONObjBuilder op;
                  op.append("op", "u");
                  op.appendBool("b", false /* no upserting */);
                  op.append("ns", ChunkType::ConfigNS);
  
                  BSONObjBuilder n(op.subobjStart("o"));
                  n.append(ChunkType::name(), Chunk::genID(ns, min));
                  myVersion.addToBSON(n, ChunkType::DEPRECATED_lastmod());
                  n.append(ChunkType::ns(), ns);
                  n.append(ChunkType::min(), min);
                  n.append(ChunkType::max(), max);
                  n.append(ChunkType::shard(), toShardName);
                  n.done();
  
                  BSONObjBuilder q(op.subobjStart("o2"));
                  q.append(ChunkType::name(), Chunk::genID(ns, min));
                  q.done();
  
                  updates.append(op.obj());
              }
  
              nextVersion = myVersion;
  
              // if we have chunks left on the FROM shard, update the version of one of them as
              // well.  we can figure that out by grabbing the metadata installed on 5.a
  
              const std::shared_ptr<CollectionMetadata> bumpedCollMetadata(
                  shardingState.getCollectionMetadata(ns));
              if (bumpedCollMetadata->getNumChunks() > 0) {
                  // get another chunk on that shard
                  ChunkType bumpChunk;
                  bool chunkRes =
                      bumpedCollMetadata->getNextChunk(bumpedCollMetadata->getMinKey(), &bumpChunk);
                  BSONObj bumpMin = bumpChunk.getMin();
                  BSONObj bumpMax = bumpChunk.getMax();
  
                  (void)chunkRes;  // for compile warning on non-debug
                  dassert(chunkRes);
                  dassert(bumpMin.woCompare(min) != 0);
  
                  BSONObjBuilder op;
                  op.append("op", "u");
                  op.appendBool("b", false);
                  op.append("ns", ChunkType::ConfigNS);
  
                  nextVersion.incMinor();  // same as used on donateChunk
                  BSONObjBuilder n(op.subobjStart("o"));
                  n.append(ChunkType::name(), Chunk::genID(ns, bumpMin));
                  nextVersion.addToBSON(n, ChunkType::DEPRECATED_lastmod());
                  n.append(ChunkType::ns(), ns);
                  n.append(ChunkType::min(), bumpMin);
                  n.append(ChunkType::max(), bumpMax);
                  n.append(ChunkType::shard(), fromShardName);
                  n.done();
  
                  BSONObjBuilder q(op.subobjStart("o2"));
                  q.append(ChunkType::name(), Chunk::genID(ns, bumpMin));
                  q.done();
  
                  updates.append(op.obj());
  
                  log() << "moveChunk updating self version to: " << nextVersion << " through "
                        << bumpMin << " -> " << bumpMax << " for collection '" << ns << "'"
                        << migrateLog;
  
              } else {
                  log() << "moveChunk moved last chunk out for collection '" << ns << "'"
                        << migrateLog;
              }
  
              BSONArrayBuilder preCond;
              {
                  BSONObjBuilder b;
                  b.append("ns", ChunkType::ConfigNS);
                  b.append("q",
                           BSON("query" << BSON(ChunkType::ns(ns)) << "orderby"
                                        << BSON(ChunkType::DEPRECATED_lastmod() << -1)));
                  {
                      BSONObjBuilder bb(b.subobjStart("res"));
                      // TODO: For backwards compatibility, we can't yet require an epoch here
                      bb.appendTimestamp(ChunkType::DEPRECATED_lastmod(), origCollVersion.toLong());
                      bb.done();
                  }
                  preCond.append(b.obj());
              }
  
              Status applyOpsStatus{Status::OK()};
              try {
                  // For testing migration failures
                  if (MONGO_FAIL_POINT(failMigrationConfigWritePrepare)) {
                      throw DBException("mock migration failure before config write",
                                        PrepareConfigsFailedCode);
                  }
  
                  applyOpsStatus =
                      grid.catalogManager()->applyChunkOpsDeprecated(updates.arr(), preCond.arr());
  
                  if (MONGO_FAIL_POINT(failMigrationApplyOps)) {
                      throw SocketException(SocketException::RECV_ERROR,
                                            shardingState.getConfigServer());
                  }
              } catch (const DBException& ex) {
                  warning() << ex << migrateLog;
                  applyOpsStatus = ex.toStatus();
              }
  
              if (applyOpsStatus == ErrorCodes::PrepareConfigsFailedCode) {
                  // In the process of issuing the migrate commit, the SyncClusterConnection
                  // checks that the config servers are reachable. If they are not, we are
                  // sure that the applyOps command was not sent to any of the configs, so we
                  // can safely back out of the migration here, by resetting the shard
                  // version that we bumped up to in the donateChunk() call above.
  
                  log() << "About to acquire moveChunk coll lock to reset shard version from "
                        << "failed migration";
  
                  {
                      ScopedTransaction transaction(txn, MODE_IX);
                      Lock::DBLock dbLock(txn->lockState(), nsToDatabaseSubstring(ns), MODE_IX);
                      Lock::CollectionLock collLock(txn->lockState(), ns, MODE_X);
  
                      // Revert the metadata back to the state before "forgetting"
                      // about the chunk.
                      shardingState.undoDonateChunk(txn, ns, origCollMetadata);
                  }
  
                  log() << "Shard version successfully reset to clean up failed migration";
  
                  errmsg = "Failed to send migrate commit to configs because " + errmsg;
                  return false;
  
              } else if (!applyOpsStatus.isOK()) {
                  // this could be a blip in the connectivity
                  // wait out a few seconds and check if the commit request made it
                  //
                  // if the commit made it to the config, we'll see the chunk in the new shard and
                  // there's no action
                  // if the commit did not make it, currently the only way to fix this state is to
                  // bounce the mongod so that the old state (before migrating) be brought in
  
                  warning() << "moveChunk commit outcome ongoing" << migrateLog;
                  sleepsecs(10);
  
                  // look for the chunk in this shard whose version got bumped
                  // we assume that if that mod made it to the config, the applyOps was successful
                  try {
                      std::vector<ChunkType> newestChunk;
                      Status status = grid.catalogManager()->getChunks(
                          BSON(ChunkType::ns(ns)),
                          BSON(ChunkType::DEPRECATED_lastmod() << -1),
                          1,
                          &newestChunk);
                      uassertStatusOK(status);
  
                      ChunkVersion checkVersion;
                      if (!newestChunk.empty()) {
                          invariant(newestChunk.size() == 1);
                          checkVersion = newestChunk[0].getVersion();
                      }
  
                      if (checkVersion.equals(nextVersion)) {
                          log() << "moveChunk commit confirmed" << migrateLog;
                          errmsg.clear();
  
                      } else {
                          error() << "moveChunk commit failed: version is at " << checkVersion
                                  << " instead of " << nextVersion << migrateLog;
                          error() << "TERMINATING" << migrateLog;
                          dbexit(EXIT_SHARDING_ERROR);
                      }
                  } catch (...) {
                      error() << "moveChunk failed to get confirmation of commit" << migrateLog;
                      error() << "TERMINATING" << migrateLog;
                      dbexit(EXIT_SHARDING_ERROR);
                  }
              }
  
              migrateFromStatus.setInCriticalSection(false);
  
              // 5.d
              BSONObjBuilder commitInfo;
              commitInfo.appendElements(chunkInfo);
              if (res["counts"].type() == Object) {
                  commitInfo.appendElements(res["counts"].Obj());
              }
  
              grid.catalogManager()->logChange(
                  txn->getClient()->clientAddress(true), "moveChunk.commit", ns, commitInfo.obj());
          }
  
          migrateFromStatus.done(txn);
          timing.done(5);
          MONGO_FP_PAUSE_WHILE(moveChunkHangAtStep5);
  
          // 6.
          // NOTE: It is important that the distributed collection lock be held for this step.
          RangeDeleter* deleter = getDeleter();
          RangeDeleterOptions deleterOptions(
              KeyRange(ns, min.getOwned(), max.getOwned(), shardKeyPattern));
          deleterOptions.writeConcern = writeConcern;
          deleterOptions.waitForOpenCursors = true;
          deleterOptions.fromMigrate = true;
          deleterOptions.onlyRemoveOrphanedDocs = true;
          deleterOptions.removeSaverReason = "post-cleanup";
  
          if (waitForDelete) {
              log() << "doing delete inline for cleanup of chunk data" << migrateLog;
  
              string errMsg;
              // This is an immediate delete, and as a consequence, there could be more
              // deletes happening simultaneously than there are deleter worker threads.
              if (!deleter->deleteNow(txn, deleterOptions, &errMsg)) {
                  log() << "Error occured while performing cleanup: " << errMsg;
              }
          } else {
              log() << "forking for cleanup of chunk data" << migrateLog;
  
              string errMsg;
              if (!deleter->queueDelete(txn,
                                        deleterOptions,
                                        NULL,  // Don't want to be notified.
                                        &errMsg)) {
                  log() << "could not queue migration cleanup: " << errMsg;
              }
          }
          timing.done(6);
          MONGO_FP_PAUSE_WHILE(moveChunkHangAtStep6);
  
          return true;
      }
  
  } moveChunkCmd;
  
  bool ShardingState::inCriticalMigrateSection() {
      return migrateFromStatus.getInCriticalSection();
  }
  
  bool ShardingState::waitTillNotInCriticalSection(int maxSecondsToWait) {
      return migrateFromStatus.waitTillNotInCriticalSection(maxSecondsToWait);
  }
  
  /* -----
     below this are the "to" side commands
  
     command to initiate
     worker thread
       does initial clone
       pulls initial change set
       keeps pulling
       keeps state
     command to get state
     commend to "commit"
  */
  
  // Enabling / disabling these fail points pauses / resumes MigrateStatus::_go(), the thread
  // that receives a chunk migration from the donor.
  MONGO_FP_DECLARE(migrateThreadHangAtStep1);
  MONGO_FP_DECLARE(migrateThreadHangAtStep2);
  MONGO_FP_DECLARE(migrateThreadHangAtStep3);
  MONGO_FP_DECLARE(migrateThreadHangAtStep4);
  MONGO_FP_DECLARE(migrateThreadHangAtStep5);
  
  class MigrateStatus {
  public:
      enum State { READY, CLONE, CATCHUP, STEADY, COMMIT_START, DONE, FAIL, ABORT };
  
      MigrateStatus()
          : _active(false),
            _numCloned(0),
            _clonedBytes(0),
            _numCatchup(0),
            _numSteady(0),
            _state(READY) {}
  
      void setState(State newState) {
          stdx::lock_guard<stdx::mutex> sl(_mutex);
          _state = newState;
      }
  
      State getState() const {
          stdx::lock_guard<stdx::mutex> sl(_mutex);
          return _state;
      }
  
      /**
       * Returns OK if preparation was successful.
       */
      Status prepare(const std::string& ns,
                     const std::string& fromShard,
                     const BSONObj& min,
                     const BSONObj& max,
                     const BSONObj& shardKeyPattern) {
          stdx::lock_guard<stdx::mutex> lk(_mutex);
  
          if (_active) {
              return Status(ErrorCodes::ConflictingOperationInProgress,
                            str::stream() << "Active migration already in progress "
                                          << "ns: " << _ns << ", from: " << _from << ", min: " << _min
                                          << ", max: " << _max);
          }
  
          _state = READY;
          _errmsg = "";
  
          _ns = ns;
          _from = fromShard;
          _min = min;
          _max = max;
          _shardKeyPattern = shardKeyPattern;
  
          _numCloned = 0;
          _clonedBytes = 0;
          _numCatchup = 0;
          _numSteady = 0;
  
          _active = true;
  
          return Status::OK();
      }
  
      void go(OperationContext* txn,
              const std::string& ns,
              BSONObj min,
              BSONObj max,
              BSONObj shardKeyPattern,
              const std::string& fromShard,
              const OID& epoch,
              const WriteConcernOptions& writeConcern) {
          try {
              _go(txn, ns, min, max, shardKeyPattern, fromShard, epoch, writeConcern);
          } catch (std::exception& e) {
              {
                  stdx::lock_guard<stdx::mutex> sl(_mutex);
                  _state = FAIL;
                  _errmsg = e.what();
              }
  
              error() << "migrate failed: " << e.what() << migrateLog;
          } catch (...) {
              {
                  stdx::lock_guard<stdx::mutex> sl(_mutex);
                  _state = FAIL;
                  _errmsg = "UNKNOWN ERROR";
              }
  
              error() << "migrate failed with unknown exception" << migrateLog;
          }
  
          if (getState() != DONE) {
              // Unprotect the range if needed/possible on unsuccessful TO migration
              ScopedTransaction transaction(txn, MODE_IX);
              Lock::DBLock dbLock(txn->lockState(), nsToDatabaseSubstring(ns), MODE_IX);
              Lock::CollectionLock collLock(txn->lockState(), ns, MODE_X);
  
              string errMsg;
              if (!shardingState.forgetPending(txn, ns, min, max, epoch, &errMsg)) {
                  warning() << errMsg;
              }
          }
  
          setActive(false);
      }
  
      void _go(OperationContext* txn,
               const std::string& ns,
               BSONObj min,
               BSONObj max,
               BSONObj shardKeyPattern,
               const std::string& fromShard,
               const OID& epoch,
               const WriteConcernOptions& writeConcern) {
          verify(getActive());
          verify(getState() == READY);
          verify(!min.isEmpty());
          verify(!max.isEmpty());
  
          DisableDocumentValidation validationDisabler(txn);
  
          log() << "starting receiving-end of migration of chunk " << min << " -> " << max
                << " for collection " << ns << " from " << fromShard << " at epoch "
                << epoch.toString();
  
          string errmsg;
          MoveTimingHelper timing(txn, "to", ns, min, max, 5 /* steps */, &errmsg, "", "");
  
          ScopedDbConnection conn(fromShard);
          conn->getLastError();  // just test connection
  
          NamespaceString nss(ns);
          {
              // 0. copy system.namespaces entry if collection doesn't already exist
              OldClientWriteContext ctx(txn, ns);
              if (!repl::getGlobalReplicationCoordinator()->canAcceptWritesFor(nss)) {
                  errmsg = str::stream() << "Not primary during migration: " << ns
                                         << ": checking if collection exists";
                  warning() << errmsg;
                  setState(FAIL);
                  return;
              }
  
              // Only copy if ns doesn't already exist
              Database* db = ctx.db();
              Collection* collection = db->getCollection(ns);
  
              if (!collection) {
                  list<BSONObj> infos = conn->getCollectionInfos(
                      nsToDatabase(ns), BSON("name" << nsToCollectionSubstring(ns)));
  
                  BSONObj options;
                  if (infos.size() > 0) {
                      BSONObj entry = infos.front();
                      if (entry["options"].isABSONObj()) {
                          options = entry["options"].Obj();
                      }
                  }
  
                  WriteUnitOfWork wuow(txn);
                  Status status = userCreateNS(txn, db, ns, options, false);
                  if (!status.isOK()) {
                      warning() << "failed to create collection [" << ns << "] "
                                << " with options " << options << ": " << status;
                  }
                  wuow.commit();
              }
          }
  
          {
              // 1. copy indexes
  
              vector<BSONObj> indexSpecs;
              {
                  const std::list<BSONObj> indexes = conn->getIndexSpecs(ns);
                  indexSpecs.insert(indexSpecs.begin(), indexes.begin(), indexes.end());
              }
  
              ScopedTransaction transaction(txn, MODE_IX);
              Lock::DBLock lk(txn->lockState(), nsToDatabaseSubstring(ns), MODE_X);
              OldClientContext ctx(txn, ns);
              if (!repl::getGlobalReplicationCoordinator()->canAcceptWritesFor(nss)) {
                  errmsg = str::stream() << "Not primary during migration: " << ns;
                  warning() << errmsg;
                  setState(FAIL);
                  return;
              }
  
              Database* db = ctx.db();
              Collection* collection = db->getCollection(ns);
              if (!collection) {
                  errmsg = str::stream() << "collection dropped during migration: " << ns;
                  warning() << errmsg;
                  setState(FAIL);
                  return;
              }
  
              MultiIndexBlock indexer(txn, collection);
  
              indexer.removeExistingIndexes(&indexSpecs);
  
              if (!indexSpecs.empty()) {
                  // Only copy indexes if the collection does not have any documents.
                  if (collection->numRecords(txn) > 0) {
                      errmsg = str::stream() << "aborting migration, shard is missing "
                                             << indexSpecs.size() << " indexes and "
                                             << "collection is not empty. Non-trivial "
                                             << "index creation should be scheduled manually";
                      warning() << errmsg;
                      setState(FAIL);
                      return;
                  }
  
                  Status status = indexer.init(indexSpecs);
                  if (!status.isOK()) {
                      errmsg = str::stream() << "failed to create index before migrating data. "
                                             << " error: " << status.toString();
                      warning() << errmsg;
                      setState(FAIL);
                      return;
                  }
  
                  status = indexer.insertAllDocumentsInCollection();
                  if (!status.isOK()) {
                      errmsg = str::stream() << "failed to create index before migrating data. "
                                             << " error: " << status.toString();
                      warning() << errmsg;
                      setState(FAIL);
                      return;
                  }
  
                  WriteUnitOfWork wunit(txn);
                  indexer.commit();
  
                  for (size_t i = 0; i < indexSpecs.size(); i++) {
                      // make sure to create index on secondaries as well
                      getGlobalServiceContext()->getOpObserver()->onCreateIndex(
                          txn, db->getSystemIndexesName(), indexSpecs[i], true /* fromMigrate */);
                  }
  
                  wunit.commit();
              }
  
              timing.done(1);
              MONGO_FP_PAUSE_WHILE(migrateThreadHangAtStep1);
          }
  
          {
              // 2. delete any data already in range
              RangeDeleterOptions deleterOptions(
                  KeyRange(ns, min.getOwned(), max.getOwned(), shardKeyPattern));
              deleterOptions.writeConcern = writeConcern;
              // No need to wait since all existing cursors will filter out this range when
              // returning the results.
              deleterOptions.waitForOpenCursors = false;
              deleterOptions.fromMigrate = true;
              deleterOptions.onlyRemoveOrphanedDocs = true;
              deleterOptions.removeSaverReason = "preCleanup";
  
              string errMsg;
  
              if (!getDeleter()->deleteNow(txn, deleterOptions, &errMsg)) {
                  warning() << "Failed to queue delete for migrate abort: " << errMsg;
                  setState(FAIL);
                  return;
              }
  
              {
                  // Protect the range by noting that we're now starting a migration to it
                  ScopedTransaction transaction(txn, MODE_IX);
                  Lock::DBLock dbLock(txn->lockState(), nsToDatabaseSubstring(ns), MODE_IX);
                  Lock::CollectionLock collLock(txn->lockState(), ns, MODE_X);
  
                  if (!shardingState.notePending(txn, ns, min, max, epoch, &errmsg)) {
                      warning() << errmsg;
                      setState(FAIL);
                      return;
                  }
              }
  
              timing.done(2);
              MONGO_FP_PAUSE_WHILE(migrateThreadHangAtStep2);
          }
  
          State currentState = getState();
          if (currentState == FAIL || currentState == ABORT) {
              string errMsg;
              RangeDeleterOptions deleterOptions(
                  KeyRange(ns, min.getOwned(), max.getOwned(), shardKeyPattern));
              deleterOptions.writeConcern = writeConcern;
              // No need to wait since all existing cursors will filter out this range when
              // returning the results.
              deleterOptions.waitForOpenCursors = false;
              deleterOptions.fromMigrate = true;
              deleterOptions.onlyRemoveOrphanedDocs = true;
  
              if (!getDeleter()->queueDelete(txn, deleterOptions, NULL /* notifier */, &errMsg)) {
                  warning() << "Failed to queue delete for migrate abort: " << errMsg;
              }
          }
  
          {
              // 3. initial bulk clone
              setState(CLONE);
  
              while (true) {
                  BSONObj res;
                  if (!conn->runCommand("admin",
                                        BSON("_migrateClone" << 1),
                                        res)) {  // gets array of objects to copy, in disk order
                      setState(FAIL);
                      errmsg = "_migrateClone failed: ";
                      errmsg += res.toString();
                      error() << errmsg << migrateLog;
                      conn.done();
                      return;
                  }
  
                  BSONObj arr = res["objects"].Obj();
                  int thisTime = 0;
  
                  BSONObjIterator i(arr);
                  while (i.more()) {
                      txn->checkForInterrupt();
  
                      if (getState() == ABORT) {
                          errmsg = str::stream() << "Migration abort requested while "
                                                 << "copying documents";
                          error() << errmsg << migrateLog;
                          return;
                      }
  
                      BSONObj docToClone = i.next().Obj();
                      {
                          OldClientWriteContext cx(txn, ns);
  
                          BSONObj localDoc;
                          if (willOverrideLocalId(txn,
                                                  ns,
                                                  min,
                                                  max,
                                                  shardKeyPattern,
                                                  cx.db(),
                                                  docToClone,
                                                  &localDoc)) {
                              string errMsg = str::stream() << "cannot migrate chunk, local document "
                                                            << localDoc << " has same _id as cloned "
                                                            << "remote document " << docToClone;
  
                              warning() << errMsg;
  
                              // Exception will abort migration cleanly
                              uasserted(16976, errMsg);
                          }
  
                          Helpers::upsert(txn, ns, docToClone, true);
                      }
                      thisTime++;
  
                      {
                          stdx::lock_guard<stdx::mutex> statsLock(_mutex);
                          _numCloned++;
                          _clonedBytes += docToClone.objsize();
                      }
  
                      if (writeConcern.shouldWaitForOtherNodes()) {
                          repl::ReplicationCoordinator::StatusAndDuration replStatus =
                              repl::getGlobalReplicationCoordinator()->awaitReplication(
                                  txn,
                                  repl::ReplClientInfo::forClient(txn->getClient()).getLastOp(),
                                  writeConcern);
                          if (replStatus.status.code() == ErrorCodes::WriteConcernFailed) {
                              warning() << "secondaryThrottle on, but doc insert timed out; "
                                           "continuing";
                          } else {
                              massertStatusOK(replStatus.status);
                          }
                      }
                  }
  
                  if (thisTime == 0)
                      break;
              }
  
              timing.done(3);
              MONGO_FP_PAUSE_WHILE(migrateThreadHangAtStep3);
          }
  
          // If running on a replicated system, we'll need to flush the docs we cloned to the
          // secondaries
          repl::OpTime lastOpApplied = repl::ReplClientInfo::forClient(txn->getClient()).getLastOp();
  
          {
              // 4. do bulk of mods
              setState(CATCHUP);
              while (true) {
                  BSONObj res;
                  if (!conn->runCommand("admin", BSON("_transferMods" << 1), res)) {
                      setState(FAIL);
                      errmsg = "_transferMods failed: ";
                      errmsg += res.toString();
                      error() << "_transferMods failed: " << res << migrateLog;
                      conn.done();
                      return;
                  }
                  if (res["size"].number() == 0)
                      break;
  
                  apply(txn, ns, min, max, shardKeyPattern, res, &lastOpApplied);
  
                  const int maxIterations = 3600 * 50;
                  int i;
                  for (i = 0; i < maxIterations; i++) {
                      txn->checkForInterrupt();
  
                      if (getState() == ABORT) {
                          errmsg = str::stream() << "Migration abort requested while waiting "
                                                 << "for replication at catch up stage";
                          error() << errmsg << migrateLog;
  
                          return;
                      }
  
                      if (opReplicatedEnough(txn, lastOpApplied, writeConcern))
                          break;
  
                      if (i > 100) {
                          warning() << "secondaries having hard time keeping up with migrate"
                                    << migrateLog;
                      }
  
                      sleepmillis(20);
                  }
  
                  if (i == maxIterations) {
                      errmsg = "secondary can't keep up with migrate";
                      error() << errmsg << migrateLog;
                      conn.done();
                      setState(FAIL);
                      return;
                  }
              }
  
              timing.done(4);
              MONGO_FP_PAUSE_WHILE(migrateThreadHangAtStep4);
          }
  
          {
              // pause to wait for replication
              // this will prevent us from going into critical section until we're ready
              Timer t;
              while (t.minutes() < 600) {
                  txn->checkForInterrupt();
  
                  if (getState() == ABORT) {
                      errmsg = "Migration abort requested while waiting for replication";
                      error() << errmsg << migrateLog;
                      return;
                  }
  
                  log() << "Waiting for replication to catch up before entering critical section";
                  if (flushPendingWrites(txn, ns, min, max, lastOpApplied, writeConcern))
                      break;
                  sleepsecs(1);
              }
  
              if (t.minutes() >= 600) {
                  setState(FAIL);
                  errmsg = "Cannot go to critical section because secondaries cannot keep up";
                  error() << errmsg << migrateLog;
                  return;
              }
          }
  
          {
              // 5. wait for commit
  
              setState(STEADY);
              bool transferAfterCommit = false;
              while (getState() == STEADY || getState() == COMMIT_START) {
                  txn->checkForInterrupt();
  
                  // Make sure we do at least one transfer after recv'ing the commit message
                  // If we aren't sure that at least one transfer happens *after* our state
                  // changes to COMMIT_START, there could be mods still on the FROM shard that
                  // got logged *after* our _transferMods but *before* the critical section.
                  if (getState() == COMMIT_START)
                      transferAfterCommit = true;
  
                  BSONObj res;
                  if (!conn->runCommand("admin", BSON("_transferMods" << 1), res)) {
                      log() << "_transferMods failed in STEADY state: " << res << migrateLog;
                      errmsg = res.toString();
                      setState(FAIL);
                      conn.done();
                      return;
                  }
  
                  if (res["size"].number() > 0 &&
                      apply(txn, ns, min, max, shardKeyPattern, res, &lastOpApplied)) {
                      continue;
                  }
  
                  if (getState() == ABORT) {
                      return;
                  }
  
                  // We know we're finished when:
                  // 1) The from side has told us that it has locked writes (COMMIT_START)
                  // 2) We've checked at least one more time for un-transmitted mods
                  if (getState() == COMMIT_START && transferAfterCommit == true) {
                      if (flushPendingWrites(txn, ns, min, max, lastOpApplied, writeConcern))
                          break;
                  }
  
                  // Only sleep if we aren't committing
                  if (getState() == STEADY)
                      sleepmillis(10);
              }
  
              if (getState() == FAIL) {
                  errmsg = "timed out waiting for commit";
                  return;
              }
  
              timing.done(5);
              MONGO_FP_PAUSE_WHILE(migrateThreadHangAtStep5);
          }
  
          setState(DONE);
          conn.done();
      }
  
      void status(BSONObjBuilder& b) {
          stdx::lock_guard<stdx::mutex> sl(_mutex);
  
          b.appendBool("active", _active);
  
          b.append("ns", _ns);
          b.append("from", _from);
          b.append("min", _min);
          b.append("max", _max);
          b.append("shardKeyPattern", _shardKeyPattern);
  
          b.append("state", stateToString(_state));
          if (_state == FAIL) {
              b.append("errmsg", _errmsg);
          }
  
          BSONObjBuilder bb(b.subobjStart("counts"));
          bb.append("cloned", _numCloned);
          bb.append("clonedBytes", _clonedBytes);
          bb.append("catchup", _numCatchup);
          bb.append("steady", _numSteady);
          bb.done();
      }
  
      bool apply(OperationContext* txn,
                 const string& ns,
                 BSONObj min,
                 BSONObj max,
                 BSONObj shardKeyPattern,
                 const BSONObj& xfer,
                 repl::OpTime* lastOpApplied) {
          repl::OpTime dummy;
          if (lastOpApplied == NULL) {
              lastOpApplied = &dummy;
          }
  
          bool didAnything = false;
  
          if (xfer["deleted"].isABSONObj()) {
              ScopedTransaction transaction(txn, MODE_IX);
              Lock::DBLock dlk(txn->lockState(), nsToDatabaseSubstring(ns), MODE_IX);
              Helpers::RemoveSaver rs("moveChunk", ns, "removedDuring");
  
              BSONObjIterator i(xfer["deleted"].Obj());
              while (i.more()) {
                  Lock::CollectionLock clk(txn->lockState(), ns, MODE_X);
                  OldClientContext ctx(txn, ns);
  
                  BSONObj id = i.next().Obj();
  
                  // do not apply deletes if they do not belong to the chunk being migrated
                  BSONObj fullObj;
                  if (Helpers::findById(txn, ctx.db(), ns.c_str(), id, fullObj)) {
                      if (!isInRange(fullObj, min, max, shardKeyPattern)) {
                          log() << "not applying out of range deletion: " << fullObj << migrateLog;
  
                          continue;
                      }
                  }
  
                  if (serverGlobalParams.moveParanoia) {
                      rs.goingToDelete(fullObj);
                  }
  
                  deleteObjects(txn,
                                ctx.db(),
                                ns,
                                id,
                                PlanExecutor::YIELD_MANUAL,
                                true /* justOne */,
                                false /* god */,
                                true /* fromMigrate */);
  
                  *lastOpApplied = repl::ReplClientInfo::forClient(txn->getClient()).getLastOp();
                  didAnything = true;
              }
          }
  
          if (xfer["reload"].isABSONObj()) {
              BSONObjIterator i(xfer["reload"].Obj());
              while (i.more()) {
                  OldClientWriteContext cx(txn, ns);
  
                  BSONObj updatedDoc = i.next().Obj();
  
                  BSONObj localDoc;
                  if (willOverrideLocalId(
                          txn, ns, min, max, shardKeyPattern, cx.db(), updatedDoc, &localDoc)) {
                      string errMsg = str::stream()
                          << "cannot migrate chunk, local document " << localDoc
                          << " has same _id as reloaded remote document " << updatedDoc;
  
                      warning() << errMsg;
  
                      // Exception will abort migration cleanly
                      uasserted(16977, errMsg);
                  }
  
                  // We are in write lock here, so sure we aren't killing
                  Helpers::upsert(txn, ns, updatedDoc, true);
  
                  *lastOpApplied = repl::ReplClientInfo::forClient(txn->getClient()).getLastOp();
                  didAnything = true;
              }
          }
  
          return didAnything;
      }
  
      /**
       * Checks if an upsert of a remote document will override a local document with the same _id
       * but in a different range on this shard.
       * Must be in WriteContext to avoid races and DBHelper errors.
       * TODO: Could optimize this check out if sharding on _id.
       */
      bool willOverrideLocalId(OperationContext* txn,
                               const string& ns,
                               BSONObj min,
                               BSONObj max,
                               BSONObj shardKeyPattern,
                               Database* db,
                               BSONObj remoteDoc,
                               BSONObj* localDoc) {
          *localDoc = BSONObj();
          if (Helpers::findById(txn, db, ns.c_str(), remoteDoc, *localDoc)) {
              return !isInRange(*localDoc, min, max, shardKeyPattern);
          }
  
          return false;
      }
  
      /**
       * Returns true if the majority of the nodes and the nodes corresponding to the given
       * writeConcern (if not empty) have applied till the specified lastOp.
       */
      bool opReplicatedEnough(OperationContext* txn,
                              const repl::OpTime& lastOpApplied,
                              const WriteConcernOptions& writeConcern) {
          WriteConcernOptions majorityWriteConcern;
          majorityWriteConcern.wTimeout = -1;
          majorityWriteConcern.wMode = WriteConcernOptions::kMajority;
          Status majorityStatus = repl::getGlobalReplicationCoordinator()
                                      ->awaitReplication(txn, lastOpApplied, majorityWriteConcern)
                                      .status;
  
          if (!writeConcern.shouldWaitForOtherNodes()) {
              return majorityStatus.isOK();
          }
  
          // Also enforce the user specified write concern after "majority" so it covers
          // the union of the 2 write concerns.
  
          WriteConcernOptions userWriteConcern(writeConcern);
          userWriteConcern.wTimeout = -1;
          Status userStatus = repl::getGlobalReplicationCoordinator()
                                  ->awaitReplication(txn, lastOpApplied, userWriteConcern)
                                  .status;
  
          return majorityStatus.isOK() && userStatus.isOK();
      }
  
      bool flushPendingWrites(OperationContext* txn,
                              const std::string& ns,
                              BSONObj min,
                              BSONObj max,
                              const repl::OpTime& lastOpApplied,
                              const WriteConcernOptions& writeConcern) {
          if (!opReplicatedEnough(txn, lastOpApplied, writeConcern)) {
              repl::OpTime op(lastOpApplied);
              OCCASIONALLY warning() << "migrate commit waiting for a majority of slaves for '" << ns
                                     << "' " << min << " -> " << max << " waiting for: " << op
                                     << migrateLog;
              return false;
          }
  
          log() << "migrate commit succeeded flushing to secondaries for '" << ns << "' " << min
                << " -> " << max << migrateLog;
  
          {
              // Get global lock to wait for write to be commited to journal.
              ScopedTransaction transaction(txn, MODE_S);
              Lock::GlobalRead lk(txn->lockState());
  
              // if durability is on, force a write to journal
              if (getDur().commitNow(txn)) {
                  log() << "migrate commit flushed to journal for '" << ns << "' " << min << " -> "
                        << max << migrateLog;
              }
          }
  
          return true;
      }
  
      static string stateToString(State state) {
          switch (state) {
              case READY:
                  return "ready";
              case CLONE:
                  return "clone";
              case CATCHUP:
                  return "catchup";
              case STEADY:
                  return "steady";
              case COMMIT_START:
                  return "commitStart";
              case DONE:
                  return "done";
              case FAIL:
                  return "fail";
              case ABORT:
                  return "abort";
          }
          verify(0);
          return "";
      }
  
      bool startCommit() {
          stdx::unique_lock<stdx::mutex> lock(_mutex);
  
          if (_state != STEADY) {
              return false;
          }
  
          const auto deadline = system_clock::now() + seconds(30);
          _state = COMMIT_START;
          while (_active) {
              if (stdx::cv_status::timeout == isActiveCV.wait_until(lock, deadline)) {
                  _state = FAIL;
                  log() << "startCommit never finished!" << migrateLog;
                  return false;
              }
          }
  
          if (_state == DONE) {
              return true;
          }
  
          log() << "startCommit failed, final data failed to transfer" << migrateLog;
          return false;
      }
  
      void abort() {
          stdx::lock_guard<stdx::mutex> sl(_mutex);
          _state = ABORT;
          _errmsg = "aborted";
      }
  
      bool getActive() const {
          stdx::lock_guard<stdx::mutex> lk(_mutex);
          return _active;
      }
      void setActive(bool b) {
          stdx::lock_guard<stdx::mutex> lk(_mutex);
          _active = b;
          isActiveCV.notify_all();
      }
  
      // Guards all fields.
      mutable stdx::mutex _mutex;
      bool _active;
      stdx::condition_variable isActiveCV;
  
      std::string _ns;
      std::string _from;
      BSONObj _min;
      BSONObj _max;
      BSONObj _shardKeyPattern;
  
      long long _numCloned;
      long long _clonedBytes;
      long long _numCatchup;
      long long _numSteady;
  
      State _state;
      std::string _errmsg;
  
  } migrateStatus;
  
  void migrateThread(std::string ns,
                     BSONObj min,
                     BSONObj max,
                     BSONObj shardKeyPattern,
                     std::string fromShard,
                     OID epoch,
                     WriteConcernOptions writeConcern) {
      Client::initThread("migrateThread");
      OperationContextImpl txn;
      if (getGlobalAuthorizationManager()->isAuthEnabled()) {
          ShardedConnectionInfo::addHook();
          AuthorizationSession::get(txn.getClient())->grantInternalAuthorization();
      }
  
      migrateStatus.go(&txn, ns, min, max, shardKeyPattern, fromShard, epoch, writeConcern);
  }
  
  /**
   * Command for initiating the recipient side of the migration to start copying data
   * from the donor shard.
   *
   * {
   *   _recvChunkStart: "namespace",
   *   congfigServer: "hostAndPort",
   *   from: "hostAndPort",
   *   fromShardName: "shardName",
   *   toShardName: "shardName",
   *   min: {},
   *   max: {},
   *   shardKeyPattern: {},
   *
   *   // optional
   *   secondaryThrottle: bool, // defaults to true
   *   writeConcern: {} // applies to individual writes.
   * }
   */
  class RecvChunkStartCommand : public ChunkCommandHelper {
  public:
      void help(std::stringstream& h) const {
          h << "internal";
      }
      RecvChunkStartCommand() : ChunkCommandHelper("_recvChunkStart") {}
  
      virtual bool isWriteCommandForConfigServer() const {
          return false;
      }
      virtual void addRequiredPrivileges(const std::string& dbname,
                                         const BSONObj& cmdObj,
                                         std::vector<Privilege>* out) {
          ActionSet actions;
          actions.addAction(ActionType::internal);
          out->push_back(Privilege(ResourcePattern::forClusterResource(), actions));
      }
      bool run(OperationContext* txn,
               const string&,
               BSONObj& cmdObj,
               int,
               string& errmsg,
               BSONObjBuilder& result) {
          // Active state of TO-side migrations (MigrateStatus) is serialized by distributed
          // collection lock.
          if (migrateStatus.getActive()) {
              errmsg = "migrate already in progress";
              return false;
          }
  
          // Pending deletes (for migrations) are serialized by the distributed collection lock,
          // we are sure we registered a delete for a range *before* we can migrate-in a
          // subrange.
          const size_t numDeletes = getDeleter()->getTotalDeletes();
          if (numDeletes > 0) {
              errmsg = str::stream() << "can't accept new chunks because "
                                     << " there are still " << numDeletes
                                     << " deletes from previous migration";
  
              warning() << errmsg;
              return false;
          }
  
          if (!shardingState.enabled()) {
              if (!cmdObj["configServer"].eoo()) {
                  dassert(cmdObj["configServer"].type() == String);
                  shardingState.initialize(cmdObj["configServer"].String());
              } else {
                  errmsg = str::stream()
                      << "cannot start recv'ing chunk, "
                      << "sharding is not enabled and no config server was provided";
  
                  warning() << errmsg;
                  return false;
              }
          }
  
          if (!cmdObj["toShardName"].eoo()) {
              dassert(cmdObj["toShardName"].type() == String);
              shardingState.gotShardName(cmdObj["toShardName"].String());
          }
  
          string ns = cmdObj.firstElement().String();
          BSONObj min = cmdObj["min"].Obj().getOwned();
          BSONObj max = cmdObj["max"].Obj().getOwned();
  
          // Refresh our collection manager from the config server, we need a collection manager
          // to start registering pending chunks.
          // We force the remote refresh here to make the behavior consistent and predictable,
          // generally we'd refresh anyway, and to be paranoid.
          ChunkVersion currentVersion;
          Status status = shardingState.refreshMetadataNow(txn, ns, &currentVersion);
  
          if (!status.isOK()) {
              errmsg = str::stream() << "cannot start recv'ing chunk "
                                     << "[" << min << "," << max << ")" << causedBy(status.reason());
  
              warning() << errmsg;
              return false;
          }
  
          // Process secondary throttle settings and assign defaults if necessary.
          WriteConcernOptions writeConcern;
          status = writeConcern.parseSecondaryThrottle(cmdObj, NULL);
  
          if (!status.isOK()) {
              if (status.code() != ErrorCodes::WriteConcernNotDefined) {
                  warning() << status.toString();
                  return appendCommandStatus(result, status);
              }
  
              writeConcern = getDefaultWriteConcern();
          } else {
              repl::ReplicationCoordinator* replCoordinator = repl::getGlobalReplicationCoordinator();
  
              if (replCoordinator->getReplicationMode() ==
                      repl::ReplicationCoordinator::modeMasterSlave &&
                  writeConcern.shouldWaitForOtherNodes()) {
                  warning() << "recvChunk cannot check if secondary throttle setting "
                            << writeConcern.toBSON()
                            << " can be enforced in a master slave configuration";
              }
  
              Status status = replCoordinator->checkIfWriteConcernCanBeSatisfied(writeConcern);
              if (!status.isOK() && status != ErrorCodes::NoReplicationEnabled) {
                  warning() << status.toString();
                  return appendCommandStatus(result, status);
              }
          }
  
          if (writeConcern.shouldWaitForOtherNodes() &&
              writeConcern.wTimeout == WriteConcernOptions::kNoTimeout) {
              // Don't allow no timeout.
              writeConcern.wTimeout = kDefaultWTimeoutMs;
          }
  
          BSONObj shardKeyPattern;
          if (cmdObj.hasField("shardKeyPattern")) {
              shardKeyPattern = cmdObj["shardKeyPattern"].Obj().getOwned();
          } else {
              // shardKeyPattern may not be provided if another shard is from pre 2.2
              // In that case, assume the shard key pattern is the same as the range
              // specifiers provided.
              BSONObj keya = Helpers::inferKeyPattern(min);
              BSONObj keyb = Helpers::inferKeyPattern(max);
              verify(keya == keyb);
  
              warning()
                  << "No shard key pattern provided by source shard for migration."
                     " This is likely because the source shard is running a version prior to 2.2."
                     " Falling back to assuming the shard key matches the pattern of the min and max"
                     " chunk range specifiers.  Inferred shard key: " << keya;
  
              shardKeyPattern = keya.getOwned();
          }
  
          const string fromShard(cmdObj["from"].String());
  
          // Set the TO-side migration to active
          Status prepareStatus = migrateStatus.prepare(ns, fromShard, min, max, shardKeyPattern);
  
          if (!prepareStatus.isOK()) {
              return appendCommandStatus(result, prepareStatus);
          }
  
          stdx::thread m(migrateThread,
                         ns,
                         min,
                         max,
                         shardKeyPattern,
                         fromShard,
                         currentVersion.epoch(),
                         writeConcern);
  
          m.detach();
          result.appendBool("started", true);
          return true;
      }
  
  } recvChunkStartCmd;
  
  class RecvChunkStatusCommand : public ChunkCommandHelper {
  public:
      void help(std::stringstream& h) const {
          h << "internal";
      }
      RecvChunkStatusCommand() : ChunkCommandHelper("_recvChunkStatus") {}
      virtual void addRequiredPrivileges(const std::string& dbname,
                                         const BSONObj& cmdObj,
                                         std::vector<Privilege>* out) {
          ActionSet actions;
          actions.addAction(ActionType::internal);
          out->push_back(Privilege(ResourcePattern::forClusterResource(), actions));
      }
      bool run(OperationContext* txn,
               const string&,
               BSONObj& cmdObj,
               int,
               string& errmsg,
               BSONObjBuilder& result) {
          migrateStatus.status(result);
          return 1;
      }
  
  } recvChunkStatusCommand;
  
  class RecvChunkCommitCommand : public ChunkCommandHelper {
  public:
      void help(std::stringstream& h) const {
          h << "internal";
      }
      RecvChunkCommitCommand() : ChunkCommandHelper("_recvChunkCommit") {}
      virtual void addRequiredPrivileges(const std::string& dbname,
                                         const BSONObj& cmdObj,
                                         std::vector<Privilege>* out) {
          ActionSet actions;
          actions.addAction(ActionType::internal);
          out->push_back(Privilege(ResourcePattern::forClusterResource(), actions));
      }
      bool run(OperationContext* txn,
               const string&,
               BSONObj& cmdObj,
               int,
               string& errmsg,
               BSONObjBuilder& result) {
          bool ok = migrateStatus.startCommit();
          migrateStatus.status(result);
          return ok;
      }
  
  } recvChunkCommitCommand;
  
  class RecvChunkAbortCommand : public ChunkCommandHelper {
  public:
      void help(std::stringstream& h) const {
          h << "internal";
      }
      RecvChunkAbortCommand() : ChunkCommandHelper("_recvChunkAbort") {}
      virtual void addRequiredPrivileges(const std::string& dbname,
                                         const BSONObj& cmdObj,
                                         std::vector<Privilege>* out) {
          ActionSet actions;
          actions.addAction(ActionType::internal);
          out->push_back(Privilege(ResourcePattern::forClusterResource(), actions));
      }
      bool run(OperationContext* txn,
               const string&,
               BSONObj& cmdObj,
               int,
               string& errmsg,
               BSONObjBuilder& result) {
          migrateStatus.abort();
          migrateStatus.status(result);
          return true;
      }
  
  } recvChunkAboortCommand;
  
  
  class IsInRangeTest : public StartupTest {
  public:
      void run() {
          BSONObj min = BSON("x" << 1);
          BSONObj max = BSON("x" << 5);
          BSONObj skey = BSON("x" << 1);
  
          verify(!isInRange(BSON("x" << 0), min, max, skey));
          verify(isInRange(BSON("x" << 1), min, max, skey));
          verify(isInRange(BSON("x" << 3), min, max, skey));
          verify(isInRange(BSON("x" << 4), min, max, skey));
          verify(!isInRange(BSON("x" << 5), min, max, skey));
          verify(!isInRange(BSON("x" << 6), min, max, skey));
  
          BSONObj obj = BSON("n" << 3);
          BSONObj min2 = BSON("x" << BSONElementHasher::hash64(obj.firstElement(), 0) - 2);
          BSONObj max2 = BSON("x" << BSONElementHasher::hash64(obj.firstElement(), 0) + 2);
          BSONObj hashedKey = BSON("x"
                                   << "hashed");
  
          verify(isInRange(BSON("x" << 3), min2, max2, hashedKey));
          verify(!isInRange(BSON("x" << 3), min, max, hashedKey));
          verify(!isInRange(BSON("x" << 4), min2, max2, hashedKey));
  
          LOG(1) << "isInRangeTest passed" << migrateLog;
      }
  } isInRangeTest;
  }
  