Make socket per core

Author: akrambek

examples/tcp.proxy/README.md

@@ -20,7 +20,6 @@
 import java.util.concurrent.ConcurrentMap;
 
 import io.aklivity.zilla.runtime.binding.tcp.internal.config.TcpServerBindingConfig;
-import io.aklivity.zilla.runtime.binding.tcp.internal.util.OperatingSystem;
 import io.aklivity.zilla.runtime.engine.EngineContext;
 import io.aklivity.zilla.runtime.engine.binding.Binding;
 import io.aklivity.zilla.runtime.engine.binding.BindingContext;
@@ -61,10 +60,8 @@ public BindingContext supply(
     }
 
     private TcpServerBindingConfig supplyServer(
-        long bindingId)
+        long index)
     {
-        return OperatingSystem.detect() == OperatingSystem.OS.MACOS
-            ? servers.computeIfAbsent(bindingId, TcpServerBindingConfig::new)
-            : new TcpServerBindingConfig(bindingId);
+        return servers.computeIfAbsent(index, TcpServerBindingConfig::new);
     }
 }

examples/tcp.proxy/compose.yaml

@@ -40,9 +40,9 @@ public final class TcpServerBindingConfig
     private volatile ServerSocketChannel[] channels;
 
     public TcpServerBindingConfig(
-        long bindingId)
+        long index)
     {
-        this.id = bindingId;
+        this.id = index;
         this.binds = new AtomicInteger();
     }
 

examples/tcp.proxy/etc/zilla.yaml

@@ -99,14 +99,15 @@ public class TcpServerFactory implements TcpStreamFactory
     private final int replyMax;
     private final int windowThreshold;
     private final int proxyTypeId;
+    private final int index;
     private final BindingHandler streamFactory;
 
     public TcpServerFactory(
         TcpConfiguration config,
         EngineContext context,
         LongFunction<TcpServerBindingConfig> servers)
     {
-        this.router = new TcpServerRouter(config, context, this::handleAccept, servers);
+        this.router = new TcpServerRouter(context.index(), config, context, this::handleAccept, servers);
         this.writeBuffer = context.writeBuffer();
         this.writeByteBuffer = ByteBuffer.allocateDirect(writeBuffer.capacity()).order(nativeOrder());
         this.bufferPool = context.bufferPool();
@@ -116,6 +117,7 @@ public TcpServerFactory(
         this.supplyPollerKey = context::supplyPollerKey;
         this.streamFactory = context.streamFactory();
         this.proxyTypeId = context.supplyTypeId("proxy");
+        this.index = context.index();
 
         final int readBufferSize = writeBuffer.capacity() - DataFW.FIELD_OFFSET_PAYLOAD;
         this.readByteBuffer = ByteBuffer.allocateDirect(readBufferSize).order(nativeOrder());

runtime/binding-tcp/src/main/java/io/aklivity/zilla/runtime/binding/tcp/internal/TcpBinding.java

@@ -41,16 +41,19 @@ public final class TcpServerRouter
     private final ToIntFunction<PollerKey> acceptHandler;
     private final Function<SelectableChannel, PollerKey> supplyPollerKey;
     private final LongFunction<TcpServerBindingConfig> lookupServer;
+    private final int index;
 
     private int remainingConnections;
     private boolean unbound;
 
     public TcpServerRouter(
+        int index,
         TcpConfiguration config,
         EngineContext context,
         ToIntFunction<PollerKey> acceptHandler,
         LongFunction<TcpServerBindingConfig> lookupServer)
     {
+        this.index = index;
         this.remainingConnections = config.maxConnections();
         this.bindings = new Long2ObjectHashMap<>();
         this.supplyPollerKey = context::supplyPollerKey;
@@ -130,7 +133,7 @@ public void close(
     private void register(
         TcpBindingConfig binding)
     {
-        TcpServerBindingConfig server = lookupServer.apply(binding.id);
+        TcpServerBindingConfig server = lookupServer.apply(index);
         ServerSocketChannel[] channels = server.bind(binding.options);
 
         PollerKey[] acceptKeys = new PollerKey[channels.length];
@@ -158,7 +161,7 @@ private void unregister(
             }
         }
 
-        TcpServerBindingConfig server = lookupServer.apply(binding.id);
+        TcpServerBindingConfig server = lookupServer.apply(index);
         server.unbind();
     }
 }