diff --git a/src/main/java/rife/bld/dependencies/ArtifactRetriever.java b/src/main/java/rife/bld/dependencies/ArtifactRetriever.java index ac98ebd..aa69070 100644 --- a/src/main/java/rife/bld/dependencies/ArtifactRetriever.java +++ b/src/main/java/rife/bld/dependencies/ArtifactRetriever.java @@ -19,6 +19,7 @@ import java.net.URLConnection; import java.nio.channels.Channels; import java.security.MessageDigest; import java.util.*; +import java.util.concurrent.ConcurrentHashMap; import static rife.tools.HttpUtils.HEADER_AUTHORIZATION; import static rife.tools.HttpUtils.basicAuthorizationHeader; @@ -35,6 +36,8 @@ import static rife.tools.StringUtils.encodeHexLower; * @since 1.5.18 */ public abstract class ArtifactRetriever { + private static final int TRANSFER_CHUNK_SIZE = 128 * 1024; + private final static ArtifactRetriever UNCACHED = new ArtifactRetriever() { String getCached(RepositoryArtifact artifact) { return null; @@ -63,7 +66,7 @@ public abstract class ArtifactRetriever { */ public static ArtifactRetriever cachingInstance() { return new ArtifactRetriever() { - private final Map artifactCache = new HashMap<>(); + private final Map artifactCache = new ConcurrentHashMap<>(); String getCached(RepositoryArtifact artifact) { return artifactCache.get(artifact); @@ -141,17 +144,17 @@ public abstract class ArtifactRetriever { var download_filename = artifact.location().substring(artifact.location().lastIndexOf("/") + 1); var download_file = new File(directory, download_filename); - System.out.print("Downloading: " + artifact.location() + " ... "); - System.out.flush(); + var transfer = TransferOutput.instance().start(artifact.location()); + var status = ""; try { if (artifact.repository().isLocal()) { var source = new File(artifact.location()); if (source.exists()) { FileUtils.copy(source, download_file); - System.out.print("done"); + status = "done"; return true; } else { - System.out.print("not found"); + status = "not found"; return false; } } else { @@ -159,7 +162,7 @@ public abstract class ArtifactRetriever { if (download_file.exists() && download_file.canRead()) { if (checkHash(artifact, download_file, ".sha256", "SHA-256") || checkHash(artifact, download_file, ".md5", "MD5")) { - System.out.print("exists"); + status = "exists"; return true; } } @@ -170,23 +173,29 @@ public abstract class ArtifactRetriever { HEADER_AUTHORIZATION, basicAuthorizationHeader(artifact.repository().username(), artifact.repository().password())); } + var content_length = connection.getContentLengthLong(); try (var input_stream = connection.getInputStream()) { var readableByteChannel = Channels.newChannel(input_stream); try (var fileOutputStream = new FileOutputStream(download_file)) { var fileChannel = fileOutputStream.getChannel(); - fileChannel.transferFrom(readableByteChannel, 0, Long.MAX_VALUE); + var position = 0L; + long transferred; + while ((transferred = fileChannel.transferFrom(readableByteChannel, position, TRANSFER_CHUNK_SIZE)) > 0) { + position += transferred; + transfer.progress(position, content_length); + } - System.out.print("done"); + status = "done"; return true; } } } catch (FileNotFoundException e) { - System.out.print("not found"); + status = "not found"; return false; } } } finally { - System.out.println(); + transfer.finish(status); } } diff --git a/src/main/java/rife/bld/dependencies/DependencySet.java b/src/main/java/rife/bld/dependencies/DependencySet.java index a81bb49..813817d 100644 --- a/src/main/java/rife/bld/dependencies/DependencySet.java +++ b/src/main/java/rife/bld/dependencies/DependencySet.java @@ -8,6 +8,10 @@ import rife.bld.dependencies.exceptions.DependencyTransferException; import java.io.File; import java.util.*; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.function.Supplier; /** * Convenience class to handle a set of {@link Dependency} objects. @@ -128,6 +132,11 @@ public class DependencySet extends AbstractSet implements Set * The destination directory must exist and be writable. + *

+ * The artifacts of different dependencies are transferred in parallel, + * the {@value VersionResolution#PROPERTY_TRANSFER_PARALLELISM} property + * can be used to change the number of simultaneous transfers, setting it + * to {@code 1} makes the transfers sequential. * * @param resolution the version resolution state that can be cached * @param retriever the retriever to use to get artifacts @@ -140,7 +149,7 @@ public class DependencySet extends AbstractSet implements Set transferIntoDirectory(VersionResolution resolution, ArtifactRetriever retriever, List repositories, File directory, File modulesDirectory, String... classifiers) { - var result = new ArrayList(); + var transfers = new ArrayList>>(); for (var dependency : this) { var transfer_directory = directory; if (dependency.isModularJar()) { @@ -159,21 +168,66 @@ public class DependencySet extends AbstractSet implements Set { + var artifacts = new ArrayList(); + var artifact = new DependencyResolver(resolution, retriever, repositories, dependency).transferIntoDirectory(target_directory); + if (artifact != null) { + artifacts.add(artifact); + } - if (classifiers != null) { - for (var classifier : classifiers) { - if (classifier != null && !dependency.excludedClassifiers().contains(classifier)) { - var classifier_artifact = new DependencyResolver(resolution, retriever, repositories, dependency.withClassifier(classifier)).transferIntoDirectory(transfer_directory); - if (classifier_artifact != null) { - result.add(classifier_artifact); + if (classifiers != null) { + for (var classifier : classifiers) { + if (classifier != null && !dependency.excludedClassifiers().contains(classifier)) { + var classifier_artifact = new DependencyResolver(resolution, retriever, repositories, dependency.withClassifier(classifier)).transferIntoDirectory(target_directory); + if (classifier_artifact != null) { + artifacts.add(classifier_artifact); + } } } } + return artifacts; + }); + } + + return executeTransfers(transfers, resolution.transferParallelism()); + } + + private static List executeTransfers(List>> transfers, int transferParallelism) { + var result = new ArrayList(); + + var parallelism = Math.min(transfers.size(), transferParallelism); + if (parallelism <= 1) { + for (var transfer : transfers) { + result.addAll(transfer.get()); } + return result; + } + + var executor = Executors.newFixedThreadPool(parallelism); + try { + var futures = new ArrayList>>(); + for (var transfer : transfers) { + futures.add(executor.submit(transfer::get)); + } + for (var future : futures) { + try { + result.addAll(future.get()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException("Artifact transfer was interrupted", e); + } catch (ExecutionException e) { + if (e.getCause() instanceof RuntimeException runtime) { + throw runtime; + } + if (e.getCause() instanceof Error error) { + throw error; + } + throw new IllegalStateException(e.getCause()); + } + } + } finally { + executor.shutdownNow(); } return result; } diff --git a/src/main/java/rife/bld/dependencies/TransferOutput.java b/src/main/java/rife/bld/dependencies/TransferOutput.java new file mode 100644 index 0000000..a04d3b4 --- /dev/null +++ b/src/main/java/rife/bld/dependencies/TransferOutput.java @@ -0,0 +1,185 @@ +/* + * Copyright 2001-2026 Geert Bevin (gbevin[remove] at uwyn dot com) + * Licensed under the Apache License, Version 2.0 (the "License") + */ +package rife.bld.dependencies; + +import java.util.ArrayList; +import java.util.List; + +/** + * Outputs the status of artifact transfers to the console. + *

+ * When the console is an interactive terminal, the transfers that are in + * progress are displayed in a live updating block at the bottom of the + * output, with each transfer reporting its progress as it happens. + * Otherwise, a single line is printed when each transfer finishes, keeping + * the output stable for CI logs, pipes and IDE consoles, while still never + * interleaving the lines of parallel transfers. + * + * @author Geert Bevin (gbevin[remove] at uwyn dot com) + * @since 2.3.1 + */ +abstract class TransferOutput { + private static final TransferOutput INSTANCE = create(); + + static TransferOutput instance() { + return INSTANCE; + } + + private static TransferOutput create() { + if (System.console() != null && !"dumb".equals(System.getenv("TERM"))) { + return new AnsiTransferOutput(); + } + return new PlainTransferOutput(); + } + + /** + * Starts tracking a single artifact transfer. + * + * @param location the location of the artifact that is being transferred + * @return the transfer to report progress and completion on + * @since 2.3.1 + */ + abstract Transfer start(String location); + + /** + * A single artifact transfer whose progress is being tracked. + * + * @since 2.3.1 + */ + abstract static class Transfer { + protected final String location_; + + protected Transfer(String location) { + location_ = location; + } + + /** + * Reports transfer progress. + * + * @param transferred the number of bytes transferred so far + * @param total the total number of bytes; or {@code -1} when unknown + * @since 2.3.1 + */ + abstract void progress(long transferred, long total); + + /** + * Reports that the transfer finished. + * + * @param status the final status of the transfer, like + * {@code done}, {@code exists} or {@code not found} + * @since 2.3.1 + */ + abstract void finish(String status); + } + + static String describe(String location, String status) { + return "Downloading: " + location + " ... " + status; + } + + private static class PlainTransferOutput extends TransferOutput { + Transfer start(String location) { + return new Transfer(location) { + void progress(long transferred, long total) { + // no-op, only the finished state is printed + } + + void finish(String status) { + System.out.println(describe(location_, status)); + } + }; + } + } + + private static class AnsiTransferOutput extends TransferOutput { + private static final long REPAINT_INTERVAL_MS = 100; + private static final int MAX_LINE_WIDTH = 79; + + private final List active_ = new ArrayList<>(); + private int paintedLines_ = 0; + private long lastRepaint_ = 0; + + synchronized Transfer start(String location) { + var transfer = new AnsiTransfer(location); + active_.add(transfer); + repaint(null); + return transfer; + } + + private synchronized void reportProgress() { + if (System.currentTimeMillis() - lastRepaint_ >= REPAINT_INTERVAL_MS) { + repaint(null); + } + } + + private synchronized void reportFinished(AnsiTransfer transfer, String status) { + active_.remove(transfer); + repaint(describe(transfer.location_, status)); + } + + private void repaint(String finishedLine) { + var output = new StringBuilder(); + if (paintedLines_ > 0) { + // move the cursor to the beginning of the live block + // and clear everything below it + output.append("\u001B[").append(paintedLines_).append("F\u001B[0J"); + } + if (finishedLine != null) { + output.append(finishedLine).append('\n'); + } + for (var transfer : active_) { + output.append(transfer.statusLine()).append('\n'); + } + paintedLines_ = active_.size(); + lastRepaint_ = System.currentTimeMillis(); + System.out.print(output); + System.out.flush(); + } + + private class AnsiTransfer extends Transfer { + private volatile long transferred_ = 0; + private volatile long total_ = -1; + + AnsiTransfer(String location) { + super(location); + } + + void progress(long transferred, long total) { + transferred_ = transferred; + total_ = total; + reportProgress(); + } + + void finish(String status) { + reportFinished(this, status); + } + + String statusLine() { + var filename = location_.substring(location_.lastIndexOf('/') + 1); + var line = new StringBuilder("Downloading: ").append(filename).append(" ... "); + if (total_ > 0) { + line.append(transferred_ * 100 / total_).append('%'); + } else if (transferred_ > 0) { + line.append(humanBytes(transferred_)); + } + // prevent lines from wrapping since that would break + // the cursor repositioning of the repaints + if (line.length() > MAX_LINE_WIDTH) { + line.setLength(MAX_LINE_WIDTH); + } + return line.toString(); + } + + private String humanBytes(long bytes) { + if (bytes >= 1024 * 1024) { + return (bytes / (1024 * 1024)) + " MB"; + } + if (bytes >= 1024) { + return (bytes / 1024) + " KB"; + } + return bytes + " B"; + } + } + } +} diff --git a/src/main/java/rife/bld/dependencies/VersionResolution.java b/src/main/java/rife/bld/dependencies/VersionResolution.java index 462f375..7741c4e 100644 --- a/src/main/java/rife/bld/dependencies/VersionResolution.java +++ b/src/main/java/rife/bld/dependencies/VersionResolution.java @@ -8,6 +8,7 @@ import rife.ioc.HierarchicalProperties; import java.util.HashMap; import java.util.Map; +import java.util.logging.Logger; /** * This class is responsible for managing version overrides for dependencies. @@ -27,6 +28,10 @@ import java.util.Map; * bld.override-tests=com.uwyn.rife2:bld-tests-badge:1.4.7 * bld.override-h2=com.h2database:h2:2.2.222 * + *

+ * It also captures other dependency resolution preferences, like the number + * of parallel artifact transfers through the "{@code bld.transferParallelism}" + * property. * @since 2.0 */ public class VersionResolution { @@ -36,7 +41,16 @@ public class VersionResolution { */ public static final String PROPERTY_OVERRIDE_PREFIX = "bld.override"; + /** + * The property key that determines how many artifact transfers are + * performed in parallel, {@code 1} makes them sequential. + * @since 2.3.1 + */ + public static final String PROPERTY_TRANSFER_PARALLELISM = "bld.transferParallelism"; + private static final int DEFAULT_TRANSFER_PARALLELISM = 6; + private final Map versionOverrides_ = new HashMap<>(); + private final int transferParallelism_; /** * Returns a dummy {@code VersionResolution} instance that doesn't override anything. @@ -59,6 +73,7 @@ public class VersionResolution { * @since 2.0 */ public VersionResolution(HierarchicalProperties properties) { + var transfer_parallelism = DEFAULT_TRANSFER_PARALLELISM; if (properties != null) { for (var name : properties.getNames()) { if (name.startsWith(PROPERTY_OVERRIDE_PREFIX)) { @@ -73,7 +88,17 @@ public class VersionResolution { } } } + + var parallelism = properties.getValueString(PROPERTY_TRANSFER_PARALLELISM); + if (parallelism != null && !parallelism.isBlank()) { + try { + transfer_parallelism = Math.max(1, Integer.parseInt(parallelism.trim())); + } catch (NumberFormatException e) { + Logger.getLogger("rife.bld").warning("Unable to parse the " + PROPERTY_TRANSFER_PARALLELISM + " property as an integer: '" + parallelism + "', using " + transfer_parallelism + " instead"); + } + } } + transferParallelism_ = transfer_parallelism; } /** @@ -122,4 +147,15 @@ public class VersionResolution { public Map versionOverrides() { return versionOverrides_; } + + /** + * Returns the number of artifact transfers that are performed in parallel, + * {@code 1} means transfers are sequential. + * + * @return the number of parallel artifact transfers + * @since 2.3.1 + */ + public int transferParallelism() { + return transferParallelism_; + } } diff --git a/src/test/java/rife/bld/dependencies/TestDependencySet.java b/src/test/java/rife/bld/dependencies/TestDependencySet.java index 7cbe32a..f6e9025 100644 --- a/src/test/java/rife/bld/dependencies/TestDependencySet.java +++ b/src/test/java/rife/bld/dependencies/TestDependencySet.java @@ -4,10 +4,19 @@ */ package rife.bld.dependencies; +import com.sun.net.httpserver.HttpServer; import org.junit.jupiter.api.Test; +import rife.ioc.HierarchicalProperties; +import rife.tools.FileUtils; import rife.tools.StringUtils; +import java.io.File; +import java.io.IOException; +import java.net.InetSocketAddress; +import java.nio.file.Files; import java.util.List; +import java.util.concurrent.Executors; +import java.util.concurrent.atomic.AtomicInteger; import static org.junit.jupiter.api.Assertions.*; import static rife.bld.dependencies.RepositoryTestHelper.getNextRepository; @@ -373,4 +382,93 @@ public class TestDependencySet { └─ org.json:json:20250107 """), dependencies.generateTransitiveDependencyTree(VersionResolution.dummy(), ArtifactRetriever.instance(), RepositoryTestHelper.getNextRepositories(), compile, runtime)); } + + @Test + void testTransferIntoDirectoryParallel() throws Exception { + var max_concurrent_transfers = new AtomicInteger(); + var server = createTransferServer(max_concurrent_transfers); + server.start(); + var tmp = Files.createTempDirectory("transfers").toFile(); + try { + var dependencies = createTransferDependencies(); + var repositories = List.of(new Repository("http://localhost:" + server.getAddress().getPort() + "/")); + + var artifacts = dependencies.transferIntoDirectory(new VersionResolution(null), ArtifactRetriever.instance(), repositories, tmp, tmp); + + assertTransferredArtifacts(dependencies, artifacts, tmp); + assertTrue(max_concurrent_transfers.get() > 1, "expected concurrent transfers, max was " + max_concurrent_transfers.get()); + } finally { + server.stop(0); + FileUtils.deleteDirectory(tmp); + } + } + + @Test + void testTransferIntoDirectorySequential() throws Exception { + var max_concurrent_transfers = new AtomicInteger(); + var server = createTransferServer(max_concurrent_transfers); + server.start(); + var tmp = Files.createTempDirectory("transfers").toFile(); + try { + var properties = new HierarchicalProperties(); + properties.put(VersionResolution.PROPERTY_TRANSFER_PARALLELISM, "1"); + var resolution = new VersionResolution(properties); + assertEquals(1, resolution.transferParallelism()); + + var dependencies = createTransferDependencies(); + var repositories = List.of(new Repository("http://localhost:" + server.getAddress().getPort() + "/")); + + var artifacts = dependencies.transferIntoDirectory(resolution, ArtifactRetriever.instance(), repositories, tmp, tmp); + + assertTransferredArtifacts(dependencies, artifacts, tmp); + assertEquals(1, max_concurrent_transfers.get(), "expected sequential transfers, max was " + max_concurrent_transfers.get()); + } finally { + server.stop(0); + FileUtils.deleteDirectory(tmp); + } + } + + private static DependencySet createTransferDependencies() { + var dependencies = new DependencySet(); + for (var i = 1; i <= 6; i++) { + dependencies.include(new Dependency("com.example", "artifact" + i, new VersionNumber(1, 0, 0))); + } + return dependencies; + } + + private static HttpServer createTransferServer(AtomicInteger maxConcurrentTransfers) + throws IOException { + var active_transfers = new AtomicInteger(); + var server = HttpServer.create(new InetSocketAddress("localhost", 0), 0); + server.createContext("/", exchange -> { + var active = active_transfers.incrementAndGet(); + maxConcurrentTransfers.accumulateAndGet(active, Math::max); + try { + // delay the response so that parallel transfers overlap + Thread.sleep(200); + + var body = exchange.getRequestURI().getPath().getBytes(); + exchange.sendResponseHeaders(200, body.length); + exchange.getResponseBody().write(body); + exchange.close(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + active_transfers.decrementAndGet(); + } + }); + server.setExecutor(Executors.newCachedThreadPool()); + return server; + } + + private static void assertTransferredArtifacts(DependencySet dependencies, List artifacts, File directory) { + assertEquals(dependencies.size(), artifacts.size()); + var index = 0; + for (var dependency : dependencies) { + var filename = dependency.artifactId() + "-" + dependency.version() + ".jar"; + assertTrue(artifacts.get(index).location().endsWith(filename), "expected artifact " + filename + " at index " + index); + assertTrue(new File(directory, filename).exists(), "expected file " + filename + " to be transferred"); + ++index; + } + } }