Fixed session pool deadlock

Author: alex268

core/src/test/java/tech/ydb/core/impl/pool/EndpointPoolTest.java

@@ -37,11 +37,12 @@ public class EndpointPoolTest {
 
     @Before
     public void setUp() throws IOException {
+        Mockito.doNothing().when(socket).connect(Mockito.any(SocketAddress.class));
+        Mockito.when(socketFactory.createSocket()).thenReturn(socket);
+
         mocks = MockitoAnnotations.openMocks(this);
         threadLocalStaticMock.when(ThreadLocalRandom::current).thenReturn(random);
         socketFactoryStaticMock.when(SocketFactory::getDefault).thenReturn(socketFactory);
-        Mockito.when(socketFactory.createSocket()).thenReturn(socket);
-        Mockito.doNothing().when(socket).connect(Mockito.any(SocketAddress.class));
     }
 
     @After

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

@@ -140,7 +140,7 @@ public void delete(T object) {
         handler.destroy(object);
 
         // After deleting one object we can try to create new pending if it needed
-        checkNextWaitingAcquire();
+        CompletableFuture.runAsync(this::checkNextWaitingAcquire);
     }
 
     @Override

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

@@ -16,6 +16,7 @@
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+
 import org.junit.Assert;
 
 /**
@@ -30,11 +31,11 @@ public class MockedScheduler implements ScheduledExecutorService {
     public MockedScheduler(MockedClock clock) {
         this.clock = clock;
     }
-    
+
     public Checker check() {
         return new Checker();
     }
-    
+
     public void runTasksTo(Instant timestamp, Runnable... runs) {
         int runIdx = 0;
         MockedTask<?> next = tasks.peek();
@@ -57,7 +58,7 @@ public void runTasksTo(Instant timestamp, Runnable... runs) {
 
         clock.goToFuture(timestamp);
     }
-    
+
     @Override
     public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
         Instant time = clock.instant().plusNanos(unit.toNanos(delay));
@@ -133,27 +134,27 @@ public <T> Future<T> submit(Callable<T> task) {
 
     @Override
     public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException {
-        throw new UnsupportedOperationException("Not supported yet."); 
+        throw new UnsupportedOperationException("Not supported yet.");
     }
 
     @Override
     public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException {
-        throw new UnsupportedOperationException("Not supported yet."); 
+        throw new UnsupportedOperationException("Not supported yet.");
     }
 
     @Override
     public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException {
-        throw new UnsupportedOperationException("Not supported yet."); 
+        throw new UnsupportedOperationException("Not supported yet.");
     }
 
     @Override
     public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
-        throw new UnsupportedOperationException("Not supported yet."); 
+        throw new UnsupportedOperationException("Not supported yet.");
     }
 
     @Override
     public void execute(Runnable command) {
-        throw new UnsupportedOperationException("Not supported yet."); 
+        throw new UnsupportedOperationException("Not supported yet.");
     }
 
     private class MockedTask<V> extends FutureTask<V> implements RunnableScheduledFuture<V> {
@@ -195,7 +196,7 @@ public int compareTo(Delayed other) {
                 MockedTask<?> x = (MockedTask<?>)other;
                 return time.compareTo(x.time);
             }
-            
+
             @SuppressWarnings("null")
             long diff = getDelay(TimeUnit.MILLISECONDS) - other.getDelay(TimeUnit.MILLISECONDS);
             return (diff < 0) ? -1 : (diff > 0) ? 1 : 0;
@@ -225,7 +226,7 @@ public void run() {
                 cancel(false);
                 return;
             }
-            
+
             if (isPeriodic()) {
                 if (super.runAndReset()) {
                     setNextRunTime();
@@ -236,14 +237,14 @@ public void run() {
             }
         }
     }
-    
+
     public class Checker {
         public Checker isClosed() {
             Assert.assertTrue("Scheduler is shutdown", isShutdown());
             Assert.assertTrue("Scheduler is terminated", isTerminated());
             return this;
         }
-    
+
         public Checker hasNoTasks() {
             Assert.assertTrue("Scheduler hasn't tasks", tasks.isEmpty());
             return this;

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

@@ -8,6 +8,8 @@
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Supplier;
 
@@ -595,6 +597,7 @@ public void checkWaitingAfterDeleteTest() {
         // After deleting current resource queue must create new pending to complete waiting
         queue.delete(r1);
 
+        ForkJoinPool.commonPool().awaitQuiescence(1, TimeUnit.SECONDS);
         check(rs).requestsCount(1).activeCount(0);
         check(queue).queueSize(1).idleSize(0).waitingsCount(3);
         futureIsPending(w1);
@@ -611,6 +614,8 @@ public void checkWaitingAfterDeleteTest() {
         Assert.assertNotEquals("After deleting waiting got different resource ", r1, r2);
 
         queue.delete(r2);
+
+        ForkJoinPool.commonPool().awaitQuiescence(1, TimeUnit.SECONDS);
         check(rs).requestsCount(1).activeCount(0);
 
         // If pending completed with exception - queue must repeat it
@@ -623,16 +628,20 @@ public void checkWaitingAfterDeleteTest() {
         futureIsPending(w3);
 
         Assert.assertNotEquals("After deleting waiting got different resource ", r2, r3);
+
+        ForkJoinPool.commonPool().awaitQuiescence(1, TimeUnit.SECONDS);
         check(rs).requestsCount(0).activeCount(1);
         check(queue).queueSize(1).idleSize(0).waitingsCount(1);
 
         queue.delete(r3);
 
         // After canceling of pending waiting queue will move resource to idle
+        ForkJoinPool.commonPool().awaitQuiescence(1, TimeUnit.SECONDS);
         check(rs).requestsCount(1).activeCount(0);
         w3.cancel(true);
         rs.completeNext();
 
+        ForkJoinPool.commonPool().awaitQuiescence(1, TimeUnit.SECONDS);
         check(rs).requestsCount(0).activeCount(1);
         check(queue).queueSize(1).idleSize(1).waitingsCount(0);