diff --git a/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientConnection.java b/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientConnection.java index 02226f6aa2ba2..f0837e2fe1aa8 100644 --- a/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientConnection.java +++ b/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientConnection.java @@ -20,6 +20,7 @@ import java.util.List; import java.util.Map; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutorService; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; @@ -29,6 +30,7 @@ import org.apache.camel.CamelContextAware; import org.apache.camel.RuntimeCamelException; import org.apache.camel.component.milo.client.internal.SubscriptionManager; +import org.apache.camel.spi.ExecutorServiceManager; import org.apache.camel.spi.ThreadPoolProfile; import org.apache.camel.util.ObjectHelper; import org.apache.camel.util.concurrent.ThreadPoolRejectedPolicy; @@ -49,7 +51,8 @@ public class MiloClientConnection implements AutoCloseable, CamelContextAware { private final Lock lock = new ReentrantLock(); private final MiloClientConfiguration configuration; private volatile CamelContext camelContext; - private ScheduledExecutorService executor; + private ExecutorService executor; + private ScheduledExecutorService reconnectExecutor; private SubscriptionManager manager; private volatile boolean initialized; @@ -101,18 +104,28 @@ protected void init() { ObjectHelper.notNull(this.camelContext, "camelContext"); // A failed request disconnects the client, which can wait for a reconnect attempt that Milo schedules on its shared - // scheduled executor, so this connection handles its requests and reconnects on an executor of its own. + // scheduled executor, so this connection handles its requests and reconnects on executors of its own. + final ExecutorServiceManager executorServiceManager = this.camelContext.getExecutorServiceManager(); final ThreadPoolProfile profile = new ThreadPoolProfile("MiloClient"); - // as many threads as that executor; with one processor or with virtual threads this is one thread, so a route - // that continues on it after an awaited request must not wait for another request of this connection - profile.setPoolSize(Runtime.getRuntime().availableProcessors()); - // not bounded, like that executor, because a rejected reconnect would end the reconnects of this connection - profile.setMaxQueueSize(-1); + // as many threads as that executor, but they end when they are idle; with virtual threads and more than one + // processor, a virtual thread for each task. Not more platform threads, because when the connection is lost, + // every request in flight fails and waits for the disconnect of the first failure. With one processor this is + // one thread, so a route that continues on it after an awaited request must not wait for another request of + // this connection. + final int threads = Runtime.getRuntime().availableProcessors(); + profile.setPoolSize(threads); + profile.setMaxPoolSize(threads); + // not bounded, so that no request outcome is rejected while the connection is open; a size of -1 would give a + // SynchronousQueue, which rejects a task while all the threads are busy + profile.setMaxQueueSize(Integer.MAX_VALUE); // a request that completes after the connection is closed fails instead of being dropped profile.setRejectedPolicy(ThreadPoolRejectedPolicy.Abort); - this.executor = this.camelContext.getExecutorServiceManager().newScheduledThreadPool(this, "MiloClient", profile); + this.executor = executorServiceManager.newThreadPool(this, "MiloClient", profile); + // the reconnects of the connection run one after the other, on a thread that starts with the first one and ends + // when the connection is closed + this.reconnectExecutor = executorServiceManager.newSingleThreadScheduledExecutor(this, "MiloClientReconnect"); - this.manager = new SubscriptionManager(this.configuration, this.executor, 10_000); + this.manager = new SubscriptionManager(this.configuration, this.reconnectExecutor, this.executor, 10_000); } @Override @@ -125,9 +138,16 @@ public void close() throws Exception { this.manager = null; } } finally { - if (this.executor != null) { - this.camelContext.getExecutorServiceManager().shutdown(this.executor); - this.executor = null; + try { + if (this.reconnectExecutor != null) { + this.camelContext.getExecutorServiceManager().shutdown(this.reconnectExecutor); + this.reconnectExecutor = null; + } + } finally { + if (this.executor != null) { + this.camelContext.getExecutorServiceManager().shutdown(this.executor); + this.executor = null; + } } } } finally { diff --git a/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/internal/SubscriptionManager.java b/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/internal/SubscriptionManager.java index 314a55b4625db..d2a6efcbd417a 100644 --- a/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/internal/SubscriptionManager.java +++ b/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/internal/SubscriptionManager.java @@ -717,7 +717,8 @@ nodeId, direction, NodeIds.References, includeSubTypes, uint(nodeClasses), } private final MiloClientConfiguration configuration; - private final ScheduledExecutorService executor; + private final ScheduledExecutorService reconnectExecutor; + private final Executor executor; private final long reconnectTimeout; private final ServerCertificateTrust serverCertificateTrust; private final Lock lock = new ReentrantLock(); @@ -728,10 +729,11 @@ nodeId, direction, NodeIds.References, includeSubTypes, uint(nodeClasses), private Future reconnectJob; private final Map subscriptions = new HashMap<>(); - public SubscriptionManager(final MiloClientConfiguration configuration, final ScheduledExecutorService executor, - final long reconnectTimeout) { + public SubscriptionManager(final MiloClientConfiguration configuration, final ScheduledExecutorService reconnectExecutor, + final Executor executor, final long reconnectTimeout) { this.configuration = configuration; + this.reconnectExecutor = reconnectExecutor; this.executor = executor; this.reconnectTimeout = reconnectTimeout; this.serverCertificateTrust = new ServerCertificateTrust(configuration); @@ -1101,6 +1103,11 @@ public void dispose() { } this.disposed = true; connected = this.connected; + if (this.reconnectJob != null) { + // a waiting reconnect would keep the thread of the reconnect executor until it is due + this.reconnectJob.cancel(false); + this.reconnectJob = null; + } } finally { lock.unlock(); } @@ -1128,9 +1135,10 @@ private void triggerReconnect(final boolean immediate) { } if (immediate) { - this.reconnectJob = this.executor.submit(this::connect); + this.reconnectJob = this.reconnectExecutor.submit(this::connect); } else { - this.reconnectJob = this.executor.schedule(this::connect, this.reconnectTimeout, TimeUnit.MILLISECONDS); + this.reconnectJob + = this.reconnectExecutor.schedule(this::connect, this.reconnectTimeout, TimeUnit.MILLISECONDS); } } finally { lock.unlock(); diff --git a/components/camel-milo/src/test/java/org/apache/camel/component/milo/MiloClientExecutorTest.java b/components/camel-milo/src/test/java/org/apache/camel/component/milo/MiloClientExecutorTest.java index a4be2a00135be..8326debb6face 100644 --- a/components/camel-milo/src/test/java/org/apache/camel/component/milo/MiloClientExecutorTest.java +++ b/components/camel-milo/src/test/java/org/apache/camel/component/milo/MiloClientExecutorTest.java @@ -27,9 +27,11 @@ import org.apache.camel.builder.RouteBuilder; import org.apache.camel.component.milo.server.MiloServerComponent; import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.test.AvailablePortFinder; import org.eclipse.milo.opcua.stack.core.Stack; import org.eclipse.milo.opcua.stack.core.types.builtin.Variant; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; import static org.apache.camel.component.milo.NodeIds.nodeValue; import static org.awaitility.Awaitility.await; @@ -45,6 +47,10 @@ public class MiloClientExecutorTest extends AbstractMiloServerTest { = "milo-client:opc.tcp://foo:bar@localhost:@@port@@?node=" + nodeValue(MiloServerComponent.DEFAULT_NAMESPACE_URI, "myitem1") + "&overrideHost=true"; + // no server listens on this port + @RegisterExtension + AvailablePortFinder.Port unreachablePort = AvailablePortFinder.find(); + @EndpointInject("mock:item") protected MockEndpoint item; @@ -55,6 +61,9 @@ protected RoutesBuilder createRouteBuilder() { public void configure() { from("milo-server:myitem1").to("mock:item"); from("direct:write").routeId("write").to(resolve(MILO_CLIENT_ITEM)); + from("milo-client:opc.tcp://localhost:" + unreachablePort.getPort() + "?node=" + + nodeValue(MiloServerComponent.DEFAULT_NAMESPACE_URI, "myitem1") + "&overrideHost=true") + .routeId("unreachable").autoStartup(false).to("mock:unreachable"); } }; } @@ -101,7 +110,8 @@ void awaitedWriteCompletesWhileTheSharedScheduledExecutorOfMiloIsBusy() throws E } /** - * The executor of a connection is shut down when the last endpoint releases the connection. + * The pool that handles the outcome of the requests of a connection is shut down when the last endpoint releases + * the connection. */ @Test void executorIsShutDownWithTheConnection() throws Exception { @@ -114,14 +124,34 @@ void executorIsShutDownWithTheConnection() throws Exception { await().atMost(30, TimeUnit.SECONDS).until(() -> executorThreads().isEmpty()); } + /** + * A connection that cannot connect waits for its next reconnect on a thread of its own. The thread ends when the + * last endpoint releases the connection, without waiting until that reconnect is due. + */ + @Test + void reconnectThreadEndsWithTheConnection() throws Exception { + context.getRouteController().startRoute("unreachable"); + await().atMost(30, TimeUnit.SECONDS).until(() -> !threads(" - MiloClientReconnect").isEmpty()); + + // the route is the only user of the connection + context.getRouteController().stopRoute("unreachable"); + + // the reconnect is due 10 seconds after the failed connect + await().atMost(5, TimeUnit.SECONDS).until(() -> threads(" - MiloClientReconnect").isEmpty()); + } + private void write(final String value) { template.sendBodyAndHeader("direct:write", new Variant(value), MiloConstants.HEADER_AWAIT, true); } private List executorThreads() { + return threads(" - MiloClient"); + } + + private List threads(final String suffix) { final String prefix = "Camel (" + context.getName() + ") thread #"; return Thread.getAllStackTraces().keySet().stream() - .filter(thread -> thread.getName().startsWith(prefix) && thread.getName().endsWith(" - MiloClient")) + .filter(thread -> thread.getName().startsWith(prefix) && thread.getName().endsWith(suffix)) .toList(); } } diff --git a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc index 08ed3d96753bf..5136676b0d7e5 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc @@ -889,13 +889,16 @@ the content of the earlier file was lost. The later entry is now added with a nu `orders (1).csv`, `orders (2).csv`, and so on. Entry names can therefore differ from the file names (or the `CamelFileName` header). Use `preserveFolderStructure=true` to keep the folders in the entry names instead. -=== camel-milo - each client connection has a thread pool of its own +=== camel-milo - each client connection has thread pools of its own A `milo-client` or `milo-browse` connection now handles the outcome of its write, method call, read and browse -requests, and its reconnects, on a thread pool of its own named `MiloClient`. Before, it used the shared scheduled -executor of Milo (`milo-shared-scheduled-executor`). A route that continues after an awaited request now runs on a -thread of this pool. A connection has up to as many of these threads as there are processors, and one with virtual -threads enabled; they end when the connection is closed. +requests on a thread pool of its own named `MiloClient`, and runs its reconnects on a single thread named +`MiloClientReconnect`. Before, it used the shared scheduled executor of Milo (`milo-shared-scheduled-executor`) for +both. A route that continues after an awaited request now runs on a thread of the `MiloClient` pool. The pool has up +to as many threads as there are processors, which end after they have been idle for the keep-alive time of the +default thread pool profile (60 seconds). With virtual threads enabled and more than one processor, it starts a +virtual thread for each task instead. The `MiloClientReconnect` thread starts with the first reconnect of the +connection, and its reconnects run one after the other. The threads of a connection end when it is closed. A `MiloClientConnection` created outside an endpoint needs a `CamelContext`, set with `setCamelContext`, before it is used.