1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
|
'use strict';
var fsm = (function() {
const kIsRunningInsideTransaction = Symbol('isRunningInsideTransaction');
function forceRunningOutsideTransaction(data) {
if (data[kIsRunningInsideTransaction]) {
const err =
new Error('Intentionally thrown to stop state function from running inside of a' +
' multi-statement transaction');
err.isNotSupported = true;
throw err;
}
}
// args.data = 'this' object of the state functions
// args.db = database object
// args.collName = collection name
// args.cluster = connection strings for all cluster nodes (see fsm_libs/cluster.js for format)
// args.passConnectionCache = boolean, whether to pass a connection cache to the workload states
// args.startState = name of initial state function
// args.states = state functions of the form
// { stateName: function(db, collName) { ... } }
// args.tid = the thread identifier
// args.transitions = transitions between state functions of the form
// { stateName: { nextState1: probability,
// nextState2: ... } }
// args.iterations = number of iterations to run the FSM for
function runFSM(args) {
if (TestData.runInsideTransaction) {
let overridePath = "jstests/libs/override_methods/";
load(overridePath + "check_for_operation_not_supported_in_transaction.js");
load("jstests/concurrency/fsm_workload_helpers/auto_retry_transaction.js");
load("jstests/libs/transactions_util.js");
}
var currentState = args.startState;
// We build a cache of connections that can be used in workload states. This cache
// allows state functions to access arbitrary cluster nodes for verification checks.
// See fsm_libs/cluster.js for the format of args.cluster.
var connCache;
if (args.passConnectionCache) {
// In order to ensure that all operations performed by a worker thread happen on the
// same session, we override the "_defaultSession" property of the connections in the
// cache to be the same as the session underlying 'args.db'.
const makeNewConnWithExistingSession = function(connStr) {
// We may fail to connect if the continuous stepdown thread had just terminated
// or killed a primary. We therefore use the connect() function defined in
// network_error_and_transaction_override.js to add automatic retries to
// connections. The override is loaded in worker_thread.js.
const conn = connect(connStr).getMongo();
conn._defaultSession = new _DelegatingDriverSession(conn, args.db.getSession());
return conn;
};
const getReplSetName = (conn) => {
const res = assert.commandWorked(conn.getDB('admin').runCommand({isMaster: 1}));
assert.eq('string',
typeof res.setName,
() => `not connected to a replica set: ${tojson(res)}`);
return res.setName;
};
const makeReplSetConnWithExistingSession = (connStrList, replSetName) => {
const conn = makeNewConnWithExistingSession(`mongodb://${
connStrList.join(',')}/?appName=tid:${args.tid}&replicaSet=${replSetName}`);
return conn;
};
connCache =
{mongos: [], config: [], shards: {}, rsConns: {config: undefined, shards: {}}};
connCache.mongos = args.cluster.mongos.map(makeNewConnWithExistingSession);
connCache.config = args.cluster.config.map(makeNewConnWithExistingSession);
connCache.rsConns.config = makeReplSetConnWithExistingSession(
args.cluster.config, getReplSetName(connCache.config[0]));
// We set _isConfigServer=true on the Mongo connection object so
// set_read_preference_secondary.js knows to avoid overriding the read preference as the
// concurrency suite may be running with a 1-node CSRS.
connCache.rsConns.config._isConfigServer = true;
var shardNames = Object.keys(args.cluster.shards);
shardNames.forEach(name => {
connCache.shards[name] =
args.cluster.shards[name].map(makeNewConnWithExistingSession);
connCache.rsConns.shards[name] = makeReplSetConnWithExistingSession(
args.cluster.shards[name], getReplSetName(connCache.shards[name][0]));
});
}
for (var i = 0; i < args.iterations; ++i) {
var fn = args.states[currentState];
assert.eq('function', typeof fn, 'states.' + currentState + ' is not a function');
if (TestData.runInsideTransaction) {
try {
// We make a deep copy of 'args.data' before running the state function in a
// transaction so that if the transaction aborts, then we haven't speculatively
// modified the thread-local state.
let data;
withTxnAndAutoRetry(args.db.getSession(), () => {
data = TransactionsUtil.deepCopyObject({}, args.data);
data[kIsRunningInsideTransaction] = true;
fn.call(data, args.db, args.collName, connCache);
});
delete data[kIsRunningInsideTransaction];
args.data = data;
} catch (e) {
// Retry state functions that threw OperationNotSupportedInTransaction or
// InvalidOptions errors outside of a transaction. Rethrow any other error.
// e.isNotSupported added by check_for_operation_not_supported_in_transaction.js
if (!e.isNotSupported) {
throw e;
}
fn.call(args.data, args.db, args.collName, connCache);
}
} else {
fn.call(args.data, args.db, args.collName, connCache);
}
var nextState = getWeightedRandomChoice(args.transitions[currentState], Random.rand());
currentState = nextState;
}
// Null out the workload connection cache and perform garbage collection to clean up,
// i.e., close, the open connections.
if (args.passConnectionCache) {
connCache = null;
gc();
}
}
// doc = document of the form
// { nextState1: probability, nextState2: ... }
// randVal = a value on the interval [0, 1)
// returns a state, weighted by its probability,
// assuming randVal was chosen randomly by the caller
function getWeightedRandomChoice(doc, randVal) {
assert.gte(randVal, 0);
assert.lt(randVal, 1);
var states = Object.keys(doc);
assert.gt(states.length, 0, 'transition must have at least one state to transition to');
// weights = [ 0.25, 0.5, 0.25 ]
// => accumulated = [ 0.25, 0.75, 1 ]
var weights = states.map(function(k) {
return doc[k];
});
var accumulated = [];
var sum = weights.reduce(function(a, b, i) {
accumulated[i] = a + b;
return accumulated[i];
}, 0);
// Scale the random value by the sum of the weights
randVal *= sum; // ~ U[0, sum)
// Find the state corresponding to randVal
for (var i = 0; i < accumulated.length; ++i) {
if (randVal < accumulated[i]) {
return states[i];
}
}
assert(false, 'not reached');
}
return {
forceRunningOutsideTransaction,
run: runFSM,
_getWeightedRandomChoice: getWeightedRandomChoice,
};
})();
|
