Repository: eclipse-vertx/vert.x

Problem statement:
Timers created outside event loop might not fire
When a timer is created from a non vertx thread, there can be a race with the timer task that executes before the internal timer task is added to the vertx timeout map. When it happens, the timer will not execute since this check is to check whether the task was cancelled by another thread.


Hints:

Patch:
diff --git a/src/main/java/io/vertx/core/impl/AbstractContext.java b/src/main/java/io/vertx/core/impl/AbstractContext.java
--- a/src/main/java/io/vertx/core/impl/AbstractContext.java
+++ b/src/main/java/io/vertx/core/impl/AbstractContext.java
@@ -18,6 +18,7 @@
 import io.vertx.core.impl.launcher.VertxCommandLauncher;
 
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 
 /**
  * A context implementation that does not hold any specific state.
@@ -84,13 +85,13 @@ public final void endDispatch(ContextInternal previous) {
   @Override
   public long setPeriodic(long delay, Handler<Long> handler) {
     VertxImpl owner = (VertxImpl) owner();
-    return owner.scheduleTimeout(this, handler, delay, true);
+    return owner.scheduleTimeout(this, true, delay, TimeUnit.MILLISECONDS, handler);
   }
 
   @Override
   public long setTimer(long delay, Handler<Long> handler) {
     VertxImpl owner = (VertxImpl) owner();
-    return owner.scheduleTimeout(this, handler, delay, false);
+    return owner.scheduleTimeout(this, false, delay, TimeUnit.MILLISECONDS,handler);
   }
 
   @Override
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
@@ -77,6 +77,7 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Supplier;
@@ -357,7 +358,7 @@ public EventBus eventBus() {
   }
 
   public long setPeriodic(long delay, Handler<Long> handler) {
-    return scheduleTimeout(getOrCreateContext(), handler, delay, true);
+    return scheduleTimeout(getOrCreateContext(), true, delay, TimeUnit.MILLISECONDS, handler);
   }
 
   @Override
@@ -366,7 +367,7 @@ public TimeoutStream periodicStream(long delay) {
   }
 
   public long setTimer(long delay, Handler<Long> handler) {
-    return scheduleTimeout(getOrCreateContext(), handler, delay, false);
+    return scheduleTimeout(getOrCreateContext(), false, delay, TimeUnit.MILLISECONDS, handler);
   }
 
   @Override
@@ -451,10 +452,9 @@ public Metrics getMetrics() {
   }
 
   public boolean cancelTimer(long id) {
-    InternalTimerHandler handler = timeouts.remove(id);
+    InternalTimerHandler handler = timeouts.get(id);
     if (handler != null) {
-      handler.cancel();
-      return true;
+      return handler.cancel();
     } else {
       return false;
     }
@@ -510,16 +510,22 @@ public DnsClient createDnsClient(DnsClientOptions options) {
     return new DnsClientImpl(this, options);
   }
 
-  public long scheduleTimeout(ContextInternal context, Handler<Long> handler, long delay, boolean periodic) {
+  public long scheduleTimeout(ContextInternal context, boolean periodic, long delay, TimeUnit timeUnit, Handler<Long> handler) {
     if (delay < 1) {
       throw new IllegalArgumentException("Cannot schedule a timer with delay < 1 ms");
     }
     long timerId = timeoutCounter.getAndIncrement();
-    InternalTimerHandler task = new InternalTimerHandler(timerId, handler, periodic, delay, context);
+    InternalTimerHandler task = new InternalTimerHandler(timerId, handler, periodic, context);
     timeouts.put(timerId, task);
     if (context.isDeployment()) {
       context.addCloseHook(task);
     }
+    EventLoop el = context.nettyEventLoop();
+    if (periodic) {
+      task.future = el.scheduleAtFixedRate(task, delay, delay, timeUnit);
+    } else {
+      task.future = el.schedule(task, delay, timeUnit);
+    }
     return timerId;
   }
 
@@ -869,19 +875,14 @@ private class InternalTimerHandler implements Handler<Void>, Closeable, Runnable
     private final boolean periodic;
     private final long timerID;
     private final ContextInternal context;
-    private final java.util.concurrent.Future<?> future;
+    private final AtomicBoolean disposed = new AtomicBoolean();
+    private volatile java.util.concurrent.Future<?> future;
 
-    InternalTimerHandler(long timerID, Handler<Long> runnable, boolean periodic, long delay, ContextInternal context) {
+    InternalTimerHandler(long timerID, Handler<Long> runnable, boolean periodic, ContextInternal context) {
       this.context = context;
       this.timerID = timerID;
       this.handler = runnable;
       this.periodic = periodic;
-      EventLoop el = context.nettyEventLoop();
-      if (periodic) {
-        future = el.scheduleAtFixedRate(this, delay, delay, TimeUnit.MILLISECONDS);
-      } else {
-        future = el.schedule(this, delay, TimeUnit.MILLISECONDS);
-      }
     }
 
     @Override
@@ -891,10 +892,11 @@ public void run() {
 
     public void handle(Void v) {
       if (periodic) {
-        if (timeouts.containsKey(timerID)) {
+        if (!disposed.get()) {
           handler.handle(timerID);
         }
-      } else if (timeouts.remove(timerID) != null) {
+      } else if (disposed.compareAndSet(false, true)) {
+        timeouts.remove(timerID);
         try {
           handler.handle(timerID);
         } finally {
@@ -904,18 +906,29 @@ public void handle(Void v) {
       }
     }
 
-    private void cancel() {
-      future.cancel(false);
-      if (context.isDeployment()) {
-        context.removeCloseHook(this);
+    private boolean cancel() {
+      boolean cancelled = tryCancel();
+      if (cancelled) {
+        if (context.isDeployment()) {
+          context.removeCloseHook(this);
+        }
+      }
+      return cancelled;
+    }
+
+    private boolean tryCancel() {
+      if  (disposed.compareAndSet(false, true)) {
+        timeouts.remove(timerID);
+        future.cancel(false);
+        return true;
+      } else {
+        return false;
       }
     }
 
     // Called via Context close hook when Verticle is undeployed
     public void close(Promise<Void> completion) {
-      if (timeouts.remove(timerID) != null) {
-        future.cancel(false);
-      }
+      tryCancel();
       completion.complete();
     }
   }
@@ -987,7 +1000,7 @@ public synchronized TimeoutStream handler(Handler<Long> handler) {
           throw new IllegalStateException();
         }
         this.handler = handler;
-        id = scheduleTimeout(getOrCreateContext(), this, delay, periodic);
+        id = scheduleTimeout(getOrCreateContext(), periodic, delay, TimeUnit.MILLISECONDS, this);
       } else {
         cancel();
       }


Test patch:
diff --git a/src/test/java/io/vertx/core/TimerTest.java b/src/test/java/io/vertx/core/TimerTest.java
--- a/src/test/java/io/vertx/core/TimerTest.java
+++ b/src/test/java/io/vertx/core/TimerTest.java
@@ -12,11 +12,14 @@
 package io.vertx.core;
 
 import io.vertx.core.impl.ContextInternal;
+import io.vertx.core.impl.VertxImpl;
 import io.vertx.core.impl.VertxInternal;
 import io.vertx.core.streams.ReadStream;
+import io.vertx.test.core.Repeat;
 import io.vertx.test.core.VertxTestBase;
 import org.junit.Test;
 
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
@@ -446,4 +449,24 @@ public void handle(Long l) {
     });
     await();
   }
+
+  @Repeat(times = 100)
+  @Test
+  public void testRaceWhenTimerCreatedOutsideEventLoop() {
+    int numThreads = 1000;
+    int numIter = 1;
+    Thread[] threads = new Thread[numThreads];
+    AtomicInteger count = new AtomicInteger(numIter * numThreads);
+    for (int i = 0;i < numThreads;i++) {
+      Thread th = new Thread(() -> {
+        // We need something more aggressive than a millisecond for this test
+        ((VertxImpl)vertx).scheduleTimeout(((VertxImpl) vertx).getOrCreateContext(), false, 1, TimeUnit.NANOSECONDS, ignore -> {
+          count.decrementAndGet();
+        });
+      });
+      th.start();
+      threads[i] = th;
+    }
+    waitUntil(() -> count.get() == 0);
+  }
 }