One-partition throughput on default delivery settings

commit_hash:830f706fc9e7b9c334b1c66737d1b78afc78b9aa

Author: yuri-pechatnov

yt/yt/flow/library/cpp/common/spec.cpp

@@ -985,9 +985,9 @@ void TDynamicBufferStateManagerSpec::Register(TRegistrar registrar)
 void TDynamicMessageDistributorSpec::Register(TRegistrar registrar)
 {
     registrar.Parameter("send_queue_max_rows_per_batch", &TThis::SendQueueMaxRowsPerBatch)
-        .Default(NYTree::TSize(1000));
+        .Default(NYTree::TSize(10'000));
     registrar.Parameter("send_queue_max_bytes_per_batch", &TThis::SendQueueMaxBytesPerBatch)
-        .Default(NYTree::TSize(1_MB));
+        .Default(NYTree::TSize(10_MB));
     registrar.Parameter("send_queue_batch_duration", &TThis::SendQueueBatchDuration)
         .Default(TDuration::MilliSeconds(100));
 

yt/yt/flow/library/cpp/worker/input_buffer_detail.cpp

@@ -325,9 +325,15 @@ void TInputBuffer::DoAddConnectionOffer(TGuid connectionId, TConnectionOffer off
             connectionState.UpdateEpoch = streamState.Epoch;
         }
 
-        // Just double function amortized complexity.
-        if (--streamState.RecalculateCounter <= 0) {
-            RecalculateStreamLimits(streamState);
+        // Just double function amortized complexity. A manager-side limit change bypasses the
+        // amortization so the raised limit turns into grants on the first offer, not the Nth;
+        // the bypass must not advance the connection-GC epoch: the recomputed limit jitters on
+        // every manage tick, and the GC horizon must stay measured in offer rounds.
+        const bool offerRoundElapsed = --streamState.RecalculateCounter <= 0;
+        if (offerRoundElapsed ||
+            streamState.LimitUsageState->GetLimitBytes() != streamState.LastRecalculatedLimitBytes)
+        {
+            RecalculateStreamLimits(streamState, /*collectStaleConnections*/ offerRoundElapsed);
             streamState.Usage.PendingInflatedBytes = GetPendingSize(streamState);
             streamState.LimitUsageState->Update(streamState.Usage);
         }
@@ -445,6 +451,7 @@ TFuture<std::vector<TInputMessageConstPtr>> TInputBuffer::DoGetInputBatch(THashS
 
     using TPriority = std::pair<TSystemTimestamp, ui64>;
     std::vector<std::pair<TMessagesPriorityQueue*, std::function<TPriority()>>> queues;
+    std::vector<TStreamState*> extractionStreamStates;
     size_t queuedMessageCount = 0;
     for (auto& [streamId, streamState] : StreamStates_) {
         if (!streamState.Messages.empty() && allowedStreams.contains(streamId)) {
@@ -454,6 +461,7 @@ TFuture<std::vector<TInputMessageConstPtr>> TInputBuffer::DoGetInputBatch(THashS
                 const auto& front = messagesPtr->front();
                 return {TSystemTimestamp(front.AlignmentTimestamp.Underlying() + bias), front.SeqNo};
             });
+            extractionStreamStates.push_back(&streamState);
             queuedMessageCount += streamState.Messages.size();
         }
     }
@@ -480,22 +488,31 @@ TFuture<std::vector<TInputMessageConstPtr>> TInputBuffer::DoGetInputBatch(THashS
         LastNotFullBatchInstant_ = TInstant::Now();
     }
 
+    // Extraction freed buffer space; regrant connection windows right away so the next
+    // PushMessages response carries fresh limits instead of waiting out the offer amortization.
+    if (!batch.empty()) {
+        for (auto* streamState : extractionStreamStates) {
+            RecalculateStreamLimits(*streamState, /*collectStaleConnections*/ false);
+            streamState->Usage.PendingInflatedBytes = GetPendingSize(*streamState);
+        }
+    }
+
     for (auto& [streamId, streamState] : StreamStates_) {
         streamState.LimitUsageState->Update(streamState.Usage);
     }
 
     return MakeFuture<std::vector<TInputMessageConstPtr>>(std::move(batch));
 }
 
-void TInputBuffer::RecalculateStreamLimits(TStreamState& streamState)
+void TInputBuffer::RecalculateStreamLimits(TStreamState& streamState, bool collectStaleConnections)
 {
     using THeapElement = std::pair<TConnectionState*, i64>; // (it, currentBucketIndex).
 
     constexpr i64 lastEpochStoreCount = 5; // TODO: something better? Configurable?
 
     // Cleanup.
     for (auto it = streamState.ConnectionStates.begin(); it != streamState.ConnectionStates.end();) {
-        if (it->second.UpdateEpoch + lastEpochStoreCount < streamState.Epoch) {
+        if (collectStaleConnections && it->second.UpdateEpoch + lastEpochStoreCount < streamState.Epoch) {
             streamState.ConnectionStates.erase(it++);
         } else {
             it->second.InflatedByteLimit = 0;
@@ -520,6 +537,7 @@ void TInputBuffer::RecalculateStreamLimits(TStreamState& streamState)
     // Match the manager's accounting: limit is inflated, so used must be inflated too.
     i64 inflatedUsedBytes = streamState.Usage.GetInflatedInflightBytes(streamState.LimitUsageState->GetInflationPerMessage());
     i64 inflatedLimitBytes = streamState.LimitUsageState->GetLimitBytes();
+    streamState.LastRecalculatedLimitBytes = inflatedLimitBytes;
     // Do not allocate small share of buffer to reduce retransmits.
     i64 inflatedFreeBytes = std::max<i64>(inflatedLimitBytes * 0.9 - inflatedUsedBytes, 0);
 
@@ -547,8 +565,10 @@ void TInputBuffer::RecalculateStreamLimits(TStreamState& streamState)
         }
     }
 
-    streamState.Epoch += 1;
-    streamState.RecalculateCounter = std::ssize(streamState.ConnectionStates);
+    if (collectStaleConnections) {
+        streamState.Epoch += 1;
+        streamState.RecalculateCounter = std::ssize(streamState.ConnectionStates);
+    }
 }
 
 i64 TInputBuffer::GetPendingSize(const TStreamState& streamState)

yt/yt/flow/library/cpp/worker/input_buffer_detail.h

@@ -76,7 +76,8 @@ class TInputBuffer
 
         TConnectionStates ConnectionStates;
         i64 RecalculateCounter = 0;
-        i64 Epoch = 0; // How many recalculations (~ updates of all connections) were done.
+        i64 Epoch = 0; // How many offer-driven recalculations (~ updates of all connections) were done.
+        i64 LastRecalculatedLimitBytes = -1;
 
         TMessagesPriorityQueue Messages;
 
@@ -119,7 +120,10 @@ class TInputBuffer
 
     TSystemTimestamp GetMinStabilizedEventTimestamp() override;
 
-    static void RecalculateStreamLimits(TStreamState& streamState);
+    // |collectStaleConnections| must be set only on the offer-driven cadence: connection GC is
+    // measured in recalculation epochs, so high-frequency extraction regrants would otherwise
+    // erase live connections between their offers.
+    static void RecalculateStreamLimits(TStreamState& streamState, bool collectStaleConnections = true);
     static i64 GetPendingSize(const TStreamState& streamState);
 
     double ComputeStreamBias(TStreamId streamId, const TInputMessageConstPtr& frontMessage) const;

yt/yt/flow/library/cpp/worker/unittests/input_buffer_ut.cpp

@@ -167,6 +167,40 @@ TEST(TInputBufferRecalculateStreamLimitsTest, Cleanup)
     EXPECT_EQ(0, streamState.ConnectionStates.at(connectionId1).InflatedByteLimit);
 }
 
+TEST(TInputBufferRecalculateStreamLimitsTest, RegrantKeepsStaleConnections)
+{
+    auto staleConnectionId = TGuid::Create();
+    auto freshConnectionId = TGuid::Create();
+    TStreamState streamState = {
+        .ConnectionStates = {
+            {
+                staleConnectionId,
+                TConnectionState{
+                    .UpdateEpoch = 0,
+                    .Offer = {{TSystemTimestamp(12), 100}, {TSystemTimestamp(5), 200}},
+                    .InflatedByteLimit = 100500,
+                },
+            },
+            {
+                freshConnectionId,
+                TConnectionState{
+                    .UpdateEpoch = 100,
+                    .Offer = {{TSystemTimestamp(10), 800}, {TSystemTimestamp(9), 400}},
+                },
+            },
+        },
+        .RecalculateCounter = 42,
+        .Epoch = 101,
+        .LimitUsageState = MakeLimitState(1000),
+    };
+    TInputBuffer::RecalculateStreamLimits(streamState, /*collectStaleConnections*/ false);
+    ASSERT_TRUE(streamState.ConnectionStates.contains(staleConnectionId));
+    EXPECT_EQ(200, streamState.ConnectionStates.at(staleConnectionId).InflatedByteLimit);
+    EXPECT_EQ(400, streamState.ConnectionStates.at(freshConnectionId).InflatedByteLimit);
+    EXPECT_EQ(101, streamState.Epoch);
+    EXPECT_EQ(42, streamState.RecalculateCounter);
+}
+
 TEST(TInputBufferGetPendingSizeTest, Simple)
 {
     auto connectionId1 = TGuid::Create();