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
|
/**
* Tests that an aggregation cursor is killed when it is timed out by the ClientCursorMonitor.
*
* This test was designed to reproduce SERVER-25585.
*/
(function() {
'use strict';
// Cursor timeout on mongod is handled by a single thread/timer that will sleep for
// "clientCursorMonitorFrequencySecs" and add the sleep value to each operation's duration when
// it wakes up, timing out those whose "now() - last accessed since" time exceeds. A cursor
// timeout of 2 seconds with a monitor frequency of 1 second means an effective timeout period
// of 1 to 2 seconds.
const cursorTimeoutMs = 2000;
const cursorMonitorFrequencySecs = 1;
const options = {
setParameter: {
internalDocumentSourceCursorBatchSizeBytes: 1,
// We use the "cursorTimeoutMillis" server parameter to decrease how long it takes for a
// non-exhausted cursor to time out. We use the "clientCursorMonitorFrequencySecs"
// server parameter to make the ClientCursorMonitor that cleans up the timed out cursors
// run more often. The combination of these server parameters reduces the amount of time
// we need to wait within this test.
cursorTimeoutMillis: cursorTimeoutMs,
clientCursorMonitorFrequencySecs: cursorMonitorFrequencySecs,
}
};
const conn = MongoRunner.runMongod(options);
assert.neq(null, conn, 'mongod was unable to start up with options: ' + tojson(options));
const testDB = conn.getDB('test');
// We use a batch size of 2 to ensure that the mongo shell does not exhaust the cursor on its
// first batch.
const batchSize = 2;
const numMatches = 5;
function assertCursorTimesOut(collName, pipeline) {
const res = assert.commandWorked(testDB.runCommand({
aggregate: collName,
pipeline: pipeline,
cursor: {
batchSize: batchSize,
},
}));
let serverStatus = assert.commandWorked(testDB.serverStatus());
const expectedNumTimedOutCursors = serverStatus.metrics.cursor.timedOut + 1;
const cursor = new DBCommandCursor(testDB, res, batchSize);
// Wait until the idle cursor background job has killed the aggregation cursor.
assert.soon(
function() {
serverStatus = assert.commandWorked(testDB.serverStatus());
return +serverStatus.metrics.cursor.timedOut === expectedNumTimedOutCursors;
},
function() {
return "aggregation cursor failed to time out: " +
tojson(serverStatus.metrics.cursor);
});
assert.eq(0, serverStatus.metrics.cursor.open.total, tojson(serverStatus));
// We attempt to exhaust the aggregation cursor to verify that sending a getMore returns an
// error due to the cursor being killed.
let err = assert.throws(function() {
cursor.itcount();
});
assert.eq(ErrorCodes.CursorNotFound, err.code, tojson(err));
}
assert.writeOK(testDB.source.insert({local: 1}));
for (let i = 0; i < numMatches; ++i) {
assert.writeOK(testDB.dest.insert({foreign: 1}));
}
// Test that a regular aggregation cursor is killed when the timeout is reached.
assertCursorTimesOut('dest', []);
// Test that an aggregation cursor with a $lookup stage is killed when the timeout is reached.
assertCursorTimesOut('source', [
{
$lookup: {
from: 'dest',
localField: 'local',
foreignField: 'foreign',
as: 'matches',
}
},
{
$unwind: "$matches",
},
]);
// Test that an aggregation cursor with nested $lookup stages is killed when the timeout is
// reached.
assertCursorTimesOut('source', [
{
$lookup: {
from: 'dest',
let : {local1: "$local"},
pipeline: [
{$match: {$expr: {$eq: ["$foreign", "$$local1"]}}},
{
$lookup: {
from: 'source',
let : {foreign1: "$foreign"},
pipeline: [{$match: {$expr: {$eq: ["$local", "$$foreign1"]}}}],
as: 'matches2'
}
},
{
$unwind: "$matches2",
},
],
as: 'matches1',
}
},
{
$unwind: "$matches1",
},
]);
MongoRunner.stopMongod(conn);
})();
|
