Repository: eclipse-vertx/vert.x

Problem statement:
Ensure EventBusMetrics message schedule event is called before the event-loop action
The `EventBusMetrics#scheduleMessage` method was changed to be called on the event-loop in order to avoid incorrect counter in metrics implementations (that needed to compensate and correct the counter value) when a message consumer is unregistered.

It is more interesting to have the event delivered on the thread scheduling the delivery action on the event-loop as it provides a better observation of the latency of a message dispatch after is has been received.

We should compensate a consumer unregistration with a discard message event so the metrics implementation will be aware of this and compensate the actual value.


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
@@ -523,13 +523,14 @@ private <T> void deliverToHandler(MessageImpl msg, HandlerHolder<T> holder) {
     // Each handler gets a fresh copy
     MessageImpl copied = msg.copyBeforeReceive();
     DeliveryContext<T> receiveContext = new InboundDeliveryContext<>(copied, holder);
+
+    if (metrics != null) {
+      metrics.scheduleMessage(holder.getHandler().getMetric(), msg.isLocal());
+    }
+
     holder.getContext().runOnContext((v) -> {
-      // Need to check handler is still there - the handler might have been removed after the message were sent but
-      // before it was received
       try {
-        if (!holder.isRemoved()) {
-          receiveContext.next();
-        }
+        receiveContext.next();
       } finally {
         if (holder.isReplyHandler()) {
           holder.getHandler().unregister();
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
@@ -16,7 +16,6 @@
 import io.vertx.core.eventbus.MessageConsumer;
 import io.vertx.core.eventbus.ReplyException;
 import io.vertx.core.eventbus.ReplyFailure;
-import io.vertx.core.eventbus.impl.clustered.ClusteredMessage;
 import io.vertx.core.impl.Arguments;
 import io.vertx.core.impl.ContextInternal;
 import io.vertx.core.logging.Logger;
@@ -98,7 +97,7 @@ public MessageConsumer<T> setMaxBufferedMessages(int maxBufferedMessages) {
     }
     for (Message<T> msg : discarded) {
       if (metrics != null) {
-        metrics.discardMessage(metric, isLocal(msg), msg);
+        metrics.discardMessage(metric, ((MessageImpl)msg).isLocal(), msg);
       }
       if (discardHandler != null) {
         discardHandler.handle(msg);
@@ -176,7 +175,7 @@ private void doUnregister(Handler<AsyncResult<Void>> doneHandler) {
       if (metrics != null) {
         Message<T> msg;
         while ((msg = pending.poll()) != null) {
-          metrics.discardMessage(metric, isLocal(msg), msg);
+          metrics.discardMessage(metric, ((MessageImpl)msg).isLocal(), msg);
         }
       } else {
         pending.clear();
@@ -224,21 +223,24 @@ public synchronized void setResult(AsyncResult<Void> result) {
 
   @Override
   public void handle(Message<T> message) {
+    boolean local = ((MessageImpl) message).isLocal();
     Handler<Message<T>> theHandler;
     ContextInternal ctx;
     synchronized (this) {
+      // Need to check handler is still there - the handler might have been removed after the message were sent but
+      // before it was received
       if (registered == null) {
+        if (metrics != null) {
+          metrics.discardMessage(metric, local, message);
+        }
         return;
       }
-      if (metrics != null) {
-        metrics.scheduleMessage(metric, isLocal(message));
-      }
       if (demand == 0L) {
         if (pending.size() < maxBufferedMessages) {
           pending.add(message);
         } else {
           if (metrics != null) {
-            metrics.discardMessage(metric, isLocal(message), message);
+            metrics.discardMessage(metric, local, message);
           }
           if (discardHandler != null) {
             discardHandler.handle(message);
@@ -271,7 +273,7 @@ private void deliver(Handler<Message<T>> theHandler, Message<T> message, Context
     }
     try {
       if (metrics != null) {
-        metrics.beginHandleMessage(metric, isLocal(message));
+        metrics.beginHandleMessage(metric, ((MessageImpl)message).isLocal());
       }
       theHandler.handle(message);
       if (metrics != null) {
@@ -391,16 +393,4 @@ public Handler<Message<T>> getHandler() {
   public Object getMetric() {
     return metric;
   }
-
-  private boolean isLocal(Message<?> message) {
-    boolean local = true;
-    if (message instanceof ClusteredMessage) {
-      // A bit hacky
-      ClusteredMessage cmsg = (ClusteredMessage)message;
-      if (cmsg.isFromWire()) {
-        return false;
-      }
-    }
-    return true;
-  }
 }


Test patch:
diff --git a/src/test/java/io/vertx/core/eventbus/EventBusRegistrationRaceTest.java b/src/test/java/io/vertx/core/eventbus/EventBusRegistrationRaceTest.java
--- a/src/test/java/io/vertx/core/eventbus/EventBusRegistrationRaceTest.java
+++ b/src/test/java/io/vertx/core/eventbus/EventBusRegistrationRaceTest.java
@@ -10,8 +10,11 @@
  */
 package io.vertx.core.eventbus;
 
-import io.vertx.core.eventbus.EventBus;
-import io.vertx.core.eventbus.MessageConsumer;
+import io.vertx.core.VertxOptions;
+import io.vertx.core.metrics.MetricsOptions;
+import io.vertx.core.spi.VertxMetricsFactory;
+import io.vertx.core.spi.metrics.EventBusMetrics;
+import io.vertx.core.spi.metrics.VertxMetrics;
 import io.vertx.test.core.VertxTestBase;
 import org.junit.Test;
 
@@ -32,6 +35,39 @@ public class EventBusRegistrationRaceTest extends VertxTestBase {
   private static final int NUM_MSG = 300_000;
   private static String TEST_ADDR = "the-addr";
 
+  private final AtomicInteger count = new AtomicInteger();
+
+  @Override
+  protected VertxOptions getOptions() {
+    VertxOptions options = super.getOptions();
+    options.setMetricsOptions(new MetricsOptions().setEnabled(true).setFactory(new VertxMetricsFactory() {
+      @Override
+      public VertxMetrics metrics(VertxOptions options) {
+        return new VertxMetrics() {
+          @Override
+          public EventBusMetrics<Void> createEventBusMetrics() {
+            return new EventBusMetrics<Void>() {
+              @Override
+              public void scheduleMessage(Void handler, boolean local) {
+                count.incrementAndGet();
+              }
+
+              @Override
+              public void beginHandleMessage(Void handler, boolean local) {
+                count.decrementAndGet();
+              }
+              @Override
+              public void discardMessage(Void handler, boolean local, Message<?> msg) {
+                count.decrementAndGet();
+              }
+            };
+          }
+        };
+      }
+    }));
+    return options;
+  }
+
   @Test
   public void theTest() throws Exception {
     AtomicInteger seq = new AtomicInteger();
@@ -43,6 +79,7 @@ public void theTest() throws Exception {
     threadB.start();
     threadA.join(20 * 1000);
     threadB.join(20 * 1000);
+    assertEquals(0, count.get());
   }
 
   private void threadA(AtomicInteger seq) {
diff --git a/src/test/java/io/vertx/core/spi/metrics/MetricsContextTest.java b/src/test/java/io/vertx/core/spi/metrics/MetricsContextTest.java
--- a/src/test/java/io/vertx/core/spi/metrics/MetricsContextTest.java
+++ b/src/test/java/io/vertx/core/spi/metrics/MetricsContextTest.java
@@ -795,6 +795,7 @@ public void testMessageHandlerEventLoop() {
   }
 
   private void testMessageHandler(BiConsumer<Vertx, Handler<Void>> runOnContext, BiConsumer<Thread, Context> checker) {
+    AtomicReference<Thread> scheduleThread = new AtomicReference<>();
     AtomicReference<Thread> consumerThread = new AtomicReference<>();
     AtomicReference<Context> consumerContext = new AtomicReference<>();
     AtomicBoolean registeredCalled = new AtomicBoolean();
@@ -819,6 +820,10 @@ public void handlerUnregistered(Void handler) {
             unregisteredCalled.set(true);
           }
           @Override
+          public void scheduleMessage(Void handler, boolean local) {
+            scheduleThread.set(Thread.currentThread());
+          }
+          @Override
           public void beginHandleMessage(Void handler, boolean local) {
             consumerThread.set(Thread.currentThread());
             consumerContext.set(Vertx.currentContext());
@@ -834,6 +839,9 @@ public void endHandleMessage(Void handler, Throwable failure) {
     };
     Vertx vertx = vertx(new VertxOptions().setMetricsOptions(new MetricsOptions().setEnabled(true).setFactory(factory)));
     EventBus eb = vertx.eventBus();
+    Thread t = new Thread(() -> {
+      eb.send("the_address", "the_msg");
+    });
     runOnContext.accept(vertx, v -> {
       MessageConsumer<Object> consumer = eb.consumer("the_address");
       consumer.handler(msg -> {
@@ -841,6 +849,7 @@ public void endHandleMessage(Void handler, Throwable failure) {
         executeInVanillaThread(() -> {
           vertx.getOrCreateContext().runOnContext(v2 -> {
             consumer.unregister(onSuccess(v3 -> {
+              assertSame(t, scheduleThread.get());
               assertTrue(registeredCalled.get());
               assertTrue(beginHandleCalled.get());
               assertTrue(endHandleCalled.get());
@@ -850,7 +859,7 @@ public void endHandleMessage(Void handler, Throwable failure) {
           });
         });
       }).completionHandler(onSuccess(v2 -> {
-        eb.send("the_address", "the_msg");
+        t.start();
       }));
     });
     await();