Repository: eclipse-vertx/vert.x

Problem statement:
Deadlock when closing Vertx and WorkerExecutor concurrently
I met a deadlock issue recently on Vertx version 4.1.1 and after investigated and looked into the VertxImpl, CloseFuture, WorkerExecutorImpl, it looks like the deadlock happened in Vertx, please take a look:

In my case, the deadlock issue is hit when the `vertx.close` and `workerExecutor.close` happened to be called at same time, one from main thread and another from eventloop thread. (the `workerExecutor` is created by the `vertx.createSharedWorkerExecutor`)

Backtrace printed by jstack:
```
Found one Java-level deadlock:
=============================
"main":
  waiting to lock monitor 0x00007fa958116580 (object 0x00000000b3329eb8, a io.vertx.core.impl.WorkerExecutorImpl),
  which is held by "vert.x-eventloop-thread-0"
"vert.x-eventloop-thread-0":
  waiting to lock monitor 0x00007fa89c01ae80 (object 0x00000000b3329f58, a io.vertx.core.impl.VertxImpl),
  which is held by "main"

Java stack information for the threads listed above:
===================================================
"main":
        at io.vertx.core.impl.WorkerExecutorImpl.close(WorkerExecutorImpl.java:90)
        - waiting to lock <0x00000000b3329eb8> (a io.vertx.core.impl.WorkerExecutorImpl)
        at io.vertx.core.impl.CloseFuture.close(CloseFuture.java:117)
        at io.vertx.core.impl.VertxImpl.close(VertxImpl.java:578)
        - locked <0x00000000b3329f58> (a io.vertx.core.impl.VertxImpl)
        at io.vertx.reactivex.core.Vertx.close(Vertx.java:433)
...
"vert.x-eventloop-thread-0":
        at io.vertx.core.impl.VertxImpl$SharedWorkerPool.close(VertxImpl.java:1080)
        - waiting to lock <0x00000000b3329f58> (a io.vertx.core.impl.VertxImpl)
        at io.vertx.core.impl.WorkerExecutorImpl.close(WorkerExecutorImpl.java:93)
        - locked <0x00000000b3329eb8> (a io.vertx.core.impl.WorkerExecutorImpl)
        at io.vertx.core.impl.WorkerExecutorImpl.close(WorkerExecutorImpl.java:78)
        at io.vertx.core.impl.WorkerExecutorImpl.close(WorkerExecutorImpl.java:84)
        at io.vertx.reactivex.core.WorkerExecutor.close(WorkerExecutor.java:207)
        at io.vertx.reactivex.core.WorkerExecutor.close(WorkerExecutor.java:214)
...

Found 1 deadlock.
```

Checked vertx 4.1.1 code and find:
When closing WorkerExecutor, it will do:
1. Lock self (WorkerExecutorImpl)
2. Remove self from CloseFuture
3. Close SharedWorkerPool, which require lock Vertx

When closing Vertx, it will do:
1. Lock self (VertxImpl)
2. Close CloseFuture, which synchronized copy the list of callbacks and invoke them outside of the synchronized block, one of the callback is to close the same WorkerExecutorImpl instance as above

So, when the copy of callbacks in CloseFuture happened before removing the WorkerExecutorImpl from CloseFuture, it will still be called and cause deadlock

Time | Main thread | Eventloop thread
-- | -- | --
T1 | Start close Vertx, locked VertxImpl instance |  
T2 |   | Start close WorkerExecutor, locked WorkerExecutorImpl instance
T3 | In CloseFuture, the list of callbacks are copied |  
T4 |   | Remove WorkerExecutorImpl instance from CloseFuture. But since the callback is already copied in T3, this doesn't prevent the callback from being called
T5 | The callback of close WorkerExecutorImpl instance is called. This require lock on the WorkerExecutorImpl instance, which was acquired by eventloop thread at T2. So it is waiting for the lock to be released by eventloop thread |  
T6 |   | Close SharedWorkerPool, and this require lock on the VertxImpl instance, which was acquired by main thread at T1. So it is waiting for the lock to be released by main thread
  | Deadlock | Deadlock

The issue seems still exist in 4.1.2


Hints:
thanks, do you have a reproducer ?
Hi @vietj ,
I tried write a test case to illustrate the issue, the code loops 100 times but it usually hit the issue in less than 10 iterations on my environment:
```java
  @Test
  public void testVertx() throws InterruptedException {
    Logger LOG = LogManager.getLogger();
    for (int i = 0; i < 100; i++) {
      LOG.debug("Cur iteration: {}", i);
      Vertx vertx = Vertx.vertx();
      CountDownLatch latch = new CountDownLatch(1);
      WorkerExecutor workerExecutor = vertx.createSharedWorkerExecutor("test");
      workerExecutor.rxExecuteBlocking(handler -> handler.complete("foo"))
          .doFinally(workerExecutor::close)
          .subscribe(s -> {
            LOG.debug("Received {}", s);
            latch.countDown();
          });
      LOG.debug("Before await");
      latch.await();
      LOG.debug("After await");
      vertx.close();
    }
  }
```

thanks for the reproducer
NOTE: here is a simpler (without RxJava) reproducer

```java
  @Repeat(times = 100)
  @Test
  public void testWorkerExecutorConcurrentCloseWithVertx() throws InterruptedException {
    Vertx vertx = Vertx.vertx();
    try {
      CountDownLatch latch = new CountDownLatch(1);
      WorkerExecutor workerExecutor = vertx.createSharedWorkerExecutor("test");
      vertx.runOnContext(v -> {
        latch.countDown();
        workerExecutor.close();
      });
      latch.await();
    } finally {
      vertx.close();
    }
  }
```

Patch:
diff --git a/src/main/java/io/vertx/core/impl/VertxImpl.java b/src/main/java/io/vertx/core/impl/VertxImpl.java
--- a/src/main/java/io/vertx/core/impl/VertxImpl.java
+++ b/src/main/java/io/vertx/core/impl/VertxImpl.java
@@ -1079,11 +1079,12 @@ class SharedWorkerPool extends WorkerPool {
     @Override
     void close() {
       synchronized (VertxImpl.this) {
-        if (--refCount == 0) {
-          namedWorkerPools.remove(name);
-          super.close();
+        if (--refCount > 0) {
+          return;
         }
+        namedWorkerPools.remove(name);
       }
+      super.close();
     }
   }
 
diff --git a/src/main/java/io/vertx/core/impl/WorkerExecutorImpl.java b/src/main/java/io/vertx/core/impl/WorkerExecutorImpl.java
--- a/src/main/java/io/vertx/core/impl/WorkerExecutorImpl.java
+++ b/src/main/java/io/vertx/core/impl/WorkerExecutorImpl.java
@@ -86,12 +86,14 @@ public void close(Handler<AsyncResult<Void>> handler) {
 
   @Override
   public void close(Promise<Void> completion) {
+    boolean close;
     synchronized (this) {
-      if (!closed) {
-        closed = true;
-        closeFuture.remove(this);
-        pool.close();
-      }
+      close = !closed;
+      closed = true;
+    }
+    if (close) {
+      closeFuture.remove(this);
+      pool.close();
     }
     completion.complete();
   }


Test patch:
diff --git a/src/test/java/io/vertx/core/VertxTest.java b/src/test/java/io/vertx/core/VertxTest.java
--- a/src/test/java/io/vertx/core/VertxTest.java
+++ b/src/test/java/io/vertx/core/VertxTest.java
@@ -21,6 +21,9 @@
 import io.vertx.core.net.NetClientOptions;
 import io.vertx.core.net.NetSocket;
 import io.vertx.test.core.AsyncTestBase;
+import io.vertx.test.core.Repeat;
+import io.vertx.test.core.RepeatRule;
+import org.junit.Rule;
 import org.junit.Test;
 import org.openjdk.jmh.runner.Runner;
 import org.openjdk.jmh.runner.options.OptionsBuilder;
@@ -41,6 +44,9 @@ public class VertxTest extends AsyncTestBase {
 
   private final org.openjdk.jmh.runner.Runner RUNNER = new Runner(new OptionsBuilder().shouldDoGC(true).build());
 
+  @Rule
+  public RepeatRule repeatRule = new RepeatRule();
+
   @Test
   public void testCloseHooksCalled() {
     AtomicInteger closedCount = new AtomicInteger();
@@ -318,4 +324,21 @@ private void testTCCL(boolean disable) {
     });
     await();
   }
+
+  @Repeat(times = 100)
+  @Test
+  public void testWorkerExecutorConcurrentCloseWithVertx() throws InterruptedException {
+    Vertx vertx = Vertx.vertx();
+    try {
+      CountDownLatch latch = new CountDownLatch(1);
+      WorkerExecutor workerExecutor = vertx.createSharedWorkerExecutor("test");
+      vertx.runOnContext(v -> {
+        latch.countDown();
+        workerExecutor.close();
+      });
+      latch.await();
+    } finally {
+      vertx.close();
+    }
+  }
 }