Context duplicate should use the delegate task queue for ordering - f…

…ixes #3950
eclipse-vertx · May 27, 2021 · a5af83e · a5af83e
1 parent ecde0f3
commit a5af83e
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 46 deletions.
diff --git a/src/main/java/io/vertx/core/impl/DuplicatedContext.java b/src/main/java/io/vertx/core/impl/DuplicatedContext.java
@@ -35,22 +35,12 @@
 class DuplicatedContext extends AbstractContext {
 
   protected final ContextImpl delegate;
-  private TaskQueue orderedTasks;
   private ConcurrentMap<Object, Object> localData;
 
   DuplicatedContext(ContextImpl delegate) {
     this.delegate = delegate;
   }
 
-  final TaskQueue orderedTasks() {
-    synchronized (this) {
-      if (orderedTasks == null) {
-        orderedTasks = new TaskQueue();
-      }
-      return orderedTasks;
-    }
-  }
-
   @Override
   boolean inThread() {
     return delegate.inThread();
@@ -154,13 +144,7 @@ public final <T> Future<T> executeBlockingInternal(Handler<Promise<T>> action, b
 
   @Override
   public final <T> Future<T> executeBlocking(Handler<Promise<T>> action, boolean ordered) {
-    TaskQueue queue;
-    if (ordered) {
-      queue = orderedTasks();
-    } else {
-      queue = null;
-    }
-    return ContextImpl.executeBlocking(this, action, delegate.workerPool, queue);
+    return ContextImpl.executeBlocking(this, action, delegate.workerPool, delegate.orderedTasks);
   }
 
   @Override

diff --git a/src/test/java/io/vertx/core/ContextTest.java b/src/test/java/io/vertx/core/ContextTest.java
@@ -563,47 +563,35 @@ public void testDuplicateWorkerExecuteBlocking() throws Exception {
   private void testDuplicateExecuteBlocking(ContextInternal ctx) throws Exception {
     ContextInternal dup1 = ctx.duplicate();
     ContextInternal dup2 = ctx.duplicate();
-    CyclicBarrier barrier = new CyclicBarrier(3);
-    dup1.executeBlocking(p -> {
+    AtomicInteger cnt = new AtomicInteger();
+    Future<Void> f1 = dup1.executeBlocking(p -> {
       assertTrue(Context.isOnWorkerThread());
+      assertEquals(1, cnt.incrementAndGet());
       try {
-        barrier.await(10, TimeUnit.SECONDS);
-      } catch (Exception e) {
+        Thread.sleep(500);
+      } catch (InterruptedException e) {
         fail(e);
+      } finally {
+        cnt.decrementAndGet();
       }
       p.complete();
     });
-    dup2.executeBlocking(p -> {
+    Future<Void> f2 = dup2.executeBlocking(p -> {
       assertTrue(Context.isOnWorkerThread());
+      assertEquals(1, cnt.incrementAndGet());
       try {
-        barrier.await(10, TimeUnit.SECONDS);
-      } catch (Exception e) {
+        Thread.sleep(500);
+      } catch (InterruptedException e) {
         fail(e);
+      } finally {
+        cnt.decrementAndGet();
       }
       p.complete();
     });
-    barrier.await(10, TimeUnit.SECONDS);
-  }
-
-  @Test
-  public void testDuplicateEventLoopExecuteBlockingOrdering() {
-    testDuplicateExecuteBlockingOrdering((ContextInternal) vertx.getOrCreateContext());
-  }
-
-  @Test
-  public void testDuplicateWorkerExecuteBlockingOrdering() {
-    testDuplicateExecuteBlockingOrdering(createWorkerContext());
-  }
-
-  private void testDuplicateExecuteBlockingOrdering(ContextInternal context) {
-    List<Consumer<Handler<Promise<Object>>>> lst = new ArrayList<>();
-    for (int i = 0;i < 2;i++) {
-      ContextInternal duplicate = context.duplicate();
-      lst.add(task -> {
-        duplicate.executeBlocking(task, ar -> {});
-      });
-    }
-    testInternalExecuteBlockingWithQueue(lst);
+    CompositeFuture.all(f1, f2).onComplete(onSuccess(v -> {
+      testComplete();
+    }));
+    await();
   }
 
   @Test