Complete (composite) namespace attach on stack when called from same …

…worker thread

runtime/engine/src/main/java/io/aklivity/zilla/runtime/engine/Engine.java

@@ -51,10 +51,8 @@
 import java.util.function.ToIntFunction;
 import java.util.stream.Collectors;
 
-import org.agrona.CloseHelper;
 import org.agrona.ErrorHandler;
 import org.agrona.collections.Int2ObjectHashMap;
-import org.agrona.concurrent.AgentRunner;
 
 import io.aklivity.zilla.runtime.engine.binding.Binding;
 import io.aklivity.zilla.runtime.engine.binding.function.MessageConsumer;
@@ -85,7 +83,6 @@ public final class Engine implements Collector, AutoCloseable
 {
     private final Collection<Binding> bindings;
     private final ExecutorService tasks;
-    private final Collection<AgentRunner> runners;
     private final Tuning tuning;
     private final List<EngineExtSpi> extensions;
     private final ContextImpl context;
@@ -224,14 +221,10 @@ else if ("http".equals(protocol) || "https".equals(protocol))
             throw new UnsupportedOperationException();
         }
 
-        List<AgentRunner> runners = new ArrayList<>(workers.size());
-        workers.forEach(d -> runners.add(d.runner()));
-
         this.bindings = bindings;
         this.tasks = tasks;
         this.extensions = extensions;
         this.context = context;
-        this.runners = runners;
         this.readonly = readonly;
     }
 
@@ -247,10 +240,11 @@ public <T> T binding(
 
     public void start() throws Exception
     {
-        for (AgentRunner runner : runners)
+        for (EngineWorker worker : workers)
         {
-            AgentRunner.startOnThread(runner, Thread::new);
+            worker.doStart();
         }
+
         watcherTaskRef = watcherTask.submit();
         if (!readonly)
         {
@@ -272,11 +266,11 @@ public void close() throws Exception
         watcherTask.close();
         watcherTaskRef.get();
 
-        for (AgentRunner runner : runners)
+        for (EngineWorker worker : workers)
         {
             try
             {
-                CloseHelper.close(runner);
+                worker.doClose();
             }
             catch (Throwable ex)
             {

runtime/engine/src/main/java/io/aklivity/zilla/runtime/engine/internal/registry/EngineManager.java

@@ -79,7 +79,7 @@ public class EngineManager
     private final IntFunction<String> supplyName;
     private final IntFunction<ToIntFunction<KindConfig>> maxWorkers;
     private final Tuning tuning;
-    private final Collection<EngineWorker> dispatchers;
+    private final Collection<EngineWorker> workers;
     private final Consumer<String> logger;
     private final EngineExtContext context;
     private final EngineConfiguration config;
@@ -98,7 +98,7 @@ public EngineManager(
         IntFunction<String> supplyName,
         IntFunction<ToIntFunction<KindConfig>> maxWorkers,
         Tuning tuning,
-        Collection<EngineWorker> dispatchers,
+        Collection<EngineWorker> workers,
         Consumer<String> logger,
         EngineExtContext context,
         EngineConfiguration config,
@@ -112,7 +112,7 @@ public EngineManager(
         this.supplyName = supplyName;
         this.maxWorkers = maxWorkers;
         this.tuning = tuning;
-        this.dispatchers = dispatchers;
+        this.workers = workers;
         this.logger = logger;
         this.context = context;
         this.config = config;
@@ -138,12 +138,14 @@ public EngineConfig reconfigure(
 
                 try
                 {
-                    register(newConfig);
                     current = newConfig;
+                    register(newConfig);
                 }
                 catch (Exception ex)
                 {
                     context.onError(ex);
+
+                    current = oldConfig;
                     register(oldConfig);
 
                     rethrowUnchecked(ex);
@@ -401,8 +403,8 @@ private void unregister(
     private void register(
         NamespaceConfig namespace)
     {
-        dispatchers.stream()
-            .map(d -> d.attach(namespace))
+        workers.stream()
+            .map(w -> w.attach(namespace))
             .reduce(CompletableFuture::allOf)
             .ifPresent(CompletableFuture::join);
     }
@@ -412,8 +414,8 @@ private void unregister(
     {
         if (namespace != null)
         {
-            dispatchers.stream()
-                .map(d -> d.detach(namespace))
+            workers.stream()
+                .map(w -> w.detach(namespace))
                 .reduce(CompletableFuture::allOf)
                 .ifPresent(CompletableFuture::join);
         }

runtime/engine/src/main/java/io/aklivity/zilla/runtime/engine/internal/registry/EngineWorker.java

@@ -37,6 +37,7 @@
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static org.agrona.CloseHelper.quietClose;
 import static org.agrona.LangUtil.rethrowUnchecked;
+import static org.agrona.concurrent.AgentRunner.startOnThread;
 
 import java.net.InetAddress;
 import java.net.MalformedURLException;
@@ -63,6 +64,7 @@
 import java.util.function.LongUnaryOperator;
 import java.util.function.Supplier;
 
+import org.agrona.CloseHelper;
 import org.agrona.DeadlineTimerWheel;
 import org.agrona.DeadlineTimerWheel.TimerHandler;
 import org.agrona.DirectBuffer;
@@ -228,6 +230,8 @@ public class EngineWorker implements EngineContext, Agent
 
     private long lastReadStreamId;
 
+    private volatile Thread thread;
+
     public EngineWorker(
         EngineConfiguration config,
         ExecutorService executor,
@@ -743,6 +747,17 @@ public String roleName()
         return agentName;
     }
 
+    public void doStart()
+    {
+        thread = startOnThread(runner, Thread::new);
+    }
+
+    public void doClose()
+    {
+        CloseHelper.close(runner);
+        thread = null;
+    }
+
     @Override
     public int doWork()
     {
@@ -849,17 +864,35 @@ public CompletableFuture<Void> attach(
         NamespaceConfig namespace)
     {
         NamespaceTask attachTask = registry.attach(namespace);
-        taskQueue.offer(attachTask);
-        signaler.signalNow(0L, 0L, 0L, supplyTraceId(), SIGNAL_TASK_QUEUED, 0);
+
+        if (thread == Thread.currentThread())
+        {
+            attachTask.run();
+        }
+        else
+        {
+            taskQueue.offer(attachTask);
+            signaler.signalNow(0L, 0L, 0L, supplyTraceId(), SIGNAL_TASK_QUEUED, 0);
+        }
+
         return attachTask.future();
     }
 
     public CompletableFuture<Void> detach(
         NamespaceConfig namespace)
     {
         NamespaceTask detachTask = registry.detach(namespace);
-        taskQueue.offer(detachTask);
-        signaler.signalNow(0L, 0L, 0L, supplyTraceId(), SIGNAL_TASK_QUEUED, 0);
+
+        if (thread == Thread.currentThread())
+        {
+            detachTask.run();
+        }
+        else
+        {
+            taskQueue.offer(detachTask);
+            signaler.signalNow(0L, 0L, 0L, supplyTraceId(), SIGNAL_TASK_QUEUED, 0);
+        }
+
         return detachTask.future();
     }