From bedf40bddc8a47ee23fa7e7298a656b3a13b29ec Mon Sep 17 00:00:00 2001 From: kruall Date: Mon, 13 Jul 2026 12:37:18 +0500 Subject: Separate read queues for channels in keyvalue tablet (#45783) --- ydb/core/keyvalue/keyvalue_events.h | 9 +- ydb/core/keyvalue/keyvalue_flat_impl.h | 3 +- ydb/core/keyvalue/keyvalue_intermediate.cpp | 3 +- ydb/core/keyvalue/keyvalue_intermediate.h | 3 + ydb/core/keyvalue/keyvalue_state.cpp | 265 ++++++++-- ydb/core/keyvalue/keyvalue_state.h | 37 +- .../keyvalue/keyvalue_storage_read_request.cpp | 3 +- ydb/core/keyvalue/keyvalue_storage_request.cpp | 3 +- ydb/core/keyvalue/keyvalue_ut.cpp | 588 +++++++++++++++++++++ ydb/core/protos/config.proto | 6 + 10 files changed, 871 insertions(+), 49 deletions(-) diff --git a/ydb/core/keyvalue/keyvalue_events.h b/ydb/core/keyvalue/keyvalue_events.h index ddf615ee276..37da171ed7c 100644 --- a/ydb/core/keyvalue/keyvalue_events.h +++ b/ydb/core/keyvalue/keyvalue_events.h @@ -171,27 +171,32 @@ namespace TEvKeyValue { ui64 Step; NKeyValue::TRequestStat Stat; NMsgBusProxy::EResponseStatus Status; + TVector AcquiredChannels; std::deque> RefCountsIncr; TEvNotify() { } TEvNotify(ui64 requestUid, ui64 generation, ui64 step, const NKeyValue::TRequestStat &stat, - NMsgBusProxy::EResponseStatus status, std::deque>&& refCountsIncr) + NMsgBusProxy::EResponseStatus status, TVector acquiredChannels, + std::deque>&& refCountsIncr) : RequestUid(requestUid) , Generation(generation) , Step(step) , Stat(stat) , Status(status) + , AcquiredChannels(std::move(acquiredChannels)) , RefCountsIncr(std::move(refCountsIncr)) {} TEvNotify(ui64 requestUid, ui64 generation, ui64 step, const NKeyValue::TRequestStat &stat, - NKikimrKeyValue::Statuses::ReplyStatus status, std::deque>&& refCountsIncr) + NKikimrKeyValue::Statuses::ReplyStatus status, TVector acquiredChannels, + std::deque>&& refCountsIncr) : RequestUid(requestUid) , Generation(generation) , Step(step) , Stat(stat) , Status(ConvertStatus(status)) + , AcquiredChannels(std::move(acquiredChannels)) , RefCountsIncr(std::move(refCountsIncr)) {} diff --git a/ydb/core/keyvalue/keyvalue_flat_impl.h b/ydb/core/keyvalue/keyvalue_flat_impl.h index a2f9b6bbd4d..b89be58e860 100644 --- a/ydb/core/keyvalue/keyvalue_flat_impl.h +++ b/ydb/core/keyvalue/keyvalue_flat_impl.h @@ -491,7 +491,8 @@ protected: {"ev", event}); CheckYellowChannels(ev->Get()->Stat); - State.OnRequestComplete(event.RequestUid, event.Generation, event.Step, ctx, Info(), event.Status, event.Stat); + State.OnRequestComplete(event.RequestUid, event.Generation, event.Step, ctx, Info(), event.Status, event.Stat, + event.AcquiredChannels); State.DropRefCountsOnError(event.RefCountsIncr, true, ctx); if (!event.RefCountsIncr.empty()) { Execute(new TTxDropRefCountsOnError(std::move(event.RefCountsIncr), this), ctx); diff --git a/ydb/core/keyvalue/keyvalue_intermediate.cpp b/ydb/core/keyvalue/keyvalue_intermediate.cpp index b48d35ef626..4e860c6b2d7 100644 --- a/ydb/core/keyvalue/keyvalue_intermediate.cpp +++ b/ydb/core/keyvalue/keyvalue_intermediate.cpp @@ -63,7 +63,8 @@ TRope TIntermediate::TRead::BuildRope() { TIntermediate::TIntermediate(TActorId respondTo, TActorId keyValueActorId, ui64 channelGeneration, ui64 channelStep, TRequestType::EType requestType, NWilson::TTraceId traceId) - : Cookie(0) + : PostponedQueuesLeft(0) + , Cookie(0) , Generation(0) , Deadline(TInstant::Max()) , HasCookie(false) diff --git a/ydb/core/keyvalue/keyvalue_intermediate.h b/ydb/core/keyvalue/keyvalue_intermediate.h index c4017e8f30c..05da58a87ec 100644 --- a/ydb/core/keyvalue/keyvalue_intermediate.h +++ b/ydb/core/keyvalue/keyvalue_intermediate.h @@ -148,6 +148,9 @@ struct TIntermediate { ui64 CopyRangeCount = 0; ui64 ConcatCount = 0; + TVector AcquiredChannels; + ui32 PostponedQueuesLeft; + ui64 Cookie; ui64 Generation; ui64 RequestUid; diff --git a/ydb/core/keyvalue/keyvalue_state.cpp b/ydb/core/keyvalue/keyvalue_state.cpp index 6d6911ebbcc..fa3d6c2c770 100644 --- a/ydb/core/keyvalue/keyvalue_state.cpp +++ b/ydb/core/keyvalue/keyvalue_state.cpp @@ -98,11 +98,47 @@ TKeyValueState::TKeyValueState() , UsePayload(UsePayload_Base) , RejectNonExistentStorageChannel_Base(0, 0, 1) , RejectNonExistentStorageChannel(RejectNonExistentStorageChannel_Base) + , UsePerChannelReadQueues_Base(0, 0, 1) + , UsePerChannelReadQueues(UsePerChannelReadQueues_Base) { TabletCounters = nullptr; Clear(); } +TKeyValueState::TPostponedChannel::TPostponedChannel(TPostponedChannel&& other) noexcept { + Queue.swap(other.Queue); + IntermediatesInFlight = other.IntermediatesInFlight; + other.IntermediatesInFlight = 0; +} + +TKeyValueState::TPostponedChannel& TKeyValueState::TPostponedChannel::operator=( + TPostponedChannel&& other) noexcept { + if (this != &other) { + ClearQueue(); + Queue.swap(other.Queue); + IntermediatesInFlight = other.IntermediatesInFlight; + other.IntermediatesInFlight = 0; + } + return *this; +} + +TKeyValueState::TPostponedChannel::~TPostponedChannel() { + ClearQueue(); +} + +void TKeyValueState::TPostponedChannel::ClearQueue() { + while (!Queue.empty()) { + TIntermediate *intermediate = Queue.front(); + Queue.pop_front(); + Y_ABORT_UNLESS(intermediate); + Y_ABORT_UNLESS(intermediate->PostponedQueuesLeft); + --intermediate->PostponedQueuesLeft; + if (intermediate->PostponedQueuesLeft == 0) { + delete intermediate; + } + } +} + bool TKeyValueState::RejectNonExistentStorageChannelEnabled(const TActorContext& ctx) { return RejectNonExistentStorageChannel.Update(ctx.Now()) != 0; } @@ -133,7 +169,8 @@ void TKeyValueState::Clear() { KeyValueActorId = TActorId(); ExecutorGeneration = 0; - Queue.clear(); + PostponedChannels.clear(); + PostponedIntermediatesCount = 0; IntermediatesInFlight = 0; RoInlineIntermediatesInFlight = 0; DeletesPerRequestLimit = 100'000; @@ -305,7 +342,7 @@ void TKeyValueState::CountRequestTakeOffOrEnqueue(TRequestType::EType requestTyp } else { TabletCounters->Simple()[COUNTER_REQ_RO_RW_IN_FLY].Set(IntermediatesInFlight); } - TabletCounters->Simple()[COUNTER_REQ_RO_RW_QUEUED].Set(Queue.size()); + TabletCounters->Simple()[COUNTER_REQ_RO_RW_QUEUED].Set(PostponedIntermediatesCount); } } @@ -553,11 +590,13 @@ void TKeyValueState::Load(const TString &key, const TString& value) { void TKeyValueState::InitExecute(ui64 tabletId, TActorId keyValueActorId, ui32 executorGeneration, ISimpleDb &db, const TActorContext &ctx, const TTabletStorageInfo *info) { - Y_UNUSED(info); Y_ABORT_UNLESS(IsEmptyDbStart || IsStatePresent); TabletId = tabletId; KeyValueActorId = keyValueActorId; ExecutorGeneration = executorGeneration; + PostponedChannels.clear(); + PostponedIntermediatesCount = 0; + PostponedChannels.resize(info->Channels.size()); if (IsDamaged) { return; } @@ -582,12 +621,15 @@ void TKeyValueState::InitExecute(ui64 tabletId, TActorId keyValueActorId, ui32 e UsePayload.ResetControl(UsePayload_Base); TControlBoard::RegisterSharedControl(RejectNonExistentStorageChannel_Base, icb->KeyValueVolumeControls.RejectNonExistentStorageChannel); RejectNonExistentStorageChannel.ResetControl(RejectNonExistentStorageChannel_Base); + TControlBoard::RegisterSharedControl(UsePerChannelReadQueues_Base, icb->KeyValueVolumeControls.UsePerChannelReadQueues); + UsePerChannelReadQueues.ResetControl(UsePerChannelReadQueues_Base); YDB_LOG_DEBUG("Init KeyValue with ICB", {"keyValue", TabletId}, {"usePayload", UsePayload.Update(ctx.Now())}, {"readRequestsInFlightLimit", ReadRequestsInFlightLimit.Update(ctx.Now())}, {"rejectNonExistentStorageChannel", RejectNonExistentStorageChannel.Update(ctx.Now())}, + {"usePerChannelReadQueues", UsePerChannelReadQueues.Update(ctx.Now())}, {"marker", "KV92"}); } @@ -998,7 +1040,8 @@ void TKeyValueState::Reply(THolder &intermediate, const TActorCon CountLatencyLocalBase(*intermediate); OnRequestComplete(intermediate->RequestUid, intermediate->CreatedAtGeneration, intermediate->CreatedAtStep, - ctx, info, (NMsgBusProxy::EResponseStatus)intermediate->Response.GetStatus(), intermediate->Stat); + ctx, info, (NMsgBusProxy::EResponseStatus)intermediate->Response.GetStatus(), intermediate->Stat, + intermediate->AcquiredChannels); } } @@ -1858,8 +1901,175 @@ void TKeyValueState::OnUpdateWeights(TChannelBalancer::TEvUpdateWeights::TPtr ev WeightManager = std::move(ev->Get()->WeightManager); } +TVector TKeyValueState::GetAcquiredChannels(const TIntermediate &intermediate) const { + std::bitset<256> acquiredChannels; + + auto addAcquiredChannel = [&](ui32 channel) { + Y_DEBUG_ABORT_UNLESS(channel < acquiredChannels.size()); + acquiredChannels.set(channel); + }; + + auto addRead = [&](const TIntermediate::TRead& read) { + if (!read.ReadItems.empty()) { + for (const TIntermediate::TRead::TReadItem& item : read.ReadItems) { + addAcquiredChannel(item.LogoBlobId.Channel()); + } + } + }; + + auto addRangeRead = [&](const TIntermediate::TRangeRead& rangeRead) { + for (const TIntermediate::TRead& read : rangeRead.Reads) { + addRead(read); + } + }; + + auto addPatch = [&](const TIntermediate::TPatch& patch) { + addAcquiredChannel(patch.OriginalBlobId.Channel()); + }; + + for (const TIntermediate::TRead& read : intermediate.Reads) { + addRead(read); + } + for (const TIntermediate::TRangeRead& rangeRead : intermediate.RangeReads) { + addRangeRead(rangeRead); + } + for (const TIntermediate::TPatch& patch : intermediate.Patches) { + addPatch(patch); + } + if (intermediate.ReadCommand) { + std::visit([&](const auto& command) { + using TCommand = std::decay_t; + if constexpr (std::is_same_v) { + addRead(command); + } else if constexpr (std::is_same_v) { + addRangeRead(command); + } + }, *intermediate.ReadCommand); + } + for (const TIntermediate::TCmd& command : intermediate.Commands) { + std::visit([&](const auto& cmd) { + using TCommand = std::decay_t; + if constexpr (std::is_same_v) { + addPatch(cmd); + } + }, command); + } + if (intermediate.TrimLeakedBlobs) { + for (const auto& [channel, _] : intermediate.TrimLeakedBlobs->ChannelGroupMap) { + addAcquiredChannel(channel); + } + } + + TVector channels; + for (size_t channel = 2; channel < PostponedChannels.size(); ++channel) { + if (acquiredChannels.test(channel)) { + channels.push_back(static_cast(channel)); + } + } + return channels; +} + +void TKeyValueState::StartChannelLimitedIntermediate(const TIntermediate &intermediate) { + Y_DEBUG_ABORT_UNLESS(intermediate.Stat.RequestType != TRequestType::WriteOnly); + Y_DEBUG_ABORT_UNLESS(intermediate.Stat.RequestType != TRequestType::ReadOnlyInline); + Y_DEBUG_ABORT_UNLESS(intermediate.PostponedQueuesLeft == 0); + Y_DEBUG_ABORT_UNLESS(!intermediate.AcquiredChannels.empty()); + ++IntermediatesInFlight; +} + +bool TKeyValueState::TryStartOrPostponeIntermediate(THolder &intermediate, const TActorContext &ctx) { + Y_DEBUG_ABORT_UNLESS(intermediate->Stat.RequestType != TRequestType::WriteOnly); + Y_DEBUG_ABORT_UNLESS(intermediate->Stat.RequestType != TRequestType::ReadOnlyInline); + const TInstant now = ctx.Now(); + intermediate->AcquiredChannels = GetAcquiredChannels(*intermediate); + intermediate->PostponedQueuesLeft = 0; + if (intermediate->AcquiredChannels.empty()) { + return true; + } + + if (!UsePerChannelReadQueues.Update(now)) { + Y_DEBUG_ABORT_UNLESS(BLOB_CHANNEL < PostponedChannels.size()); + intermediate->AcquiredChannels = {BLOB_CHANNEL}; + } + + const ui64 limit = ReadRequestsInFlightLimit.Update(now); + TIntermediate *rawIntermediate = intermediate.Get(); + for (ui32 channel : intermediate->AcquiredChannels) { + TPostponedChannel& channelState = PostponedChannels[channel]; + if (!channelState.Queue.empty() || channelState.IntermediatesInFlight >= limit) { + channelState.Queue.push_back(rawIntermediate); + ++intermediate->PostponedQueuesLeft; + } else { + ++channelState.IntermediatesInFlight; + } + } + + if (intermediate->PostponedQueuesLeft == 0) { + StartChannelLimitedIntermediate(*intermediate); + return true; + } + + intermediate->Stat.EnqueuedAs = PostponedIntermediatesCount + 1; + ++PostponedIntermediatesCount; + // Queued raw pointers keep this intermediate alive via PostponedQueuesLeft. + Y_UNUSED(intermediate.Release()); + return false; +} + +void TKeyValueState::ReleaseChannelLimitedIntermediate(const TVector &acquiredChannels) { + if (acquiredChannels.empty()) { + return; + } + + Y_DEBUG_ABORT_UNLESS(IntermediatesInFlight); + --IntermediatesInFlight; + + for (ui32 channel : acquiredChannels) { + TPostponedChannel& channelState = PostponedChannels[channel]; + Y_DEBUG_ABORT_UNLESS(channelState.IntermediatesInFlight); + --channelState.IntermediatesInFlight; + } +} + +void TKeyValueState::ProcessPostponedChannel(ui32 channel, const TActorContext &ctx, const TTabletStorageInfo *info) { + TPostponedChannel& channelState = PostponedChannels[channel]; + const ui64 limit = ReadRequestsInFlightLimit.Update(ctx.Now()); + + while (!channelState.Queue.empty() && channelState.IntermediatesInFlight < limit) { + TIntermediate *intermediate = channelState.Queue.front(); + channelState.Queue.pop_front(); + Y_DEBUG_ABORT_UNLESS(intermediate); + + ++channelState.IntermediatesInFlight; + Y_DEBUG_ABORT_UNLESS(intermediate->PostponedQueuesLeft); + --intermediate->PostponedQueuesLeft; + if (intermediate->PostponedQueuesLeft != 0) { + continue; + } + + Y_DEBUG_ABORT_UNLESS(PostponedIntermediatesCount); + --PostponedIntermediatesCount; + THolder ready(intermediate); + + StartChannelLimitedIntermediate(*ready); + CountLatencyQueue(ready->Stat); + const TRequestType::EType requestType = ready->Stat.RequestType; + + ProcessPostponedIntermediate(ctx, std::move(ready), info); + CountRequestTakeOffOrEnqueue(requestType); + } +} + +void TKeyValueState::ProcessPostponedChannels(const TVector &channels, const TActorContext &ctx, + const TTabletStorageInfo *info) { + for (ui32 channel : channels) { + ProcessPostponedChannel(channel, ctx, info); + } +} + void TKeyValueState::OnRequestComplete(ui64 requestUid, ui64 generation, ui64 step, const TActorContext &ctx, - const TTabletStorageInfo *info, NMsgBusProxy::EResponseStatus status, const TRequestStat &stat) { + const TTabletStorageInfo *info, NMsgBusProxy::EResponseStatus status, const TRequestStat &stat, + const TVector &acquiredChannels) { YDB_LOG_DEBUG("OnRequestComplete", {"keyValue", TabletId}, {"uid", requestUid}, @@ -1872,28 +2082,22 @@ void TKeyValueState::OnRequestComplete(ui64 requestUid, ui64 generation, ui64 st RequestInputTime.erase(requestUid); + TVector releasedChannels; if (stat.RequestType != TRequestType::WriteOnly) { if (stat.RequestType == TRequestType::ReadOnlyInline) { RoInlineIntermediatesInFlight--; } else { - IntermediatesInFlight--; + ReleaseChannelLimitedIntermediate(acquiredChannels); + // Processing postponed queues may re-enter and destroy the source intermediate. + releasedChannels = acquiredChannels; } } CountRequestComplete(status, stat, ctx); ResourceMetrics->TryUpdate(ctx); - ui64 limit = ReadRequestsInFlightLimit.Update(ctx.Now()); - if (Queue.size() && IntermediatesInFlight < limit) { - TRequestType::EType requestType = Queue.front()->Stat.RequestType; - - CountLatencyQueue(Queue.front()->Stat); - - ProcessPostponedIntermediate(ctx, std::move(Queue.front()), info); - Queue.pop_front(); - ++IntermediatesInFlight; - - CountRequestTakeOffOrEnqueue(requestType); + if (!releasedChannels.empty()) { + ProcessPostponedChannels(releasedChannels, ctx, info); } CmdTrimLeakedBlobsUids.erase(requestUid); @@ -2947,7 +3151,7 @@ void TKeyValueState::ReplyError(const TActorContext &ctx, TString errorDescripti if (info) { intermediate->UpdateStat(); OnRequestComplete(intermediate->RequestUid, intermediate->CreatedAtGeneration, intermediate->CreatedAtStep, - ctx, info, oldStatus, intermediate->Stat); + ctx, info, oldStatus, intermediate->Stat, intermediate->AcquiredChannels); } else { //metrics change report in OnRequestComplete is not done ResourceMetrics->TryUpdate(ctx); RequestInputTime.erase(intermediate->RequestUid); @@ -3277,6 +3481,11 @@ void TKeyValueState::ProcessPostponedTrims(const TActorContext& ctx, const TTabl if (!IsCollectEventSent) { for (auto& interm : CmdTrimLeakedBlobsPostponed) { CmdTrimLeakedBlobsUids.insert(interm->RequestUid); + const TRequestType::EType requestType = interm->Stat.RequestType; + if (requestType == TRequestType::ReadOnlyInline) { + ++RoInlineIntermediatesInFlight; + CountRequestTakeOffOrEnqueue(requestType); + } RegisterRequestActor(ctx, std::move(interm), info, ExecutorGeneration); } CmdTrimLeakedBlobsPostponed.clear(); @@ -3302,22 +3511,17 @@ void TKeyValueState::OnEvReadRequest(TEvKeyValue::TEvRead::TPtr &ev, const TActo RegisterReadRequestActor(ctx, std::move(intermediate), info, ExecutorGeneration); ++RoInlineIntermediatesInFlight; } else { - ui64 limit = ReadRequestsInFlightLimit.Update(ctx.Now()); - if (IntermediatesInFlight < limit) { - ++IntermediatesInFlight; + if (TryStartOrPostponeIntermediate(intermediate, ctx)) { YDB_LOG_DEBUG("Create storage read request, /", {"keyValue", TabletId}, {"inFlight", IntermediatesInFlight}, - {"limit", limit}, {"marker", "KV54"}); RegisterReadRequestActor(ctx, std::move(intermediate), info, ExecutorGeneration); } else { YDB_LOG_DEBUG("Enqueue storage read request /", {"keyValue", TabletId}, {"intermediatesInFlight", IntermediatesInFlight}, - {"limit", limit}, {"marker", "KV56"}); - PostponeIntermediate(std::move(intermediate)); } } CountRequestTakeOffOrEnqueue(requestType); @@ -3346,20 +3550,16 @@ void TKeyValueState::OnEvReadRangeRequest(TEvKeyValue::TEvReadRange::TPtr &ev, c RegisterReadRequestActor(ctx, std::move(intermediate), info, ExecutorGeneration); ++RoInlineIntermediatesInFlight; } else { - ui64 limit = ReadRequestsInFlightLimit.Update(ctx.Now()); - if (IntermediatesInFlight < limit) { - ++IntermediatesInFlight; + if (TryStartOrPostponeIntermediate(intermediate, ctx)) { YDB_LOG_DEBUG("Create storage read range request, /", {"keyValue", TabletId}, {"inFlight", IntermediatesInFlight}, - {"limit", limit}, {"marker", "KV66"}); RegisterReadRequestActor(ctx, std::move(intermediate), info, ExecutorGeneration); } else { YDB_LOG_DEBUG("Enqueue storage read range request,", {"keyValue", TabletId}, {"marker", "KV59"}); - PostponeIntermediate(std::move(intermediate)); } } CountRequestTakeOffOrEnqueue(requestType); @@ -3411,7 +3611,6 @@ void TKeyValueState::OnEvGetStorageChannelStatus(TEvKeyValue::TEvGetStorageChann {"keyValue", TabletId}, {"marker", "KV75"}); RegisterRequestActor(ctx, std::move(intermediate), info, ExecutorGeneration); - ++IntermediatesInFlight; CountRequestTakeOffOrEnqueue(requestType); } else { intermediate->UpdateStat(); @@ -3497,20 +3696,16 @@ void TKeyValueState::OnEvRequest(TEvKeyValue::TEvRequest::TPtr &ev, const TActor RegisterRequestActor(ctx, std::move(intermediate), info, ExecutorGeneration); ++RoInlineIntermediatesInFlight; } else { - ui64 limit = ReadRequestsInFlightLimit.Update(ctx.Now()); - if (IntermediatesInFlight < limit) { - ++IntermediatesInFlight; + if (TryStartOrPostponeIntermediate(intermediate, ctx)) { YDB_LOG_DEBUG("Create storage request for RO/RW, /", {"keyValue", TabletId}, {"inFlight", IntermediatesInFlight}, - {"limit", limit}, {"marker", "KV43"}); RegisterRequestActor(ctx, std::move(intermediate), info, ExecutorGeneration); } else { YDB_LOG_DEBUG("Enqueue storage request for RO/RW,", {"keyValue", TabletId}, {"marker", "KV44"}); - PostponeIntermediate(std::move(intermediate)); } } } diff --git a/ydb/core/keyvalue/keyvalue_state.h b/ydb/core/keyvalue/keyvalue_state.h index d3253f7f827..4b692b2d028 100644 --- a/ydb/core/keyvalue/keyvalue_state.h +++ b/ydb/core/keyvalue/keyvalue_state.h @@ -277,7 +277,22 @@ protected: TActorId ChannelBalancerActorId; ui64 InitialCollectsSent = 0; - TDeque> Queue; + struct TPostponedChannel { + TPostponedChannel() = default; + TPostponedChannel(TPostponedChannel&& other) noexcept; + TPostponedChannel& operator=(TPostponedChannel&& other) noexcept; + TPostponedChannel(const TPostponedChannel&) = delete; + TPostponedChannel& operator=(const TPostponedChannel&) = delete; + ~TPostponedChannel(); + + void ClearQueue(); + + TDeque Queue; + ui64 IntermediatesInFlight = 0; + }; + + TVector PostponedChannels; + ui64 PostponedIntermediatesCount = 0; ui64 IntermediatesInFlight; ui64 RoInlineIntermediatesInFlight; ui64 DeletesPerRequestLimit; @@ -303,6 +318,8 @@ protected: TMemorizableControlWrapper UsePayload; TControlWrapper RejectNonExistentStorageChannel_Base; TMemorizableControlWrapper RejectNonExistentStorageChannel; + TControlWrapper UsePerChannelReadQueues_Base; + TMemorizableControlWrapper UsePerChannelReadQueues; std::shared_ptr LifetimeToken = std::make_shared(); @@ -474,7 +491,8 @@ public: void OnUpdateWeights(TChannelBalancer::TEvUpdateWeights::TPtr ev); void OnRequestComplete(ui64 requestUid, ui64 generation, ui64 step, const TActorContext &ctx, - const TTabletStorageInfo *info, NMsgBusProxy::EResponseStatus status, const TRequestStat &stat); + const TTabletStorageInfo *info, NMsgBusProxy::EResponseStatus status, const TRequestStat &stat, + const TVector &acquiredChannels); void CancelInFlight(ui64 requestUid); void OnEvIntermediate(TIntermediate &intermediate); @@ -628,7 +646,8 @@ public: if (info) { intermediate->UpdateStat(); OnRequestComplete(intermediate->RequestUid, intermediate->CreatedAtGeneration, intermediate->CreatedAtStep, - ctx, info, TEvKeyValue::TEvNotify::ConvertStatus(status), intermediate->Stat); + ctx, info, TEvKeyValue::TEvNotify::ConvertStatus(status), intermediate->Stat, + intermediate->AcquiredChannels); } else { //metrics change report in OnRequestComplete is not done ResourceMetrics->TryUpdate(ctx); RequestInputTime.erase(intermediate->RequestUid); @@ -648,11 +667,13 @@ public: bool PrepareAcquireLockRequest(const TActorContext &ctx, TEvKeyValue::TEvAcquireLock::TPtr &ev, THolder &intermediate); - template - void PostponeIntermediate(THolder &&intermediate) { - intermediate->Stat.EnqueuedAs = Queue.size() + 1; - Queue.push_back(std::move(intermediate)); - } + TVector GetAcquiredChannels(const TIntermediate &intermediate) const; + bool TryStartOrPostponeIntermediate(THolder &intermediate, const TActorContext &ctx); + void StartChannelLimitedIntermediate(const TIntermediate &intermediate); + void ReleaseChannelLimitedIntermediate(const TVector &acquiredChannels); + void ProcessPostponedChannel(ui32 channel, const TActorContext &ctx, const TTabletStorageInfo *info); + void ProcessPostponedChannels(const TVector &channels, const TActorContext &ctx, + const TTabletStorageInfo *info); void ProcessPostponedIntermediate(const TActorContext& ctx, THolder &&intermediate, const TTabletStorageInfo *info); diff --git a/ydb/core/keyvalue/keyvalue_storage_read_request.cpp b/ydb/core/keyvalue/keyvalue_storage_read_request.cpp index 4b345480079..b2eee4f2432 100644 --- a/ydb/core/keyvalue/keyvalue_storage_read_request.cpp +++ b/ydb/core/keyvalue/keyvalue_storage_read_request.cpp @@ -359,7 +359,8 @@ public: Send(IntermediateResult->KeyValueActorId, new TEvKeyValue::TEvNotify( IntermediateResult->RequestUid, IntermediateResult->CreatedAtGeneration, IntermediateResult->CreatedAtStep, - IntermediateResult->Stat, status, std::move(IntermediateResult->RefCountsIncr))); + IntermediateResult->Stat, status, std::move(IntermediateResult->AcquiredChannels), + std::move(IntermediateResult->RefCountsIncr))); } std::unique_ptr CreateReadResponse(NKikimrKeyValue::Statuses::ReplyStatus status, diff --git a/ydb/core/keyvalue/keyvalue_storage_request.cpp b/ydb/core/keyvalue/keyvalue_storage_request.cpp index 80c405e0b62..6eace78aaa8 100644 --- a/ydb/core/keyvalue/keyvalue_storage_request.cpp +++ b/ydb/core/keyvalue/keyvalue_storage_request.cpp @@ -572,7 +572,8 @@ public: ctx.Send(keyValueActorId, new TEvKeyValue::TEvNotify( IntermediateResults->RequestUid, IntermediateResults->CreatedAtGeneration, IntermediateResults->CreatedAtStep, - IntermediateResults->Stat, status, std::move(IntermediateResults->RefCountsIncr)), 0, 0, Span.GetTraceId()); + IntermediateResults->Stat, status, std::move(IntermediateResults->AcquiredChannels), + std::move(IntermediateResults->RefCountsIncr)), 0, 0, Span.GetTraceId()); Span.EndError(TStringBuilder() << status << ": " << errorDescription); Die(ctx); diff --git a/ydb/core/keyvalue/keyvalue_ut.cpp b/ydb/core/keyvalue/keyvalue_ut.cpp index dd55e89ebac..47b866afb7e 100644 --- a/ydb/core/keyvalue/keyvalue_ut.cpp +++ b/ydb/core/keyvalue/keyvalue_ut.cpp @@ -9,6 +9,8 @@ #include #include #include +#include +#include const bool ENABLE_DETAILED_KV_LOG = false; const bool ENABLE_TESTLOG_OUTPUT = false; @@ -692,6 +694,163 @@ auto ReceiveResponse(TTestContext &tc) -> decltype(std::declval( return response->Record; } +std::unique_ptr MakeReadRequest(ui64 cookie, const TVector &keys) { + auto request = std::make_unique(); + request->Record.SetCookie(cookie); + for (const TString& key : keys) { + auto read = request->Record.AddCmdRead(); + read->SetKey(key); + read->SetPriority(NKikimrClient::TKeyValueRequest::REALTIME); + } + return request; +} + +std::unique_ptr MakeWriteRequest(ui64 cookie, const TString &key, const TString &value, + NKikimrClient::TKeyValueRequest::EStorageChannel storageChannel) { + auto request = std::make_unique(); + request->Record.SetCookie(cookie); + auto write = request->Record.AddCmdWrite(); + write->SetKey(key); + write->SetValue(value); + write->SetStorageChannel(storageChannel); + write->SetPriority(NKikimrClient::TKeyValueRequest::REALTIME); + return request; +} + +NKikimrClient::TResponse ReceiveKeyValueResponse(TTestContext &tc) { + return ReceiveResponse(tc); +} + +void CheckReadResponse(const NKikimrClient::TResponse &response, ui64 cookie, const TVector &values) { + UNIT_ASSERT(response.HasCookie()); + UNIT_ASSERT_VALUES_EQUAL(response.GetCookie(), cookie); + UNIT_ASSERT(response.HasStatus()); + UNIT_ASSERT_VALUES_EQUAL(response.GetStatus(), NMsgBusProxy::MSTATUS_OK); + UNIT_ASSERT_VALUES_EQUAL(response.ReadResultSize(), values.size()); + for (ui32 i = 0; i < values.size(); ++i) { + const auto& result = response.GetReadResult(i); + UNIT_ASSERT(result.HasStatus()); + UNIT_ASSERT_VALUES_EQUAL(result.GetStatus(), static_cast(NKikimrProto::OK)); + UNIT_ASSERT(result.HasValue()); + UNIT_ASSERT_VALUES_EQUAL(result.GetValue(), values[i]); + } +} + +void CheckWriteResponse(const NKikimrClient::TResponse &response, ui64 cookie) { + UNIT_ASSERT(response.HasCookie()); + UNIT_ASSERT_VALUES_EQUAL(response.GetCookie(), cookie); + UNIT_ASSERT(response.HasStatus()); + UNIT_ASSERT_VALUES_EQUAL(response.GetStatus(), NMsgBusProxy::MSTATUS_OK); + UNIT_ASSERT_VALUES_EQUAL(response.WriteResultSize(), 1); + const auto& result = response.GetWriteResult(0); + UNIT_ASSERT(result.HasStatus()); + UNIT_ASSERT_VALUES_EQUAL(result.GetStatus(), static_cast(NKikimrProto::OK)); +} + +class TGetBlocker { +public: + explicit TGetBlocker(TTestActorRuntime& runtime) + : Runtime(runtime) + , Holder(Runtime.AddObserver( + [this](TEvBlobStorage::TEvGet::TPtr& ev) { + Process(ev); + })) + {} + + void BlockChannel(ui32 channel) { + Y_ABORT_UNLESS(channel < BlockedChannels.size()); + BlockedChannels.set(channel); + } + + ui32 Seen(ui32 channel) const { + Y_ABORT_UNLESS(channel < SeenByChannel.size()); + return SeenByChannel[channel]; + } + + ui32 TotalSeen() const { + ui32 total = 0; + for (ui32 count : SeenByChannel) { + total += count; + } + return total; + } + + size_t BlockedSize() const { + return Blocked.size(); + } + + size_t BlockedCount(ui32 channel) const { + size_t count = 0; + for (const auto& ev : Blocked) { + if (TouchesChannel(ev, channel)) { + ++count; + } + } + return count; + } + + void UnblockOne() { + Y_ABORT_UNLESS(!Blocked.empty()); + auto ev = std::move(Blocked.front()); + Blocked.pop_front(); + SendBlocked(std::move(ev)); + } + + void UnblockAll() { + while (!Blocked.empty()) { + UnblockOne(); + } + } + +private: + static bool TouchesChannel(const TEvBlobStorage::TEvGet::TPtr& ev, ui32 channel) { + const auto *msg = ev->Get(); + for (ui32 i = 0; i < msg->QuerySize; ++i) { + if (msg->Queries[i].Id.Channel() == channel) { + return true; + } + } + return false; + } + + void Process(TEvBlobStorage::TEvGet::TPtr& ev) { + IEventHandle *ptr = ev.Get(); + if (UnblockedOnce.erase(ptr)) { + return; + } + + bool shouldBlock = false; + const auto *msg = ev->Get(); + for (ui32 i = 0; i < msg->QuerySize; ++i) { + const ui32 channel = msg->Queries[i].Id.Channel(); + Y_ABORT_UNLESS(channel < SeenByChannel.size()); + ++SeenByChannel[channel]; + if (channel < BlockedChannels.size() && BlockedChannels.test(channel)) { + shouldBlock = true; + } + } + + if (shouldBlock) { + Blocked.emplace_back(std::move(ev)); + } + } + + void SendBlocked(TEvBlobStorage::TEvGet::TPtr ev) { + IEventHandle *ptr = ev.Get(); + UnblockedOnce.insert(ptr); + const ui32 nodeIdx = ev->GetRecipientRewrite().NodeId() - Runtime.GetFirstNodeId(); + Runtime.Send(ev.Release(), nodeIdx, /* viaActorSystem */ true); + } + +private: + TTestActorRuntime& Runtime; + TTestActorRuntime::TEventObserverHolder Holder; + std::array SeenByChannel = {}; + std::bitset<256> BlockedChannels; + TDeque Blocked; + THashSet UnblockedOnce; +}; + template void ExecuteEvent(TDesiredPair &dp, TTestContext &tc) { DoWithRetry([&] { @@ -2971,5 +3130,434 @@ Y_UNIT_TEST(TestGetStatusNonExistentChannelReturnsErrorNewApi) { } +Y_UNIT_TEST(TestPerChannelReadLimitKeepsOtherChannelsAvailable) +{ + TTestContext tc; + TFinalizer finalizer(tc); + bool activeZone = false; + tc.Prepare(INITIAL_TEST_DISPATCH_NAME, [](TTestActorRuntime &){}, activeZone); + + auto &icb = tc.Runtime->GetAppData().Icb; + TControlWrapper readRequestInFlightLimit(5, 1, 4096); + TControlBoard::RegisterSharedControl(readRequestInFlightLimit, icb->KeyValueVolumeControls.ReadRequestsInFlightLimit); + readRequestInFlightLimit = 1; + TControlWrapper usePerChannelReadQueues(0, 0, 1); + TControlBoard::RegisterSharedControl(usePerChannelReadQueues, icb->KeyValueVolumeControls.UsePerChannelReadQueues); + usePerChannelReadQueues = 1; + + constexpr ui32 mainBlobChannel = NKeyValue::BLOB_CHANNEL; + constexpr ui32 extraBlobChannel = NKeyValue::BLOB_CHANNEL + 1; + + CmdWrite("main-hold", "value-main-hold", + NKikimrClient::TKeyValueRequest::MAIN, + NKikimrClient::TKeyValueRequest::REALTIME, tc); + CmdWrite("main-wait", "value-main-wait", + NKikimrClient::TKeyValueRequest::MAIN, + NKikimrClient::TKeyValueRequest::REALTIME, tc); + CmdWrite("extra-free", "value-extra-free", + NKikimrClient::TKeyValueRequest::EXTRA, + NKikimrClient::TKeyValueRequest::REALTIME, tc); + + TGetBlocker gets(*tc.Runtime); + gets.BlockChannel(mainBlobChannel); + + SendRequestEvent(MakeReadRequest(1, {"main-hold"}), tc); + tc.Runtime->WaitFor("blocked main read", [&] { + return gets.BlockedCount(mainBlobChannel) == 1; + }, TDuration::Seconds(1)); + UNIT_ASSERT_VALUES_EQUAL(gets.Seen(mainBlobChannel), 1); + + SendRequestEvent(MakeReadRequest(2, {"main-wait"}), tc); + tc.Runtime->DispatchEvents(TDispatchOptions(), TDuration::MilliSeconds(50)); + UNIT_ASSERT_VALUES_EQUAL(gets.Seen(mainBlobChannel), 1); + + SendRequestEvent(MakeReadRequest(3, {"extra-free"}), tc); + CheckReadResponse(ReceiveKeyValueResponse(tc), 3, {"value-extra-free"}); + UNIT_ASSERT_VALUES_EQUAL(gets.Seen(extraBlobChannel), 1); + + gets.UnblockOne(); + CheckReadResponse(ReceiveKeyValueResponse(tc), 1, {"value-main-hold"}); + + tc.Runtime->WaitFor("second main read starts after first one completes", [&] { + return gets.BlockedCount(mainBlobChannel) == 1 && gets.Seen(mainBlobChannel) == 2; + }, TDuration::Seconds(1)); + gets.UnblockOne(); + CheckReadResponse(ReceiveKeyValueResponse(tc), 2, {"value-main-wait"}); +} + +Y_UNIT_TEST(TestReadLimitUsesSingleQueueByDefault) +{ + TTestContext tc; + TFinalizer finalizer(tc); + bool activeZone = false; + tc.Prepare(INITIAL_TEST_DISPATCH_NAME, [](TTestActorRuntime &){}, activeZone); + + auto &icb = tc.Runtime->GetAppData().Icb; + TControlWrapper readRequestInFlightLimit(5, 1, 4096); + TControlBoard::RegisterSharedControl(readRequestInFlightLimit, icb->KeyValueVolumeControls.ReadRequestsInFlightLimit); + readRequestInFlightLimit = 1; + + constexpr ui32 mainBlobChannel = NKeyValue::BLOB_CHANNEL; + constexpr ui32 extraBlobChannel = NKeyValue::BLOB_CHANNEL + 1; + + CmdWrite("main-hold", "value-main-hold", + NKikimrClient::TKeyValueRequest::MAIN, + NKikimrClient::TKeyValueRequest::REALTIME, tc); + CmdWrite("extra-wait", "value-extra-wait", + NKikimrClient::TKeyValueRequest::EXTRA, + NKikimrClient::TKeyValueRequest::REALTIME, tc); + + TGetBlocker gets(*tc.Runtime); + gets.BlockChannel(mainBlobChannel); + + SendRequestEvent(MakeReadRequest(1, {"main-hold"}), tc); + tc.Runtime->WaitFor("blocked main read", [&] { + return gets.BlockedCount(mainBlobChannel) == 1; + }, TDuration::Seconds(1)); + UNIT_ASSERT_VALUES_EQUAL(gets.Seen(mainBlobChannel), 1); + + SendRequestEvent(MakeReadRequest(2, {"extra-wait"}), tc); + tc.Runtime->DispatchEvents(TDispatchOptions(), TDuration::MilliSeconds(50)); + UNIT_ASSERT_VALUES_EQUAL(gets.Seen(extraBlobChannel), 0); + + gets.UnblockOne(); + CheckReadResponse(ReceiveKeyValueResponse(tc), 1, {"value-main-hold"}); + + tc.Runtime->WaitFor("extra read starts after main queue is released", [&] { + return gets.Seen(extraBlobChannel) == 1; + }, TDuration::Seconds(1)); + CheckReadResponse(ReceiveKeyValueResponse(tc), 2, {"value-extra-wait"}); +} + +Y_UNIT_TEST(TestReadLimitDoesNotBlockGetStorageChannelStatus) +{ + TTestContext tc; + TFinalizer finalizer(tc); + bool activeZone = false; + tc.Prepare(INITIAL_TEST_DISPATCH_NAME, [](TTestActorRuntime &){}, activeZone); + + auto &icb = tc.Runtime->GetAppData().Icb; + TControlWrapper readRequestInFlightLimit(5, 1, 4096); + TControlBoard::RegisterSharedControl(readRequestInFlightLimit, icb->KeyValueVolumeControls.ReadRequestsInFlightLimit); + readRequestInFlightLimit = 1; + + constexpr ui32 mainBlobChannel = NKeyValue::BLOB_CHANNEL; + + CmdWrite("main-hold", "value-main-hold", + NKikimrClient::TKeyValueRequest::MAIN, + NKikimrClient::TKeyValueRequest::REALTIME, tc); + + TGetBlocker gets(*tc.Runtime); + gets.BlockChannel(mainBlobChannel); + + SendRequestEvent(MakeReadRequest(1, {"main-hold"}), tc); + tc.Runtime->WaitFor("blocked main read", [&] { + return gets.BlockedCount(mainBlobChannel) == 1; + }, TDuration::Seconds(1)); + UNIT_ASSERT_VALUES_EQUAL(gets.Seen(mainBlobChannel), 1); + + ExecuteGetStatus(tc, {1}, 0); + UNIT_ASSERT_VALUES_EQUAL(gets.Seen(mainBlobChannel), 1); + + gets.UnblockOne(); + CheckReadResponse(ReceiveKeyValueResponse(tc), 1, {"value-main-hold"}); +} + +Y_UNIT_TEST(TestPerChannelReadLimitDoesNotReserveGetStatusChannelInCompositeRequest) +{ + TTestContext tc; + TFinalizer finalizer(tc); + bool activeZone = false; + tc.Prepare(INITIAL_TEST_DISPATCH_NAME, [](TTestActorRuntime &){}, activeZone); + + auto &icb = tc.Runtime->GetAppData().Icb; + TControlWrapper readRequestInFlightLimit(5, 1, 4096); + TControlBoard::RegisterSharedControl(readRequestInFlightLimit, icb->KeyValueVolumeControls.ReadRequestsInFlightLimit); + readRequestInFlightLimit = 1; + TControlWrapper usePerChannelReadQueues(0, 0, 1); + TControlBoard::RegisterSharedControl(usePerChannelReadQueues, icb->KeyValueVolumeControls.UsePerChannelReadQueues); + usePerChannelReadQueues = 1; + + constexpr ui32 mainBlobChannel = NKeyValue::BLOB_CHANNEL; + constexpr ui32 extraBlobChannel = NKeyValue::BLOB_CHANNEL + 1; + + CmdWrite("main-hold", "value-main-hold", + NKikimrClient::TKeyValueRequest::MAIN, + NKikimrClient::TKeyValueRequest::REALTIME, tc); + CmdWrite("main-wait", "value-main-wait", + NKikimrClient::TKeyValueRequest::MAIN, + NKikimrClient::TKeyValueRequest::REALTIME, tc); + CmdWrite("extra-free", "value-extra-free", + NKikimrClient::TKeyValueRequest::EXTRA, + NKikimrClient::TKeyValueRequest::REALTIME, tc); + + TGetBlocker gets(*tc.Runtime); + gets.BlockChannel(mainBlobChannel); + + SendRequestEvent(MakeReadRequest(1, {"main-hold"}), tc); + tc.Runtime->WaitFor("blocked main read", [&] { + return gets.BlockedCount(mainBlobChannel) == 1; + }, TDuration::Seconds(1)); + UNIT_ASSERT_VALUES_EQUAL(gets.Seen(mainBlobChannel), 1); + + auto compositeRequest = MakeReadRequest(2, {"main-wait"}); + compositeRequest->Record.AddCmdGetStatus()->SetStorageChannel(NKikimrClient::TKeyValueRequest::EXTRA); + SendRequestEvent(std::move(compositeRequest), tc); + tc.Runtime->DispatchEvents(TDispatchOptions(), TDuration::MilliSeconds(50)); + UNIT_ASSERT_VALUES_EQUAL(gets.Seen(mainBlobChannel), 1); + UNIT_ASSERT_VALUES_EQUAL(gets.Seen(extraBlobChannel), 0); + + SendRequestEvent(MakeReadRequest(3, {"extra-free"}), tc); + CheckReadResponse(ReceiveKeyValueResponse(tc), 3, {"value-extra-free"}); + UNIT_ASSERT_VALUES_EQUAL(gets.Seen(extraBlobChannel), 1); + + gets.UnblockOne(); + CheckReadResponse(ReceiveKeyValueResponse(tc), 1, {"value-main-hold"}); + + tc.Runtime->WaitFor("composite read starts after main queue is released", [&] { + return gets.BlockedCount(mainBlobChannel) == 1 && gets.Seen(mainBlobChannel) == 2; + }, TDuration::Seconds(1)); + + gets.UnblockOne(); + const NKikimrClient::TResponse compositeResponse = ReceiveKeyValueResponse(tc); + CheckReadResponse(compositeResponse, 2, {"value-main-wait"}); + UNIT_ASSERT_VALUES_EQUAL(compositeResponse.GetStatusResultSize(), 1); + UNIT_ASSERT(compositeResponse.GetGetStatusResult(0).HasStatus()); + UNIT_ASSERT_EQUAL(compositeResponse.GetGetStatusResult(0).GetStatus(), NKikimrProto::OK); +} + +Y_UNIT_TEST(TestPerChannelReadLimitDoesNotBlockWritesAndInlineReads) +{ + TTestContext tc; + TFinalizer finalizer(tc); + bool activeZone = false; + tc.Prepare(INITIAL_TEST_DISPATCH_NAME, [](TTestActorRuntime &){}, activeZone); + + auto &icb = tc.Runtime->GetAppData().Icb; + TControlWrapper readRequestInFlightLimit(5, 1, 4096); + TControlBoard::RegisterSharedControl(readRequestInFlightLimit, icb->KeyValueVolumeControls.ReadRequestsInFlightLimit); + readRequestInFlightLimit = 1; + TControlWrapper usePerChannelReadQueues(0, 0, 1); + TControlBoard::RegisterSharedControl(usePerChannelReadQueues, icb->KeyValueVolumeControls.UsePerChannelReadQueues); + usePerChannelReadQueues = 1; + + constexpr ui32 mainBlobChannel = NKeyValue::BLOB_CHANNEL; + + CmdWrite("main-hold", "value-main-hold", + NKikimrClient::TKeyValueRequest::MAIN, + NKikimrClient::TKeyValueRequest::REALTIME, tc); + CmdWrite("inline-key", "value-inline", + NKikimrClient::TKeyValueRequest::INLINE, + NKikimrClient::TKeyValueRequest::REALTIME, tc); + + TGetBlocker gets(*tc.Runtime); + gets.BlockChannel(mainBlobChannel); + + SendRequestEvent(MakeReadRequest(1, {"main-hold"}), tc); + tc.Runtime->WaitFor("blocked main read", [&] { + return gets.BlockedCount(mainBlobChannel) == 1; + }, TDuration::Seconds(1)); + + SendRequestEvent(MakeWriteRequest(2, "main-write-while-read-blocked", "value-write", + NKikimrClient::TKeyValueRequest::MAIN), tc); + CheckWriteResponse(ReceiveKeyValueResponse(tc), 2); + + SendRequestEvent(MakeReadRequest(3, {"inline-key"}), tc); + CheckReadResponse(ReceiveKeyValueResponse(tc), 3, {"value-inline"}); + UNIT_ASSERT_VALUES_EQUAL(gets.Seen(mainBlobChannel), 1); + + gets.UnblockOne(); + CheckReadResponse(ReceiveKeyValueResponse(tc), 1, {"value-main-hold"}); +} + +Y_UNIT_TEST(TestPerChannelReadLimitMultiChannelReadWaitsForBlockedChannel) +{ + TTestContext tc; + TFinalizer finalizer(tc); + bool activeZone = false; + tc.Prepare(INITIAL_TEST_DISPATCH_NAME, [](TTestActorRuntime &){}, activeZone); + + auto &icb = tc.Runtime->GetAppData().Icb; + TControlWrapper readRequestInFlightLimit(5, 1, 4096); + TControlBoard::RegisterSharedControl(readRequestInFlightLimit, icb->KeyValueVolumeControls.ReadRequestsInFlightLimit); + readRequestInFlightLimit = 1; + TControlWrapper usePerChannelReadQueues(0, 0, 1); + TControlBoard::RegisterSharedControl(usePerChannelReadQueues, icb->KeyValueVolumeControls.UsePerChannelReadQueues); + usePerChannelReadQueues = 1; + + constexpr ui32 mainBlobChannel = NKeyValue::BLOB_CHANNEL; + constexpr ui32 extraBlobChannel = NKeyValue::BLOB_CHANNEL + 1; + + CmdWrite("main-hold", "value-main-hold", + NKikimrClient::TKeyValueRequest::MAIN, + NKikimrClient::TKeyValueRequest::REALTIME, tc); + CmdWrite("main-part", "value-main-part", + NKikimrClient::TKeyValueRequest::MAIN, + NKikimrClient::TKeyValueRequest::REALTIME, tc); + CmdWrite("extra-part", "value-extra-part", + NKikimrClient::TKeyValueRequest::EXTRA, + NKikimrClient::TKeyValueRequest::REALTIME, tc); + + TGetBlocker gets(*tc.Runtime); + gets.BlockChannel(mainBlobChannel); + + SendRequestEvent(MakeReadRequest(1, {"main-hold"}), tc); + tc.Runtime->WaitFor("blocked main read", [&] { + return gets.BlockedCount(mainBlobChannel) == 1; + }, TDuration::Seconds(1)); + UNIT_ASSERT_VALUES_EQUAL(gets.Seen(mainBlobChannel), 1); + UNIT_ASSERT_VALUES_EQUAL(gets.Seen(extraBlobChannel), 0); + + SendRequestEvent(MakeReadRequest(2, {"main-part", "extra-part"}), tc); + tc.Runtime->DispatchEvents(TDispatchOptions(), TDuration::MilliSeconds(50)); + UNIT_ASSERT_VALUES_EQUAL(gets.Seen(mainBlobChannel), 1); + UNIT_ASSERT_VALUES_EQUAL(gets.Seen(extraBlobChannel), 0); + + gets.UnblockOne(); + CheckReadResponse(ReceiveKeyValueResponse(tc), 1, {"value-main-hold"}); + + tc.Runtime->WaitFor("multi-channel read starts after blocked channel is released", [&] { + return gets.BlockedCount(mainBlobChannel) == 1 && gets.Seen(mainBlobChannel) == 2; + }, TDuration::Seconds(1)); + UNIT_ASSERT_VALUES_EQUAL(gets.Seen(extraBlobChannel), 1); + + gets.UnblockOne(); + CheckReadResponse(ReceiveKeyValueResponse(tc), 2, {"value-main-part", "value-extra-part"}); +} + +Y_UNIT_TEST(TestPerChannelReadLimitRandomizedNoDeadlock) +{ + TTestContext tc; + TFinalizer finalizer(tc); + bool activeZone = false; + tc.Prepare(INITIAL_TEST_DISPATCH_NAME, [](TTestActorRuntime &){}, activeZone); + + auto &icb = tc.Runtime->GetAppData().Icb; + TControlWrapper readRequestInFlightLimit(5, 1, 4096); + TControlBoard::RegisterSharedControl(readRequestInFlightLimit, icb->KeyValueVolumeControls.ReadRequestsInFlightLimit); + readRequestInFlightLimit = 1; + TControlWrapper usePerChannelReadQueues(0, 0, 1); + TControlBoard::RegisterSharedControl(usePerChannelReadQueues, icb->KeyValueVolumeControls.UsePerChannelReadQueues); + usePerChannelReadQueues = 1; + + struct TChannelKey { + TString Key; + TString Value; + NKikimrClient::TKeyValueRequest::EStorageChannel StorageChannel; + ui32 BlobChannel; + }; + + const TVector keys = { + {"random-main", "value-random-main", NKikimrClient::TKeyValueRequest::MAIN, NKeyValue::BLOB_CHANNEL}, + {"random-extra", "value-random-extra", NKikimrClient::TKeyValueRequest::EXTRA, NKeyValue::BLOB_CHANNEL + 1}, + {"random-extra2", "value-random-extra2", NKikimrClient::TKeyValueRequest::EXTRA2, NKeyValue::BLOB_CHANNEL + 2}, + }; + + for (const TChannelKey& key : keys) { + CmdWrite(key.Key, key.Value, key.StorageChannel, NKikimrClient::TKeyValueRequest::REALTIME, tc); + } + + TGetBlocker gets(*tc.Runtime); + for (const TChannelKey& key : keys) { + gets.BlockChannel(key.BlobChannel); + } + + THashMap> expectedValuesByCookie; + TReallyFastRng32 rng(17); + constexpr ui32 requestCount = 36; + auto makeRandomRequest = [&](ui32 i) { + std::bitset<8> usedKeys; + const ui32 keyCount = 1 + rng.Uniform(3); + TVector requestKeys; + TVector expectedValues; + while (requestKeys.size() < keyCount) { + const ui32 keyIdx = rng.Uniform(keys.size()); + if (!usedKeys.test(keyIdx)) { + usedKeys.set(keyIdx); + requestKeys.push_back(keys[keyIdx].Key); + expectedValues.push_back(keys[keyIdx].Value); + } + } + + const ui64 cookie = 1000 + i; + expectedValuesByCookie.emplace(cookie, expectedValues); + return MakeReadRequest(cookie, requestKeys); + }; + + THashSet seenCookies; + auto handleResponse = [&](const NKikimrClient::TResponse& response) { + UNIT_ASSERT(response.HasCookie()); + const ui64 cookie = response.GetCookie(); + UNIT_ASSERT_C(seenCookies.insert(cookie).second, "duplicate cookie# " << cookie); + auto it = expectedValuesByCookie.find(cookie); + UNIT_ASSERT_C(it != expectedValuesByCookie.end(), "unexpected cookie# " << cookie); + CheckReadResponse(response, cookie, it->second); + }; + + auto waitDescription = [&](TStringBuf action, ui32 targetGets, ui32 targetResponses) { + return TStringBuilder() << action + << " received# " << seenCookies.size() << "/" << targetResponses + << " blockedGets# " << gets.BlockedSize() + << " totalGets# " << gets.TotalSeen() << "/" << targetGets; + }; + + constexpr ui32 batchSize = 6; + const TDuration progressTimeout = TDuration::Seconds(5); + for (ui32 batchBegin = 0; batchBegin < requestCount; batchBegin += batchSize) { + const ui32 batchEnd = Min(requestCount, batchBegin + batchSize); + const ui32 targetResponses = seenCookies.size() + (batchEnd - batchBegin); + const ui32 initialSeenGets = gets.TotalSeen(); + ui32 batchGets = 0; + + SendRequestEvent(makeRandomRequest(batchBegin), tc); + batchGets += expectedValuesByCookie[1000 + batchBegin].size(); + + tc.Runtime->WaitFor(waitDescription("initial randomized read batch", initialSeenGets + batchGets, targetResponses), [&] { + return gets.BlockedSize() > 0; + }, progressTimeout); + + for (ui32 i = batchBegin + 1; i < batchEnd; ++i) { + SendRequestEvent(makeRandomRequest(i), tc); + batchGets += expectedValuesByCookie[1000 + i].size(); + } + tc.Runtime->DispatchEvents(TDispatchOptions(), TDuration::MilliSeconds(50)); + + const ui32 targetGets = initialSeenGets + batchGets; + for (ui32 iteration = 0; (gets.TotalSeen() < targetGets || gets.BlockedSize() > 0) && iteration < batchGets * 2; ++iteration) { + if (gets.BlockedSize() == 0) { + tc.Runtime->WaitFor(waitDescription("more randomized blob reads", targetGets, targetResponses), [&] { + return gets.BlockedSize() > 0; + }, progressTimeout); + } + + if (gets.BlockedSize() > 0) { + gets.UnblockAll(); + tc.Runtime->DispatchEvents(TDispatchOptions(), TDuration::MilliSeconds(50)); + } + } + + UNIT_ASSERT_VALUES_EQUAL_C(gets.TotalSeen(), targetGets, + waitDescription("all randomized blob reads must start", targetGets, targetResponses)); + UNIT_ASSERT_VALUES_EQUAL_C(gets.BlockedSize(), 0, + waitDescription("all randomized blob reads must be unblocked", targetGets, targetResponses)); + + while (seenCookies.size() < targetResponses) { + TAutoPtr handle; + TEvKeyValue::TEvResponse *response = tc.Runtime->GrabEdgeEvent( + handle, progressTimeout); + UNIT_ASSERT_C(response, waitDescription("randomized response timeout", targetGets, targetResponses)); + handleResponse(response->Record); + } + + UNIT_ASSERT_VALUES_EQUAL_C(seenCookies.size(), targetResponses, + "received# " << seenCookies.size() << " expected# " << targetResponses + << " blockedGets# " << gets.BlockedSize() << " totalGets# " << gets.TotalSeen()); + } + + UNIT_ASSERT_VALUES_EQUAL_C(seenCookies.size(), requestCount, + "received# " << seenCookies.size() << " expected# " << requestCount + << " blockedGets# " << gets.BlockedSize() << " totalGets# " << gets.TotalSeen()); +} + } // TKeyValueTest } // NKikimr diff --git a/ydb/core/protos/config.proto b/ydb/core/protos/config.proto index ae7f495c554..7b68d7a174f 100644 --- a/ydb/core/protos/config.proto +++ b/ydb/core/protos/config.proto @@ -2158,6 +2158,12 @@ message TImmediateControlsConfig { MaxValue: 1, DefaultValue: 0 }]; + optional uint64 UsePerChannelReadQueues = 5 [(ControlOptions) = { + Description: "Use separate KeyValue read request queues per storage channel", + MinValue: 0, + MaxValue: 1, + DefaultValue: 0 + }]; } message TKQPSessionControls { -- cgit v1.3