summaryrefslogtreecommitdiff
path: root/src/mongo/executor/task_executor_cursor_test.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'src/mongo/executor/task_executor_cursor_test.cpp')
-rw-r--r--src/mongo/executor/task_executor_cursor_test.cpp1061
1 files changed, 301 insertions, 760 deletions
diff --git a/src/mongo/executor/task_executor_cursor_test.cpp b/src/mongo/executor/task_executor_cursor_test.cpp
index b99edbf715d..09d1a1c301b 100644
--- a/src/mongo/executor/task_executor_cursor_test.cpp
+++ b/src/mongo/executor/task_executor_cursor_test.cpp
@@ -32,564 +32,65 @@
#include "mongo/platform/basic.h"
#include "mongo/db/concurrency/locker_noop_client_observer.h"
-#include "mongo/executor/pinned_connection_task_executor.h"
-#include "mongo/executor/pinned_connection_task_executor_test_fixture.h"
#include "mongo/executor/task_executor_cursor.h"
#include "mongo/executor/thread_pool_task_executor_test_fixture.h"
-#include "mongo/rpc/op_msg_rpc_impls.h"
#include "mongo/unittest/bson_test_util.h"
-#include "mongo/unittest/thread_assertion_monitor.h"
#include "mongo/unittest/unittest.h"
namespace mongo {
namespace executor {
namespace {
-BSONObj buildCursorResponse(StringData fieldName, size_t start, size_t end, size_t cursorId) {
- BSONObjBuilder bob;
- {
- BSONObjBuilder cursor(bob.subobjStart("cursor"));
- {
- BSONArrayBuilder batch(cursor.subarrayStart(fieldName));
-
- for (size_t i = start; i <= end; ++i) {
- BSONObjBuilder doc(batch.subobjStart());
- doc.append("x", int(i));
- }
- }
- cursor.append("id", (long long)(cursorId));
- cursor.append("ns", "test.test");
- }
- bob.append("ok", int(1));
- return bob.obj();
-}
-
-BSONObj buildMultiCursorResponse(StringData fieldName,
- size_t start,
- size_t end,
- std::vector<size_t> cursorIds) {
- BSONObjBuilder bob;
- {
- BSONArrayBuilder cursors;
- int baseCursorValue = 1;
- for (auto cursorId : cursorIds) {
- BSONObjBuilder cursor;
- BSONArrayBuilder batch;
- ASSERT(start < end && end < INT_MAX);
- for (size_t i = start; i <= end; ++i) {
- batch.append(BSON("x" << static_cast<int>(i) * baseCursorValue).getOwned());
- }
- cursor.append(fieldName, batch.arr());
- cursor.append("id", (long long)(cursorId));
- cursor.append("ns", "test.test");
- auto cursorObj = BSON("cursor" << cursor.done() << "ok" << 1);
- cursors.append(cursorObj.getOwned());
- ++baseCursorValue;
- }
- bob.append("cursors", cursors.arr());
- }
- bob.append("ok", 1);
- return bob.obj();
-}
-
/**
* Fixture for the task executor cursor tests which offers some convenience methods to help with
- * scheduling responses. Uses the CRTP pattern so that the tests can be shared between child-classes
- * that provide their own implementations of the network-mocking needed for the tests.
+ * scheduling responses
*/
-template <typename Derived, typename Base>
-class TaskExecutorCursorTestFixture : public Base {
+class TaskExecutorCursorFixture : public ThreadPoolExecutorTest {
public:
- TaskExecutorCursorTestFixture() {
+ TaskExecutorCursorFixture() {
serviceCtx->registerClientObserver(std::make_unique<LockerNoopClientObserver>());
}
void setUp() override {
- Base::setUp();
+ ThreadPoolExecutorTest::setUp();
+
client = serviceCtx->makeClient("TaskExecutorCursorTest");
opCtx = client->makeOperationContext();
- static_cast<Derived*>(this)->postSetUp();
+
+ launchExecutorThread();
}
void tearDown() override {
opCtx.reset();
client.reset();
- Base::tearDown();
+ ThreadPoolExecutorTest::tearDown();
}
BSONObj scheduleSuccessfulCursorResponse(StringData fieldName,
size_t start,
size_t end,
size_t cursorId) {
- return static_cast<Derived*>(this)->scheduleSuccessfulCursorResponse(
- fieldName, start, end, cursorId);
- }
-
- BSONObj scheduleSuccessfulMultiCursorResponse(StringData fieldName,
- size_t start,
- size_t end,
- std::vector<size_t> cursorIds) {
- return static_cast<Derived*>(this)->scheduleSuccessfulMultiCursorResponse(
- fieldName, start, end, cursorIds);
- }
-
- void scheduleErrorResponse(Status error) {
- return static_cast<Derived*>(this)->scheduleErrorResponse(error);
- }
- void blackHoleNextOutgoingRequest() {
- return static_cast<Derived*>(this)->blackHoleNextOutgoingRequest();
- }
-
- BSONObj scheduleSuccessfulKillCursorResponse(size_t cursorId) {
- return static_cast<Derived*>(this)->scheduleSuccessfulKillCursorResponse(cursorId);
- }
-
- TaskExecutorCursor makeTec(RemoteCommandRequest rcr,
- TaskExecutorCursor::Options&& options = {}) {
- return static_cast<Derived*>(this)->makeTec(rcr, std::move(options));
- }
-
- bool hasReadyRequests() {
- return static_cast<Derived*>(this)->hasReadyRequests();
- }
-
- Base& asBase() {
- return *this;
- }
-
- /**
- * Ensure we work for a single simple batch
- */
- void SingleBatchWorksTest() {
- const auto findCmd = BSON("find"
- << "test"
- << "batchSize" << 2);
- const CursorId cursorId = 0;
-
- RemoteCommandRequest rcr(HostAndPort("localhost"), "test", findCmd, opCtx.get());
-
- TaskExecutorCursor tec = makeTec(rcr);
-
- ASSERT_BSONOBJ_EQ(findCmd, scheduleSuccessfulCursorResponse("firstBatch", 1, 2, cursorId));
-
- ASSERT_EQUALS(tec.getNext(opCtx.get()).value()["x"].Int(), 1);
-
- ASSERT_FALSE(hasReadyRequests());
-
- ASSERT_EQUALS(tec.getNext(opCtx.get()).value()["x"].Int(), 2);
-
- ASSERT_FALSE(tec.getNext(opCtx.get()));
- }
-
- /**
- * Ensure the firstBatch can be read correctly when multiple cursors are returned.
- */
- void MultipleCursorsSingleBatchSucceedsTest() {
- const auto aggCmd = BSON("aggregate"
- << "test"
- << "pipeline"
- << BSON_ARRAY(BSON("returnMultipleCursors" << true)));
-
- RemoteCommandRequest rcr(HostAndPort("localhost"), "test", aggCmd, opCtx.get());
-
- TaskExecutorCursor tec = makeTec(rcr);
-
- ASSERT_BSONOBJ_EQ(aggCmd,
- scheduleSuccessfulMultiCursorResponse("firstBatch", 1, 2, {0, 0}));
-
- ASSERT_EQUALS(tec.getNext(opCtx.get()).value()["x"].Int(), 1);
-
- ASSERT_EQUALS(tec.getNext(opCtx.get()).value()["x"].Int(), 2);
-
- ASSERT_FALSE(tec.getNext(opCtx.get()));
-
- auto cursorVec = tec.releaseAdditionalCursors();
- ASSERT_EQUALS(cursorVec.size(), 1);
- auto secondCursor = std::move(cursorVec[0]);
-
- ASSERT_EQUALS(secondCursor.getNext(opCtx.get()).value()["x"].Int(), 2);
- ASSERT_EQUALS(secondCursor.getNext(opCtx.get()).value()["x"].Int(), 4);
- ASSERT_FALSE(hasReadyRequests());
-
- ASSERT_FALSE(secondCursor.getNext(opCtx.get()));
- }
- /**
- * The operation context under which we send the original cursor-establishing command
- * can be destructed before getNext is called with new opCtx. Ensure that 'child'
- * TaskExecutorCursors created from the original TEC's multi-cursor-response can safely
- * operate if this happens/don't try and use the now-destroyed operation context.
- * See SERVER-69702 for context
- */
- void ChildTaskExecutorCursorsAreSafeIfOriginalOpCtxDestructedTest() {
- auto lsid = makeLogicalSessionIdForTest();
- opCtx->setLogicalSessionId(lsid);
- const auto aggCmd = BSON("aggregate"
- << "test"
- << "pipeline"
- << BSON_ARRAY(BSON("returnMultipleCursors" << true)));
- RemoteCommandRequest rcr(HostAndPort("localhost"), "test", aggCmd, opCtx.get());
- TaskExecutorCursor tec = makeTec(rcr);
- auto expected = BSON("aggregate"
- << "test"
- << "pipeline" << BSON_ARRAY(BSON("returnMultipleCursors" << true))
- << "lsid" << lsid.toBSON());
- ASSERT_BSONOBJ_EQ(expected,
- scheduleSuccessfulMultiCursorResponse("firstBatch", 1, 2, {0, 0}));
- // Before calling getNext (and therefore spawning child TECs), destroy the opCtx
- // we used to send the initial query and make a new one.
- opCtx.reset();
- opCtx = client->makeOperationContext();
- opCtx->setLogicalSessionId(lsid);
- // Use the new opCtx to call getNext. The child TECs should not attempt to read from the
- // now dead original opCtx.
- ASSERT_EQUALS(tec.getNext(opCtx.get()).value()["x"].Int(), 1);
-
- ASSERT_EQUALS(tec.getNext(opCtx.get()).value()["x"].Int(), 2);
-
- ASSERT_FALSE(tec.getNext(opCtx.get()));
-
- auto cursorVec = tec.releaseAdditionalCursors();
- ASSERT_EQUALS(cursorVec.size(), 1);
- auto secondCursor = std::move(cursorVec[0]);
-
- ASSERT_EQUALS(secondCursor.getNext(opCtx.get()).value()["x"].Int(), 2);
- ASSERT_EQUALS(secondCursor.getNext(opCtx.get()).value()["x"].Int(), 4);
- ASSERT_FALSE(hasReadyRequests());
-
- ASSERT_FALSE(secondCursor.getNext(opCtx.get()));
- }
-
- void MultipleCursorsGetMoreWorksTest() {
- const auto aggCmd = BSON("aggregate"
- << "test"
- << "pipeline"
- << BSON_ARRAY(BSON("returnMultipleCursors" << true)));
-
- std::vector<size_t> cursorIds{1, 2};
- RemoteCommandRequest rcr(HostAndPort("localhost"), "test", aggCmd, opCtx.get());
-
- TaskExecutorCursor tec = makeTec(rcr);
-
- ASSERT_BSONOBJ_EQ(aggCmd,
- scheduleSuccessfulMultiCursorResponse("firstBatch", 1, 2, cursorIds));
-
- ASSERT_EQUALS(tec.getNext(opCtx.get()).value()["x"].Int(), 1);
-
- ASSERT_EQUALS(tec.getNext(opCtx.get()).value()["x"].Int(), 2);
-
- auto cursorVec = tec.releaseAdditionalCursors();
- ASSERT_EQUALS(cursorVec.size(), 1);
-
- // If we try to getNext() at this point, we are interruptible and can timeout
- ASSERT_THROWS_CODE(opCtx->runWithDeadline(Date_t::now() + Milliseconds(100),
- ErrorCodes::ExceededTimeLimit,
- [&] { tec.getNext(opCtx.get()); }),
- DBException,
- ErrorCodes::ExceededTimeLimit);
-
- // We can pick up after that interruption though
- ASSERT_BSONOBJ_EQ(BSON("getMore" << 1LL << "collection"
- << "test"),
- scheduleSuccessfulCursorResponse("nextBatch", 3, 5, cursorIds[0]));
-
- // Repeat for second cursor.
- auto secondCursor = std::move(cursorVec[0]);
-
- ASSERT_EQUALS(secondCursor.getNext(opCtx.get()).value()["x"].Int(), 2);
- ASSERT_EQUALS(secondCursor.getNext(opCtx.get()).value()["x"].Int(), 4);
-
- ASSERT_THROWS_CODE(opCtx->runWithDeadline(Date_t::now() + Milliseconds(100),
- ErrorCodes::ExceededTimeLimit,
- [&] { secondCursor.getNext(opCtx.get()); }),
- DBException,
- ErrorCodes::ExceededTimeLimit);
-
- ASSERT_BSONOBJ_EQ(BSON("getMore" << 2LL << "collection"
- << "test"),
- scheduleSuccessfulCursorResponse("nextBatch", 6, 8, cursorIds[1]));
- // Read second batch, then schedule EOF on both cursors.
- // Then read final document for each.
- ASSERT_EQUALS(tec.getNext(opCtx.get()).value()["x"].Int(), 3);
- ASSERT_EQUALS(tec.getNext(opCtx.get()).value()["x"].Int(), 4);
- ASSERT_EQUALS(tec.getNext(opCtx.get()).value()["x"].Int(), 5);
- scheduleSuccessfulCursorResponse("nextBatch", 6, 6, 0);
- ASSERT_EQUALS(tec.getNext(opCtx.get()).value()["x"].Int(), 6);
-
- ASSERT_EQUALS(secondCursor.getNext(opCtx.get()).value()["x"].Int(), 6);
- ASSERT_EQUALS(secondCursor.getNext(opCtx.get()).value()["x"].Int(), 7);
- ASSERT_EQUALS(secondCursor.getNext(opCtx.get()).value()["x"].Int(), 8);
- scheduleSuccessfulCursorResponse("nextBatch", 12, 12, 0);
- ASSERT_EQUALS(secondCursor.getNext(opCtx.get()).value()["x"].Int(), 12);
-
- // Shouldn't have any more requests, both cursors are closed.
- ASSERT_FALSE(hasReadyRequests());
-
- ASSERT_FALSE(tec.getNext(opCtx.get()));
- ASSERT_FALSE(secondCursor.getNext(opCtx.get()));
- }
-
- /**
- * Ensure we work if find fails (and that we receive the error code it failed with)
- */
- void FailureInFindTest() {
- const auto findCmd = BSON("find"
- << "test"
- << "batchSize" << 2);
-
- RemoteCommandRequest rcr(HostAndPort("localhost"), "test", findCmd, opCtx.get());
-
- TaskExecutorCursor tec = makeTec(rcr);
-
- scheduleErrorResponse(Status(ErrorCodes::BadValue, "an error"));
-
- ASSERT_THROWS_CODE(tec.getNext(opCtx.get()), DBException, ErrorCodes::BadValue);
- }
-
-
- /**
- * Ensure multiple batches works correctly
- */
- void MultipleBatchesWorksTest() {
- const auto findCmd = BSON("find"
- << "test"
- << "batchSize" << 2);
- CursorId cursorId = 1;
-
- RemoteCommandRequest rcr(HostAndPort("localhost"), "test", findCmd, opCtx.get());
-
- TaskExecutorCursor tec = makeTec(rcr, [] {
- TaskExecutorCursor::Options opts;
- opts.batchSize = 3;
- return opts;
- }());
-
- scheduleSuccessfulCursorResponse("firstBatch", 1, 2, cursorId);
-
- ASSERT_EQUALS(tec.getNext(opCtx.get()).value()["x"].Int(), 1);
-
- // ASSERT(hasReadyRequests());
-
- ASSERT_EQUALS(tec.getNext(opCtx.get()).value()["x"].Int(), 2);
-
- // If we try to getNext() at this point, we are interruptible and can timeout
- ASSERT_THROWS_CODE(opCtx->runWithDeadline(Date_t::now() + Milliseconds(100),
- ErrorCodes::ExceededTimeLimit,
- [&] { tec.getNext(opCtx.get()); }),
- DBException,
- ErrorCodes::ExceededTimeLimit);
-
- // We can pick up after that interruption though
- ASSERT_BSONOBJ_EQ(BSON("getMore" << 1LL << "collection"
- << "test"
- << "batchSize" << 3),
- scheduleSuccessfulCursorResponse("nextBatch", 3, 5, cursorId));
-
- ASSERT_EQUALS(tec.getNext(opCtx.get()).value()["x"].Int(), 3);
- ASSERT_EQUALS(tec.getNext(opCtx.get()).value()["x"].Int(), 4);
- ASSERT_EQUALS(tec.getNext(opCtx.get()).value()["x"].Int(), 5);
-
- cursorId = 0;
- scheduleSuccessfulCursorResponse("nextBatch", 6, 6, cursorId);
-
- ASSERT_EQUALS(tec.getNext(opCtx.get()).value()["x"].Int(), 6);
-
- // We don't issue extra getmores after returning a 0 cursor id
- ASSERT_FALSE(hasReadyRequests());
-
- ASSERT_FALSE(tec.getNext(opCtx.get()));
- }
-
- /**
- * Ensure we allow empty firstBatch.
- */
- void EmptyFirstBatchTest() {
- const auto findCmd = BSON("find"
- << "test"
- << "batchSize" << 2);
- const auto getMoreCmd = BSON("getMore" << 1LL << "collection"
- << "test"
- << "batchSize" << 3);
- const CursorId cursorId = 1;
-
- RemoteCommandRequest rcr(HostAndPort("localhost"), "test", findCmd, opCtx.get());
-
- TaskExecutorCursor tec = makeTec(rcr, [] {
- TaskExecutorCursor::Options opts;
- opts.batchSize = 3;
- return opts;
- }());
-
- // Schedule a cursor response with an empty "firstBatch". Use end < start so we don't
- // append any doc to "firstBatch".
- ASSERT_BSONOBJ_EQ(findCmd, scheduleSuccessfulCursorResponse("firstBatch", 1, 0, cursorId));
-
- stdx::thread th([&] {
- // Wait for the getMore run by the getNext() below to be ready, and schedule a
- // cursor response with a non-empty "nextBatch".
- while (!hasReadyRequests()) {
- sleepmillis(10);
- }
-
- ASSERT_BSONOBJ_EQ(getMoreCmd, scheduleSuccessfulCursorResponse("nextBatch", 1, 1, 0));
- });
-
- // Verify that the first doc is the doc from the second batch.
- ASSERT_EQUALS(tec.getNext(opCtx.get()).value()["x"].Int(), 1);
-
- th.join();
- }
-
- /**
- * Ensure we allow any empty non-initial batch.
- */
- void EmptyNonInitialBatchTest() {
- const auto findCmd = BSON("find"
- << "test"
- << "batchSize" << 2);
- const auto getMoreCmd = BSON("getMore" << 1LL << "collection"
- << "test"
- << "batchSize" << 3);
- const CursorId cursorId = 1;
-
- RemoteCommandRequest rcr(HostAndPort("localhost"), "test", findCmd, opCtx.get());
-
- TaskExecutorCursor tec = makeTec(rcr, [] {
- TaskExecutorCursor::Options opts;
- opts.batchSize = 3;
- return opts;
- }());
-
- // Schedule a cursor response with a non-empty "firstBatch".
- ASSERT_BSONOBJ_EQ(findCmd, scheduleSuccessfulCursorResponse("firstBatch", 1, 1, cursorId));
-
- ASSERT_EQUALS(tec.getNext(opCtx.get()).value()["x"].Int(), 1);
-
- // Schedule two consecutive cursor responses with empty "nextBatch". Use end < start so
- // we don't append any doc to "nextBatch".
- ASSERT_BSONOBJ_EQ(getMoreCmd,
- scheduleSuccessfulCursorResponse("nextBatch", 1, 0, cursorId));
-
- stdx::thread th([&] {
- // Wait for the first getMore run by the getNext() below to be ready, and schedule a
- // cursor response with a non-empty "nextBatch".
- while (!hasReadyRequests()) {
- sleepmillis(10);
- }
-
- ASSERT_BSONOBJ_EQ(getMoreCmd,
- scheduleSuccessfulCursorResponse("nextBatch", 1, 0, cursorId));
-
- // Wait for the second getMore run by the getNext() below to be ready, and schedule a
- // cursor response with a non-empty "nextBatch".
- while (!hasReadyRequests()) {
- sleepmillis(10);
- }
-
- ASSERT_BSONOBJ_EQ(getMoreCmd, scheduleSuccessfulCursorResponse("nextBatch", 2, 2, 0));
- });
-
- // Verify that the next doc is the doc from the fourth batch.
- ASSERT_EQUALS(tec.getNext(opCtx.get()).value()["x"].Int(), 2);
-
- th.join();
- }
-
- /**
- * Test that if 'preFetchNextBatch' is false, the TaskExecutorCursor does not request GetMores
- * until the current batch is exhausted and 'getNext()' is invoked.
- */
- void NoPrefetchGetMore() {
- unittest::threadAssertionMonitoredTest([&](auto& monitor) {
- CursorId cursorId = 1;
- RemoteCommandRequest rcr(HostAndPort("localhost"),
- "test",
- BSON("search"
- << "foo"),
- opCtx.get());
-
- // The lambda that will be used to augment the getMore request sent below is passed into
- // the TEC constructor.
- auto augmentGetMore = [](BSONObjBuilder& bob) { bob.append("test", 1); };
-
- // Construction of the TaskExecutorCursor enqueues a request in the
- // NetworkInterfaceMock.
- TaskExecutorCursor tec = makeTec(rcr, [&augmentGetMore] {
- TaskExecutorCursor::Options opts;
- opts.batchSize = 2;
- opts.preFetchNextBatch = false;
- opts.getMoreAugmentationWriter = augmentGetMore;
- return opts;
- }());
-
- // Mock the response for the first batch.
- scheduleSuccessfulCursorResponse("firstBatch", 1, 2, cursorId);
-
- // Exhaust the first batch.
- ASSERT_EQUALS(tec.getNext(opCtx.get()).value()["x"].Int(), 1);
- ASSERT_EQUALS(tec.getNext(opCtx.get()).value()["x"].Int(), 2);
-
- // Assert that the TaskExecutorCursor has not requested a GetMore. This enforces that
- // 'preFetchNextBatch' works as expected.
- ASSERT_FALSE(hasReadyRequests());
-
- // As soon as 'getNext()' is invoked, the TaskExecutorCursor will try to send a GetMore
- // and that will block this thread in the NetworkInterfaceMock until there is a
- // scheduled response. However, we cannot schedule the cursor response on the main
- // thread before we call 'getNext()' as that will cause the NetworkInterfaceMock to
- // block until there is request enqueued ('getNext()' is the function which will enqueue
- // such as request). To avoid this deadlock, we start a new thread which will schedule a
- // response on the NetworkInterfaceMock.
- auto responseSchedulerThread = monitor.spawn([&] {
- auto recievedGetMoreCmd = scheduleSuccessfulCursorResponse("nextBatch", 3, 4, 0);
-
- // Assert that the command processed for the above response matches with the
- // lambda to augment the getMore command used during construction of the TEC
- // above.
- const auto expectedGetMoreCmd = BSON("getMore" << 1LL << "collection"
- << "test"
- << "batchSize" << 2 << "test" << 1);
- ASSERT_BSONOBJ_EQ(expectedGetMoreCmd, recievedGetMoreCmd);
- });
-
- // Schedules the GetMore request and exhausts the cursor.
- ASSERT_EQUALS(tec.getNext(opCtx.get()).value()["x"].Int(), 3);
- ASSERT_EQUALS(tec.getNext(opCtx.get()).value()["x"].Int(), 4);
-
- // Joining the thread which schedules the cursor response for the GetMore here forces
- // the destructor of NetworkInterfaceMock::InNetworkGuard to run, which ensures that the
- // 'NetworkInterfaceMock' stops executing as the network thread. This is required before
- // we invoke 'hasReadyRequests()' which enters the network again.
- responseSchedulerThread.join();
-
- // Assert no GetMore is requested.
- ASSERT_FALSE(hasReadyRequests());
- });
- }
-
- ServiceContext::UniqueServiceContext serviceCtx = ServiceContext::make();
- ServiceContext::UniqueClient client;
- ServiceContext::UniqueOperationContext opCtx;
-};
-
-class NonPinningTaskExecutorCursorTestFixture
- : public TaskExecutorCursorTestFixture<NonPinningTaskExecutorCursorTestFixture,
- ThreadPoolExecutorTest> {
-public:
- void postSetUp() {
- launchExecutorThread();
- }
-
- virtual BSONObj scheduleSuccessfulCursorResponse(StringData fieldName,
- size_t start,
- size_t end,
- size_t cursorId) {
NetworkInterfaceMock::InNetworkGuard ing(getNet());
+ BSONObjBuilder bob;
+ {
+ BSONObjBuilder cursor(bob.subobjStart("cursor"));
+ {
+ BSONArrayBuilder batch(cursor.subarrayStart(fieldName));
+
+ for (size_t i = start; i <= end; ++i) {
+ BSONObjBuilder doc(batch.subobjStart());
+ doc.append("x", int(i));
+ }
+ }
+ cursor.append("id", (long long)(cursorId));
+ cursor.append("ns", "test.test");
+ }
+ bob.append("ok", int(1));
ASSERT(getNet()->hasReadyRequests());
- auto rcr = getNet()->scheduleSuccessfulResponse(
- buildCursorResponse(fieldName, start, end, cursorId));
+ auto rcr = getNet()->scheduleSuccessfulResponse(bob.obj());
getNet()->runReadyNetworkOperations();
return rcr.cmdObj.getOwned();
@@ -601,10 +102,30 @@ public:
std::vector<size_t> cursorIds) {
NetworkInterfaceMock::InNetworkGuard ing(getNet());
+ BSONObjBuilder bob;
+ {
+ BSONArrayBuilder cursors;
+ int baseCursorValue = 1;
+ for (auto cursorId : cursorIds) {
+ BSONObjBuilder cursor;
+ BSONArrayBuilder batch;
+ ASSERT(start < end && end < INT_MAX);
+ for (size_t i = start; i <= end; ++i) {
+ batch.append(BSON("x" << static_cast<int>(i) * baseCursorValue).getOwned());
+ }
+ cursor.append(fieldName, batch.arr());
+ cursor.append("id", (long long)(cursorId));
+ cursor.append("ns", "test.test");
+ auto cursorObj = BSON("cursor" << cursor.done() << "ok" << 1);
+ cursors.append(cursorObj.getOwned());
+ ++baseCursorValue;
+ }
+ bob.append("cursors", cursors.arr());
+ }
+ bob.append("ok", 1);
ASSERT(getNet()->hasReadyRequests());
- auto rcr = getNet()->scheduleSuccessfulResponse(
- buildMultiCursorResponse(fieldName, start, end, cursorIds));
+ auto rcr = getNet()->scheduleSuccessfulResponse(bob.obj());
getNet()->runReadyNetworkOperations();
return rcr.cmdObj.getOwned();
@@ -623,235 +144,167 @@ public:
return rcr.cmdObj.getOwned();
}
- void scheduleErrorResponse(Status error) {
- NetworkInterfaceMock::InNetworkGuard ing(getNet());
-
- ASSERT(getNet()->hasReadyRequests());
- getNet()->scheduleErrorResponse(error);
- getNet()->runReadyNetworkOperations();
- }
-
bool hasReadyRequests() {
NetworkInterfaceMock::InNetworkGuard ing(getNet());
return getNet()->hasReadyRequests();
}
- void blackHoleNextOutgoingRequest() {
- NetworkInterfaceMock::InNetworkGuard guard(getNet());
- getNet()->blackHole(getNet()->getFrontOfUnscheduledQueue());
- }
-
- TaskExecutorCursor makeTec(RemoteCommandRequest rcr,
- TaskExecutorCursor::Options&& options = {}) {
- options.pinConnection = false;
- return TaskExecutorCursor(getExecutorPtr(), rcr, std::move(options));
- }
+ ServiceContext::UniqueServiceContext serviceCtx = ServiceContext::make();
+ ServiceContext::UniqueClient client;
+ ServiceContext::UniqueOperationContext opCtx;
};
-class PinnedConnTaskExecutorCursorTestFixture
- : public TaskExecutorCursorTestFixture<PinnedConnTaskExecutorCursorTestFixture,
- PinnedConnectionTaskExecutorTest> {
-public:
- void postSetUp() {}
-
- BSONObj scheduleResponse(StatusWith<BSONObj> response) {
- int32_t responseToId;
- BSONObj cmdObjReceived;
- auto pf = makePromiseFuture<void>();
- expectSinkMessage([&](Message m) {
- responseToId = m.header().getId();
- auto opMsg = OpMsgRequest::parse(m);
- cmdObjReceived = opMsg.body.removeField("$db").getOwned();
- pf.promise.emplaceValue();
- return Status::OK();
- });
- // Wait until we recieved the command request.
- pf.future.get();
-
- // Now we expect source message to be called and provide the response
- expectSourceMessage([=]() {
- rpc::OpMsgReplyBuilder replyBuilder;
- replyBuilder.setCommandReply(response);
- auto message = replyBuilder.done();
- message.header().setResponseToMsgId(responseToId);
- return message;
- });
- return cmdObjReceived;
- }
+/**
+ * Ensure we work for a single simple batch
+ */
+TEST_F(TaskExecutorCursorFixture, SingleBatchWorks) {
+ const auto findCmd = BSON("find"
+ << "test"
+ << "batchSize" << 2);
+ const CursorId cursorId = 0;
- BSONObj scheduleSuccessfulCursorResponse(StringData fieldName,
- size_t start,
- size_t end,
- size_t cursorId) {
- auto cursorResponse = buildCursorResponse(fieldName, start, end, cursorId);
- return scheduleResponse(cursorResponse);
- }
+ RemoteCommandRequest rcr(HostAndPort("localhost"), "test", findCmd, opCtx.get());
- BSONObj scheduleSuccessfulMultiCursorResponse(StringData fieldName,
- size_t start,
- size_t end,
- std::vector<size_t> cursorIds) {
- auto cursorResponse = buildMultiCursorResponse(fieldName, start, end, cursorIds);
- return scheduleResponse(cursorResponse);
- }
+ TaskExecutorCursor tec(&getExecutor(), rcr);
- void scheduleErrorResponse(Status error) {
- scheduleResponse(error);
- }
+ ASSERT_BSONOBJ_EQ(findCmd, scheduleSuccessfulCursorResponse("firstBatch", 1, 2, cursorId));
- BSONObj scheduleSuccessfulKillCursorResponse(size_t cursorId) {
+ ASSERT_EQUALS(tec.getNext(opCtx.get()).get()["x"].Int(), 1);
- auto cursorResponse =
- BSON("cursorsKilled" << BSON_ARRAY((long long)(cursorId)) << "cursorsNotFound"
- << BSONArray() << "cursorsAlive" << BSONArray() << "cursorsUnknown"
- << BSONArray() << "ok" << 1);
- return scheduleResponse(cursorResponse);
- }
+ ASSERT_FALSE(hasReadyRequests());
- TaskExecutorCursor makeTec(RemoteCommandRequest rcr,
- TaskExecutorCursor::Options&& options = {}) {
- options.pinConnection = true;
- return TaskExecutorCursor(getExecutorPtr(), rcr, std::move(options));
- }
+ ASSERT_EQUALS(tec.getNext(opCtx.get()).get()["x"].Int(), 2);
- bool hasReadyRequests() {
- return asBase().hasReadyRequests();
- }
+ ASSERT_FALSE(tec.getNext(opCtx.get()));
+}
- void blackHoleNextOutgoingRequest() {
- auto pf = makePromiseFuture<void>();
- expectSinkMessage([&](Message m) {
- pf.promise.emplaceValue();
- return Status(ErrorCodes::SocketException, "test");
- });
- pf.future.get();
- }
-};
+/**
+ * Ensure the firstBatch can be read correctly when multiple cursors are returned.
+ */
+TEST_F(TaskExecutorCursorFixture, MultipleCursorsSingleBatchSucceeds) {
+ const auto aggCmd = BSON("aggregate"
+ << "test"
+ << "pipeline" << BSON_ARRAY(BSON("returnMultipleCursors" << true)));
-class NoPrefetchTaskExecutorCursorTestFixture : public NonPinningTaskExecutorCursorTestFixture {
-public:
- TaskExecutorCursor makeTec(RemoteCommandRequest rcr,
- TaskExecutorCursor::Options&& options = {}) {
- options.preFetchNextBatch = false;
- return TaskExecutorCursor(getExecutorPtr(), rcr, std::move(options));
- }
+ RemoteCommandRequest rcr(HostAndPort("localhost"), "test", aggCmd, opCtx.get());
- BSONObj scheduleSuccessfulCursorResponse(StringData fieldName,
- size_t start,
- size_t end,
- size_t cursorId) {
- NetworkInterfaceMock::InNetworkGuard ing(getNet());
- // Don't assert that the network has requests like we do in other classes. This is to enable
- // the test in 'NoPrefetchGetMore'.
- auto rcr =
- ing->scheduleSuccessfulResponse(buildCursorResponse(fieldName, start, end, cursorId));
- ing->runReadyNetworkOperations();
- return rcr.cmdObj.getOwned();
- }
-};
+ TaskExecutorCursor tec(&getExecutor(), rcr);
-class NoPrefetchPinnedTaskExecutorCursorTestFixture
- : public PinnedConnTaskExecutorCursorTestFixture {
-public:
- TaskExecutorCursor makeTec(RemoteCommandRequest rcr,
- TaskExecutorCursor::Options&& options = {}) {
- options.preFetchNextBatch = false;
- options.pinConnection = true;
- return TaskExecutorCursor(getExecutorPtr(), rcr, std::move(options));
- }
-};
+ ASSERT_BSONOBJ_EQ(aggCmd, scheduleSuccessfulMultiCursorResponse("firstBatch", 1, 2, {0, 0}));
-TEST_F(NonPinningTaskExecutorCursorTestFixture, SingleBatchWorks) {
- SingleBatchWorksTest();
-}
+ ASSERT_EQUALS(tec.getNext(opCtx.get()).get()["x"].Int(), 1);
-TEST_F(NoPrefetchTaskExecutorCursorTestFixture, SingleBatchWorks) {
- SingleBatchWorksTest();
-}
+ ASSERT_EQUALS(tec.getNext(opCtx.get()).get()["x"].Int(), 2);
-TEST_F(NoPrefetchPinnedTaskExecutorCursorTestFixture, SingleBatchWorks) {
- SingleBatchWorksTest();
-}
+ ASSERT_FALSE(tec.getNext(opCtx.get()));
-TEST_F(PinnedConnTaskExecutorCursorTestFixture, SingleBatchWorks) {
- SingleBatchWorksTest();
-}
+ auto cursorVec = tec.releaseAdditionalCursors();
+ ASSERT_EQUALS(cursorVec.size(), 1);
+ auto secondCursor = std::move(cursorVec[0]);
-TEST_F(NonPinningTaskExecutorCursorTestFixture, MultipleCursorsSingleBatchSucceeds) {
- MultipleCursorsSingleBatchSucceedsTest();
-}
+ ASSERT_EQUALS(secondCursor.getNext(opCtx.get()).get()["x"].Int(), 2);
+ ASSERT_EQUALS(secondCursor.getNext(opCtx.get()).get()["x"].Int(), 4);
+ ASSERT_FALSE(hasReadyRequests());
-TEST_F(NoPrefetchTaskExecutorCursorTestFixture, MultipleCursorsSingleBatchSucceeds) {
- MultipleCursorsSingleBatchSucceedsTest();
+ ASSERT_FALSE(secondCursor.getNext(opCtx.get()));
}
-TEST_F(NoPrefetchPinnedTaskExecutorCursorTestFixture, MultipleCursorsSingleBatchSucceeds) {
- MultipleCursorsSingleBatchSucceedsTest();
-}
+TEST_F(TaskExecutorCursorFixture, MultipleCursorsGetMoreWorks) {
+ const auto aggCmd = BSON("aggregate"
+ << "test"
+ << "pipeline" << BSON_ARRAY(BSON("returnMultipleCursors" << true)));
-TEST_F(PinnedConnTaskExecutorCursorTestFixture, MultipleCursorsSingleBatchSucceeds) {
- MultipleCursorsSingleBatchSucceedsTest();
-}
+ std::vector<size_t> cursorIds{1, 2};
+ RemoteCommandRequest rcr(HostAndPort("localhost"), "test", aggCmd, opCtx.get());
-TEST_F(NonPinningTaskExecutorCursorTestFixture,
- ChildTaskExecutorCursorsAreSafeIfOriginalOpCtxDestructed) {
- ChildTaskExecutorCursorsAreSafeIfOriginalOpCtxDestructedTest();
-}
+ TaskExecutorCursor tec(&getExecutor(), rcr);
-TEST_F(NoPrefetchTaskExecutorCursorTestFixture,
- ChildTaskExecutorCursorsAreSafeIfOriginalOpCtxDestructed) {
- ChildTaskExecutorCursorsAreSafeIfOriginalOpCtxDestructedTest();
-}
+ ASSERT_BSONOBJ_EQ(aggCmd, scheduleSuccessfulMultiCursorResponse("firstBatch", 1, 2, cursorIds));
-TEST_F(NoPrefetchPinnedTaskExecutorCursorTestFixture,
- ChildTaskExecutorCursorsAreSafeIfOriginalOpCtxDestructed) {
- ChildTaskExecutorCursorsAreSafeIfOriginalOpCtxDestructedTest();
-}
+ ASSERT_EQUALS(tec.getNext(opCtx.get()).get()["x"].Int(), 1);
-TEST_F(PinnedConnTaskExecutorCursorTestFixture,
- ChildTaskExecutorCursorsAreSafeIfOriginalOpCtxDestructed) {
- ChildTaskExecutorCursorsAreSafeIfOriginalOpCtxDestructedTest();
-}
+ ASSERT_EQUALS(tec.getNext(opCtx.get()).get()["x"].Int(), 2);
-TEST_F(NonPinningTaskExecutorCursorTestFixture, MultipleCursorsGetMoreWorks) {
- MultipleCursorsGetMoreWorksTest();
-}
+ auto cursorVec = tec.releaseAdditionalCursors();
+ ASSERT_EQUALS(cursorVec.size(), 1);
-TEST_F(NoPrefetchTaskExecutorCursorTestFixture, MultipleCursorsGetMoreWorks) {
- MultipleCursorsGetMoreWorksTest();
-}
+ // If we try to getNext() at this point, we are interruptible and can timeout
+ ASSERT_THROWS_CODE(opCtx->runWithDeadline(Date_t::now() + Milliseconds(100),
+ ErrorCodes::ExceededTimeLimit,
+ [&] { tec.getNext(opCtx.get()); }),
+ DBException,
+ ErrorCodes::ExceededTimeLimit);
-TEST_F(NoPrefetchPinnedTaskExecutorCursorTestFixture, MultipleCursorsGetMoreWorks) {
- MultipleCursorsGetMoreWorksTest();
-}
+ // We can pick up after that interruption though
+ ASSERT_BSONOBJ_EQ(BSON("getMore" << 1LL << "collection"
+ << "test"),
+ scheduleSuccessfulCursorResponse("nextBatch", 3, 5, cursorIds[0]));
+
+ // Repeat for second cursor.
+ auto secondCursor = std::move(cursorVec[0]);
+
+ ASSERT_EQUALS(secondCursor.getNext(opCtx.get()).get()["x"].Int(), 2);
+ ASSERT_EQUALS(secondCursor.getNext(opCtx.get()).get()["x"].Int(), 4);
+
+ ASSERT_THROWS_CODE(opCtx->runWithDeadline(Date_t::now() + Milliseconds(100),
+ ErrorCodes::ExceededTimeLimit,
+ [&] { secondCursor.getNext(opCtx.get()); }),
+ DBException,
+ ErrorCodes::ExceededTimeLimit);
+
+ ASSERT_BSONOBJ_EQ(BSON("getMore" << 2LL << "collection"
+ << "test"),
+ scheduleSuccessfulCursorResponse("nextBatch", 6, 8, cursorIds[1]));
+ // Read second batch on both cursors.
+ ASSERT_EQUALS(tec.getNext(opCtx.get()).get()["x"].Int(), 3);
+ ASSERT_EQUALS(tec.getNext(opCtx.get()).get()["x"].Int(), 4);
+ ASSERT_EQUALS(tec.getNext(opCtx.get()).get()["x"].Int(), 5);
+ ASSERT_EQUALS(secondCursor.getNext(opCtx.get()).get()["x"].Int(), 6);
+ ASSERT_EQUALS(secondCursor.getNext(opCtx.get()).get()["x"].Int(), 7);
+ ASSERT_EQUALS(secondCursor.getNext(opCtx.get()).get()["x"].Int(), 8);
+
+ // Schedule EOF on both cursors.
+ scheduleSuccessfulCursorResponse("nextBatch", 6, 6, 0);
+ scheduleSuccessfulCursorResponse("nextBatch", 12, 12, 0);
+
+ // Read final document.
+ ASSERT_EQUALS(tec.getNext(opCtx.get()).get()["x"].Int(), 6);
+ ASSERT_EQUALS(secondCursor.getNext(opCtx.get()).get()["x"].Int(), 12);
+
+ // Shouldn't have any more requests, both cursors are closed.
+ ASSERT_FALSE(hasReadyRequests());
-TEST_F(PinnedConnTaskExecutorCursorTestFixture, MultipleCursorsGetMoreWorks) {
- MultipleCursorsGetMoreWorksTest();
+ ASSERT_FALSE(tec.getNext(opCtx.get()));
+ ASSERT_FALSE(secondCursor.getNext(opCtx.get()));
}
-TEST_F(NonPinningTaskExecutorCursorTestFixture, FailureInFind) {
- FailureInFindTest();
-}
+/**
+ * Ensure we work if find fails (and that we receive the error code it failed with)
+ */
+TEST_F(TaskExecutorCursorFixture, FailureInFind) {
+ const auto findCmd = BSON("find"
+ << "test"
+ << "batchSize" << 2);
-TEST_F(NoPrefetchTaskExecutorCursorTestFixture, FailureInFind) {
- FailureInFindTest();
-}
+ RemoteCommandRequest rcr(HostAndPort("localhost"), "test", findCmd, opCtx.get());
-TEST_F(NoPrefetchPinnedTaskExecutorCursorTestFixture, FailureInFind) {
- FailureInFindTest();
-}
+ TaskExecutorCursor tec(&getExecutor(), rcr);
-TEST_F(PinnedConnTaskExecutorCursorTestFixture, FailureInFind) {
- FailureInFindTest();
+ {
+ NetworkInterfaceMock::InNetworkGuard ing(getNet());
+
+ ASSERT(getNet()->hasReadyRequests());
+ getNet()->scheduleErrorResponse(Status(ErrorCodes::BadValue, "an error"));
+ getNet()->runReadyNetworkOperations();
+ }
+
+ ASSERT_THROWS_CODE(tec.getNext(opCtx.get()), DBException, ErrorCodes::BadValue);
}
/**
* Ensure early termination of the cursor calls killCursor (if we know about the cursor id)
- * Only applicable to the unpinned case - if the connection is pinned, and a getMore is
- * in progress and/or fails, the most we can do is kill the connection. We can't re-use
- * the connection to send killCursors.
*/
-TEST_F(NonPinningTaskExecutorCursorTestFixture, EarlyReturnKillsCursor) {
+TEST_F(TaskExecutorCursorFixture, EarlyReturnKillsCursor) {
const auto findCmd = BSON("find"
<< "test"
<< "batchSize" << 2);
@@ -860,76 +313,172 @@ TEST_F(NonPinningTaskExecutorCursorTestFixture, EarlyReturnKillsCursor) {
RemoteCommandRequest rcr(HostAndPort("localhost"), "test", findCmd, opCtx.get());
{
- TaskExecutorCursor tec = makeTec(rcr);
+ TaskExecutorCursor tec(&getExecutor(), rcr);
scheduleSuccessfulCursorResponse("firstBatch", 1, 2, cursorId);
ASSERT(tec.getNext(opCtx.get()));
-
- // Black hole the pending `getMore` operation scheduled by the `TaskExecutorCursor`.
- blackHoleNextOutgoingRequest();
}
-
ASSERT_BSONOBJ_EQ(BSON("killCursors"
<< "test"
<< "cursors" << BSON_ARRAY(1)),
scheduleSuccessfulKillCursorResponse(1));
}
-TEST_F(NonPinningTaskExecutorCursorTestFixture, MultipleBatchesWorks) {
- MultipleBatchesWorksTest();
-}
+/**
+ * Ensure multiple batches works correctly
+ */
+TEST_F(TaskExecutorCursorFixture, MultipleBatchesWorks) {
+ const auto findCmd = BSON("find"
+ << "test"
+ << "batchSize" << 2);
+ CursorId cursorId = 1;
-TEST_F(NoPrefetchTaskExecutorCursorTestFixture, MultipleBatchesWorks) {
- MultipleBatchesWorksTest();
-}
+ RemoteCommandRequest rcr(HostAndPort("localhost"), "test", findCmd, opCtx.get());
-TEST_F(NoPrefetchPinnedTaskExecutorCursorTestFixture, MultipleBatchesWorks) {
- MultipleBatchesWorksTest();
-}
+ TaskExecutorCursor tec(&getExecutor(), rcr, [] {
+ TaskExecutorCursor::Options opts;
+ opts.batchSize = 3;
+ return opts;
+ }());
-TEST_F(PinnedConnTaskExecutorCursorTestFixture, MultipleBatchesWorks) {
- MultipleBatchesWorksTest();
-}
+ scheduleSuccessfulCursorResponse("firstBatch", 1, 2, cursorId);
-TEST_F(NonPinningTaskExecutorCursorTestFixture, EmptyFirstBatch) {
- EmptyFirstBatchTest();
-}
+ ASSERT_EQUALS(tec.getNext(opCtx.get()).get()["x"].Int(), 1);
-TEST_F(NoPrefetchTaskExecutorCursorTestFixture, EmptyFirstBatch) {
- EmptyFirstBatchTest();
-}
+ ASSERT(hasReadyRequests());
-TEST_F(NoPrefetchPinnedTaskExecutorCursorTestFixture, EmptyFirstBatch) {
- EmptyFirstBatchTest();
-}
+ ASSERT_EQUALS(tec.getNext(opCtx.get()).get()["x"].Int(), 2);
-TEST_F(PinnedConnTaskExecutorCursorTestFixture, EmptyFirstBatch) {
- EmptyFirstBatchTest();
-}
+ // If we try to getNext() at this point, we are interruptible and can timeout
+ ASSERT_THROWS_CODE(opCtx->runWithDeadline(Date_t::now() + Milliseconds(100),
+ ErrorCodes::ExceededTimeLimit,
+ [&] { tec.getNext(opCtx.get()); }),
+ DBException,
+ ErrorCodes::ExceededTimeLimit);
-TEST_F(NonPinningTaskExecutorCursorTestFixture, EmptyNonInitialBatch) {
- EmptyNonInitialBatchTest();
-}
+ // We can pick up after that interruption though
+ ASSERT_BSONOBJ_EQ(BSON("getMore" << 1LL << "collection"
+ << "test"
+ << "batchSize" << 3),
+ scheduleSuccessfulCursorResponse("nextBatch", 3, 5, cursorId));
+
+ ASSERT_EQUALS(tec.getNext(opCtx.get()).get()["x"].Int(), 3);
+ ASSERT_EQUALS(tec.getNext(opCtx.get()).get()["x"].Int(), 4);
+ ASSERT_EQUALS(tec.getNext(opCtx.get()).get()["x"].Int(), 5);
+
+ cursorId = 0;
+ scheduleSuccessfulCursorResponse("nextBatch", 6, 6, cursorId);
-TEST_F(NoPrefetchTaskExecutorCursorTestFixture, EmptyNonInitialBatch) {
- EmptyNonInitialBatchTest();
+ // We don't issue extra getmores after returning a 0 cursor id
+ ASSERT_FALSE(hasReadyRequests());
+
+ ASSERT_EQUALS(tec.getNext(opCtx.get()).get()["x"].Int(), 6);
+
+ ASSERT_FALSE(tec.getNext(opCtx.get()));
}
-TEST_F(NoPrefetchPinnedTaskExecutorCursorTestFixture, EmptyNonInitialBatch) {
- EmptyNonInitialBatchTest();
+/**
+ * Ensure we allow empty firstBatch.
+ */
+TEST_F(TaskExecutorCursorFixture, EmptyFirstBatch) {
+ const auto findCmd = BSON("find"
+ << "test"
+ << "batchSize" << 2);
+ const auto getMoreCmd = BSON("getMore" << 1LL << "collection"
+ << "test"
+ << "batchSize" << 3);
+ const CursorId cursorId = 1;
+
+ RemoteCommandRequest rcr(HostAndPort("localhost"), "test", findCmd, opCtx.get());
+
+ TaskExecutorCursor tec(&getExecutor(), rcr, [] {
+ TaskExecutorCursor::Options opts;
+ opts.batchSize = 3;
+ return opts;
+ }());
+
+ // Schedule a cursor response with an empty "firstBatch". Use end < start so we don't
+ // append any doc to "firstBatch".
+ ASSERT_BSONOBJ_EQ(findCmd, scheduleSuccessfulCursorResponse("firstBatch", 1, 0, cursorId));
+
+ stdx::thread th([&] {
+ // Wait for the getMore run by the getNext() below to be ready, and schedule a
+ // cursor response with a non-empty "nextBatch".
+ while (!hasReadyRequests()) {
+ sleepmillis(10);
+ }
+
+ ASSERT_BSONOBJ_EQ(getMoreCmd,
+ scheduleSuccessfulCursorResponse("nextBatch", 1, 1, cursorId));
+ });
+
+ // Verify that the first doc is the doc from the second batch.
+ ASSERT_EQUALS(tec.getNext(opCtx.get()).get()["x"].Int(), 1);
+
+ th.join();
}
-TEST_F(PinnedConnTaskExecutorCursorTestFixture, EmptyNonInitialBatch) {
- EmptyNonInitialBatchTest();
+/**
+ * Ensure we allow any empty non-initial batch.
+ */
+TEST_F(TaskExecutorCursorFixture, EmptyNonInitialBatch) {
+ const auto findCmd = BSON("find"
+ << "test"
+ << "batchSize" << 2);
+ const auto getMoreCmd = BSON("getMore" << 1LL << "collection"
+ << "test"
+ << "batchSize" << 3);
+ const CursorId cursorId = 1;
+
+ RemoteCommandRequest rcr(HostAndPort("localhost"), "test", findCmd, opCtx.get());
+
+ TaskExecutorCursor tec(&getExecutor(), rcr, [] {
+ TaskExecutorCursor::Options opts;
+ opts.batchSize = 3;
+ return opts;
+ }());
+
+ // Schedule a cursor response with a non-empty "firstBatch".
+ ASSERT_BSONOBJ_EQ(findCmd, scheduleSuccessfulCursorResponse("firstBatch", 1, 1, cursorId));
+
+ ASSERT_EQUALS(tec.getNext(opCtx.get()).get()["x"].Int(), 1);
+
+ // Schedule two consecutive cursor responses with empty "nextBatch". Use end < start so
+ // we don't append any doc to "nextBatch".
+ ASSERT_BSONOBJ_EQ(getMoreCmd, scheduleSuccessfulCursorResponse("nextBatch", 1, 0, cursorId));
+
+ stdx::thread th([&] {
+ // Wait for the first getMore run by the getNext() below to be ready, and schedule a
+ // cursor response with a non-empty "nextBatch".
+ while (!hasReadyRequests()) {
+ sleepmillis(10);
+ }
+
+ ASSERT_BSONOBJ_EQ(getMoreCmd,
+ scheduleSuccessfulCursorResponse("nextBatch", 1, 0, cursorId));
+
+ // Wait for the second getMore run by the getNext() below to be ready, and schedule a
+ // cursor response with a non-empty "nextBatch".
+ while (!hasReadyRequests()) {
+ sleepmillis(10);
+ }
+
+ ASSERT_BSONOBJ_EQ(getMoreCmd,
+ scheduleSuccessfulCursorResponse("nextBatch", 2, 2, cursorId));
+ });
+
+ // Verify that the next doc is the doc from the fourth batch.
+ ASSERT_EQUALS(tec.getNext(opCtx.get()).get()["x"].Int(), 2);
+
+ th.join();
}
/**
- * Ensure the LSID is passed in all stages of querying. Need to test the
- * pinning case separately because of difference around killCursor.
+ * Ensure lsid is passed in all stages of querying
*/
-TEST_F(NonPinningTaskExecutorCursorTestFixture, LsidIsPassed) {
+TEST_F(TaskExecutorCursorFixture, LsidIsPassed) {
auto lsid = makeLogicalSessionIdForTest();
opCtx->setLogicalSessionId(lsid);
@@ -941,11 +490,11 @@ TEST_F(NonPinningTaskExecutorCursorTestFixture, LsidIsPassed) {
RemoteCommandRequest rcr(HostAndPort("localhost"), "test", findCmd, opCtx.get());
boost::optional<TaskExecutorCursor> tec;
- tec.emplace(makeTec(rcr, []() {
+ tec.emplace(&getExecutor(), rcr, []() {
TaskExecutorCursor::Options opts;
opts.batchSize = 1;
return opts;
- }()));
+ }());
// lsid in the first batch
ASSERT_BSONOBJ_EQ(BSON("find"
@@ -972,14 +521,6 @@ TEST_F(NonPinningTaskExecutorCursorTestFixture, LsidIsPassed) {
ASSERT_FALSE(hasReadyRequests());
}
-TEST_F(NoPrefetchTaskExecutorCursorTestFixture, NoPrefetchGetMore) {
- NoPrefetchGetMore();
-}
-
-TEST_F(NoPrefetchPinnedTaskExecutorCursorTestFixture, NoPrefetchWithPinning) {
- NoPrefetchGetMore();
-}
-
} // namespace
} // namespace executor
} // namespace mongo
generated by cgit v1.2.3 (git 2.39.1) at 2026-09-11 17:13:18 +0000