1 
2 /**
3  *    Copyright (C) 2018-present MongoDB, Inc.
4  *
5  *    This program is free software: you can redistribute it and/or modify
6  *    it under the terms of the Server Side Public License, version 1,
7  *    as published by MongoDB, Inc.
8  *
9  *    This program is distributed in the hope that it will be useful,
10  *    but WITHOUT ANY WARRANTY; without even the implied warranty of
11  *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
12  *    Server Side Public License for more details.
13  *
14  *    You should have received a copy of the Server Side Public License
15  *    along with this program. If not, see
16  *    <http://www.mongodb.com/licensing/server-side-public-license>.
17  *
18  *    As a special exception, the copyright holders give permission to link the
19  *    code of portions of this program with the OpenSSL library under certain
20  *    conditions as described in each individual source file and distribute
21  *    linked combinations including the program with the OpenSSL library. You
22  *    must comply with the Server Side Public License in all respects for
23  *    all of the code used other than as permitted herein. If you modify file(s)
24  *    with this exception, you may extend this exception to your version of the
25  *    file(s), but you are not obligated to do so. If you do not wish to do so,
26  *    delete this exception statement from your version. If you delete this
27  *    exception statement from all source files in the program, then also delete
28  *    it in the license file.
29  */
30 
31 #define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kExecutor
32 
33 #include "mongo/platform/basic.h"
34 
35 #include "mongo/executor/task_executor_test_common.h"
36 
37 #include <memory>
38 
39 #include "mongo/base/disallow_copying.h"
40 #include "mongo/db/operation_context.h"
41 #include "mongo/executor/network_interface.h"
42 #include "mongo/executor/network_interface_mock.h"
43 #include "mongo/executor/task_executor.h"
44 #include "mongo/executor/task_executor_test_fixture.h"
45 #include "mongo/platform/unordered_map.h"
46 #include "mongo/stdx/memory.h"
47 #include "mongo/stdx/thread.h"
48 #include "mongo/unittest/unittest.h"
49 #include "mongo/util/clock_source_mock.h"
50 #include "mongo/util/log.h"
51 #include "mongo/util/mongoutils/str.h"
52 
53 namespace mongo {
54 namespace executor {
55 namespace {
56 
57 using ExecutorFactory =
58     stdx::function<std::unique_ptr<TaskExecutor>(std::unique_ptr<NetworkInterfaceMock>)>;
59 
60 class CommonTaskExecutorTestFixture : public TaskExecutorTest {
61 public:
CommonTaskExecutorTestFixture(ExecutorFactory makeExecutor)62     CommonTaskExecutorTestFixture(ExecutorFactory makeExecutor)
63         : _makeExecutor(std::move(makeExecutor)) {}
64 
65 private:
makeTaskExecutor(std::unique_ptr<NetworkInterfaceMock> net)66     std::unique_ptr<TaskExecutor> makeTaskExecutor(
67         std::unique_ptr<NetworkInterfaceMock> net) override {
68         return _makeExecutor(std::move(net));
69     }
70 
71     ExecutorFactory _makeExecutor;
72 };
73 
74 using ExecutorTestCaseFactory =
75     stdx::function<std::unique_ptr<CommonTaskExecutorTestFixture>(ExecutorFactory)>;
76 using ExecutorTestCaseMap = unordered_map<std::string, ExecutorTestCaseFactory>;
77 
executorTestCaseRegistry()78 static ExecutorTestCaseMap& executorTestCaseRegistry() {
79     static ExecutorTestCaseMap registry;
80     return registry;
81 }
82 
83 class CetRegistrationAgent {
84     MONGO_DISALLOW_COPYING(CetRegistrationAgent);
85 
86 public:
CetRegistrationAgent(const std::string & name,ExecutorTestCaseFactory makeTest)87     CetRegistrationAgent(const std::string& name, ExecutorTestCaseFactory makeTest) {
88         auto& entry = executorTestCaseRegistry()[name];
89         if (entry) {
90             severe() << "Multiple attempts to register ExecutorTest named " << name;
91             fassertFailed(28713);
92         }
93         entry = std::move(makeTest);
94     }
95 };
96 
97 #define COMMON_EXECUTOR_TEST(TEST_NAME)                                         \
98     class CET_##TEST_NAME : public CommonTaskExecutorTestFixture {              \
99     public:                                                                     \
100         CET_##TEST_NAME(ExecutorFactory makeExecutor)                           \
101             : CommonTaskExecutorTestFixture(std::move(makeExecutor)) {}         \
102                                                                                 \
103     private:                                                                    \
104         void _doTest() override;                                                \
105         static const CetRegistrationAgent _agent;                               \
106     };                                                                          \
107     const CetRegistrationAgent CET_##TEST_NAME::_agent(                         \
108         #TEST_NAME, [](ExecutorFactory makeExecutor) {                          \
109             return stdx::make_unique<CET_##TEST_NAME>(std::move(makeExecutor)); \
110         });                                                                     \
111     void CET_##TEST_NAME::_doTest()
112 
setStatus(const TaskExecutor::CallbackArgs & cbData,Status * target)113 void setStatus(const TaskExecutor::CallbackArgs& cbData, Status* target) {
114     *target = cbData.status;
115 }
116 
setStatusAndShutdown(const TaskExecutor::CallbackArgs & cbData,Status * target)117 void setStatusAndShutdown(const TaskExecutor::CallbackArgs& cbData, Status* target) {
118     setStatus(cbData, target);
119     if (cbData.status != ErrorCodes::CallbackCanceled)
120         cbData.executor->shutdown();
121 }
122 
setStatusAndTriggerEvent(const TaskExecutor::CallbackArgs & cbData,Status * outStatus,TaskExecutor::EventHandle event)123 void setStatusAndTriggerEvent(const TaskExecutor::CallbackArgs& cbData,
124                               Status* outStatus,
125                               TaskExecutor::EventHandle event) {
126     *outStatus = cbData.status;
127     if (!cbData.status.isOK())
128         return;
129     cbData.executor->signalEvent(event);
130 }
131 
scheduleSetStatusAndShutdown(const TaskExecutor::CallbackArgs & cbData,Status * outStatus1,Status * outStatus2)132 void scheduleSetStatusAndShutdown(const TaskExecutor::CallbackArgs& cbData,
133                                   Status* outStatus1,
134                                   Status* outStatus2) {
135     if (!cbData.status.isOK()) {
136         *outStatus1 = cbData.status;
137         return;
138     }
139     *outStatus1 =
140         cbData.executor
141             ->scheduleWork(stdx::bind(setStatusAndShutdown, stdx::placeholders::_1, outStatus2))
142             .getStatus();
143 }
144 
COMMON_EXECUTOR_TEST(RunOne)145 COMMON_EXECUTOR_TEST(RunOne) {
146     TaskExecutor& executor = getExecutor();
147     Status status = getDetectableErrorStatus();
148     ASSERT_OK(
149         executor.scheduleWork(stdx::bind(setStatusAndShutdown, stdx::placeholders::_1, &status))
150             .getStatus());
151     launchExecutorThread();
152     joinExecutorThread();
153     ASSERT_OK(status);
154 }
155 
COMMON_EXECUTOR_TEST(Schedule1ButShutdown)156 COMMON_EXECUTOR_TEST(Schedule1ButShutdown) {
157     TaskExecutor& executor = getExecutor();
158     Status status = getDetectableErrorStatus();
159     ASSERT_OK(
160         executor.scheduleWork(stdx::bind(setStatusAndShutdown, stdx::placeholders::_1, &status))
161             .getStatus());
162     executor.shutdown();
163     launchExecutorThread();
164     joinExecutorThread();
165     ASSERT_EQUALS(status, ErrorCodes::CallbackCanceled);
166 }
167 
COMMON_EXECUTOR_TEST(Schedule2Cancel1)168 COMMON_EXECUTOR_TEST(Schedule2Cancel1) {
169     TaskExecutor& executor = getExecutor();
170     Status status1 = getDetectableErrorStatus();
171     Status status2 = getDetectableErrorStatus();
172     TaskExecutor::CallbackHandle cb = unittest::assertGet(
173         executor.scheduleWork(stdx::bind(setStatusAndShutdown, stdx::placeholders::_1, &status1)));
174     executor.cancel(cb);
175     ASSERT_OK(
176         executor.scheduleWork(stdx::bind(setStatusAndShutdown, stdx::placeholders::_1, &status2))
177             .getStatus());
178     launchExecutorThread();
179     joinExecutorThread();
180     ASSERT_EQUALS(status1, ErrorCodes::CallbackCanceled);
181     ASSERT_OK(status2);
182 }
183 
COMMON_EXECUTOR_TEST(OneSchedulesAnother)184 COMMON_EXECUTOR_TEST(OneSchedulesAnother) {
185     TaskExecutor& executor = getExecutor();
186     Status status1 = getDetectableErrorStatus();
187     Status status2 = getDetectableErrorStatus();
188     ASSERT_OK(executor
189                   .scheduleWork(stdx::bind(
190                       scheduleSetStatusAndShutdown, stdx::placeholders::_1, &status1, &status2))
191                   .getStatus());
192     launchExecutorThread();
193     joinExecutorThread();
194     ASSERT_OK(status1);
195     ASSERT_OK(status2);
196 }
197 
198 class EventChainAndWaitingTest {
199     MONGO_DISALLOW_COPYING(EventChainAndWaitingTest);
200 
201 public:
202     EventChainAndWaitingTest(TaskExecutor* exec, NetworkInterfaceMock* network);
203     ~EventChainAndWaitingTest();
204 
205     void run();
206     void assertSuccess();
207 
208 private:
209     void onGo(const TaskExecutor::CallbackArgs& cbData);
210     void onGoAfterTriggered(const TaskExecutor::CallbackArgs& cbData);
211 
212     NetworkInterfaceMock* net;
213     TaskExecutor* executor;
214     const TaskExecutor::EventHandle goEvent;
215     const TaskExecutor::EventHandle event2;
216     const TaskExecutor::EventHandle event3;
217     TaskExecutor::EventHandle triggerEvent;
218     TaskExecutor::CallbackFn triggered2;
219     TaskExecutor::CallbackFn triggered3;
220     Status status1;
221     Status status2;
222     Status status3;
223     Status status4;
224     Status status5;
225     stdx::thread neverSignaledWaiter;
226 };
227 
COMMON_EXECUTOR_TEST(EventChainAndWaiting)228 COMMON_EXECUTOR_TEST(EventChainAndWaiting) {
229     launchExecutorThread();
230     EventChainAndWaitingTest theTest(&getExecutor(), getNet());
231     theTest.run();
232     joinExecutorThread();
233     theTest.assertSuccess();
234 }
235 
EventChainAndWaitingTest(TaskExecutor * exec,NetworkInterfaceMock * network)236 EventChainAndWaitingTest::EventChainAndWaitingTest(TaskExecutor* exec,
237                                                    NetworkInterfaceMock* network)
238     : net(network),
239       executor(exec),
240       goEvent(unittest::assertGet(executor->makeEvent())),
241       event2(unittest::assertGet(executor->makeEvent())),
242       event3(unittest::assertGet(executor->makeEvent())),
243       status1(ErrorCodes::InternalError, "Not mutated"),
244       status2(ErrorCodes::InternalError, "Not mutated"),
245       status3(ErrorCodes::InternalError, "Not mutated"),
246       status4(ErrorCodes::InternalError, "Not mutated"),
247       status5(ErrorCodes::InternalError, "Not mutated") {
248     triggered2 = stdx::bind(setStatusAndTriggerEvent, stdx::placeholders::_1, &status2, event2);
249     triggered3 = stdx::bind(setStatusAndTriggerEvent, stdx::placeholders::_1, &status3, event3);
250 }
251 
~EventChainAndWaitingTest()252 EventChainAndWaitingTest::~EventChainAndWaitingTest() {
253     if (neverSignaledWaiter.joinable()) {
254         neverSignaledWaiter.join();
255     }
256 }
257 
run()258 void EventChainAndWaitingTest::run() {
259     executor
260         ->onEvent(goEvent,
261                   stdx::bind(&EventChainAndWaitingTest::onGo, this, stdx::placeholders::_1))
262         .status_with_transitional_ignore();
263     executor->signalEvent(goEvent);
264     executor->waitForEvent(goEvent);
265     executor->waitForEvent(event2);
266     executor->waitForEvent(event3);
267 
268     TaskExecutor::EventHandle neverSignaledEvent = unittest::assertGet(executor->makeEvent());
269     auto waitForeverCallback = [this, neverSignaledEvent]() {
270         executor->waitForEvent(neverSignaledEvent);
271     };
272     neverSignaledWaiter = stdx::thread(waitForeverCallback);
273     TaskExecutor::CallbackHandle shutdownCallback = unittest::assertGet(
274         executor->scheduleWork(stdx::bind(setStatusAndShutdown, stdx::placeholders::_1, &status5)));
275     executor->wait(shutdownCallback);
276 }
277 
assertSuccess()278 void EventChainAndWaitingTest::assertSuccess() {
279     neverSignaledWaiter.join();
280     ASSERT_OK(status1);
281     ASSERT_OK(status2);
282     ASSERT_OK(status3);
283     ASSERT_OK(status4);
284     ASSERT_OK(status5);
285 }
286 
onGo(const TaskExecutor::CallbackArgs & cbData)287 void EventChainAndWaitingTest::onGo(const TaskExecutor::CallbackArgs& cbData) {
288     if (!cbData.status.isOK()) {
289         status1 = cbData.status;
290         return;
291     }
292     executor::TaskExecutor* executor = cbData.executor;
293     StatusWith<TaskExecutor::EventHandle> errorOrTriggerEvent = executor->makeEvent();
294     if (!errorOrTriggerEvent.isOK()) {
295         status1 = errorOrTriggerEvent.getStatus();
296         executor->shutdown();
297         return;
298     }
299     triggerEvent = errorOrTriggerEvent.getValue();
300     StatusWith<TaskExecutor::CallbackHandle> cbHandle = executor->onEvent(triggerEvent, triggered2);
301     if (!cbHandle.isOK()) {
302         status1 = cbHandle.getStatus();
303         executor->shutdown();
304         return;
305     }
306     cbHandle = executor->onEvent(triggerEvent, triggered3);
307     if (!cbHandle.isOK()) {
308         status1 = cbHandle.getStatus();
309         executor->shutdown();
310         return;
311     }
312 
313     cbHandle = executor->onEvent(
314         goEvent,
315         stdx::bind(&EventChainAndWaitingTest::onGoAfterTriggered, this, stdx::placeholders::_1));
316     if (!cbHandle.isOK()) {
317         status1 = cbHandle.getStatus();
318         executor->shutdown();
319         return;
320     }
321     status1 = Status::OK();
322 }
323 
onGoAfterTriggered(const TaskExecutor::CallbackArgs & cbData)324 void EventChainAndWaitingTest::onGoAfterTriggered(const TaskExecutor::CallbackArgs& cbData) {
325     status4 = cbData.status;
326     if (!cbData.status.isOK()) {
327         return;
328     }
329     cbData.executor->signalEvent(triggerEvent);
330 }
331 
COMMON_EXECUTOR_TEST(EventWaitingWithTimeoutTest)332 COMMON_EXECUTOR_TEST(EventWaitingWithTimeoutTest) {
333     TaskExecutor& executor = getExecutor();
334     launchExecutorThread();
335 
336     auto eventThatWillNeverBeTriggered = unittest::assertGet(executor.makeEvent());
337 
338     auto serviceContext = getGlobalServiceContext();
339 
340     serviceContext->setFastClockSource(stdx::make_unique<ClockSourceMock>());
341     auto mockClock = static_cast<ClockSourceMock*>(serviceContext->getFastClockSource());
342 
343     auto client = serviceContext->makeClient("for testing");
344     auto opCtx = client->makeOperationContext();
345 
346     auto deadline = mockClock->now() + Milliseconds{1};
347     mockClock->advance(Milliseconds(2));
348     ASSERT(stdx::cv_status::timeout ==
349            executor.waitForEvent(opCtx.get(), eventThatWillNeverBeTriggered, deadline));
350     executor.shutdown();
351     joinExecutorThread();
352 }
353 
COMMON_EXECUTOR_TEST(EventSignalWithTimeoutTest)354 COMMON_EXECUTOR_TEST(EventSignalWithTimeoutTest) {
355     TaskExecutor& executor = getExecutor();
356     launchExecutorThread();
357 
358     auto eventSignalled = unittest::assertGet(executor.makeEvent());
359 
360     auto serviceContext = getGlobalServiceContext();
361 
362     serviceContext->setFastClockSource(stdx::make_unique<ClockSourceMock>());
363     auto mockClock = static_cast<ClockSourceMock*>(serviceContext->getFastClockSource());
364 
365     auto client = serviceContext->makeClient("for testing");
366     auto opCtx = client->makeOperationContext();
367 
368     auto deadline = mockClock->now() + Milliseconds{1};
369     mockClock->advance(Milliseconds(1));
370 
371     executor.signalEvent(eventSignalled);
372 
373     ASSERT(stdx::cv_status::no_timeout ==
374            executor.waitForEvent(opCtx.get(), eventSignalled, deadline));
375     executor.shutdown();
376     joinExecutorThread();
377 }
378 
COMMON_EXECUTOR_TEST(ScheduleWorkAt)379 COMMON_EXECUTOR_TEST(ScheduleWorkAt) {
380     NetworkInterfaceMock* net = getNet();
381     TaskExecutor& executor = getExecutor();
382     launchExecutorThread();
383     Status status1 = getDetectableErrorStatus();
384     Status status2 = getDetectableErrorStatus();
385     Status status3 = getDetectableErrorStatus();
386     Status status4 = getDetectableErrorStatus();
387 
388     const Date_t now = net->now();
389     const TaskExecutor::CallbackHandle cb1 = unittest::assertGet(executor.scheduleWorkAt(
390         now + Milliseconds(100), stdx::bind(setStatus, stdx::placeholders::_1, &status1)));
391     const TaskExecutor::CallbackHandle cb4 = unittest::assertGet(executor.scheduleWorkAt(
392         now - Milliseconds(50), stdx::bind(setStatus, stdx::placeholders::_1, &status4)));
393     unittest::assertGet(executor.scheduleWorkAt(
394         now + Milliseconds(5000), stdx::bind(setStatus, stdx::placeholders::_1, &status3)));
395     const TaskExecutor::CallbackHandle cb2 = unittest::assertGet(executor.scheduleWorkAt(
396         now + Milliseconds(200),
397         stdx::bind(setStatusAndShutdown, stdx::placeholders::_1, &status2)));
398 
399     executor.wait(cb4);
400     ASSERT_OK(status4);
401 
402     const Date_t startTime = net->now();
403     net->enterNetwork();
404     net->runUntil(startTime + Milliseconds(200));
405     net->exitNetwork();
406     ASSERT_EQUALS(startTime + Milliseconds(200), net->now());
407     executor.wait(cb1);
408     executor.wait(cb2);
409     ASSERT_OK(status1);
410     ASSERT_OK(status2);
411     executor.shutdown();
412     joinExecutorThread();
413     ASSERT_EQUALS(status3, ErrorCodes::CallbackCanceled);
414 }
415 
getRequestDescription(const RemoteCommandRequest & request)416 std::string getRequestDescription(const RemoteCommandRequest& request) {
417     return mongoutils::str::stream() << "Request(" << request.target.toString() << ", "
418                                      << request.dbname << ", " << request.cmdObj << ')';
419 }
420 
setStatusOnRemoteCommandCompletion(const TaskExecutor::RemoteCommandCallbackArgs & cbData,const RemoteCommandRequest & expectedRequest,Status * outStatus)421 static void setStatusOnRemoteCommandCompletion(
422     const TaskExecutor::RemoteCommandCallbackArgs& cbData,
423     const RemoteCommandRequest& expectedRequest,
424     Status* outStatus) {
425     if (cbData.request != expectedRequest) {
426         *outStatus = Status(ErrorCodes::BadValue,
427                             mongoutils::str::stream() << "Actual request: "
428                                                       << getRequestDescription(cbData.request)
429                                                       << "; expected: "
430                                                       << getRequestDescription(expectedRequest));
431         return;
432     }
433     *outStatus = cbData.response.status;
434 }
435 
COMMON_EXECUTOR_TEST(ScheduleRemoteCommand)436 COMMON_EXECUTOR_TEST(ScheduleRemoteCommand) {
437     NetworkInterfaceMock* net = getNet();
438     TaskExecutor& executor = getExecutor();
439     launchExecutorThread();
440     Status status1 = getDetectableErrorStatus();
441     const RemoteCommandRequest request(HostAndPort("localhost", 27017),
442                                        "mydb",
443                                        BSON("whatsUp"
444                                             << "doc"),
445                                        nullptr);
446     TaskExecutor::CallbackHandle cbHandle = unittest::assertGet(executor.scheduleRemoteCommand(
447         request,
448         stdx::bind(setStatusOnRemoteCommandCompletion, stdx::placeholders::_1, request, &status1)));
449     net->enterNetwork();
450     ASSERT(net->hasReadyRequests());
451     NetworkInterfaceMock::NetworkOperationIterator noi = net->getNextReadyRequest();
452     net->scheduleResponse(noi, net->now(), {ErrorCodes::NoSuchKey, "I'm missing"});
453     net->runReadyNetworkOperations();
454     ASSERT(!net->hasReadyRequests());
455     net->exitNetwork();
456     executor.wait(cbHandle);
457     executor.shutdown();
458     joinExecutorThread();
459     ASSERT_EQUALS(ErrorCodes::NoSuchKey, status1);
460 }
461 
COMMON_EXECUTOR_TEST(ScheduleAndCancelRemoteCommand)462 COMMON_EXECUTOR_TEST(ScheduleAndCancelRemoteCommand) {
463     TaskExecutor& executor = getExecutor();
464     Status status1 = getDetectableErrorStatus();
465     const RemoteCommandRequest request(HostAndPort("localhost", 27017),
466                                        "mydb",
467                                        BSON("whatsUp"
468                                             << "doc"),
469                                        nullptr);
470     TaskExecutor::CallbackHandle cbHandle = unittest::assertGet(executor.scheduleRemoteCommand(
471         request,
472         stdx::bind(setStatusOnRemoteCommandCompletion, stdx::placeholders::_1, request, &status1)));
473     executor.cancel(cbHandle);
474     launchExecutorThread();
475     getNet()->enterNetwork();
476     getNet()->runReadyNetworkOperations();
477     getNet()->exitNetwork();
478     executor.wait(cbHandle);
479     executor.shutdown();
480     joinExecutorThread();
481     ASSERT_EQUALS(ErrorCodes::CallbackCanceled, status1);
482 }
483 
484 
COMMON_EXECUTOR_TEST(RemoteCommandWithTimeout)485 COMMON_EXECUTOR_TEST(RemoteCommandWithTimeout) {
486     NetworkInterfaceMock* net = getNet();
487     TaskExecutor& executor = getExecutor();
488     Status status(ErrorCodes::InternalError, "");
489     launchExecutorThread();
490     const RemoteCommandRequest request(
491         HostAndPort("lazy", 27017), "admin", BSON("sleep" << 1), nullptr, Milliseconds(1));
492     TaskExecutor::CallbackHandle cbHandle = unittest::assertGet(executor.scheduleRemoteCommand(
493         request,
494         stdx::bind(setStatusOnRemoteCommandCompletion, stdx::placeholders::_1, request, &status)));
495     net->enterNetwork();
496     ASSERT(net->hasReadyRequests());
497     const Date_t startTime = net->now();
498     NetworkInterfaceMock::NetworkOperationIterator noi = net->getNextReadyRequest();
499     net->scheduleResponse(noi, startTime + Milliseconds(2), {});
500     net->runUntil(startTime + Milliseconds(2));
501     ASSERT_EQUALS(startTime + Milliseconds(2), net->now());
502     net->exitNetwork();
503     executor.wait(cbHandle);
504     ASSERT_EQUALS(ErrorCodes::NetworkTimeout, status);
505 }
506 
COMMON_EXECUTOR_TEST(CallbackHandleComparison)507 COMMON_EXECUTOR_TEST(CallbackHandleComparison) {
508     TaskExecutor& executor = getExecutor();
509     auto status1 = getDetectableErrorStatus();
510     auto status2 = getDetectableErrorStatus();
511     const RemoteCommandRequest request(
512         HostAndPort("lazy", 27017), "admin", BSON("cmd" << 1), nullptr);
513     TaskExecutor::CallbackHandle cbHandle1 = unittest::assertGet(executor.scheduleRemoteCommand(
514         request,
515         stdx::bind(setStatusOnRemoteCommandCompletion, stdx::placeholders::_1, request, &status1)));
516     TaskExecutor::CallbackHandle cbHandle2 = unittest::assertGet(executor.scheduleRemoteCommand(
517         request,
518         stdx::bind(setStatusOnRemoteCommandCompletion, stdx::placeholders::_1, request, &status2)));
519 
520     // test equality
521     ASSERT_TRUE(cbHandle1 == cbHandle1);
522     ASSERT_TRUE(cbHandle2 == cbHandle2);
523     ASSERT_FALSE(cbHandle1 != cbHandle1);
524     ASSERT_FALSE(cbHandle2 != cbHandle2);
525 
526     // test inequality
527     ASSERT_TRUE(cbHandle1 != cbHandle2);
528     ASSERT_TRUE(cbHandle2 != cbHandle1);
529     ASSERT_FALSE(cbHandle1 == cbHandle2);
530     ASSERT_FALSE(cbHandle2 == cbHandle1);
531 
532     TaskExecutor::CallbackHandle cbHandle1Copy = cbHandle1;
533     ASSERT_TRUE(cbHandle1 == cbHandle1Copy);
534     ASSERT_TRUE(cbHandle1Copy == cbHandle1);
535     ASSERT_FALSE(cbHandle1Copy != cbHandle1);
536     ASSERT_FALSE(cbHandle1 != cbHandle1Copy);
537 
538     std::vector<TaskExecutor::CallbackHandle> cbs;
539     cbs.push_back(cbHandle1);
540     cbs.push_back(cbHandle2);
541     ASSERT(cbHandle1 != cbHandle2);
542     std::vector<TaskExecutor::CallbackHandle>::iterator foundHandle =
543         std::find(cbs.begin(), cbs.end(), cbHandle1);
544     ASSERT_TRUE(cbs.end() != foundHandle);
545     ASSERT_TRUE(cbHandle1 == *foundHandle);
546     launchExecutorThread();
547     executor.shutdown();
548     joinExecutorThread();
549 }
550 
551 }  // namespace
552 
addTestsForExecutor(const std::string & suiteName,ExecutorFactory makeExecutor)553 void addTestsForExecutor(const std::string& suiteName, ExecutorFactory makeExecutor) {
554     auto suite = unittest::Suite::getSuite(suiteName);
555     for (auto testCase : executorTestCaseRegistry()) {
556         suite->add(str::stream() << suiteName << "::" << testCase.first,
557                    [testCase, makeExecutor] { testCase.second(makeExecutor)->run(); });
558     }
559 }
560 
561 }  // namespace executor
562 }  // namespace mongo
563