From c397bafcd0a35474b8dc91bac10d2831dc9c9011 Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Wed, 29 Apr 2020 15:46:17 +0200 Subject: [PATCH] #30 - Introduce Executor bean to split worker pools. The latest JGit upgrade runs tasks in parallel on the common fork-join pool. To avoid pool exhaustion and cyclic locks, we're offloading our work to an own pool. --- .../data/release/git/GitOperations.java | 27 ++++++++------ .../release/issues/IssueTrackerCommands.java | 18 +++++---- .../release/sagan/SaganConfiguration.java | 6 ++- .../data/release/sagan/SaganOperations.java | 8 ++-- .../data/release/utils/ExecutionUtils.java | 18 ++++++--- .../release/utils/ExecutorConfiguration.java | 37 +++++++++++++++++++ 6 files changed, 83 insertions(+), 31 deletions(-) create mode 100644 release-tools/src/main/java/org/springframework/data/release/utils/ExecutorConfiguration.java diff --git a/release-tools/src/main/java/org/springframework/data/release/git/GitOperations.java b/release-tools/src/main/java/org/springframework/data/release/git/GitOperations.java index da991b8..63d456f 100644 --- a/release-tools/src/main/java/org/springframework/data/release/git/GitOperations.java +++ b/release-tools/src/main/java/org/springframework/data/release/git/GitOperations.java @@ -27,6 +27,7 @@ import java.util.List; import java.util.Locale; import java.util.Map; import java.util.Optional; +import java.util.concurrent.Executor; import java.util.concurrent.atomic.AtomicBoolean; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -86,6 +87,7 @@ public class GitOperations { } GitServer server = new GitServer(); + Executor executor; Workspace workspace; Logger logger; PluginRegistry issueTracker; @@ -111,7 +113,7 @@ public class GitOperations { Assert.notNull(train, "Train must not be null!"); - ExecutionUtils.run(train, module -> { + ExecutionUtils.run(executor, train, module -> { reset(module.getProject(), Branch.from(module)); }); } @@ -128,7 +130,7 @@ public class GitOperations { update(train); AtomicBoolean masterSwitch = new AtomicBoolean(); - ExecutionUtils.run(train, module -> { + ExecutionUtils.run(executor, train, module -> { Project project = module.getProject(); @@ -179,7 +181,7 @@ public class GitOperations { Assert.notNull(iteration, "Train iteration must not be null!"); - ExecutionUtils.run(iteration, module -> { + ExecutionUtils.run(executor, iteration, module -> { Project project = module.getProject(); ArtifactVersion artifactVersion = ArtifactVersion.of(module); @@ -198,7 +200,7 @@ public class GitOperations { public void prepare(TrainIteration iteration) { - ExecutionUtils.run(iteration, module -> { + ExecutionUtils.run(executor, iteration, module -> { Project project = module.getProject(); Branch branch = Branch.from(module); @@ -223,12 +225,12 @@ public class GitOperations { } public void update(Train train) { - ExecutionUtils.run(train, module -> update(module.getProject())); + ExecutionUtils.run(executor, train, module -> update(module.getProject())); } public void push(TrainIteration iteration) { - ExecutionUtils.run(iteration, module -> { + ExecutionUtils.run(executor, iteration, module -> { Branch branch = Branch.from(module); logger.log(module, "git push origin %s", branch); @@ -254,7 +256,7 @@ public class GitOperations { public void pushTags(Train train) { - ExecutionUtils.run(train.getModules(), module -> { + ExecutionUtils.run(executor, train.getModules(), module -> { logger.log(module.getProject(), "git push --tags origin"); @@ -364,7 +366,7 @@ public class GitOperations { Assert.notNull(iteration, "Train iteration must not be null!"); - ExecutionUtils.run(iteration, module -> { + ExecutionUtils.run(executor, iteration, module -> { Project project = module.getProject(); ObjectId hash = getReleaseHash(module); @@ -408,7 +410,8 @@ public class GitOperations { Assert.notNull(iteration, "Train iteration must not be null!"); Assert.hasText(summary, "Summary must not be null or empty!"); - ExecutionUtils.run(iteration, module -> commit(module, expandSummary(summary, module, iteration), details)); + ExecutionUtils.run(executor, iteration, + module -> commit(module, expandSummary(summary, module, iteration), details)); } /** @@ -550,7 +553,7 @@ public class GitOperations { checkout(iteration); - ExecutionUtils.run(iteration, module -> { + ExecutionUtils.run(executor, iteration, module -> { Branch branch = createMaintenanceBranch(module); checkout(module.getProject(), branch, BranchCheckoutMode.CREATE_ONLY); @@ -559,7 +562,7 @@ public class GitOperations { public void removeTags(TrainIteration iteration) { - ExecutionUtils.run(iteration, module -> { + ExecutionUtils.run(executor, iteration, module -> { Project project = module.getProject(); ArtifactVersion artifactVersion = ArtifactVersion.of(module); @@ -591,7 +594,7 @@ public class GitOperations { Assert.notNull(iteration, "Train iteration must not be null!"); Assert.notNull(targets, "Target trains must not be null!"); - ExecutionUtils.run(iteration, module -> { + ExecutionUtils.run(executor, iteration, module -> { BackportTargets backportTargets = new BackportTargets(module, targets); Project project = module.getProject(); diff --git a/release-tools/src/main/java/org/springframework/data/release/issues/IssueTrackerCommands.java b/release-tools/src/main/java/org/springframework/data/release/issues/IssueTrackerCommands.java index 3045775..ed18854 100644 --- a/release-tools/src/main/java/org/springframework/data/release/issues/IssueTrackerCommands.java +++ b/release-tools/src/main/java/org/springframework/data/release/issues/IssueTrackerCommands.java @@ -22,6 +22,7 @@ import lombok.NonNull; import lombok.RequiredArgsConstructor; import lombok.experimental.FieldDefaults; +import java.util.concurrent.Executor; import java.util.stream.Collectors; import java.util.stream.StreamSupport; @@ -47,6 +48,7 @@ import org.springframework.util.StringUtils; class IssueTrackerCommands extends TimedCommand { @NonNull PluginRegistry tracker; + @NonNull Executor executor; @CliCommand("tracker evict") public void evict() { @@ -65,7 +67,7 @@ class IssueTrackerCommands extends TimedCommand { @CliCommand(value = "tracker releasetickets") public String releaseTickets(@CliOption(key = "", mandatory = true) TrainIteration iteration) { - return runAndReturn(iteration, module -> getTrackerFor(module).getReleaseTicketFor(module), + return runAndReturn(executor, iteration, module -> getTrackerFor(module).getReleaseTicketFor(module), Tickets.toTicketsCollector()).toString(); } @@ -100,25 +102,25 @@ class IssueTrackerCommands extends TimedCommand { @CliCommand(value = "tracker self-assign releasetickets") public String jiraSelfAssignReleaseTickets(@CliOption(key = "", mandatory = true) TrainIteration iteration) { - return runAndReturn(iteration, module -> getTrackerFor(module).assignReleaseTicketToMe(module), + return runAndReturn(executor, iteration, module -> getTrackerFor(module).assignReleaseTicketToMe(module), Tickets.toTicketsCollector()).toString(); } public String jiraStartProgress(@CliOption(key = "", mandatory = true) TrainIteration iteration) { - return runAndReturn(iteration, module -> getTrackerFor(module).startReleaseTicketProgress(module), + return runAndReturn(executor, iteration, module -> getTrackerFor(module).startReleaseTicketProgress(module), Tickets.toTicketsCollector()).toString(); } @CliCommand(value = "tracker create releaseversions") public void jiraCreateReleaseVersions(@CliOption(key = "", mandatory = true) TrainIteration iteration) { - run(iteration, module -> getTrackerFor(module).createReleaseVersion(module)); + run(executor, iteration, module -> getTrackerFor(module).createReleaseVersion(module)); } @CliCommand(value = "tracker create releasetickets") public String createReleaseTickets(@CliOption(key = "", mandatory = true) TrainIteration iteration) { - run(iteration, module -> getTrackerFor(module).createReleaseTicket(module)); + run(executor, iteration, module -> getTrackerFor(module).createReleaseTicket(module)); evict(); return releaseTickets(iteration); @@ -136,18 +138,18 @@ class IssueTrackerCommands extends TimedCommand { return getTrackerFor(module).getChangelogFor(module).toString(); } - return ExecutionUtils.runAndReturn(iteration, this::getChangelog).// + return ExecutionUtils.runAndReturn(executor, iteration, this::getChangelog).// stream().map(it -> it.toString()).collect(Collectors.joining("\n")); } @CliCommand("tracker close") public void closeIteration(@CliOption(key = "", mandatory = true) TrainIteration iteration) { - run(iteration, module -> getTrackerFor(module).closeIteration(module)); + run(executor, iteration, module -> getTrackerFor(module).closeIteration(module)); } @CliCommand("tracker archive") public void archiveIteration(@CliOption(key = "", mandatory = true) TrainIteration iteration) { - run(iteration, module -> getTrackerFor(module).archiveReleaseVersion(module)); + run(executor, iteration, module -> getTrackerFor(module).archiveReleaseVersion(module)); } private Changelog getChangelog(ModuleIteration module) { diff --git a/release-tools/src/main/java/org/springframework/data/release/sagan/SaganConfiguration.java b/release-tools/src/main/java/org/springframework/data/release/sagan/SaganConfiguration.java index ed6840e..999d773 100644 --- a/release-tools/src/main/java/org/springframework/data/release/sagan/SaganConfiguration.java +++ b/release-tools/src/main/java/org/springframework/data/release/sagan/SaganConfiguration.java @@ -15,6 +15,8 @@ */ package org.springframework.data.release.sagan; +import java.util.concurrent.Executor; + import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.web.client.RestTemplateBuilder; import org.springframework.context.annotation.Bean; @@ -35,8 +37,8 @@ class SaganConfiguration { @Autowired Logger logger; @Bean - public SaganOperations saganOperations(GitOperations operations) { - return new SaganOperations(operations, saganClient(), logger); + public SaganOperations saganOperations(GitOperations operations, Executor executor) { + return new SaganOperations(operations, executor, saganClient(), logger); } @Bean diff --git a/release-tools/src/main/java/org/springframework/data/release/sagan/SaganOperations.java b/release-tools/src/main/java/org/springframework/data/release/sagan/SaganOperations.java index 147c6db..312cf40 100644 --- a/release-tools/src/main/java/org/springframework/data/release/sagan/SaganOperations.java +++ b/release-tools/src/main/java/org/springframework/data/release/sagan/SaganOperations.java @@ -24,6 +24,7 @@ import java.util.Collection; import java.util.Comparator; import java.util.List; import java.util.Map; +import java.util.concurrent.Executor; import java.util.stream.Collectors; import org.springframework.data.release.git.GitOperations; @@ -51,6 +52,7 @@ class SaganOperations { Projects.KEY_VALUE); GitOperations git; + Executor executor; SaganClient client; Logger logger; @@ -72,7 +74,7 @@ class SaganOperations { Map versions = findVersions(trains); - ExecutionUtils.run(Streamable.of(versions.entrySet()), + ExecutionUtils.run(executor, Streamable.of(versions.entrySet()), entry -> client.updateProjectMetadata(entry.getKey(), entry.getValue())); } @@ -93,8 +95,8 @@ class SaganOperations { Assert.notNull(trains, "Trains must not be null!"); - return ExecutionUtils.runAndReturn(Streamable.of(trains), train -> { - return ExecutionUtils.runAndReturn( + return ExecutionUtils.runAndReturn(executor, Streamable.of(trains), train -> { + return ExecutionUtils.runAndReturn(executor, Streamable.of(() -> train.stream().filter(module -> !TO_FILTER.contains(module.getProject()))), module -> { return getLatestVersion(module, train); }); diff --git a/release-tools/src/main/java/org/springframework/data/release/utils/ExecutionUtils.java b/release-tools/src/main/java/org/springframework/data/release/utils/ExecutionUtils.java index 755aa20..be5ae44 100644 --- a/release-tools/src/main/java/org/springframework/data/release/utils/ExecutionUtils.java +++ b/release-tools/src/main/java/org/springframework/data/release/utils/ExecutionUtils.java @@ -20,6 +20,7 @@ import lombok.extern.slf4j.Slf4j; import java.util.Collection; import java.util.Objects; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Executor; import java.util.function.Function; import java.util.stream.Collector; import java.util.stream.Collectors; @@ -31,6 +32,7 @@ import org.springframework.util.Assert; * Utility method to easily execute functionality in parallel. * * @author Oliver Gierke + * @author Mark Paluch */ @Slf4j public class ExecutionUtils { @@ -40,11 +42,13 @@ public class ExecutionUtils { * all executions to complete before returning. Exceptions being thrown in the {@link ConsumerWithException} will be * converted into {@link RuntimeException}s. * + * @param executor must not be {@literal null}. * @param streamable must not be {@literal null}. * @param consumer must not be {@literal null}. */ - public static void run(Streamable streamable, ConsumerWithException consumer) { + public static void run(Executor executor, Streamable streamable, ConsumerWithException consumer) { + Assert.notNull(executor, "Executor must not be null!"); Assert.notNull(streamable, "Streamable must not be null!"); Assert.notNull(consumer, "Consumer must not be null!"); @@ -56,29 +60,31 @@ public class ExecutionUtils { log.error(o_O.getMessage(), o_O); throw new RuntimeException(o_O); } - })).collect(Collectors.toList()).forEach(CompletableFuture::join); + }, executor)).collect(Collectors.toList()).forEach(CompletableFuture::join); } /** * Runs the given {@link Function} for each element in the given {@link Streamable} in parallel waiting for all * executions to complete before returning the results. * + * @param executor must not be {@literal null}. * @param streamable must not be {@literal null}. * @param function must not be {@literal null}. * @return */ - public static Collection runAndReturn(Streamable streamable, Function function) { - return runAndReturn(streamable, function, Collectors.toList()); + public static Collection runAndReturn(Executor executor, Streamable streamable, + Function function) { + return runAndReturn(executor, streamable, function, Collectors.toList()); } - public static R runAndReturn(Streamable streamable, Function function, + public static R runAndReturn(Executor executor, Streamable streamable, Function function, Collector collector) { Assert.notNull(streamable, "Iterable must not be null!"); Assert.notNull(function, "Function must not be null!"); return streamable.stream().// - map(it -> CompletableFuture.supplyAsync(() -> function.apply(it))).// + map(it -> CompletableFuture.supplyAsync(() -> function.apply(it), executor)).// filter(Objects::nonNull).// collect(Collectors.toList()).// stream().// diff --git a/release-tools/src/main/java/org/springframework/data/release/utils/ExecutorConfiguration.java b/release-tools/src/main/java/org/springframework/data/release/utils/ExecutorConfiguration.java new file mode 100644 index 0000000..e084008 --- /dev/null +++ b/release-tools/src/main/java/org/springframework/data/release/utils/ExecutorConfiguration.java @@ -0,0 +1,37 @@ +/* + * Copyright 2020 the original author or authors. + * + * Licensed 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. + */ +package org.springframework.data.release.utils; + +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.scheduling.concurrent.ThreadPoolExecutorFactoryBean; + +/** + * @author Mark Paluch + */ +@Configuration(proxyBeanMethods = false) +class ExecutorConfiguration { + + @Bean + public ThreadPoolExecutorFactoryBean executorService() { + + ThreadPoolExecutorFactoryBean scheduler = new ThreadPoolExecutorFactoryBean(); + scheduler.setCorePoolSize(Runtime.getRuntime().availableProcessors()); + scheduler.setQueueCapacity(32); + + return scheduler; + } +}