Skip to content

Commit 987ac86

Browse files
q3769q3769
committed
+ google style indent
1 parent 4c6997a commit 987ac86

File tree

7 files changed

+277
-263
lines changed

7 files changed

+277
-263
lines changed

‎pom.xml‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -150,8 +150,8 @@
150150
<configuration>
151151
<java>
152152
<palantirJavaFormat>
153-
<version>2.46.0</version> <!-- optional -->
154-
<style>PALANTIR</style> <!-- or AOSP/GOOGLE (optional) -->
153+
<version>2.50.0</version> <!-- optional -->
154+
<style>GOOGLE</style> <!-- or AOSP/GOOGLE (optional) -->
155155
<formatJavadoc>true
156156
</formatJavadoc> <!-- defaults to false (optional, requires at least Palantir 2.39.0) -->
157157
</palantirJavaFormat>

‎src/main/java/conottle/ClientTaskExecutor.java‎

Lines changed: 19 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -29,25 +29,25 @@
2929
import java.util.concurrent.Future;
3030

3131
/**
32-
* Implementation should provide throttle capability on concurrent tasks of each client, and optionally on total number
33-
* of clients serviced in parallel.
32+
* Implementation should provide throttle capability on concurrent tasks of each client, and
33+
* optionally on total number of clients serviced in parallel.
3434
*/
3535
public interface ClientTaskExecutor {
36-
/**
37-
* @param command {@link Runnable} command to run asynchronously. All such commands under the same {@code clientId}
38-
* are run in parallel, albeit throttled at a maximum concurrency.
39-
* @param clientId A key representing a client whose tasks are throttled while running in parallel
40-
* @return {@link Future} holding the run status of the {@code command}
41-
*/
42-
default Future<Void> execute(Runnable command, Object clientId) {
43-
return submit(Executors.callable(command, null), clientId);
44-
}
45-
/**
46-
* @param task {@link Callable} task to run asynchronously. All such tasks under the same {@code clientId} are run
47-
* in parallel, albeit throttled at a maximum concurrency.
48-
* @param clientId A key representing a client whose tasks are throttled while running in parallel
49-
* @param <V> Type of the task result
50-
* @return {@link Future} representing the result of the {@code task}
51-
*/
52-
<V> Future<V> submit(Callable<V> task, Object clientId);
36+
/**
37+
* @param command {@link Runnable} command to run asynchronously. All such commands under the same
38+
* {@code clientId} are run in parallel, albeit throttled at a maximum concurrency.
39+
* @param clientId A key representing a client whose tasks are throttled while running in parallel
40+
* @return {@link Future} holding the run status of the {@code command}
41+
*/
42+
default Future<Void> execute(Runnable command, Object clientId) {
43+
return submit(Executors.callable(command, null), clientId);
44+
}
45+
/**
46+
* @param task {@link Callable} task to run asynchronously. All such tasks under the same
47+
* {@code clientId} are run in parallel, albeit throttled at a maximum concurrency.
48+
* @param clientId A key representing a client whose tasks are throttled while running in parallel
49+
* @param <V> Type of the task result
50+
* @return {@link Future} representing the result of the {@code task}
51+
*/
52+
<V> Future<V> submit(Callable<V> task, Object clientId);
5353
}

‎src/main/java/conottle/Conottle.java‎

Lines changed: 73 additions & 65 deletions
Original file line numberDiff line numberDiff line change
@@ -37,81 +37,89 @@
3737
import lombok.NonNull;
3838
import lombok.ToString;
3939

40-
/** Provides throttling on concurrent tasks per client, and on total number of clients serviced concurrently. */
40+
/**
41+
* Provides throttling on concurrent tasks per client, and on total number of clients serviced
42+
* concurrently.
43+
*/
4144
@ThreadSafe
4245
@ToString
4346
public final class Conottle implements ClientTaskExecutor, AutoCloseable {
44-
private static final Logger logger = Logger.instance();
45-
private static final int DEFAULT_MAX = Math.max(16, Runtime.getRuntime().availableProcessors());
46-
private final ExecutorService adminExecutorService =
47-
Executors.newCachedThreadPool(ThreadFactories.newPlatformThreadFactory("conottle-admin"));
47+
private static final Logger logger = Logger.instance();
48+
private static final int DEFAULT_MAX = Math.max(16, Runtime.getRuntime().availableProcessors());
49+
private final ExecutorService adminExecutorService =
50+
Executors.newCachedThreadPool(ThreadFactories.newPlatformThreadFactory("conottle-admin"));
4851

49-
private final ConcurrentMap<Object, SingleClientThrottlingTaskExecutor> activeThrottlingExecutors;
50-
private final Semaphore clientTotalThrottle;
51-
private final int maxParallelismPerClient;
52-
private final ExecutorService workerExecutorService;
52+
private final ConcurrentMap<Object, SingleClientThrottlingTaskExecutor> activeThrottlingExecutors;
53+
private final Semaphore clientTotalThrottle;
54+
private final int maxParallelismPerClient;
55+
private final ExecutorService workerExecutorService;
5356

54-
@Builder
55-
private Conottle(int maxClientsInParallel, int maxParallelismPerClient, ExecutorService workerExecutorService) {
56-
this.activeThrottlingExecutors = new ConcurrentHashMap<>(requireNonNegativeOrThrow(maxClientsInParallel));
57-
this.clientTotalThrottle = new Semaphore(requireNonNegativeOrThrow(maxClientsInParallel));
58-
this.maxParallelismPerClient = requireNonNegativeOrThrow(maxParallelismPerClient);
59-
this.workerExecutorService =
60-
workerExecutorService == null ? Executors.newWorkStealingPool(DEFAULT_MAX) : workerExecutorService;
61-
logger.log("Constructed: {}", this);
62-
}
57+
@Builder
58+
private Conottle(
59+
int maxClientsInParallel,
60+
int maxParallelismPerClient,
61+
ExecutorService workerExecutorService) {
62+
this.activeThrottlingExecutors =
63+
new ConcurrentHashMap<>(requireNonNegativeOrThrow(maxClientsInParallel));
64+
this.clientTotalThrottle = new Semaphore(requireNonNegativeOrThrow(maxClientsInParallel));
65+
this.maxParallelismPerClient = requireNonNegativeOrThrow(maxParallelismPerClient);
66+
this.workerExecutorService = workerExecutorService == null
67+
? Executors.newWorkStealingPool(DEFAULT_MAX)
68+
: workerExecutorService;
69+
logger.log("Constructed: {}", this);
70+
}
6371

64-
private static int requireNonNegativeOrThrow(int value) {
65-
if (value < 0) {
66-
throw new IllegalArgumentException("Negative value is not allowed: " + value);
67-
}
68-
if (value == 0) {
69-
return DEFAULT_MAX;
70-
}
71-
return value;
72+
private static int requireNonNegativeOrThrow(int value) {
73+
if (value < 0) {
74+
throw new IllegalArgumentException("Negative value is not allowed: " + value);
7275
}
73-
74-
@Override
75-
public <V> @NonNull Future<V> submit(@NonNull Callable<V> task, @NonNull Object clientId) {
76-
CompletableFutureHolder<V> taskCompletableFutureHolder = new CompletableFutureHolder<>();
77-
activeThrottlingExecutors.compute(clientId, (k, valueExecutor) -> {
78-
SingleClientThrottlingTaskExecutor executor =
79-
(valueExecutor == null) ? acquireClientExecutor() : valueExecutor;
80-
taskCompletableFutureHolder.setCompletableFuture(executor.submit(task));
81-
return executor;
82-
});
83-
CompletableFuture<V> taskCompletableFuture = taskCompletableFutureHolder.getCompletableFuture();
84-
CompletableFuture<V> copy = taskCompletableFuture.thenApply(r -> r);
85-
taskCompletableFuture.whenCompleteAsync(
86-
(r, e) -> activeThrottlingExecutors.computeIfPresent(clientId, (k, valueExecutor) -> {
87-
if (valueExecutor.hasMoreWork()) {
88-
return valueExecutor;
89-
}
90-
releaseClientExecutor();
91-
return null;
92-
}),
93-
adminExecutorService);
94-
return new DefensiveFuture<>(copy);
76+
if (value == 0) {
77+
return DEFAULT_MAX;
9578
}
79+
return value;
80+
}
9681

97-
@Override
98-
public void close() {
99-
workerExecutorService.shutdown();
100-
await().forever().until(workerExecutorService::isTerminated);
101-
adminExecutorService.shutdown();
102-
}
82+
@Override
83+
public <V> @NonNull Future<V> submit(@NonNull Callable<V> task, @NonNull Object clientId) {
84+
CompletableFutureHolder<V> taskCompletableFutureHolder = new CompletableFutureHolder<>();
85+
activeThrottlingExecutors.compute(clientId, (k, valueExecutor) -> {
86+
SingleClientThrottlingTaskExecutor executor =
87+
(valueExecutor == null) ? acquireClientExecutor() : valueExecutor;
88+
taskCompletableFutureHolder.setCompletableFuture(executor.submit(task));
89+
return executor;
90+
});
91+
CompletableFuture<V> taskCompletableFuture = taskCompletableFutureHolder.getCompletableFuture();
92+
CompletableFuture<V> copy = taskCompletableFuture.thenApply(r -> r);
93+
taskCompletableFuture.whenCompleteAsync(
94+
(r, e) -> activeThrottlingExecutors.computeIfPresent(clientId, (k, valueExecutor) -> {
95+
if (valueExecutor.hasMoreWork()) {
96+
return valueExecutor;
97+
}
98+
releaseClientExecutor();
99+
return null;
100+
}),
101+
adminExecutorService);
102+
return new DefensiveFuture<>(copy);
103+
}
103104

104-
int countActiveExecutors() {
105-
return activeThrottlingExecutors.size();
106-
}
105+
@Override
106+
public void close() {
107+
workerExecutorService.shutdown();
108+
await().forever().until(workerExecutorService::isTerminated);
109+
adminExecutorService.shutdown();
110+
}
107111

108-
private @NonNull SingleClientThrottlingTaskExecutor acquireClientExecutor() {
109-
Semaphores.acquireInterruptiblyUnchecked(clientTotalThrottle);
110-
return new SingleClientThrottlingTaskExecutor(
111-
maxParallelismPerClient, workerExecutorService, adminExecutorService);
112-
}
112+
int countActiveExecutors() {
113+
return activeThrottlingExecutors.size();
114+
}
113115

114-
private void releaseClientExecutor() {
115-
clientTotalThrottle.release();
116-
}
116+
private @NonNull SingleClientThrottlingTaskExecutor acquireClientExecutor() {
117+
Semaphores.acquireInterruptiblyUnchecked(clientTotalThrottle);
118+
return new SingleClientThrottlingTaskExecutor(
119+
maxParallelismPerClient, workerExecutorService, adminExecutorService);
120+
}
121+
122+
private void releaseClientExecutor() {
123+
clientTotalThrottle.release();
124+
}
117125
}

‎src/main/java/conottle/SingleClientThrottlingTaskExecutor.java‎

Lines changed: 19 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -32,26 +32,26 @@
3232

3333
@ToString
3434
final class SingleClientThrottlingTaskExecutor {
35-
private final int maxParallelism;
36-
private final Semaphore clientParallelismThrottle;
37-
private final ExecutorService workerExecutor;
38-
private final ExecutorService adminExecutor;
35+
private final int maxParallelism;
36+
private final Semaphore clientParallelismThrottle;
37+
private final ExecutorService workerExecutor;
38+
private final ExecutorService adminExecutor;
3939

40-
public SingleClientThrottlingTaskExecutor(
41-
int maxParallelismPerClient, ExecutorService workerExecutor, ExecutorService adminExecutor) {
42-
this.maxParallelism = maxParallelismPerClient;
43-
this.clientParallelismThrottle = new Semaphore(maxParallelismPerClient);
44-
this.workerExecutor = workerExecutor;
45-
this.adminExecutor = adminExecutor;
46-
}
40+
public SingleClientThrottlingTaskExecutor(
41+
int maxParallelismPerClient, ExecutorService workerExecutor, ExecutorService adminExecutor) {
42+
this.maxParallelism = maxParallelismPerClient;
43+
this.clientParallelismThrottle = new Semaphore(maxParallelismPerClient);
44+
this.workerExecutor = workerExecutor;
45+
this.adminExecutor = adminExecutor;
46+
}
4747

48-
public boolean hasMoreWork() {
49-
return clientParallelismThrottle.availablePermits() < maxParallelism;
50-
}
48+
public boolean hasMoreWork() {
49+
return clientParallelismThrottle.availablePermits() < maxParallelism;
50+
}
5151

52-
public @NonNull <V> CompletableFuture<V> submit(Callable<V> task) {
53-
Semaphores.acquireInterruptiblyUnchecked(clientParallelismThrottle);
54-
return CompletableFuture.supplyAsync(Tasks.asSupplierUnchecked(task), workerExecutor)
55-
.whenCompleteAsync((r, e) -> clientParallelismThrottle.release(), adminExecutor);
56-
}
52+
public @NonNull <V> CompletableFuture<V> submit(Callable<V> task) {
53+
Semaphores.acquireInterruptiblyUnchecked(clientParallelismThrottle);
54+
return CompletableFuture.supplyAsync(Tasks.asSupplierUnchecked(task), workerExecutor)
55+
.whenCompleteAsync((r, e) -> clientParallelismThrottle.release(), adminExecutor);
56+
}
5757
}

0 commit comments

Comments
 (0)