void BatchSafeWriter::safeWriteBatch( DBClientBase* conn,
                                      const BatchedCommandRequest& request,
                                      BatchedCommandResponse* response ) {

    // N starts at zero, and we add to it for each item
    response->setN( 0 );

    for ( size_t i = 0; i < request.sizeWriteOps(); ++i ) {

        BatchItemRef itemRef( &request, static_cast<int>( i ) );
        LastError lastError;

        _safeWriter->safeWrite( conn, itemRef, &lastError );

        // Register the error if we need to
        WriteErrorDetail* batchError = lastErrorToBatchError( lastError );
        if ( batchError ) {
            batchError->setIndex( i );
            response->addToErrDetails( batchError );
        }

        response->setN( response->getN() + lastError.nObjects );

        if ( !lastError.upsertedId.isEmpty() ) {
            BatchedUpsertDetail* upsertedId = new BatchedUpsertDetail;
            upsertedId->setIndex( i );
            upsertedId->setUpsertedID( lastError.upsertedId );
            response->addToUpsertDetails( upsertedId );
        }

        // Break on first error if we're ordered
        if ( request.getOrdered() && BatchSafeWriter::isFailedOp( lastError ) ) break;
    }

    if ( request.sizeWriteOps() == 1 && response->isErrDetailsSet()
            && !response->isErrCodeSet() ) {

        // Promote single error to batch error
        const WriteErrorDetail* error = response->getErrDetailsAt( 0 );
        response->setErrCode( error->getErrCode() );
        if ( error->isErrInfoSet() ) response->setErrInfo( error->getErrInfo() );
        response->setErrMessage( error->getErrMessage() );

        response->unsetErrDetails();
    }

    if ( request.sizeWriteOps() == 1 && response->isUpsertDetailsSet() ) {

        // Promote single upsert to batch upsert
        const BatchedUpsertDetail* upsertedId = response->getUpsertDetailsAt( 0 );
        response->setSingleUpserted( upsertedId->getUpsertedID() );

        response->unsetUpsertDetails();
    }

    response->setOk( !response->isErrCodeSet() );
    dassert( response->isValid( NULL ) );
}

    void WriteBatchExecutor::execInserts( const BatchedCommandRequest& request,
                                          std::vector<WriteErrorDetail*>* errors ) {

        // Theory of operation:
        //
        // Instantiates an ExecInsertsState, which represents all of the state involved in the batch
        // insert execution algorithm.  Most importantly, encapsulates the lock state.
        //
        // Every iteration of the loop in execInserts() processes one document insertion, by calling
        // insertOne() exactly once for a given value of state.currIndex.
        //
        // If the ExecInsertsState indicates that the requisite write locks are not held, insertOne
        // acquires them and performs lock-acquisition-time checks.  However, on non-error
        // execution, it does not release the locks.  Therefore, the yielding logic in the while
        // loop in execInserts() is solely responsible for lock release in the non-error case.
        //
        // Internally, insertOne loops performing the single insert until it completes without a
        // PageFaultException, or until it fails with some kind of error.  Errors are mostly
        // propagated via the request->error field, but DBExceptions or std::exceptions may escape,
        // particularly on operation interruption.  These kinds of errors necessarily prevent
        // further insertOne calls, and stop the batch.  As a result, the only expected source of
        // such exceptions are interruptions.
        ExecInsertsState state(&request);
        normalizeInserts(request, &state.normalizedInserts, &state.pregeneratedKeys);

        ElapsedTracker elapsedTracker(128, 10); // 128 hits or 10 ms, matching RunnerYieldPolicy's

        for (state.currIndex = 0;
             state.currIndex < state.request->sizeWriteOps();
             ++state.currIndex) {

            if (elapsedTracker.intervalHasElapsed()) {
                // Consider yielding between inserts.

                if (state.hasLock()) {
                    int micros = ClientCursor::suggestYieldMicros();
                    if (micros > 0) {
                        state.unlock();
                        killCurrentOp.checkForInterrupt();
                        sleepmicros(micros);
                    }
                }
                killCurrentOp.checkForInterrupt();
                elapsedTracker.resetLastTime();
            }

            WriteErrorDetail* error = NULL;
            execOneInsert(&state, &error);
            if (error) {
                errors->push_back(error);
                error->setIndex(state.currIndex);
                if (request.getOrdered())
                    return;
            }
        }
    }

    static WriteErrorDetail* toWriteError( const Status& status ) {

        WriteErrorDetail* error = new WriteErrorDetail;

        // TODO: Complex transform here?
        error->setErrCode( status.code() );
        error->setErrMessage( status.reason() );

        return error;
    }

static bool checkIsMasterForCollection(const NamespaceString& ns, WriteErrorDetail** error) {
    if (!isMasterNs(ns.ns().c_str())) {
        WriteErrorDetail* errorDetail = *error = new WriteErrorDetail;
        errorDetail->setErrCode(ErrorCodes::NotMaster);
        errorDetail->setErrMessage(std::string(mongoutils::str::stream() <<
                                               "Not primary while writing to " << ns.ns()));
        return false;
    }
    return true;
}

 static bool checkIsMasterForCollection(const std::string& ns, WriteOpResult* result) {
     if (!isMasterNs(ns.c_str())) {
         WriteErrorDetail* errorDetail = new WriteErrorDetail;
         result->setError(errorDetail);
         errorDetail->setErrCode(ErrorCodes::NotMaster);
         errorDetail->setErrMessage("Not primary while writing to " + ns);
         return false;
     }
     return true;
 }

 static bool checkIsMasterForDatabase(const std::string& ns, WriteOpResult* result) {
     if (!repl::getGlobalReplicationCoordinator()->canAcceptWritesForDatabase(
             NamespaceString(ns).db())) {
         WriteErrorDetail* errorDetail = new WriteErrorDetail;
         result->setError(errorDetail);
         errorDetail->setErrCode(ErrorCodes::NotMaster);
         errorDetail->setErrMessage("Not primary while writing to " + ns);
         return false;
     }
     return true;
 }

void WriteOp::setOpError(const WriteErrorDetail& error) {
    dassert(_state == WriteOpState_Ready);
    _error.reset(new WriteErrorDetail);
    error.cloneTo(_error.get());
    _error->setIndex(_itemRef.getItemIndex());
    _state = WriteOpState_Error;
    // No need to updateOpState, set directly
}

void WriteOp::noteWriteError(const TargetedWrite& targetedWrite, const WriteErrorDetail& error) {
    const WriteOpRef& ref = targetedWrite.writeOpRef;
    auto& childOp = _childOps[ref.second];

    childOp.pendingWrite = NULL;
    childOp.endpoint.reset(new ShardEndpoint(targetedWrite.endpoint));
    childOp.error.reset(new WriteErrorDetail);
    error.cloneTo(childOp.error.get());
    dassert(ref.first == _itemRef.getItemIndex());
    childOp.error->setIndex(_itemRef.getItemIndex());
    childOp.state = WriteOpState_Error;
    _updateOpState();
}

static void toWriteErrorResponse(const WriteErrorDetail& error,
                                 bool ordered,
                                 int numWrites,
                                 BatchedCommandResponse* writeErrResponse) {
    writeErrResponse->setOk(true);
    writeErrResponse->setN(0);

    int numErrors = ordered ? 1 : numWrites;
    for (int i = 0; i < numErrors; i++) {
        unique_ptr<WriteErrorDetail> errorClone(new WriteErrorDetail);
        error.cloneTo(errorClone.get());
        errorClone->setIndex(i);
        writeErrResponse->addToErrDetails(errorClone.release());
    }

    dassert(writeErrResponse->isValid(NULL));
}

WriteErrorDetail* BatchSafeWriter::lastErrorToBatchError( const LastError& lastError ) {

    bool isFailedOp = lastError.msg != "";
    bool isStaleOp = lastError.writebackId.isSet();
    dassert( !( isFailedOp && isStaleOp ) );

    if ( isFailedOp ) {
        WriteErrorDetail* batchError = new WriteErrorDetail;
        if ( lastError.code != 0 ) batchError->setErrCode( lastError.code );
        else batchError->setErrCode( ErrorCodes::UnknownError );
        batchError->setErrMessage( lastError.msg );
        return batchError;
    }
    else if ( isStaleOp ) {
        WriteErrorDetail* batchError = new WriteErrorDetail;
        batchError->setErrCode( ErrorCodes::StaleShardVersion );
        batchError->setErrInfo( BSON( "downconvert" << true ) ); // For debugging
        batchError->setErrMessage( "shard version was stale" );
        return batchError;
    }

    return NULL;
}

bool batchErrorToLastError(const BatchedCommandRequest& request,
                           const BatchedCommandResponse& response,
                           LastError* error) {
    unique_ptr<WriteErrorDetail> commandError;
    WriteErrorDetail* lastBatchError = NULL;

    if (!response.getOk()) {
        // Command-level error, all writes failed

        commandError.reset(new WriteErrorDetail);
        buildErrorFromResponse(response, commandError.get());
        lastBatchError = commandError.get();
    } else if (response.isErrDetailsSet()) {
        // The last error in the batch is always reported - this matches expected COE
        // semantics for insert batches. For updates and deletes, error is only reported
        // if the error was on the last item.

        const bool lastOpErrored = response.getErrDetails().back()->getIndex() ==
            static_cast<int>(request.sizeWriteOps() - 1);
        if (request.getBatchType() == BatchedCommandRequest::BatchType_Insert || lastOpErrored) {
            lastBatchError = response.getErrDetails().back();
        }
    } else {
        // We don't care about write concern errors, these happen in legacy mode in GLE.
    }

    // Record an error if one exists
    if (lastBatchError) {
        string errMsg = lastBatchError->getErrMessage();
        error->setLastError(lastBatchError->getErrCode(),
                            errMsg.empty() ? "see code for details" : errMsg.c_str());
        return true;
    }

    // Record write stats otherwise
    // NOTE: For multi-write batches, our semantics change a little because we don't have
    // un-aggregated "n" stats.
    if (request.getBatchType() == BatchedCommandRequest::BatchType_Update) {
        BSONObj upsertedId;
        if (response.isUpsertDetailsSet()) {
            // Only report the very last item's upserted id if applicable
            if (response.getUpsertDetails().back()->getIndex() + 1 ==
                static_cast<int>(request.sizeWriteOps())) {
                upsertedId = response.getUpsertDetails().back()->getUpsertedID();
            }
        }

        int numUpserted = 0;
        if (response.isUpsertDetailsSet())
            numUpserted = response.sizeUpsertDetails();

        int numMatched = response.getN() - numUpserted;
        dassert(numMatched >= 0);

        // Wrap upserted id in "upserted" field
        BSONObj leUpsertedId;
        if (!upsertedId.isEmpty())
            leUpsertedId = upsertedId.firstElement().wrap(kUpsertedFieldName);

        error->recordUpdate(numMatched > 0, response.getN(), leUpsertedId);
    } else if (request.getBatchType() == BatchedCommandRequest::BatchType_Delete) {
        error->recordDelete(response.getN());
    }

    return false;
}