Skip to content

Commit 1cd1265

Browse files
BenHuddlestondaverigby
BenHuddleston
authored and
daverigby
committed
MB-41719: Make sendConsumerMutationsNearThreshold ST
Change-Id: I1cde07ccd5fececbeb22e7dc92893bf622787c52 Reviewed-on: http://review.couchbase.org/c/kv_engine/+/137425 Tested-by: Build Bot <[email protected]> Reviewed-by: Dave Rigby <[email protected]>
1 parent 5057149 commit 1cd1265

File tree

2 files changed

+151
-149
lines changed

2 files changed

+151
-149
lines changed

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

Lines changed: 151 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -94,6 +94,15 @@ class STDcpTest : public STParameterizedBucketTest {
9494
* threshold or just below it
9595
*/
9696
void processConsumerMutationsNearThreshold(bool beyondThreshold);
97+
98+
/**
99+
* Creates a consumer conn and sends items on the conn with memory usage
100+
* near to replication threshold
101+
*
102+
* @param beyondThreshold indicates if the memory usage should above the
103+
* threshold or just below it
104+
*/
105+
void sendConsumerMutationsNearThreshold(bool beyondThreshold);
97106
};
98107

99108
/*
@@ -696,6 +705,148 @@ TEST_P(STDcpTest, test_producer_no_stream_end_on_client_close_stream) {
696705
connMap.manageConnections();
697706
}
698707

708+
void STDcpTest::sendConsumerMutationsNearThreshold(bool beyondThreshold) {
709+
setVBucketStateAndRunPersistTask(vbid, vbucket_state_replica);
710+
711+
const void* cookie = create_mock_cookie(engine.get());
712+
const uint32_t opaque = 1;
713+
const uint64_t snapStart = 1;
714+
const uint64_t snapEnd = std::numeric_limits<uint64_t>::max();
715+
uint64_t bySeqno = snapStart;
716+
717+
/* Set up a consumer connection */
718+
auto& connMap = static_cast<MockDcpConnMap&>(engine->getDcpConnMap());
719+
auto consumer =
720+
std::make_shared<MockDcpConsumer>(*engine, cookie, "test_consumer");
721+
connMap.addConn(cookie, consumer);
722+
723+
/* Passive stream */
724+
ASSERT_EQ(ENGINE_SUCCESS,
725+
consumer->addStream(/*opaque*/ 0,
726+
vbid,
727+
/*flags*/ 0));
728+
MockPassiveStream* stream = static_cast<MockPassiveStream*>(
729+
(consumer->getVbucketStream(vbid)).get());
730+
ASSERT_TRUE(stream->isActive());
731+
732+
/* Send a snapshotMarker before sending items for replication */
733+
EXPECT_EQ(ENGINE_SUCCESS,
734+
consumer->snapshotMarker(opaque,
735+
vbid,
736+
snapStart,
737+
snapEnd,
738+
/* in-memory snapshot */ 0x1,
739+
{} /*HCS*/,
740+
{} /*maxVisibleSeq*/));
741+
742+
/* Send an item for replication */
743+
const DocKey docKey{nullptr, 0, DocKeyEncodesCollectionId::No};
744+
EXPECT_EQ(ENGINE_SUCCESS,
745+
consumer->mutation(opaque,
746+
docKey,
747+
{}, // value
748+
0, // priv bytes
749+
PROTOCOL_BINARY_RAW_BYTES,
750+
0, // cas
751+
vbid,
752+
0, // flags
753+
bySeqno,
754+
0, // rev seqno
755+
0, // exptime
756+
0, // locktime
757+
{}, // meta
758+
0)); // nru
759+
760+
/* Set 'mem_used' beyond the 'replication threshold' */
761+
EPStats& stats = engine->getEpStats();
762+
if (beyondThreshold) {
763+
engine->setMaxDataSize(stats.getPreciseTotalMemoryUsed());
764+
} else {
765+
/* Set 'mem_used' just 1 byte less than the 'replication threshold'.
766+
That is we are below 'replication threshold', but not enough space
767+
for the new item */
768+
engine->setMaxDataSize(stats.getPreciseTotalMemoryUsed() + 1);
769+
/* Simpler to set the replication threshold to 1 and test, rather than
770+
testing with maxData = (memUsed / replicationThrottleThreshold);
771+
that is, we are avoiding a division */
772+
engine->getConfiguration().setReplicationThrottleThreshold(100);
773+
}
774+
775+
if ((engine->getConfiguration().getBucketType() == "ephemeral") &&
776+
(engine->getConfiguration().getEphemeralFullPolicy()) ==
777+
"fail_new_data") {
778+
/* Expect disconnect signal in Ephemeral with "fail_new_data" policy */
779+
while (true) {
780+
/* Keep sending items till the memory usage goes above the
781+
threshold and the connection is disconnected */
782+
if (ENGINE_DISCONNECT ==
783+
consumer->mutation(opaque,
784+
docKey,
785+
{}, // value
786+
0, // priv bytes
787+
PROTOCOL_BINARY_RAW_BYTES,
788+
0, // cas
789+
vbid,
790+
0, // flags
791+
++bySeqno,
792+
0, // rev seqno
793+
0, // exptime
794+
0, // locktime
795+
{}, // meta
796+
0)) {
797+
break;
798+
}
799+
}
800+
} else {
801+
/* In 'couchbase' buckets we buffer the replica items and indirectly
802+
throttle replication by not sending flow control acks to the
803+
producer. Hence we do not drop the connection here */
804+
EXPECT_EQ(ENGINE_SUCCESS,
805+
consumer->mutation(opaque,
806+
docKey,
807+
{}, // value
808+
0, // priv bytes
809+
PROTOCOL_BINARY_RAW_BYTES,
810+
0, // cas
811+
vbid,
812+
0, // flags
813+
bySeqno + 1,
814+
0, // rev seqno
815+
0, // exptime
816+
0, // locktime
817+
{}, // meta
818+
0)); // nru
819+
}
820+
821+
/* Close stream before deleting the connection */
822+
EXPECT_EQ(ENGINE_SUCCESS, consumer->closeStream(opaque, vbid));
823+
824+
connMap.disconnect(cookie);
825+
EXPECT_FALSE(connMap.isDeadConnectionsEmpty());
826+
connMap.manageConnections();
827+
EXPECT_TRUE(connMap.isDeadConnectionsEmpty());
828+
}
829+
830+
/* Here we test how the DCP consumer handles the scenario where the memory
831+
usage is beyond the replication throttle threshold.
832+
In case of Ephemeral buckets with 'fail_new_data' policy it is expected to
833+
indicate close of the consumer conn and in other cases it is expected to
834+
just defer processing. */
835+
TEST_P(STDcpTest, ReplicateAfterThrottleThreshold) {
836+
sendConsumerMutationsNearThreshold(true);
837+
}
838+
839+
/* Here we test how the DCP consumer handles the scenario where the memory
840+
usage is just below the replication throttle threshold, but will go over the
841+
threshold when it adds the new mutation from the processor buffer to the
842+
hashtable.
843+
In case of Ephemeral buckets with 'fail_new_data' policy it is expected to
844+
indicate close of the consumer conn and in other cases it is expected to
845+
just defer processing. */
846+
TEST_P(STDcpTest, ReplicateJustBeforeThrottleThreshold) {
847+
sendConsumerMutationsNearThreshold(false);
848+
}
849+
699850
INSTANTIATE_TEST_SUITE_P(PersistentAndEphemeral,
700851
STDcpTest,
701852
STParameterizedBucketTest::allConfigValues());

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

Lines changed: 0 additions & 149 deletions
Original file line numberDiff line numberDiff line change
@@ -824,15 +824,6 @@ class ConnectionTest : public DCPTest,
824824
vbid, state, {}, TransferVB::Yes);
825825
}
826826

827-
/**
828-
* Creates a consumer conn and sends items on the conn with memory usage
829-
* near to replication threshold
830-
*
831-
* @param beyondThreshold indicates if the memory usage should above the
832-
* threshold or just below it
833-
*/
834-
void sendConsumerMutationsNearThreshold(bool beyondThreshold);
835-
836827
/* vbucket associated with this connection */
837828
Vbid vbid;
838829
};
@@ -2428,146 +2419,6 @@ TEST_P(ConnectionTest, test_mb24424_mutationResponse) {
24282419
destroy_mock_cookie(cookie);
24292420
}
24302421

2431-
void ConnectionTest::sendConsumerMutationsNearThreshold(bool beyondThreshold) {
2432-
const void* cookie = create_mock_cookie(engine);
2433-
const uint32_t opaque = 1;
2434-
const uint64_t snapStart = 1;
2435-
const uint64_t snapEnd = std::numeric_limits<uint64_t>::max();
2436-
uint64_t bySeqno = snapStart;
2437-
2438-
/* Set up a consumer connection */
2439-
auto consumer =
2440-
std::make_shared<MockDcpConsumer>(*engine, cookie, "test_consumer");
2441-
2442-
/* Replica vbucket */
2443-
ASSERT_EQ(ENGINE_SUCCESS, set_vb_state(vbid, vbucket_state_replica));
2444-
2445-
/* Passive stream */
2446-
ASSERT_EQ(ENGINE_SUCCESS,
2447-
consumer->addStream(/*opaque*/ 0,
2448-
vbid,
2449-
/*flags*/ 0));
2450-
MockPassiveStream* stream = static_cast<MockPassiveStream*>(
2451-
(consumer->getVbucketStream(vbid)).get());
2452-
ASSERT_TRUE(stream->isActive());
2453-
2454-
/* Send a snapshotMarker before sending items for replication */
2455-
EXPECT_EQ(ENGINE_SUCCESS,
2456-
consumer->snapshotMarker(opaque,
2457-
vbid,
2458-
snapStart,
2459-
snapEnd,
2460-
/* in-memory snapshot */ 0x1,
2461-
{} /*HCS*/,
2462-
{} /*maxVisibleSeq*/));
2463-
2464-
/* Send an item for replication */
2465-
const DocKey docKey{nullptr, 0, DocKeyEncodesCollectionId::No};
2466-
EXPECT_EQ(ENGINE_SUCCESS,
2467-
consumer->mutation(opaque,
2468-
docKey,
2469-
{}, // value
2470-
0, // priv bytes
2471-
PROTOCOL_BINARY_RAW_BYTES,
2472-
0, // cas
2473-
vbid,
2474-
0, // flags
2475-
bySeqno,
2476-
0, // rev seqno
2477-
0, // exptime
2478-
0, // locktime
2479-
{}, // meta
2480-
0)); // nru
2481-
2482-
/* Set 'mem_used' beyond the 'replication threshold' */
2483-
EPStats& stats = engine->getEpStats();
2484-
if (beyondThreshold) {
2485-
engine->setMaxDataSize(stats.getPreciseTotalMemoryUsed());
2486-
} else {
2487-
/* Set 'mem_used' just 1 byte less than the 'replication threshold'.
2488-
That is we are below 'replication threshold', but not enough space
2489-
for the new item */
2490-
engine->setMaxDataSize(stats.getPreciseTotalMemoryUsed() + 1);
2491-
/* Simpler to set the replication threshold to 1 and test, rather than
2492-
testing with maxData = (memUsed / replicationThrottleThreshold);
2493-
that is, we are avoiding a division */
2494-
engine->getConfiguration().setReplicationThrottleThreshold(100);
2495-
}
2496-
2497-
if ((engine->getConfiguration().getBucketType() == "ephemeral") &&
2498-
(engine->getConfiguration().getEphemeralFullPolicy()) ==
2499-
"fail_new_data") {
2500-
/* Expect disconnect signal in Ephemeral with "fail_new_data" policy */
2501-
while (true) {
2502-
/* Keep sending items till the memory usage goes above the
2503-
threshold and the connection is disconnected */
2504-
if (ENGINE_DISCONNECT ==
2505-
consumer->mutation(opaque,
2506-
docKey,
2507-
{}, // value
2508-
0, // priv bytes
2509-
PROTOCOL_BINARY_RAW_BYTES,
2510-
0, // cas
2511-
vbid,
2512-
0, // flags
2513-
++bySeqno,
2514-
0, // rev seqno
2515-
0, // exptime
2516-
0, // locktime
2517-
{}, // meta
2518-
0)) {
2519-
break;
2520-
}
2521-
}
2522-
} else {
2523-
/* In 'couchbase' buckets we buffer the replica items and indirectly
2524-
throttle replication by not sending flow control acks to the
2525-
producer. Hence we do not drop the connection here */
2526-
EXPECT_EQ(ENGINE_SUCCESS,
2527-
consumer->mutation(opaque,
2528-
docKey,
2529-
{}, // value
2530-
0, // priv bytes
2531-
PROTOCOL_BINARY_RAW_BYTES,
2532-
0, // cas
2533-
vbid,
2534-
0, // flags
2535-
bySeqno + 1,
2536-
0, // rev seqno
2537-
0, // exptime
2538-
0, // locktime
2539-
{}, // meta
2540-
0)); // nru
2541-
}
2542-
2543-
/* Close stream before deleting the connection */
2544-
EXPECT_EQ(ENGINE_SUCCESS, consumer->closeStream(opaque, vbid));
2545-
2546-
destroy_mock_cookie(cookie);
2547-
}
2548-
2549-
/* Here we test how the DCP consumer handles the scenario where the memory
2550-
usage is beyond the replication throttle threshold.
2551-
In case of Ephemeral buckets with 'fail_new_data' policy it is expected to
2552-
indicate close of the consumer conn and in other cases it is expected to
2553-
just defer processing. */
2554-
TEST_P(ConnectionTest, ReplicateAfterThrottleThreshold) {
2555-
sendConsumerMutationsNearThreshold(true);
2556-
}
2557-
2558-
/* Here we test how the DCP consumer handles the scenario where the memory
2559-
usage is just below the replication throttle threshold, but will go over the
2560-
threshold when it adds the new mutation from the processor buffer to the
2561-
hashtable.
2562-
In case of Ephemeral buckets with 'fail_new_data' policy it is expected to
2563-
indicate close of the consumer conn and in other cases it is expected to
2564-
just defer processing. */
2565-
TEST_P(ConnectionTest, ReplicateJustBeforeThrottleThreshold) {
2566-
sendConsumerMutationsNearThreshold(false);
2567-
}
2568-
2569-
2570-
25712422
TEST_P(ConnectionTest, ProducerEnablesDeleteXattr) {
25722423
const void* cookie = create_mock_cookie();
25732424

0 commit comments

Comments
 (0)