Spring/spring-data-dev-tools
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

#30 - Introduce Executor bean to split worker pools.

Browse Source 
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.
This commit is contained in:
Mark Paluch
2020-04-29 15:46:17 +02:00
parent 0884570d34
commit c397bafcd0
6 changed files with 83 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
release-tools/src/main/java/org/springframework/data/release/git/GitOperations.java
View File

@@ -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, Project> 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();

18
release-tools/src/main/java/org/springframework/data/release/issues/IssueTrackerCommands.java
View File

@@ -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<IssueTracker, Project> 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) {

6
release-tools/src/main/java/org/springframework/data/release/sagan/SaganConfiguration.java
View File

@@ -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

8
release-tools/src/main/java/org/springframework/data/release/sagan/SaganOperations.java
View File

@@ -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<Project, MaintainedVersions> 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);
});

18
release-tools/src/main/java/org/springframework/data/release/utils/ExecutionUtils.java
View File

@@ -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 <T> void run(Streamable<T> streamable, ConsumerWithException<T> consumer) {
public static <T> void run(Executor executor, Streamable<T> streamable, ConsumerWithException<T> 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 <T, S> Collection<S> runAndReturn(Streamable<T> streamable, Function<T, S> function) {
return runAndReturn(streamable, function, Collectors.toList());
public static <T, S> Collection<S> runAndReturn(Executor executor, Streamable<T> streamable,
Function<T, S> function) {
return runAndReturn(executor, streamable, function, Collectors.toList());
}
public static <T, S, R> R runAndReturn(Streamable<T> streamable, Function<T, S> function,
public static <T, S, R> R runAndReturn(Executor executor, Streamable<T> streamable, Function<T, S> function,
Collector<? super S, ?, R> 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().//

37
release-tools/src/main/java/org/springframework/data/release/utils/ExecutorConfiguration.java Normal file
View File

@@ -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;
}
}