Skip to content

[MRESOLVER-283] Shared executor service #213

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
cstamas wants to merge 3 commits into from
Closed

[MRESOLVER-283] Shared executor service #213

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
fb96453
[MRESOLVER-283] Shared executor service
cstamas Nov 2, 2022
d5f7403
Do not SHARE
cstamas Nov 8, 2022
39970d0
runnable -> runnables
cstamas Nov 8, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
89 changes: 30 additions & 59 deletions ...ctor-basic/src/main/java/org/eclipse/aether/connector/basic/BasicRepositoryConnector.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,17 +31,14 @@
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import org.eclipse.aether.ConfigurationProperties;
import org.eclipse.aether.RepositorySystemSession;
import org.eclipse.aether.RequestTrace;
import org.eclipse.aether.repository.RemoteRepository;
import org.eclipse.aether.spi.concurrency.ResolverExecutor;
import org.eclipse.aether.spi.concurrency.ResolverExecutorService;
import org.eclipse.aether.spi.connector.ArtifactDownload;
import org.eclipse.aether.spi.connector.ArtifactUpload;
import org.eclipse.aether.spi.connector.MetadataDownload;
Expand Down Expand Up @@ -70,7 +67,6 @@
import org.eclipse.aether.util.ConfigUtils;
import org.eclipse.aether.util.FileUtils;
import org.eclipse.aether.util.concurrency.RunnableErrorForwarder;
import org.eclipse.aether.util.concurrency.WorkerThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

Expand All @@ -81,14 +77,24 @@ final class BasicRepositoryConnector
implements RepositoryConnector
{

/**
* The count of threads to be used when downloading artifacts in parallel, default value 5.
*/
private static final String CONFIG_PROP_THREADS = "aether.connector.basic.threads";

/**
* The default value for {@link #CONFIG_PROP_THREADS}.
*/
private static final int CONFIG_PROP_THREADS_DEFAULT = 5;

private static final String CONFIG_PROP_SMART_CHECKSUMS = "aether.connector.smartChecksums";

private static final Logger LOGGER = LoggerFactory.getLogger( BasicRepositoryConnector.class );

private final Map<String, ProvidedChecksumsSource> providedChecksumsSources;

private final ResolverExecutor resolverExecutor;

private final FileProcessor fileProcessor;

private final RemoteRepository repository;
Expand All @@ -101,23 +107,21 @@ final class BasicRepositoryConnector

private final ChecksumPolicyProvider checksumPolicyProvider;

private final int maxThreads;

private final boolean smartChecksums;

private final boolean persistedChecksums;

private Executor executor;

private final AtomicBoolean closed;

@SuppressWarnings( "checkstyle:parameternumber" )
BasicRepositoryConnector( RepositorySystemSession session,
RemoteRepository repository,
TransporterProvider transporterProvider,
RepositoryLayoutProvider layoutProvider,
ChecksumPolicyProvider checksumPolicyProvider,
FileProcessor fileProcessor,
Map<String, ProvidedChecksumsSource> providedChecksumsSources )
Map<String, ProvidedChecksumsSource> providedChecksumsSources,
ResolverExecutorService resolverExecutorService )
throws NoRepositoryConnectorException
{
try
Expand All @@ -143,36 +147,17 @@ final class BasicRepositoryConnector
this.fileProcessor = fileProcessor;
this.providedChecksumsSources = providedChecksumsSources;
this.closed = new AtomicBoolean( false );
this.resolverExecutor = resolverExecutorService.getResolverExecutor(
resolverExecutorService.getName( BasicRepositoryConnector.class, repository.getId() ),
ConfigUtils.getInteger( session, CONFIG_PROP_THREADS_DEFAULT, CONFIG_PROP_THREADS,
"maven.artifact.threads" ) );
Copy link
Copy Markdown
Member

@michael-o michael-o Nov 6, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We should also deprecate this property since it does not reflect reality. MD is not artifacts, but this connector does both.

Copy link
Copy Markdown
Member Author

@cstamas cstamas Nov 7, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe we should just remove it? We are going for 1.9 after all....


maxThreads = ConfigUtils.getInteger( session, 5, CONFIG_PROP_THREADS, "maven.artifact.threads" );
smartChecksums = ConfigUtils.getBoolean( session, true, CONFIG_PROP_SMART_CHECKSUMS );
persistedChecksums =
this.smartChecksums = ConfigUtils.getBoolean( session, true, CONFIG_PROP_SMART_CHECKSUMS );
this.persistedChecksums =
ConfigUtils.getBoolean( session, ConfigurationProperties.DEFAULT_PERSISTED_CHECKSUMS,
ConfigurationProperties.PERSISTED_CHECKSUMS );
}

private Executor getExecutor( Collection<?> artifacts, Collection<?> metadatas )
{
if ( maxThreads <= 1 )
{
return DirectExecutor.INSTANCE;
}
int tasks = safe( artifacts ).size() + safe( metadatas ).size();
if ( tasks <= 1 )
{
return DirectExecutor.INSTANCE;
}
if ( executor == null )
{
executor =
new ThreadPoolExecutor( maxThreads, maxThreads, 3L, TimeUnit.SECONDS,
new LinkedBlockingQueue<>(),
new WorkerThreadFactory( getClass().getSimpleName() + '-'
+ repository.getHost() + '-' ) );
}
return executor;
}

@Override
protected void finalize()
throws Throwable
Expand All @@ -192,10 +177,7 @@ public void close()
{
if ( closed.compareAndSet( false, true ) )
{
if ( executor instanceof ExecutorService )
{
( (ExecutorService) executor ).shutdown();
}
resolverExecutor.close();
transporter.close();
}
}
Expand All @@ -214,11 +196,13 @@ public void get( Collection<? extends ArtifactDownload> artifactDownloads,
{
failIfClosed();

Executor executor = getExecutor( artifactDownloads, metadataDownloads );
RunnableErrorForwarder errorForwarder = new RunnableErrorForwarder();
List<ChecksumAlgorithmFactory> checksumAlgorithmFactories = layout.getChecksumAlgorithmFactories();
Collection<? extends MetadataDownload> mds = safe( metadataDownloads );
Collection<? extends ArtifactDownload> ads = safe( artifactDownloads );
ArrayList<Runnable> runnables = new ArrayList<>( mds.size() + ads.size() );

for ( MetadataDownload transfer : safe( metadataDownloads ) )
for ( MetadataDownload transfer : mds )
{
URI location = layout.getLocation( transfer.getMetadata(), false );

Expand All @@ -235,10 +219,10 @@ public void get( Collection<? extends ArtifactDownload> artifactDownloads,

Runnable task = new GetTaskRunner( location, transfer.getFile(), checksumPolicy,
checksumAlgorithmFactories, checksumLocations, null, listener );
executor.execute( errorForwarder.wrap( task ) );
runnables.add( errorForwarder.wrap( task ) );
}

for ( ArtifactDownload transfer : safe( artifactDownloads ) )
for ( ArtifactDownload transfer : ads )
{
Map<String, String> providedChecksums = Collections.emptyMap();
for ( ProvidedChecksumsSource providedChecksumsSource : providedChecksumsSources.values() )
Expand Down Expand Up @@ -276,9 +260,10 @@ public void get( Collection<? extends ArtifactDownload> artifactDownloads,
task = new GetTaskRunner( location, transfer.getFile(), checksumPolicy,
checksumAlgorithmFactories, checksumLocations, providedChecksums, listener );
}
executor.execute( errorForwarder.wrap( task ) );
runnables.add( errorForwarder.wrap( task ) );
}

resolverExecutor.submitBatch( runnables );
errorForwarder.await();
}

Expand Down Expand Up @@ -622,18 +607,4 @@ private void uploadChecksum( URI location, Object checksum )
}

}

private static class DirectExecutor
implements Executor
{

static final Executor INSTANCE = new DirectExecutor();

@Override
public void execute( Runnable command )
{
command.run();
}

Copy link
Copy Markdown
Contributor

@caiwei-ebay caiwei-ebay Nov 3, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Given this being deleted, I think it may introduce additional cost for DF.
DF downloads pom one by one, originally it will be downloaded by main thread directly and now it will be downloaded in a thread and the main thread has to wait

Copy link
Copy Markdown
Member Author

@cstamas cstamas Nov 3, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

reintroduced direct strategy as well, two out of 3 users of resolver executor now relies on it.

}
}
25 changes: 23 additions & 2 deletions ...sic/src/main/java/org/eclipse/aether/connector/basic/BasicRepositoryConnectorFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@

import org.eclipse.aether.RepositorySystemSession;
import org.eclipse.aether.repository.RemoteRepository;
import org.eclipse.aether.spi.concurrency.ResolverExecutorService;
import org.eclipse.aether.spi.connector.RepositoryConnector;
import org.eclipse.aether.spi.connector.RepositoryConnectorFactory;
import org.eclipse.aether.spi.connector.checksum.ChecksumPolicyProvider;
Expand Down Expand Up @@ -59,6 +60,8 @@ public final class BasicRepositoryConnectorFactory

private Map<String, ProvidedChecksumsSource> providedChecksumsSources;

private ResolverExecutorService resolverExecutorService;

private float priority;

/**
Expand All @@ -76,13 +79,15 @@ public BasicRepositoryConnectorFactory()
RepositoryLayoutProvider layoutProvider,
ChecksumPolicyProvider checksumPolicyProvider,
FileProcessor fileProcessor,
Map<String, ProvidedChecksumsSource> providedChecksumsSources )
Map<String, ProvidedChecksumsSource> providedChecksumsSources,
ResolverExecutorService resolverExecutorService )
{
setTransporterProvider( transporterProvider );
setRepositoryLayoutProvider( layoutProvider );
setChecksumPolicyProvider( checksumPolicyProvider );
setFileProcessor( fileProcessor );
setProvidedChecksumSources( providedChecksumsSources );
setResolverExecutorService( resolverExecutorService );
}

public void initService( ServiceLocator locator )
Expand All @@ -92,6 +97,7 @@ public void initService( ServiceLocator locator )
setChecksumPolicyProvider( locator.getService( ChecksumPolicyProvider.class ) );
setFileProcessor( locator.getService( FileProcessor.class ) );
setProvidedChecksumSources( Collections.emptyMap() );
setResolverExecutorService( locator.getService( ResolverExecutorService.class ) );
}

/**
Expand Down Expand Up @@ -159,6 +165,21 @@ public BasicRepositoryConnectorFactory setProvidedChecksumSources(
return this;
}

/**
* Sets the resolver executor to use for this component.
*
* @param resolverExecutorService The resolver executor to use, must not be {@code null}.
* @return This component for chaining, never {@code null}.
* @since 1.9.0
*/
public BasicRepositoryConnectorFactory setResolverExecutorService( ResolverExecutorService resolverExecutorService )
{
this.resolverExecutorService = requireNonNull(
resolverExecutorService, "resolver executor service cannot be null"
);
return this;
}

public float getPriority()
{
return priority;
Expand All @@ -183,7 +204,7 @@ public RepositoryConnector newInstance( RepositorySystemSession session, RemoteR
requireNonNull( repository, "repository cannot be null" );

return new BasicRepositoryConnector( session, repository, transporterProvider, layoutProvider,
checksumPolicyProvider, fileProcessor, providedChecksumsSources );
checksumPolicyProvider, fileProcessor, providedChecksumsSources, resolverExecutorService );
}

}
3 changes: 3 additions & 0 deletions maven-resolver-impl/src/main/java/org/eclipse/aether/impl/DefaultServiceLocator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,9 +57,11 @@
import org.eclipse.aether.internal.impl.EnhancedLocalRepositoryManagerFactory;
import org.eclipse.aether.internal.impl.Maven2RepositoryLayoutFactory;
import org.eclipse.aether.internal.impl.SimpleLocalRepositoryManagerFactory;
import org.eclipse.aether.internal.impl.concurrency.DefaultResolverExecutorService;
import org.eclipse.aether.internal.impl.filter.DefaultRemoteRepositoryFilterManager;
import org.eclipse.aether.internal.impl.slf4j.Slf4jLoggerFactory;
import org.eclipse.aether.internal.impl.synccontext.DefaultSyncContextFactory;
import org.eclipse.aether.spi.concurrency.ResolverExecutorService;
import org.eclipse.aether.spi.connector.checksum.ChecksumAlgorithmFactorySelector;
import org.eclipse.aether.spi.connector.checksum.ChecksumPolicyProvider;
import org.eclipse.aether.spi.connector.layout.RepositoryLayoutFactory;
Expand Down Expand Up @@ -231,6 +233,7 @@ public DefaultServiceLocator()
addService( LocalPathComposer.class, DefaultLocalPathComposer.class );
addService( RemoteRepositoryFilterManager.class, DefaultRemoteRepositoryFilterManager.class );
addService( RepositorySystemLifecycle.class, DefaultRepositorySystemLifecycle.class );
addService( ResolverExecutorService.class, DefaultResolverExecutorService.class );
}

private <T> Entry<T> getEntry( Class<T> type, boolean create )
Expand Down
5 changes: 5 additions & 0 deletions maven-resolver-impl/src/main/java/org/eclipse/aether/impl/guice/AetherModule.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,7 @@
import org.eclipse.aether.internal.impl.collect.DependencyCollectorDelegate;
import org.eclipse.aether.internal.impl.collect.bf.BfDependencyCollector;
import org.eclipse.aether.internal.impl.collect.df.DfDependencyCollector;
import org.eclipse.aether.internal.impl.concurrency.DefaultResolverExecutorService;
import org.eclipse.aether.internal.impl.filter.DefaultRemoteRepositoryFilterManager;
import org.eclipse.aether.internal.impl.filter.GroupIdRemoteRepositoryFilterSource;
import org.eclipse.aether.internal.impl.filter.PrefixesRemoteRepositoryFilterSource;
Expand Down Expand Up @@ -102,6 +103,7 @@
import org.eclipse.aether.internal.impl.slf4j.Slf4jLoggerFactory;
import org.eclipse.aether.named.providers.NoopNamedLockFactory;
import org.eclipse.aether.spi.checksums.TrustedChecksumsSource;
import org.eclipse.aether.spi.concurrency.ResolverExecutorService;
import org.eclipse.aether.spi.connector.checksum.ProvidedChecksumsSource;
import org.eclipse.aether.spi.connector.checksum.ChecksumAlgorithmFactorySelector;
import org.eclipse.aether.spi.connector.checksum.ChecksumPolicyProvider;
Expand Down Expand Up @@ -229,6 +231,9 @@ protected void configure()
bind( RepositorySystemLifecycle.class )
.to( DefaultRepositorySystemLifecycle.class ).in( Singleton.class );

bind( ResolverExecutorService.class )
.to( DefaultResolverExecutorService.class ).in( Singleton.class );

bind( NamedLockFactorySelector.class ).toInstance( new ParameterizedNamedLockFactorySelector() );
bind( SyncContextFactory.class ).to( DefaultSyncContextFactory.class ).in( Singleton.class );
bind( org.eclipse.aether.impl.SyncContextFactory.class )
Expand Down
Loading