(ignite-3) branch main updated: IGNITE-22603 Add InetSocketAddress caching for DefaultMessagingService (#4011)

[tk...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

tkalkirill pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git


The following commit(s) were added to refs/heads/main by this push:
     new 079225e576 IGNITE-22603 Add InetSocketAddress caching for DefaultMessagingService (#4011)
079225e576 is described below

commit 079225e5764e5537844c6e499f0a513c44a1e62d
Author: Kirill Tkalenko <tk...@yandex.ru>
AuthorDate: Fri Jun 28 15:28:18 2024 +0300

    IGNITE-22603 Add InetSocketAddress caching for DefaultMessagingService (#4011)
---
 .../internal/network/DefaultMessagingService.java  | 39 ++++++++++++++++++++--
 1 file changed, 36 insertions(+), 3 deletions(-)

diff --git a/modules/network/src/main/java/org/apache/ignite/internal/network/DefaultMessagingService.java b/modules/network/src/main/java/org/apache/ignite/internal/network/DefaultMessagingService.java
index 9ba36ae69d..5f035e4c7e 100644
--- a/modules/network/src/main/java/org/apache/ignite/internal/network/DefaultMessagingService.java
+++ b/modules/network/src/main/java/org/apache/ignite/internal/network/DefaultMessagingService.java
@@ -31,6 +31,7 @@ import java.net.InetSocketAddress;
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
@@ -114,14 +115,23 @@ public class DefaultMessagingService extends AbstractMessagingService {
     @Nullable
     private volatile BiPredicate<String, NetworkMessage> dropMessagesPredicate;
 
+    /**
+     * Cache of {@link InetSocketAddress} of recipient nodes ({@link ClusterNode}) that are in the topology and not stale.
+     *
+     * <p>Introduced for optimization - reducing the number of address resolving for the same nodes.</p>
+     */
+    private final Map<String, InetSocketAddress> recipientInetAddrByNodeId = new ConcurrentHashMap<>();
+
     /**
      * Constructor.
      *
+     * @param nodeName Consistent ID (aka name) of the local node associated with the service to create.
      * @param factory Network messages factory.
      * @param topologyService Topology service.
      * @param staleIdDetector Used to detect stale node IDs.
      * @param classDescriptorRegistry Descriptor registry.
      * @param marshaller Marshaller.
+     * @param criticalWorkerRegistry Used to register critical threads managed by the new service and its components.
      */
     public DefaultMessagingService(
             String nodeName,
@@ -566,6 +576,13 @@ public class DefaultMessagingService extends AbstractMessagingService {
      */
     public void start() {
         criticalWorkerRegistry.register(outboundExecutor);
+
+        topologyService.addEventHandler(new TopologyEventHandler() {
+            @Override
+            public void onDisappeared(ClusterNode member) {
+                recipientInetAddrByNodeId.remove(member.id());
+            }
+        });
     }
 
     /**
@@ -580,6 +597,8 @@ public class DefaultMessagingService extends AbstractMessagingService {
 
         criticalWorkerRegistry.unregister(outboundExecutor);
 
+        recipientInetAddrByNodeId.clear();
+
         inboundExecutors.close();
         IgniteUtils.shutdownAndAwaitTermination(outboundExecutor, 10, TimeUnit.SECONDS);
     }
@@ -674,15 +693,15 @@ public class DefaultMessagingService extends AbstractMessagingService {
      * @param recipientNode Target cluster node.
      */
     private @Nullable InetSocketAddress resolveRecipientAddress(ClusterNode recipientNode) {
-        NetworkAddress recipientAddress = recipientNode.address();
-
         // Node name is {@code null} if the node has not been added to the topology.
         if (recipientNode.name() != null) {
-            return connectionManager.consistentId().equals(recipientNode.name()) ? null : createResolved(recipientAddress);
+            return connectionManager.consistentId().equals(recipientNode.name()) ? null : getFromCacheOrCreateResolved(recipientNode);
         }
 
         InetSocketAddress localAddress = connectionManager.localAddress();
 
+        NetworkAddress recipientAddress = recipientNode.address();
+
         if (localAddress.getPort() != recipientAddress.port()) {
             return createResolved(recipientAddress);
         }
@@ -705,4 +724,18 @@ public class DefaultMessagingService extends AbstractMessagingService {
     private static InetSocketAddress createResolved(NetworkAddress address) {
         return new InetSocketAddress(address.host(), address.port());
     }
+
+    private InetSocketAddress getFromCacheOrCreateResolved(ClusterNode recipientNode) {
+        assert recipientNode.name() != null : "Node has not been added to the topology: " + recipientNode.id();
+
+        InetSocketAddress address = recipientInetAddrByNodeId.compute(recipientNode.id(), (nodeId, inetSocketAddress) -> {
+            if (staleIdDetector.isIdStale(nodeId)) {
+                return null;
+            }
+
+            return inetSocketAddress != null ? inetSocketAddress : createResolved(recipientNode.address());
+        });
+
+        return address != null ? address : createResolved(recipientNode.address());
+    }
 }