/**
     * The core config write functionality.
     *
     * Config writes run in two passes - the first is a quick check to ensure the config servers
     * are all reachable, the second runs the actual write.
     *
     * TODO: Upgrade and move this logic to the config servers, a state machine implementation
     * is probably the next step.
     */
    void ConfigCoordinator::executeBatch( const BatchedCommandRequest& clientRequest,
                                          BatchedCommandResponse* clientResponse,
                                          bool fsyncCheck ) {

        NamespaceString nss( clientRequest.getNS() );
        dassert( nss.db() == "config" || nss.db() == "admin" );
        dassert( clientRequest.sizeWriteOps() == 1u );

        if ( fsyncCheck ) {

            //
            // Sanity check that all configs are still reachable using fsync, preserving legacy
            // behavior
            //

            OwnedPointerVector<ConfigFsyncResponse> fsyncResponsesOwned;
            vector<ConfigFsyncResponse*>& fsyncResponses = fsyncResponsesOwned.mutableVector();

            //
            // Send side
            //

            for ( vector<ConnectionString>::iterator it = _configHosts.begin();
                it != _configHosts.end(); ++it ) {
                ConnectionString& configHost = *it;
                FsyncRequest fsyncRequest;
                _dispatcher->addCommand( configHost, "admin", fsyncRequest );
            }

            _dispatcher->sendAll();

            //
            // Recv side
            //

            bool fsyncError = false;
            while ( _dispatcher->numPending() > 0 ) {

                fsyncResponses.push_back( new ConfigFsyncResponse() );
                ConfigFsyncResponse& fsyncResponse = *fsyncResponses.back();
                Status dispatchStatus = _dispatcher->recvAny( &fsyncResponse.configHost,
                                                              &fsyncResponse.response );

                // We've got to recv everything, no matter what
                if ( !dispatchStatus.isOK() ) {
                    fsyncError = true;
                    buildFsyncErrorFrom( dispatchStatus, &fsyncResponse.response );
                }
                else if ( !fsyncResponse.response.getOk() ) {
                    fsyncError = true;
                }
            }

            if ( fsyncError ) {
                combineFsyncErrors( fsyncResponses, clientResponse );
                return;
            }
            else {
                fsyncResponsesOwned.clear();
            }
        }

        //
        // Do the actual writes
        //

        BatchedCommandRequest configRequest( clientRequest.getBatchType() );
        clientRequest.cloneTo( &configRequest );
        configRequest.setNS( nss.coll() );

        OwnedPointerVector<ConfigResponse> responsesOwned;
        vector<ConfigResponse*>& responses = responsesOwned.mutableVector();

        //
        // Send the actual config writes
        //

        // Get as many batches as we can at once
        for ( vector<ConnectionString>::iterator it = _configHosts.begin();
            it != _configHosts.end(); ++it ) {
            ConnectionString& configHost = *it;
            _dispatcher->addCommand( configHost, nss.db(), configRequest );
        }

        // Send them all out
        _dispatcher->sendAll();

        //
        // Recv side
        //

//.........这里部分代码省略.........

//.........这里部分代码省略.........

        //
        // Try to enforce the write concern if everything succeeded (unordered or ordered)
        // OR if something succeeded and we're unordered.
        //

        auto_ptr<WCErrorDetail> wcError;
        bool needToEnforceWC = writeErrors.empty()
                               || ( !request.getOrdered()
                                    && writeErrors.size() < request.sizeWriteOps() );

        if ( needToEnforceWC ) {

            _client->curop()->setMessage( "waiting for write concern" );

            WriteConcernResult res;
            Status status = waitForWriteConcern( _txn, writeConcern, _client->getLastOp(), &res );

            if ( !status.isOK() ) {
                wcError.reset( toWriteConcernError( status, res ) );
            }
        }

        //
        // Refresh metadata if needed
        //

        bool staleBatch = !writeErrors.empty()
                          && writeErrors.back()->getErrCode() == ErrorCodes::StaleShardVersion;

        if ( staleBatch ) {

            const BatchedRequestMetadata* requestMetadata = request.getMetadata();
            dassert( requestMetadata );

            // Make sure our shard name is set or is the same as what was set previously
            if ( shardingState.setShardName( requestMetadata->getShardName() ) ) {

                //
                // First, we refresh metadata if we need to based on the requested version.
                //

                ChunkVersion latestShardVersion;
                shardingState.refreshMetadataIfNeeded( request.getTargetingNS(),
                                                       requestMetadata->getShardVersion(),
                                                       &latestShardVersion );

                // Report if we're still changing our metadata
                // TODO: Better reporting per-collection
                if ( shardingState.inCriticalMigrateSection() ) {
                    noteInCriticalSection( writeErrors.back() );
                }

                if ( queueForMigrationCommit ) {

                    //
                    // Queue up for migration to end - this allows us to be sure that clients will
                    // not repeatedly try to refresh metadata that is not yet written to the config
                    // server.  Not necessary for correctness.
                    // Exposed as optional parameter to allow testing of queuing behavior with
                    // different network timings.
                    //

                    const ChunkVersion& requestShardVersion = requestMetadata->getShardVersion();

                    //

void CScriptDebugging::OnLuaMainDestroy ( CLuaMain* pLuaMain )
{
    dassert ( !ListContains ( m_LuaMainStack, pLuaMain ) );
    ListRemove ( m_LuaMainStack, pLuaMain );
}

const UpdateIndexData& CollectionInfoCache::getIndexKeys(OperationContext* opCtx) const {
    // This requires "some" lock, and MODE_IS is an expression for that, for now.
    dassert(opCtx->lockState()->isCollectionLockedForMode(_collection->ns().ns(), MODE_IS));
    invariant(_keysComputed);
    return _indexedPaths;
}

        void nfs_client_impl::end_get_file_size(
            ::dsn::error_code err,
            const ::dsn::service::get_file_size_response& resp,
            void* context)
        {
            user_request* ureq = (user_request*)context;

            if (err != ::dsn::ERR_OK)
            {
                derror("remote copy request failed");
                ureq->nfs_task->enqueue(err, 0, ureq->nfs_task->node());
                delete ureq;
                return;
            }

            err.set(resp.error);
            if (err != ::dsn::ERR_OK)
            {
                derror("remote copy request failed");
                error_code resp_err;
                resp_err.set(resp.error);
                ureq->nfs_task->enqueue(resp_err, 0, ureq->nfs_task->node());
                delete ureq;
                return;
            }

            for (size_t i = 0; i < resp.size_list.size(); i++) // file list
            {
                file_context *filec;
                uint64_t size = resp.size_list[i];

                filec = new file_context(ureq, resp.file_list[i], resp.size_list[i]);
                ureq->file_context_map.insert(std::pair<std::string, file_context*>(
                    ureq->file_size_req.dst_dir + resp.file_list[i], filec));

                //dinfo("this file size is %d, name is %s", size, resp.file_list[i].c_str());

                // new all the copy requests                

                uint64_t req_offset = 0;
                uint32_t req_size;
                if (size > _opts.nfs_copy_block_bytes)
                    req_size = _opts.nfs_copy_block_bytes;
                else
                    req_size = static_cast<uint32_t>(size);

                int idx = 0;
                for (;;) // send one file with multi-round rpc
                {
                    auto req = boost::intrusive_ptr<copy_request_ex>(new copy_request_ex(filec, idx++));
                    filec->copy_requests.push_back(req);

                    {
                        zauto_lock l(_copy_requests_lock);
                        _copy_requests.push(req);
                    }

                    req->copy_req.source = ureq->file_size_req.source;
                    req->copy_req.file_name = resp.file_list[i];
                    req->copy_req.offset = req_offset;
                    req->copy_req.size = req_size;
                    req->copy_req.dst_dir = ureq->file_size_req.dst_dir;
                    req->copy_req.source_dir = ureq->file_size_req.source_dir;
                    req->copy_req.overwrite = ureq->file_size_req.overwrite;
                    req->copy_req.is_last = (size <= req_size);

                    req_offset += req_size;
                    size -= req_size;
                    if (size <= 0)
                    {
                        dassert(size == 0, "last request must read exactly the remaing size of the file");
                        break;
                    }

                    if (size > _opts.nfs_copy_block_bytes)
                        req_size = _opts.nfs_copy_block_bytes;
                    else
                        req_size = static_cast<uint32_t>(size);
                }
            }

            continue_copy(0);
        }

    Status DBClientShardResolver::chooseWriteHost( const string& shardName,
                                                   ConnectionString* shardHost ) const {

        // Declare up here for parsing later
        string errMsg;

        // Special-case for config and admin
        if ( shardName == "config" || shardName == "admin" ) {
            *shardHost = ConnectionString::parse( configServer.modelServer(), errMsg );
            dassert( errMsg == "" );
            return Status::OK();
        }

        //
        // First get the information about the shard from the shard cache
        //

        // Internally uses our shard cache, does no reload
        Shard shard = Shard::findIfExists( shardName );
        if ( shard.getName() == "" ) {
            return Status( ErrorCodes::ShardNotFound,
                           string("unknown shard name ") + shardName );
        }

        ConnectionString rawShardHost = ConnectionString::parse( shard.getConnString(), errMsg );
        dassert( errMsg == "" );
        dassert( rawShardHost.type() == ConnectionString::SET
                 || rawShardHost.type() == ConnectionString::MASTER );

        if ( rawShardHost.type() == ConnectionString::MASTER ) {
            *shardHost = rawShardHost;
            return Status::OK();
        }

        //
        // If we need to, then get the particular node we're targeting in the replica set
        //

        // Does not reload the monitor if it doesn't currently exist
        ReplicaSetMonitorPtr replMonitor = ReplicaSetMonitor::get( rawShardHost.getSetName(),
                                                                   false );
        if ( !replMonitor ) {
            return Status( ErrorCodes::ReplicaSetNotFound,
                           string("unknown replica set ") + rawShardHost.getSetName() );
        }

        try {
            // This can throw when we don't find a master!
            HostAndPort masterHostAndPort = replMonitor->getMaster();
            *shardHost = ConnectionString::parse( masterHostAndPort.toString( true ), errMsg );
            dassert( errMsg == "" );
            return Status::OK();
        }
        catch ( const DBException& ) {
            return Status( ErrorCodes::HostNotFound,
                           string("could not contact primary for replica set ")
                           + replMonitor->getName() );
        }

        // Unreachable
        dassert( false );
        return Status( ErrorCodes::UnknownError, "" );
    }

KeyV1Owned::KeyV1Owned(const KeyV1& rhs) {
    b.appendBuf(rhs.data(), rhs.dataSize());
    _keyData = (const unsigned char*)b.buf();
    dassert(b.len() == dataSize());  // check datasize method is correct
    dassert((*_keyData & cNOTUSED) == 0);
}

        error_code io_looper::bind_io_handle(
            dsn_handle_t handle,
            io_loop_callback* cb,
            unsigned int events,
            ref_counter* ctx
            )
        {
            int fd;
            short filters[2];
            int nr_filters;
            struct kevent e;           

            if (cb == nullptr)
            {
                derror("cb == nullptr");
                return ERR_INVALID_PARAMETERS;
            }

            fd = (int)(intptr_t)(handle);
            if (fd < 0)
            {
                if (fd != IO_LOOPER_USER_NOTIFICATION_FD)
                {
                    derror("The fd %d is less than 0.", fd);
                    return ERR_INVALID_PARAMETERS;
                }
            }

            if (_filters.find((short)events) == _filters.end())
            {
                derror("The filter %u is unsupported.", events);
                return ERR_INVALID_PARAMETERS;
            }

            if (fd > 0)
            {
                int flags = fcntl(fd, F_GETFL, 0);
                dassert (flags != -1, "fcntl failed, err = %s, fd = %d", strerror(errno), fd);

                if (!(flags & O_NONBLOCK))
                {
                    flags |= O_NONBLOCK;
                    flags = fcntl(fd, F_SETFL, flags);
                    dassert(flags != -1, "fcntl failed, err = %s, fd = %d", strerror(errno), fd);
                }
            }

            uintptr_t cb0 = (uintptr_t)cb;
            dassert((cb0 & 0x1) == 0, "the least one bit must be zero for the callback address");

            if (ctx)
            {
                cb0 |= 0x1; // has ref_counter

                utils::auto_lock<utils::ex_lock_nr_spin> l(_io_sessions_lock);
                auto pr = _io_sessions.insert(io_sessions::value_type(cb, ctx));
                dassert(pr.second, "the callback must not be registered before");
            }
            
            if ((short)events == EVFILT_READ_WRITE)
            {
                filters[0] = EVFILT_READ;
                filters[1] = EVFILT_WRITE;
                nr_filters = 2;
            }
            else
            {
                filters[0] = (short)events;
                nr_filters = 1;
            }

            for (int i = 0; i < nr_filters; i++)
            {
                EV_SET(&e, fd, filters[i], (EV_ADD | EV_ENABLE | EV_CLEAR), 0, 0, (void*)cb0);

                if (kevent(_io_queue, &e, 1, nullptr, 0, nullptr) == -1)
                {
                    derror("bind io handler to kqueue failed, err = %s, fd = %d", strerror(errno), fd);

                    if (ctx)
                    {
                        utils::auto_lock<utils::ex_lock_nr_spin> l(_io_sessions_lock);
                        auto r = _io_sessions.erase(cb);
                        dassert(r > 0, "the callback must be present");
                    }

                    for (int j = 0; j < i; j++)
                    {
                        EV_SET(&e, fd, filters[j], EV_DELETE, 0, 0, nullptr);
                        if (kevent(_io_queue, &e, 1, nullptr, 0, nullptr) == -1)
                        {
                            derror("Unregister kqueue failed, filter = %d, err = %s, fd = %d", filters[j], strerror(errno), fd);
                        }
                    }

                    return ERR_BIND_IOCP_FAILED;
                }
            }

            return ERR_OK;
//.........这里部分代码省略.........

    bool run(OperationContext* txn,
             const string& dbname,
             BSONObj& cmdObj,
             int,
             string& errmsg,
             BSONObjBuilder& result) {
        BSONElement first = cmdObj.firstElement();
        uassert(28528,
                str::stream() << "Argument to listIndexes must be of type String, not "
                              << typeName(first.type()),
                first.type() == String);
        StringData collectionName = first.valueStringData();
        uassert(28529,
                str::stream() << "Argument to listIndexes must be a collection name, "
                              << "not the empty string",
                !collectionName.empty());
        const NamespaceString ns(dbname, collectionName);

        const long long defaultBatchSize = std::numeric_limits<long long>::max();
        long long batchSize;
        Status parseCursorStatus = parseCommandCursorOptions(cmdObj, defaultBatchSize, &batchSize);
        if (!parseCursorStatus.isOK()) {
            return appendCommandStatus(result, parseCursorStatus);
        }

        AutoGetCollectionForRead autoColl(txn, ns);
        if (!autoColl.getDb()) {
            return appendCommandStatus(result,
                                       Status(ErrorCodes::NamespaceNotFound, "no database"));
        }

        const Collection* collection = autoColl.getCollection();
        if (!collection) {
            return appendCommandStatus(result,
                                       Status(ErrorCodes::NamespaceNotFound, "no collection"));
        }

        const CollectionCatalogEntry* cce = collection->getCatalogEntry();
        invariant(cce);

        vector<string> indexNames;
        MONGO_WRITE_CONFLICT_RETRY_LOOP_BEGIN {
            indexNames.clear();
            cce->getAllIndexes(txn, &indexNames);
        }
        MONGO_WRITE_CONFLICT_RETRY_LOOP_END(txn, "listIndexes", ns.ns());

        auto ws = make_unique<WorkingSet>();
        auto root = make_unique<QueuedDataStage>(txn, ws.get());

        for (size_t i = 0; i < indexNames.size(); i++) {
            BSONObj indexSpec;
            MONGO_WRITE_CONFLICT_RETRY_LOOP_BEGIN {
                indexSpec = cce->getIndexSpec(txn, indexNames[i]);
            }
            MONGO_WRITE_CONFLICT_RETRY_LOOP_END(txn, "listIndexes", ns.ns());

            WorkingSetID id = ws->allocate();
            WorkingSetMember* member = ws->get(id);
            member->keyData.clear();
            member->loc = RecordId();
            member->obj = Snapshotted<BSONObj>(SnapshotId(), indexSpec.getOwned());
            member->transitionToOwnedObj();
            root->pushBack(id);
        }

        std::string cursorNamespace = str::stream() << dbname << ".$cmd." << name << "."
                                                    << ns.coll();
        dassert(NamespaceString(cursorNamespace).isValid());
        dassert(NamespaceString(cursorNamespace).isListIndexesCursorNS());
        dassert(ns == NamespaceString(cursorNamespace).getTargetNSForListIndexes());

        auto statusWithPlanExecutor = PlanExecutor::make(
            txn, std::move(ws), std::move(root), cursorNamespace, PlanExecutor::YIELD_MANUAL);
        if (!statusWithPlanExecutor.isOK()) {
            return appendCommandStatus(result, statusWithPlanExecutor.getStatus());
        }
        unique_ptr<PlanExecutor> exec = std::move(statusWithPlanExecutor.getValue());

        BSONArrayBuilder firstBatch;

        const int byteLimit = FindCommon::kMaxBytesToReturnToClientAtOnce;
        for (long long objCount = 0; objCount < batchSize && firstBatch.len() < byteLimit;
             objCount++) {
            BSONObj next;
            PlanExecutor::ExecState state = exec->getNext(&next, NULL);
            if (state == PlanExecutor::IS_EOF) {
                break;
            }
            invariant(state == PlanExecutor::ADVANCED);
            firstBatch.append(next);
        }

        CursorId cursorId = 0LL;
        if (!exec->isEOF()) {
            exec->saveState();
            exec->detachFromOperationContext();
            ClientCursor* cursor =
                new ClientCursor(CursorManager::getGlobalCursorManager(),
                                 exec.release(),
//.........这里部分代码省略.........

    Status MetadataLoader::initChunks( const string& ns,
                                       const string& shard,
                                       const CollectionMetadata* oldMetadata,
                                       CollectionMetadata* metadata ) const
    {
        map<string, ChunkVersion> versionMap;

        // Preserve the epoch
        versionMap[shard] = metadata->_shardVersion;
        OID epoch = metadata->getCollVersion().epoch();
        bool fullReload = true;

        // Check to see if we should use the old version or not.
        if ( oldMetadata ) {

            // If our epochs are compatible, it's useful to use the old metadata for diffs
            if ( oldMetadata->getCollVersion().hasCompatibleEpoch( epoch ) ) {

                fullReload = false;
                dassert( oldMetadata->isValid() );

                versionMap[shard] = oldMetadata->_shardVersion;
                metadata->_collVersion = oldMetadata->_collVersion;

                // TODO: This could be made more efficient if copying not required, but
                // not as frequently reloaded as in mongos.
                metadata->_chunksMap = oldMetadata->_chunksMap;

                LOG( 2 ) << "loading new chunks for collection " << ns
                         << " using old metadata w/ version " << oldMetadata->getShardVersion()
                         << " and " << metadata->_chunksMap.size() << " chunks" << endl;
            }
            else {
                warning() << "reloading collection metadata for " << ns << " with new epoch "
                          << epoch.toString() << ", the current epoch is "
                          << oldMetadata->getCollVersion().epoch().toString() << endl;
            }
        }


        // Exposes the new metadata's range map and version to the "differ," who
        // would ultimately be responsible of filling them up.
        SCMConfigDiffTracker differ( shard );
        differ.attach( ns, metadata->_chunksMap, metadata->_collVersion, versionMap );

        try {

            ScopedDbConnection conn( _configLoc.toString(), 30 );

            auto_ptr<DBClientCursor> cursor = conn->query( ChunkType::ConfigNS,
                                                           differ.configDiffQuery() );

            if ( !cursor.get() ) {

                // Make our metadata invalid
                metadata->_collVersion = ChunkVersion( 0, 0, OID() );
                metadata->_chunksMap.clear();
                conn.done();

                return Status( ErrorCodes::HostUnreachable,
                               "problem opening chunk metadata cursor" );
            }

            //
            // The diff tracker should always find at least one chunk (the highest chunk we saw
            // last time).  If not, something has changed on the config server (potentially between
            // when we read the collection data and when we read the chunks data).
            //

            int diffsApplied = differ.calculateConfigDiff( *cursor );
            if ( diffsApplied > 0 ) {

                // Chunks found, return ok

                LOG(2) << "loaded " << diffsApplied << " chunks into new metadata for " << ns
                           << " with version " << metadata->_collVersion << endl;

                metadata->_shardVersion = versionMap[shard];
                metadata->fillRanges();
                conn.done();

                dassert( metadata->isValid() );
                return Status::OK();
            }
            else if ( diffsApplied == 0 ) {

                // No chunks found, the collection is dropping or we're confused
                // If this is a full reload, assume it is a drop for backwards compatibility
                // TODO: drop the config.collections entry *before* the chunks and eliminate this
                // ambiguity

                string errMsg =
                    str::stream() << "no chunks found when reloading " << ns
                                  << ", previous version was "
                                  << metadata->_collVersion.toString()
                                  << ( fullReload ? ", this is a drop" : "" );

                warning() << errMsg << endl;

                metadata->_collVersion = ChunkVersion( 0, 0, OID() );
//.........这里部分代码省略.........

    Status MetadataLoader::initCollection( const string& ns,
                                           const string& shard,
                                           CollectionMetadata* metadata ) const
    {
        //
        // Bring collection entry from the config server.
        //

        BSONObj collDoc;
        {
            try {
                ScopedDbConnection conn( _configLoc.toString(), 30 );
                collDoc = conn->findOne( CollectionType::ConfigNS, QUERY(CollectionType::ns()<<ns));
                conn.done();
            }
            catch ( const DBException& e ) {
                string errMsg = str::stream() << "could not query collection metadata"
                                              << causedBy( e );

                // We deliberately do not return conn to the pool, since it was involved
                // with the error here.

                return Status( ErrorCodes::HostUnreachable, errMsg );
            }
        }

        string errMsg;
        if ( collDoc.isEmpty() ) {

            errMsg = str::stream() << "could not load metadata, collection " << ns << " not found";
            warning() << errMsg << endl;

            return Status( ErrorCodes::NamespaceNotFound, errMsg );
        }

        CollectionType collInfo;
        if ( !collInfo.parseBSON( collDoc, &errMsg ) || !collInfo.isValid( &errMsg ) ) {

            errMsg = str::stream() << "could not parse metadata for collection " << ns
                                   << causedBy( errMsg );
            warning() << errMsg << endl;

            return Status( ErrorCodes::FailedToParse, errMsg );
        }

        if ( collInfo.isDroppedSet() && collInfo.getDropped() ) {

            errMsg = str::stream() << "could not load metadata, collection " << ns
                                   << " was dropped";
            warning() << errMsg << endl;

            return Status( ErrorCodes::NamespaceNotFound, errMsg );
        }

        if ( collInfo.isKeyPatternSet() && !collInfo.getKeyPattern().isEmpty() ) {

            // Sharded collection, need to load chunks

            metadata->_keyPattern = collInfo.getKeyPattern();
            metadata->_shardVersion = ChunkVersion( 0, 0, collInfo.getEpoch() );
            metadata->_collVersion = ChunkVersion( 0, 0, collInfo.getEpoch() );

            return Status::OK();
        }
        else if ( collInfo.isPrimarySet() && collInfo.getPrimary() == shard ) {

            // A collection with a non-default primary

            // Empty primary field not allowed if set
            dassert( collInfo.getPrimary() != "" );

            metadata->_keyPattern = BSONObj();
            metadata->_shardVersion = ChunkVersion( 1, 0, collInfo.getEpoch() );
            metadata->_collVersion = metadata->_shardVersion;

            return Status::OK();
        }
        else {

            // A collection with a primary that doesn't match this shard or is empty, the primary
            // may have changed before we loaded.

            errMsg = // br
                    str::stream() << "collection " << ns << " does not have a shard key "
                                  << "and primary "
                                  << ( collInfo.isPrimarySet() ? collInfo.getPrimary() : "" )
                                  << " does not match this shard " << shard;

            warning() << errMsg << endl;

            metadata->_collVersion = ChunkVersion( 0, 0, OID() );

            return Status( ErrorCodes::RemoteChangeDetected, errMsg );
        }
    }

Status ReadConcernArgs::initialize(const BSONElement& readConcernElem) {
    invariant(isEmpty());  // only legal to call on uninitialized object.

    if (readConcernElem.eoo()) {
        return Status::OK();
    }

    dassert(readConcernElem.fieldNameStringData() == kReadConcernFieldName);

    if (readConcernElem.type() != Object) {
        return Status(ErrorCodes::FailedToParse,
                      str::stream() << kReadConcernFieldName << " field should be an object");
    }

    BSONObj readConcernObj = readConcernElem.Obj();
    for (auto&& field : readConcernObj) {
        auto fieldName = field.fieldNameStringData();
        if (fieldName == kAfterOpTimeFieldName) {
            OpTime opTime;
            // TODO pass field in rather than scanning again.
            auto opTimeStatus =
                bsonExtractOpTimeField(readConcernObj, kAfterOpTimeFieldName, &opTime);
            if (!opTimeStatus.isOK()) {
                return opTimeStatus;
            }
            _opTime = opTime;
        } else if (fieldName == kAfterClusterTimeFieldName) {
            Timestamp afterClusterTime;
            auto afterClusterTimeStatus = bsonExtractTimestampField(
                readConcernObj, kAfterClusterTimeFieldName, &afterClusterTime);
            if (!afterClusterTimeStatus.isOK()) {
                return afterClusterTimeStatus;
            }
            _afterClusterTime = LogicalTime(afterClusterTime);
        } else if (fieldName == kAtClusterTimeFieldName) {
            Timestamp atClusterTime;
            auto atClusterTimeStatus =
                bsonExtractTimestampField(readConcernObj, kAtClusterTimeFieldName, &atClusterTime);
            if (!atClusterTimeStatus.isOK()) {
                return atClusterTimeStatus;
            }
            _atClusterTime = LogicalTime(atClusterTime);
        } else if (fieldName == kLevelFieldName) {
            std::string levelString;
            // TODO pass field in rather than scanning again.
            auto readCommittedStatus =
                bsonExtractStringField(readConcernObj, kLevelFieldName, &levelString);

            if (!readCommittedStatus.isOK()) {
                return readCommittedStatus;
            }

            if (levelString == kLocalReadConcernStr) {
                _level = ReadConcernLevel::kLocalReadConcern;
            } else if (levelString == kMajorityReadConcernStr) {
                _level = ReadConcernLevel::kMajorityReadConcern;
            } else if (levelString == kLinearizableReadConcernStr) {
                _level = ReadConcernLevel::kLinearizableReadConcern;
            } else if (levelString == kAvailableReadConcernStr) {
                _level = ReadConcernLevel::kAvailableReadConcern;
            } else if (levelString == kSnapshotReadConcernStr) {
                _level = ReadConcernLevel::kSnapshotReadConcern;
            } else {
                return Status(ErrorCodes::FailedToParse,
                              str::stream() << kReadConcernFieldName << '.' << kLevelFieldName
                                            << " must be either 'local', 'majority', "
                                               "'linearizable', 'available', or 'snapshot'");
            }
        } else {
            return Status(ErrorCodes::InvalidOptions,
                          str::stream() << "Unrecognized option in " << kReadConcernFieldName
                                        << ": "
                                        << fieldName);
        }
    }

    if (_afterClusterTime && _opTime) {
        return Status(ErrorCodes::InvalidOptions,
                      str::stream() << "Can not specify both " << kAfterClusterTimeFieldName
                                    << " and "
                                    << kAfterOpTimeFieldName);
    }

    if (_afterClusterTime && _atClusterTime) {
        return Status(ErrorCodes::InvalidOptions,
                      str::stream() << "Can not specify both " << kAfterClusterTimeFieldName
                                    << " and "
                                    << kAtClusterTimeFieldName);
    }

    // Note: 'available' should not be used with after cluster time, as cluster time can wait for
    // replication whereas the premise of 'available' is to avoid waiting. 'linearizable' should not
    // be used with after cluster time, since linearizable reads are inherently causally consistent.
    if (_afterClusterTime && getLevel() != ReadConcernLevel::kMajorityReadConcern &&
        getLevel() != ReadConcernLevel::kLocalReadConcern &&
        getLevel() != ReadConcernLevel::kSnapshotReadConcern) {
        return Status(ErrorCodes::InvalidOptions,
                      str::stream() << kAfterClusterTimeFieldName << " field can be set only if "
                                    << kLevelFieldName
                                    << " is equal to "
//.........这里部分代码省略.........

void FindAndModifyRequest::setUpsert(bool upsert) {
    dassert(_update);
    _isUpsert = upsert;
}

    // @param reconf true if this is a reconfiguration and not an initial load of the configuration.
    // @return true if ok; throws if config really bad; false if config doesn't include self
    bool ReplSetImpl::initFromConfig(OperationContext* txn, ReplSetConfig& c, bool reconf) {
        // NOTE: haveNewConfig() writes the new config to disk before we get here.  So
        //       we cannot error out at this point, except fatally.  Check errors earlier.
        lock lk(this);

        if (!getLastErrorDefault.isEmpty() || !c.getLastErrorDefaults.isEmpty()) {
            getLastErrorDefault = c.getLastErrorDefaults;
        }

        list<ReplSetConfig::MemberCfg*> newOnes;
        // additive short-cuts the new config setup. If we are just adding a
        // node/nodes and nothing else is changing, this is additive. If it's
        // not a reconfig, we're not adding anything
        bool additive = reconf;
        bool updateConfigs = false;
        {
            unsigned nfound = 0;
            int me = 0;
            for (vector<ReplSetConfig::MemberCfg>::iterator i = c.members.begin();
                    i != c.members.end();
                    i++) {
                
                ReplSetConfig::MemberCfg& m = *i;
                if (isSelf(m.h)) {
                    me++;
                }
                
                if (reconf) {
                    const Member *old = findById(m._id);
                    if (old) {
                        nfound++;
                        verify((int) old->id() == m._id);
                        if (!old->config().isSameIgnoringTags(m)) {
                            additive = false;
                        }
                        if (!updateConfigs && old->config() != m) {
                            updateConfigs = true;
                        }
                    }
                    else {
                        newOnes.push_back(&m);
                    }
                }
            }
            if (me == 0) { // we're not in the config -- we must have been removed
                if (state().shunned()) {
                    // already took note of our ejection from the set
                    // so just sit tight and poll again
                    return false;
                }

                _members.orphanAll();

                // kill off rsHealthPoll threads (because they Know Too Much about our past)
                endOldHealthTasks();

                // close sockets to force clients to re-evaluate this member
                MessagingPort::closeAllSockets(0);

                // take note of our ejection
                changeState(MemberState::RS_SHUNNED);

                // go into holding pattern
                log() << "replSet info self not present in the repl set configuration:" << rsLog;
                log() << c.toString() << rsLog;

                loadConfig(txn);  // redo config from scratch
                return false; 
            }
            uassert(13302, "replSet error self appears twice in the repl set configuration", me<=1);

            // if we found different members that the original config, reload everything
            if (reconf && config().members.size() != nfound)
                additive = false;
        }

        // If we are changing chaining rules, we don't want this to be an additive reconfig so that
        // the primary can step down and the sync targets change.
        // TODO: This can be removed once SERVER-5208 is fixed.
        if (reconf && config().chainingAllowed() != c.chainingAllowed()) {
            additive = false;
        }

        _cfg = new ReplSetConfig(c);
        // config() is same thing but const, so we use that when we can for clarity below
        dassert(&config() == _cfg);
        verify(config().ok());
        verify(_name.empty() || _name == config()._id);
        _name = config()._id;
        verify(!_name.empty());
        // this is a shortcut for simple changes
        if (additive) {
            log() << "replSet info : additive change to configuration" << rsLog;
            if (updateConfigs) {
                // we have new configs for existing members, so we need to repopulate _members
                // with the most recent configs
                _members.orphanAll();

//.........这里部分代码省略.........

        error_code deploy_svc_service_impl::start()
        {
            std::string pdir = utils::filesystem::path_combine(dsn_get_current_app_data_dir(), "services");
            _service_dir = dsn_config_get_value_string("deploy.service",
                "deploy_dir",
                pdir.c_str(),
                "where to put temporal deployment resources"
                );

            // load clusters
            const char* clusters[100];
            int sz = 100;
            int count = dsn_config_get_all_keys("deploy.service.clusters", clusters, &sz);
            dassert(count <= 100, "too many clusters");

            for (int i = 0; i < count; i++)
            {
                std::string cluster_name = dsn_config_get_value_string(
                    clusters[i],
                    "name",
                    "",
                    "cluster name"
                    );

                if (nullptr != get_cluster(cluster_name))
                {
                    derror("cluster %s already defined", cluster_name.c_str());
                    return ERR_CLUSTER_ALREADY_EXIST;
                }

                std::string cluster_factory_type = dsn_config_get_value_string(
                    clusters[i],
                    "factory",
                    "",
                    "factory name to create the target cluster scheduler"
                    );

                auto cluster = ::dsn::utils::factory_store<cluster_scheduler>::create(
                    cluster_factory_type.c_str(),
                    PROVIDER_TYPE_MAIN
                    );

                if (nullptr == cluster)
                {
                    derror("cluster type %s is not defined", cluster_factory_type.c_str());
                    return ERR_OBJECT_NOT_FOUND;
                }

                std::shared_ptr<cluster_ex> ce(new cluster_ex);
                ce->scheduler.reset(cluster);
                ce->cluster.name = cluster_name;
                ce->cluster.type = cluster->type();

                _clusters[cluster_name] = ce;
            }

            _cli_deploy = dsn_cli_app_register(
                "deploy",
                "deploy deploy_request(in json format)",
                "deploy an app via our deployment service",
                (void*)this,
                [](void *context, int argc, const char **argv, dsn_cli_reply *reply)
                {
                    auto this_ = (deploy_svc_service_impl*)context;
                    this_->on_deploy_cli(context, argc, argv, reply);
                },
                __svc_cli_freeer__
                );

            _cli_undeploy = dsn_cli_app_register(
                "undeploy",
                "undeploy service_name(in json format)",
                "undeploy an app via our deployment service",
                (void*)this,
                [](void *context, int argc, const char **argv, dsn_cli_reply *reply)
                {
                    auto this_ = (deploy_svc_service_impl*)context;
                    this_->on_undeploy_cli(context, argc, argv, reply);
                },
                __svc_cli_freeer__
                );

            _cli_get_service_list = dsn_cli_app_register(
                "service_list",
                "service