You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by pt...@apache.org on 2025/06/30 11:40:28 UTC
(ignite-3) branch main updated: IGNITE-25257 .NET: Optimize platform job execution (#6120)
This is an automated email from the ASF dual-hosted git repository.
ptupitsyn 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 1fce76fa7b4 IGNITE-25257 .NET: Optimize platform job execution (#6120)
1fce76fa7b4 is described below
commit 1fce76fa7b4f2c9d3c587b9d9e9528b1b3a60dca
Author: Pavel Tupitsyn <pt...@apache.org>
AuthorDate: Mon Jun 30 14:40:22 2025 +0300
IGNITE-25257 .NET: Optimize platform job execution (#6120)
* Do not create Java classloader for platform jobs (lazy `JobClassLoader.classLoader()`)
* Cache `JobLoadContext` and resolved job types in .NET
* Time-based expiration
* Clear cache on unit undeploy
Results:
* Isolated .NET `ExecuteJobAsync` is ~500x faster
* End-to-end .NET job execution is ~4x faster, on par with Java jobs
---
.../handler/ClientInboundMessageHandler.java | 12 +-
.../apache/ignite/client/fakes/FakeCompute.java | 5 +-
.../internal/deployunit/DeploymentManagerImpl.java | 10 +-
.../internal/compute/ComputeComponentImpl.java | 2 +-
.../ignite/internal/compute/ComputeUtils.java | 2 +-
.../internal/compute/JobExecutionContextImpl.java | 7 +-
.../compute/executor/ComputeExecutorImpl.java | 15 ++
.../platform/dotnet/DotNetComputeExecutor.java | 32 +++
.../internal/compute/loader/JobClassLoader.java | 106 ++++-----
.../compute/loader/JobClassLoaderFactory.java | 78 +------
...JobClassLoader.java => JobClassLoaderImpl.java} | 47 +---
...obClassLoaderFactory.java => JobClasspath.java} | 39 +---
.../compute/streamer/StreamerReceiverJob.java | 2 +-
.../internal/compute/ComputeComponentImplTest.java | 3 +-
.../compute/JobExecutionContextImplTest.java | 4 +-
.../compute/executor/ComputeExecutorTest.java | 3 +-
.../compute/loader/JobClassLoaderFactoryTest.java | 35 +--
.../compute/loader/JobClassLoaderTest.java | 21 +-
.../compute/loader/JobContextManagerTest.java | 4 +-
.../Compute/ComputeJobExecutorBenchmarks.cs | 6 +-
.../Compute/PlatformJobBenchmarks.cs | 24 +-
.../Compute/Executor/JobLoadContextCacheTests.cs | 131 +++++++++++
.../TestHelpers/ManagementApi.cs | 4 +-
.../Apache.Ignite/Internal/ClientSocket.Server.cs | 7 +-
.../Compute/Executor/ComputeJobExecutor.cs | 54 +++--
.../Compute/Executor/DeploymentUnitPaths.cs | 3 +-
.../Internal/Compute/Executor/JobLoadContext.cs | 49 ++++-
.../Compute/Executor/JobLoadContextCache.cs | 241 +++++++++++++++++++++
.../org/apache/ignite/internal/app/IgniteImpl.java | 13 +-
29 files changed, 633 insertions(+), 326 deletions(-)
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientInboundMessageHandler.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientInboundMessageHandler.java
index 72585c31d6d..034bef8a369 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientInboundMessageHandler.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientInboundMessageHandler.java
@@ -1342,14 +1342,22 @@ public class ClientInboundMessageHandler
public CompletableFuture<Boolean> cancelJobAsync(long jobId) {
return sendServerToClientRequest(ServerOp.COMPUTE_JOB_CANCEL,
packer -> packer.packLong(jobId))
- .thenApply(ClientMessageUnpacker::unpackBoolean);
+ .thenApply(unpacker -> {
+ try (unpacker) {
+ return unpacker.unpackBoolean();
+ }
+ });
}
@Override
public CompletableFuture<Boolean> undeployUnitsAsync(List<String> deploymentUnitPaths) {
return sendServerToClientRequest(ServerOp.DEPLOYMENT_UNITS_UNDEPLOY,
packer -> packDeploymentUnitPaths(deploymentUnitPaths, packer))
- .thenApply(ClientMessageUnpacker::unpackBoolean);
+ .thenApply(unpacker -> {
+ try (unpacker) {
+ return unpacker.unpackBoolean();
+ }
+ });
}
@Override
diff --git a/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeCompute.java b/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeCompute.java
index 1a111c4be7d..c3c7dcf5a7e 100644
--- a/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeCompute.java
+++ b/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeCompute.java
@@ -24,7 +24,6 @@ import static org.apache.ignite.compute.JobStatus.FAILED;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
import static org.apache.ignite.internal.util.CompletableFutures.trueCompletedFuture;
-import java.net.URL;
import java.time.Instant;
import java.util.Collection;
import java.util.List;
@@ -128,11 +127,11 @@ public class FakeCompute implements IgniteComputeInternal {
}
if (jobClassName.startsWith("org.apache.ignite")) {
- JobClassLoader jobClassLoader = new JobClassLoader(List.of(), new URL[]{}, this.getClass().getClassLoader());
+ JobClassLoader jobClassLoader = new JobClassLoader(List.of(), this.getClass().getClassLoader());
Class<ComputeJob<Object, Object>> jobClass = ComputeUtils.jobClass(jobClassLoader, jobClassName);
ComputeJob<Object, Object> job = ComputeUtils.instantiateJob(jobClass);
CompletableFuture<Object> jobFut = job.executeAsync(
- new JobExecutionContextImpl(ignite, new AtomicBoolean(), this.getClass().getClassLoader(), null),
+ new JobExecutionContextImpl(ignite, new AtomicBoolean(), jobClassLoader, null),
SharedComputeUtils.unmarshalArgOrResult(arg, null, null));
return jobExecution(jobFut != null ? jobFut : nullCompletedFuture());
diff --git a/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeploymentManagerImpl.java b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeploymentManagerImpl.java
index 77fa718b417..7aefafd2dc7 100644
--- a/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeploymentManagerImpl.java
+++ b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeploymentManagerImpl.java
@@ -37,6 +37,7 @@ import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
import java.util.stream.Collectors;
import org.apache.ignite.deployment.version.Version;
import org.apache.ignite.internal.cluster.management.ClusterManagementGroupManager;
@@ -140,6 +141,7 @@ public class DeploymentManagerImpl implements IgniteDeployment {
* @param configuration Deployment configuration.
* @param cmgManager Cluster management group manager.
* @param nodeName Node consistent ID.
+ * @param onPathRemoving Consumer to call when a deployment unit path is being removed.
*/
public DeploymentManagerImpl(
ClusterService clusterService,
@@ -148,7 +150,8 @@ public class DeploymentManagerImpl implements IgniteDeployment {
Path workDir,
DeploymentConfiguration configuration,
ClusterManagementGroupManager cmgManager,
- String nodeName
+ String nodeName,
+ Consumer<Path> onPathRemoving
) {
this.deploymentUnitStore = deploymentUnitStore;
this.configuration = configuration;
@@ -161,7 +164,10 @@ public class DeploymentManagerImpl implements IgniteDeployment {
undeployer = new DeploymentUnitAcquiredWaiter(
nodeName,
deploymentUnitAccessor,
- unit -> deploymentUnitStore.updateNodeStatus(nodeName, unit.name(), unit.version(), REMOVING)
+ unit -> {
+ onPathRemoving.accept(deployer.unitPath(unit.name(), unit.version(), false));
+ deploymentUnitStore.updateNodeStatus(nodeName, unit.name(), unit.version(), REMOVING);
+ }
);
messaging = new DeployMessagingService(clusterService, cmgManager, deployer, tracker);
unitDownloader = new UnitDownloader(deploymentUnitStore, nodeName, deployer, tracker, messaging);
diff --git a/modules/compute/src/main/java/org/apache/ignite/internal/compute/ComputeComponentImpl.java b/modules/compute/src/main/java/org/apache/ignite/internal/compute/ComputeComponentImpl.java
index f1b91d57712..a0ad567f9d7 100644
--- a/modules/compute/src/main/java/org/apache/ignite/internal/compute/ComputeComponentImpl.java
+++ b/modules/compute/src/main/java/org/apache/ignite/internal/compute/ComputeComponentImpl.java
@@ -354,7 +354,7 @@ public class ComputeComponentImpl implements ComputeComponent, SystemViewProvide
I input
) {
try {
- return executor.executeTask(jobSubmitter, taskClass(context.classLoader(), taskClassName), input);
+ return executor.executeTask(jobSubmitter, taskClass(context.classLoader().classLoader(), taskClassName), input);
} catch (Throwable e) {
context.close();
throw e;
diff --git a/modules/compute/src/main/java/org/apache/ignite/internal/compute/ComputeUtils.java b/modules/compute/src/main/java/org/apache/ignite/internal/compute/ComputeUtils.java
index 2b91be4791c..eb2ce887791 100644
--- a/modules/compute/src/main/java/org/apache/ignite/internal/compute/ComputeUtils.java
+++ b/modules/compute/src/main/java/org/apache/ignite/internal/compute/ComputeUtils.java
@@ -101,7 +101,7 @@ public class ComputeUtils {
*/
public static <T, R> Class<ComputeJob<T, R>> jobClass(JobClassLoader jobClassLoader, String jobClassName) {
try {
- return (Class<ComputeJob<T, R>>) Class.forName(jobClassName, true, jobClassLoader);
+ return (Class<ComputeJob<T, R>>) Class.forName(jobClassName, true, jobClassLoader.classLoader());
} catch (ClassNotFoundException e) {
String message = "Cannot load job class by name '" + jobClassName + "'";
if (jobClassLoader.units().isEmpty()) {
diff --git a/modules/compute/src/main/java/org/apache/ignite/internal/compute/JobExecutionContextImpl.java b/modules/compute/src/main/java/org/apache/ignite/internal/compute/JobExecutionContextImpl.java
index e924e8882b2..16ce3614cd2 100644
--- a/modules/compute/src/main/java/org/apache/ignite/internal/compute/JobExecutionContextImpl.java
+++ b/modules/compute/src/main/java/org/apache/ignite/internal/compute/JobExecutionContextImpl.java
@@ -20,6 +20,7 @@ package org.apache.ignite.internal.compute;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.ignite.Ignite;
import org.apache.ignite.compute.JobExecutionContext;
+import org.apache.ignite.internal.compute.loader.JobClassLoader;
import org.apache.ignite.table.partition.Partition;
import org.jetbrains.annotations.Nullable;
@@ -31,7 +32,7 @@ public class JobExecutionContextImpl implements JobExecutionContext {
private final AtomicBoolean isInterrupted;
- private final ClassLoader classLoader;
+ private final JobClassLoader classLoader;
private final @Nullable Partition partition;
@@ -43,7 +44,7 @@ public class JobExecutionContextImpl implements JobExecutionContext {
* @param classLoader Job class loader.
* @param partition Partition associated with this job.
*/
- public JobExecutionContextImpl(Ignite ignite, AtomicBoolean isInterrupted, ClassLoader classLoader, @Nullable Partition partition) {
+ public JobExecutionContextImpl(Ignite ignite, AtomicBoolean isInterrupted, JobClassLoader classLoader, @Nullable Partition partition) {
this.ignite = ignite;
this.isInterrupted = isInterrupted;
this.classLoader = classLoader;
@@ -70,7 +71,7 @@ public class JobExecutionContextImpl implements JobExecutionContext {
*
* @return Job class loader.
*/
- public ClassLoader classLoader() {
+ public JobClassLoader classLoader() {
return classLoader;
}
}
diff --git a/modules/compute/src/main/java/org/apache/ignite/internal/compute/executor/ComputeExecutorImpl.java b/modules/compute/src/main/java/org/apache/ignite/internal/compute/executor/ComputeExecutorImpl.java
index 5972c70dfbc..fe94921de1a 100644
--- a/modules/compute/src/main/java/org/apache/ignite/internal/compute/executor/ComputeExecutorImpl.java
+++ b/modules/compute/src/main/java/org/apache/ignite/internal/compute/executor/ComputeExecutorImpl.java
@@ -24,6 +24,7 @@ import static org.apache.ignite.internal.thread.ThreadOperation.STORAGE_READ;
import static org.apache.ignite.internal.thread.ThreadOperation.STORAGE_WRITE;
import java.io.IOException;
+import java.nio.file.Path;
import java.util.ArrayList;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
@@ -258,4 +259,18 @@ public class ComputeExecutorImpl implements ComputeExecutor {
dotNetExec.stop();
}
}
+
+ /**
+ * Handles the removal of a deployment unit.
+ *
+ * @param unitPath Path to the deployment unit that is being removed.
+ */
+ public void onUnitRemoving(Path unitPath) {
+ DotNetComputeExecutor dotNetExec = dotNetComputeExecutor;
+
+ if (dotNetExec != null) {
+ // Start async undeployment, do not wait. Won't throw.
+ dotNetExec.beginUndeployUnit(unitPath);
+ }
+ }
}
diff --git a/modules/compute/src/main/java/org/apache/ignite/internal/compute/executor/platform/dotnet/DotNetComputeExecutor.java b/modules/compute/src/main/java/org/apache/ignite/internal/compute/executor/platform/dotnet/DotNetComputeExecutor.java
index a6761cfca66..9a85c6747c1 100644
--- a/modules/compute/src/main/java/org/apache/ignite/internal/compute/executor/platform/dotnet/DotNetComputeExecutor.java
+++ b/modules/compute/src/main/java/org/apache/ignite/internal/compute/executor/platform/dotnet/DotNetComputeExecutor.java
@@ -96,6 +96,38 @@ public class DotNetComputeExecutor {
return () -> executeJobAsync(deploymentUnitPaths, jobClassName, input, context);
}
+ /**
+ * Starts undeploy process for the specified deployment units.
+ *
+ * @param unitPath Paths to deployment units to undeploy.
+ */
+ public void beginUndeployUnit(Path unitPath) {
+ try {
+ String unitPathStr = unitPath.toRealPath().toString();
+
+ getPlatformComputeConnectionWithRetryAsync()
+ .thenCompose(conn -> conn.connectionFut()
+ .thenCompose(c -> c.undeployUnitsAsync(List.of(unitPathStr)))
+ .exceptionally(e -> {
+ var cause = unwrapCause(e);
+
+ if (cause instanceof TraceableException) {
+ TraceableException te = (TraceableException) cause;
+
+ if (te.code() == Client.SERVER_TO_CLIENT_REQUEST_ERR) {
+ // Connection was lost (process exited), nothing to do.
+ return true;
+ }
+ }
+
+ LOG.warn(".NET unit undeploy error: " + e.getMessage(), e);
+ return false;
+ }));
+ } catch (Throwable t) {
+ LOG.warn(".NET unit undeploy error: " + t.getMessage(), t);
+ }
+ }
+
/**
* Stops the executor.
*/
diff --git a/modules/compute/src/main/java/org/apache/ignite/internal/compute/loader/JobClassLoader.java b/modules/compute/src/main/java/org/apache/ignite/internal/compute/loader/JobClassLoader.java
index 20f55b5ade0..fcf024ae142 100644
--- a/modules/compute/src/main/java/org/apache/ignite/internal/compute/loader/JobClassLoader.java
+++ b/modules/compute/src/main/java/org/apache/ignite/internal/compute/loader/JobClassLoader.java
@@ -18,102 +18,73 @@
package org.apache.ignite.internal.compute.loader;
import java.net.URL;
-import java.net.URLClassLoader;
+import java.security.AccessController;
+import java.security.PrivilegedAction;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
-import java.util.regex.Pattern;
import org.apache.ignite.internal.deployunit.DisposableDeploymentUnit;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.lang.ErrorGroups.Compute;
import org.apache.ignite.lang.IgniteException;
/**
* Implementation of {@link ClassLoader} that loads classes from specified component directories.
*/
-public class JobClassLoader extends URLClassLoader {
- /**
- * Pattern to match system packages.
- */
- private static final Pattern SYSTEM_PACKAGES_PATTERN = Pattern.compile("^(java|x|javax|org\\.apache\\.ignite)\\..*$");
+public class JobClassLoader implements AutoCloseable {
+ private static final IgniteLogger LOG = Loggers.forClass(JobClassLoaderFactory.class);
private final List<DisposableDeploymentUnit> units;
- /**
- * Parent class loader.
- */
private final ClassLoader parent;
+ private volatile JobClassLoaderImpl impl;
+
/**
* Creates new instance of {@link JobClassLoader}.
*
- * @param urls URLs to load classes from.
* @param units Units to load classes from.
* @param parent Parent class loader.
*/
- public JobClassLoader(List<DisposableDeploymentUnit> units, URL[] urls, ClassLoader parent) {
- super("compute-job", urls, parent);
+ public JobClassLoader(List<DisposableDeploymentUnit> units, ClassLoader parent) {
this.units = units;
this.parent = parent;
}
+ /**
+ * Returns the list of deployment units.
+ *
+ * @return List of deployment units.
+ */
public List<DisposableDeploymentUnit> units() {
return units;
}
/**
- * Loads the class with the specified <a href="#binary-name">binary name</a>. The implementation of this method searches for classes in
- * the following order:
- *
- * <ol>
- *
- * <li><p> If the name starts with one of {@link JobClassLoader#SYSTEM_PACKAGES_PATTERN},
- * the loader delegates search to the parent. </p></li>
- *
- * <li><p> If the name doesn't start with one of {@link JobClassLoader#SYSTEM_PACKAGES_PATTERN}:
+ * Returns the inner class loader.
*
- * <ol>
- * <li><p> Invoke {@link #findLoadedClass(String)} to check if the class has already been loaded. </p></li>
- * <li><p> Invoke the {@link #findClass(String)} method to find the
- * class. </p></li>
- * <li><p> Invoke the {@link #loadClass(String) loadClass} method
- * on the parent class loader. </p></li>
- * </ol></p></li>
- *
- * </ol>
- *
- * @param name The <a href="#binary-name">binary name</a> of the class.
- * @return The resulting {@code Class} object.
- * @throws ClassNotFoundException If the class was not found.
+ * @return Class loader.
*/
- @Override
- protected Class<?> loadClass(String name, boolean resolve) throws ClassNotFoundException {
- boolean isSystem = SYSTEM_PACKAGES_PATTERN.matcher(name).find();
- if (isSystem) {
- try {
- return parent.loadClass(name);
- } catch (ClassNotFoundException exception) {
- return loadClassFromClasspath(name, resolve);
- }
- } else {
- try {
- return loadClassFromClasspath(name, resolve);
- } catch (ClassNotFoundException exception) {
- return parent.loadClass(name);
- }
+ public ClassLoader classLoader() {
+ if (impl != null) {
+ return impl;
}
- }
- private Class<?> loadClassFromClasspath(String name, boolean resolve) throws ClassNotFoundException {
- Class<?> loadedClass = findLoadedClass(name);
- if (loadedClass == null) {
- Class<?> clazz = findClass(name);
- if (resolve) {
- resolveClass(clazz);
+ impl = AccessController.doPrivileged((PrivilegedAction<JobClassLoaderImpl>) () -> {
+ URL[] classpath = units.stream()
+ .map(DisposableDeploymentUnit::path)
+ .flatMap(JobClasspath::collectClasspath)
+ .toArray(URL[]::new);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Created class loader with classpath: {}", Arrays.toString(classpath));
}
- return clazz;
- } else {
- return loadedClass;
- }
+
+ return new JobClassLoaderImpl(units, classpath, parent);
+ });
+
+ return impl;
}
@Override
@@ -129,7 +100,11 @@ public class JobClassLoader extends URLClassLoader {
}
try {
- super.close();
+ JobClassLoaderImpl impl0 = impl;
+
+ if (impl0 != null) {
+ impl0.close();
+ }
} catch (Exception e) {
exceptions.add(e);
}
@@ -146,11 +121,4 @@ public class JobClassLoader extends URLClassLoader {
throw igniteException;
}
}
-
- @Override
- public String toString() {
- return "JobClassLoader{"
- + "classpath=" + Arrays.toString(getURLs())
- + '}';
- }
}
diff --git a/modules/compute/src/main/java/org/apache/ignite/internal/compute/loader/JobClassLoaderFactory.java b/modules/compute/src/main/java/org/apache/ignite/internal/compute/loader/JobClassLoaderFactory.java
index d5e356a4978..bb277dffbff 100644
--- a/modules/compute/src/main/java/org/apache/ignite/internal/compute/loader/JobClassLoaderFactory.java
+++ b/modules/compute/src/main/java/org/apache/ignite/internal/compute/loader/JobClassLoaderFactory.java
@@ -17,32 +17,13 @@
package org.apache.ignite.internal.compute.loader;
-import java.io.IOException;
-import java.net.MalformedURLException;
-import java.net.URL;
-import java.nio.file.FileVisitResult;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.SimpleFileVisitor;
-import java.nio.file.attribute.BasicFileAttributes;
-import java.security.AccessController;
-import java.security.PrivilegedAction;
-import java.util.ArrayList;
-import java.util.Arrays;
import java.util.List;
-import java.util.stream.Stream;
-import org.apache.ignite.compute.ComputeException;
import org.apache.ignite.internal.deployunit.DisposableDeploymentUnit;
-import org.apache.ignite.internal.logger.IgniteLogger;
-import org.apache.ignite.internal.logger.Loggers;
-import org.apache.ignite.lang.ErrorGroups.Compute;
/**
* Creates a class loader for a job.
*/
public class JobClassLoaderFactory {
- private static final IgniteLogger LOG = Loggers.forClass(JobClassLoaderFactory.class);
-
/**
* Create a class loader for the specified units.
*
@@ -50,63 +31,6 @@ public class JobClassLoaderFactory {
* @return The class loader.
*/
public JobClassLoader createClassLoader(List<DisposableDeploymentUnit> units) {
- return AccessController.doPrivileged(new PrivilegedAction<>() {
- @Override
- public JobClassLoader run() {
- URL[] classpath = units.stream()
- .map(DisposableDeploymentUnit::path)
- .flatMap(JobClassLoaderFactory::collectClasspath)
- .toArray(URL[]::new);
-
- if (LOG.isDebugEnabled()) {
- LOG.debug("Created class loader with classpath: {}", Arrays.toString(classpath));
- }
-
- return new JobClassLoader(units, classpath, getClass().getClassLoader());
- }
- });
- }
-
- private static Stream<URL> collectClasspath(Path unitDir) {
- if (Files.notExists(unitDir)) {
- throw new IllegalArgumentException("Unit does not exist: " + unitDir);
- }
-
- if (!Files.isDirectory(unitDir)) {
- throw new IllegalArgumentException("Unit is not a directory: " + unitDir);
- }
-
- // Construct the "class path" for this unit
- try {
- ClasspathCollector classpathCollector = new ClasspathCollector(unitDir);
- Files.walkFileTree(unitDir, classpathCollector);
- return classpathCollector.classpathAsStream();
- } catch (IOException e) {
- throw new ComputeException(
- Compute.CLASS_PATH_ERR,
- "Failed to construct classpath for job: " + unitDir,
- e
- );
- }
- }
-
- private static class ClasspathCollector extends SimpleFileVisitor<Path> {
- private final List<URL> classpath = new ArrayList<>();
-
- private ClasspathCollector(Path base) throws MalformedURLException {
- classpath.add(base.toAbsolutePath().toUri().toURL());
- }
-
- @Override
- public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
- if (Files.isDirectory(file) || file.toString().endsWith(".jar")) {
- classpath.add(file.toAbsolutePath().toUri().toURL());
- }
- return FileVisitResult.CONTINUE;
- }
-
- Stream<URL> classpathAsStream() {
- return classpath.stream();
- }
+ return new JobClassLoader(units, getClass().getClassLoader());
}
}
diff --git a/modules/compute/src/main/java/org/apache/ignite/internal/compute/loader/JobClassLoader.java b/modules/compute/src/main/java/org/apache/ignite/internal/compute/loader/JobClassLoaderImpl.java
similarity index 73%
copy from modules/compute/src/main/java/org/apache/ignite/internal/compute/loader/JobClassLoader.java
copy to modules/compute/src/main/java/org/apache/ignite/internal/compute/loader/JobClassLoaderImpl.java
index 20f55b5ade0..0bd1db89b09 100644
--- a/modules/compute/src/main/java/org/apache/ignite/internal/compute/loader/JobClassLoader.java
+++ b/modules/compute/src/main/java/org/apache/ignite/internal/compute/loader/JobClassLoaderImpl.java
@@ -19,18 +19,12 @@ package org.apache.ignite.internal.compute.loader;
import java.net.URL;
import java.net.URLClassLoader;
-import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.regex.Pattern;
import org.apache.ignite.internal.deployunit.DisposableDeploymentUnit;
-import org.apache.ignite.lang.ErrorGroups.Compute;
-import org.apache.ignite.lang.IgniteException;
-/**
- * Implementation of {@link ClassLoader} that loads classes from specified component directories.
- */
-public class JobClassLoader extends URLClassLoader {
+class JobClassLoaderImpl extends URLClassLoader {
/**
* Pattern to match system packages.
*/
@@ -50,7 +44,7 @@ public class JobClassLoader extends URLClassLoader {
* @param units Units to load classes from.
* @param parent Parent class loader.
*/
- public JobClassLoader(List<DisposableDeploymentUnit> units, URL[] urls, ClassLoader parent) {
+ JobClassLoaderImpl(List<DisposableDeploymentUnit> units, URL[] urls, ClassLoader parent) {
super("compute-job", urls, parent);
this.units = units;
this.parent = parent;
@@ -66,10 +60,10 @@ public class JobClassLoader extends URLClassLoader {
*
* <ol>
*
- * <li><p> If the name starts with one of {@link JobClassLoader#SYSTEM_PACKAGES_PATTERN},
+ * <li><p> If the name starts with one of {@link #SYSTEM_PACKAGES_PATTERN},
* the loader delegates search to the parent. </p></li>
*
- * <li><p> If the name doesn't start with one of {@link JobClassLoader#SYSTEM_PACKAGES_PATTERN}:
+ * <li><p> If the name doesn't start with one of {@link #SYSTEM_PACKAGES_PATTERN}:
*
* <ol>
* <li><p> Invoke {@link #findLoadedClass(String)} to check if the class has already been loaded. </p></li>
@@ -77,7 +71,7 @@ public class JobClassLoader extends URLClassLoader {
* class. </p></li>
* <li><p> Invoke the {@link #loadClass(String) loadClass} method
* on the parent class loader. </p></li>
- * </ol></p></li>
+ * </ol></li>
*
* </ol>
*
@@ -116,37 +110,6 @@ public class JobClassLoader extends URLClassLoader {
}
}
- @Override
- public void close() {
- List<Exception> exceptions = new ArrayList<>();
-
- for (DisposableDeploymentUnit unit : units) {
- try {
- unit.release();
- } catch (Exception e) {
- exceptions.add(e);
- }
- }
-
- try {
- super.close();
- } catch (Exception e) {
- exceptions.add(e);
- }
-
- if (!exceptions.isEmpty()) {
- IgniteException igniteException = new IgniteException(
- Compute.CLASS_LOADER_ERR,
- "Failed to close class loader"
- );
-
- for (Exception exception : exceptions) {
- igniteException.addSuppressed(exception);
- }
- throw igniteException;
- }
- }
-
@Override
public String toString() {
return "JobClassLoader{"
diff --git a/modules/compute/src/main/java/org/apache/ignite/internal/compute/loader/JobClassLoaderFactory.java b/modules/compute/src/main/java/org/apache/ignite/internal/compute/loader/JobClasspath.java
similarity index 66%
copy from modules/compute/src/main/java/org/apache/ignite/internal/compute/loader/JobClassLoaderFactory.java
copy to modules/compute/src/main/java/org/apache/ignite/internal/compute/loader/JobClasspath.java
index d5e356a4978..ec909fe33d0 100644
--- a/modules/compute/src/main/java/org/apache/ignite/internal/compute/loader/JobClassLoaderFactory.java
+++ b/modules/compute/src/main/java/org/apache/ignite/internal/compute/loader/JobClasspath.java
@@ -25,49 +25,14 @@ import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.SimpleFileVisitor;
import java.nio.file.attribute.BasicFileAttributes;
-import java.security.AccessController;
-import java.security.PrivilegedAction;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.List;
import java.util.stream.Stream;
import org.apache.ignite.compute.ComputeException;
-import org.apache.ignite.internal.deployunit.DisposableDeploymentUnit;
-import org.apache.ignite.internal.logger.IgniteLogger;
-import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.lang.ErrorGroups.Compute;
-/**
- * Creates a class loader for a job.
- */
-public class JobClassLoaderFactory {
- private static final IgniteLogger LOG = Loggers.forClass(JobClassLoaderFactory.class);
-
- /**
- * Create a class loader for the specified units.
- *
- * @param units The units of the job.
- * @return The class loader.
- */
- public JobClassLoader createClassLoader(List<DisposableDeploymentUnit> units) {
- return AccessController.doPrivileged(new PrivilegedAction<>() {
- @Override
- public JobClassLoader run() {
- URL[] classpath = units.stream()
- .map(DisposableDeploymentUnit::path)
- .flatMap(JobClassLoaderFactory::collectClasspath)
- .toArray(URL[]::new);
-
- if (LOG.isDebugEnabled()) {
- LOG.debug("Created class loader with classpath: {}", Arrays.toString(classpath));
- }
-
- return new JobClassLoader(units, classpath, getClass().getClassLoader());
- }
- });
- }
-
- private static Stream<URL> collectClasspath(Path unitDir) {
+class JobClasspath {
+ static Stream<URL> collectClasspath(Path unitDir) {
if (Files.notExists(unitDir)) {
throw new IllegalArgumentException("Unit does not exist: " + unitDir);
}
diff --git a/modules/compute/src/main/java/org/apache/ignite/internal/compute/streamer/StreamerReceiverJob.java b/modules/compute/src/main/java/org/apache/ignite/internal/compute/streamer/StreamerReceiverJob.java
index fb34d8ec4cd..30c15635576 100644
--- a/modules/compute/src/main/java/org/apache/ignite/internal/compute/streamer/StreamerReceiverJob.java
+++ b/modules/compute/src/main/java/org/apache/ignite/internal/compute/streamer/StreamerReceiverJob.java
@@ -46,7 +46,7 @@ public class StreamerReceiverJob implements ComputeJob<byte[], byte[]> {
buf.slice().order(ByteOrder.LITTLE_ENDIAN),
payloadElementCount,
receiverClassName -> {
- ClassLoader classLoader = ((JobExecutionContextImpl) context).classLoader();
+ ClassLoader classLoader = ((JobExecutionContextImpl) context).classLoader().classLoader();
Class<DataStreamerReceiver<Object, Object, Object>> receiverClass = ComputeUtils.receiverClass(
classLoader, receiverClassName);
diff --git a/modules/compute/src/test/java/org/apache/ignite/internal/compute/ComputeComponentImplTest.java b/modules/compute/src/test/java/org/apache/ignite/internal/compute/ComputeComponentImplTest.java
index 935e9ec703a..ba5daaeee3a 100644
--- a/modules/compute/src/test/java/org/apache/ignite/internal/compute/ComputeComponentImplTest.java
+++ b/modules/compute/src/test/java/org/apache/ignite/internal/compute/ComputeComponentImplTest.java
@@ -56,7 +56,6 @@ import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
-import java.net.URL;
import java.time.Duration;
import java.time.Instant;
import java.util.List;
@@ -158,7 +157,7 @@ class ComputeComponentImplTest extends BaseIgniteAbstractTest {
lenient().when(ignite.name()).thenReturn(INSTANCE_NAME);
lenient().when(topologyService.localMember().name()).thenReturn(INSTANCE_NAME);
- JobClassLoader classLoader = new JobClassLoader(List.of(), new URL[0], getClass().getClassLoader());
+ JobClassLoader classLoader = new JobClassLoader(List.of(), getClass().getClassLoader());
JobContext jobContext = new JobContext(classLoader, ignored -> {});
lenient().when(jobContextManager.acquireClassLoader(anyList()))
.thenReturn(completedFuture(jobContext));
diff --git a/modules/compute/src/test/java/org/apache/ignite/internal/compute/JobExecutionContextImplTest.java b/modules/compute/src/test/java/org/apache/ignite/internal/compute/JobExecutionContextImplTest.java
index 4e4814c7bbb..9151ae6fdbd 100644
--- a/modules/compute/src/test/java/org/apache/ignite/internal/compute/JobExecutionContextImplTest.java
+++ b/modules/compute/src/test/java/org/apache/ignite/internal/compute/JobExecutionContextImplTest.java
@@ -37,7 +37,7 @@ class JobExecutionContextImplTest extends BaseIgniteAbstractTest {
@Test
void returnsIgnite() {
- JobExecutionContext context = new JobExecutionContextImpl(ignite, new AtomicBoolean(), ClassLoader.getSystemClassLoader(), null);
+ JobExecutionContext context = new JobExecutionContextImpl(ignite, new AtomicBoolean(), null, null);
assertThat(context.ignite(), is(sameInstance(ignite)));
}
@@ -46,7 +46,7 @@ class JobExecutionContextImplTest extends BaseIgniteAbstractTest {
void returnsInterruptedFlag() {
AtomicBoolean isInterrupted = new AtomicBoolean();
- JobExecutionContext context = new JobExecutionContextImpl(ignite, isInterrupted, ClassLoader.getSystemClassLoader(), null);
+ JobExecutionContext context = new JobExecutionContextImpl(ignite, isInterrupted, null, null);
assertThat(context.isCancelled(), is(false));
diff --git a/modules/compute/src/test/java/org/apache/ignite/internal/compute/executor/ComputeExecutorTest.java b/modules/compute/src/test/java/org/apache/ignite/internal/compute/executor/ComputeExecutorTest.java
index 65346142e29..c62e3fc54f3 100644
--- a/modules/compute/src/test/java/org/apache/ignite/internal/compute/executor/ComputeExecutorTest.java
+++ b/modules/compute/src/test/java/org/apache/ignite/internal/compute/executor/ComputeExecutorTest.java
@@ -32,7 +32,6 @@ import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.is;
import static org.mockito.Answers.RETURNS_DEEP_STUBS;
-import java.net.URL;
import java.util.List;
import java.util.Map;
import java.util.UUID;
@@ -77,7 +76,7 @@ class ComputeExecutorTest extends BaseIgniteAbstractTest {
private ComputeExecutor computeExecutor;
- private final JobClassLoader jobClassLoader = new JobClassLoader(List.of(), new URL[0], getClass().getClassLoader());
+ private final JobClassLoader jobClassLoader = new JobClassLoader(List.of(), getClass().getClassLoader());
@BeforeEach
void setUp() {
diff --git a/modules/compute/src/test/java/org/apache/ignite/internal/compute/loader/JobClassLoaderFactoryTest.java b/modules/compute/src/test/java/org/apache/ignite/internal/compute/loader/JobClassLoaderFactoryTest.java
index a3b8a1a48ba..08562dc3c38 100644
--- a/modules/compute/src/test/java/org/apache/ignite/internal/compute/loader/JobClassLoaderFactoryTest.java
+++ b/modules/compute/src/test/java/org/apache/ignite/internal/compute/loader/JobClassLoaderFactoryTest.java
@@ -65,13 +65,13 @@ class JobClassLoaderFactoryTest extends BaseIgniteAbstractTest {
try (JobClassLoader classLoader1 = jobClassLoaderFactory.createClassLoader(units1);
JobClassLoader classLoader2 = jobClassLoaderFactory.createClassLoader(units2)) {
// then classes from the first unit are loaded from the first class loader
- Class<?> clazz1 = classLoader1.loadClass(UNIT_JOB_CLASS_NAME);
+ Class<?> clazz1 = classLoader1.classLoader().loadClass(UNIT_JOB_CLASS_NAME);
ComputeJob<Void, Integer> job1 = (ComputeJob<Void, Integer>) clazz1.getDeclaredConstructor().newInstance();
CompletableFuture<Integer> result1 = job1.executeAsync(null, null);
assertThat(result1, willBe(1));
// and classes from the second unit are loaded from the second class loader
- Class<?> clazz2 = classLoader2.loadClass(UNIT_JOB_CLASS_NAME);
+ Class<?> clazz2 = classLoader2.classLoader().loadClass(UNIT_JOB_CLASS_NAME);
ComputeJob<Void, String> job2 = (ComputeJob<Void, String>) clazz2.getDeclaredConstructor().newInstance();
CompletableFuture<String> result2 = job2.executeAsync(null, null);
assertThat(result2, willBe("Hello World!"));
@@ -88,7 +88,7 @@ class JobClassLoaderFactoryTest extends BaseIgniteAbstractTest {
);
try (JobClassLoader classLoader = jobClassLoaderFactory.createClassLoader(units)) {
- Class<?> unitJobClass = classLoader.loadClass(UNIT_JOB_CLASS_NAME);
+ Class<?> unitJobClass = classLoader.classLoader().loadClass(UNIT_JOB_CLASS_NAME);
assertNotNull(unitJobClass);
// and classes are loaded in the aplhabetical order
@@ -96,10 +96,10 @@ class JobClassLoaderFactoryTest extends BaseIgniteAbstractTest {
CompletableFuture<Integer> result1 = job1.executeAsync(null, null);
assertThat(result1, willBe(1));
- Class<?> job1UtilityClass = classLoader.loadClass(JOB1_UTILITY_CLASS_NAME);
+ Class<?> job1UtilityClass = classLoader.classLoader().loadClass(JOB1_UTILITY_CLASS_NAME);
assertNotNull(job1UtilityClass);
- Class<?> job2UtilityClass = classLoader.loadClass(JOB2_UTILITY_CLASS_NAME);
+ Class<?> job2UtilityClass = classLoader.classLoader().loadClass(JOB2_UTILITY_CLASS_NAME);
assertNotNull(job2UtilityClass);
// classes from the different units are loaded from the same class loader
@@ -116,13 +116,13 @@ class JobClassLoaderFactoryTest extends BaseIgniteAbstractTest {
// then class from all jars are loaded
try (JobClassLoader classLoader = jobClassLoaderFactory.createClassLoader(units)) {
- Class<?> unitJobClass = classLoader.loadClass(UNIT_JOB_CLASS_NAME);
+ Class<?> unitJobClass = classLoader.classLoader().loadClass(UNIT_JOB_CLASS_NAME);
assertNotNull(unitJobClass);
- Class<?> job1UtilityClass = classLoader.loadClass(JOB1_UTILITY_CLASS_NAME);
+ Class<?> job1UtilityClass = classLoader.classLoader().loadClass(JOB1_UTILITY_CLASS_NAME);
assertNotNull(job1UtilityClass);
- Class<?> job2UtilityClass = classLoader.loadClass(JOB2_UTILITY_CLASS_NAME);
+ Class<?> job2UtilityClass = classLoader.classLoader().loadClass(JOB2_UTILITY_CLASS_NAME);
assertNotNull(job2UtilityClass);
}
}
@@ -136,13 +136,13 @@ class JobClassLoaderFactoryTest extends BaseIgniteAbstractTest {
// then class from all jars are loaded
try (JobClassLoader classLoader = jobClassLoaderFactory.createClassLoader(units)) {
- Class<?> unitJobClass = classLoader.loadClass(UNIT_JOB_CLASS_NAME);
+ Class<?> unitJobClass = classLoader.classLoader().loadClass(UNIT_JOB_CLASS_NAME);
assertNotNull(unitJobClass);
- Class<?> job1UtilityClass = classLoader.loadClass(JOB1_UTILITY_CLASS_NAME);
+ Class<?> job1UtilityClass = classLoader.classLoader().loadClass(JOB1_UTILITY_CLASS_NAME);
assertNotNull(job1UtilityClass);
- Class<?> job2UtilityClass = classLoader.loadClass(JOB2_UTILITY_CLASS_NAME);
+ Class<?> job2UtilityClass = classLoader.classLoader().loadClass(JOB2_UTILITY_CLASS_NAME);
assertNotNull(job2UtilityClass);
}
}
@@ -156,7 +156,7 @@ class JobClassLoaderFactoryTest extends BaseIgniteAbstractTest {
// then class loader throws an exception
try (JobClassLoader classLoader = jobClassLoaderFactory.createClassLoader(units)) {
- assertThrows(ClassNotFoundException.class, () -> classLoader.loadClass(UNIT_JOB_CLASS_NAME));
+ assertThrows(ClassNotFoundException.class, () -> classLoader.classLoader().loadClass(UNIT_JOB_CLASS_NAME));
}
}
@@ -171,7 +171,8 @@ class JobClassLoaderFactoryTest extends BaseIgniteAbstractTest {
List<DisposableDeploymentUnit> units = toDisposableDeploymentUnits(new DeploymentUnit("unit1", "5.0.0"));
// then the files are accessible
- try (JobClassLoader classLoader = jobClassLoaderFactory.createClassLoader(units)) {
+ try (JobClassLoader jobClassLoader = jobClassLoaderFactory.createClassLoader(units)) {
+ ClassLoader classLoader = jobClassLoader.classLoader();
String resource = Files.readString(getPath(classLoader.getResource("test.txt")));
String subDirResource = Files.readString(getPath(classLoader.getResource("subdir/test.txt")));
assertEquals(expectedContent, resource);
@@ -199,7 +200,9 @@ class JobClassLoaderFactoryTest extends BaseIgniteAbstractTest {
}
);
- assertThrows(IllegalArgumentException.class, () -> jobClassLoaderFactory.createClassLoader(List.of(disposableDeploymentUnit)));
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> jobClassLoaderFactory.createClassLoader(List.of(disposableDeploymentUnit)).classLoader());
}
@Test
@@ -213,7 +216,9 @@ class JobClassLoaderFactoryTest extends BaseIgniteAbstractTest {
}
);
- assertThrows(IllegalArgumentException.class, () -> jobClassLoaderFactory.createClassLoader(List.of(disposableDeploymentUnit)));
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> jobClassLoaderFactory.createClassLoader(List.of(disposableDeploymentUnit)).classLoader());
}
private List<DisposableDeploymentUnit> toDisposableDeploymentUnits(DeploymentUnit... units) {
diff --git a/modules/compute/src/test/java/org/apache/ignite/internal/compute/loader/JobClassLoaderTest.java b/modules/compute/src/test/java/org/apache/ignite/internal/compute/loader/JobClassLoaderTest.java
index 722de43c4c1..014fce1c001 100644
--- a/modules/compute/src/test/java/org/apache/ignite/internal/compute/loader/JobClassLoaderTest.java
+++ b/modules/compute/src/test/java/org/apache/ignite/internal/compute/loader/JobClassLoaderTest.java
@@ -65,7 +65,7 @@ class JobClassLoaderTest extends BaseIgniteAbstractTest {
doReturn(desiredClass).when(parentClassLoader).loadClass(className);
- try (TestJobClassLoader jobClassLoader = spy(new TestJobClassLoader(new URL[0], List.of(), parentClassLoader))) {
+ try (TestJobClassLoaderImpl jobClassLoader = spy(new TestJobClassLoaderImpl(new URL[0], List.of(), parentClassLoader))) {
assertSame(desiredClass, jobClassLoader.loadClass(className));
verify(jobClassLoader, never()).findClass(className);
}
@@ -76,8 +76,8 @@ class JobClassLoaderTest extends BaseIgniteAbstractTest {
public void loadsOwnClassIfSystemAbsent(String className) throws Exception {
doThrow(ClassNotFoundException.class).when(parentClassLoader).loadClass(className);
- try (TestJobClassLoader jobClassLoader = spy(new TestJobClassLoader(new URL[0], List.of(), parentClassLoader))) {
- Class<TestJobClassLoader> toBeReturned = TestJobClassLoader.class;
+ try (TestJobClassLoaderImpl jobClassLoader = spy(new TestJobClassLoaderImpl(new URL[0], List.of(), parentClassLoader))) {
+ Class<TestJobClassLoaderImpl> toBeReturned = TestJobClassLoaderImpl.class;
doReturn(toBeReturned).when(jobClassLoader).findClass(className);
assertSame(toBeReturned, jobClassLoader.loadClass(className));
@@ -90,7 +90,7 @@ class JobClassLoaderTest extends BaseIgniteAbstractTest {
DisposableDeploymentUnit unit1 = mock(DisposableDeploymentUnit.class);
DisposableDeploymentUnit unit2 = mock(DisposableDeploymentUnit.class);
- try (TestJobClassLoader jobClassLoader = new TestJobClassLoader(new URL[0], List.of(unit1, unit2), parentClassLoader)) {
+ try (TestJobClassLoader jobClassLoader = new TestJobClassLoader(List.of(unit1, unit2), parentClassLoader)) {
jobClassLoader.close();
verify(unit1, times(1)).release();
verify(unit2, times(1)).release();
@@ -104,14 +104,13 @@ class JobClassLoaderTest extends BaseIgniteAbstractTest {
RuntimeException toBeThrown = new RuntimeException("Expected exception");
doThrow(toBeThrown).when(unit1).release();
- TestJobClassLoader jobClassLoader = new TestJobClassLoader(new URL[0], List.of(unit1, unit2), parentClassLoader);
+ TestJobClassLoader jobClassLoader = new TestJobClassLoader(List.of(unit1, unit2), parentClassLoader);
IgniteException igniteException = assertThrows(IgniteException.class, jobClassLoader::close);
assertThat(igniteException.getSuppressed(), arrayContaining(toBeThrown));
}
-
- private static class TestJobClassLoader extends JobClassLoader {
- TestJobClassLoader(URL[] urls, List<DisposableDeploymentUnit> units, ClassLoader parent) {
+ private static class TestJobClassLoaderImpl extends JobClassLoaderImpl {
+ TestJobClassLoaderImpl(URL[] urls, List<DisposableDeploymentUnit> units, ClassLoader parent) {
super(units, urls, parent);
}
@@ -120,4 +119,10 @@ class JobClassLoaderTest extends BaseIgniteAbstractTest {
return super.findClass(name);
}
}
+
+ private static class TestJobClassLoader extends JobClassLoader {
+ TestJobClassLoader(List<DisposableDeploymentUnit> units, ClassLoader parent) {
+ super(units, parent);
+ }
+ }
}
diff --git a/modules/compute/src/test/java/org/apache/ignite/internal/compute/loader/JobContextManagerTest.java b/modules/compute/src/test/java/org/apache/ignite/internal/compute/loader/JobContextManagerTest.java
index 1df4f944977..486e8215f05 100644
--- a/modules/compute/src/test/java/org/apache/ignite/internal/compute/loader/JobContextManagerTest.java
+++ b/modules/compute/src/test/java/org/apache/ignite/internal/compute/loader/JobContextManagerTest.java
@@ -100,7 +100,7 @@ class JobContextManagerTest extends BaseIgniteAbstractTest {
.collect(Collectors.toList());
- JobClassLoader toBeReturned = new JobClassLoader(deploymentUnits, new URL[0], getClass().getClassLoader());
+ JobClassLoader toBeReturned = new JobClassLoader(deploymentUnits, getClass().getClassLoader());
doReturn(toBeReturned)
.when(jobClassLoaderFactory).createClassLoader(deploymentUnits);
@@ -128,11 +128,9 @@ class JobContextManagerTest extends BaseIgniteAbstractTest {
try (JobClassLoader toBeReturned1 = new JobClassLoader(
disposableVersion1,
- extractUrls(disposableVersion1),
getClass().getClassLoader());
JobClassLoader toBeReturned2 = new JobClassLoader(
disposableVersion2,
- extractUrls(disposableVersion2),
getClass().getClassLoader())
) {
diff --git a/modules/platforms/dotnet/Apache.Ignite.Benchmarks/Compute/ComputeJobExecutorBenchmarks.cs b/modules/platforms/dotnet/Apache.Ignite.Benchmarks/Compute/ComputeJobExecutorBenchmarks.cs
index 7c21f9c248f..9bed8934874 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Benchmarks/Compute/ComputeJobExecutorBenchmarks.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Benchmarks/Compute/ComputeJobExecutorBenchmarks.cs
@@ -28,9 +28,9 @@ using Tests.TestHelpers;
/// Measures the performance of assembly loading and job execution.
/// Results on i9-12900H, .NET SDK 8.0.15, Ubuntu 22.04:
///
-/// | Method | Mean | Error | StdDev | Median | Gen0 | Gen1 | Gen2 | Allocated |
-/// |---------------- |---------:|--------:|---------:|---------:|-------:|-------:|-------:|----------:|
-/// | ExecuteJobAsync | 282.2 us | 9.58 us | 28.25 us | 267.0 us | 7.0801 | 7.0801 | 7.0801 | 4.64 KB |.
+/// | Method | Mean | Error | StdDev | Gen0 | Allocated |
+/// |---------------- |---------:|--------:|--------:|-------:|----------:|
+/// | ExecuteJobAsync | 412.7 ns | 3.36 ns | 2.81 ns | 0.0014 | 464 B |.
/// </summary>
[MemoryDiagnoser]
public class ComputeJobExecutorBenchmarks
diff --git a/modules/platforms/dotnet/Apache.Ignite.Benchmarks/Compute/PlatformJobBenchmarks.cs b/modules/platforms/dotnet/Apache.Ignite.Benchmarks/Compute/PlatformJobBenchmarks.cs
index aa2d7f4af9e..3112bc15019 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Benchmarks/Compute/PlatformJobBenchmarks.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Benchmarks/Compute/PlatformJobBenchmarks.cs
@@ -31,20 +31,18 @@ using Tests.TestHelpers;
/// Not entirely fair, because the Java job does not require class loading.
/// Results on i9-12900H, .NET SDK 8.0.15, Ubuntu 22.04:
///
-/// | Method | Mean | Error | StdDev | Median | Ratio | RatioSD |
-/// |---------- |---------:|---------:|---------:|---------:|------:|--------:|
-/// | DotNetJob | 776.0 us | 15.16 us | 28.09 us | 772.3 us | 2.86 | 0.40 |
-/// | JavaJob | 276.4 us | 14.63 us | 41.27 us | 258.6 us | 1.02 | 0.21 |
+/// | Method | Mean | Error | StdDev | Ratio | RatioSD |
+/// |---------- |---------:|---------:|---------:|------:|--------:|
+/// | DotNetJob | 170.1 us | 3.40 us | 9.30 us | 0.91 | 0.12 |
+/// | JavaJob | 191.2 us | 10.18 us | 27.34 us | 1.02 | 0.19 |
///
/// Notes:
-/// 1. Presence of a deployment unit in the job descriptor
-/// slows down the Java side of things by ~150us, even when no Java classes are loaded.
-/// 2. Dev mode executor uses Debug build (hardcoded path) - we should measure Release to be more fair.
-/// 3. Disable Netty leak detector.
+/// 1. Change DotNetComputeExecutor$resolveDotNetBinaryDir in Java to use Release executor in dev mode.
+/// 2. Disable Netty leak detector.
/// </summary>
public class PlatformJobBenchmarks : ServerBenchmarkBase
{
- private static readonly DeploymentUnit Unit = new(nameof(PlatformJobBenchmarks), "1.0.0");
+ private DeploymentUnit _unit = null!;
private JobDescriptor<object?, object?> _echoJobJava = null!;
@@ -62,15 +60,15 @@ public class PlatformJobBenchmarks : ServerBenchmarkBase
var asmName = nameof(PlatformJobBenchmarks);
var asmDll = JobGenerator.EmitEchoJob(_tempDir, asmName);
- await ManagementApi.UnitDeploy(Unit.Name, Unit.Version, [asmDll]);
+ _unit = await ManagementApi.UnitDeploy(nameof(PlatformJobBenchmarks), ManagementApi.GetRandomUnitVersion(), [asmDll]);
_echoJobDotNet = new JobDescriptor<object?, object?>(
JobClassName: $"TestNamespace.EchoJob, {asmName}",
- DeploymentUnits: [Unit],
+ DeploymentUnits: [_unit],
Options: new() { ExecutorType = JobExecutorType.DotNetSidecar });
_echoJobJava = new(
- "org.apache.ignite.internal.runner.app.client.ItThinClientComputeTest$EchoJob", [Unit]);
+ "org.apache.ignite.internal.runner.app.client.ItThinClientComputeTest$EchoJob", [_unit]);
var nodes = await Client.GetClusterNodesAsync();
var firstNode = nodes.Single(x => x.Name.EndsWith("PlatformTestNodeRunner", StringComparison.Ordinal));
@@ -81,7 +79,7 @@ public class PlatformJobBenchmarks : ServerBenchmarkBase
{
await base.GlobalCleanup();
- await ManagementApi.UnitUndeploy(Unit);
+ await ManagementApi.UnitUndeploy(_unit);
_tempDir.Dispose();
}
diff --git a/modules/platforms/dotnet/Apache.Ignite.Tests/Compute/Executor/JobLoadContextCacheTests.cs b/modules/platforms/dotnet/Apache.Ignite.Tests/Compute/Executor/JobLoadContextCacheTests.cs
new file mode 100644
index 00000000000..c68c1f092c9
--- /dev/null
+++ b/modules/platforms/dotnet/Apache.Ignite.Tests/Compute/Executor/JobLoadContextCacheTests.cs
@@ -0,0 +1,131 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.
+ */
+
+namespace Apache.Ignite.Tests.Compute.Executor;
+
+using System;
+using System.Threading.Tasks;
+using Internal.Compute.Executor;
+using NUnit.Framework;
+
+/// <summary>
+/// Tests for <see cref="JobLoadContextCache"/>.
+/// </summary>
+public class JobLoadContextCacheTests
+{
+ [Test]
+ public async Task TestGetOrAddJobLoadContextReturnsSameInstanceForSamePaths()
+ {
+ using var cache = new JobLoadContextCache();
+ var paths = new DeploymentUnitPaths(["foo", "bar"]);
+
+ var ctx1 = await cache.GetOrAddJobLoadContext(paths);
+ var ctx2 = await cache.GetOrAddJobLoadContext(paths);
+
+ Assert.AreSame(ctx1.AssemblyLoadContext, ctx2.AssemblyLoadContext);
+ }
+
+ [Test]
+ public async Task TestGetOrAddJobLoadContextReturnsDifferentInstanceForDifferentPaths()
+ {
+ using var cache = new JobLoadContextCache();
+
+ var paths1 = new DeploymentUnitPaths(["foo", "bar"]);
+ var paths2 = new DeploymentUnitPaths(["foo", "bar", "baz"]);
+
+ var ctx1 = await cache.GetOrAddJobLoadContext(paths1);
+ var ctx2 = await cache.GetOrAddJobLoadContext(paths2);
+
+ Assert.AreNotSame(ctx1.AssemblyLoadContext, ctx2.AssemblyLoadContext);
+ }
+
+ [Test]
+ public async Task TestExpiredContextsAreCleanedUp()
+ {
+ using var cache = new JobLoadContextCache(ttlMs: 100, cacheCleanupIntervalMs: 50);
+ var paths = new DeploymentUnitPaths(["x", "y", "z"]);
+
+ var ctx0 = await cache.GetOrAddJobLoadContext(paths);
+ var ctx1 = await cache.GetOrAddJobLoadContext(paths);
+
+ await Task.Delay(200);
+ var ctx2 = await cache.GetOrAddJobLoadContext(paths);
+
+ Assert.AreSame(ctx0.AssemblyLoadContext, ctx1.AssemblyLoadContext);
+ Assert.AreNotSame(ctx1.AssemblyLoadContext, ctx2.AssemblyLoadContext);
+ }
+
+ [Test]
+ public async Task TestFrequentlyUsedContextsAreNotCleanedUp()
+ {
+ using var cache = new JobLoadContextCache(ttlMs: 300, cacheCleanupIntervalMs: 10);
+ var paths1 = new DeploymentUnitPaths(["a", "b", "c"]);
+ var paths2 = new DeploymentUnitPaths(["c", "b", "a"]);
+
+ var ctx1 = await cache.GetOrAddJobLoadContext(paths1);
+ var ctx2 = await cache.GetOrAddJobLoadContext(paths2);
+
+ for (int i = 0; i < 33; i++)
+ {
+ await Task.Delay(millisecondsDelay: 10);
+
+ var ctx10 = await cache.GetOrAddJobLoadContext(paths1);
+ Assert.AreSame(ctx1.AssemblyLoadContext, ctx10.AssemblyLoadContext, "Iteration {0}", i);
+ }
+
+ // ctx2 was not used and should be cleaned up.
+ var ctx20 = await cache.GetOrAddJobLoadContext(paths2);
+ Assert.AreNotSame(ctx2.AssemblyLoadContext, ctx20.AssemblyLoadContext);
+ }
+
+ [Test]
+ public async Task TestUseAfterDisposeThrows()
+ {
+ var cache = new JobLoadContextCache();
+ var paths = new DeploymentUnitPaths(["a", "b"]);
+
+ await cache.GetOrAddJobLoadContext(paths);
+ cache.Dispose();
+
+ Assert.ThrowsAsync<ObjectDisposedException>(async () => await cache.GetOrAddJobLoadContext(paths));
+ }
+
+ [Test]
+ public async Task TestUndeployUnitsRemovesAllRelatedContexts()
+ {
+ using var cache = new JobLoadContextCache();
+
+ var paths1 = new DeploymentUnitPaths(["unit1", "unit2"]);
+ var paths2 = new DeploymentUnitPaths(["unit2", "unit3"]);
+ var paths3 = new DeploymentUnitPaths(["unit3", "unit4"]);
+
+ var ctx1 = await cache.GetOrAddJobLoadContext(paths1);
+ var ctx2 = await cache.GetOrAddJobLoadContext(paths2);
+ var ctx3 = await cache.GetOrAddJobLoadContext(paths3);
+
+ await cache.UndeployUnits(["unit1", "unit2"]);
+
+ var ctx10 = await cache.GetOrAddJobLoadContext(paths1);
+ var ctx20 = await cache.GetOrAddJobLoadContext(paths2);
+ var ctx30 = await cache.GetOrAddJobLoadContext(paths3);
+
+ // ctx1 and ctx2 should be disposed, ctx3 should remain intact.
+ Assert.AreNotSame(ctx1.AssemblyLoadContext, ctx10.AssemblyLoadContext);
+ Assert.AreNotSame(ctx2.AssemblyLoadContext, ctx20.AssemblyLoadContext);
+ Assert.AreSame(ctx3.AssemblyLoadContext, ctx30.AssemblyLoadContext);
+ }
+}
diff --git a/modules/platforms/dotnet/Apache.Ignite.Tests/TestHelpers/ManagementApi.cs b/modules/platforms/dotnet/Apache.Ignite.Tests/TestHelpers/ManagementApi.cs
index 4c4d3a51782..da584fb2758 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Tests/TestHelpers/ManagementApi.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Tests/TestHelpers/ManagementApi.cs
@@ -106,7 +106,7 @@ public static class ManagementApi
var testsDll = typeof(ManagementApi).Assembly.Location;
var unitId0 = unitId ?? TestContext.CurrentContext.Test.FullName;
- var unitVersion0 = unitVersion ?? DateTime.Now.TimeOfDay.ToString(@"m\.s\.f");
+ var unitVersion0 = unitVersion ?? GetRandomUnitVersion();
return await UnitDeploy(
unitId: unitId0,
@@ -114,6 +114,8 @@ public static class ManagementApi
unitContent: [testsDll]);
}
+ public static string GetRandomUnitVersion() => DateTime.Now.TimeOfDay.ToString(@"m\.s\.f");
+
private static async Task<DeploymentUnitStatus[]?> GetUnitStatus(string unitId)
{
using var client = new HttpClient();
diff --git a/modules/platforms/dotnet/Apache.Ignite/Internal/ClientSocket.Server.cs b/modules/platforms/dotnet/Apache.Ignite/Internal/ClientSocket.Server.cs
index 54112bc4782..6b74a0fbf63 100644
--- a/modules/platforms/dotnet/Apache.Ignite/Internal/ClientSocket.Server.cs
+++ b/modules/platforms/dotnet/Apache.Ignite/Internal/ClientSocket.Server.cs
@@ -40,7 +40,7 @@ internal sealed partial class ClientSocket
switch (op)
{
case ServerOp.Ping:
- // No-op.
+ response.MessageWriter.Write(0); // Response flags: success.
break;
case ServerOp.ComputeJobExec:
@@ -50,11 +50,14 @@ internal sealed partial class ClientSocket
case ServerOp.ComputeJobCancel:
// TODO IGNITE-25153: Add cancellation support for platform jobs.
+ response.MessageWriter.Write(0); // Response flags: success.
response.MessageWriter.Write(false);
break;
case ServerOp.DeploymentUnitsUndeploy:
- response.MessageWriter.Write(false);
+ var res = await ComputeJobExecutor.UndeployUnits(request).ConfigureAwait(false);
+ response.MessageWriter.Write(0); // Response flags: success.
+ response.MessageWriter.Write(res);
break;
default:
diff --git a/modules/platforms/dotnet/Apache.Ignite/Internal/Compute/Executor/ComputeJobExecutor.cs b/modules/platforms/dotnet/Apache.Ignite/Internal/Compute/Executor/ComputeJobExecutor.cs
index 73c9be23e68..1f79b01f0fc 100644
--- a/modules/platforms/dotnet/Apache.Ignite/Internal/Compute/Executor/ComputeJobExecutor.cs
+++ b/modules/platforms/dotnet/Apache.Ignite/Internal/Compute/Executor/ComputeJobExecutor.cs
@@ -23,6 +23,7 @@ using System.Diagnostics.CodeAnalysis;
using System.Threading;
using System.Threading.Tasks;
using Buffers;
+using Proto.MsgPack;
using Table.StreamerReceiverExecutor;
/// <summary>
@@ -35,6 +36,8 @@ internal static class ComputeJobExecutor
/// </summary>
internal static readonly string? IgniteComputeExecutorId = Environment.GetEnvironmentVariable("IGNITE_COMPUTE_EXECUTOR_ID");
+ private static readonly JobLoadContextCache Cache = new JobLoadContextCache();
+
/// <summary>
/// Executes compute job.
/// </summary>
@@ -55,21 +58,8 @@ internal static class ComputeJobExecutor
var r = request.GetReader();
long jobId = r.ReadInt64();
string jobClassName = r.ReadString();
-
- int cnt = r.ReadInt32();
- List<string> deploymentUnitPaths = new List<string>(cnt);
- for (int i = 0; i < cnt; i++)
- {
- deploymentUnitPaths.Add(r.ReadString());
- }
-
- bool retainDeploymentUnits = r.ReadBoolean();
-
- if (retainDeploymentUnits)
- {
- // TODO IGNITE-25257 Cache deployment units and JobLoadContext.
- throw new NotSupportedException("Caching deployment units is not supported yet.");
- }
+ List<string> deploymentUnitPaths = ReadDeploymentUnitPaths(ref r);
+ _ = r.ReadBoolean(); // Retain deployment units.
request.Position += r.Consumed;
@@ -77,15 +67,29 @@ internal static class ComputeJobExecutor
}
}
+ /// <summary>
+ /// Cleans up deployment units.
+ /// </summary>
+ /// <param name="request">Request.</param>
+ /// <returns>Whether units were cleaned up.</returns>
+ internal static async ValueTask<bool> UndeployUnits(PooledBuffer request)
+ {
+ return await Cache.UndeployUnits(Read()).ConfigureAwait(false);
+
+ List<string> Read()
+ {
+ var r = request.GetReader();
+ return ReadDeploymentUnitPaths(ref r);
+ }
+ }
+
private static async ValueTask ExecuteJobAsync(
JobExecuteRequest req,
PooledBuffer argBuf,
PooledArrayBuffer resBuf,
IgniteApiAccessor context)
{
- // Unload assemblies after job execution.
- // TODO IGNITE-25257 Cache deployment units and JobLoadContext - see ComputeJobExecutorBenchmarks, expensive.
- using JobLoadContext jobLoadCtx = DeploymentUnitLoader.GetJobLoadContext(req.DeploymentUnitPaths);
+ JobLoadContext jobLoadCtx = await Cache.GetOrAddJobLoadContext(req.DeploymentUnitPaths).ConfigureAwait(false);
resBuf.MessageWriter.Write(0); // Response flags: success.
@@ -103,6 +107,20 @@ internal static class ComputeJobExecutor
await jobWrapper.ExecuteAsync(context, argBuf, resBuf, CancellationToken.None).ConfigureAwait(false);
}
+ [SuppressMessage("Design", "CA1002:Do not expose generic lists", Justification = "Internal.")]
+ private static List<string> ReadDeploymentUnitPaths(ref MsgPackReader r)
+ {
+ int cnt = r.ReadInt32();
+ List<string> deploymentUnitPaths = new List<string>(cnt);
+
+ for (int i = 0; i < cnt; i++)
+ {
+ deploymentUnitPaths.Add(r.ReadString());
+ }
+
+ return deploymentUnitPaths;
+ }
+
[SuppressMessage("ReSharper", "NotAccessedPositionalProperty.Local", Justification = "DTO.")]
private record JobExecuteRequest(long JobId, DeploymentUnitPaths DeploymentUnitPaths, string JobClassName);
}
diff --git a/modules/platforms/dotnet/Apache.Ignite/Internal/Compute/Executor/DeploymentUnitPaths.cs b/modules/platforms/dotnet/Apache.Ignite/Internal/Compute/Executor/DeploymentUnitPaths.cs
index ccbda47c64c..b301299451c 100644
--- a/modules/platforms/dotnet/Apache.Ignite/Internal/Compute/Executor/DeploymentUnitPaths.cs
+++ b/modules/platforms/dotnet/Apache.Ignite/Internal/Compute/Executor/DeploymentUnitPaths.cs
@@ -27,8 +27,7 @@ using System.Text;
/// - A set of deployment unit paths is considered equal if it contains the same paths in the same order.
/// - Unit order affects assembly resolution order.
/// - A given deployment unit (name+ver) is always located at the same path (after it is uploaded to the cluster).
-/// Example: ignite-3/target/work/PlatformTestNodeRunner/o.a.i.internal.runner.app.PlatformTestNodeRunner/deployment/UNIT_NAME/UNIT_VER
-/// The facts above can be used to cache load contexts (IGNITE-25257).
+/// Example: ignite-3/target/work/PlatformTestNodeRunner/o.a.i.internal.runner.app.PlatformTestNodeRunner/deployment/UNIT_NAME/UNIT_VER.
/// </summary>
/// <param name="Paths">The list of deployment unit paths.</param>
internal readonly record struct DeploymentUnitPaths(IReadOnlyList<string> Paths)
diff --git a/modules/platforms/dotnet/Apache.Ignite/Internal/Compute/Executor/JobLoadContext.cs b/modules/platforms/dotnet/Apache.Ignite/Internal/Compute/Executor/JobLoadContext.cs
index 13ed4c60b6b..bfd625a5ba9 100644
--- a/modules/platforms/dotnet/Apache.Ignite/Internal/Compute/Executor/JobLoadContext.cs
+++ b/modules/platforms/dotnet/Apache.Ignite/Internal/Compute/Executor/JobLoadContext.cs
@@ -18,6 +18,7 @@
namespace Apache.Ignite.Internal.Compute.Executor;
using System;
+using System.Collections.Concurrent;
using System.Reflection;
using System.Runtime.Loader;
using Ignite.Compute;
@@ -30,6 +31,8 @@ using Table.StreamerReceiverExecutor;
/// <param name="AssemblyLoadContext">Assembly load context.</param>
internal readonly record struct JobLoadContext(AssemblyLoadContext AssemblyLoadContext) : IDisposable
{
+ private readonly ConcurrentDictionary<(string TypeName, Type OpenInterfaceType), (Type Type, Type ClosedWrapperType)> _typeCache = new();
+
/// <summary>
/// Gets or creates a job delegate for the specified type name.
/// </summary>
@@ -37,7 +40,7 @@ internal readonly record struct JobLoadContext(AssemblyLoadContext AssemblyLoadC
/// <returns>Job execution delegate.</returns>
public IComputeJobWrapper CreateJobWrapper(string typeName) =>
CreateWrapper<IComputeJobWrapper>(
- typeName, typeof(IComputeJob<,>), typeof(ComputeJobWrapper<,,>), AssemblyLoadContext);
+ typeName, typeof(IComputeJob<,>), typeof(ComputeJobWrapper<,,>));
/// <summary>
/// Gets or creates a receiver delegate for the specified type name.
@@ -46,30 +49,54 @@ internal readonly record struct JobLoadContext(AssemblyLoadContext AssemblyLoadC
/// <returns>Receiver execution delegate.</returns>
public IDataStreamerReceiverWrapper CreateReceiverWrapper(string typeName) =>
CreateWrapper<IDataStreamerReceiverWrapper>(
- typeName, typeof(IDataStreamerReceiver<,,>), typeof(DataStreamerReceiverWrapper<,,,>), AssemblyLoadContext);
+ typeName, typeof(IDataStreamerReceiver<,,>), typeof(DataStreamerReceiverWrapper<,,,>));
/// <inheritdoc/>
public void Dispose() => AssemblyLoadContext.Unload();
- private static T CreateWrapper<T>(string wrappedTypeName, Type openInterfaceType, Type openWrapperType, AssemblyLoadContext ctx)
+ private T CreateWrapper<T>(string wrappedTypeName, Type openInterfaceType, Type openWrapperType)
+ {
+ var (type, closedWrapperType) = _typeCache.GetOrAdd(
+ key: (wrappedTypeName, openInterfaceType),
+ valueFactory: static (key, arg) =>
+ GetClosedWrapperType(key.TypeName, key.OpenInterfaceType, arg.openWrapperType, arg.AssemblyLoadContext),
+ factoryArgument: (openWrapperType, AssemblyLoadContext));
+
+ try
+ {
+ return (T)Activator.CreateInstance(closedWrapperType)!;
+ }
+ catch (Exception e)
+ {
+ CheckPublicCtor(type, e);
+ throw;
+ }
+ }
+
+ private static void CheckPublicCtor(Type type, Exception e)
{
- var type = LoadType(wrappedTypeName, ctx);
+ if (type.GetConstructor(BindingFlags.Public, []) == null)
+ {
+ throw new InvalidOperationException($"No public parameterless constructor for type '{type.AssemblyQualifiedName}'", e);
+ }
+ }
+
+ private static (Type Type, Type ClosedWrapperType) GetClosedWrapperType(
+ string typeName, Type openInterfaceType, Type openWrapperType, AssemblyLoadContext ctx)
+ {
+ var type = LoadType(typeName, ctx);
var closedInterfaceType = FindInterface(type, openInterfaceType);
try
{
var genericArgs = closedInterfaceType.GenericTypeArguments;
- var jobWrapperType = openWrapperType.MakeGenericType([type, .. genericArgs]);
+ var closedWrapperType = openWrapperType.MakeGenericType([type, .. genericArgs]);
- return (T)Activator.CreateInstance(jobWrapperType)!;
+ return (type, closedWrapperType);
}
catch (Exception e)
{
- if (type.GetConstructor(BindingFlags.Public, []) == null)
- {
- throw new InvalidOperationException($"No public parameterless constructor for type '{wrappedTypeName}'", e);
- }
-
+ CheckPublicCtor(type, e);
throw;
}
}
diff --git a/modules/platforms/dotnet/Apache.Ignite/Internal/Compute/Executor/JobLoadContextCache.cs b/modules/platforms/dotnet/Apache.Ignite/Internal/Compute/Executor/JobLoadContextCache.cs
new file mode 100644
index 00000000000..9b545d4cdae
--- /dev/null
+++ b/modules/platforms/dotnet/Apache.Ignite/Internal/Compute/Executor/JobLoadContextCache.cs
@@ -0,0 +1,241 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.
+ */
+
+namespace Apache.Ignite.Internal.Compute.Executor;
+
+using System;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.Diagnostics.CodeAnalysis;
+using System.Runtime.InteropServices;
+using System.Threading;
+using System.Threading.Tasks;
+
+/// <summary>
+/// Cache for job load contexts.
+/// <para />
+/// Every job load context is associated with a set of deployment unit paths. For example, [unit1, unit2] and [unit1, unit3] sets
+/// represent a different set of assemblies, so they will have different job load contexts. Unit order matters too.
+/// <para />
+/// Loading assemblies and resolving types can take from ~100us to several seconds, depending on the number of assemblies,
+/// type initializers (static constructors), and other factors.
+/// <para />
+/// If a deployment unit U is undeployed, all job load contexts with U in their set will be removed from the cache immediately.
+/// </summary>
+internal sealed class JobLoadContextCache : IDisposable
+{
+ private static readonly long TicksPerMs = Stopwatch.Frequency / 1000L;
+
+ private readonly long _ttlTicks;
+
+ /** Main cache of job load contexts. */
+ private readonly Dictionary<DeploymentUnitPaths, (JobLoadContext Ctx, long Ts)> _jobLoadContextCache = new();
+
+ /** Additional map to quickly find all job load contexts that use a given deployment unit path. */
+ private readonly Dictionary<string, List<DeploymentUnitPaths>> _deploymentUnitSets = new();
+
+ private readonly SemaphoreSlim _cacheLock = new(1);
+
+ private readonly CancellationTokenSource _cts = new();
+
+ /// <summary>
+ /// Initializes a new instance of the <see cref="JobLoadContextCache"/> class.
+ /// </summary>
+ /// <param name="ttlMs">Cache entry time-to-live in milliseconds.</param>
+ /// <param name="cacheCleanupIntervalMs">Cache cleanup interval.</param>
+ internal JobLoadContextCache(int ttlMs = 30_000, int cacheCleanupIntervalMs = 5_000)
+ {
+ _ttlTicks = ttlMs * TicksPerMs;
+
+ _ = StartCacheCleanupAsync(cacheCleanupIntervalMs);
+ }
+
+ /// <summary>
+ /// Gets or adds a job load context for the specified deployment unit paths.
+ /// </summary>
+ /// <param name="paths">Deployment unit paths.</param>
+ /// <returns>Job load context.</returns>
+ public async ValueTask<JobLoadContext> GetOrAddJobLoadContext(DeploymentUnitPaths paths)
+ {
+ await _cacheLock.WaitAsync().ConfigureAwait(false);
+
+ try
+ {
+ return GetOrAddImpl();
+ }
+ finally
+ {
+ _cacheLock.Release();
+ }
+
+ JobLoadContext GetOrAddImpl()
+ {
+ ref (JobLoadContext Ctx, long Ts) valRef = ref CollectionsMarshal.GetValueRefOrAddDefault(_jobLoadContextCache, paths, out var exists);
+
+ if (!exists)
+ {
+ valRef.Ctx = DeploymentUnitLoader.GetJobLoadContext(paths);
+
+ foreach (var path in paths.Paths)
+ {
+ ref var listRef = ref CollectionsMarshal.GetValueRefOrAddDefault(_deploymentUnitSets, path, out _);
+ listRef ??= new List<DeploymentUnitPaths>();
+
+ listRef.Add(paths);
+ }
+ }
+
+ valRef.Ts = NowTicks();
+
+ return valRef.Ctx;
+ }
+ }
+
+ /// <summary>
+ /// Un-deploys the specified deployment unit paths and cleans up associated job load contexts.
+ /// </summary>
+ /// <param name="deploymentUnitPaths">Deployment unit paths to undeploy.</param>
+ /// <returns><see cref="Task"/> representing the asynchronous operation.</returns>
+ public async ValueTask<bool> UndeployUnits(ICollection<string> deploymentUnitPaths)
+ {
+ await _cacheLock.WaitAsync().ConfigureAwait(false);
+
+ try
+ {
+ bool res = false;
+
+ foreach (var deploymentUnitPath in deploymentUnitPaths)
+ {
+ if (!_deploymentUnitSets.TryGetValue(deploymentUnitPath, out var unitSet))
+ {
+ continue;
+ }
+
+ foreach (DeploymentUnitPaths paths in unitSet)
+ {
+ if (_jobLoadContextCache.TryGetValue(paths, out var cachedJobCtx))
+ {
+ cachedJobCtx.Ctx.Dispose();
+ _jobLoadContextCache.Remove(paths);
+ }
+ }
+
+ _deploymentUnitSets.Remove(deploymentUnitPath);
+ res = true;
+ }
+
+ return res;
+ }
+ finally
+ {
+ _cacheLock.Release();
+ }
+ }
+
+ /// <inheritdoc/>
+ public void Dispose()
+ {
+ _cts.Cancel();
+
+ _cacheLock.Wait();
+
+ try
+ {
+ foreach (var cachedJobCtx in _jobLoadContextCache)
+ {
+ cachedJobCtx.Value.Ctx.Dispose();
+ }
+
+ _jobLoadContextCache.Clear();
+ _deploymentUnitSets.Clear();
+ }
+ finally
+ {
+ _cacheLock.Release();
+ _cacheLock.Dispose();
+ _cts.Dispose();
+ }
+ }
+
+ private static long NowTicks() => Stopwatch.GetTimestamp();
+
+ [SuppressMessage("Design", "CA1031:Do not catch general exception types", Justification = "Thread root.")]
+ private async Task StartCacheCleanupAsync(int cacheCleanupIntervalMs)
+ {
+ var ct = _cts.Token;
+
+ while (!ct.IsCancellationRequested)
+ {
+ await Task.Delay(cacheCleanupIntervalMs, ct).ConfigureAwait(false);
+
+ await _cacheLock.WaitAsync(ct).ConfigureAwait(false);
+
+ try
+ {
+ CleanUpExpiredJobContexts();
+ }
+ catch (Exception)
+ {
+ // Ignore.
+ }
+ finally
+ {
+ _cacheLock.Release();
+ }
+ }
+ }
+
+ private void CleanUpExpiredJobContexts()
+ {
+ List<KeyValuePair<DeploymentUnitPaths, (JobLoadContext Ctx, long Ts)>>? toRemove = null;
+ long nowTicks = NowTicks();
+
+ foreach (KeyValuePair<DeploymentUnitPaths, (JobLoadContext Ctx, long Ts)> cachedJobCtx in _jobLoadContextCache)
+ {
+ if (cachedJobCtx.Value.Ts + _ttlTicks < nowTicks)
+ {
+ toRemove ??= new();
+ toRemove.Add(cachedJobCtx);
+ }
+ }
+
+ if (toRemove is null)
+ {
+ return;
+ }
+
+ foreach (KeyValuePair<DeploymentUnitPaths, (JobLoadContext Ctx, long Ts)> cachedJobCtx in toRemove)
+ {
+ DeploymentUnitPaths paths = cachedJobCtx.Key;
+ _jobLoadContextCache.Remove(paths);
+ cachedJobCtx.Value.Ctx.Dispose();
+
+ foreach (var unitPath in paths.Paths)
+ {
+ if (_deploymentUnitSets.TryGetValue(unitPath, out var unitSet))
+ {
+ unitSet.Remove(paths);
+
+ if (unitSet.Count == 0)
+ {
+ _deploymentUnitSets.Remove(unitPath);
+ }
+ }
+ }
+ }
+ }
+}
diff --git a/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java b/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
index 71eecf0324c..03e3537ff47 100644
--- a/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
+++ b/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
@@ -1195,6 +1195,11 @@ public class IgniteImpl implements Ignite {
sql = new IgniteSqlImpl(qryEngine, observableTimestampTracker, threadPoolsManager.commonScheduler());
+ ComputeConfiguration computeCfg = nodeConfigRegistry.getConfiguration(ComputeExtensionConfiguration.KEY).compute();
+ InMemoryComputeStateMachine stateMachine = new InMemoryComputeStateMachine(computeCfg, name);
+ ComputeExecutorImpl computeExecutor = new ComputeExecutorImpl(
+ this, stateMachine, computeCfg, clusterSvc.topologyService(), clockService);
+
var deploymentManagerImpl = new DeploymentManagerImpl(
clusterSvc,
new DeploymentUnitStoreImpl(metaStorageMgr),
@@ -1202,15 +1207,11 @@ public class IgniteImpl implements Ignite {
workDir,
nodeConfigRegistry.getConfiguration(DeploymentExtensionConfiguration.KEY).deployment(),
cmgMgr,
- name
+ name,
+ computeExecutor::onUnitRemoving // TODO IGNITE-25759 cleaner platform unit removal.
);
deploymentManager = deploymentManagerImpl;
- ComputeConfiguration computeCfg = nodeConfigRegistry.getConfiguration(ComputeExtensionConfiguration.KEY).compute();
- InMemoryComputeStateMachine stateMachine = new InMemoryComputeStateMachine(computeCfg, name);
- ComputeExecutorImpl computeExecutor = new ComputeExecutorImpl(
- this, stateMachine, computeCfg, clusterSvc.topologyService(), clockService);
-
computeComponent = new ComputeComponentImpl(
name,
clusterSvc.messagingService(),