Skip to content

Commit f133e71

Browse files
BenHuddlestondaverigby
BenHuddleston
authored and
daverigby
committed
MB-41719: Move processConsumerMutationsNearThreshold to ST test suite
Change-Id: I8b51b0eb9d7a779a652f8c73ee3ce3b5410e9f30 Reviewed-on: http://review.couchbase.org/c/kv_engine/+/137170 Tested-by: Build Bot <[email protected]> Reviewed-by: Dave Rigby <[email protected]>
1 parent 643773a commit f133e71

File tree

2 files changed

+153
-145
lines changed

2 files changed

+153
-145
lines changed

‎engines/ep/tests/module_tests/dcp_single_threaded_test.cc‎

Lines changed: 153 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,15 @@ class STDcpTest : public STParameterizedBucketTest {
4444
*/
4545
void testConsumerNegotiatesIncludeDeletedUserXattrs(
4646
IncludeDeletedUserXattrs producerState);
47+
48+
/**
49+
* Creates a consumer conn and makes the consumer processor task run with
50+
* memory usage near to replication threshold
51+
*
52+
* @param beyondThreshold indicates if the memory usage should above the
53+
* threshold or just below it
54+
*/
55+
void processConsumerMutationsNearThreshold(bool beyondThreshold);
4756
};
4857

4958
/*
@@ -412,6 +421,150 @@ TEST_P(STDcpTest, ConsumerNegotiatesDeletedUserXattrs_EnabledAtProducer) {
412421
IncludeDeletedUserXattrs::Yes);
413422
}
414423

424+
void STDcpTest::processConsumerMutationsNearThreshold(bool beyondThreshold) {
425+
const void* cookie = create_mock_cookie(engine.get());
426+
const uint32_t opaque = 1;
427+
const uint64_t snapStart = 1, snapEnd = 10;
428+
const uint64_t bySeqno = snapStart;
429+
430+
/* Set up a consumer connection */
431+
auto& connMap = engine->getDcpConnMap();
432+
auto& mockConnMap = static_cast<MockDcpConnMap&>(connMap);
433+
auto consumer =
434+
std::make_shared<MockDcpConsumer>(*engine, cookie, "test_consumer");
435+
mockConnMap.addConn(cookie, consumer);
436+
437+
/* Replica vbucket */
438+
setVBucketStateAndRunPersistTask(vbid, vbucket_state_replica);
439+
440+
/* Passive stream */
441+
ASSERT_EQ(ENGINE_SUCCESS,
442+
consumer->addStream(/*opaque*/ 0,
443+
vbid,
444+
/*flags*/ 0));
445+
MockPassiveStream* stream = static_cast<MockPassiveStream*>(
446+
(consumer->getVbucketStream(vbid)).get());
447+
ASSERT_TRUE(stream->isActive());
448+
449+
/* Send a snapshotMarker before sending items for replication */
450+
EXPECT_EQ(ENGINE_SUCCESS,
451+
consumer->snapshotMarker(opaque,
452+
vbid,
453+
snapStart,
454+
snapEnd,
455+
/* in-memory snapshot */ 0x1,
456+
/*HCS*/ {},
457+
/*maxVisibleSeqno*/ {}));
458+
459+
/* Simulate a situation where adding a mutation temporarily fails
460+
and hence adds the mutation to a replication buffer. For that, we
461+
set vbucket::takeover_backed_up to true */
462+
engine->getKVBucket()->getVBucket(vbid)->setTakeoverBackedUpState(true);
463+
464+
/* Send an item for replication and expect it to be buffered */
465+
const DocKey docKey{"mykey", DocKeyEncodesCollectionId::No};
466+
EXPECT_EQ(ENGINE_SUCCESS,
467+
consumer->mutation(opaque,
468+
docKey,
469+
{}, // value
470+
0, // priv bytes
471+
PROTOCOL_BINARY_RAW_BYTES,
472+
0, // cas
473+
vbid,
474+
0, // flags
475+
bySeqno,
476+
0, // rev seqno
477+
0, // exptime
478+
0, // locktime
479+
{}, // meta
480+
0)); // nru
481+
EXPECT_EQ(1, stream->getNumBufferItems());
482+
483+
/* Set back the vbucket::takeover_backed_up to false */
484+
engine->getKVBucket()->getVBucket(vbid)->setTakeoverBackedUpState(false);
485+
486+
/* Set 'mem_used' beyond the 'replication threshold' */
487+
EPStats& stats = engine->getEpStats();
488+
if (beyondThreshold) {
489+
/* Actually setting it well above also, as there can be a drop in memory
490+
usage during testing */
491+
stats.setMaxDataSize(stats.getPreciseTotalMemoryUsed() / 4);
492+
} else {
493+
/* set max size to a value just over */
494+
stats.setMaxDataSize(stats.getPreciseTotalMemoryUsed() + 1);
495+
/* Simpler to set the replication threshold to 1 and test, rather than
496+
testing with maxData = (memUsed / replicationThrottleThreshold); that
497+
is, we are avoiding a division */
498+
engine->getConfiguration().setReplicationThrottleThreshold(100);
499+
}
500+
501+
MockDcpMessageProducers producers(engine.get());
502+
if ((engine->getConfiguration().getBucketType() == "ephemeral") &&
503+
(engine->getConfiguration().getEphemeralFullPolicy()) ==
504+
"fail_new_data") {
505+
/* Make a call to the function that would be called by the processor
506+
task here */
507+
EXPECT_EQ(stop_processing, consumer->processBufferedItems());
508+
509+
/* Expect the connection to be notified */
510+
EXPECT_FALSE(consumer->isPaused());
511+
512+
/* Expect disconnect signal in Ephemeral with "fail_new_data" policy */
513+
EXPECT_EQ(ENGINE_DISCONNECT, consumer->step(&producers));
514+
} else {
515+
uint32_t backfoffs = consumer->getNumBackoffs();
516+
517+
/* Make a call to the function that would be called by the processor
518+
task here */
519+
if (beyondThreshold) {
520+
EXPECT_EQ(more_to_process, consumer->processBufferedItems());
521+
} else {
522+
EXPECT_EQ(cannot_process, consumer->processBufferedItems());
523+
}
524+
525+
EXPECT_EQ(backfoffs + 1, consumer->getNumBackoffs());
526+
527+
/* In 'couchbase' buckets we buffer the replica items and indirectly
528+
throttle replication by not sending flow control acks to the
529+
producer. Hence we do not drop the connection here */
530+
EXPECT_EQ(ENGINE_SUCCESS, consumer->step(&producers));
531+
532+
/* Close stream before deleting the connection */
533+
EXPECT_EQ(ENGINE_SUCCESS, consumer->closeStream(opaque, vbid));
534+
}
535+
536+
connMap.disconnect(cookie);
537+
EXPECT_FALSE(connMap.isDeadConnectionsEmpty());
538+
connMap.manageConnections();
539+
EXPECT_TRUE(connMap.isDeadConnectionsEmpty());
540+
}
541+
542+
/* Here we test how the Processor task in DCP consumer handles the scenario
543+
where the memory usage is beyond the replication throttle threshold.
544+
In case of Ephemeral buckets with 'fail_new_data' policy it is expected to
545+
indicate close of the consumer conn and in other cases it is expected to
546+
just defer processing. */
547+
TEST_P(STDcpTest, ProcessReplicationBufferAfterThrottleThreshold) {
548+
processConsumerMutationsNearThreshold(true);
549+
}
550+
551+
/* Here we test how the Processor task in DCP consumer handles the scenario
552+
where the memory usage is just below the replication throttle threshold,
553+
but will go over the threshold when it adds the new mutation from the
554+
processor buffer to the hashtable.
555+
In case of Ephemeral buckets with 'fail_new_data' policy it is expected to
556+
indicate close of the consumer conn and in other cases it is expected to
557+
just defer processing. */
558+
TEST_P(STDcpTest,
559+
DISABLED_ProcessReplicationBufferJustBeforeThrottleThreshold) {
560+
/* There are sporadic failures seen while testing this. The problem is
561+
we need to have a memory usage just below max_size, so we need to
562+
start at that point. But sometimes the memory usage goes further below
563+
resulting in the test failure (a hang). Hence commenting out the test.
564+
Can be run locally as and when needed. */
565+
processConsumerMutationsNearThreshold(false);
566+
}
567+
415568
INSTANTIATE_TEST_SUITE_P(PersistentAndEphemeral,
416569
STDcpTest,
417570
STParameterizedBucketTest::allConfigValues());

‎engines/ep/tests/module_tests/dcp_test.cc‎

Lines changed: 0 additions & 145 deletions
Original file line numberDiff line numberDiff line change
@@ -833,15 +833,6 @@ class ConnectionTest : public DCPTest,
833833
*/
834834
void sendConsumerMutationsNearThreshold(bool beyondThreshold);
835835

836-
/**
837-
* Creates a consumer conn and makes the consumer processor task run with
838-
* memory usage near to replication threshold
839-
*
840-
* @param beyondThreshold indicates if the memory usage should above the
841-
* threshold or just below it
842-
*/
843-
void processConsumerMutationsNearThreshold(bool beyondThreshold);
844-
845836
/* vbucket associated with this connection */
846837
Vbid vbid;
847838
};
@@ -2671,143 +2662,7 @@ TEST_P(ConnectionTest, ReplicateJustBeforeThrottleThreshold) {
26712662
sendConsumerMutationsNearThreshold(false);
26722663
}
26732664

2674-
void ConnectionTest::processConsumerMutationsNearThreshold(
2675-
bool beyondThreshold) {
2676-
const void* cookie = create_mock_cookie(engine);
2677-
const uint32_t opaque = 1;
2678-
const uint64_t snapStart = 1, snapEnd = 10;
2679-
const uint64_t bySeqno = snapStart;
2680-
2681-
/* Set up a consumer connection */
2682-
auto consumer =
2683-
std::make_shared<MockDcpConsumer>(*engine, cookie, "test_consumer");
2684-
2685-
/* Replica vbucket */
2686-
ASSERT_EQ(ENGINE_SUCCESS, set_vb_state(vbid, vbucket_state_replica));
2687-
2688-
/* Passive stream */
2689-
ASSERT_EQ(ENGINE_SUCCESS,
2690-
consumer->addStream(/*opaque*/ 0,
2691-
vbid,
2692-
/*flags*/ 0));
2693-
MockPassiveStream* stream = static_cast<MockPassiveStream*>(
2694-
(consumer->getVbucketStream(vbid)).get());
2695-
ASSERT_TRUE(stream->isActive());
2696-
2697-
/* Send a snapshotMarker before sending items for replication */
2698-
EXPECT_EQ(ENGINE_SUCCESS,
2699-
consumer->snapshotMarker(opaque,
2700-
vbid,
2701-
snapStart,
2702-
snapEnd,
2703-
/* in-memory snapshot */ 0x1,
2704-
/*HCS*/ {},
2705-
/*maxVisibleSeqno*/ {}));
2706-
2707-
/* Simulate a situation where adding a mutation temporarily fails
2708-
and hence adds the mutation to a replication buffer. For that, we
2709-
set vbucket::takeover_backed_up to true */
2710-
engine->getKVBucket()->getVBucket(vbid)->setTakeoverBackedUpState(true);
2711-
2712-
/* Send an item for replication and expect it to be buffered */
2713-
const DocKey docKey{"mykey", DocKeyEncodesCollectionId::No};
2714-
EXPECT_EQ(ENGINE_SUCCESS,
2715-
consumer->mutation(opaque,
2716-
docKey,
2717-
{}, // value
2718-
0, // priv bytes
2719-
PROTOCOL_BINARY_RAW_BYTES,
2720-
0, // cas
2721-
vbid,
2722-
0, // flags
2723-
bySeqno,
2724-
0, // rev seqno
2725-
0, // exptime
2726-
0, // locktime
2727-
{}, // meta
2728-
0)); // nru
2729-
EXPECT_EQ(1, stream->getNumBufferItems());
2730-
2731-
/* Set back the vbucket::takeover_backed_up to false */
2732-
engine->getKVBucket()->getVBucket(vbid)->setTakeoverBackedUpState(false);
2733-
2734-
/* Set 'mem_used' beyond the 'replication threshold' */
2735-
EPStats& stats = engine->getEpStats();
2736-
if (beyondThreshold) {
2737-
/* Actually setting it well above also, as there can be a drop in memory
2738-
usage during testing */
2739-
stats.setMaxDataSize(stats.getPreciseTotalMemoryUsed() / 4);
2740-
} else {
2741-
/* set max size to a value just over */
2742-
stats.setMaxDataSize(stats.getPreciseTotalMemoryUsed() + 1);
2743-
/* Simpler to set the replication threshold to 1 and test, rather than
2744-
testing with maxData = (memUsed / replicationThrottleThreshold); that
2745-
is, we are avoiding a division */
2746-
engine->getConfiguration().setReplicationThrottleThreshold(100);
2747-
}
2748-
2749-
MockDcpMessageProducers producers(handle);
2750-
if ((engine->getConfiguration().getBucketType() == "ephemeral") &&
2751-
(engine->getConfiguration().getEphemeralFullPolicy()) ==
2752-
"fail_new_data") {
2753-
/* Make a call to the function that would be called by the processor
2754-
task here */
2755-
EXPECT_EQ(stop_processing, consumer->processBufferedItems());
2756-
2757-
/* Expect the connection to be notified */
2758-
EXPECT_FALSE(consumer->isPaused());
2759-
2760-
/* Expect disconnect signal in Ephemeral with "fail_new_data" policy */
2761-
EXPECT_EQ(ENGINE_DISCONNECT, consumer->step(&producers));
2762-
} else {
2763-
uint32_t backfoffs = consumer->getNumBackoffs();
2764-
2765-
/* Make a call to the function that would be called by the processor
2766-
task here */
2767-
if (beyondThreshold) {
2768-
EXPECT_EQ(more_to_process, consumer->processBufferedItems());
2769-
} else {
2770-
EXPECT_EQ(cannot_process, consumer->processBufferedItems());
2771-
}
2772-
2773-
EXPECT_EQ(backfoffs + 1, consumer->getNumBackoffs());
2774-
2775-
/* In 'couchbase' buckets we buffer the replica items and indirectly
2776-
throttle replication by not sending flow control acks to the
2777-
producer. Hence we do not drop the connection here */
2778-
EXPECT_EQ(ENGINE_SUCCESS, consumer->step(&producers));
2779-
2780-
/* Close stream before deleting the connection */
2781-
EXPECT_EQ(ENGINE_SUCCESS, consumer->closeStream(opaque, vbid));
2782-
}
2783-
destroy_mock_cookie(cookie);
2784-
}
2785-
2786-
/* Here we test how the Processor task in DCP consumer handles the scenario
2787-
where the memory usage is beyond the replication throttle threshold.
2788-
In case of Ephemeral buckets with 'fail_new_data' policy it is expected to
2789-
indicate close of the consumer conn and in other cases it is expected to
2790-
just defer processing. */
2791-
TEST_P(ConnectionTest, ProcessReplicationBufferAfterThrottleThreshold) {
2792-
processConsumerMutationsNearThreshold(true);
2793-
}
27942665

2795-
/* Here we test how the Processor task in DCP consumer handles the scenario
2796-
where the memory usage is just below the replication throttle threshold,
2797-
but will go over the threshold when it adds the new mutation from the
2798-
processor buffer to the hashtable.
2799-
In case of Ephemeral buckets with 'fail_new_data' policy it is expected to
2800-
indicate close of the consumer conn and in other cases it is expected to
2801-
just defer processing. */
2802-
TEST_P(ConnectionTest,
2803-
DISABLED_ProcessReplicationBufferJustBeforeThrottleThreshold) {
2804-
/* There are sporadic failures seen while testing this. The problem is
2805-
we need to have a memory usage just below max_size, so we need to
2806-
start at that point. But sometimes the memory usage goes further below
2807-
resulting in the test failure (a hang). Hence commenting out the test.
2808-
Can be run locally as and when needed. */
2809-
processConsumerMutationsNearThreshold(false);
2810-
}
28112666

28122667
TEST_P(ConnectionTest, ProducerEnablesDeleteXattr) {
28132668
const void* cookie = create_mock_cookie();

0 commit comments

Comments
 (0)