Merge pull request #568 from alex268/master

Fixed session creation loop

Author: alex268

table/src/main/java/tech/ydb/table/impl/pool/WaitingQueue.java

@@ -61,7 +61,7 @@ public interface Handler<T> {
 
     @VisibleForTesting
     WaitingQueue(Handler<T> handler, int maxSize, int waitingsLimit) {
-        Preconditions.checkArgument(maxSize > 0, "WaitingQueue max size (%d) must be positive", maxSize);
+        Preconditions.checkArgument(maxSize > 0, "WaitingQueue max size (%s) must be positive", maxSize);
         Preconditions.checkArgument(handler != null, "WaitingQueue handler must be not null");
 
         this.handler = handler;
@@ -267,23 +267,20 @@ private boolean tryToCompleteWaiting(T object) {
     }
 
     private void checkNextWaitingAcquire() {
-        if (stopped || waitingAcquires.isEmpty()) {
+        if (stopped) {
             return;
         }
 
-        // Try to create new pending request
-        CompletableFuture<T> pending = new CompletableFuture<>();
-        if (tryToCreateNewPending(pending)) {
-            pending.whenComplete((object, th) -> {
-                if (th != null) {
-                    checkNextWaitingAcquire();
-                }
-                if (object != null) {
-                    if (!tryToCompleteWaiting(object)) {
-                        idle.offerFirst(object);
-                    }
-                }
-            });
+        CompletableFuture<T> next = waitingAcquires.peek();
+        while (next != null && next.isDone()) {
+            waitingAcquires.remove();
+            waitingAcqueireCount.decrementAndGet();
+            next = waitingAcquires.peek();
+        }
+
+        if (next != null && tryToCreateNewPending(next)) {
+            waitingAcquires.remove();
+            waitingAcqueireCount.decrementAndGet();
         }
     }
 
@@ -343,15 +340,19 @@ public void accept(T object, Throwable th) {
             if (th != null) {
                 queueSize.decrementAndGet();
                 acquire.completeExceptionally(th);
+                checkNextWaitingAcquire();
                 return;
             }
 
-            if (!acquire.isDone() && safeAcquireObject(acquire, object)) {
+            if (safeAcquireObject(acquire, object)) {
                 return;
             }
 
-            // If acquire future is already canceled, put new object to hot queue
-            idle.offerFirst(object); // ConcurrentLinkedQueue always return true
+            // If acquire future is already canceled, try to complete waiting or put to hot queue
+            if (!tryToCompleteWaiting(object)) {
+                idle.offerFirst(object); // ConcurrentLinkedQueue always return true
+            }
+
             if (stopped) {
                 clear();
             }

table/src/test/java/tech/ydb/table/impl/pool/SessionPoolTest.java

@@ -117,11 +117,10 @@ public void increaseLimitsTest() {
         // increase pool limit
         pool.updateMaxSize(2);
         tableRpc.check().sessionRequests(2);
-        // NOTE: we have increasing of pendings because one request may be in both queues (pendings + waitings)
-        check(pool).idle(0).acquired(0).pending(2 + wsize).size(0, 2).totalClient(0, 0).totalServer(2, 0, 0, 0);
+        check(pool).idle(0).acquired(0).pending(1 + wsize).size(0, 2).totalClient(0, 0).totalServer(2, 0, 0, 0);
 
         tableRpc.nextCreateSession().completeSuccess();
-        check(pool).idle(0).acquired(1).pending(1 + wsize).size(0, 2).totalClient(1, 0).totalServer(2, 1, 0, 0);
+        check(pool).idle(0).acquired(1).pending(wsize).size(0, 2).totalClient(1, 0).totalServer(2, 1, 0, 0);
 
         tableRpc.nextCreateSession().completeSuccess();
         check(pool).idle(0).acquired(2).pending(-1 + wsize).size(0, 2).totalClient(2, 0).totalServer(2, 2, 0, 0);

table/src/test/java/tech/ydb/table/impl/pool/WaitingQueueTest.java

@@ -579,14 +579,16 @@ public void canceledWaitingTest() {
     @Test
     public void checkWaitingAfterDeleteTest() {
         ResourceHandler rs = new ResourceHandler();
-        WaitingQueue<Resource> queue = new WaitingQueue<>(rs, 1, 3);
+        WaitingQueue<Resource> queue = new WaitingQueue<>(rs, 1, 5);
 
         CompletableFuture<Resource> f1 = pendingFuture(acquire(queue));
         CompletableFuture<Resource> w1 = pendingFuture(acquire(queue));
         CompletableFuture<Resource> w2 = pendingFuture(acquire(queue));
         CompletableFuture<Resource> w3 = pendingFuture(acquire(queue));
+        CompletableFuture<Resource> w4 = pendingFuture(acquire(queue));
+        CompletableFuture<Resource> w5 = pendingFuture(acquire(queue));
 
-        check(queue).queueSize(1).idleSize(0).waitingsCount(3);
+        check(queue).queueSize(1).idleSize(0).waitingsCount(5);
         rs.completeNext();
         check(rs).requestsCount(0).activeCount(1);
 
@@ -596,41 +598,46 @@ public void checkWaitingAfterDeleteTest() {
         queue.delete(r1);
 
         check(rs).requestsCount(1).activeCount(0);
-        check(queue).queueSize(1).idleSize(0).waitingsCount(3);
+        check(queue).queueSize(1).idleSize(0).waitingsCount(4);
         futureIsPending(w1);
         futureIsPending(w2);
         futureIsPending(w3);
+        futureIsPending(w4);
 
         rs.completeNext();
         check(rs).requestsCount(0).activeCount(1);
-        check(queue).queueSize(1).idleSize(0).waitingsCount(2);
+        check(queue).queueSize(1).idleSize(0).waitingsCount(4);
 
         Resource r2 = pendingIsReady(w1);
         futureIsPending(w2);
+        futureIsPending(w3);
+        w3.cancel(true);
 
         Assert.assertNotEquals("After deleting waiting got different resource ", r1, r2);
 
         queue.delete(r2);
         check(rs).requestsCount(1).activeCount(0);
 
-        // If pending completed with exception - queue must repeat it
+        // If pending completed with exception - next waiting got that trouble
         rs.completeNextWithException(new RuntimeException("Trouble"));
+        futureIsExceptionally(w2, "Trouble");
+
         check(rs).requestsCount(1).activeCount(0);
 
         rs.completeNext();
 
-        Resource r3 = pendingIsReady(w2);
-        futureIsPending(w3);
+        Resource r4 = pendingIsReady(w4);
+        futureIsPending(w5);
 
-        Assert.assertNotEquals("After deleting waiting got different resource ", r2, r3);
+        Assert.assertNotEquals("After deleting waiting got different resource ", r2, r4);
         check(rs).requestsCount(0).activeCount(1);
         check(queue).queueSize(1).idleSize(0).waitingsCount(1);
 
-        queue.delete(r3);
+        queue.delete(r4);
 
         // After canceling of pending waiting queue will move resource to idle
         check(rs).requestsCount(1).activeCount(0);
-        w3.cancel(true);
+        w5.cancel(true);
         rs.completeNext();
 
         check(rs).requestsCount(0).activeCount(1);
@@ -660,14 +667,17 @@ public void cleanWaitingAfterClosingTest() {
         queue.release(r1);
     }
 
-    @Test(expected = IllegalArgumentException.class)
-    public void validateQueueMaxSize() {
-        new WaitingQueue<>(new ResourceHandler(), 0, 3);
-    }
-
-    @Test(expected = IllegalArgumentException.class)
-    public void validateQueueHandler() {
-        new WaitingQueue<>(null, 1);
+    @Test
+    public void validateQueueArguments() {
+        IllegalArgumentException ex1 = Assert.assertThrows("Invalid max size expected", IllegalArgumentException.class,
+                () -> new WaitingQueue<>(new ResourceHandler(), 0, 3)
+        );
+        Assert.assertEquals("WaitingQueue max size (0) must be positive", ex1.getMessage());
+
+        IllegalArgumentException ex2 = Assert.assertThrows("Invalid handler expected", IllegalArgumentException.class,
+                () -> new WaitingQueue<>(null, 1, 3)
+        );
+        Assert.assertEquals("WaitingQueue handler must be not null", ex2.getMessage());
     }
 
     @Test