Repository: eclipse-vertx/vert.x

Problem statement:
Possible concurrency issue in Event Bus
It looks like, there is possible concurrency issue in Event Bus, when we're adding and removing local registrations.

**addLocalRegistration**: https://github.com/eclipse/vert.x/blob/5338f6353a49a4e7e9a6350e4878bcc5f11028a0/src/main/java/io/vertx/core/eventbus/impl/EventBusImpl.java#L232
**removeLocalRegistration**: https://github.com/eclipse/vert.x/blob/5338f6353a49a4e7e9a6350e4878bcc5f11028a0/src/main/java/io/vertx/core/eventbus/impl/EventBusImpl.java#L277

There is very very little possibility (but non 0), that when we're removing last handler in one thread and adding the new one for the same address in another, we can lost handlers list for that address.
Is this real issue or not, because of it's very little possibility, or I'm missing some additional synchronization?

Topic in user group: https://groups.google.com/forum/#!topic/vertx/T3F3CuG1n3w

Possible concurrency issue in Event Bus
It looks like, there is possible concurrency issue in Event Bus, when we're adding and removing local registrations.

**addLocalRegistration**: https://github.com/eclipse/vert.x/blob/5338f6353a49a4e7e9a6350e4878bcc5f11028a0/src/main/java/io/vertx/core/eventbus/impl/EventBusImpl.java#L232
**removeLocalRegistration**: https://github.com/eclipse/vert.x/blob/5338f6353a49a4e7e9a6350e4878bcc5f11028a0/src/main/java/io/vertx/core/eventbus/impl/EventBusImpl.java#L277

There is very very little possibility (but non 0), that when we're removing last handler in one thread and adding the new one for the same address in another, we can lost handlers list for that address.
Is this real issue or not, because of it's very little possibility, or I'm missing some additional synchronization?

Topic in user group: https://groups.google.com/forum/#!topic/vertx/T3F3CuG1n3w



Hints:

Patch:
diff --git a/src/main/java/io/vertx/core/eventbus/impl/EventBusImpl.java b/src/main/java/io/vertx/core/eventbus/impl/EventBusImpl.java
--- a/src/main/java/io/vertx/core/eventbus/impl/EventBusImpl.java
+++ b/src/main/java/io/vertx/core/eventbus/impl/EventBusImpl.java
@@ -28,6 +28,7 @@
 import io.vertx.core.eventbus.ReplyFailure;
 import io.vertx.core.eventbus.SendContext;
 import io.vertx.core.impl.VertxInternal;
+import io.vertx.core.impl.utils.ConcurrentCyclicSequence;
 import io.vertx.core.logging.Logger;
 import io.vertx.core.logging.LoggerFactory;
 import io.vertx.core.spi.metrics.EventBusMetrics;
@@ -55,7 +56,7 @@ public class EventBusImpl implements EventBus, MetricsProvider {
   private final AtomicLong replySequence = new AtomicLong(0);
   protected final VertxInternal vertx;
   protected final EventBusMetrics metrics;
-  protected final ConcurrentMap<String, Handlers> handlerMap = new ConcurrentHashMap<>();
+  protected final ConcurrentMap<String, ConcurrentCyclicSequence<HandlerHolder>> handlerMap = new ConcurrentHashMap<>();
   protected final CodecManager codecManager = new CodecManager();
   protected volatile boolean started;
 
@@ -227,11 +228,12 @@ protected MessageImpl createMessage(boolean send, String address, MultiMap heade
     return msg;
   }
 
-  protected <T> void addRegistration(String address, HandlerRegistration<T> registration,
+  protected <T> HandlerHolder<T> addRegistration(String address, HandlerRegistration<T> registration,
                                      boolean replyHandler, boolean localOnly) {
     Objects.requireNonNull(registration.getHandler(), "handler");
-    boolean newAddress = addLocalRegistration(address, registration, replyHandler, localOnly);
-    addRegistration(newAddress, address, replyHandler, localOnly, registration::setResult);
+    LocalRegistrationResult<T> result = addLocalRegistration(address, registration, replyHandler, localOnly);
+    addRegistration(result.newAddress, address, replyHandler, localOnly, registration::setResult);
+    return result.holder;
   }
 
   protected <T> void addRegistration(boolean newAddress, String address,
@@ -240,8 +242,17 @@ protected <T> void addRegistration(boolean newAddress, String address,
     completionHandler.handle(Future.succeededFuture());
   }
 
-  protected <T> boolean addLocalRegistration(String address, HandlerRegistration<T> registration,
-                                             boolean replyHandler, boolean localOnly) {
+  private static class LocalRegistrationResult<T> {
+    final HandlerHolder<T> holder;
+    final boolean newAddress;
+    LocalRegistrationResult(HandlerHolder<T> holder, boolean newAddress) {
+      this.holder = holder;
+      this.newAddress = newAddress;
+    }
+  }
+
+  private <T> LocalRegistrationResult<T> addLocalRegistration(String address, HandlerRegistration<T> registration,
+                                                           boolean replyHandler, boolean localOnly) {
     Objects.requireNonNull(address, "address");
 
     Context context = Vertx.currentContext();
@@ -252,63 +263,46 @@ protected <T> boolean addLocalRegistration(String address, HandlerRegistration<T
     }
     registration.setHandlerContext(context);
 
-    boolean newAddress = false;
-
-    HandlerHolder holder = new HandlerHolder<>(metrics, registration, replyHandler, localOnly, context);
+    HandlerHolder<T> holder = new HandlerHolder<>(metrics, registration, replyHandler, localOnly, context);
 
-    Handlers handlers = handlerMap.get(address);
-    if (handlers == null) {
-      handlers = new Handlers();
-      Handlers prevHandlers = handlerMap.putIfAbsent(address, handlers);
-      if (prevHandlers != null) {
-        handlers = prevHandlers;
-      }
-      newAddress = true;
-    }
-    handlers.list.add(holder);
+    ConcurrentCyclicSequence<HandlerHolder> handlers = new ConcurrentCyclicSequence<HandlerHolder>().add(holder);
+    ConcurrentCyclicSequence<HandlerHolder> actualHandlers = handlerMap.merge(
+      address,
+      handlers,
+      (old, prev) -> old.add(prev.first()));
 
     if (hasContext) {
       HandlerEntry entry = new HandlerEntry<>(address, registration);
       context.addCloseHook(entry);
     }
 
-    return newAddress;
+    boolean newAddress = handlers == actualHandlers;
+    return new LocalRegistrationResult<>(holder, newAddress);
   }
 
-  protected <T> void removeRegistration(String address, HandlerRegistration<T> handler, Handler<AsyncResult<Void>> completionHandler) {
-    HandlerHolder holder = removeLocalRegistration(address, handler);
-    removeRegistration(holder, address, completionHandler);
+  protected <T> void removeRegistration(HandlerHolder<T> holder, Handler<AsyncResult<Void>> completionHandler) {
+    boolean last = removeLocalRegistration(holder);
+    removeRegistration(last ? holder : null, holder.getHandler().address(), completionHandler);
   }
 
-  protected <T> void removeRegistration(HandlerHolder handlerHolder, String address,
+  protected <T> void removeRegistration(HandlerHolder<T> handlerHolder, String address,
                                         Handler<AsyncResult<Void>> completionHandler) {
     callCompletionHandlerAsync(completionHandler);
   }
 
-  protected <T> HandlerHolder removeLocalRegistration(String address, HandlerRegistration<T> handler) {
-    Handlers handlers = handlerMap.get(address);
-    HandlerHolder lastHolder = null;
-    if (handlers != null) {
-      synchronized (handlers) {
-        int size = handlers.list.size();
-        // Requires a list traversal. This is tricky to optimise since we can't use a set since
-        // we need fast ordered traversal for the round robin
-        for (int i = 0; i < size; i++) {
-          HandlerHolder holder = handlers.list.get(i);
-          if (holder.getHandler() == handler) {
-            handlers.list.remove(i);
-            holder.setRemoved();
-            if (handlers.list.isEmpty()) {
-              handlerMap.remove(address);
-              lastHolder = holder;
-            }
-            holder.getContext().removeCloseHook(new HandlerEntry<>(address, holder.getHandler()));
-            break;
-          }
-        }
+  private <T> boolean removeLocalRegistration(HandlerHolder<T> holder) {
+    String address = holder.getHandler().address();
+    boolean last = handlerMap.compute(address, (key, val) -> {
+      if (val == null) {
+        return null;
       }
+      ConcurrentCyclicSequence<HandlerHolder> next = val.remove(holder);
+      return next.size() == 0 ? null : next;
+    }) == null;
+    if (holder.setRemoved()) {
+      holder.getContext().removeCloseHook(new HandlerEntry<>(address, holder.getHandler()));
     }
-    return lastHolder;
+    return last;
   }
 
   protected <T> void sendReply(MessageImpl replyMessage, MessageImpl replierMessage, DeliveryOptions options,
@@ -377,11 +371,11 @@ protected boolean isMessageLocal(MessageImpl msg) {
 
   protected <T> boolean deliverMessageLocally(MessageImpl msg) {
     msg.setBus(this);
-    Handlers handlers = handlerMap.get(msg.address());
+    ConcurrentCyclicSequence<HandlerHolder> handlers = handlerMap.get(msg.address());
     if (handlers != null) {
       if (msg.isSend()) {
         //Choose one
-        HandlerHolder holder = handlers.choose();
+        HandlerHolder holder = handlers.next();
         if (metrics != null) {
           metrics.messageReceived(msg.address(), !msg.isSend(), isMessageLocal(msg), holder != null ? 1 : 0);
         }
@@ -391,9 +385,9 @@ protected <T> boolean deliverMessageLocally(MessageImpl msg) {
       } else {
         // Publish
         if (metrics != null) {
-          metrics.messageReceived(msg.address(), !msg.isSend(), isMessageLocal(msg), handlers.list.size());
+          metrics.messageReceived(msg.address(), !msg.isSend(), isMessageLocal(msg), handlers.size());
         }
-        for (HandlerHolder holder: handlers.list) {
+        for (HandlerHolder holder: handlers) {
           deliverToHandler(msg, holder);
         }
       }
@@ -509,8 +503,8 @@ public void next() {
 
   private void unregisterAll() {
     // Unregister all handlers explicitly - don't rely on context hooks
-    for (Handlers handlers: handlerMap.values()) {
-      for (HandlerHolder holder: handlers.list) {
+    for (ConcurrentCyclicSequence<HandlerHolder> handlers: handlerMap.values()) {
+      for (HandlerHolder holder: handlers) {
         holder.getHandler().unregister();
       }
     }
diff --git a/src/main/java/io/vertx/core/eventbus/impl/HandlerHolder.java b/src/main/java/io/vertx/core/eventbus/impl/HandlerHolder.java
--- a/src/main/java/io/vertx/core/eventbus/impl/HandlerHolder.java
+++ b/src/main/java/io/vertx/core/eventbus/impl/HandlerHolder.java
@@ -36,17 +36,18 @@ public HandlerHolder(EventBusMetrics metrics, HandlerRegistration<T> handler, bo
   }
 
   // We use a synchronized block to protect removed as it can be unregistered from a different thread
-  public void setRemoved() {
-    boolean unregisterMetric = false;
+  public boolean setRemoved() {
+    boolean unregistered = false;
     synchronized (this) {
       if (!removed) {
         removed = true;
-        unregisterMetric = true;
+        unregistered = true;
       }
     }
-    if (metrics != null && unregisterMetric) {
+    if (metrics != null && unregistered) {
       metrics.handlerUnregistered(handler.getMetric());
     }
+    return unregistered;
   }
 
   // Because of biased locks the overhead of the synchronized lock should be very low as it's almost always
diff --git a/src/main/java/io/vertx/core/eventbus/impl/HandlerRegistration.java b/src/main/java/io/vertx/core/eventbus/impl/HandlerRegistration.java
--- a/src/main/java/io/vertx/core/eventbus/impl/HandlerRegistration.java
+++ b/src/main/java/io/vertx/core/eventbus/impl/HandlerRegistration.java
@@ -48,7 +48,7 @@ public class HandlerRegistration<T> implements MessageConsumer<T>, Handler<Messa
   private final boolean localOnly;
   private final Handler<AsyncResult<Message<T>>> asyncResultHandler;
   private long timeoutID = -1;
-  private boolean registered;
+  private HandlerHolder<T> registered;
   private Handler<Message<T>> handler;
   private Context handlerContext;
   private AsyncResult<Void> result;
@@ -141,9 +141,10 @@ private void doUnregister(Handler<AsyncResult<Void>> completionHandler) {
         }
       };
     }
-    if (registered) {
-      registered = false;
-      eventBus.removeRegistration(address, this, completionHandler);
+    HandlerHolder<T> holder = registered;
+    if (holder != null) {
+      registered = null;
+      eventBus.removeRegistration(holder, completionHandler);
     } else {
       callCompletionHandlerAsync(completionHandler);
     }
@@ -260,10 +261,9 @@ public synchronized void discardHandler(Handler<Message<T>> handler) {
   @Override
   public synchronized MessageConsumer<T> handler(Handler<Message<T>> handler) {
     this.handler = handler;
-    if (this.handler != null && !registered) {
-      registered = true;
-      eventBus.addRegistration(address, this, repliedAddress != null, localOnly);
-    } else if (this.handler == null && registered) {
+    if (this.handler != null && registered == null) {
+      registered = eventBus.addRegistration(address, this, repliedAddress != null, localOnly);
+    } else if (this.handler == null && registered != null) {
       // This will set registered to false
       this.unregister();
     }
@@ -277,7 +277,7 @@ public ReadStream<T> bodyStream() {
 
   @Override
   public synchronized boolean isRegistered() {
-    return registered;
+    return registered != null;
   }
 
   @Override
diff --git a/src/main/java/io/vertx/core/eventbus/impl/Handlers.java b/src/main/java/io/vertx/core/eventbus/impl/Handlers.java
deleted file mode 100644
--- a/src/main/java/io/vertx/core/eventbus/impl/Handlers.java
+++ /dev/null
@@ -1,45 +0,0 @@
-/*
- * Copyright (c) 2011-2017 Contributors to the Eclipse Foundation
- *
- * This program and the accompanying materials are made available under the
- * terms of the Eclipse Public License 2.0 which is available at
- * http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
- * which is available at https://www.apache.org/licenses/LICENSE-2.0.
- *
- * SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
- */
-
-package io.vertx.core.eventbus.impl;
-
-import java.util.List;
-import java.util.concurrent.CopyOnWriteArrayList;
-import java.util.concurrent.atomic.AtomicInteger;
-
-/**
- * @author <a href="http://tfox.org">Tim Fox</a>
- */
-public class Handlers {
-
-  private final AtomicInteger pos = new AtomicInteger(0);
-  public final List<HandlerHolder> list = new CopyOnWriteArrayList<>();
-
-  public HandlerHolder choose() {
-    while (true) {
-      int size = list.size();
-      if (size == 0) {
-        return null;
-      }
-      int p = pos.getAndIncrement();
-      if (p >= size - 1) {
-        pos.set(0);
-      }
-      try {
-        return list.get(p);
-      } catch (IndexOutOfBoundsException e) {
-        // Can happen
-        pos.set(0);
-      }
-    }
-  }
-}
-
diff --git a/src/main/java/io/vertx/core/eventbus/impl/clustered/ClusteredEventBus.java b/src/main/java/io/vertx/core/eventbus/impl/clustered/ClusteredEventBus.java
--- a/src/main/java/io/vertx/core/eventbus/impl/clustered/ClusteredEventBus.java
+++ b/src/main/java/io/vertx/core/eventbus/impl/clustered/ClusteredEventBus.java
@@ -215,7 +215,7 @@ protected <T> void addRegistration(boolean newAddress, String address,
   }
 
   @Override
-  protected <T> void removeRegistration(HandlerHolder lastHolder, String address,
+  protected <T> void removeRegistration(HandlerHolder<T> lastHolder, String address,
                                         Handler<AsyncResult<Void>> completionHandler) {
     if (lastHolder != null && subs != null && !lastHolder.isLocalOnly()) {
       ownSubs.remove(address);
diff --git a/src/main/java/io/vertx/core/impl/utils/ConcurrentCyclicSequence.java b/src/main/java/io/vertx/core/impl/utils/ConcurrentCyclicSequence.java
new file mode 100644
--- /dev/null
+++ b/src/main/java/io/vertx/core/impl/utils/ConcurrentCyclicSequence.java
@@ -0,0 +1,151 @@
+/*
+ * Copyright (c) 2011-2018 Contributors to the Eclipse Foundation
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
+ * which is available at https://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
+ */
+
+package io.vertx.core.impl.utils;
+
+import java.util.*;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * A concurrent cyclic sequence of elements that can be used for round robin.
+ * <p/>
+ * The sequence is immutable and modifications are done with copy-on-write using
+ * {@link #add(Object)} and {@link #remove(Object)} to return a modified copy of the current instance.
+ * <p/>
+ * The internal counter uses a volatile index, so it can be incremented concurrently by several
+ * threads without locking.
+ *
+ * @author <a href="http://tfox.org">Tim Fox</a>
+ * @author <a href="mailto:julien@julienviet.com">Julien Viet</a>
+ */
+public class ConcurrentCyclicSequence<T> implements Iterable<T>, Iterator<T> {
+
+  private static final Object[] EMPTY_ARRAY = new Object[0];
+
+  private final AtomicInteger pos;
+  private final Object[] elements;
+
+  /**
+   * Create a new empty sequence.
+   */
+  public ConcurrentCyclicSequence() {
+    this(0, EMPTY_ARRAY);
+  }
+
+  /**
+   * Create a new empty sequence.
+   */
+  @SafeVarargs
+  public ConcurrentCyclicSequence(T... elements) {
+    this(0, Arrays.copyOf(elements.clone(), elements.length, Object[].class));
+  }
+
+  private ConcurrentCyclicSequence(int pos, Object[] elements) {
+    this.pos = new AtomicInteger(pos);
+    this.elements = elements;
+  }
+
+  /**
+   * @return the current index
+   */
+  public int index() {
+    return elements.length > 0 ? pos.get() % elements.length : 0;
+  }
+
+  /**
+   * @return the first element or {@code null} when the sequence is empty
+   */
+  @SuppressWarnings("unchecked")
+  public T first() {
+    return (T) (elements.length > 0 ? elements[0] : null);
+  }
+
+  /**
+   * Copy the current sequence, add {@code element} at the tail of this sequence and returns it.
+   * @param element the element to add
+   * @return the resulting sequence
+   */
+  public ConcurrentCyclicSequence<T> add(T element) {
+    int len = elements.length;
+    Object[] copy = Arrays.copyOf(elements, len + 1);
+    copy[len] = element;
+    return new ConcurrentCyclicSequence<>(pos.get(), copy);
+  }
+
+  /**
+   * Remove the first occurrence of {@code element} in this sequence and returns it.
+   * <p/>
+   * If the sequence does not contains {@code element}, this instance is returned instead.
+   *
+   * @param element the element to remove
+   * @return the resulting sequence
+   */
+  public ConcurrentCyclicSequence<T> remove(T element) {
+    int len = elements.length;
+    for (int i = 0;i < len;i++) {
+      if (Objects.equals(element, elements[i])) {
+        if (len > 1) {
+          Object[] copy = new Object[len - 1];
+          System.arraycopy(elements,0, copy, 0, i);
+          System.arraycopy(elements, i + 1, copy, i, len - i - 1);
+          return new ConcurrentCyclicSequence<>(pos.get() % copy.length, copy);
+        } else {
+          return new ConcurrentCyclicSequence<>();
+        }
+      }
+    }
+    return this;
+  }
+
+  /**
+   * @return always {@code true}
+   */
+  @Override
+  public boolean hasNext() {
+    return true;
+  }
+
+  /**
+   * @return the next element in the sequence
+   */
+  @SuppressWarnings("unchecked")
+  @Override
+  public T next() {
+    int len = elements.length;
+    switch (len) {
+      case 0:
+        return null;
+      case 1:
+        return (T) elements[0];
+      default:
+        int p;
+        p = pos.getAndIncrement();
+        return (T) elements[Math.abs(p % len)];
+    }
+  }
+
+  /**
+   * @return the size of this sequence
+   */
+  public int size() {
+    return elements.length;
+  }
+
+  /**
+   * @return an iterator starting at the first element of the sequence, the iterator will not throw {@link ConcurrentModificationException}
+   */
+  @Override
+  @SuppressWarnings("unchecked")
+  public Iterator<T> iterator() {
+    return Arrays.asList((T[]) elements).iterator();
+  }
+}
+


Test patch:
diff --git a/src/test/benchmarks/io/vertx/benchmarks/ConcurrentCyclicSequenceBenchmark.java b/src/test/benchmarks/io/vertx/benchmarks/ConcurrentCyclicSequenceBenchmark.java
new file mode 100644
--- /dev/null
+++ b/src/test/benchmarks/io/vertx/benchmarks/ConcurrentCyclicSequenceBenchmark.java
@@ -0,0 +1,65 @@
+/*
+ * Copyright (c) 2011-2018 Contributors to the Eclipse Foundation
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
+ * which is available at https://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
+ */
+package io.vertx.benchmarks;
+
+import io.vertx.core.impl.utils.ConcurrentCyclicSequence;
+import org.openjdk.jmh.annotations.*;
+
+import java.util.stream.IntStream;
+
+@State(Scope.Benchmark)
+@Threads(8)
+public class ConcurrentCyclicSequenceBenchmark extends BenchmarkBase {
+
+  private ConcurrentCyclicSequence<String> seq1;
+  private ConcurrentCyclicSequence<String> seq2;
+  private ConcurrentCyclicSequence<String> seq4;
+  private ConcurrentCyclicSequence<String> seq8;
+  private ConcurrentCyclicSequence<String> seq16;
+
+  private static ConcurrentCyclicSequence<String> gen(int size) {
+    return new ConcurrentCyclicSequence<>(IntStream.range(0, size + 1).mapToObj(i -> "" + i).toArray(String[]::new));
+  }
+
+  @Setup
+  public void setup() {
+    seq1 = gen(1);
+    seq2 = gen(2);
+    seq4 = gen(4);
+    seq8 = gen(8);
+    seq16 = gen(16);
+  }
+
+  @Benchmark
+  public String size1() {
+    return seq1.next();
+  }
+
+  @Benchmark
+  public String size2() {
+    return seq2.next();
+  }
+
+  @Benchmark
+  public String size4() {
+    return seq4.next();
+  }
+
+  @Benchmark
+  public String size8() {
+    return seq8.next();
+  }
+
+  @Benchmark
+  public String size16() {
+    return seq16.next();
+  }
+}
diff --git a/src/test/java/io/vertx/test/core/EventBusRegistrationRaceTest.java b/src/test/java/io/vertx/test/core/EventBusRegistrationRaceTest.java
new file mode 100644
--- /dev/null
+++ b/src/test/java/io/vertx/test/core/EventBusRegistrationRaceTest.java
@@ -0,0 +1,82 @@
+/*
+ * Copyright (c) 2011-2018 Contributors to the Eclipse Foundation
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
+ * which is available at https://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
+ */
+package io.vertx.test.core;
+
+import io.vertx.core.eventbus.EventBus;
+import io.vertx.core.eventbus.MessageConsumer;
+import org.junit.Test;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * This test checks race conditions in EventBus registration and unregistration.
+ *
+ * We use two separate threads to register and unregister consumers on the same address.
+ *
+ * Thread are coordinated by an volatile sequence and we check at each iteration that Thread-A
+ * registration / un-registration of a consumer does not prevent the Thread-B registration
+ * to receive the message it sends to its consumer, since both threads will access the same
+ * handlers list concurrently.
+ */
+public class EventBusRegistrationRaceTest extends VertxTestBase {
+
+  private static final int NUM_MSG = 300_000;
+  private static String TEST_ADDR = "the-addr";
+
+  @Test
+  public void theTest() throws Exception {
+    AtomicInteger seq = new AtomicInteger();
+    Thread threadA = new Thread(() -> threadA(seq));
+    threadA.setName("Thread-A");
+    Thread threadB = new Thread(() -> threadB(seq));
+    threadB.setName("Thread-B");
+    threadA.start();
+    threadB.start();
+    threadA.join(20 * 1000);
+    threadB.join(20 * 1000);
+  }
+
+  private void threadA(AtomicInteger seq) {
+    EventBus eventBus = vertx.eventBus();
+    int count = 0;
+    while (count < NUM_MSG) {
+      while (count > seq.get()) {
+        Thread.yield();
+      }
+      count++;
+      MessageConsumer<Object> consumer = eventBus.consumer(TEST_ADDR, msg -> { });
+      consumer.unregister();
+    }
+  }
+
+  private void threadB(AtomicInteger seq) {
+    EventBus eventBus = vertx.eventBus();
+    MessageConsumer<Object> consumer = null;
+    int count = 0;
+    while (count < NUM_MSG) {
+      while (count > seq.get()) {
+        Thread.yield();
+      }
+      count++;
+      if (consumer != null) {
+        consumer.unregister();
+      }
+      consumer = eventBus.consumer(TEST_ADDR);
+      consumer.handler(event -> {
+        // Missing a message prevents the progression of the test
+        seq.incrementAndGet();
+      });
+      // We use publish because send might deliver the message to a Thread-A's consumer
+      // so we are sure we always get a message
+      eventBus.publish(TEST_ADDR, count);
+    }
+  }
+}
diff --git a/src/test/java/io/vertx/test/core/utils/ConcurrentCyclicSequenceTest.java b/src/test/java/io/vertx/test/core/utils/ConcurrentCyclicSequenceTest.java
new file mode 100644
--- /dev/null
+++ b/src/test/java/io/vertx/test/core/utils/ConcurrentCyclicSequenceTest.java
@@ -0,0 +1,116 @@
+/*
+ * Copyright (c) 2011-2018 Contributors to the Eclipse Foundation
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
+ * which is available at https://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
+ */
+package io.vertx.test.core.utils;
+
+import io.vertx.core.impl.utils.ConcurrentCyclicSequence;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class ConcurrentCyclicSequenceTest {
+
+  @Test
+  public void testEmpty() {
+    ConcurrentCyclicSequence<String> empty = new ConcurrentCyclicSequence<>();
+    for (int j = 0;j < 1;j++) {
+      for (int i = 0;i < 3;i++) {
+        assertEquals(0, empty.size());
+        assertEquals(0, empty.index());
+        assertEquals(null, empty.first());
+        empty.next();
+      }
+      empty = empty.remove("does-not-exist");
+    }
+  }
+
+  @Test
+  public void testAdd() {
+    ConcurrentCyclicSequence<String> seq = new ConcurrentCyclicSequence<String>().add("s1");
+    assertEquals(Collections.singletonList("s1"), toList(seq));
+    assertEquals(Arrays.asList("s1", "s2"), toList(seq.add("s2")));
+    assertEquals(Collections.singletonList("s1"), toList(seq));
+  }
+
+  @Test
+  public void testRemove() {
+    ConcurrentCyclicSequence<String> seq = new ConcurrentCyclicSequence<String>().add("s1").add("s2").add("s1").add("s2");
+    assertEquals(Arrays.asList("s1", "s2", "s1", "s2"), toList(seq));
+    assertEquals(Arrays.asList("s1", "s1", "s2"), toList(seq.remove("s2")));
+    assertEquals(Arrays.asList("s1", "s1"), toList(seq.remove("s2").remove("s2")));
+    assertEquals(Arrays.asList("s2", "s1", "s2"), toList(seq.remove("s1")));
+    assertEquals(Arrays.asList("s2", "s2"), toList(seq.remove("s1").remove("s1")));
+    assertEquals(Arrays.asList("s1", "s2"), toList(seq.remove("s1").remove("s2")));
+    assertEquals(Collections.emptyList(), toList(seq.remove("s1").remove("s2").remove("s1").remove("s2")));
+    assertEquals(Arrays.asList("s1", "s2", "s1", "s2"), toList(seq));
+  }
+
+  @Test
+  public void testNullElement() {
+    ConcurrentCyclicSequence<String> seq = new ConcurrentCyclicSequence<>("s1", null, "s2", null);
+    assertEquals(Arrays.asList("s1", null, "s2", null), toList(seq));
+    assertEquals(Arrays.asList("s1", "s2", null), toList(seq.remove(null)));
+  }
+
+  @Test
+  public void testRoundRobin() throws Exception {
+    int iter = 1_000_000;
+    int range = 10;
+    ConcurrentCyclicSequence<AtomicInteger> tmp = new ConcurrentCyclicSequence<>();
+    for (int i = 0; i < range; i++) {
+      tmp = tmp.add(new AtomicInteger());
+    }
+    ConcurrentCyclicSequence<AtomicInteger> handlers = tmp;
+    AtomicBoolean failed = new AtomicBoolean();
+    int numThreads = 10;
+    Thread[] threads = new Thread[numThreads];
+    for (int i = 0;i < numThreads;i++) {
+      threads[i] = new Thread(() -> {
+        try {
+          for (int j = 0;j < iter;j++) {
+            handlers.next().incrementAndGet();
+          }
+        } catch (Exception e) {
+          e.printStackTrace();
+          failed.set(true);
+        }
+      });
+    }
+    for (Thread thread : threads) {
+      thread.start();
+    }
+    for (Thread thread : threads) {
+      thread.join();
+    }
+    for (AtomicInteger i : handlers) {
+      assertEquals(iter, i.get());
+    }
+    assertFalse(failed.get());
+    int pos = handlers.index();
+    assertTrue("Incorrect pos value " + pos, pos <= range);
+  }
+
+  private static <T> List<T> toList(ConcurrentCyclicSequence<T> seq) {
+    ArrayList<T> ret = new ArrayList<>();
+    for (T elt : seq) {
+      ret.add(elt);
+    }
+    return ret;
+  }
+}