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
|
/**
* Helpers for doing a snapshot read in concurrency suites. Specifically, the read is a find that
* spans a getmore.
*/
load('jstests/concurrency/fsm_workload_helpers/cleanup_txns.js');
load('jstests/libs/transactions_util.js');
/**
* Parses a cursor from cmdResult, if possible.
*/
function parseCursor(cmdResult) {
if (cmdResult.hasOwnProperty("cursor")) {
assert(cmdResult.cursor.hasOwnProperty("id"));
return cmdResult.cursor;
}
return null;
}
/**
* Performs a snapshot find.
*/
function doSnapshotFind(sortByAscending, collName, data, findErrorCodes) {
// Reset txnNumber and stmtId for this transaction.
abortTransaction(data.sessionDb, data.txnNumber);
data.txnNumber++;
data.stmtId = 0;
const sortOrder = sortByAscending ? {_id: 1} : {_id: -1};
const findCmd = {
find: collName,
sort: sortOrder,
batchSize: 0,
readConcern: {level: "snapshot"},
txnNumber: NumberLong(data.txnNumber),
stmtId: NumberInt(data.stmtId++),
startTransaction: true,
autocommit: false
};
let res = data.sessionDb.runCommand(findCmd);
// A transaction request can always fail with a transient transaction error, so only check the
// specific error code if it is not labeled as transient.
if (!TransactionsUtil.isTransientTransactionError(res)) {
assert.commandWorkedOrFailedWithCode(res, findErrorCodes, () => `cmd: ${tojson(findCmd)}`);
}
const cursor = parseCursor(res);
if (!cursor) {
abortTransaction(
data.sessionDb, data.txnNumber, [ErrorCodes.NoSuchTransaction, ErrorCodes.Interrupted]);
data.cursorId = NumberLong(0);
} else {
assert(cursor.hasOwnProperty("firstBatch"), tojson(res));
assert.eq(0, cursor.firstBatch.length, tojson(res));
assert.neq(cursor.id, 0);
// Store the cursor Id in the data object.
data.cursorId = cursor.id;
}
}
/**
* Performs a snapshot getmore. This function is to be used in conjunction with doSnapshotFind.
*/
function doSnapshotGetMore(collName, data, getMoreErrorCodes, commitTransactionErrorCodes) {
// doSnapshotGetMore may be called even if doSnapshotFind fails to obtain a cursor.
if (bsonWoCompare({_: data.cursorId}, {_: NumberLong(0)}) === 0) {
return;
}
const getMoreCmd = {
getMore: data.cursorId,
collection: collName,
batchSize: data.batchSize,
txnNumber: NumberLong(data.txnNumber),
stmtId: NumberInt(data.stmtId++),
autocommit: false
};
let res = data.sessionDb.runCommand(getMoreCmd);
// A transaction request can always fail with a transient transaction error, so only check the
// specific error code if it is not labeled as transient.
if (!TransactionsUtil.isTransientTransactionError(res)) {
assert.commandWorkedOrFailedWithCode(
res, getMoreErrorCodes, () => `cmd: ${tojson(getMoreCmd)}`);
}
const commitCmd = {
commitTransaction: 1,
txnNumber: NumberLong(data.txnNumber),
autocommit: false
};
res = data.sessionDb.adminCommand(commitCmd);
assert.commandWorkedOrFailedWithCode(
res, commitTransactionErrorCodes, () => `cmd: ${tojson(commitCmd)}`);
}
/**
* This function can be used to share session data across threads.
*/
function insertSessionDoc(db, collName, tid, sessionId) {
const sessionDoc = {"_id": "sessionDoc" + tid, "id": sessionId};
const res = db[collName].insert(sessionDoc);
assert.commandWorked(res);
assert.eq(1, res.nInserted);
}
/**
* This function can be used in conjunction with insertSessionDoc to kill any active sessions on
* teardown or iteration completion.
*/
function killSessionsFromDocs(db, collName, tid) {
// Cleanup up all sessions, unless 'tid' is supplied.
let docs = {$regex: /^sessionDoc/};
if (tid !== undefined) {
docs = "sessionDoc" + tid;
}
let sessionIds = db[collName].find({"_id": docs}, {_id: 0, id: 1}).toArray();
assert.commandWorked(db.runCommand({killSessions: sessionIds}));
}
|
