Imported Upstream version 2.6.11upstream/2.6.11

1 files changed, 386 insertions, 0 deletions

diff --git a/src/mongo/s/write_ops/batch_write_exec.cpp b/src/mongo/s/write_ops/batch_write_exec.cpp
new file mode 100644
index 00000000000..0069a2788de
--- /dev/null
+++ b/src/mongo/s/write_ops/batch_write_exec.cpp
@@ -0,0 +1,386 @@
+/**
+ *    Copyright (C) 2013 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the GNU Affero General Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include "mongo/s/write_ops/batch_write_exec.h"
+
+#include "mongo/base/error_codes.h"
+#include "mongo/base/owned_pointer_map.h"
+#include "mongo/base/status.h"
+#include "mongo/bson/util/builder.h"
+#include "mongo/client/dbclientinterface.h" // ConnectionString (header-only)
+#include "mongo/s/write_ops/batch_write_op.h"
+#include "mongo/s/write_ops/write_error_detail.h"
+
+namespace mongo {
+
+    BatchWriteExec::BatchWriteExec( NSTargeter* targeter,
+                                    ShardResolver* resolver,
+                                    MultiCommandDispatch* dispatcher ) :
+        _targeter( targeter ),
+        _resolver( resolver ),
+        _dispatcher( dispatcher ),
+        _stats( new BatchWriteExecStats ) {
+    }
+
+    namespace {
+
+        //
+        // Map which allows associating ConnectionString hosts with TargetedWriteBatches
+        // This is needed since the dispatcher only returns hosts with responses.
+        //
+
+        // TODO: Unordered map?
+        typedef OwnedPointerMap<ConnectionString, TargetedWriteBatch> OwnedHostBatchMap;
+    }
+
+    static void buildErrorFrom( const Status& status, WriteErrorDetail* error ) {
+        error->setErrCode( status.code() );
+        error->setErrMessage( status.reason() );
+    }
+
+    // Helper to note several stale errors from a response
+    static void noteStaleResponses( const vector<ShardError*>& staleErrors, NSTargeter* targeter ) {
+        for ( vector<ShardError*>::const_iterator it = staleErrors.begin(); it != staleErrors.end();
+            ++it ) {
+            const ShardError* error = *it;
+            targeter->noteStaleResponse( error->endpoint,
+                                         error->error.isErrInfoSet() ? error->error.getErrInfo() :
+                                                                       BSONObj() );
+        }
+    }
+
+    static bool isShardMetadataChanging( const vector<ShardError*>& staleErrors ) {
+        if ( !staleErrors.empty() && staleErrors.back()->error.isErrInfoSet() )
+            return staleErrors.back()->error.getErrInfo()["inCriticalSection"].trueValue();
+        return false;
+    }
+
+    // The number of times we'll try to continue a batch op if no progress is being made
+    // This only applies when no writes are occurring and metadata is not changing on reload
+    static const int kMaxRoundsWithoutProgress( 5 );
+
+    void BatchWriteExec::executeBatch( const BatchedCommandRequest& clientRequest,
+                                       BatchedCommandResponse* clientResponse ) {
+
+        LOG( 4 ) << "starting execution of write batch of size "
+                 << static_cast<int>( clientRequest.sizeWriteOps() )
+                 << " for " << clientRequest.getNS() << endl;
+
+        BatchWriteOp batchOp;
+        batchOp.initClientRequest( &clientRequest );
+
+        // Current batch status
+        bool refreshedTargeter = false;
+        int rounds = 0;
+        int numCompletedOps = 0;
+        int numRoundsWithoutProgress = 0;
+
+        while ( !batchOp.isFinished() ) {
+
+            //
+            // Get child batches to send using the targeter
+            //
+            // Targeting errors can be caused by remote metadata changing (the collection could have
+            // been dropped and recreated, for example with a new shard key).  If a remote metadata
+            // change occurs *before* a client sends us a batch, we need to make sure that we don't
+            // error out just because we're staler than the client - otherwise mongos will be have
+            // unpredictable behavior.
+            //
+            // (If a metadata change happens *during* or *after* a client sends us a batch, however,
+            // we make no guarantees about delivery.)
+            //
+            // For this reason, we don't record targeting errors until we've refreshed our targeting
+            // metadata at least once *after* receiving the client batch - at that point, we know:
+            //
+            // 1) our new metadata is the same as the metadata when the client sent a batch, and so
+            //    targeting errors are real.
+            // OR
+            // 2) our new metadata is a newer version than when the client sent a batch, and so
+            //    the metadata must have changed after the client batch was sent.  We don't need to
+            //    deliver in this case, since for all the client knows we may have gotten the batch
+            //    exactly when the metadata changed.
+            //
+
+            OwnedPointerVector<TargetedWriteBatch> childBatchesOwned;
+            vector<TargetedWriteBatch*>& childBatches = childBatchesOwned.mutableVector();
+
+            // If we've already had a targeting error, we've refreshed the metadata once and can
+            // record target errors definitively.
+            bool recordTargetErrors = refreshedTargeter;
+            Status targetStatus = batchOp.targetBatch( *_targeter,
+                                                       recordTargetErrors,
+                                                       &childBatches );
+            if ( !targetStatus.isOK() ) {
+                // Don't do anything until a targeter refresh
+                _targeter->noteCouldNotTarget();
+                refreshedTargeter = true;
+                ++_stats->numTargetErrors;
+                dassert( childBatches.size() == 0u );
+            }
+
+            //
+            // Send all child batches
+            //
+
+            size_t numSent = 0;
+            size_t numToSend = childBatches.size();
+            bool remoteMetadataChanging = false;
+            while ( numSent != numToSend ) {
+
+                // Collect batches out on the network, mapped by endpoint
+                OwnedHostBatchMap ownedPendingBatches;
+                OwnedHostBatchMap::MapType& pendingBatches = ownedPendingBatches.mutableMap();
+
+                //
+                // Send side
+                //
+
+                // Get as many batches as we can at once
+                for ( vector<TargetedWriteBatch*>::iterator it = childBatches.begin();
+                    it != childBatches.end(); ++it ) {
+
+                    //
+                    // Collect the info needed to dispatch our targeted batch
+                    //
+
+                    TargetedWriteBatch* nextBatch = *it;
+                    // If the batch is NULL, we sent it previously, so skip
+                    if ( nextBatch == NULL ) continue;
+
+                    // Figure out what host we need to dispatch our targeted batch
+                    ConnectionString shardHost;
+                    Status resolveStatus = _resolver->chooseWriteHost( nextBatch->getEndpoint()
+                                                                           .shardName,
+                                                                       &shardHost );
+                    if ( !resolveStatus.isOK() ) {
+
+                        ++_stats->numResolveErrors;
+
+                        // Record a resolve failure
+                        // TODO: It may be necessary to refresh the cache if stale, or maybe just
+                        // cancel and retarget the batch
+                        WriteErrorDetail error;
+                        buildErrorFrom( resolveStatus, &error );
+
+                        LOG( 4 ) << "unable to send write batch to " << shardHost.toString()
+                                 << causedBy( resolveStatus.toString() ) << endl;
+
+                        batchOp.noteBatchError( *nextBatch, error );
+
+                        // We're done with this batch
+                        // Clean up when we can't resolve a host
+                        delete *it;
+                        *it = NULL;
+                        --numToSend;
+                        continue;
+                    }
+
+                    // If we already have a batch for this host, wait until the next time
+                    OwnedHostBatchMap::MapType::iterator pendingIt = pendingBatches.find( shardHost );
+                    if ( pendingIt != pendingBatches.end() ) continue;
+
+                    //
+                    // We now have all the info needed to dispatch the batch
+                    //
+
+                    BatchedCommandRequest request( clientRequest.getBatchType() );
+                    batchOp.buildBatchRequest( *nextBatch, &request );
+
+                    // Internally we use full namespaces for request/response, but we send the
+                    // command to a database with the collection name in the request.
+                    NamespaceString nss( request.getNS() );
+                    request.setNS( nss.coll() );
+
+                    LOG( 4 ) << "sending write batch to " << shardHost.toString() << ": "
+                             << request.toString() << endl;
+
+                    _dispatcher->addCommand( shardHost, nss.db(), request );
+
+                    // Indicate we're done by setting the batch to NULL
+                    // We'll only get duplicate hostEndpoints if we have broadcast and non-broadcast
+                    // endpoints for the same host, so this should be pretty efficient without
+                    // moving stuff around.
+                    *it = NULL;
+
+                    // Recv-side is responsible for cleaning up the nextBatch when used
+                    pendingBatches.insert( make_pair( shardHost, nextBatch ) );
+                }
+
+                // Send them all out
+                _dispatcher->sendAll();
+                numSent += pendingBatches.size();
+
+                //
+                // Recv side
+                //
+
+                while ( _dispatcher->numPending() > 0 ) {
+
+                    // Get the response
+                    ConnectionString shardHost;
+                    BatchedCommandResponse response;
+                    Status dispatchStatus = _dispatcher->recvAny( &shardHost, &response );
+
+                    // Get the TargetedWriteBatch to find where to put the response
+                    dassert( pendingBatches.find( shardHost ) != pendingBatches.end() );
+                    TargetedWriteBatch* batch = pendingBatches.find( shardHost )->second;
+
+                    if ( dispatchStatus.isOK() ) {
+
+                        TrackedErrors trackedErrors;
+                        trackedErrors.startTracking( ErrorCodes::StaleShardVersion );
+
+                        LOG( 4 ) << "write results received from " << shardHost.toString() << ": "
+                                 << response.toString() << endl;
+
+                        // Dispatch was ok, note response
+                        batchOp.noteBatchResponse( *batch, response, &trackedErrors );
+
+                        // Note if anything was stale
+                        const vector<ShardError*>& staleErrors =
+                            trackedErrors.getErrors( ErrorCodes::StaleShardVersion );
+
+                        if ( staleErrors.size() > 0 ) {
+                            noteStaleResponses( staleErrors, _targeter );
+                            ++_stats->numStaleBatches;
+                        }
+
+                        // Remember if the shard is actively changing metadata right now
+                        if ( isShardMetadataChanging( staleErrors ) ) {
+                            remoteMetadataChanging = true;
+                        }
+
+                        // Remember that we successfully wrote to this shard
+                        // NOTE: This will record lastOps for shards where we actually didn't update
+                        // or delete any documents, which preserves old behavior but is conservative
+                        _stats->noteWriteAt( shardHost,
+                                             response.isLastOpSet() ? 
+                                             response.getLastOp() : OpTime(),
+                                             response.isElectionIdSet() ?
+                                             response.getElectionId() : OID());
+                    }
+                    else {
+
+                        // Error occurred dispatching, note it
+
+                        stringstream msg;
+                        msg << "write results unavailable from " << shardHost.toString()
+                            << causedBy( dispatchStatus.toString() );
+
+                        WriteErrorDetail error;
+                        buildErrorFrom( Status( ErrorCodes::RemoteResultsUnavailable, msg.str() ),
+                                        &error );
+
+                        LOG( 4 ) << "unable to receive write results from " << shardHost.toString()
+                                 << causedBy( dispatchStatus.toString() ) << endl;
+
+                        batchOp.noteBatchError( *batch, error );
+                    }
+                }
+            }
+
+            ++rounds;
+            ++_stats->numRounds;
+
+            // If we're done, get out
+            if ( batchOp.isFinished() )
+                break;
+
+            // MORE WORK TO DO
+
+            //
+            // Refresh the targeter if we need to (no-op if nothing stale)
+            //
+
+            bool targeterChanged = false;
+            Status refreshStatus = _targeter->refreshIfNeeded( &targeterChanged );
+
+            if ( !refreshStatus.isOK() ) {
+
+                // It's okay if we can't refresh, we'll just record errors for the ops if
+                // needed.
+                warning() << "could not refresh targeter" << causedBy( refreshStatus.reason() )
+                          << endl;
+            }
+
+            //
+            // Ensure progress is being made toward completing the batch op
+            //
+
+            int currCompletedOps = batchOp.numWriteOpsIn( WriteOpState_Completed );
+            if ( currCompletedOps == numCompletedOps && !targeterChanged
+                 && !remoteMetadataChanging ) {
+                ++numRoundsWithoutProgress;
+            }
+            else {
+                numRoundsWithoutProgress = 0;
+            }
+            numCompletedOps = currCompletedOps;
+
+            if ( numRoundsWithoutProgress > kMaxRoundsWithoutProgress ) {
+
+                stringstream msg;
+                msg << "no progress was made executing batch write op in " << clientRequest.getNS()
+                    << " after " << kMaxRoundsWithoutProgress << " rounds (" << numCompletedOps
+                    << " ops completed in " << rounds << " rounds total)";
+
+                WriteErrorDetail error;
+                buildErrorFrom( Status( ErrorCodes::NoProgressMade, msg.str() ), &error );
+                batchOp.abortBatch( error );
+                break;
+            }
+        }
+
+        batchOp.buildClientResponse( clientResponse );
+
+        LOG( 4 ) << "finished execution of write batch"
+                 << ( clientResponse->isErrDetailsSet() ? " with write errors" : "")
+                 << ( clientResponse->isErrDetailsSet() &&
+                      clientResponse->isWriteConcernErrorSet() ? " and" : "" )
+                 << ( clientResponse->isWriteConcernErrorSet() ? " with write concern error" : "" )
+                 << " for " << clientRequest.getNS() << endl;
+    }
+
+    const BatchWriteExecStats& BatchWriteExec::getStats() {
+        return *_stats;
+    }
+
+    BatchWriteExecStats* BatchWriteExec::releaseStats() {
+        return _stats.release();
+    }
+
+    void BatchWriteExecStats::noteWriteAt(const ConnectionString& host,
+                                          OpTime opTime,
+                                          const OID& electionId) {
+        _writeOpTimes[host] = HostOpTime(opTime, electionId);
+    }
+
+    const HostOpTimeMap& BatchWriteExecStats::getWriteOpTimes() const {
+        return _writeOpTimes;
+    }
+}