void WriteBatchExecutor::execUpdate( const BatchItemRef& updateItem,
                                     BSONObj* upsertedId,
                                     WriteErrorDetail** error ) {

    // BEGIN CURRENT OP
    scoped_ptr<CurOp> currentOp( beginCurrentOp( _client, updateItem ) );
    incOpStats( updateItem );

    WriteOpResult result;

    WriteUnitOfWork wunit(_txn->recoveryUnit());
    multiUpdate( _txn, updateItem, &result );
    wunit.commit();

    if ( !result.getStats().upsertedID.isEmpty() ) {
        *upsertedId = result.getStats().upsertedID;
    }
    // END CURRENT OP
    incWriteStats( updateItem, result.getStats(), result.getError(), currentOp.get() );
    finishCurrentOp( _txn, _client, currentOp.get(), result.getError() );

    if ( result.getError() ) {
        result.getError()->setIndex( updateItem.getItemIndex() );
        *error = result.releaseError();
    }
}

    void WriteBatchExecutor::execRemove( const BatchItemRef& removeItem,
                                         WriteErrorDetail** error ) {

        // Removes are similar to updates, but page faults are handled externally

        // BEGIN CURRENT OP
        scoped_ptr<CurOp> currentOp( beginCurrentOp( _client, removeItem ) );
        incOpStats( removeItem );

        WriteOpResult result;

        // NOTE: Deletes will not fault outside the lock once any data has been written
        PageFaultRetryableSection pageFaultSection;
        while ( true ) {
            try {
                multiRemove( removeItem, &result );
                break;
            }
            catch (PageFaultException& pfe) {
                pfe.touch();
                invariant(!result.getError());
                continue;
            }
            fassertFailed(17429);
        }

        // END CURRENT OP
        incWriteStats( removeItem, result.getStats(), result.getError(), currentOp.get() );
        finishCurrentOp( _client, currentOp.get(), result.getError() );

        if ( result.getError() ) {
            result.getError()->setIndex( removeItem.getItemIndex() );
            *error = result.releaseError();
        }
    }

    void WriteBatchExecutor::execUpdate( const BatchItemRef& updateItem,
                                         BSONObj* upsertedId,
                                         WriteErrorDetail** error ) {

        // Updates currently do a lot of the lock management internally

        const BatchedCommandRequest& request = *updateItem.getRequest();
        const NamespaceString nss( updateItem.getRequest()->getNS() );

        // BEGIN CURRENT OP
        scoped_ptr<CurOp> currentOp( beginCurrentOp( _client, updateItem ) );
        incOpStats( updateItem );

        WriteOpResult result;

        {
            ///////////////////////////////////////////
            Lock::DBWrite writeLock( nss.ns() );
            ///////////////////////////////////////////

            // Check version once we're locked

            if ( checkShardVersion( &shardingState, request, &result.error ) ) {

                // Context once we're locked, to set more details in currentOp()
                // TODO: better constructor?
                Client::Context writeContext( nss.ns(),
                                              storageGlobalParams.dbpath,
                                              false /* don't check version */);

                multiUpdate( updateItem, &result );

                incWriteStats( updateItem, result.stats, result.error, currentOp.get() );

                if ( !result.stats.upsertedID.isEmpty() ) {
                    *upsertedId = result.stats.upsertedID.getOwned();
                }
            }
        }

        // END CURRENT OP
        finishCurrentOp( _client, currentOp.get(), result.error );

        if ( result.error ) {
            result.error->setIndex( updateItem.getItemIndex() );
            *error = result.releaseError();
        }
    }

    void WriteBatchExecutor::execOneInsert(ExecInsertsState* state, WriteErrorDetail** error) {
        BatchItemRef currInsertItem(state->request, state->currIndex);
        scoped_ptr<CurOp> currentOp(beginCurrentOp(_client, currInsertItem));
        incOpStats(currInsertItem);

        WriteOpResult result;
        insertOne(state, &result);

        if (state->hasLock()) {
            // Normally, unlocking records lock time stats on the active CurOp.  However,
            // insertOne() may not release the lock. In that case, record time by hand.
            state->getLock().recordTime();
            // If we deschedule here, there could be substantial unaccounted locked time.
            // Any time from here will be attributed to the next insert in the batch, or
            // not attributed to any operation if this is the last op in the batch.
            state->getLock().resetTime();
        }

        incWriteStats(currInsertItem,
                      result.getStats(),
                      result.getError(),
                      currentOp.get());
        finishCurrentOp(_txn, _client, currentOp.get(), result.getError());

        if (result.getError()) {
            *error = result.releaseError();
        }
    }

    void WriteBatchExecutor::execRemove( const BatchItemRef& removeItem,
                                         WriteErrorDetail** error ) {

        // Removes are similar to updates, but page faults are handled externally

        // BEGIN CURRENT OP
        scoped_ptr<CurOp> currentOp( beginCurrentOp( _client, removeItem ) );
        incOpStats( removeItem );

        WriteOpResult result;

        multiRemove( _txn, removeItem, &result );

        // END CURRENT OP
        incWriteStats( removeItem, result.getStats(), result.getError(), currentOp.get() );
        finishCurrentOp( _txn, _client, currentOp.get(), result.getError() );

        if ( result.getError() ) {
            result.getError()->setIndex( removeItem.getItemIndex() );
            *error = result.releaseError();
        }
    }