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
|
'use strict';
/**
* Test a snapshot read spanning a find and getmore that runs concurrently with killSessions,
* killOp, killCursors, and txnNumber change.
*
* @tags: [uses_transactions, state_functions_share_transaction]
*/
load('jstests/concurrency/fsm_workload_helpers/snapshot_read_utils.js');
var $config = (function() {
const data = {numIds: 100, batchSize: 50};
const states = {
init: function init(db, collName) {
let session = db.getMongo().startSession({causalConsistency: true});
// Store the session ID in the database so any unterminated transactions can be aborted
// at teardown.
insertSessionDoc(db, collName, this.tid, session.getSessionId().id);
this.sessionDb = session.getDatabase(db.getName());
this.txnNumber = 0;
this.stmtId = 0;
this.iteration = 1;
},
snapshotFind: function snapshotFind(db, collName) {
const sortByAscending = false;
doSnapshotFind(sortByAscending, collName, this, [
ErrorCodes.NoSuchTransaction,
ErrorCodes.LockTimeout,
ErrorCodes.Interrupted,
ErrorCodes.SnapshotTooOld,
]);
},
snapshotGetMore: function snapshotGetMore(db, collName) {
doSnapshotGetMore(collName,
this,
[
ErrorCodes.CursorKilled,
ErrorCodes.CursorNotFound,
ErrorCodes.Interrupted,
ErrorCodes.LockTimeout,
ErrorCodes.NoSuchTransaction,
],
[ErrorCodes.NoSuchTransaction, ErrorCodes.Interrupted]);
},
incrementTxnNumber: function incrementTxnNumber(db, collName) {
const abortErrorCodes = [
ErrorCodes.NoSuchTransaction,
ErrorCodes.TransactionCommitted,
ErrorCodes.TransactionTooOld,
ErrorCodes.Interrupted
];
abortTransaction(this.sessionDb, this.txnNumber, abortErrorCodes);
this.txnNumber++;
},
killSessions: function killSessions(db, collName) {
// Kill a random active session.
let idToKill = "sessionDoc" + Math.floor(Math.random() * this.threadCount);
let sessionDocToKill = db[collName].findOne({"_id": idToKill});
// Retry the find on idToKill in case the ID corresponds to a thread that has not
// inserted its sessionDoc yet, and make sure we don't kill our own thread.
while (!sessionDocToKill || idToKill == "sessionDoc" + this.tid) {
idToKill = "sessionDoc" + Math.floor(Math.random() * this.threadCount);
sessionDocToKill = db[collName].findOne({"_id": idToKill});
}
// This command may get interrupted by another thread's killSessions.
assert.commandWorkedOrFailedWithCode(
this.sessionDb.runCommand({killSessions: [{id: sessionDocToKill.id}]}),
ErrorCodes.Interrupted);
},
killOp: function killOp(db, collName) {
// Find the object ID of the getMore in the snapshot read, if it is running, and attempt
// to kill the operation. This command may get interrupted by another thread's
// killSessions.
const res = assert.commandWorkedOrFailedWithCode(
this.sessionDb.adminCommand(
{currentOp: 1, ns: {$regex: db.getName() + "\." + collName}, op: "getmore"}),
ErrorCodes.Interrupted);
if (res.hasOwnProperty("inprog") && res.inprog.length) {
const killOpCmd = {killOp: 1, op: res.inprog[0].opid};
const killRes = this.sessionDb.adminCommand(killOpCmd);
assert.commandWorkedOrFailedWithCode(killRes, ErrorCodes.Interrupted);
}
},
killCursors: function killCursors(db, collName) {
const killCursorCmd = {killCursors: collName, cursors: [this.cursorId]};
const res = this.sessionDb.runCommand(killCursorCmd);
// This command may get interrupted by another thread's killSessions.
assert.commandWorkedOrFailedWithCode(
res,
[ErrorCodes.CursorNotFound, ErrorCodes.Interrupted],
() => `cmd: ${tojson(killCursorCmd)}`);
},
};
// Wrap each state in a cleanupOnLastIteration() invocation.
for (let stateName of Object.keys(states)) {
const stateFn = states[stateName];
const abortErrorCodes = [
ErrorCodes.NoSuchTransaction,
ErrorCodes.TransactionCommitted,
ErrorCodes.TransactionTooOld,
ErrorCodes.Interrupted
];
states[stateName] = function(db, collName) {
cleanupOnLastIteration(this, () => stateFn.apply(this, arguments), abortErrorCodes);
};
}
const transitions = {
init: {snapshotFind: 1.0},
snapshotFind: {
incrementTxnNumber: 0.20,
killSessions: 0.20,
killOp: 0.20,
killCursors: 0.20,
snapshotGetMore: 0.20
},
incrementTxnNumber: {snapshotGetMore: 1.0},
killSessions: {snapshotGetMore: 1.0},
killOp: {snapshotGetMore: 1.0},
killCursors: {snapshotGetMore: 1.0},
snapshotGetMore: {snapshotFind: 1.0}
};
function setup(db, collName, cluster) {
assertWhenOwnColl.commandWorked(db.runCommand({create: collName}));
for (let i = 0; i < this.numIds; ++i) {
const res = db[collName].insert({_id: i, value: i});
assert.writeOK(res);
assert.eq(1, res.nInserted);
}
}
function teardown(db, collName, cluster) {
// Make sure any currently running transactions are aborted.
killSessionsFromDocs(db, collName);
}
return {
threadCount: 5,
iterations: 10,
startState: 'init',
states: states,
transitions: transitions,
setup: setup,
teardown: teardown,
data: data,
};
})();
|
