From 0f285fc9544d9437cabb69f9d1a67b4aaf9530ff Mon Sep 17 00:00:00 2001 From: Alexandr Gorshenin Date: Sun, 30 Aug 2026 14:16:38 +0100 Subject: [PATCH 1/4] Added waiting queue test for resource autoreleasing --- .../ydb/table/impl/pool/WaitingQueueTest.java | 26 +++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/table/src/test/java/tech/ydb/table/impl/pool/WaitingQueueTest.java b/table/src/test/java/tech/ydb/table/impl/pool/WaitingQueueTest.java index 76233a173..93f6b29a4 100644 --- a/table/src/test/java/tech/ydb/table/impl/pool/WaitingQueueTest.java +++ b/table/src/test/java/tech/ydb/table/impl/pool/WaitingQueueTest.java @@ -28,6 +28,10 @@ public static class Resource { Resource(int id) { this.id = id; } + + public int getId() { + return id; + } } private static class ResourceHandler implements WaitingQueue.Handler { @@ -576,6 +580,28 @@ public void canceledWaitingTest() { check(rs).requestsCount(0).activeCount(0); } + @Test + public void immediatellyCompletedWaitingsTest() { + ResourceHandler rs = new ResourceHandler(); + WaitingQueue queue = new WaitingQueue<>(rs, 1, 9999); + + @SuppressWarnings("unchecked") + CompletableFuture[] wa = (CompletableFuture[]) new CompletableFuture[10000]; + for (int idx = 0; idx < 10000; idx++) { + wa[idx] = new CompletableFuture<>(); + queue.acquire(wa[idx]); + wa[idx].thenAccept(queue::release); + } + + check(queue).queueSize(1).idleSize(0).waitingsCount(9999); + rs.completeNext(); // all 10000 acquires must be completed immediatelly + check(queue).queueSize(1).idleSize(1).waitingsCount(0); + + queue.close(); + check(queue).queueSize(0).idleSize(0).waitingsCount(0); + check(rs).requestsCount(0).activeCount(0); + } + @Test public void checkWaitingAfterDeleteTest() { ResourceHandler rs = new ResourceHandler(); From 9c51d4b5685c36b4fba9a1a5b282227bf6537212 Mon Sep 17 00:00:00 2001 From: Alexandr Gorshenin Date: Sun, 30 Aug 2026 14:17:16 +0100 Subject: [PATCH 2/4] Fixed StackOverflow for autoreleasing resources --- .../java/tech/ydb/table/impl/pool/WaitingQueue.java | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/table/src/main/java/tech/ydb/table/impl/pool/WaitingQueue.java b/table/src/main/java/tech/ydb/table/impl/pool/WaitingQueue.java index 494e6a21a..c01ce89de 100644 --- a/table/src/main/java/tech/ydb/table/impl/pool/WaitingQueue.java +++ b/table/src/main/java/tech/ydb/table/impl/pool/WaitingQueue.java @@ -60,6 +60,8 @@ default void destroy(T object, PoolMetrics.Reason reason) { /** Size of waiting acquires queue */ private final AtomicInteger waitingAcqueireCount = new AtomicInteger(); + private final ThreadLocal localResource = new ThreadLocal<>(); + @VisibleForTesting WaitingQueue(Handler handler, int maxSize, int waitingsLimit) { Preconditions.checkArgument(maxSize > 0, "WaitingQueue max size (%s) must be positive", maxSize); @@ -110,8 +112,11 @@ public void release(T object) { return; } + boolean insideWaitingLoop = localResource.get() == object; + localResource.remove(); + // Try to complete waiting request - if (!tryToCompleteWaiting(object)) { + if (!insideWaitingLoop && !tryToCompleteWaiting(object)) { // if queue is overflowed if (queueSize.get() > limits.maxSize) { queueSize.decrementAndGet(); @@ -256,8 +261,12 @@ private boolean tryToCompleteWaiting(T object) { while (next != null) { waitingAcqueireCount.decrementAndGet(); + localResource.set(object); if (safeAcquireObject(next, object)) { - return true; + if (localResource.get() != null) { + localResource.remove(); + return true; + } } next = waitingAcquires.poll(); From e491040866b6f3353f557abc82018d284742aac8 Mon Sep 17 00:00:00 2001 From: Alexandr Gorshenin Date: Sun, 30 Aug 2026 14:17:46 +0100 Subject: [PATCH 3/4] Avoid EndpointPool locks for expired requests --- .../tech/ydb/core/impl/BaseGrpcTransport.java | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/core/src/main/java/tech/ydb/core/impl/BaseGrpcTransport.java b/core/src/main/java/tech/ydb/core/impl/BaseGrpcTransport.java index bdd3504f5..e15eb5c5a 100644 --- a/core/src/main/java/tech/ydb/core/impl/BaseGrpcTransport.java +++ b/core/src/main/java/tech/ydb/core/impl/BaseGrpcTransport.java @@ -119,13 +119,14 @@ public CompletableFuture> unaryCall( String traceId = settings.getTraceId(); try { - GrpcChannel channel = getChannel(settings); - EndpointRecord endpoint = channel.getEndpoint(); CallOptions options = prepareCallOptions(settings); if (options == null) { return CompletableFuture.completedFuture(deadlineExpiredResult(method, settings)); } + GrpcChannel channel = getChannel(settings); + EndpointRecord endpoint = channel.getEndpoint(); + ClientCall call = channel.getReadyChannel().newCall(method, options); ChannelStatusHandler handler = new ChannelStatusHandler(channel, settings); @@ -158,13 +159,14 @@ public GrpcReadStream readStreamCall( String traceId = settings.getTraceId(); try { - GrpcChannel channel = getChannel(settings); - EndpointRecord endpoint = channel.getEndpoint(); CallOptions options = prepareCallOptions(settings); if (options == null) { return new EmptyStream<>(deadlineExpiredStatus(method, settings)); } + GrpcChannel channel = getChannel(settings); + EndpointRecord endpoint = channel.getEndpoint(); + ClientCall call = channel.getReadyChannel().newCall(method, options); ChannelStatusHandler handler = new ChannelStatusHandler(channel, settings); @@ -200,13 +202,14 @@ public GrpcReadWriteStream readWriteStreamCall( String traceId = settings.getTraceId(); try { - GrpcChannel channel = getChannel(settings); - EndpointRecord endpoint = channel.getEndpoint(); CallOptions options = prepareCallOptions(settings); if (options == null) { return new EmptyStream<>(deadlineExpiredStatus(method, settings)); } + GrpcChannel channel = getChannel(settings); + EndpointRecord endpoint = channel.getEndpoint(); + ClientCall call = channel.getReadyChannel().newCall(method, options); ChannelStatusHandler hdlr = new ChannelStatusHandler(channel, settings); From 7f7218b1c3ad2b18c65f149754ac3af008df6005 Mon Sep 17 00:00:00 2001 From: Alexandr Gorshenin Date: Mon, 31 Aug 2026 13:43:50 +0100 Subject: [PATCH 4/4] Fixed waiting guard --- .../ydb/table/impl/pool/WaitingQueue.java | 63 ++++++++++++++----- .../ydb/table/impl/pool/WaitingQueueTest.java | 49 +++++++++++++++ 2 files changed, 97 insertions(+), 15 deletions(-) diff --git a/table/src/main/java/tech/ydb/table/impl/pool/WaitingQueue.java b/table/src/main/java/tech/ydb/table/impl/pool/WaitingQueue.java index c01ce89de..a82572118 100644 --- a/table/src/main/java/tech/ydb/table/impl/pool/WaitingQueue.java +++ b/table/src/main/java/tech/ydb/table/impl/pool/WaitingQueue.java @@ -60,7 +60,7 @@ default void destroy(T object, PoolMetrics.Reason reason) { /** Size of waiting acquires queue */ private final AtomicInteger waitingAcqueireCount = new AtomicInteger(); - private final ThreadLocal localResource = new ThreadLocal<>(); + private final ThreadLocal localGuard = new ThreadLocal<>(); @VisibleForTesting WaitingQueue(Handler handler, int maxSize, int waitingsLimit) { @@ -112,11 +112,13 @@ public void release(T object) { return; } - boolean insideWaitingLoop = localResource.get() == object; - localResource.remove(); + WaitingGuard guard = localGuard.get(); + if (guard != null && !guard.isAllowed(object)) { + return; + } // Try to complete waiting request - if (!insideWaitingLoop && !tryToCompleteWaiting(object)) { + if (!tryToCompleteWaiting(object)) { // if queue is overflowed if (queueSize.get() > limits.maxSize) { queueSize.decrementAndGet(); @@ -257,22 +259,20 @@ private boolean tryToCompleteWaiting(T object) { return false; } - CompletableFuture next = waitingAcquires.poll(); - while (next != null) { - waitingAcqueireCount.decrementAndGet(); - - localResource.set(object); - if (safeAcquireObject(next, object)) { - if (localResource.get() != null) { - localResource.remove(); + try (WaitingGuard guard = new WaitingGuard(object)) { + CompletableFuture next = waitingAcquires.poll(); + while (next != null) { + guard.init(); + waitingAcqueireCount.decrementAndGet(); + if (safeAcquireObject(next, object) && !guard.isBroken()) { return true; } + + next = waitingAcquires.poll(); } - next = waitingAcquires.poll(); + return false; } - - return false; } private void checkNextWaitingAcquire() { @@ -316,6 +316,39 @@ private void clear() { } } + private final class WaitingGuard implements AutoCloseable { + private final WaitingGuard prev; + private final Object obj; + private boolean isBroken = false; + + WaitingGuard(Object obj) { + this.prev = localGuard.get(); + this.obj = obj; + } + + public void init() { + localGuard.set(this); + isBroken = false; + } + + @Override + public void close() { + localGuard.set(prev); + } + + private boolean isAllowed(Object object) { + if (object != obj) { + return true; + } + isBroken = true; + return false; + } + + public boolean isBroken() { + return isBroken; + } + } + private static class Limits { private final int maxSize; private final int waitingsLimit; diff --git a/table/src/test/java/tech/ydb/table/impl/pool/WaitingQueueTest.java b/table/src/test/java/tech/ydb/table/impl/pool/WaitingQueueTest.java index 93f6b29a4..9616ebf2e 100644 --- a/table/src/test/java/tech/ydb/table/impl/pool/WaitingQueueTest.java +++ b/table/src/test/java/tech/ydb/table/impl/pool/WaitingQueueTest.java @@ -602,6 +602,55 @@ public void immediatellyCompletedWaitingsTest() { check(rs).requestsCount(0).activeCount(0); } + @Test + public void immediatellyCompletedOtherTest() { + ResourceHandler rs = new ResourceHandler(); + WaitingQueue queue = new WaitingQueue<>(rs, 2, 3); + + CompletableFuture r1 = pendingFuture(acquire(queue)); + CompletableFuture r2 = pendingFuture(acquire(queue)); + check(queue).queueSize(2).idleSize(0).waitingsCount(0); + + rs.completeNext().completeNext(); + Resource a = pendingIsReady(r1); + Resource b = pendingIsReady(r2); + check(queue).queueSize(2).idleSize(0).waitingsCount(0); + + CompletableFuture w1 = pendingFuture(acquire(queue)); + CompletableFuture w2 = pendingFuture(acquire(queue)); + w1.thenAccept(ignored -> queue.release(b)); + check(queue).queueSize(2).idleSize(0).waitingsCount(2); + + queue.release(a); + + Assert.assertSame(a, pendingIsReady(w1)); + Assert.assertSame(b, pendingIsReady(w2)); + check(queue).queueSize(2).idleSize(0).waitingsCount(0); + } + + @Test + public void currentReleaseBeforeOtherMustNotBeLost() { + ResourceHandler rs = new ResourceHandler(); + WaitingQueue queue = new WaitingQueue<>(rs, 2, 3); + + CompletableFuture first = pendingFuture(acquire(queue)); + CompletableFuture second = pendingFuture(acquire(queue)); + rs.completeNext().completeNext(); + Resource a = pendingIsReady(first); + Resource b = pendingIsReady(second); + + CompletableFuture waiting = pendingFuture(acquire(queue)); + waiting.thenAccept(ignored -> { + queue.release(a); + queue.release(b); + }); + + queue.release(a); + + Assert.assertSame(a, pendingIsReady(waiting)); + check(queue).queueSize(2).idleSize(2).waitingsCount(0); + } + @Test public void checkWaitingAfterDeleteTest() { ResourceHandler rs = new ResourceHandler();