自学教程：C++ uassert函数代码示例

//.........这里部分代码省略.........                chunks.push_back( chunk );            }            cursor.reset();            if (shardToChunksMap.empty()) {                LOG(1) << "skipping empty collection (" << ns << ")";                continue;            }            for ( vector<Shard>::iterator i=allShards.begin(); i!=allShards.end(); ++i ) {                // this just makes sure there is an entry in shardToChunksMap for every shard                Shard s = *i;                shardToChunksMap[s.getName()].size();            }            DistributionStatus status( shardInfo, shardToChunksMap );            // load tags            Status result = clusterCreateIndex(TagsType::ConfigNS,                                               BSON(TagsType::ns() << 1 << TagsType::min() << 1),                                               true, // unique                                               WriteConcernOptions::AllConfigs,                                               NULL);            if ( !result.isOK() ) {                warning() << "could not create index tags_1_min_1: " << result.reason() << endl;                continue;            }            cursor = conn.query(TagsType::ConfigNS,                                QUERY(TagsType::ns(ns)).sort(TagsType::min()));            vector<TagRange> ranges;            while ( cursor->more() ) {                BSONObj tag = cursor->nextSafe();                TagRange tr(tag[TagsType::min()].Obj().getOwned(),                            tag[TagsType::max()].Obj().getOwned(),                            tag[TagsType::tag()].String());                ranges.push_back(tr);                uassert(16356,                        str::stream() << "tag ranges not valid for: " << ns,                        status.addTagRange(tr) );            }            cursor.reset();            DBConfigPtr cfg = grid.getDBConfig( ns );            if ( !cfg ) {                warning() << "could not load db config to balance " << ns << " collection" << endl;                continue;            }            // This line reloads the chunk manager once if this process doesn't know the collection            // is sharded yet.            ChunkManagerPtr cm = cfg->getChunkManagerIfExists( ns, true );            if ( !cm ) {                warning() << "could not load chunks to balance " << ns << " collection" << endl;                continue;            }            // loop through tags to make sure no chunk spans tags; splits on tag min. for all chunks            bool didAnySplits = false;            for ( unsigned i = 0; i < ranges.size(); i++ ) {                BSONObj min = ranges[i].min;                min = cm->getShardKey().extendRangeBound( min, false );                if ( allChunkMinimums.count( min ) > 0 )                    continue;                didAnySplits = true;                log() << "ns: " << ns << " need to split on "                      << min << " because there is a range there" << endl;                ChunkPtr c = cm->findIntersectingChunk( min );                vector<BSONObj> splitPoints;                splitPoints.push_back( min );                Status status = c->multiSplit( splitPoints );                if ( !status.isOK() ) {                    error() << "split failed: " << status << endl;                }                else {                    LOG(1) << "split worked" << endl;                }                break;            }            if ( didAnySplits ) {                // state change, just wait till next round                continue;            }            CandidateChunk* p = _policy->balance( ns, status, _balancedLastTime );            if ( p ) candidateChunks->push_back( CandidateChunkPtr( p ) );        }    }

intrusive_ptr<DocumentSource> DocumentSourceSort::createFromBson(    BSONElement elem, const intrusive_ptr<ExpressionContext>& pExpCtx) {    uassert(15973, "the $sort key specification must be an object", elem.type() == Object);    return create(pExpCtx, elem.embeddedObject());}

        bool run(const string& dbname, BSONObj& cmdObj, int, string& errmsg, BSONObjBuilder& result, bool fromRepl) {            const string ns = dbname + "." + cmdObj.firstElement().valuestr();            if (!cmdObj["start"].eoo()) {                errmsg = "using deprecated 'start' argument to geoNear";                return false;            }            Client::ReadContext ctx(ns);            Database* db = cc().database();            if ( !db ) {                errmsg = "can't find ns";                return false;            }            Collection* collection = db->getCollection( ns );            if ( !collection ) {                errmsg = "can't find ns";                return false;            }            IndexCatalog* indexCatalog = collection->getIndexCatalog();            // cout << "raw cmd " << cmdObj.toString() << endl;            // We seek to populate this.            string nearFieldName;            bool using2DIndex = false;            if (!getFieldName(collection, indexCatalog, &nearFieldName, &errmsg, &using2DIndex)) {                return false;            }            uassert(17304, "'near' field must be point",                    !cmdObj["near"].eoo() && cmdObj["near"].isABSONObj()                    && GeoParser::isPoint(cmdObj["near"].Obj()));            bool isSpherical = cmdObj["spherical"].trueValue();            if (!using2DIndex) {                uassert(17301, "2dsphere index must have spherical: true", isSpherical);            }            // Build the $near expression for the query.            BSONObjBuilder nearBob;            if (isSpherical) {                nearBob.append("$nearSphere", cmdObj["near"].Obj());            }            else {                nearBob.append("$near", cmdObj["near"].Obj());            }            if (!cmdObj["maxDistance"].eoo()) {                uassert(17299, "maxDistance must be a number",cmdObj["maxDistance"].isNumber());                nearBob.append("$maxDistance", cmdObj["maxDistance"].number());            }            if (!cmdObj["minDistance"].eoo()) {                uassert(17298, "minDistance doesn't work on 2d index", !using2DIndex);                uassert(17300, "minDistance must be a number",cmdObj["minDistance"].isNumber());                nearBob.append("$minDistance", cmdObj["minDistance"].number());            }            if (!cmdObj["uniqueDocs"].eoo()) {                warning() << ns << ": ignoring deprecated uniqueDocs option in geoNear command";            }            // And, build the full query expression.            BSONObjBuilder queryBob;            queryBob.append(nearFieldName, nearBob.obj());            if (!cmdObj["query"].eoo() && cmdObj["query"].isABSONObj()) {                queryBob.appendElements(cmdObj["query"].Obj());            }            BSONObj rewritten = queryBob.obj();            // cout << "rewritten query: " << rewritten.toString() << endl;            int numWanted = 100;            const char* limitName = !cmdObj["num"].eoo() ? "num" : "limit";            BSONElement eNumWanted = cmdObj[limitName];            if (!eNumWanted.eoo()) {                uassert(17303, "limit must be number", eNumWanted.isNumber());                numWanted = eNumWanted.numberInt();                uassert(17302, "limit must be >=0", numWanted >= 0);            }            bool includeLocs = false;            if (!cmdObj["includeLocs"].eoo()) {                includeLocs = cmdObj["includeLocs"].trueValue();            }            double distanceMultiplier = 1.0;            BSONElement eDistanceMultiplier = cmdObj["distanceMultiplier"];            if (!eDistanceMultiplier.eoo()) {                uassert(17296, "distanceMultiplier must be a number", eDistanceMultiplier.isNumber());                distanceMultiplier = eDistanceMultiplier.number();                uassert(17297, "distanceMultiplier must be non-negative", distanceMultiplier >= 0);            }            BSONObj projObj = BSON("$pt" << BSON("$meta" << LiteParsedQuery::metaGeoNearPoint) <<                                   "$dis" << BSON("$meta" << LiteParsedQuery::metaGeoNearDistance));//.........这里部分代码省略.........

 void GridFile::_exists() const {     uassert( 10015 ,  "doesn't exists" , exists() ); }

    // @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();//.........这里部分代码省略.........

 static void _logOpUninitialized(const char *opstr, const char *ns, const char *logNS, const BSONObj& obj, BSONObj *o2, bool *bb ) {     uassert(13288, "replSet error write op to db before replSet initialized", str::startsWith(ns, "local.") || *opstr == 'n'); }

 void SyncClusterConnection::insert( const string &ns, const vector< BSONObj >& v , int flags) {     uassert( 10023 , "SyncClusterConnection bulk insert not implemented" , 0); }

        void JSReducer::_reduce( const BSONList& tuples , BSONObj& key , int& endSizeEstimate ) {            uassert( 10074 ,  "need values" , tuples.size() );            int sizeEstimate = ( tuples.size() * tuples.begin()->getField( "value" ).size() ) + 128;            BSONObjBuilder reduceArgs( sizeEstimate );            boost::scoped_ptr<BSONArrayBuilder>  valueBuilder;            int sizeSoFar = 0;            unsigned n = 0;            for ( ; n<tuples.size(); n++ ) {                BSONObjIterator j(tuples[n]);                BSONElement keyE = j.next();                if ( n == 0 ) {                    reduceArgs.append( keyE );                    key = keyE.wrap();                    sizeSoFar = 5 + keyE.size();                    valueBuilder.reset(new BSONArrayBuilder( reduceArgs.subarrayStart( "tuples" ) ));                }                BSONElement ee = j.next();                uassert( 13070 , "value to large to reduce" , ee.size() < ( BSONObjMaxUserSize / 2 ) );                if ( sizeSoFar + ee.size() > BSONObjMaxUserSize ) {                    assert( n > 1 ); // if not, inf. loop                    break;                }                valueBuilder->append( ee );                sizeSoFar += ee.size();            }            assert(valueBuilder);            valueBuilder->done();            BSONObj args = reduceArgs.obj();            Scope * s = _func.scope();            s->invokeSafe( _func.func() , args );            if ( s->type( "return" ) == Array ) {                uasserted( 10075 , "reduce -> multiple not supported yet");                return;            }            endSizeEstimate = key.objsize() + ( args.objsize() / tuples.size() );            if ( n == tuples.size() )                return;            // the input list was too large            BSONList x;            for ( ; n < tuples.size(); n++ ) {                x.push_back( tuples[n] );            }            BSONObjBuilder temp( endSizeEstimate );            temp.append( key.firstElement() );            s->append( temp , "1" , "return" );            x.push_back( temp.obj() );            _reduce( x , key , endSizeEstimate );        }

        virtual bool run(const char *ns, BSONObj& cmdObj, string& errmsg, BSONObjBuilder& result, bool fromRepl) {            string source = cmdObj.getStringField( name.c_str() );            string target = cmdObj.getStringField( "to" );            if ( source.empty() || target.empty() ) {                errmsg = "invalid command syntax";                return false;            }                        setClient( source.c_str() );            NamespaceDetails *nsd = nsdetails( source.c_str() );            uassert( "source namespace does not exist", nsd );            bool capped = nsd->capped;            long long size = 0;            if ( capped )                for( DiskLoc i = nsd->firstExtent; !i.isNull(); i = i.ext()->xnext )                    size += i.ext()->length;                        setClient( target.c_str() );            uassert( "target namespace exists", !nsdetails( target.c_str() ) );            {                char from[256];                nsToClient( source.c_str(), from );                char to[256];                nsToClient( target.c_str(), to );                if ( strcmp( from, to ) == 0 ) {                    renameNamespace( source.c_str(), target.c_str() );                    return true;                }            }            BSONObjBuilder spec;            if ( capped ) {                spec.appendBool( "capped", true );                spec.append( "size", double( size ) );            }            if ( !userCreateNS( target.c_str(), spec.done(), errmsg, false ) )                return false;                        auto_ptr< DBClientCursor > c;            DBDirectClient bridge;            {                c = bridge.query( source, BSONObj() );            }            while( 1 ) {                {                    if ( !c->more() )                        break;                }                BSONObj o = c->next();                theDataFileMgr.insert( target.c_str(), o );            }                        char cl[256];            nsToClient( source.c_str(), cl );            string sourceIndexes = string( cl ) + ".system.indexes";            nsToClient( target.c_str(), cl );            string targetIndexes = string( cl ) + ".system.indexes";            {                c = bridge.query( sourceIndexes, QUERY( "ns" << source ) );            }            while( 1 ) {                {                    if ( !c->more() )                        break;                }                BSONObj o = c->next();                BSONObjBuilder b;                BSONObjIterator i( o );                while( i.moreWithEOO() ) {                    BSONElement e = i.next();                    if ( e.eoo() )                        break;                    if ( strcmp( e.fieldName(), "ns" ) == 0 ) {                        b.append( "ns", target );                    } else {                        b.append( e );                    }                }                BSONObj n = b.done();                theDataFileMgr.insert( targetIndexes.c_str(), n );            }            setClientTempNs( source.c_str() );            dropCollection( source, errmsg, result );            return true;        }

 BSONObj SerialServerClusteredCursor::next(){     uassert( 10018 ,  "no more items" , more() );     return _current.next(); }

void WireProtocolWriter::write(const StringData& ns,                               const std::vector<WriteOperation*>& write_operations,                               bool ordered,                               bool bypassDocumentValidation,                               const WriteConcern* writeConcern,                               WriteResult* writeResult) {    if (_client->getMaxWireVersion() >= 4) {        // Per DRIVERS-250:        // If your driver sends unacknowledged writes using op codes (OP_INSERT, OP_UPDATE,        // OP_DELETE), you MUST raise an error when bypassDocumentValidation is explicitly set by a        // user on >= 3.2 servers.        //        uassert(0,                "bypassDocumentValidation is not supported for unacknowledged writes with MongoDB "                "3.2 and later.",                !bypassDocumentValidation);    }    // Effectively a map of batch relative indexes to WriteOperations    std::vector<WriteOperation*> batchOps;    BufBuilder builder;    std::vector<WriteOperation*>::const_iterator batch_begin = write_operations.begin();    const std::vector<WriteOperation*>::const_iterator end = write_operations.end();    while (batch_begin != end) {        std::vector<WriteOperation*>::const_iterator batch_iter = batch_begin;        // We must be able to fit the first item of the batch. Otherwise, the calling code        // passed an over size write operation in violation of our contract.        invariant(_fits(&builder, *batch_iter));        // Set the current operation type for this batch        const WriteOpType batchOpType = (*batch_iter)->operationType();        // Begin the command for this batch.        (*batch_iter)->startRequest(ns.toString(), ordered, &builder);        while (true) {            // Always safe to append here: either we just entered the loop, or all the            // below checks passed.            (*batch_iter)->appendSelfToRequest(&builder);            // Associate batch index with WriteOperation            batchOps.push_back(*batch_iter);            // If the operation we just queued isn't batchable, issue what we have.            if (!_batchableRequest(batchOpType, writeResult))                break;            // Peek at the next operation.            const std::vector<WriteOperation*>::const_iterator next = boost::next(batch_iter);            // If we are out of operations, issue what we have.            if (next == end)                break;            // If the next operation is of a different type, issue what we have.            if ((*next)->operationType() != batchOpType)                break;            // If adding the next op would put us over the limit of ops in a batch, issue            // what we have.            if (std::distance(batch_begin, next) >= _client->getMaxWriteBatchSize())                break;            // If we can't put the next item into the current batch, issue what we have.            if (!_fits(&builder, *next))                break;            // OK to proceed to next op            batch_iter = next;        }        // Issue the complete command.        BSONObj batchResult = _send(batchOpType, builder, writeConcern, ns);        // Merge this batch's result into the result for all batches written.        writeResult->_mergeGleResult(batchOps, batchResult);        batchOps.clear();        // Check write result for errors if we are doing ordered processing or last op        bool lastOp = *batch_iter == write_operations.back();        if (ordered || lastOp)            writeResult->_check(lastOp);        // Reset the builder so we can build the next request.        builder.reset();        // The next batch begins with the op after the last one in the just issued batch.        batch_begin = ++batch_iter;    }}

    UpdateResult update(            OperationContext* txn,            Database* db,            const UpdateRequest& request,            OpDebug* opDebug,            UpdateDriver* driver,            CanonicalQuery* cq) {        LOG(3) << "processing update : " << request;        std::auto_ptr<CanonicalQuery> cqHolder(cq);        const NamespaceString& nsString = request.getNamespaceString();        UpdateLifecycle* lifecycle = request.getLifecycle();        Collection* collection = db->getCollection(nsString.ns());        validateUpdate(nsString.ns().c_str(), request.getUpdates(), request.getQuery());        // TODO: This seems a bit circuitious.        opDebug->updateobj = request.getUpdates();        if (lifecycle) {            lifecycle->setCollection(collection);            driver->refreshIndexKeys(lifecycle->getIndexKeys());        }        Runner* rawRunner;        Status status = cq ?            getRunner(collection, cqHolder.release(), &rawRunner) :            getRunner(collection, nsString.ns(), request.getQuery(), &rawRunner, &cq);        uassert(17243,                "could not get runner " + request.getQuery().toString() + "; " + causedBy(status),                status.isOK());        // Create the runner and setup all deps.        auto_ptr<Runner> runner(rawRunner);        // Register Runner with ClientCursor        const ScopedRunnerRegistration safety(runner.get());        //        // We'll start assuming we have one or more documents for this update. (Otherwise,        // we'll fall-back to insert case (if upsert is true).)        //        // We are an update until we fall into the insert case below.        driver->setContext(ModifierInterface::ExecInfo::UPDATE_CONTEXT);        int numMatched = 0;        // If the update was in-place, we may see it again.  This only matters if we're doing        // a multi-update; if we're not doing a multi-update we stop after one update and we        // won't see any more docs.        //        // For example: If we're scanning an index {x:1} and performing {$inc:{x:5}}, we'll keep        // moving the document forward and it will continue to reappear in our index scan.        // Unless the index is multikey, the underlying query machinery won't de-dup.        //        // If the update wasn't in-place we may see it again.  Our query may return the new        // document and we wouldn't want to update that.        //        // So, no matter what, we keep track of where the doc wound up.        typedef unordered_set<DiskLoc, DiskLoc::Hasher> DiskLocSet;        const scoped_ptr<DiskLocSet> updatedLocs(request.isMulti() ? new DiskLocSet : NULL);        // Reset these counters on each call. We might re-enter this function to retry this        // update if we throw a page fault exception below, and we rely on these counters        // reflecting only the actions taken locally. In particlar, we must have the no-op        // counter reset so that we can meaningfully comapre it with numMatched above.        opDebug->nscanned = 0;        opDebug->nscannedObjects = 0;        opDebug->nModified = 0;        // Get the cached document from the update driver.        mutablebson::Document& doc = driver->getDocument();        mutablebson::DamageVector damages;        // Used during iteration of docs        BSONObj oldObj;        // Get first doc, and location        Runner::RunnerState state = Runner::RUNNER_ADVANCED;        uassert(ErrorCodes::NotMaster,                mongoutils::str::stream() << "Not primary while updating " << nsString.ns(),                !request.shouldCallLogOp() || isMasterNs(nsString.ns().c_str()));        while (true) {            // Get next doc, and location            DiskLoc loc;            state = runner->getNext(&oldObj, &loc);            if (state != Runner::RUNNER_ADVANCED) {                if (state == Runner::RUNNER_EOF) {                    // We have reached the logical end of the loop, so do yielding recovery                    break;                }                else {                    uassertStatusOK(Status(ErrorCodes::InternalError,//.........这里部分代码省略.........

Status ClusterAggregate::runAggregate(OperationContext* txn,                                      const Namespaces& namespaces,                                      BSONObj cmdObj,                                      int options,                                      BSONObjBuilder* result) {    auto scopedShardDbStatus = ScopedShardDatabase::getExisting(txn, namespaces.executionNss.db());    if (!scopedShardDbStatus.isOK()) {        appendEmptyResultSet(            *result, scopedShardDbStatus.getStatus(), namespaces.requestedNss.ns());        return Status::OK();    }    auto request = AggregationRequest::parseFromBSON(namespaces.executionNss, cmdObj);    if (!request.isOK()) {        return request.getStatus();    }    const auto conf = scopedShardDbStatus.getValue().db();    // Determine the appropriate collation and 'resolve' involved namespaces to make the    // ExpressionContext.    // We won't try to execute anything on a mongos, but we still have to populate this map so that    // any $lookups, etc. will be able to have a resolved view definition. It's okay that this is    // incorrect, we will repopulate the real resolved namespace map on the mongod. Note that we    // need to check if any involved collections are sharded before forwarding an aggregation    // command on an unsharded collection.    StringMap<ExpressionContext::ResolvedNamespace> resolvedNamespaces;    LiteParsedPipeline liteParsedPipeline(request.getValue());    for (auto&& ns : liteParsedPipeline.getInvolvedNamespaces()) {        uassert(28769, str::stream() << ns.ns() << " cannot be sharded", !conf->isSharded(ns.ns()));        resolvedNamespaces[ns.coll()] = {ns, std::vector<BSONObj>{}};    }    if (!conf->isSharded(namespaces.executionNss.ns())) {        return aggPassthrough(txn, namespaces, conf, cmdObj, result, options);    }    auto chunkMgr = conf->getChunkManager(txn, namespaces.executionNss.ns());    std::unique_ptr<CollatorInterface> collation;    if (!request.getValue().getCollation().isEmpty()) {        collation = uassertStatusOK(CollatorFactoryInterface::get(txn->getServiceContext())                                        ->makeFromBSON(request.getValue().getCollation()));    } else if (chunkMgr->getDefaultCollator()) {        collation = chunkMgr->getDefaultCollator()->clone();    }    boost::intrusive_ptr<ExpressionContext> mergeCtx = new ExpressionContext(        txn, request.getValue(), std::move(collation), std::move(resolvedNamespaces));    mergeCtx->inRouter = true;    // explicitly *not* setting mergeCtx->tempDir    // Parse and optimize the pipeline specification.    auto pipeline = Pipeline::parse(request.getValue().getPipeline(), mergeCtx);    if (!pipeline.isOK()) {        return pipeline.getStatus();    }    pipeline.getValue()->optimizePipeline();    // If the first $match stage is an exact match on the shard key (with a simple collation or    // no string matching), we only have to send it to one shard, so send the command to that    // shard.    BSONObj firstMatchQuery = pipeline.getValue()->getInitialQuery();    BSONObj shardKeyMatches;    shardKeyMatches = uassertStatusOK(        chunkMgr->getShardKeyPattern().extractShardKeyFromQuery(txn, firstMatchQuery));    bool singleShard = false;    if (!shardKeyMatches.isEmpty()) {        auto chunk = chunkMgr->findIntersectingChunk(            txn, shardKeyMatches, request.getValue().getCollation());        if (chunk.isOK()) {            singleShard = true;        }    }    // Don't need to split pipeline if the first $match is an exact match on shard key, unless    // there is a stage that needs to be run on the primary shard.    const bool needPrimaryShardMerger = pipeline.getValue()->needsPrimaryShardMerger();    const bool needSplit = !singleShard || needPrimaryShardMerger;    // Split the pipeline into pieces for mongod(s) and this mongos. If needSplit is true,    // 'pipeline' will become the merger side.    boost::intrusive_ptr<Pipeline> shardPipeline(needSplit ? pipeline.getValue()->splitForSharded()                                                           : pipeline.getValue());    // Create the command for the shards. The 'fromRouter' field means produce output to be    // merged.    MutableDocument commandBuilder(request.getValue().serializeToCommandObj());    commandBuilder[AggregationRequest::kPipelineName] = Value(shardPipeline->serialize());    if (needSplit) {        commandBuilder[AggregationRequest::kFromRouterName] = Value(true);        commandBuilder[AggregationRequest::kCursorName] =            Value(DOC(AggregationRequest::kBatchSizeName << 0));    }    // These fields are not part of the AggregationRequest since they are not handled by the    // aggregation subsystem, so we serialize them separately.    const std::initializer_list<StringData> fieldsToPropagateToShards = {//.........这里部分代码省略.........

OpMsg OpMsg::parse(const Message& message) try {    // TODO some validation may make more sense in the IDL parser. I've tagged them with comments.    OpMsg msg;    // Use a separate BufReader for the flags since the flags can change how much room we have    // for sections.    BufReader(message.singleData().data(), message.dataSize()).read(msg.flags);    uassert(40429,            str::stream() << "Message contains illegal flags value: " << msg.flags,            !containsUnknownRequiredFlags(msg.flags));    invariant(!msg.isFlagSet(kChecksumPresent));  // TODO SERVER-28679 check checksum here.    constexpr int kCrc32Size = 4;    const int checksumSize = msg.isFlagSet(kChecksumPresent) ? kCrc32Size : 0;    BufReader sectionsBuf(message.singleData().data() + sizeof(msg.flags),                          message.dataSize() - sizeof(msg.flags) - checksumSize);    bool haveBody = false;    while (!sectionsBuf.atEof()) {        const auto sectionKind = sectionsBuf.read<Section>();        switch (sectionKind) {            case Section::kBody: {                uassert(40430, "Multiple body sections in message", !haveBody);                haveBody = true;                msg.body = sectionsBuf.read<Validated<BSONObj>>();                break;            }            case Section::kDocSequence: {                // The first 4 bytes are the total size, including themselves.                const auto remainingSize = sectionsBuf.read<int32_t>() - sizeof(int32_t);                BufReader seqBuf(sectionsBuf.skip(remainingSize), remainingSize);                const auto name = seqBuf.readCStr();                uassert(40431,                        str::stream() << "Duplicate document sequence: " << name,                        !msg.getSequence(name));  // TODO IDL                msg.sequences.push_back({name.toString()});                while (!seqBuf.atEof()) {                    msg.sequences.back().objs.push_back(seqBuf.read<Validated<BSONObj>>());                }                break;            }            default:                // Using uint32_t so we append as a decimal number rather than as a char.                uasserted(40432, str::stream() << "Unknown section kind " << uint32_t(sectionKind));        }    }    // Detect duplicates between doc sequences and body. TODO IDL    // Technically this is O(N*M) but N is at most 2.    for (const auto& docSeq : msg.sequences) {        const char* name = docSeq.name.c_str();  // Pointer is redirected by next call.        auto inBody =            !dotted_path_support::extractElementAtPathOrArrayAlongPath(msg.body, name).eoo();        uassert(40433,                str::stream() << "Duplicate field between body and document sequence "                              << docSeq.name,                !inBody);    }    return msg;} catch (const DBException& ex) {    // TODO change to LOG(1).    log() << "invalid message: " << redact(ex) << ' '          << hexdump(message.singleData().view2ptr(), message.size());    throw;}

bool MessagingPort::recv(Message& m) {    try {#ifdef MONGO_CONFIG_SSL    again:#endif        // mmm( log() << "*  recv() sock:" << this->sock << endl; )        MSGHEADER::Value header;        int headerLen = sizeof(MSGHEADER::Value);        psock->recv((char*)&header, headerLen);        int len = header.constView().getMessageLength();        if (len == 542393671) {            // an http GET            string msg =                "It looks like you are trying to access MongoDB over HTTP on the native driver "                "port./n";            LOG(psock->getLogLevel()) << msg;            std::stringstream ss;            ss << "HTTP/1.0 200 OK/r/nConnection: close/r/nContent-Type: "                  "text/plain/r/nContent-Length: " << msg.size() << "/r/n/r/n" << msg;            string s = ss.str();            send(s.c_str(), s.size(), "http");            return false;        }        // If responseTo is not 0 or -1 for first packet assume SSL        else if (psock->isAwaitingHandshake()) {#ifndef MONGO_CONFIG_SSL            if (header.constView().getResponseTo() != 0 &&                header.constView().getResponseTo() != -1) {                uasserted(17133,                          "SSL handshake requested, SSL feature not available in this build");            }#else            if (header.constView().getResponseTo() != 0 &&                header.constView().getResponseTo() != -1) {                uassert(17132,                        "SSL handshake received but server is started without SSL support",                        sslGlobalParams.sslMode.load() != SSLParams::SSLMode_disabled);                setX509SubjectName(                    psock->doSSLHandshake(reinterpret_cast<const char*>(&header), sizeof(header)));                psock->setHandshakeReceived();                goto again;            }            uassert(17189,                    "The server is configured to only allow SSL connections",                    sslGlobalParams.sslMode.load() != SSLParams::SSLMode_requireSSL);#endif  // MONGO_CONFIG_SSL        }        if (static_cast<size_t>(len) < sizeof(MSGHEADER::Value) ||            static_cast<size_t>(len) > MaxMessageSizeBytes) {            LOG(0) << "recv(): message len " << len << " is invalid. "                   << "Min " << sizeof(MSGHEADER::Value) << " Max: " << MaxMessageSizeBytes;            return false;        }        psock->setHandshakeReceived();        int z = (len + 1023) & 0xfffffc00;        verify(z >= len);        MsgData::View md = reinterpret_cast<char*>(mongoMalloc(z));        ScopeGuard guard = MakeGuard(free, md.view2ptr());        verify(md.view2ptr());        memcpy(md.view2ptr(), &header, headerLen);        int left = len - headerLen;        psock->recv(md.data(), left);        guard.Dismiss();        m.setData(md.view2ptr(), true);        return true;    } catch (const SocketException& e) {        logger::LogSeverity severity = psock->getLogLevel();        if (!e.shouldPrint())            severity = severity.lessSevere();        LOG(severity) << "SocketException: remote: " << remote() << " error: " << e;        m.reset();        return false;    }}

 // Technically lat/long bounds, not really tied to earth radius. void checkEarthBounds(const Point &p) {     uassert(14808, str::stream() << "point " << p.toString()                                  << " must be in earth-like bounds of long "                                  << ": [-180, 180], lat : [-90, 90] ",             p.x >= -180 && p.x <= 180 && p.y >= -90 && p.y <= 90); }

    long long BSONTool::processFile( const boost::filesystem::path& root ) {        _fileName = root.string();        unsigned long long fileLength = file_size( root );        if ( fileLength == 0 ) {            out() << "file " << _fileName << " empty, skipping" << endl;            return 0;        }        FILE* file = fopen( _fileName.c_str() , "rb" );        if ( ! file ) {            log() << "error opening file: " << _fileName << " " << errnoWithDescription() << endl;            return 0;        }#if !defined(__sunos__) && defined(POSIX_FADV_SEQUENTIAL)        posix_fadvise(fileno(file), 0, fileLength, POSIX_FADV_SEQUENTIAL);#endif        log(1) << "/t file size: " << fileLength << endl;        unsigned long long read = 0;        unsigned long long num = 0;        unsigned long long processed = 0;        const int BUF_SIZE = BSONObjMaxUserSize + ( 1024 * 1024 );        boost::scoped_array<char> buf_holder(new char[BUF_SIZE]);        char * buf = buf_holder.get();        ProgressMeter m( fileLength );        m.setUnits( "bytes" );        while ( read < fileLength ) {            size_t amt = fread(buf, 1, 4, file);            verify( amt == 4 );            int size = little<int>::ref( buf );            uassert( 10264 , str::stream() << "invalid object size: " << size , size < BUF_SIZE );            amt = fread(buf+4, 1, size-4, file);            verify( amt == (size_t)( size - 4 ) );            BSONObj o( buf );            if ( _objcheck && ! o.valid() ) {                cerr << "INVALID OBJECT - going try and pring out " << endl;                cerr << "size: " << size << endl;                BSONObjIterator i(o);                while ( i.more() ) {                    BSONElement e = i.next();                    try {                        e.validate();                    }                    catch ( ... ) {                        cerr << "/t/t NEXT ONE IS INVALID" << endl;                    }                    cerr << "/t name : " << e.fieldName() << " " << e.type() << endl;                    cerr << "/t " << e << endl;                }            }            if ( _matcher.get() == 0 || _matcher->matches( o ) ) {                gotObject( o );                processed++;            }            read += o.objsize();            num++;            m.hit( o.objsize() );        }        fclose( file );        uassert( 10265 ,  "counts don't match" , m.done() == fileLength );        (_usesstdout ? cout : cerr ) << m.hits() << " objects found" << endl;        if ( _matcher.get() )            (_usesstdout ? cout : cerr ) << processed << " objects processed" << endl;        return processed;    }

    int HttpClient::_go( const char * command , string url , const char * body , Result * result ) {        bool ssl = false;        if ( url.find( "https://" ) == 0 ) {            ssl = true;            url = url.substr( 8 );        }        else {            uassert( 10271 ,  "invalid url" , url.find( "http://" ) == 0 );            url = url.substr( 7 );        }        string host , path;        if ( url.find( "/" ) == string::npos ) {            host = url;            path = "/";        }        else {            host = url.substr( 0 , url.find( "/" ) );            path = url.substr( url.find( "/" ) );        }        HD( "host [" << host << "]" );        HD( "path [" << path << "]" );        string server = host;        int port = ssl ? 443 : 80;        string::size_type idx = host.find( ":" );        if ( idx != string::npos ) {            server = host.substr( 0 , idx );            string t = host.substr( idx + 1 );            port = atoi( t.c_str() );        }        HD( "server [" << server << "]" );        HD( "port [" << port << "]" );        string req;        {            stringstream ss;            ss << command << " " << path << " HTTP/1.1/r/n";            ss << "Host: " << host << "/r/n";            ss << "Connection: Close/r/n";            ss << "User-Agent: mongodb http client/r/n";            if ( body ) {                ss << "Content-Length: " << strlen( body ) << "/r/n";            }            ss << "/r/n";            if ( body ) {                ss << body;            }            req = ss.str();        }        SockAddr addr( server.c_str() , port );        uassert( 15000 ,  "server socket addr is invalid" , addr.isValid() );        HD( "addr: " << addr.toString() );        Socket sock;        if ( ! sock.connect( addr ) )            return -1;                if ( ssl ) {#ifdef MONGO_SSL            // pointer to global singleton instance            SSLManagerInterface* mgr = getSSLManager();            sock.secure(mgr, "");#else            uasserted( 15862 , "no ssl support" );#endif        }        {            const char * out = req.c_str();            int toSend = req.size();            sock.send( out , toSend, "_go" );        }        char buf[4097];        int got = sock.unsafe_recv( buf , 4096 );        buf[got] = 0;        int rc;        char version[32];        verify( sscanf( buf , "%s %d" , version , &rc ) == 2 );        HD( "rc: " << rc );        StringBuilder sb;        if ( result )            sb << buf;        // SERVER-8864, unsafe_recv will throw when recv returns 0 indicating closed socket.        try {            while ( ( got = sock.unsafe_recv( buf , 4096 ) ) > 0) {                buf[got] = 0;                if ( result )                    sb << buf;//.........这里部分代码省略.........

 int Scope::invoke(const char* code, const BSONObj* args, const BSONObj* recv, int timeoutMs) {     ScriptingFunction func = createFunction(code);     uassert(10207,  "compile failed", func);     return invoke(func, args, recv, timeoutMs); }

 void Scope::validateObjectIdString(const string& str) {     uassert(10448, "invalid object id: length", str.size() == 24);     for (size_t i = 0; i < str.size(); i++)         uassert(10430,  "invalid object id: not hex", std::isxdigit(str.at(i))); }

    void Balancer::run() {        // this is the body of a BackgroundJob so if we throw here we're basically ending the balancer thread prematurely        while ( ! inShutdown() ) {            if ( ! _init() ) {                log() << "will retry to initialize balancer in one minute" << endl;                sleepsecs( 60 );                continue;            }            break;        }        int sleepTime = 10;        // getConnectioString and dist lock constructor does not throw, which is what we expect on while        // on the balancer thread        ConnectionString config = configServer.getConnectionString();        DistributedLock balanceLock( config , "balancer" );        while ( ! inShutdown() ) {            try {                ScopedDbConnection conn(config.toString(), 30);                // ping has to be first so we keep things in the config server in sync                _ping();                // use fresh shard state                Shard::reloadShardInfo();                // refresh chunk size (even though another balancer might be active)                Chunk::refreshChunkSize();                SettingsType balancerConfig;                string errMsg;                if (!grid.getBalancerSettings(&balancerConfig, &errMsg)) {                    warning() << errMsg;                    return ;                }                // now make sure we should even be running                if (balancerConfig.isKeySet() && // balancer config doc exists                        !grid.shouldBalance(balancerConfig)) {                    LOG(1) << "skipping balancing round because balancing is disabled" << endl;                    // Ping again so scripts can determine if we're active without waiting                    _ping( true );                    conn.done();                    sleepsecs( sleepTime );                    continue;                }                uassert( 13258 , "oids broken after resetting!" , _checkOIDs() );                {                    dist_lock_try lk( &balanceLock , "doing balance round" );                    if ( ! lk.got() ) {                        LOG(1) << "skipping balancing round because another balancer is active" << endl;                        // Ping again so scripts can determine if we're active without waiting                        _ping( true );                        conn.done();                                                sleepsecs( sleepTime ); // no need to wake up soon                        continue;                    }                    if ( !isConfigServerConsistent() ) {                        conn.done();                        warning() << "Skipping balancing round because data inconsistency"                                  << " was detected amongst the config servers." << endl;                        sleepsecs( sleepTime );                        continue;                    }                    const bool waitForDelete = (balancerConfig.isWaitForDeleteSet() ?                            balancerConfig.getWaitForDelete() : false);                    scoped_ptr<WriteConcernOptions> writeConcern;                    if (balancerConfig.isKeySet()) { // if balancer doc exists.                        StatusWith<WriteConcernOptions*> extractStatus =                                balancerConfig.extractWriteConcern();                        if (extractStatus.isOK()) {                            writeConcern.reset(extractStatus.getValue());                        }                        else {                            warning() << extractStatus.toString();                        }                    }                    LOG(1) << "*** start balancing round. "                           << "waitForDelete: " << waitForDelete//.........这里部分代码省略.........

    void _initAndListen(int listenPort ) {        Client::initThread("initandlisten");        bool is32bit = sizeof(int*) == 4;        {            ProcessId pid = ProcessId::getCurrent();            LogstreamBuilder l = log();            l << "MongoDB starting : pid=" << pid              << " port=" << serverGlobalParams.port              << " dbpath=" << storageGlobalParams.dbpath;            if( replSettings.master ) l << " master=" << replSettings.master;            if( replSettings.slave )  l << " slave=" << (int) replSettings.slave;            l << ( is32bit ? " 32" : " 64" ) << "-bit host=" << getHostNameCached() << endl;        }        DEV log() << "_DEBUG build (which is slower)" << endl;        logStartupWarnings();#if defined(_WIN32)        printTargetMinOS();#endif        logProcessDetails();        {            stringstream ss;            ss << endl;            ss << "*********************************************************************" << endl;            ss << " ERROR: dbpath (" << storageGlobalParams.dbpath << ") does not exist." << endl;            ss << " Create this directory or give existing directory in --dbpath." << endl;            ss << " See http://dochub.mongodb.org/core/startingandstoppingmongo" << endl;            ss << "*********************************************************************" << endl;            uassert(10296,  ss.str().c_str(), boost::filesystem::exists(storageGlobalParams.dbpath));        }        {            stringstream ss;            ss << "repairpath (" << storageGlobalParams.repairpath << ") does not exist";            uassert(12590,  ss.str().c_str(),                    boost::filesystem::exists(storageGlobalParams.repairpath));        }        // TODO check non-journal subdirs if using directory-per-db        checkReadAhead(storageGlobalParams.dbpath);        acquirePathLock(mongodGlobalParams.repair);        boost::filesystem::remove_all(storageGlobalParams.dbpath + "/_tmp/");        FileAllocator::get()->start();        // TODO:  This should go into a MONGO_INITIALIZER once we have figured out the correct        // dependencies.        if (snmpInit) {            snmpInit();        }        MONGO_ASSERT_ON_EXCEPTION_WITH_MSG( clearTmpFiles(), "clear tmp files" );        dur::startup();        if (storageGlobalParams.durOptions & StorageGlobalParams::DurRecoverOnly)            return;        unsigned long long missingRepl = checkIfReplMissingFromCommandLine();        if (missingRepl) {            log() << startupWarningsLog;            log() << "** WARNING: mongod started without --replSet yet " << missingRepl                  << " documents are present in local.system.replset" << startupWarningsLog;            log() << "**          Restart with --replSet unless you are doing maintenance and no"                  << " other clients are connected." << startupWarningsLog;            log() << "**          The TTL collection monitor will not start because of this." << startupWarningsLog;            log() << "**          For more info see http://dochub.mongodb.org/core/ttlcollections" << startupWarningsLog;            log() << startupWarningsLog;        }        if (mongodGlobalParams.scriptingEnabled) {            ScriptEngine::setup();            globalScriptEngine->setCheckInterruptCallback( jsInterruptCallback );            globalScriptEngine->setGetCurrentOpIdCallback( jsGetCurrentOpIdCallback );        }        // On replica set members we only clear temp collections on DBs other than "local" during        // promotion to primary. On pure slaves, they are only cleared when the oplog tells them to.        // The local DB is special because it is not replicated.  See SERVER-10927 for more details.        const bool shouldClearNonLocalTmpCollections = !(missingRepl                                                         || replSettings.usingReplSets()                                                         || replSettings.slave == SimpleSlave);        repairDatabasesAndCheckVersion(shouldClearNonLocalTmpCollections);        if (mongodGlobalParams.upgrade)            return;        uassertStatusOK(getGlobalAuthorizationManager()->initialize());        /* this is for security on certain platforms (nonce generation) */        srand((unsigned) (curTimeMicros() ^ startupSrandTimer.micros()));        // The snapshot thread provides historical collection level and lock statistics for use        // by the web interface. Only needed when HTTP is enabled.        if (serverGlobalParams.isHttpInterfaceEnabled)            snapshotThread.go();        d.clientCursorMonitor.go();//.........这里部分代码省略.........

    void Grid::removeDB( const std::string& database ) {        uassert( 10186 ,  "removeDB expects db name" , database.find( '.' ) == string::npos );        scoped_lock l( _lock );        _databases.erase( database );    }