xiangfu0 commented on code in PR #19864:
URL: https://github.com/apache/pinot/pull/19864#discussion_r4278281659
##########
pinot-spi/src/test/java/org/apache/pinot/spi/executor/HardLimitExecutorTest.java:
##########
@@ -127,4 +133,83 @@ public void testGetMultiStageExecutorHardLimit() {
PinotConfiguration config5 = new PinotConfiguration(configMap5);
assertEquals(HardLimitExecutor.getMultiStageExecutorHardLimit(config5),
-1);
}
+
+ /// A task that passes the submit-time check must run even if the limit is
reached before it starts. The limit used
+ /// to be re-checked at start time, which silently dropped the task: the
exception was captured by the FutureTask
+ /// created by AbstractExecutorService.submit and never reached the
submitter.
+ @Test
+ public void testAcceptedTaskRunsEvenIfLimitIsReachedBeforeItStarts()
+ throws Exception {
+ DeferredExecutor deferred = new DeferredExecutor();
+ HardLimitExecutor ex = new HardLimitExecutor(1, deferred);
+ CountDownLatch firstStarted = new CountDownLatch(1);
+ CountDownLatch firstRelease = new CountDownLatch(1);
+ CountDownLatch secondRan = new CountDownLatch(1);
+
+ // Both submissions pass the synchronous check because nothing has started
yet.
+ ex.execute(() -> {
+ firstStarted.countDown();
+ try {
+ firstRelease.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+ ex.execute(secondRan::countDown);
+ assertEquals(deferred._queued.size(), 2);
+
+ Thread first = new Thread(deferred._queued.get(0));
+ first.start();
+ assertTrue(firstStarted.await(10, TimeUnit.SECONDS), "first task did not
start");
+
+ // The running counter is at the limit now, but the second task was
already accepted and must run.
+ deferred._queued.get(1).run();
+ assertTrue(secondRan.await(10, TimeUnit.SECONDS), "accepted task was
dropped at start time");
+
+ // A third submission is rejected synchronously because the first task is
still running.
+ try {
+ ex.execute(() -> {
+ });
+ fail("Should not allow more than 1 task");
+ } catch (IllegalStateException e) {
Review Comment:
Nit: this `DeferredExecutor` is byte-for-byte identical to the one in
`HardLimitExecutorOpChainRegressionTest`. They live in different modules so
sharing isn't free, but if a test utils module is already visible to both it'd
be nice to dedupe. Totally fine to leave as-is.
##########
pinot-spi/src/main/java/org/apache/pinot/spi/executor/HardLimitExecutor.java:
##########
@@ -29,8 +29,16 @@
/// An Executor that allows a maximum of tasks running at the same time,
rejecting immediately any excess.
+///
+/// The limit is checked once, synchronously, when the task is submitted. A
task that passes that check is always
+/// run. The running counter is incremented when the task starts and
decremented when it finishes, so the number of
+/// running tasks can exceed the limit by the number of tasks that were
accepted but had not started yet when a new
+/// task was checked. On a cached thread pool this window is negligible.
Review Comment:
Nit: the "negligible on a cached pool" claim is true for the default
(`cached`) executor, but the MSE executor type is configurable and a `fixed`
pool can keep accepted tasks queued for a while. In that case the overshoot
isn't just a brief window — it can be as large as the queue backlog and stay
there. Might be worth softening the wording to note it's bounded by
concurrently-accepted-but-not-yet-started tasks, which is negligible only when
tasks start promptly (as on a cached pool).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
