jstests/concurrency/fsm_libs/worker_thread.js
// run = callback that takes a map of workloads to their associated $config
run = callback that takes a map of workloads to their associated $config
run = callback that takes a map of workloads to their associated $config
function main(workloads, args, run) {
        var myDB;
        var configs = {};
        var connectionString = 'mongodb://' + args.host + '/?appName=tid:' + args.tid;
        if (typeof args.replSetName !== 'undefined') {
            connectionString += '&replicaSet=' + args.replSetName;
        }
        globalAssertLevel = args.globalAssertLevel;
        TestData = (TestData !== undefined) ? TestData : {};
        try {
            if (typeof db !== 'undefined') {
                db = null;
                gc();
            }
            let mongo;
            if (TestData.pinningSecondary) {
                mongo = new SpecificSecondaryReaderMongo(connectionString, args.secondaryHost);
            } else {
                mongo = new Mongo(connectionString);
            }
            load('jstests/libs/override_methods/implicitly_retry_on_background_op_in_progress.js');
            if (typeof args.sessionOptions !== 'undefined') {
                let initialClusterTime;
                let initialOperationTime;
                if (typeof args.sessionOptions.initialClusterTime === 'string') {
                    initialClusterTime = eval('(' + args.sessionOptions.initialClusterTime + ')');
                    delete args.sessionOptions.initialClusterTime;
                }
                if (typeof args.sessionOptions.initialOperationTime === 'string') {
                    initialOperationTime =
                        eval('(' + args.sessionOptions.initialOperationTime + ')');
                    delete args.sessionOptions.initialOperationTime;
                }
                const session = mongo.startSession(args.sessionOptions);
                const readPreference = session.getOptions().getReadPreference();
                if (readPreference && readPreference.mode === 'secondary') {
                    session.getOptions().setReadPreference(undefined);
                    load('jstests/libs/override_methods/set_read_preference_secondary.js');
                }
                if (typeof initialClusterTime !== 'undefined') {
                    session.advanceClusterTime(initialClusterTime);
                }
                if (typeof initialOperationTime !== 'undefined') {
                    session.advanceOperationTime(initialOperationTime);
                }
                myDB = session.getDatabase(args.dbName);
            } else {
                myDB = mongo.getDB(args.dbName);
            }
            {
                let connectionDesc = '';
                if (!Cluster.isSharded(args.clusterOptions)) {
                    let res = assert.commandWorked(myDB.runCommand({whatsmyuri: 1}));
                    const myUri = res.you;
                    res = assert.commandWorked(myDB.adminCommand({currentOp: 1, client: myUri}));
                    connectionDesc = ', conn:' + res.inprog[0].desc;
                }
                const printOriginal = print;
                print = function() {
                    const printArgs = Array.from(arguments);
                    const prefix = '[tid:' + args.tid + connectionDesc + ']';
                    printArgs.unshift(prefix);
                    return printOriginal.apply(this, printArgs);
                };
            }
            if (Cluster.isReplication(args.clusterOptions)) {
                if (args.clusterOptions.hasOwnProperty('sharded') &&
                    args.clusterOptions.sharded.hasOwnProperty('stepdownOptions') &&
                    args.clusterOptions.sharded.stepdownOptions.shardStepdown) {
                    const newOptions = {
                        alwaysInjectTransactionNumber: true,
                        defaultReadConcernLevel: "majority",
                        logRetryAttempts: true,
                        overrideRetryAttempts: 3
                    };
                    Object.assign(TestData, newOptions);
                    assert(!TestData.hasOwnProperty('networkErrorAndTxnOverrideConfig'), TestData);
                    TestData.networkErrorAndTxnOverrideConfig = {retryOnNetworkErrors: true};
                    load('jstests/libs/override_methods/network_error_and_txn_override.js');
                }
                load('jstests/libs/override_methods/implicitly_retry_on_database_drop_pending.js');
            }
            if (TestData.defaultReadConcernLevel || TestData.defaultWriteConcern) {
                load('jstests/libs/override_methods/set_read_and_write_concerns.js');
            }
            workloads.forEach(function(workload) {
                load(workload);                     
                var config = parseConfig($config);  
                var data = Object.extend({}, args.data[workload], true);
                data = Object.extend(data, config.data, true);
                Object.defineProperties(data, {
                    'iterations': {configurable: false, writable: false, value: data.iterations},
                    'threadCount': {configurable: false, writable: false, value: data.threadCount}
                });
                data.tid = args.tid;
                configs[workload] = {
                    data: data,
                    db: myDB,
                    collName: args.collName,
                    cluster: args.cluster,
                    iterations: data.iterations,
                    passConnectionCache: config.passConnectionCache,
                    startState: config.startState,
                    states: config.states,
                    tid: args.tid,
                    transitions: config.transitions
                };
            });
            args.latch.countDown();
            try {
                args.latch.await();  
                Random.setRandomSeed(args.seed);
                run(configs);
                return {ok: 1};
            } catch (e) {
                args.errorLatch.countDown();
                return {
                    ok: 0,
                    err: e.toString(),
                    stack: e.stack,
                    tid: args.tid,
                    workloads: workloads,
                };
            }
        } finally {
            configs = null;
            myDB = null;
            gc();
        }
    }
function main ( workloads , args , run ) { var myDB ; var configs = { } ; var connectionString = 'mongodb://' + args . host + '/?appName=tid:' + args . tid ; if ( typeof args . replSetName !== 'undefined' ) { connectionString += '&replicaSet=' + args . replSetName ; } globalAssertLevel = args . globalAssertLevel ; TestData = ( TestData !== undefined ) ? TestData : { } ; try { if ( typeof db !== 'undefined' ) { db = null ; gc ( ) ; } let mongo ; if ( TestData . pinningSecondary ) { mongo = new SpecificSecondaryReaderMongo ( connectionString , args . secondaryHost ) ; } else { mongo = new Mongo ( connectionString ) ; } load ( 'jstests/libs/override_methods/implicitly_retry_on_background_op_in_progress.js' ) ; if ( typeof args . sessionOptions !== 'undefined' ) { let initialClusterTime ; let initialOperationTime ; if ( typeof args . sessionOptions . initialClusterTime === 'string' ) { initialClusterTime = eval ( '(' + args . sessionOptions . initialClusterTime + ')' ) ; delete args . sessionOptions . initialClusterTime ; } if ( typeof args . sessionOptions . initialOperationTime === 'string' ) { initialOperationTime = eval ( '(' + args . sessionOptions . initialOperationTime + ')' ) ; delete args . sessionOptions . initialOperationTime ; } const session = mongo . startSession ( args . sessionOptions ) ; const readPreference = session . getOptions ( ) . getReadPreference ( ) ; if ( readPreference && readPreference . mode === 'secondary' ) { session . getOptions ( ) . setReadPreference ( undefined ) ; load ( 'jstests/libs/override_methods/set_read_preference_secondary.js' ) ; } if ( typeof initialClusterTime !== 'undefined' ) { session . advanceClusterTime ( initialClusterTime ) ; } if ( typeof initialOperationTime !== 'undefined' ) { session . advanceOperationTime ( initialOperationTime ) ; } myDB = session . getDatabase ( args . dbName ) ; } else { myDB = mongo . getDB ( args . dbName ) ; } { let connectionDesc = '' ; if ( ! Cluster . isSharded ( args . clusterOptions ) ) { let res = assert . commandWorked ( myDB . runCommand ( { whatsmyuri : 1 } ) ) ; const myUri = res . you ; res = assert . commandWorked ( myDB . adminCommand ( { currentOp : 1 , client : myUri } ) ) ; connectionDesc = ', conn:' + res . inprog [ 0 ] . desc ; } const printOriginal = print ; print = function ( ) { const printArgs = Array . from ( arguments ) ; const prefix = '[tid:' + args . tid + connectionDesc + ']' ; printArgs . unshift ( prefix ) ; return printOriginal . apply ( this , printArgs ) ; } ; } if ( Cluster . isReplication ( args . clusterOptions ) ) { if ( args . clusterOptions . hasOwnProperty ( 'sharded' ) && args . clusterOptions . sharded . hasOwnProperty ( 'stepdownOptions' ) && args . clusterOptions . sharded . stepdownOptions . shardStepdown ) { const newOptions = { alwaysInjectTransactionNumber : true , defaultReadConcernLevel : "majority" , logRetryAttempts : true , overrideRetryAttempts : 3 } ; Object . assign ( TestData , newOptions ) ; assert ( ! TestData . hasOwnProperty ( 'networkErrorAndTxnOverrideConfig' ) , TestData ) ; TestData . networkErrorAndTxnOverrideConfig = { retryOnNetworkErrors : true } ; load ( 'jstests/libs/override_methods/network_error_and_txn_override.js' ) ; } load ( 'jstests/libs/override_methods/implicitly_retry_on_database_drop_pending.js' ) ; } if ( TestData . defaultReadConcernLevel || TestData . defaultWriteConcern ) { load ( 'jstests/libs/override_methods/set_read_and_write_concerns.js' ) ; } workloads . forEach ( function ( workload ) { load ( workload ) ; var config = parseConfig ( $config ) ; var data = Object . extend ( { } , args . data [ workload ] , true ) ; data = Object . extend ( data , config . data , true ) ; Object . defineProperties ( data , { 'iterations' : { configurable : false , writable : false , value : data . iterations } , 'threadCount' : { configurable : false , writable : false , value : data . threadCount } } ) ; data . tid = args . tid ; configs [ workload ] = { data : data , db : myDB , collName : args . collName , cluster : args . cluster , iterations : data . iterations , passConnectionCache : config . passConnectionCache , startState : config . startState , states : config . states , tid : args . tid , transitions : config . transitions } ; } ) ; args . latch . countDown ( ) ; try { args . latch . await ( ) ; Random . setRandomSeed ( args . seed ) ; run ( configs ) ; return { ok : 1 } ; } catch ( e ) { args . errorLatch . countDown ( ) ; return { ok : 0 , err : e . toString ( ) , stack : e . stack , tid : args . tid , workloads : workloads , } ; } } finally { configs = null ; myDB = null ; gc ( ) ; } }
run = callback that takes a map of workloads to their associated $config
run = callback that takes a map of workloads to their associated $config
// The global 'TestData' object may still be undefined if the concurrency suite isn't being // run by resmoke.py (e.g. if it is being run via a parallel shell in the backup/restore // tests). // The implicit database connection created within the thread's scope // is unneeded, so forcibly clean it up. // Retry operations that fail due to in-progress background operations. Load this early // so that later overrides can be retried. // JavaScript objects backed by C++ objects (e.g. BSON values from a command // response) do not serialize correctly when passed through the Thread // constructor. To work around this behavior, we instead pass a stringified form // of the JavaScript object through the Thread constructor and use eval() // to rehydrate it. // The initialClusterTime property was removed from SessionOptions in a // later revision of the Driver's specification, so we remove the property // and call advanceClusterTime() ourselves. // The initialOperationTime property was removed from SessionOptions in a // later revision of the Driver's specification, so we remove the property // and call advanceOperationTime() ourselves. // Unset the explicit read preference so set_read_preference_secondary.js can do // the right thing based on the DB. // We load() set_read_preference_secondary.js in order to avoid running // commands against the "admin" and "config" databases via mongos with // readPreference={mode: "secondary"} when there's only a single node in // the CSRS. // In sharded environments, mongos is acting as a proxy for the mongo shell and // therefore has a different outbound port than the 'whatsmyuri' command returns. // Operations that run after a "dropDatabase" command has been issued may fail with // a "DatabaseDropPending" error response if they would create a new collection on // that database while we're waiting for a majority of nodes in the replica set to // confirm it has been dropped. We load the // implicitly_retry_on_database_drop_pending.js file to make it so that the clients // started by the concurrency framework automatically retry their operation in the // face of this particular error response. // for $config // to normalize // Copy any modifications that were made to $config.data // during the setup function of the workload (see caveat // below). // XXX: Changing the order of extend calls causes problems // for workloads that reference $super. // Suppose you have workloads A and B, where workload B extends // workload A. The $config.data of workload B can define a // function that closes over the $config object of workload A // (known as $super to workload B). This reference is lost when // the config object is serialized to BSON, which results in // undefined variables in the derived workload. // Object.extend() defines all properties added to the destination object as // configurable, enumerable, and writable. To prevent workloads from changing // the iterations and threadCount properties in their state functions, we redefine // them here as non-configurable and non-writable. // Converts any exceptions to a return status. In order for the // parent thread to call countDown() on our behalf, we must throw // an exception. Nothing prior to (and including) args.latch.countDown() // should be wrapped in a try/catch statement. // wait for all threads to start // Avoid retention of connection object