From 1960d73474845f447b04327fa420bb7d0ae3a0b4 Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Thu, 15 Oct 2026 18:31:56 +0800 Subject: [PATCH 1/2] [Pipe] Fix busy-wait in pipe rate limiter when a request exceeds the burst capacity PipeEndPointRateLimiter and GlobalRateLimiter acquired a whole request with RateLimiter.tryAcquire(permits, timeout) and retried forever on failure. Since RateLimiter.create(rate) builds a SmoothBursty limiter whose burst capacity is only one second's worth of permits, a request costing more than the timeout (permits / rate > timeout) can never be granted by a timeout-bounded attempt: once the initial burst is consumed, every attempt fails instantly and the retry loop spins at ~100% CPU without releasing any logs. The default configuration hits this as soon as the rate limit is below the object batch size: the batch is at most PipeSinkReadFileBufferSize (5 MB) and the timeout is RateLimiterHotReloadCheckIntervalMs (1 s), so any rate-limit-bytes-per-second below 5 MB/s makes the sink spin instead of transferring at the configured pace. Because the loop logged nothing, the pipe kept reporting RUNNING with an empty ExceptionMessage and a plausible EstimatedRemainingSeconds, which made the stall look like a slow sync. RateLimitUtils now splits a request into pieces that the burst capacity can actually grant and gives each piece a timeout covering its own cost, falling back to a blocking acquisition if the rate is reloaded downwards while waiting, so throughput still converges to the configured rate. The Integer.MAX_VALUE fragmentation branches are folded into the same logic, which also fixes the int overflow when rate limiting is disabled via Double.MAX_VALUE. A warning is emitted once per request when a request has to be split, so such a configuration is no longer silent. --- .../iotdb/commons/i18n/PipeMessages.java | 2 + .../iotdb/commons/i18n/PipeMessages.java | 2 + .../pipe/sink/limiter/GlobalRateLimiter.java | 25 +-- .../sink/limiter/PipeEndPointRateLimiter.java | 32 +-- .../pipe/sink/limiter/RateLimitUtils.java | 187 ++++++++++++++++++ .../pipe/sink/limiter/RateLimitUtilsTest.java | 150 ++++++++++++++ 6 files changed, 353 insertions(+), 45 deletions(-) create mode 100644 iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/limiter/RateLimitUtils.java create mode 100644 iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/sink/limiter/RateLimitUtilsTest.java diff --git a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java index 0aafc35842770..0692900470b61 100644 --- a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java @@ -965,5 +965,7 @@ private PipeMessages() {} public static final String MESSAGE_DATAPARTITIONTABLE_GENERATION_COMPLETED_SUCCESSFULLY_E076E3B2 = "DataPartitionTable generation completed successfully"; public static final String MESSAGE_DATAPARTITIONTABLE_GENERATION_FAILED_D85CD23A = "DataPartitionTable generation failed: "; public static final String MESSAGE_UNKNOWN_TASK_STATUS_E05D98F0 = "Unknown task status: "; + public static final String LOG_PIPE_RATE_LIMIT_IS_SMALLER_THAN_THE_SINGLE_REQUEST_SIZE_SO_ARG_BYTES_WILL_BE_TRANSFERRED_IN_ARG_SIZED_PIECES_AT_ARG_BYTES_PER_SECOND_CONSIDER_SETTING_A_LARGER_RATE_LIMIT_FOR_BETTER_PERFORMANCE_4450D4DC = + "Pipe rate limit is smaller than the single request size, so %s bytes will be transferred in %s-sized pieces at %s bytes per second. Consider setting a larger rate limit for better performance."; } diff --git a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java index 9746b71500d2d..5c7d269c08a5c 100644 --- a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java @@ -931,5 +931,7 @@ private PipeMessages() {} public static final String MESSAGE_DATAPARTITIONTABLE_GENERATION_COMPLETED_SUCCESSFULLY_E076E3B2 = "DataPartitionTable 生成已成功完成"; public static final String MESSAGE_DATAPARTITIONTABLE_GENERATION_FAILED_D85CD23A = "DataPartitionTable 生成失败："; public static final String MESSAGE_UNKNOWN_TASK_STATUS_E05D98F0 = "未知任务状态："; + public static final String LOG_PIPE_RATE_LIMIT_IS_SMALLER_THAN_THE_SINGLE_REQUEST_SIZE_SO_ARG_BYTES_WILL_BE_TRANSFERRED_IN_ARG_SIZED_PIECES_AT_ARG_BYTES_PER_SECOND_CONSIDER_SETTING_A_LARGER_RATE_LIMIT_FOR_BETTER_PERFORMANCE_4450D4DC = + "Pipe 限速值小于单次请求大小，因此 %s 字节将按 %s 大小的分片以 %s 字节每秒的速度传输。建议设置更大的限速值以获得更好的性能。"; } diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/limiter/GlobalRateLimiter.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/limiter/GlobalRateLimiter.java index 358e6c844ec90..35e07aba23d2e 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/limiter/GlobalRateLimiter.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/limiter/GlobalRateLimiter.java @@ -24,8 +24,6 @@ import com.google.common.util.concurrent.AtomicDouble; import com.google.common.util.concurrent.RateLimiter; -import java.util.concurrent.TimeUnit; - public abstract class GlobalRateLimiter { private final AtomicDouble throughputBytesPerSecond = @@ -47,26 +45,11 @@ public void acquire(long bytes) { return; } - while (bytes > 0) { - if (bytes > Integer.MAX_VALUE) { - tryAcquireWithRateCheck(Integer.MAX_VALUE); - bytes -= Integer.MAX_VALUE; - } else { - tryAcquireWithRateCheck((int) bytes); - return; - } - } - } - - private void tryAcquireWithRateCheck(final int bytes) { - while (!rateLimiter.tryAcquire( + RateLimitUtils.acquireWithRateLimit( + rateLimiter, bytes, - PipeConfig.getInstance().getRateLimiterHotReloadCheckIntervalMs(), - TimeUnit.MILLISECONDS)) { - if (reloadParams()) { - return; - } - } + () -> !reloadParams(), + PipeConfig.getInstance().getRateLimiterHotReloadCheckIntervalMs()); } private boolean reloadParams() { diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/limiter/PipeEndPointRateLimiter.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/limiter/PipeEndPointRateLimiter.java index 2701e5bb23702..95a586030d360 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/limiter/PipeEndPointRateLimiter.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/limiter/PipeEndPointRateLimiter.java @@ -28,7 +28,6 @@ import java.util.Objects; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.TimeUnit; public class PipeEndPointRateLimiter { @@ -64,30 +63,15 @@ public void acquire(final TEndPoint endPoint, long bytes) { endPointRateLimiterMap.computeIfAbsent( endPoint, e -> RateLimiter.create(bytesPerSecondLimit)); - while (bytes > 0) { - if (bytes > Integer.MAX_VALUE) { - if (!tryAcquireWithPipeCheck(rateLimiter, Integer.MAX_VALUE)) { - return; - } - bytes -= Integer.MAX_VALUE; - } else { - tryAcquireWithPipeCheck(rateLimiter, (int) bytes); - return; - } - } + RateLimitUtils.acquireWithRateLimit( + rateLimiter, + bytes, + () -> isPipeExisted(), + PipeConfig.getInstance().getRateLimiterHotReloadCheckIntervalMs()); } - private boolean tryAcquireWithPipeCheck(final RateLimiter rateLimiter, final int bytes) { - while (!rateLimiter.tryAcquire( - bytes, - PipeConfig.getInstance().getRateLimiterHotReloadCheckIntervalMs(), - TimeUnit.MILLISECONDS)) { - final PipeTaskAgent finalTaskAgent = taskAgent; - if (Objects.nonNull(finalTaskAgent) - && !finalTaskAgent.isPipeExisted(pipeName, creationTime)) { - return false; - } - } - return true; + private boolean isPipeExisted() { + final PipeTaskAgent finalTaskAgent = taskAgent; + return Objects.isNull(finalTaskAgent) || finalTaskAgent.isPipeExisted(pipeName, creationTime); } } diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/limiter/RateLimitUtils.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/limiter/RateLimitUtils.java new file mode 100644 index 0000000000000..a3fa8363119b1 --- /dev/null +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/limiter/RateLimitUtils.java @@ -0,0 +1,187 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.commons.pipe.sink.limiter; + +import org.apache.iotdb.commons.i18n.PipeMessages; + +import com.google.common.util.concurrent.RateLimiter; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.concurrent.TimeUnit; + +/** + * Bridges a byte-based request to {@link RateLimiter}, which is a permit-based limiter. + * + *

{@link RateLimiter#create(double)} builds a {@code SmoothBursty} limiter whose burst capacity + * is at most one second's worth of permits, and {@link RateLimiter#tryAcquire(int, long, TimeUnit)} + * only grants a request when its whole cost fits inside the given timeout. When a single request + * costs more than the timeout, every attempt fails instantly and the surrounding retry loop + * degenerates into a 100% CPU busy-wait that never logs anything. Such a request must therefore + * never be passed to {@code tryAcquire} in one piece, which is what this class takes care of. + */ +public class RateLimitUtils { + + private static final Logger LOGGER = LoggerFactory.getLogger(RateLimitUtils.class); + + /** The default value of {@code RateLimiterHotReloadCheckIntervalMs}. */ + public static final long DEFAULT_CHECK_INTERVAL_MS = 1000; + + // The loop below is a busy-wait guard as well: a healthy chunk needs at most one timed attempt + // before it falls back to a blocking acquisition, so exceeding this budget means something is + // wrong (e.g. the limiter is stuck) rather than that we merely have to wait longer. + private static final int MAX_ATTEMPTS_PER_CHUNK = 20; + + private RateLimitUtils() { + // Utility class + } + + /** Decides whether to keep waiting for the rate limiter. */ + @FunctionalInterface + public interface ContinueChecker { + /** + * @return {@code true} to keep acquiring, {@code false} to abort the whole acquisition + */ + boolean shouldContinue(); + } + + /** + * Acquires {@code bytes} permits from {@code rateLimiter}, checking the limiter every {@link + * #DEFAULT_CHECK_INTERVAL_MS}. + * + * @param rateLimiter the permit-based limiter to acquire from + * @param bytes the number of bytes to acquire, non-positive values return immediately + * @param continueChecker consulted between pieces to stop waiting for a dead pipe + * @return {@code true} if the whole request was acquired, {@code false} if {@code + * continueChecker} asked to abort + */ + public static boolean acquireWithRateLimit( + final RateLimiter rateLimiter, final long bytes, final ContinueChecker continueChecker) { + return acquireWithRateLimit(rateLimiter, bytes, continueChecker, DEFAULT_CHECK_INTERVAL_MS); + } + + /** + * Acquires {@code bytes} permits from {@code rateLimiter}. + * + *

The request is split into pieces that the burst capacity can actually grant, each piece is + * acquired with a timeout that covers its own cost, and a piece that keeps timing out (which can + * only happen when the limiter's rate is reloaded downwards) falls back to a blocking acquisition + * so that the caller stops burning CPU. The average throughput always converges to the configured + * rate. + * + * @param rateLimiter the permit-based limiter to acquire from + * @param bytes the number of bytes to acquire, non-positive values return immediately + * @param continueChecker consulted between pieces and on a timeout, to stop waiting for a dead + * pipe + * @param checkIntervalMs how often {@code continueChecker} is consulted while waiting + * @return {@code true} if the whole request was acquired, {@code false} if {@code + * continueChecker} asked to abort + */ + public static boolean acquireWithRateLimit( + final RateLimiter rateLimiter, + final long bytes, + final ContinueChecker continueChecker, + final long checkIntervalMs) { + final long boundedCheckIntervalMs = Math.max(1L, checkIntervalMs); + + long remaining = bytes; + boolean logPipeSizeMismatch = true; + + while (remaining > 0) { + // A request that fits into a single piece is served by the limiter's initial burst without + // waiting, so the checker only matters for requests that need several pieces, i.e. the ones + // that actually take time to transfer. + if (remaining < bytes && !continueChecker.shouldContinue()) { + return false; + } + + final long piece = calculatePieceSize(rateLimiter, remaining); + if (logPipeSizeMismatch && piece < remaining) { + logPipeSizeMismatch = false; + LOGGER.warn( + PipeMessages + .LOG_PIPE_RATE_LIMIT_IS_SMALLER_THAN_THE_SINGLE_REQUEST_SIZE_SO_ARG_BYTES_WILL_BE_TRANSFERRED_IN_ARG_SIZED_PIECES_AT_ARG_BYTES_PER_SECOND_CONSIDER_SETTING_A_LARGER_RATE_LIMIT_FOR_BETTER_PERFORMANCE_4450D4DC, + bytes, + piece, + rateLimiter.getRate()); + } + + final long timeoutMs = calculateTimeoutMs(rateLimiter, piece, boundedCheckIntervalMs); + int attempts = 0; + boolean acquired = false; + while (!acquired) { + acquired = rateLimiter.tryAcquire((int) piece, timeoutMs, TimeUnit.MILLISECONDS); + if (acquired) { + break; + } + + if (!continueChecker.shouldContinue()) { + return false; + } + + if (++attempts >= MAX_ATTEMPTS_PER_CHUNK) { + // The rate has most likely been reloaded downwards while we were waiting, so our + // estimate is far too optimistic. Wait for the permits without a deadline instead of + // retrying with a deadline that we now know we cannot meet, which would only spin. + rateLimiter.acquire((int) piece); + acquired = true; + } + } + + remaining -= piece; + } + + return true; + } + + /** + * @return a piece size that is small enough for the limiter's burst capacity to grant + */ + private static long calculatePieceSize(final RateLimiter rateLimiter, final long bytes) { + // The burst capacity of RateLimiter.create(rate) is one second's worth of permits, so anything + // larger than that can never be granted by a single timeout-bounded attempt. + final double bytesPerSecond = rateLimiter.getRate(); + if (!(bytesPerSecond > 0) || bytesPerSecond >= Integer.MAX_VALUE) { + // Rate limiting is effectively disabled (e.g. the rate is set to Double.MAX_VALUE), but the + // piece still has to stay within the int range of the limiter's API. + return Math.min(bytes, Integer.MAX_VALUE); + } + return Math.min(bytes, (long) bytesPerSecond); + } + + /** + * @return a timeout that covers the expected wait for {@code piece}, so that a legitimate wait is + * not mistaken for a timeout + */ + private static long calculateTimeoutMs( + final RateLimiter rateLimiter, final long piece, final long checkIntervalMs) { + final double rate = rateLimiter.getRate(); + if (!(rate > 0)) { + return checkIntervalMs; + } + + final double expectedWaitMs = piece / rate * 1000.0; + if (!(expectedWaitMs > 0) || expectedWaitMs >= Long.MAX_VALUE - checkIntervalMs) { + return checkIntervalMs; + } + + return Math.max(checkIntervalMs, (long) expectedWaitMs + checkIntervalMs); + } +} diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/sink/limiter/RateLimitUtilsTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/sink/limiter/RateLimitUtilsTest.java new file mode 100644 index 0000000000000..211fb90db48e7 --- /dev/null +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/sink/limiter/RateLimitUtilsTest.java @@ -0,0 +1,150 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.commons.pipe.sink.limiter; + +import com.google.common.util.concurrent.RateLimiter; +import org.junit.Assert; +import org.junit.Test; + +import java.util.concurrent.atomic.AtomicBoolean; + +public class RateLimitUtilsTest { + + /** The rate limit and the request size reported from the field reproduction. */ + private static final double ONE_MB_PER_SECOND = 1024 * 1024; + + private static final long FIVE_MB = 5 * 1024 * 1024; + + private static final long CHECK_INTERVAL_MS = 1000; + + private static boolean acquire(final RateLimiter rateLimiter, final long bytes) { + return RateLimitUtils.acquireWithRateLimit(rateLimiter, bytes, () -> true, CHECK_INTERVAL_MS); + } + + /** + * A single request larger than the limiter's burst capacity (which is one second's worth of + * permits) must still be acquired. Before the fix, such a request made every attempt time out + * immediately, so the caller spun at 100% CPU without ever making progress. + */ + @Test(timeout = 60_000) + public void testAcquireRequestLargerThanBurstCapacity() { + final RateLimiter rateLimiter = RateLimiter.create(ONE_MB_PER_SECOND); + + final long startTime = System.nanoTime(); + final boolean acquired = acquire(rateLimiter, FIVE_MB); + final double elapsedSeconds = (System.nanoTime() - startTime) / 1_000_000_000.0; + + Assert.assertTrue(acquired); + // 5 MB at 1 MB/s, minus the one-second initial burst, must take about 4 seconds. + Assert.assertTrue( + "expected to be paced by the rate limiter, but took " + elapsedSeconds + "s", + elapsedSeconds >= 3.0); + } + + /** The average throughput must converge to the configured rate limit. */ + @Test(timeout = 60_000) + public void testAcquireIsPacedAtTheConfiguredRate() { + final RateLimiter rateLimiter = RateLimiter.create(FIVE_MB / 10.0); + + final long startTime = System.nanoTime(); + acquire(rateLimiter, FIVE_MB); + final double elapsedSeconds = (System.nanoTime() - startTime) / 1_000_000_000.0; + + // 5 MB at 0.5 MB/s is about 10 seconds, minus the one-second initial burst. + Assert.assertTrue( + "expected to be paced at the configured rate, but took " + elapsedSeconds + "s", + elapsedSeconds >= 7.0); + } + + /** A request of more than Integer.MAX_VALUE bytes must not overflow nor spin. */ + @Test(timeout = 120_000) + public void testAcquireRequestLargerThanIntegerMaxValue() { + final long bytes = 3L * Integer.MAX_VALUE; + + final RateLimiter rateLimiter = RateLimiter.create(256 * ONE_MB_PER_SECOND); + final long startTime = System.nanoTime(); + + Assert.assertTrue(acquire(rateLimiter, bytes)); + Assert.assertTrue( + "3 * Integer.MAX_VALUE bytes should be acquired in a few seconds", + (System.nanoTime() - startTime) / 1_000_000_000.0 < 60.0); + } + + /** + * The same must hold when rate limiting is disabled via {@code Double.MAX_VALUE}, even though + * such a rate makes the piece size calculation take its dedicated branch. + */ + @Test(timeout = 60_000) + public void testAcquireRequestLargerThanIntegerMaxValueWithRateLimitingDisabled() { + final RateLimiter rateLimiter = RateLimiter.create(Double.MAX_VALUE); + final long startTime = System.nanoTime(); + + // More than two pieces are needed, and every piece but the first is charged to the unlimited + // burst capacity, so this must complete without waiting. + Assert.assertTrue(acquire(rateLimiter, 3L * Integer.MAX_VALUE)); + Assert.assertTrue( + "an unlimited limiter should not make the caller wait", + (System.nanoTime() - startTime) / 1_000_000_000.0 < 10.0); + } + + /** + * When the checker asks to stop (e.g. the pipe has been dropped), a multi-piece request must + * abort instead of waiting for the remaining pieces. + */ + @Test(timeout = 60_000) + public void testAcquireAbortsWhenCheckerSaysStop() { + // The first piece is served by the initial burst, so the checker is consulted when the second + // one is about to be acquired and the request is dropped there. + final RateLimiter rateLimiter = RateLimiter.create(ONE_MB_PER_SECOND); + final AtomicBoolean checked = new AtomicBoolean(false); + + final long startTime = System.nanoTime(); + final boolean acquired = + RateLimitUtils.acquireWithRateLimit( + rateLimiter, + 4 * 1024 * 1024, + () -> { + checked.set(true); + return false; + }, + CHECK_INTERVAL_MS); + final double elapsedSeconds = (System.nanoTime() - startTime) / 1_000_000_000.0; + + Assert.assertTrue(checked.get()); + Assert.assertFalse(acquired); + // 4 MB at 1 MB/s would need about 3 seconds if the whole request were awaited. + Assert.assertTrue( + "expected to abort before waiting for the remaining pieces, but took " + + elapsedSeconds + + "s", + elapsedSeconds < 1.0); + } + + /** Non-positive requests and unlimited limiters are passed through. */ + @Test + public void testAcquireEdgeCases() { + final RateLimiter rateLimiter = RateLimiter.create(ONE_MB_PER_SECOND); + + Assert.assertTrue(acquire(rateLimiter, 0)); + Assert.assertTrue(acquire(rateLimiter, -1)); + // Double.MAX_VALUE is used to disable rate limiting, so nothing needs to be chunked or waited. + Assert.assertTrue(acquire(RateLimiter.create(Double.MAX_VALUE), FIVE_MB)); + } +} From 9ecfc4c2b1da2c2d4996e52f3d44efbf218c1c4c Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Thu, 15 Oct 2026 18:47:27 +0800 Subject: [PATCH 2/2] [Test] Add a unit test for the rate limiter configuration of obj-10-table-big-object-kill-restart Reproduces the field configuration of the pipe regression case: a rate-limit-bytes-per-second of 1 MB/s with an OBJECT batch as large as PipeSinkReadFileBufferSize (5 MB by default). The cost of one batch is 5 seconds at that rate while the hot-reload check interval used as the tryAcquire timeout is 1 second, so before the fix every attempt failed instantly and the sink thread spun at about 100% CPU. Since such a request is not rejected outright - the quota keeps accumulating during the spin and the transfer does make progress at the configured rate - the defect is wasted CPU rather than a wrong result, so a plain "does it transfer" assertion cannot tell the fixed and the pre-fix implementations apart. The test therefore asserts on the CPU time consumed by the calling thread while three batches are paced: the pre-fix loop consumes about 64% of the wall time as CPU, while the fixed implementation consumes 0%. Two tests are added: - testFieldScenarioDoesNotBusyWait paces three 5 MB batches at 1 MB/s and requires the CPU time ratio to stay below 20% while the transfer is still paced by the rate limit (at least 12s for 15 MB minus the initial burst). - testConfiguredDefaultsStayWithinTheBurstCapacity reads the effective defaults and asserts that one batch exceeds what the limiter can grant within the check interval, i.e. that splitting the request is genuinely required for this configuration. --- .../PipeRateLimiterFieldScenarioTest.java | 136 ++++++++++++++++++ 1 file changed, 136 insertions(+) create mode 100644 iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/sink/limiter/PipeRateLimiterFieldScenarioTest.java diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/sink/limiter/PipeRateLimiterFieldScenarioTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/sink/limiter/PipeRateLimiterFieldScenarioTest.java new file mode 100644 index 0000000000000..f20bae180dc41 --- /dev/null +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/sink/limiter/PipeRateLimiterFieldScenarioTest.java @@ -0,0 +1,136 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.commons.pipe.sink.limiter; + +import org.apache.iotdb.commons.conf.CommonConfig; +import org.apache.iotdb.commons.conf.CommonDescriptor; +import org.apache.iotdb.commons.pipe.config.PipeConfig; + +import com.google.common.util.concurrent.RateLimiter; +import org.junit.Assert; +import org.junit.Test; + +import java.lang.management.ManagementFactory; +import java.lang.management.ThreadMXBean; +import java.util.concurrent.TimeUnit; + +/** + * Reproduces the configuration of the pipe regression case {@code + * obj-10-table-big-object-kill-restart}: a {@code rate-limit-bytes-per-second} of 1 MB/s together + * with an OBJECT batch that is as large as {@code PipeSinkReadFileBufferSize} (5 MB by default). + * + *

With those two defaults the cost of one batch (5 seconds at 1 MB/s) is larger than the + * hot-reload check interval that used to double as the {@code tryAcquire} timeout (1 second), so + * every attempt failed instantly and the sink thread spun at roughly 100% CPU while the pipe kept + * reporting {@code RUNNING} with an empty {@code ExceptionMessage}. The stall was silent because + * that retry loop logged nothing. + * + *

Such a request is never rejected outright - the quota keeps accumulating during the spin and + * the transfer does make progress at the configured rate - so the defect shows up as wasted CPU + * rather than as a wrong result. These tests therefore assert on the CPU time consumed while the + * requests are being paced. + */ +public class PipeRateLimiterFieldScenarioTest { + + /** The {@code sink('rate-limit-bytes-per-second')} used by the regression case. */ + private static final double RATE_LIMIT_BYTES_PER_SECOND = 1048576; + + /** One OBJECT batch is bounded by {@code PipeSinkReadFileBufferSize}, 5 MB by default. */ + private static final int OBJECT_BATCH_BYTES = 5 * 1024 * 1024; + + /** + * A spin consumes CPU for almost the whole waiting time, whereas a correct implementation sleeps + * and consumes almost none. Measured with the pre-fix logic the ratio is about 64%, and with the + * fix it is 0%, so this threshold separates the two with a wide margin. + */ + private static final double MAX_CPU_TIME_RATIO = 0.2; + + private static final ThreadMXBean THREAD_MX_BEAN = ManagementFactory.getThreadMXBean(); + + @Test(timeout = 120_000) + public void testFieldScenarioDoesNotBusyWait() { + final RateLimiter rateLimiter = RateLimiter.create(RATE_LIMIT_BYTES_PER_SECOND); + + // Three batches, i.e. 15 MB, so that the initial one-second burst no longer deserves the + // transfer and the limiter has to pace the remaining ones. The complete object is 1 GB, which + // would take about 17 minutes at this rate and cannot be part of a unit test. + final long wallTimeNanos = System.nanoTime(); + final long cpuTimeNanos = THREAD_MX_BEAN.getCurrentThreadCpuTime(); + for (int batch = 0; batch < 3; batch++) { + Assert.assertTrue( + RateLimitUtils.acquireWithRateLimit( + rateLimiter, + OBJECT_BATCH_BYTES, + () -> true, + PipeConfig.getInstance().getRateLimiterHotReloadCheckIntervalMs())); + } + final long elapsedNanos = System.nanoTime() - wallTimeNanos; + final long consumedCpuNanos = THREAD_MX_BEAN.getCurrentThreadCpuTime() - cpuTimeNanos; + + final double elapsedSeconds = elapsedNanos / 1_000_000_000.0; + final double consumedCpuSeconds = consumedCpuNanos / 1_000_000_000.0; + final double cpuTimeRatio = consumedCpuSeconds / elapsedSeconds; + + // The transfer must be paced by the rate limit instead of being served by the burst, which is + // the + // case exactly when the limiter is still being asked for one batch at a time. + Assert.assertTrue( + "15 MB at 1 MB/s minus the initial burst should take at least 12s, but took " + + elapsedSeconds + + "s", + elapsedSeconds >= 12.0); + Assert.assertTrue( + "the CPU time ratio is " + + cpuTimeRatio + + " (" + + consumedCpuSeconds + + "s CPU over " + + elapsedSeconds + + "s wall), which means the sink thread was busy-waiting instead of waiting for the" + + " rate limiter", + cpuTimeRatio <= MAX_CPU_TIME_RATIO); + } + + /** + * The burst capacity of {@code RateLimiter.create(rate)} is one second's worth of permits, so a + * single batch must stay below it for one timeout-bounded attempt to be able to grant the whole + * batch. + */ + @Test + public void testConfiguredDefaultsStayWithinTheBurstCapacity() { + final CommonConfig config = CommonDescriptor.getInstance().getConfig(); + + final long burstCapacity = (long) RATE_LIMIT_BYTES_PER_SECOND; + final long timeoutNanos = + TimeUnit.MILLISECONDS.toNanos(config.getRateLimiterHotReloadCheckIntervalMs()); + final long grantableBytes = burstCapacity * timeoutNanos / 1_000_000_000L; + + // A batch larger than this cannot be granted by a timeout-bounded attempt, which is the trap + // the + // regression case fell into. + Assert.assertTrue( + "one OBJECT batch (" + + OBJECT_BATCH_BYTES + + " bytes) exceeds what the rate limiter can grant within the check interval (" + + grantableBytes + + " bytes), so the limiter has to split the request", + OBJECT_BATCH_BYTES > grantableBytes); + } +}