Separate read queues for channels in keyvalue tablet (#45783)

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<ui32> AcquiredChannels;
         std::deque<std::pair<TLogoBlobID, bool>> RefCountsIncr;
 
         TEvNotify() { }
 
         TEvNotify(ui64 requestUid, ui64 generation, ui64 step, const NKeyValue::TRequestStat &stat,
-                NMsgBusProxy::EResponseStatus status, std::deque<std::pair<TLogoBlobID, bool>>&& refCountsIncr)
+                NMsgBusProxy::EResponseStatus status, TVector<ui32> acquiredChannels,
+                std::deque<std::pair<TLogoBlobID, bool>>&& 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<std::pair<TLogoBlobID, bool>>&& refCountsIncr)
+                NKikimrKeyValue::Statuses::ReplyStatus status, TVector<ui32> acquiredChannels,
+                std::deque<std::pair<TLogoBlobID, bool>>&& 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<ui32> 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<TIntermediate> &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<ui32> 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<decltype(command)>;
+            if constexpr (std::is_same_v<TCommand, TIntermediate::TRead>) {
+                addRead(command);
+            } else if constexpr (std::is_same_v<TCommand, TIntermediate::TRangeRead>) {
+                addRangeRead(command);
+            }
+        }, *intermediate.ReadCommand);
+    }
+    for (const TIntermediate::TCmd& command : intermediate.Commands) {
+        std::visit([&](const auto& cmd) {
+            using TCommand = std::decay_t<decltype(cmd)>;
+            if constexpr (std::is_same_v<TCommand, TIntermediate::TPatch>) {
+                addPatch(cmd);
+            }
+        }, command);
+    }
+    if (intermediate.TrimLeakedBlobs) {
+        for (const auto& [channel, _] : intermediate.TrimLeakedBlobs->ChannelGroupMap) {
+            addAcquiredChannel(channel);
+        }
+    }
+
+    TVector<ui32> channels;
+    for (size_t channel = 2; channel < PostponedChannels.size(); ++channel) {
+        if (acquiredChannels.test(channel)) {
+            channels.push_back(static_cast<ui32>(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<TIntermediate> &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<ui32> &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<TIntermediate> 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<ui32> &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<ui32> &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<ui32> 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<TEvKeyValue::TEvRead>(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<TEvKeyValue::TEvReadRange>(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<TEvKeyValue::TEvRequest>(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<TAutoPtr<TIntermediate>> 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<TIntermediate*> Queue;
+        ui64 IntermediatesInFlight = 0;
+    };
+
+    TVector<TPostponedChannel> 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<TKeyValueStateLifetimeToken> LifetimeToken = std::make_shared<TKeyValueStateLifetimeToken>();
 
@@ -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<ui32> &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<TIntermediate> &intermediate);
 
-    template <typename TRequestType>
-    void PostponeIntermediate(THolder<TIntermediate> &&intermediate) {
-        intermediate->Stat.EnqueuedAs = Queue.size() + 1;
-        Queue.push_back(std::move(intermediate));
-    }
+    TVector<ui32> GetAcquiredChannels(const TIntermediate &intermediate) const;
+    bool TryStartOrPostponeIntermediate(THolder<TIntermediate> &intermediate, const TActorContext &ctx);
+    void StartChannelLimitedIntermediate(const TIntermediate &intermediate);
+    void ReleaseChannelLimitedIntermediate(const TVector<ui32> &acquiredChannels);
+    void ProcessPostponedChannel(ui32 channel, const TActorContext &ctx, const TTabletStorageInfo *info);
+    void ProcessPostponedChannels(const TVector<ui32> &channels, const TActorContext &ctx,
+        const TTabletStorageInfo *info);
     void ProcessPostponedIntermediate(const TActorContext& ctx, THolder<TIntermediate> &&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<TEvKeyValue::TEvReadResponse> 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 <library/cpp/testing/unittest/registar.h>
 #include <util/random/fast.h>
 #include <ydb/core/base/blobstorage.h>
+#include <array>
+#include <bitset>
 
 const bool ENABLE_DETAILED_KV_LOG = false;
 const bool ENABLE_TESTLOG_OUTPUT = false;
@@ -692,6 +694,163 @@ auto ReceiveResponse(TTestContext &tc) -> decltype(std::declval<TResponseEvent>(
     return response->Record;
 }
 
+std::unique_ptr<TEvKeyValue::TEvRequest> MakeReadRequest(ui64 cookie, const TVector<TString> &keys) {
+    auto request = std::make_unique<TEvKeyValue::TEvRequest>();
+    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<TEvKeyValue::TEvRequest> MakeWriteRequest(ui64 cookie, const TString &key, const TString &value,
+        NKikimrClient::TKeyValueRequest::EStorageChannel storageChannel) {
+    auto request = std::make_unique<TEvKeyValue::TEvRequest>();
+    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<TEvKeyValue::TEvResponse>(tc);
+}
+
+void CheckReadResponse(const NKikimrClient::TResponse &response, ui64 cookie, const TVector<TString> &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<ui32>(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<ui32>(NKikimrProto::OK));
+}
+
+class TGetBlocker {
+public:
+    explicit TGetBlocker(TTestActorRuntime& runtime)
+        : Runtime(runtime)
+        , Holder(Runtime.AddObserver<TEvBlobStorage::TEvGet>(
+            [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<ui32, 256> SeenByChannel = {};
+    std::bitset<256> BlockedChannels;
+    TDeque<TEvBlobStorage::TEvGet::TPtr> Blocked;
+    THashSet<IEventHandle*> UnblockedOnce;
+};
+
 template <typename TRequestEvent>
 void ExecuteEvent(TDesiredPair<TRequestEvent> &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<TChannelKey> 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<ui64, TVector<TString>> expectedValuesByCookie;
+    TReallyFastRng32 rng(17);
+    constexpr ui32 requestCount = 36;
+    auto makeRandomRequest = [&](ui32 i) {
+        std::bitset<8> usedKeys;
+        const ui32 keyCount = 1 + rng.Uniform(3);
+        TVector<TString> requestKeys;
+        TVector<TString> 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<ui64> 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<ui32>(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<IEventHandle> handle;
+            TEvKeyValue::TEvResponse *response = tc.Runtime->GrabEdgeEvent<TEvKeyValue::TEvResponse>(
+                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 {