Update smithy-build to be streaming

smithy-build/src/main/java/software/amazon/smithy/build/SmithyBuild.java

@@ -19,9 +19,13 @@
 import java.nio.file.Paths;
 import java.util.Collections;
 import java.util.HashSet;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
+import java.util.TreeMap;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Predicate;
 import java.util.function.Supplier;
@@ -89,16 +93,69 @@ public static SmithyBuild create(ClassLoader classLoader, Supplier<ModelAssemble
     /**
      * Builds the model and applies all projections.
      *
+     * <p>This method loads all projections, projected models, and their
+     * results into memory so that a {@link SmithyBuildResult} can be
+     * returned. See {@link #build(Consumer, BiConsumer)} for a streaming
+     * approach that uses callbacks and does not load all projections into
+     * memory at once.
+     *
+     * <p>Errors are aggregated together into a single
+     * {@link SmithyBuildException} that contains an aggregated error
+     * message and each encountered exception is registered to the aggregate
+     * exception through {@link Throwable#addSuppressed(Throwable)}.
+     *
+     * @return Returns the result of building the model.
+     * @throws IllegalStateException if a {@link SmithyBuildConfig} is not set.
+     * @throws SmithyBuildException if the build fails.
+     * @see #build(Consumer, BiConsumer)
+     */
+    public SmithyBuildResult build() {
+        SmithyBuildResult.Builder resultBuilder = SmithyBuildResult.builder();
+        Map<String, Throwable> errors = Collections.synchronizedMap(new TreeMap<>());
+        build(resultBuilder::addProjectionResult, errors::put);
+
+        if (!errors.isEmpty()) {
+            StringBuilder message = new StringBuilder();
+            message.append(errors.size()).append(" Smithy build projections failed.");
+            message.append(System.lineSeparator()).append(System.lineSeparator());
+
+            for (Map.Entry<String, Throwable> e : errors.entrySet()) {
+                message.append("(").append(e.getKey()).append("): ")
+                        .append(e.getValue())
+                        .append(System.lineSeparator());
+            }
+
+            SmithyBuildException buildException = new SmithyBuildException(message.toString());
+            errors.values().forEach(buildException::addSuppressed);
+            throw buildException;
+        }
+
+        return resultBuilder.build();
+    }
+
+    /**
+     * Builds the model and applies all projections, passing each
+     * {@link ProjectionResult} to the provided callback as they are
+     * completed and each encountered exception to the provided
+     * {@code exceptionCallback} as they are encountered.
+     *
+     * <p>This method differs from {@link #build()} in that it does not
+     * require every projection and projection result to be loaded into
+     * memory.
+     *
      * <p>The result each projection is placed in the outputDirectory.
      * A {@code [projection]-build-info.json} file is created in the output
      * directory. A directory is created for each projection using the
      * projection name, and a file named model.json is place in each directory.
      *
-     * @return Returns the result of building the model.
+     * @param resultCallback A thread-safe callback that receives projection
+     *   results as they complete.
+     * @param exceptionCallback A thread-safe callback that receives the name
+     *   of each failed projection and the exception that occurred.
      * @throws IllegalStateException if a {@link SmithyBuildConfig} is not set.
      */
-    public SmithyBuildResult build() {
-        return new SmithyBuildImpl(this).applyAllProjections();
+    public void build(Consumer<ProjectionResult> resultCallback, BiConsumer<String, Throwable> exceptionCallback) {
+        new SmithyBuildImpl(this).applyAllProjections(resultCallback, exceptionCallback);
     }
 
     /**

smithy-build/src/main/java/software/amazon/smithy/build/SmithyBuildImpl.java

@@ -17,20 +17,27 @@
 
 import java.nio.file.Path;
 import java.nio.file.Paths;
-import java.util.Comparator;
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.LinkedHashSet;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 import java.util.TreeMap;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.function.BiConsumer;
 import java.util.function.BiFunction;
+import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Predicate;
 import java.util.function.Supplier;
 import java.util.logging.Logger;
 import java.util.regex.Pattern;
-import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import software.amazon.smithy.build.model.ProjectionConfig;
 import software.amazon.smithy.build.model.SmithyBuildConfig;
@@ -150,46 +157,99 @@ private static void validatePluginName(String projection, String plugin) {
         }
     }
 
-    SmithyBuildResult applyAllProjections() {
+    void applyAllProjections(
+            Consumer<ProjectionResult> projectionResultConsumer,
+            BiConsumer<String, Throwable> projectionExceptionConsumer
+    ) {
         Model resolvedModel = createBaseModel();
-        SmithyBuildResult.Builder builder = SmithyBuildResult.builder();
-
-        // The projections are being split up here because we need to be able to break out non-parallelizeable plugins.
-        // Right now the only parallelization that occurs is at the projection level though, which is why the split is
-        // here instead of somewhere else.
-        // TODO: Run all parallelizeable plugins across all projections in parallel, followed by all serial plugins
-        Map<String, ProjectionConfig> serialProjections = new TreeMap<>();
-        Map<String, ProjectionConfig> parallelProjections = new TreeMap<>();
-        config.getProjections().entrySet().stream()
-                .filter(e -> !e.getValue().isAbstract())
-                .filter(e -> projectionFilter.test(e.getKey()))
-                .sorted(Comparator.comparing(Map.Entry::getKey))
-                .forEach(e -> {
-                    // Check to see if any of the plugins in the projection require the projection be run serially
-                    boolean isSerial = resolvePlugins(e.getValue()).keySet().stream().anyMatch(pluginName -> {
-                        Optional<SmithyBuildPlugin> plugin = pluginFactory.apply(pluginName);
-                        return plugin.isPresent() && plugin.get().isSerial();
-                    });
-                    // Only run a projection in parallel if all its plugins are parallelizeable.
-                    if (isSerial) {
-                        serialProjections.put(e.getKey(), e.getValue());
-                    } else {
-                        parallelProjections.put(e.getKey(), e.getValue());
-                    }
+
+        // The projections are being split up here because we need to be able
+        // to break out non-parallelizeable plugins. Right now the only
+        // parallelization that occurs is at the projection level.
+        List<Callable<Void>> parallelProjections = new ArrayList<>();
+        List<String> parallelProjectionNameOrder = new ArrayList<>();
+
+        for (Map.Entry<String, ProjectionConfig> entry : config.getProjections().entrySet()) {
+            String name = entry.getKey();
+            ProjectionConfig config = entry.getValue();
+
+            if (config.isAbstract() || !projectionFilter.test(name)) {
+                continue;
+            }
+
+            // Check to see if any of the plugins in the projection require the projection be run serially
+            boolean isSerial = resolvePlugins(config).keySet().stream().anyMatch(pluginName -> {
+                Optional<SmithyBuildPlugin> plugin = pluginFactory.apply(pluginName);
+                return plugin.isPresent() && plugin.get().isSerial();
+            });
+
+            if (isSerial) {
+                executeSerialProjection(resolvedModel, name, config,
+                                        projectionResultConsumer, projectionExceptionConsumer);
+            } else {
+                parallelProjectionNameOrder.add(name);
+                parallelProjections.add(() -> {
+                    ProjectionResult projectionResult = applyProjection(name, config, resolvedModel);
+                    projectionResultConsumer.accept(projectionResult);
+                    return null;
                 });
+            }
+        }
 
-        serialProjections.entrySet().stream()
-                .map(e -> applyProjection(e.getKey(), e.getValue(), resolvedModel))
-                .collect(Collectors.toList())
-                .forEach(builder::addProjectionResult);
+        if (!parallelProjections.isEmpty()) {
+            executeParallelProjections(parallelProjections, parallelProjectionNameOrder, projectionExceptionConsumer);
+        }
+    }
 
-        parallelProjections.entrySet().stream()
-                .parallel()
-                .map(e -> applyProjection(e.getKey(), e.getValue(), resolvedModel))
-                .collect(Collectors.toList())
-                .forEach(builder::addProjectionResult);
+    private void executeSerialProjection(
+            Model resolvedModel,
+            String name,
+            ProjectionConfig config,
+            Consumer<ProjectionResult> projectionResultConsumer,
+            BiConsumer<String, Throwable> projectionExceptionConsumer
+    ) {
+        // Errors that occur while invoking the result callback must not
+        // cause the exception callback to be invoked.
+        ProjectionResult result = null;
+
+        try {
+            result = applyProjection(name, config, resolvedModel);
+        } catch (Throwable e) {
+            projectionExceptionConsumer.accept(name, e);
+        }
 
-        return builder.build();
+        if (result != null) {
+            projectionResultConsumer.accept(result);
+        }
+    }
+
+    private void executeParallelProjections(
+            List<Callable<Void>> parallelProjections,
+            List<String> parallelProjectionNameOrder,
+            BiConsumer<String, Throwable> projectionExceptionConsumer
+    ) {
+        // Except for writing files to disk, projections are mostly CPU bound.
+        int numberOfCores = Runtime.getRuntime().availableProcessors();
+        ExecutorService executor = Executors.newFixedThreadPool(numberOfCores);
+
+        try {
+            List<Future<Void>> futures = executor.invokeAll(parallelProjections);
+            executor.shutdown();
+            // Futures are returned in the same order they were added, so
+            // use the list of ordered names to know which projections failed.
+            for (int i = 0; i < futures.size(); i++) {
+                try {
+                    futures.get(i).get();
+                } catch (ExecutionException e) {
+                    Throwable cause = e.getCause() != null ? e.getCause() : e;
+                    String failedProjectionName = parallelProjectionNameOrder.get(i);
+                    projectionExceptionConsumer.accept(failedProjectionName, cause);
+                }
+            }
+        } catch (InterruptedException e) {
+            executor.shutdownNow();
+            throw new SmithyBuildException(e.getMessage(), e);
+        }
     }
 
     private Model createBaseModel() {

smithy-build/src/main/java/software/amazon/smithy/build/SmithyBuildResult.java

@@ -17,6 +17,7 @@
 
 import java.nio.file.Path;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -123,7 +124,7 @@ public boolean isEmpty() {
      * Creates a SmithyBuildResult.
      */
     public static final class Builder implements SmithyBuilder<SmithyBuildResult> {
-        private final List<ProjectionResult> results = new ArrayList<>();
+        private final List<ProjectionResult> results = Collections.synchronizedList(new ArrayList<>());
 
         private Builder() {}
 
@@ -135,6 +136,9 @@ public SmithyBuildResult build() {
         /**
          * Adds a projection result to the builder.
          *
+         * <p>This method is thread-safe as a synchronized list is updated each
+         * time this is called.
+         *
          * @param result Result to add.
          * @return Returns the builder.
          */

smithy-build/src/test/java/software/amazon/smithy/build/SmithyBuildTest.java

@@ -16,12 +16,15 @@
 package software.amazon.smithy.build;
 
 import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.contains;
 import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.hasItem;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.not;
 import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import java.io.File;
@@ -31,6 +34,7 @@
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.Comparator;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -482,4 +486,42 @@ public void canFilterPlugins() throws URISyntaxException {
         assertFalse(a.getPluginManifest("model").isPresent());
         assertTrue(a.getPluginManifest("build-info").isPresent());
     }
+
+    @Test
+    public void throwsWhenErrorsOccur() throws Exception {
+        Path badConfig = Paths.get(getClass().getResource("trigger-plugin-error.json").toURI());
+        Model model = Model.assembler()
+                .addImport(getClass().getResource("simple-model.json"))
+                .assemble()
+                .unwrap();
+
+        RuntimeException canned = new RuntimeException("Hi");
+        Map<String, SmithyBuildPlugin> plugins = new HashMap<>();
+        plugins.put("foo", new SmithyBuildPlugin() {
+            @Override
+            public String getName() {
+                return "foo";
+            }
+
+            @Override
+            public void execute(PluginContext context) {
+                throw canned;
+            }
+        });
+
+        Function<String, Optional<SmithyBuildPlugin>> factory = SmithyBuildPlugin.createServiceFactory();
+        Function<String, Optional<SmithyBuildPlugin>> composed = name -> OptionalUtils.or(
+                Optional.ofNullable(plugins.get(name)), () -> factory.apply(name));
+
+        SmithyBuild builder = new SmithyBuild()
+                .model(model)
+                .fileManifestFactory(MockManifest::new)
+                .pluginFactory(composed)
+                .config(SmithyBuildConfig.load(badConfig));
+
+        SmithyBuildException e = Assertions.assertThrows(SmithyBuildException.class, builder::build);
+        assertThat(e.getMessage(), containsString("1 Smithy build projections failed"));
+        assertThat(e.getMessage(), containsString("(exampleProjection): java.lang.RuntimeException: Hi"));
+        assertThat(e.getSuppressed(), equalTo(new Throwable[]{canned}));
+    }
 }

smithy-build/src/test/resources/software/amazon/smithy/build/trigger-plugin-error.json

@@ -0,0 +1,10 @@
+{
+  "version": "1.0",
+  "projections": {
+    "exampleProjection": {
+      "plugins": {
+        "foo": {}
+      }
+    }
+  }
+}