You are viewing a plain text version of this content. The canonical link for it is here.
Posted to pr@cassandra.apache.org by "jasonk000 (via GitHub)" <gi...@apache.org> on 2024/09/09 19:27:25 UTC
[PR] Reduce lock held duration in ConcurrencyLimitingRequestThrottler [cassandra-java-driver]
jasonk000 opened a new pull request, #1957:
URL: https://github.com/apache/cassandra-java-driver/pull/1957
It might take some (small) time for callback handling when the throttler request proceeds to submission.
Before this change, the throttler proceed request will happen while holding the lock, preventing other tasks from proceeding when there is spare capacity and even preventing tasks from enqueuing until the callback completes.
By tracking the expected outcome, we can perform the callback outside of the lock. This means that request registration and submission can proceed even when a long callback is being processed.
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Reduce lock held duration in ConcurrencyLimitingRequestThrottler [cassandra-java-driver]
Posted by "tolbertam (via GitHub)" <gi...@apache.org>.
tolbertam commented on code in PR #1957:
URL: https://github.com/apache/cassandra-java-driver/pull/1957#discussion_r1752989503
##########
core/src/test/java/com/datastax/oss/driver/internal/core/session/throttling/ConcurrencyLimitingRequestThrottlerTest.java:
##########
@@ -239,4 +243,83 @@ public void should_reject_enqueued_when_closing() {
assertThatStage(request.started)
.isFailed(error -> assertThat(error).isInstanceOf(RequestThrottlingException.class));
}
+
+ @Test
+ public void should_run_throttle_callbacks_concurrently()
+ throws ExecutionException, InterruptedException, TimeoutException {
+ // Given
+
+ // a task is enqueued, which when in onThrottleReady, will stall for 100ms
+ // register() should automatically start onThrottleReady on same thread
+
+ // start a parallel thread
+ Runnable r =
+ () -> {
+ MockThrottled first = new MockThrottled(100);
+ throttler.register(first);
+ first.started.toCompletableFuture().thenRun(() -> throttler.signalSuccess(first));
+ };
+ Thread t = new Thread(r);
+ t.start();
+
+ // wait for the registration threads to be launched
+ Uninterruptibles.sleepUninterruptibly(5, TimeUnit.MILLISECONDS);
Review Comment:
Took me a moment to wrap my head around this, but it does seem that sleeping will allow deference to the scheduler to run those other threads?
It seems to deterministically work for me, but I wonder if it entirely is? Another thing I tried was adding a CountDownLatch(5) and then updated MockThrottled to optionally take a latch and calling latch.countDown() in MockThrottled.doSleep() (before sleeping then using assertTrue(latch.await(1, TimeUnit.SECONDS));. That seems like it would be more reliable to me, but I can't seem to reproduce a failure with the way it is.
I do worry about possible CI sensitivity around timing, but I think we should accept this as is and if it causes problems we could then reevaluate.
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Reduce lock held duration in ConcurrencyLimitingRequestThrottler [cassandra-java-driver]
Posted by "jasonk000 (via GitHub)" <gi...@apache.org>.
jasonk000 commented on PR #1957:
URL: https://github.com/apache/cassandra-java-driver/pull/1957#issuecomment-2349348990
Thank you @tolbertam , appreciate the review feedback & guidance. Positive experience. Thanks!
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Reduce lock held duration in ConcurrencyLimitingRequestThrottler [cassandra-java-driver]
Posted by "jasonk000 (via GitHub)" <gi...@apache.org>.
jasonk000 commented on code in PR #1957:
URL: https://github.com/apache/cassandra-java-driver/pull/1957#discussion_r1757613653
##########
core/src/test/java/com/datastax/oss/driver/internal/core/session/throttling/ConcurrencyLimitingRequestThrottlerTest.java:
##########
@@ -236,7 +239,116 @@ public void should_reject_enqueued_when_closing() {
throttler.register(request);
// Then
- assertThatStage(request.started)
+ assertThatStage(request.ended)
.isFailed(error -> assertThat(error).isInstanceOf(RequestThrottlingException.class));
}
+
+ @Test
+ public void should_run_throttle_callbacks_concurrently()
+ throws ExecutionException, InterruptedException, TimeoutException {
+ // Given
+
+ // a task is enqueued, which when in onThrottleReady, will stall for 100ms
+ // register() should automatically start onThrottleReady on same thread
+
+ // start a parallel thread
+ CountDownLatch firstRelease = new CountDownLatch(1);
+ MockThrottled first = new MockThrottled(firstRelease);
+ Runnable r =
+ () -> {
+ throttler.register(first);
+ first.ended.toCompletableFuture().thenRun(() -> throttler.signalSuccess(first));
+ };
+ Thread t = new Thread(r);
+ t.start();
+
+ // wait for the registration threads to reach await state
+ first.started.toCompletableFuture().join();
+ assertThatStage(first.started).isDone();
+ assertThatStage(first.ended).isNotDone();
+
+ // When
+ // we concurrently submit a second shorter task
+ MockThrottled second = new MockThrottled();
+ throttler.register(second);
+
+ // Then
+ // registration will trigger callback, should complete ~immediately
+ assertThatStage(second.ended).isSuccess(wasDelayed -> assertThat(wasDelayed).isFalse());
+ // first should still be unfinished
+ assertThatStage(first.started).isDone();
+ assertThatStage(first.ended).isNotDone();
+ // now finish, and verify
+ firstRelease.countDown();
+ assertThatStage(first.ended).isSuccess(wasDelayed -> assertThat(wasDelayed).isFalse());
+
+ t.join();
+ }
+
+ @Test
+ public void should_enqueue_tasks_quickly_when_callbacks_blocked() throws InterruptedException {
+ // Given
+
+ // Multiple tasks are registered, up to the limit, and proceed into their
+ // callback
+
+ // start five parallel threads
+ Thread[] threads = new Thread[5];
Review Comment:
i retained this, and added a `join()` at the end to clean up
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Reduce lock held duration in ConcurrencyLimitingRequestThrottler [cassandra-java-driver]
Posted by "adutra (via GitHub)" <gi...@apache.org>.
adutra commented on PR #1957:
URL: https://github.com/apache/cassandra-java-driver/pull/1957#issuecomment-2426254906
@charispav this PR _may_ reduce the symptoms you are seeing but imho won't get rid of them completely.
In fact this is explained in [this chapter](https://github.com/apache/cassandra-java-driver/tree/4.18.1/manual/core/non_blocking) of the docs:
> [Request throttlers] use locks internally, and depending on how many requests are being executed in parallel, the thread contention on these locks can be high: in short, if your application enforces strict lock-freedom, then these components should not be used.
If you are using the reactive API, I'd suggest that you try instead to throttle the upstream, e.g. using a token-bucket algorithm.
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Reduce lock held duration in ConcurrencyLimitingRequestThrottler [cassandra-java-driver]
Posted by "jasonk000 (via GitHub)" <gi...@apache.org>.
jasonk000 commented on PR #1957:
URL: https://github.com/apache/cassandra-java-driver/pull/1957#issuecomment-2429807874
@charispav
I think the current implementation should be good except under very extreme scenarios, in which case you probably want a different design anyway.
It may be possible to develop a more-lock-free implementation, and I did prototype it for this, but it likely needs change to semantics/behaviour to go more significantly lock-free, and I haven't personally yet seen the need to develop it.
If you suspect the driver itself has some deadlock or performance issues, the best way to proceed will be to share some stack traces and in particular the lines where the code is stalled and state of queue/counters.
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Reduce lock held duration in ConcurrencyLimitingRequestThrottler [cassandra-java-driver]
Posted by "tolbertam (via GitHub)" <gi...@apache.org>.
tolbertam commented on code in PR #1957:
URL: https://github.com/apache/cassandra-java-driver/pull/1957#discussion_r1755762998
##########
core/src/test/java/com/datastax/oss/driver/internal/core/session/throttling/ConcurrencyLimitingRequestThrottlerTest.java:
##########
@@ -236,7 +239,116 @@ public void should_reject_enqueued_when_closing() {
throttler.register(request);
// Then
- assertThatStage(request.started)
+ assertThatStage(request.ended)
.isFailed(error -> assertThat(error).isInstanceOf(RequestThrottlingException.class));
}
+
+ @Test
+ public void should_run_throttle_callbacks_concurrently()
+ throws ExecutionException, InterruptedException, TimeoutException {
+ // Given
+
+ // a task is enqueued, which when in onThrottleReady, will stall for 100ms
Review Comment:
since it no longer sleeps, should change to "will stall until latch countdown"
##########
core/src/test/java/com/datastax/oss/driver/internal/core/session/throttling/ConcurrencyLimitingRequestThrottlerTest.java:
##########
@@ -236,7 +239,116 @@ public void should_reject_enqueued_when_closing() {
throttler.register(request);
// Then
- assertThatStage(request.started)
+ assertThatStage(request.ended)
.isFailed(error -> assertThat(error).isInstanceOf(RequestThrottlingException.class));
}
+
+ @Test
+ public void should_run_throttle_callbacks_concurrently()
+ throws ExecutionException, InterruptedException, TimeoutException {
Review Comment:
small nit: `ExecutionException` and `TimeoutException` are no longer thrown by code in this method, so those can be removed from the signature
##########
core/src/test/java/com/datastax/oss/driver/internal/core/session/throttling/ConcurrencyLimitingRequestThrottlerTest.java:
##########
@@ -236,7 +239,116 @@ public void should_reject_enqueued_when_closing() {
throttler.register(request);
// Then
- assertThatStage(request.started)
+ assertThatStage(request.ended)
.isFailed(error -> assertThat(error).isInstanceOf(RequestThrottlingException.class));
}
+
+ @Test
+ public void should_run_throttle_callbacks_concurrently()
+ throws ExecutionException, InterruptedException, TimeoutException {
+ // Given
+
+ // a task is enqueued, which when in onThrottleReady, will stall for 100ms
+ // register() should automatically start onThrottleReady on same thread
+
+ // start a parallel thread
+ CountDownLatch firstRelease = new CountDownLatch(1);
+ MockThrottled first = new MockThrottled(firstRelease);
+ Runnable r =
+ () -> {
+ throttler.register(first);
+ first.ended.toCompletableFuture().thenRun(() -> throttler.signalSuccess(first));
+ };
+ Thread t = new Thread(r);
+ t.start();
+
+ // wait for the registration threads to reach await state
+ first.started.toCompletableFuture().join();
+ assertThatStage(first.started).isDone();
+ assertThatStage(first.ended).isNotDone();
+
+ // When
+ // we concurrently submit a second shorter task
+ MockThrottled second = new MockThrottled();
+ throttler.register(second);
+
+ // Then
+ // registration will trigger callback, should complete ~immediately
+ assertThatStage(second.ended).isSuccess(wasDelayed -> assertThat(wasDelayed).isFalse());
+ // first should still be unfinished
+ assertThatStage(first.started).isDone();
+ assertThatStage(first.ended).isNotDone();
+ // now finish, and verify
+ firstRelease.countDown();
+ assertThatStage(first.ended).isSuccess(wasDelayed -> assertThat(wasDelayed).isFalse());
+
+ t.join();
+ }
+
+ @Test
+ public void should_enqueue_tasks_quickly_when_callbacks_blocked() throws InterruptedException {
+ // Given
+
+ // Multiple tasks are registered, up to the limit, and proceed into their
+ // callback
+
+ // start five parallel threads
+ Thread[] threads = new Thread[5];
+ CountDownLatch[] latches = new CountDownLatch[5];
+ MockThrottled[] throttled = new MockThrottled[5];
+ for (int i = 0; i < threads.length; i++) {
+ latches[i] = new CountDownLatch(1);
+ final MockThrottled itThrottled = new MockThrottled(latches[i]);
+ throttled[i] = itThrottled;
+ Thread t =
+ new Thread(
+ () -> {
+ throttler.register(itThrottled);
+ itThrottled
+ .ended
+ .toCompletableFuture()
+ .thenRun(() -> throttler.signalSuccess(itThrottled));
+ });
+ threads[i] = t;
+ t.start();
+ }
+
+ // wait for the registration threads to be launched
+ // they are all waiting now
+ for (int i = 0; i < throttled.length; i++) {
+ throttled[i].started.toCompletableFuture().join();
+ assertThatStage(throttled[i].started).isDone();
+ assertThatStage(throttled[i].ended).isNotDone();
+ }
+
+ // When
+ // we concurrently submit another task
+ MockThrottled last = new MockThrottled();
+ throttler.register(last);
+
+ // Then
+ // registration will enqueue the callback, and it should not
+ // take any time to proceed (ie: we should not be blocked)
+ // and there should be an element in the queue
+ assertThatStage(last.started).isNotDone();
+ assertThatStage(last.ended).isNotDone();
+ assertThat(throttler.getQueue()).containsExactly(last);
+
+ // we still have not released, so old throttled threads should be waiting
+ for (int i = 0; i < throttled.length; i++) {
+ assertThatStage(throttled[i].started).isDone();
+ assertThatStage(throttled[i].ended).isNotDone();
+ }
+
+ // now let us release ..
+ for (int i = 0; i < latches.length; i++) {
+ latches[i].countDown();
+ }
+
+ // .. and check everything finished up OK
+ for (int i = 0; i < latches.length; i++) {
+ assertThatStage(throttled[i].started).isDone();
+ assertThatStage(throttled[i].ended).isSuccess();
+ }
Review Comment:
Would also be good to ensure that last completes as well:
```
assertThatStage(last.started).isSuccess();
assertThatStage(last.ended).isSuccess();
```
##########
core/src/test/java/com/datastax/oss/driver/internal/core/session/throttling/ConcurrencyLimitingRequestThrottlerTest.java:
##########
@@ -236,7 +239,116 @@ public void should_reject_enqueued_when_closing() {
throttler.register(request);
// Then
- assertThatStage(request.started)
+ assertThatStage(request.ended)
.isFailed(error -> assertThat(error).isInstanceOf(RequestThrottlingException.class));
}
+
+ @Test
+ public void should_run_throttle_callbacks_concurrently()
+ throws ExecutionException, InterruptedException, TimeoutException {
+ // Given
+
+ // a task is enqueued, which when in onThrottleReady, will stall for 100ms
+ // register() should automatically start onThrottleReady on same thread
+
+ // start a parallel thread
+ CountDownLatch firstRelease = new CountDownLatch(1);
+ MockThrottled first = new MockThrottled(firstRelease);
+ Runnable r =
+ () -> {
+ throttler.register(first);
+ first.ended.toCompletableFuture().thenRun(() -> throttler.signalSuccess(first));
+ };
+ Thread t = new Thread(r);
+ t.start();
+
+ // wait for the registration threads to reach await state
+ first.started.toCompletableFuture().join();
+ assertThatStage(first.started).isDone();
Review Comment:
One possible issue with using `join()` is that it will block until the future completes. I think if everything works as it should, this is fine, but if for whatever reason a bug is introduced it may just hang tests.
Was looking around and i saw that `CompletionStateAssert.isSuccess()` will actually do `get(2, TimeUnit.SECONDS)`, so maybe we can just change these two lines to:
```suggestion
assertThatStage(first.started).isSuccess();
```
We could do then do this anywhere else that `join` is used
##########
core/src/test/java/com/datastax/oss/driver/internal/core/session/throttling/ConcurrencyLimitingRequestThrottlerTest.java:
##########
@@ -236,7 +239,116 @@ public void should_reject_enqueued_when_closing() {
throttler.register(request);
// Then
- assertThatStage(request.started)
+ assertThatStage(request.ended)
.isFailed(error -> assertThat(error).isInstanceOf(RequestThrottlingException.class));
}
+
+ @Test
+ public void should_run_throttle_callbacks_concurrently()
+ throws ExecutionException, InterruptedException, TimeoutException {
+ // Given
+
+ // a task is enqueued, which when in onThrottleReady, will stall for 100ms
+ // register() should automatically start onThrottleReady on same thread
+
+ // start a parallel thread
+ CountDownLatch firstRelease = new CountDownLatch(1);
+ MockThrottled first = new MockThrottled(firstRelease);
+ Runnable r =
+ () -> {
+ throttler.register(first);
+ first.ended.toCompletableFuture().thenRun(() -> throttler.signalSuccess(first));
+ };
+ Thread t = new Thread(r);
+ t.start();
+
+ // wait for the registration threads to reach await state
+ first.started.toCompletableFuture().join();
+ assertThatStage(first.started).isDone();
+ assertThatStage(first.ended).isNotDone();
+
+ // When
+ // we concurrently submit a second shorter task
+ MockThrottled second = new MockThrottled();
+ throttler.register(second);
+
+ // Then
+ // registration will trigger callback, should complete ~immediately
+ assertThatStage(second.ended).isSuccess(wasDelayed -> assertThat(wasDelayed).isFalse());
+ // first should still be unfinished
+ assertThatStage(first.started).isDone();
+ assertThatStage(first.ended).isNotDone();
+ // now finish, and verify
+ firstRelease.countDown();
+ assertThatStage(first.ended).isSuccess(wasDelayed -> assertThat(wasDelayed).isFalse());
+
+ t.join();
+ }
+
+ @Test
+ public void should_enqueue_tasks_quickly_when_callbacks_blocked() throws InterruptedException {
+ // Given
+
+ // Multiple tasks are registered, up to the limit, and proceed into their
+ // callback
+
+ // start five parallel threads
+ Thread[] threads = new Thread[5];
Review Comment:
`threads` never used now so can probably get rid of this array.
##########
core/src/test/java/com/datastax/oss/driver/internal/core/session/throttling/ConcurrencyLimitingRequestThrottlerTest.java:
##########
@@ -236,7 +239,116 @@ public void should_reject_enqueued_when_closing() {
throttler.register(request);
// Then
- assertThatStage(request.started)
+ assertThatStage(request.ended)
.isFailed(error -> assertThat(error).isInstanceOf(RequestThrottlingException.class));
}
+
+ @Test
+ public void should_run_throttle_callbacks_concurrently()
+ throws ExecutionException, InterruptedException, TimeoutException {
+ // Given
+
+ // a task is enqueued, which when in onThrottleReady, will stall for 100ms
+ // register() should automatically start onThrottleReady on same thread
+
+ // start a parallel thread
+ CountDownLatch firstRelease = new CountDownLatch(1);
+ MockThrottled first = new MockThrottled(firstRelease);
+ Runnable r =
+ () -> {
+ throttler.register(first);
+ first.ended.toCompletableFuture().thenRun(() -> throttler.signalSuccess(first));
+ };
+ Thread t = new Thread(r);
+ t.start();
+
+ // wait for the registration threads to reach await state
+ first.started.toCompletableFuture().join();
+ assertThatStage(first.started).isDone();
+ assertThatStage(first.ended).isNotDone();
+
+ // When
+ // we concurrently submit a second shorter task
+ MockThrottled second = new MockThrottled();
+ throttler.register(second);
+
+ // Then
+ // registration will trigger callback, should complete ~immediately
+ assertThatStage(second.ended).isSuccess(wasDelayed -> assertThat(wasDelayed).isFalse());
+ // first should still be unfinished
+ assertThatStage(first.started).isDone();
+ assertThatStage(first.ended).isNotDone();
+ // now finish, and verify
+ firstRelease.countDown();
+ assertThatStage(first.ended).isSuccess(wasDelayed -> assertThat(wasDelayed).isFalse());
+
+ t.join();
+ }
+
+ @Test
+ public void should_enqueue_tasks_quickly_when_callbacks_blocked() throws InterruptedException {
+ // Given
+
+ // Multiple tasks are registered, up to the limit, and proceed into their
+ // callback
+
+ // start five parallel threads
+ Thread[] threads = new Thread[5];
+ CountDownLatch[] latches = new CountDownLatch[5];
+ MockThrottled[] throttled = new MockThrottled[5];
+ for (int i = 0; i < threads.length; i++) {
+ latches[i] = new CountDownLatch(1);
+ final MockThrottled itThrottled = new MockThrottled(latches[i]);
+ throttled[i] = itThrottled;
+ Thread t =
+ new Thread(
+ () -> {
+ throttler.register(itThrottled);
+ itThrottled
+ .ended
+ .toCompletableFuture()
+ .thenRun(() -> throttler.signalSuccess(itThrottled));
+ });
+ threads[i] = t;
+ t.start();
+ }
+
+ // wait for the registration threads to be launched
+ // they are all waiting now
+ for (int i = 0; i < throttled.length; i++) {
+ throttled[i].started.toCompletableFuture().join();
+ assertThatStage(throttled[i].started).isDone();
+ assertThatStage(throttled[i].ended).isNotDone();
+ }
+
+ // When
+ // we concurrently submit another task
+ MockThrottled last = new MockThrottled();
+ throttler.register(last);
+
+ // Then
+ // registration will enqueue the callback, and it should not
+ // take any time to proceed (ie: we should not be blocked)
+ // and there should be an element in the queue
+ assertThatStage(last.started).isNotDone();
+ assertThatStage(last.ended).isNotDone();
+ assertThat(throttler.getQueue()).containsExactly(last);
+
+ // we still have not released, so old throttled threads should be waiting
+ for (int i = 0; i < throttled.length; i++) {
+ assertThatStage(throttled[i].started).isDone();
+ assertThatStage(throttled[i].ended).isNotDone();
+ }
+
+ // now let us release ..
+ for (int i = 0; i < latches.length; i++) {
+ latches[i].countDown();
+ }
+
+ // .. and check everything finished up OK
+ for (int i = 0; i < latches.length; i++) {
+ assertThatStage(throttled[i].started).isDone();
Review Comment:
It doesn't seem to be a problem, but just to be sure can we also replace with `assertThatStage(throttled[i].started).isSuccess()` so it will block until it completes? I suspect that it's probably the case that the future will be completed by time it gets here but wasn't certain so figured would be good to add that just in case.
##########
core/src/test/java/com/datastax/oss/driver/internal/core/session/throttling/ConcurrencyLimitingRequestThrottlerTest.java:
##########
@@ -236,7 +239,116 @@ public void should_reject_enqueued_when_closing() {
throttler.register(request);
// Then
- assertThatStage(request.started)
+ assertThatStage(request.ended)
.isFailed(error -> assertThat(error).isInstanceOf(RequestThrottlingException.class));
}
+
+ @Test
+ public void should_run_throttle_callbacks_concurrently()
+ throws ExecutionException, InterruptedException, TimeoutException {
+ // Given
+
+ // a task is enqueued, which when in onThrottleReady, will stall for 100ms
+ // register() should automatically start onThrottleReady on same thread
+
+ // start a parallel thread
+ CountDownLatch firstRelease = new CountDownLatch(1);
+ MockThrottled first = new MockThrottled(firstRelease);
+ Runnable r =
+ () -> {
+ throttler.register(first);
+ first.ended.toCompletableFuture().thenRun(() -> throttler.signalSuccess(first));
+ };
+ Thread t = new Thread(r);
+ t.start();
+
+ // wait for the registration threads to reach await state
+ first.started.toCompletableFuture().join();
+ assertThatStage(first.started).isDone();
+ assertThatStage(first.ended).isNotDone();
+
+ // When
+ // we concurrently submit a second shorter task
+ MockThrottled second = new MockThrottled();
+ throttler.register(second);
+
+ // Then
+ // registration will trigger callback, should complete ~immediately
+ assertThatStage(second.ended).isSuccess(wasDelayed -> assertThat(wasDelayed).isFalse());
+ // first should still be unfinished
+ assertThatStage(first.started).isDone();
+ assertThatStage(first.ended).isNotDone();
+ // now finish, and verify
+ firstRelease.countDown();
+ assertThatStage(first.ended).isSuccess(wasDelayed -> assertThat(wasDelayed).isFalse());
+
+ t.join();
+ }
+
+ @Test
+ public void should_enqueue_tasks_quickly_when_callbacks_blocked() throws InterruptedException {
Review Comment:
InterruptedException no longer thrown so can be removed.
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Reduce lock held duration in ConcurrencyLimitingRequestThrottler [cassandra-java-driver]
Posted by "jasonk000 (via GitHub)" <gi...@apache.org>.
jasonk000 commented on code in PR #1957:
URL: https://github.com/apache/cassandra-java-driver/pull/1957#discussion_r1758182023
##########
core/src/test/java/com/datastax/oss/driver/internal/core/session/throttling/ConcurrencyLimitingRequestThrottlerTest.java:
##########
@@ -236,7 +237,125 @@ public void should_reject_enqueued_when_closing() {
throttler.register(request);
// Then
- assertThatStage(request.started)
+ assertThatStage(request.ended)
.isFailed(error -> assertThat(error).isInstanceOf(RequestThrottlingException.class));
}
+
+ @Test
+ public void should_run_throttle_callbacks_concurrently() throws InterruptedException {
+ // Given
+
+ // a task is enqueued, which when in onThrottleReady, will stall latch countDown()ed
+ // register() should automatically start onThrottleReady on same thread
+
+ // start a parallel thread
+ CountDownLatch firstRelease = new CountDownLatch(1);
+ MockThrottled first = new MockThrottled(firstRelease);
+ Runnable r =
+ () -> {
+ throttler.register(first);
+ first.ended.toCompletableFuture().thenRun(() -> throttler.signalSuccess(first));
+ };
+ Thread t = new Thread(r);
+ t.start();
+
+ // wait for the registration threads to reach await state
+ assertThatStage(first.started).isSuccess();
+ assertThatStage(first.ended).isNotDone();
+
+ // When
+ // we concurrently submit a second shorter task
+ MockThrottled second = new MockThrottled();
+ // (on a second thread, so that we can join and force a timeout in case
+ // registration is delayed)
+ Thread t2 = new Thread(() -> throttler.register(second));
+ t2.start();
+ t2.join(1_000);
+
+ // Then
+ // registration will trigger callback, should complete ~immediately
+ assertThatStage(second.ended).isSuccess(wasDelayed -> assertThat(wasDelayed).isFalse());
+ // first should still be unfinished
+ assertThatStage(first.started).isDone();
+ assertThatStage(first.ended).isNotDone();
+ // now finish, and verify
+ firstRelease.countDown();
+ assertThatStage(first.ended).isSuccess(wasDelayed -> assertThat(wasDelayed).isFalse());
+
+ t.join(1_000);
+ }
+
+ @Test
+ public void should_enqueue_tasks_quickly_when_callbacks_blocked() throws InterruptedException {
+ // Given
+
+ // Multiple tasks are registered, up to the limit, and proceed into their
+ // callback
+
+ // start five parallel threads
+ final int THREADS = 5;
+ Thread[] threads = new Thread[THREADS];
+ CountDownLatch[] latches = new CountDownLatch[THREADS];
+ MockThrottled[] throttled = new MockThrottled[THREADS];
+ for (int i = 0; i < threads.length; i++) {
+ latches[i] = new CountDownLatch(1);
+ final MockThrottled itThrottled = new MockThrottled(latches[i]);
+ throttled[i] = itThrottled;
+ threads[i] =
+ new Thread(
+ () -> {
+ throttler.register(itThrottled);
+ itThrottled
+ .ended
+ .toCompletableFuture()
+ .thenRun(() -> throttler.signalSuccess(itThrottled));
+ });
+ threads[i].start();
+ }
+
+ // wait for the registration threads to be launched
+ // they are all waiting now
+ for (int i = 0; i < throttled.length; i++) {
+ assertThatStage(throttled[i].started).isSuccess();
+ assertThatStage(throttled[i].ended).isNotDone();
+ }
+
+ // When
+ // we concurrently submit another task
+ MockThrottled last = new MockThrottled();
+ throttler.register(last);
+
+ // Then
+ // registration will enqueue the callback, and it should not
+ // take any time to proceed (ie: we should not be blocked)
+ // and there should be an element in the queue
+ assertThatStage(last.started).isNotDone();
+ assertThatStage(last.ended).isNotDone();
+ assertThat(throttler.getQueue()).containsExactly(last);
+
+ // we still have not released, so old throttled threads should be waiting
+ for (int i = 0; i < throttled.length; i++) {
+ assertThatStage(throttled[i].started).isDone();
+ assertThatStage(throttled[i].ended).isNotDone();
+ }
+
+ // now let us release ..
+ for (int i = 0; i < latches.length; i++) {
+ latches[i].countDown();
+ }
+
+ // .. and check everything finished up OK
+ for (int i = 0; i < latches.length; i++) {
+ assertThatStage(throttled[i].started).isSuccess();
+ assertThatStage(throttled[i].ended).isSuccess();
+ }
+
+ // for good measure, we will also wait for the enqueued to complete
+ assertThatStage(last.started).isSuccess();
+ assertThatStage(last.ended).isSuccess();
+
+ for (int i = 0; i < threads.length; i++) {
+ threads[i].join(1_000);
Review Comment:
Oops!, .. learn something new every day!
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Reduce lock held duration in ConcurrencyLimitingRequestThrottler [cassandra-java-driver]
Posted by "jasonk000 (via GitHub)" <gi...@apache.org>.
jasonk000 commented on PR #1957:
URL: https://github.com/apache/cassandra-java-driver/pull/1957#issuecomment-2341822210
Some additional details
Before -- we can see the contention over on the right hand-side showing up as futex calls; only one thread at a time is allowed to do a `LoadBalancing.. ::newQueryPlan` or `CqlRequestHandler::sendRequest` even when there is plenty of capacity in the limiter. Even though those components are designed to operate concurrently and the CQL `execution` will happen in parallel, the preparation is forced to be single-threaded.
After -- this results in the free-flowing query-plan generation where multiple queries can do their own plan & submit queries completely in parallel.
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Reduce lock held duration in ConcurrencyLimitingRequestThrottler [cassandra-java-driver]
Posted by "jasonk000 (via GitHub)" <gi...@apache.org>.
jasonk000 commented on code in PR #1957:
URL: https://github.com/apache/cassandra-java-driver/pull/1957#discussion_r1757652041
##########
core/src/test/java/com/datastax/oss/driver/internal/core/session/throttling/ConcurrencyLimitingRequestThrottlerTest.java:
##########
@@ -236,7 +239,116 @@ public void should_reject_enqueued_when_closing() {
throttler.register(request);
// Then
- assertThatStage(request.started)
+ assertThatStage(request.ended)
.isFailed(error -> assertThat(error).isInstanceOf(RequestThrottlingException.class));
}
+
+ @Test
+ public void should_run_throttle_callbacks_concurrently()
+ throws ExecutionException, InterruptedException, TimeoutException {
+ // Given
+
+ // a task is enqueued, which when in onThrottleReady, will stall for 100ms
+ // register() should automatically start onThrottleReady on same thread
+
+ // start a parallel thread
+ CountDownLatch firstRelease = new CountDownLatch(1);
+ MockThrottled first = new MockThrottled(firstRelease);
+ Runnable r =
+ () -> {
+ throttler.register(first);
+ first.ended.toCompletableFuture().thenRun(() -> throttler.signalSuccess(first));
+ };
+ Thread t = new Thread(r);
+ t.start();
+
+ // wait for the registration threads to reach await state
+ first.started.toCompletableFuture().join();
+ assertThatStage(first.started).isDone();
Review Comment:
Added this and also pushed the register() off to a second thread so that we can catch if it blocks without blocking the whole test.
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Reduce lock held duration in ConcurrencyLimitingRequestThrottler [cassandra-java-driver]
Posted by "tolbertam (via GitHub)" <gi...@apache.org>.
tolbertam commented on PR #1957:
URL: https://github.com/apache/cassandra-java-driver/pull/1957#issuecomment-2349368708
Thank you for the great fix and the tests @jasonk000! :rocket:
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Reduce lock held duration in ConcurrencyLimitingRequestThrottler [cassandra-java-driver]
Posted by "tolbertam (via GitHub)" <gi...@apache.org>.
tolbertam commented on PR #1957:
URL: https://github.com/apache/cassandra-java-driver/pull/1957#issuecomment-2348938102
Everything looks great, thank you!
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Reduce lock held duration in ConcurrencyLimitingRequestThrottler [cassandra-java-driver]
Posted by "tolbertam (via GitHub)" <gi...@apache.org>.
tolbertam commented on PR #1957:
URL: https://github.com/apache/cassandra-java-driver/pull/1957#issuecomment-2349191505
@jasonk000, we just got #1950 in, I rebased locally and there were no conflicts on this branch (does not compile though).
I went ahead and created a JIRA for this: [CASSANDRA-19922](https://issues.apache.org/jira/browse/CASSANDRA-19922)
With #1950 merged, I think we just need to have `signalCancel` updated in the same way as the changes you made for `signalTimeout`.
Woud you mind rebasing your branch, making that change and then squashing all of your commits and including this in the last line of the commit commit message?
```
patch by Jason Koch; Reviewed by Andy Tolbert and Chris Lohfink for CASSANDRA-19922
```
After that I can merge it and we'll get this included in the next release! :tada:
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Reduce lock held duration in ConcurrencyLimitingRequestThrottler [cassandra-java-driver]
Posted by "jasonk000 (via GitHub)" <gi...@apache.org>.
jasonk000 commented on PR #1957:
URL: https://github.com/apache/cassandra-java-driver/pull/1957#issuecomment-2338944933
fwiw, looks like this would conflict with #1950 , should be similar transform to combine the two
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Reduce lock held duration in ConcurrencyLimitingRequestThrottler [cassandra-java-driver]
Posted by "tolbertam (via GitHub)" <gi...@apache.org>.
tolbertam merged PR #1957:
URL: https://github.com/apache/cassandra-java-driver/pull/1957
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Reduce lock held duration in ConcurrencyLimitingRequestThrottler [cassandra-java-driver]
Posted by "tolbertam (via GitHub)" <gi...@apache.org>.
tolbertam commented on code in PR #1957:
URL: https://github.com/apache/cassandra-java-driver/pull/1957#discussion_r1758097884
##########
core/src/test/java/com/datastax/oss/driver/internal/core/session/throttling/ConcurrencyLimitingRequestThrottlerTest.java:
##########
@@ -236,7 +237,125 @@ public void should_reject_enqueued_when_closing() {
throttler.register(request);
// Then
- assertThatStage(request.started)
+ assertThatStage(request.ended)
.isFailed(error -> assertThat(error).isInstanceOf(RequestThrottlingException.class));
}
+
+ @Test
+ public void should_run_throttle_callbacks_concurrently() throws InterruptedException {
+ // Given
+
+ // a task is enqueued, which when in onThrottleReady, will stall latch countDown()ed
+ // register() should automatically start onThrottleReady on same thread
+
+ // start a parallel thread
+ CountDownLatch firstRelease = new CountDownLatch(1);
+ MockThrottled first = new MockThrottled(firstRelease);
+ Runnable r =
+ () -> {
+ throttler.register(first);
+ first.ended.toCompletableFuture().thenRun(() -> throttler.signalSuccess(first));
+ };
+ Thread t = new Thread(r);
+ t.start();
+
+ // wait for the registration threads to reach await state
+ assertThatStage(first.started).isSuccess();
+ assertThatStage(first.ended).isNotDone();
+
+ // When
+ // we concurrently submit a second shorter task
+ MockThrottled second = new MockThrottled();
+ // (on a second thread, so that we can join and force a timeout in case
+ // registration is delayed)
+ Thread t2 = new Thread(() -> throttler.register(second));
+ t2.start();
+ t2.join(1_000);
Review Comment:
From what I can tell `join(x)` doesn't throw if the thread doesn't complete, i suppose we could add `assertThat(t2.isAlive()).isFalse();` where `join(1_000)` is used to ensure the thread completes.
```suggestion
t2.join(1_000);
assertThat(t2.isAlive()).isFalse();
```
##########
core/src/test/java/com/datastax/oss/driver/internal/core/session/throttling/ConcurrencyLimitingRequestThrottlerTest.java:
##########
@@ -236,7 +237,125 @@ public void should_reject_enqueued_when_closing() {
throttler.register(request);
// Then
- assertThatStage(request.started)
+ assertThatStage(request.ended)
.isFailed(error -> assertThat(error).isInstanceOf(RequestThrottlingException.class));
}
+
+ @Test
+ public void should_run_throttle_callbacks_concurrently() throws InterruptedException {
+ // Given
+
+ // a task is enqueued, which when in onThrottleReady, will stall latch countDown()ed
+ // register() should automatically start onThrottleReady on same thread
+
+ // start a parallel thread
+ CountDownLatch firstRelease = new CountDownLatch(1);
+ MockThrottled first = new MockThrottled(firstRelease);
+ Runnable r =
+ () -> {
+ throttler.register(first);
+ first.ended.toCompletableFuture().thenRun(() -> throttler.signalSuccess(first));
+ };
+ Thread t = new Thread(r);
+ t.start();
+
+ // wait for the registration threads to reach await state
+ assertThatStage(first.started).isSuccess();
+ assertThatStage(first.ended).isNotDone();
+
+ // When
+ // we concurrently submit a second shorter task
+ MockThrottled second = new MockThrottled();
+ // (on a second thread, so that we can join and force a timeout in case
+ // registration is delayed)
+ Thread t2 = new Thread(() -> throttler.register(second));
+ t2.start();
+ t2.join(1_000);
+
+ // Then
+ // registration will trigger callback, should complete ~immediately
+ assertThatStage(second.ended).isSuccess(wasDelayed -> assertThat(wasDelayed).isFalse());
+ // first should still be unfinished
+ assertThatStage(first.started).isDone();
+ assertThatStage(first.ended).isNotDone();
+ // now finish, and verify
+ firstRelease.countDown();
+ assertThatStage(first.ended).isSuccess(wasDelayed -> assertThat(wasDelayed).isFalse());
+
+ t.join(1_000);
Review Comment:
```suggestion
t.join(1_000);
assertThat(t.isAlive()).isFalse();
```
##########
core/src/test/java/com/datastax/oss/driver/internal/core/session/throttling/ConcurrencyLimitingRequestThrottlerTest.java:
##########
@@ -236,7 +237,125 @@ public void should_reject_enqueued_when_closing() {
throttler.register(request);
// Then
- assertThatStage(request.started)
+ assertThatStage(request.ended)
.isFailed(error -> assertThat(error).isInstanceOf(RequestThrottlingException.class));
}
+
+ @Test
+ public void should_run_throttle_callbacks_concurrently() throws InterruptedException {
+ // Given
+
+ // a task is enqueued, which when in onThrottleReady, will stall latch countDown()ed
+ // register() should automatically start onThrottleReady on same thread
+
+ // start a parallel thread
+ CountDownLatch firstRelease = new CountDownLatch(1);
+ MockThrottled first = new MockThrottled(firstRelease);
+ Runnable r =
+ () -> {
+ throttler.register(first);
+ first.ended.toCompletableFuture().thenRun(() -> throttler.signalSuccess(first));
+ };
+ Thread t = new Thread(r);
+ t.start();
+
+ // wait for the registration threads to reach await state
+ assertThatStage(first.started).isSuccess();
+ assertThatStage(first.ended).isNotDone();
+
+ // When
+ // we concurrently submit a second shorter task
+ MockThrottled second = new MockThrottled();
+ // (on a second thread, so that we can join and force a timeout in case
+ // registration is delayed)
+ Thread t2 = new Thread(() -> throttler.register(second));
+ t2.start();
+ t2.join(1_000);
+
+ // Then
+ // registration will trigger callback, should complete ~immediately
+ assertThatStage(second.ended).isSuccess(wasDelayed -> assertThat(wasDelayed).isFalse());
+ // first should still be unfinished
+ assertThatStage(first.started).isDone();
+ assertThatStage(first.ended).isNotDone();
+ // now finish, and verify
+ firstRelease.countDown();
+ assertThatStage(first.ended).isSuccess(wasDelayed -> assertThat(wasDelayed).isFalse());
+
+ t.join(1_000);
+ }
+
+ @Test
+ public void should_enqueue_tasks_quickly_when_callbacks_blocked() throws InterruptedException {
+ // Given
+
+ // Multiple tasks are registered, up to the limit, and proceed into their
+ // callback
+
+ // start five parallel threads
+ final int THREADS = 5;
+ Thread[] threads = new Thread[THREADS];
+ CountDownLatch[] latches = new CountDownLatch[THREADS];
+ MockThrottled[] throttled = new MockThrottled[THREADS];
+ for (int i = 0; i < threads.length; i++) {
+ latches[i] = new CountDownLatch(1);
+ final MockThrottled itThrottled = new MockThrottled(latches[i]);
+ throttled[i] = itThrottled;
+ threads[i] =
+ new Thread(
+ () -> {
+ throttler.register(itThrottled);
+ itThrottled
+ .ended
+ .toCompletableFuture()
+ .thenRun(() -> throttler.signalSuccess(itThrottled));
+ });
+ threads[i].start();
+ }
+
+ // wait for the registration threads to be launched
+ // they are all waiting now
+ for (int i = 0; i < throttled.length; i++) {
+ assertThatStage(throttled[i].started).isSuccess();
+ assertThatStage(throttled[i].ended).isNotDone();
+ }
+
+ // When
+ // we concurrently submit another task
+ MockThrottled last = new MockThrottled();
+ throttler.register(last);
+
+ // Then
+ // registration will enqueue the callback, and it should not
+ // take any time to proceed (ie: we should not be blocked)
+ // and there should be an element in the queue
+ assertThatStage(last.started).isNotDone();
+ assertThatStage(last.ended).isNotDone();
+ assertThat(throttler.getQueue()).containsExactly(last);
+
+ // we still have not released, so old throttled threads should be waiting
+ for (int i = 0; i < throttled.length; i++) {
+ assertThatStage(throttled[i].started).isDone();
+ assertThatStage(throttled[i].ended).isNotDone();
+ }
+
+ // now let us release ..
+ for (int i = 0; i < latches.length; i++) {
+ latches[i].countDown();
+ }
+
+ // .. and check everything finished up OK
+ for (int i = 0; i < latches.length; i++) {
+ assertThatStage(throttled[i].started).isSuccess();
+ assertThatStage(throttled[i].ended).isSuccess();
+ }
+
+ // for good measure, we will also wait for the enqueued to complete
+ assertThatStage(last.started).isSuccess();
+ assertThatStage(last.ended).isSuccess();
+
+ for (int i = 0; i < threads.length; i++) {
+ threads[i].join(1_000);
Review Comment:
```suggestion
threads[i].join(1_000);
assertThat(threads[i].isAlive()).isFalse();
```
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Reduce lock held duration in ConcurrencyLimitingRequestThrottler [cassandra-java-driver]
Posted by "charispav (via GitHub)" <gi...@apache.org>.
charispav commented on PR #1957:
URL: https://github.com/apache/cassandra-java-driver/pull/1957#issuecomment-2426011945
Hi all,
While using the ConcurrencyLimitingRequestThrottler in our application (java driver 4.17.0), in case Cassandra gets overloaded and the throttling mechanism kicks in, we have a blocked thread situation:
As it is evident, the lock mechanism of throttler leads to the Vert.x IO thread (worker thread) being blocked. In general, the IO thread pool is used for possibly blocking operations. This block causes a deadlock, hence makes the entire application unresponsive.
From implementation perspective, we are using the reactive API (executeReactive method) for async query execution which in its turn uses the throttling mechanism internally.
Does the improvement for the lock mechanism in this PR also leads for the blocking issue to be fixed?
If not, how should this thread-block bug be handled? Should a separate Jira ticket be created?
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Reduce lock held duration in ConcurrencyLimitingRequestThrottler [cassandra-java-driver]
Posted by "charispav (via GitHub)" <gi...@apache.org>.
charispav commented on PR #1957:
URL: https://github.com/apache/cassandra-java-driver/pull/1957#issuecomment-2426865158
@adutra thanks for your prompt response.
In fact, even when Cassandra eventually returns to normal operation, the block remains leaving the application in a hanging state.
How would you explain this? Why does the blocked threads not get released afterwards?
Does the deadlock happen between our own Vert.x threads and DataStax threads?
How could we eliminate (or, at least mitigate) the issue if we prefer keeping this throttler without developing our own token-bucket-based solution? Is there any configuration option that might help?
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Reduce lock held duration in ConcurrencyLimitingRequestThrottler [cassandra-java-driver]
Posted by "jasonk000 (via GitHub)" <gi...@apache.org>.
jasonk000 commented on code in PR #1957:
URL: https://github.com/apache/cassandra-java-driver/pull/1957#discussion_r1755223849
##########
core/src/test/java/com/datastax/oss/driver/internal/core/session/throttling/ConcurrencyLimitingRequestThrottlerTest.java:
##########
@@ -239,4 +243,83 @@ public void should_reject_enqueued_when_closing() {
assertThatStage(request.started)
.isFailed(error -> assertThat(error).isInstanceOf(RequestThrottlingException.class));
}
+
+ @Test
+ public void should_run_throttle_callbacks_concurrently()
+ throws ExecutionException, InterruptedException, TimeoutException {
+ // Given
+
+ // a task is enqueued, which when in onThrottleReady, will stall for 100ms
+ // register() should automatically start onThrottleReady on same thread
+
+ // start a parallel thread
+ Runnable r =
+ () -> {
+ MockThrottled first = new MockThrottled(100);
+ throttler.register(first);
+ first.started.toCompletableFuture().thenRun(() -> throttler.signalSuccess(first));
+ };
+ Thread t = new Thread(r);
+ t.start();
+
+ // wait for the registration threads to be launched
+ Uninterruptibles.sleepUninterruptibly(5, TimeUnit.MILLISECONDS);
Review Comment:
Good feedback. [I've swapped over to latch-based, in this commit](https://github.com/apache/cassandra-java-driver/pull/1957/commits/670e677aab8b5354387e51fe28f6ca410b1348ec).
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org