Repository navigation
Fix/pipe ratelimiter busy spin - #18897
luoluoyuyu wants to merge 2 commits into
Conversation
…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.
…able-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.
Caideyipi
left a comment
Caideyipi
left a comment
There was a problem hiding this comment.
Reviewed 9ecfc4c2b1da2c2d4996e52f3d44efbf218c1c4c. I found two behavioral regressions in the limiter and one diagnostic issue, detailed inline. Please address the two behavioral regressions before merging.
Validation:
- Ran the two added test classes in
node-commons: all 8 tests passed. - Reproduced the cancellation regression under contention and the zero-permit exception at
0.5bytes/s using the PR sources and Guava32.1.2-jre. - Confirmed that the new warning prints literal
%splaceholders instead of the supplied values.
| 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); |
There was a problem hiding this comment.
[P2] Keep cancellation and reload checks active under contention
Concurrent callers can reach this fallback without any rate change: Guava returns false immediately when the earliest available permit is beyond the timeout, so all 20 failures can occur without waiting. acquire() then reserves a slot and sleeps uninterruptibly, with no further continueChecker calls during that wait. Because the endpoint and global limiters are shared across callers, a dropped pipe or a hot reload that disables limiting can leave workers waiting far beyond the configured check interval.
Reproduced with eight threads sharing a 1-byte/s limiter, each requesting one byte, and a 1000 ms check interval. After setting the checker to false and interrupting the workers, five threads were still blocked in this acquire() 2.5 seconds later; the original timed retry loop had released every worker. The delay grows with the number of queued callers. Please retain bounded waits and periodic checker calls, with backoff to prevent spinning, and add a contention/cancellation regression test.
| // 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); |
There was a problem hiding this comment.
[P2] Ensure positive requests produce at least one permit
Positive rates below 1 byte/s are currently accepted by the sink configuration. For example, sink.rate-limit-bytes-per-second='0.5' passes validation, but (long) bytesPerSecond becomes 0 here. Any positive request then calls tryAcquire(0, ...) and throws IllegalArgumentException: Requested permits (0) must be positive, so the pipe cannot transfer data at that accepted rate. This is also reproducible directly with acquireWithRateLimit(RateLimiter.create(0.5), 1, () -> true, 1000); the original implementation accepts the first one-byte request. Please preserve a positive piece size for these rates and cover this boundary in a regression test.
| 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."; |
There was a problem hiding this comment.
[P3] Use SLF4J placeholders in both locale templates
This constant is passed directly to LOGGER.warn(message, bytes, piece, rate), which substitutes {} rather than %s. The added tests actually print this warning with all three literal %s placeholders, so the request size, piece size, and configured rate are missing from the diagnostic. Please change the placeholders to {} in both the English and Chinese templates.
2 participants
Description
This PR has:
for an unfamiliar reader.
for code coverage.
Key changed/added classes (or packages if there are too many classes) in this PR