Optimize service RPC delivery by preferring local handlers

continuum-core-vertx/src/main/java/org/kinotic/continuum/internal/core/api/event/DefaultEventBusService.java

@@ -56,6 +56,8 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
  * Default implementation of {@link EventBusService} using the vertx {@link io.vertx.core.eventbus.EventBus} as a backend
@@ -76,6 +78,15 @@ public class DefaultEventBusService implements EventBusService {
     private IgniteCache<String, Set<IgniteRegistrationInfo>> subscriptionsCache;
     @Autowired
     private Vertx vertx;
+    /**
+     * Reference counted set of event bus addresses this node hosts a local handler for.
+     * An entry is added by {@link #_listen} when a consumer's handler is attached on this node and removed
+     * when that consumer is unregistered, so membership means "a local consumer exists at this address".
+     * This is local by construction: {@link #_listen} only ever registers consumers on this node, other nodes'
+     * registrations live solely in the cluster's subscription registry. The send path uses this to prefer
+     * local delivery for service RPCs and avoid an unnecessary cluster hop.
+     */
+    private final Map<String, Integer> localListenerCounts = new ConcurrentHashMap<>();
 
     @PostConstruct
     public void init(){
@@ -111,7 +122,7 @@ public Mono<Flux<Event<byte[]>>> listenWithAck(String cri) {
 
         return Mono.create(sink -> {
             final MessageConsumer<byte[]> consumer = vertx.eventBus().consumer(cri);
-            final ConnectableFlux<Event<byte[]>> flux = _listen(null, consumer).publish();
+            final ConnectableFlux<Event<byte[]>> flux = _listen(cri, consumer).publish();
             consumer.completion().onComplete(ar ->{
                 if(ar.succeeded()){
                     sink.success(flux);
@@ -185,8 +196,9 @@ public Flux<ListenerStatus> monitorListenerStatus(String cri) {
 
     @Override
     public void send(Event<byte[]> event) {
-        DeliveryOptions deliveryOptions = createDeliveryOptions(event);
-        vertx.eventBus().send(event.cri().baseResource(),
+        String baseResource = event.cri().baseResource();
+        DeliveryOptions deliveryOptions = createDeliveryOptions(event, baseResource);
+        vertx.eventBus().send(baseResource,
                               event.data(),
                               deliveryOptions);
     }
@@ -195,10 +207,11 @@ public void send(Event<byte[]> event) {
     public Mono<Void> sendWithAck(Event<byte[]> event) {
         Validate.notNull(event, "Event must not be null");
         return Mono.create(sink -> {
-            DeliveryOptions deliveryOptions = createDeliveryOptions(event);
+            String baseResource = event.cri().baseResource();
+            DeliveryOptions deliveryOptions = createDeliveryOptions(event, baseResource);
             // We expect that a response will be sent upon receipt. This will happen automatically if the listener is created with this class.
             vertx.eventBus()
-                 .request(event.cri().baseResource(),
+                 .request(baseResource,
                           event.data(),
                           deliveryOptions)
                  .onComplete(reply -> {
@@ -220,8 +233,20 @@ private Flux<Event<byte[]>> _listen(String cri, MessageConsumer<byte[]> vertxEve
         }
 
         Flux<Event<byte[]>> ret = Flux.create(fluxSink -> {
+            // Record that this node now hosts a local handler for cri. Done here, where the consumer's handler is
+            // attached, and undone in onDispose, where the consumer is unregistered, so the count is coupled to the
+            // consumer's actual registration lifecycle on this node.
+            localListenerCounts.merge(cri, 1, Integer::sum);
+            final AtomicBoolean unregistered = new AtomicBoolean(false);
+
             // Setup all required handlers that are needed prior to consuming messages
-            fluxSink.onDispose(consumer::unregister);
+            fluxSink.onDispose(() -> {
+                // Decrement exactly once on the first unregister, removing the entry when no local handlers remain.
+                if(unregistered.compareAndSet(false, true)){
+                    localListenerCounts.computeIfPresent(cri, (k, n) -> n == 1 ? null : n - 1);
+                }
+                consumer.unregister();
+            });
 
             // TODO: deal with back pressure properly.. ?
             //fluxSink.onRequest()
@@ -245,7 +270,7 @@ private Flux<Event<byte[]>> _listen(String cri, MessageConsumer<byte[]> vertxEve
         return ret; // ensure message delivery happens on vertx event loop, not sure but this by itself did not move the next above to the work loop
     }
 
-    private DeliveryOptions createDeliveryOptions(Event<?> event){
+    private DeliveryOptions createDeliveryOptions(Event<?> event, String baseResource){
         DeliveryOptions deliveryOptions = new DeliveryOptions();
         deliveryOptions.setTracingPolicy(TracingPolicy.IGNORE);
         // fast path for MultiMapMetadataAdapter's
@@ -257,7 +282,29 @@ private DeliveryOptions createDeliveryOptions(Event<?> event){
             }
         }
         deliveryOptions.addHeader(EventConstants.CRI_HEADER, event.cri().raw());
+
+        // Prefer local delivery when this node already hosts a handler for the target service address.
+        // In the clustered event bus the point-to-point selector round-robins sends across every node
+        // registered for the address, so without this ~half of a node's own service calls get shipped to a
+        // remote node. localOnly=true bypasses the cluster selector and delivers to the local handler.
+        // Only set when this node actually hosts a handler, otherwise localOnly would fail with NO_HANDLERS.
+        if(shouldPreferLocalDelivery(event, baseResource)){
+            deliveryOptions.setLocalOnly(true);
+        }
+
         return deliveryOptions;
     }
 
+    /**
+     * Determines if a send to the given {@code baseResource} should be delivered to a local handler only.
+     * True only for point-to-point {@link EventConstants#SERVICE_DESTINATION_SCHEME service} sends whose
+     * address is currently hosted by a local handler on this node. This is a lock free
+     * {@link Map#containsKey} plus a scheme check, no cluster or registry query.
+     * Package private for testing.
+     */
+    boolean shouldPreferLocalDelivery(Event<?> event, String baseResource){
+        return EventConstants.SERVICE_DESTINATION_SCHEME.equals(event.cri().scheme())
+                && localListenerCounts.containsKey(baseResource);
+    }
+
 }

continuum-core-vertx/src/test/java/org/kinotic/continuum/internal/core/api/event/DefaultEventBusServiceLocalDeliveryTest.java

@@ -0,0 +1,144 @@
+/*
+ *
+ * Copyright 2008-2021 Kinotic and the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.kinotic.continuum.internal.core.api.event;
+
+import io.vertx.core.Vertx;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+import org.kinotic.continuum.core.api.event.Event;
+import org.springframework.test.util.ReflectionTestUtils;
+import reactor.core.Disposable;
+
+import java.nio.charset.StandardCharsets;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Verifies the {@code localOnly} delivery predicate of {@link DefaultEventBusService}: when this node hosts a
+ * local handler for a service address, sends to that address should prefer local delivery, otherwise they should
+ * fall back to normal cluster routing.
+ *
+ * Uses a real (non clustered) {@link Vertx} so the consumer registration / unregistration lifecycle that drives
+ * the predicate is exercised end to end rather than mocked.
+ */
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+public class DefaultEventBusServiceLocalDeliveryTest {
+
+    private Vertx vertx;
+    private DefaultEventBusService eventBusService;
+
+    @BeforeAll
+    public void startVertx() {
+        vertx = Vertx.vertx();
+    }
+
+    @AfterAll
+    public void stopVertx() {
+        if (vertx != null) {
+            vertx.close();
+        }
+    }
+
+    @BeforeEach
+    public void setUp() {
+        // a fresh service per test so the tracked local listener state starts empty
+        eventBusService = new DefaultEventBusService();
+        ReflectionTestUtils.setField(eventBusService, "vertx", vertx);
+        eventBusService.init();
+    }
+
+    @AfterEach
+    public void tearDown() {
+        eventBusService = null;
+    }
+
+    private static Event<byte[]> event(String cri) {
+        return Event.create(cri, "data".getBytes(StandardCharsets.UTF_8));
+    }
+
+    @Test
+    public void prefersLocalDeliveryForLocallyHostedServiceAddress() {
+        String baseResource = "srv://org.kinotic.continuum.tests.LocalDeliveryService";
+        Disposable subscription = eventBusService.listen(baseResource).subscribe();
+        try {
+            // a send targets the base resource of the request cri (scheme + optional scope + resource name)
+            Event<byte[]> request = event(baseResource + "/someMethod");
+            assertThat(request.cri().baseResource())
+                    .as("the send address must equal the listened base resource")
+                    .isEqualTo(baseResource);
+
+            assertThat(eventBusService.shouldPreferLocalDelivery(request, request.cri().baseResource()))
+                    .as("a service address hosted by a local handler should prefer local delivery")
+                    .isTrue();
+        } finally {
+            subscription.dispose();
+        }
+    }
+
+    @Test
+    public void doesNotPreferLocalDeliveryForAddressNotHostedLocally() {
+        String hosted = "srv://org.kinotic.continuum.tests.HostedService";
+        Disposable subscription = eventBusService.listen(hosted).subscribe();
+        try {
+            Event<byte[]> request = event("srv://org.kinotic.continuum.tests.OtherService/someMethod");
+            assertThat(eventBusService.shouldPreferLocalDelivery(request, request.cri().baseResource()))
+                    .as("a service address not hosted locally should fall back to cluster routing")
+                    .isFalse();
+        } finally {
+            subscription.dispose();
+        }
+    }
+
+    @Test
+    public void doesNotPreferLocalDeliveryForNonServiceScheme() {
+        // This node hosts a local handler at this exact address, but the non service scheme must still
+        // suppress the optimization so reply / stream / other sends are left untouched.
+        String streamAddress = "stream://org.kinotic.continuum.tests.SomeStream";
+        Disposable subscription = eventBusService.listen(streamAddress).subscribe();
+        try {
+            Event<byte[]> request = event(streamAddress + "/someMethod");
+            assertThat(eventBusService.shouldPreferLocalDelivery(request, request.cri().baseResource()))
+                    .as("a non service scheme should not prefer local delivery even when hosted locally")
+                    .isFalse();
+        } finally {
+            subscription.dispose();
+        }
+    }
+
+    @Test
+    public void clearsAfterConsumerUnregistered() {
+        String baseResource = "srv://org.kinotic.continuum.tests.EphemeralService";
+        Disposable subscription = eventBusService.listen(baseResource).subscribe();
+        Event<byte[]> request = event(baseResource + "/someMethod");
+
+        assertThat(eventBusService.shouldPreferLocalDelivery(request, request.cri().baseResource()))
+                .as("local delivery should be preferred while the consumer is registered")
+                .isTrue();
+
+        // unregistering the only local handler must clear the preference so a later send falls back to the cluster
+        subscription.dispose();
+
+        assertThat(eventBusService.shouldPreferLocalDelivery(request, request.cri().baseResource()))
+                .as("local delivery preference should clear once the consumer is unregistered")
+                .isFalse();
+    }
+}