SERVER-13643 Plumb TransactionExperiment through Cloner

6 files changed, 107 insertions, 53 deletions

diff --git a/src/mongo/db/cloner.cpp b/src/mongo/db/cloner.cpp
index be4bc6118f9..11e6f7394c3 100644
--- a/src/mongo/db/cloner.cpp
+++ b/src/mongo/db/cloner.cpp
@@ -122,11 +122,14 @@ namespace mongo {
     Cloner::Cloner() { }
 
     struct Cloner::Fun {
-        Fun( Client::Context& ctx ) : lastLog(0), context( ctx ) { }
+        Fun( TransactionExperiment* txn, Client::Context& ctx )
+            :lastLog(0),
+             txn(txn),
+             context(ctx)
+        {}
 
         void operator()( DBClientCursorBatchIterator &i ) {
             Lock::GlobalWrite lk;
-            DurTransaction txn;
             context.relocked();
 
             bool createdCollection = false;
@@ -155,7 +158,7 @@ namespace mongo {
                                  << to_collection << "]",
                                  !createdCollection );
                         createdCollection = true;
-                        collection = context.db()->createCollection( &txn, to_collection );
+                        collection = context.db()->createCollection( txn, to_collection );
                         verify( collection );
                     }
                 }
@@ -182,14 +185,14 @@ namespace mongo {
 
                 verify(nsToCollectionSubstring(from_collection) != "system.indexes");
 
-                StatusWith<DiskLoc> loc = collection->insertDocument( &txn, js, true );
+                StatusWith<DiskLoc> loc = collection->insertDocument( txn, js, true );
                 if ( !loc.isOK() ) {
                     error() << "error: exception cloning object in " << from_collection
                             << ' ' << loc.toString() << " obj:" << js;
                 }
                 uassertStatusOK( loc.getStatus() );
                 if ( logForRepl )
-                    logOp(&txn, "i", to_collection, js);
+                    logOp(txn, "i", to_collection, js);
 
                 getDur().commitIfNeeded();
 
@@ -201,6 +204,7 @@ namespace mongo {
         }
 
         time_t lastLog;
+        TransactionExperiment* txn;
         Client::Context& context;
 
         int64_t numSeen;
@@ -217,17 +221,22 @@ namespace mongo {
     /* copy the specified collection
        isindex - if true, this is system.indexes collection, in which we do some transformation when copying.
     */
-    void Cloner::copy(Client::Context& ctx,
-                      const char *from_collection, const char *to_collection, bool isindex,
-                      bool logForRepl, bool masterSameProcess, bool slaveOk, bool mayYield,
-                      bool mayBeInterrupted, Query query) {
-
-        DurTransaction txn; // XXX
+    void Cloner::copy(TransactionExperiment* txn,
+                      Client::Context& ctx,
+                      const char *from_collection,
+                      const char *to_collection,
+                      bool isindex,
+                      bool logForRepl,
+                      bool masterSameProcess,
+                      bool slaveOk,
+                      bool mayYield,
+                      bool mayBeInterrupted,
+                      Query query) {
 
         list<BSONObj> indexesToBuild;
         LOG(2) << "\t\tcloning collection " << from_collection << " to " << to_collection << " on " << _conn->getServerAddress() << " with filter " << query.toString() << endl;
 
-        Fun f( ctx );
+        Fun f( txn, ctx );
         f.numSeen = 0;
         f.isindex = isindex;
         f.from_collection = from_collection;
@@ -253,9 +262,9 @@ namespace mongo {
 
                 BSONObj spec = *i;
                 string ns = spec["ns"].String(); // this was fixed when pulled off network
-                Collection* collection = f.context.db()->getCollection( ns );
+                Collection* collection = f.context.db()->getCollection( txn, ns );
                 if ( !collection ) {
-                    collection = f.context.db()->createCollection( &txn, ns );
+                    collection = f.context.db()->createCollection( txn, ns );
                     verify( collection );
                 }
 
@@ -270,7 +279,7 @@ namespace mongo {
                 }
 
                 if ( logForRepl )
-                    logOp(&txn, "i", to_collection, spec);
+                    logOp(txn, "i", to_collection, spec);
 
                 getDur().commitIfNeeded();
 
@@ -297,7 +306,10 @@ namespace mongo {
         return true;
     }
 
-    bool Cloner::copyCollectionFromRemote(const string& host, const string& ns, string& errmsg) {
+    bool Cloner::copyCollectionFromRemote(TransactionExperiment* txn,
+                                          const string& host,
+                                          const string& ns,
+                                          string& errmsg) {
         Cloner cloner;
 
         DBClientConnection *tmpConn = new DBClientConnection();
@@ -305,21 +317,25 @@ namespace mongo {
         cloner.setConnection(tmpConn);
         uassert(15908, errmsg, tmpConn->connect(host, errmsg) && replAuthenticate(tmpConn));
 
-        return cloner.copyCollection(ns, BSONObj(), errmsg, true, false, true, false);
+        return cloner.copyCollection(txn, ns, BSONObj(), errmsg, true, false, true, false);
     }
 
-    bool Cloner::copyCollection(const string& ns, const BSONObj& query, string& errmsg,
-                                bool mayYield, bool mayBeInterrupted, bool copyIndexes,
+    bool Cloner::copyCollection(TransactionExperiment* txn,
+                                const string& ns,
+                                const BSONObj& query,
+                                string& errmsg,
+                                bool mayYield,
+                                bool mayBeInterrupted,
+                                bool copyIndexes,
                                 bool logForRepl) {
 
         Client::WriteContext ctx(ns);
-        DurTransaction txn; // XXX
 
         // config
         string temp = ctx.ctx().db()->name() + ".system.namespaces";
         BSONObj config = _conn->findOne(temp , BSON("name" << ns));
         if (config["options"].isABSONObj()) {
-            Status status = userCreateNS(&txn, ctx.ctx().db(), ns, config["options"].Obj(), logForRepl, 0);
+            Status status = userCreateNS(txn, ctx.ctx().db(), ns, config["options"].Obj(), logForRepl, 0);
             if ( !status.isOK() ) {
                 errmsg = status.toString();
                 return false;
@@ -327,7 +343,7 @@ namespace mongo {
         }
 
         // main data
-        copy(ctx.ctx(),
+        copy(txn, ctx.ctx(),
              ns.c_str(), ns.c_str(), false, logForRepl, false, true, mayYield, mayBeInterrupted,
              Query(query).snapshot());
 
@@ -338,7 +354,7 @@ namespace mongo {
 
         // indexes
         temp = ctx.ctx().db()->name() + ".system.indexes";
-        copy(ctx.ctx(), temp.c_str(), temp.c_str(), true, logForRepl, false, true, mayYield,
+        copy(txn, ctx.ctx(), temp.c_str(), temp.c_str(), true, logForRepl, false, true, mayYield,
              mayBeInterrupted, BSON( "ns" << ns ));
 
         getDur().commitIfNeeded();
@@ -347,10 +363,13 @@ namespace mongo {
 
     extern bool inDBRepair;
 
-    bool Cloner::go(Client::Context& context,
-                    const string& masterHost, const CloneOptions& opts, set<string>* clonedColls,
-                    string& errmsg, int* errCode) {
-        DurTransaction txn; // XXX
+    bool Cloner::go(TransactionExperiment* txn,
+                    Client::Context& context,
+                    const string& masterHost,
+                    const CloneOptions& opts,
+                    set<string>* clonedColls,
+                    string& errmsg,
+                    int* errCode) {
         if ( errCode ) {
             *errCode = 0;
         }
@@ -474,13 +493,13 @@ namespace mongo {
 
             {
                 /* we defer building id index for performance - building it in batch is much faster */
-                userCreateNS(&txn, context.db(), to_name, options, opts.logForRepl, false);
+                userCreateNS(txn, context.db(), to_name, options, opts.logForRepl, false);
             }
             LOG(1) << "\t\t cloning " << from_name << " -> " << to_name << endl;
             Query q;
             if( opts.snapshot )
                 q.snapshot();
-            copy(context,from_name, to_name.c_str(), false, opts.logForRepl, masterSameProcess,
+            copy(txn, context,from_name, to_name.c_str(), false, opts.logForRepl, masterSameProcess,
                  opts.slaveOk, opts.mayYield, opts.mayBeInterrupted, q);
 
             {
@@ -522,16 +541,27 @@ namespace mongo {
             BSONObj query = BSON( "name" << NE << "_id_" << "ns" << NIN << arr );
             
             // won't need a snapshot of the query of system.indexes as there can never be very many.
-            copy(context,system_indexes_from.c_str(), system_indexes_to.c_str(), true,
+            copy(txn, context,system_indexes_from.c_str(), system_indexes_to.c_str(), true,
                  opts.logForRepl, masterSameProcess, opts.slaveOk, opts.mayYield, opts.mayBeInterrupted, query );
         }
         return true;
     }
 
-    bool Cloner::cloneFrom(Client::Context& context, const string& masterHost, const CloneOptions& options,
-                           string& errmsg, int* errCode, set<string>* clonedCollections) {
+    bool Cloner::cloneFrom(TransactionExperiment* txn,
+                           Client::Context& context,
+                           const string& masterHost,
+                           const CloneOptions& options,
+                           string& errmsg,
+                           int* errCode,
+                           set<string>* clonedCollections) {
         Cloner cloner;
-        return cloner.go(context, masterHost.c_str(), options, clonedCollections, errmsg, errCode);
+        return cloner.go(txn,
+                         context,
+                         masterHost.c_str(),
+                         options,
+                         clonedCollections,
+                         errmsg,
+                         errCode);
     }
 
     /* Usage:
@@ -587,9 +617,10 @@ namespace mongo {
 
             Lock::DBWrite dbXLock(dbname);
             Client::Context context( dbname );
+            DurTransaction txn;
 
             Cloner cloner;
-            bool rval = cloner.go(context, from, opts, &clonedColls, errmsg);
+            bool rval = cloner.go(&txn, context, from, opts, &clonedColls, errmsg);
 
             BSONArrayBuilder barr;
             barr.append( clonedColls );
@@ -669,7 +700,8 @@ namespace mongo {
 
             cloner.setConnection( myconn.release() );
 
-            return cloner.copyCollection(collection, query, errmsg, true, false, copyIndexes);
+            DurTransaction txn;
+            return cloner.copyCollection(&txn, collection, query, errmsg, true, false, copyIndexes);
         }
     } cmdCloneCollection;
 
@@ -815,6 +847,8 @@ namespace mongo {
                                              static_cast<Lock::ScopedLock*>( new Lock::GlobalWrite() ) :
                                              static_cast<Lock::ScopedLock*>( new Lock::DBWrite( todb ) ) );
 
+            DurTransaction txn;
+
             Cloner cloner;
             string username = cmdObj.getStringField( "username" );
             string nonce = cmdObj.getStringField( "nonce" );
@@ -847,7 +881,7 @@ namespace mongo {
                 cloner.setConnection(conn);
             }
             Client::Context ctx(todb);
-            return cloner.go(ctx, fromhost, cloneOptions, NULL, errmsg );
+            return cloner.go(&txn, ctx, fromhost, cloneOptions, NULL, errmsg );
         }
     } cmdCopyDB;
 
diff --git a/src/mongo/db/cloner.h b/src/mongo/db/cloner.h
index 255c1442f0b..1cd9e06989f 100644
--- a/src/mongo/db/cloner.h
+++ b/src/mongo/db/cloner.h
@@ -40,6 +40,7 @@ namespace mongo {
     class DBClientBase;
     class DBClientCursor;
     class Query;
+    class TransactionExperiment;
 
     class Cloner: boost::noncopyable {
     public:
@@ -54,13 +55,20 @@ namespace mongo {
         void setConnection( DBClientBase *c ) { _conn.reset( c ); }
 
         /** copy the entire database */
-        bool go(Client::Context& ctx,
-                const string& masterHost, const CloneOptions& opts,
+        bool go(TransactionExperiment* txn,
+                Client::Context& ctx,
+                const string& masterHost,
+                const CloneOptions& opts,
                 set<string>* clonedColls,
                 string& errmsg, int *errCode = 0);
 
-        bool copyCollection(const string& ns, const BSONObj& query, string& errmsg,
-                            bool mayYield, bool mayBeInterrupted, bool copyIndexes = true,
+        bool copyCollection(TransactionExperiment* txn,
+                            const string& ns,
+                            const BSONObj& query,
+                            string& errmsg,
+                            bool mayYield,
+                            bool mayBeInterrupted,
+                            bool copyIndexes = true,
                             bool logForRepl = true );
         /**
          * validate the cloner query was successful
@@ -77,20 +85,31 @@ namespace mongo {
          *                      Currently this will only be set if there is an error in the initial
          *                      system.namespaces query.
          */
-        static bool cloneFrom(Client::Context& context,
-                              const string& masterHost, const CloneOptions& options,
-                              string& errmsg, int* errCode = 0,
+        static bool cloneFrom(TransactionExperiment* txn,
+                              Client::Context& context,
+                              const string& masterHost,
+                              const CloneOptions& options,
+                              string& errmsg,
+                              int* errCode = 0,
                               set<string>* clonedCollections = 0);
 
         /**
          * Copy a collection (and indexes) from a remote host
          */
-        static bool copyCollectionFromRemote(const string& host, const string& ns, string& errmsg);
+        static bool copyCollectionFromRemote(TransactionExperiment* txn,
+                                             const string& host, const string& ns, string& errmsg);
 
     private:
-        void copy(Client::Context& ctx,
-                  const char *from_ns, const char *to_ns, bool isindex, bool logForRepl,
-                  bool masterSameProcess, bool slaveOk, bool mayYield, bool mayBeInterrupted,
+        void copy(TransactionExperiment* txn,
+                  Client::Context& ctx,
+                  const char *from_ns,
+                  const char *to_ns,
+                  bool isindex,
+                  bool logForRepl,
+                  bool masterSameProcess,
+                  bool slaveOk,
+                  bool mayYield,
+                  bool mayBeInterrupted,
                   Query q);
 
         struct Fun;
diff --git a/src/mongo/db/repl/master_slave.cpp b/src/mongo/db/repl/master_slave.cpp
index 40def198015..13d9963c5dd 100644
--- a/src/mongo/db/repl/master_slave.cpp
+++ b/src/mongo/db/repl/master_slave.cpp
@@ -361,7 +361,7 @@ namespace mongo {
     }
 
     /* grab initial copy of a database from the master */
-    void ReplSource::resync(const std::string& dbName) {
+    void ReplSource::resync(TransactionExperiment* txn, const std::string& dbName) {
         const std::string db(dbName);   // need local copy of the name, we're dropping the original
         resyncDrop( db );
         Client::Context ctx( db );
@@ -378,7 +378,7 @@ namespace mongo {
             cloneOptions.snapshot = true;
             cloneOptions.mayYield = true;
             cloneOptions.mayBeInterrupted = false;
-            bool ok = Cloner::cloneFrom(ctx,hostName, cloneOptions, errmsg, &errCode);
+            bool ok = Cloner::cloneFrom(txn, ctx,hostName, cloneOptions, errmsg, &errCode);
 
             if ( !ok ) {
                 if ( errCode == DatabaseDifferCaseCode ) {
@@ -639,7 +639,7 @@ namespace mongo {
                 save();
                 Client::Context ctx(ns);
                 nClonedThisPass++;
-                resync(ctx.db()->name());
+                resync(&txn, ctx.db()->name());
                 addDbNextPass.erase(clientName);
                 incompleteCloneDbs.erase( clientName );
             }
diff --git a/src/mongo/db/repl/master_slave.h b/src/mongo/db/repl/master_slave.h
index 97255cbecde..6529480f53f 100644
--- a/src/mongo/db/repl/master_slave.h
+++ b/src/mongo/db/repl/master_slave.h
@@ -77,7 +77,7 @@ namespace mongo {
     class ReplSource {
         shared_ptr<threadpool::ThreadPool> tp;
 
-        void resync(const std::string& dbName);
+        void resync(TransactionExperiment* txn, const std::string& dbName);
 
         /** @param alreadyLocked caller already put us in write lock if true */
         void sync_pullOpLog_applyOperation(BSONObj& op, bool alreadyLocked);
diff --git a/src/mongo/db/repl/rs_initialsync.cpp b/src/mongo/db/repl/rs_initialsync.cpp
index 76c39c88d9a..506e9a8d9a6 100644
--- a/src/mongo/db/repl/rs_initialsync.cpp
+++ b/src/mongo/db/repl/rs_initialsync.cpp
@@ -98,6 +98,7 @@ namespace mongo {
                 sethbmsg( str::stream() << "initial sync cloning indexes for : " << db , 0);
 
             Client::WriteContext ctx(db);
+            DurTransaction txn;
 
             string err;
             int errCode;
@@ -112,7 +113,7 @@ namespace mongo {
             options.syncData = dataPass;
             options.syncIndexes = ! dataPass;
 
-            if (!cloner.go(ctx.ctx(), master, options, NULL, err, &errCode)) {
+            if (!cloner.go(&txn, ctx.ctx(), master, options, NULL, err, &errCode)) {
                 sethbmsg(str::stream() << "initial sync: error while "
                                        << (dataPass ? "cloning " : "indexing ") << db
                                        << ".  " << (err.empty() ? "" : err + ".  ")
diff --git a/src/mongo/db/repl/rs_rollback.cpp b/src/mongo/db/repl/rs_rollback.cpp
index 529ea548462..5079ba8bea1 100644
--- a/src/mongo/db/repl/rs_rollback.cpp
+++ b/src/mongo/db/repl/rs_rollback.cpp
@@ -413,7 +413,7 @@ namespace mongo {
                     {
                         string errmsg;
                         dbtemprelease r;
-                        bool ok = Cloner::copyCollectionFromRemote(them->getServerAddress(), ns, errmsg);
+                        bool ok = Cloner::copyCollectionFromRemote(&txn, them->getServerAddress(), ns, errmsg);
                         uassert(15909, str::stream() << "replSet rollback error resyncing collection " << ns << ' ' << errmsg, ok);
                     }
                 }