Skip to content

Commit 346c7be

Browse files
BlackBaronessBlackBaroness
committed
rework core scheduler and use it for bungee
1 parent bed5711 commit 346c7be

File tree

4 files changed

+170
-163
lines changed

4 files changed

+170
-163
lines changed

‎litecommands-bungee/src/dev/rollczi/litecommands/bungee/BungeeSchedulerImpl.java

Lines changed: 0 additions & 121 deletions
This file was deleted.

‎litecommands-core/src/dev/rollczi/litecommands/scheduler/SchedulerExecutorPoolBuilder.java

Lines changed: 67 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,67 @@
1+
package dev.rollczi.litecommands.scheduler;
2+
3+
import java.util.concurrent.ScheduledExecutorService;
4+
import java.util.concurrent.ScheduledThreadPoolExecutor;
5+
import java.util.concurrent.TimeUnit;
6+
import java.util.logging.Logger;
7+
8+
public class SchedulerExecutorPoolBuilder {
9+
10+
private Logger logger;
11+
12+
private ScheduledExecutorService mainExecutor;
13+
private boolean closeMainExecutorOnShutdown;
14+
15+
private ScheduledExecutorService asyncExecutor;
16+
private boolean closeAsyncExecutorOnShutdown;
17+
18+
public void setMainExecutor(ScheduledExecutorService mainExecutor, boolean closeOnShutdown) {
19+
this.mainExecutor = mainExecutor;
20+
this.closeMainExecutorOnShutdown = closeOnShutdown;
21+
}
22+
23+
public void setAsyncExecutor(ScheduledExecutorService asyncExecutor, boolean closeOnShutdown) {
24+
this.asyncExecutor = asyncExecutor;
25+
this.closeAsyncExecutorOnShutdown = closeOnShutdown;
26+
}
27+
28+
public SchedulerExecutorPoolImpl build() {
29+
if (logger == null) {
30+
logger = Logger.getLogger("LiteCommands");
31+
}
32+
33+
if (mainExecutor == null) {
34+
/*
35+
In some cases, the main executor might never be used. We don't want to have to create a useless thread
36+
37+
This may look like a premature optimization, but a typical server has 40+ plugins
38+
As the framework spreads, more and more of them can use LiteCommands,
39+
so we better handle this situation, there are practically no losses anyway
40+
*/
41+
ScheduledThreadPoolExecutor mainExecutor = new ScheduledThreadPoolExecutor(1);
42+
mainExecutor.allowCoreThreadTimeOut(true);
43+
mainExecutor.setKeepAliveTime(1, TimeUnit.HOURS);
44+
45+
this.mainExecutor = mainExecutor;
46+
closeMainExecutorOnShutdown = true;
47+
}
48+
49+
if (asyncExecutor == null) {
50+
/*
51+
We want to create a thread pool that both is fast and uses as few threads as possible.
52+
So we use a fixed core pool size to not hold an unreasonable number of threads on server CPUs
53+
54+
We could also decrease the max cap, but long-running commands are those that use I/O,
55+
so it's fine and meaningful to create a bigger pool in this case
56+
*/
57+
ScheduledThreadPoolExecutor asyncExecutor = new ScheduledThreadPoolExecutor(4);
58+
asyncExecutor.setMaximumPoolSize(Runtime.getRuntime().availableProcessors() * 2);
59+
asyncExecutor.setKeepAliveTime(3, TimeUnit.MINUTES);
60+
61+
this.asyncExecutor = asyncExecutor;
62+
closeAsyncExecutorOnShutdown = true;
63+
}
64+
65+
return new SchedulerExecutorPoolImpl(logger, mainExecutor, closeMainExecutorOnShutdown, asyncExecutor, closeAsyncExecutorOnShutdown);
66+
}
67+
}

‎litecommands-core/src/dev/rollczi/litecommands/scheduler/SchedulerExecutorPoolImpl.java

Lines changed: 101 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -1,91 +1,151 @@
11
package dev.rollczi.litecommands.scheduler;
22

3+
import dev.rollczi.litecommands.shared.ThrowingRunnable;
34
import dev.rollczi.litecommands.shared.ThrowingSupplier;
5+
import org.jetbrains.annotations.ApiStatus;
46

7+
import java.io.PrintWriter;
8+
import java.io.StringWriter;
59
import java.time.Duration;
610
import java.util.concurrent.CompletableFuture;
7-
import java.util.concurrent.ExecutorService;
11+
import java.util.concurrent.CompletionException;
812
import java.util.concurrent.Executors;
913
import java.util.concurrent.ScheduledExecutorService;
1014
import java.util.concurrent.ThreadFactory;
1115
import java.util.concurrent.TimeUnit;
1216
import java.util.concurrent.atomic.AtomicInteger;
17+
import java.util.logging.Logger;
1318

1419
public class SchedulerExecutorPoolImpl implements Scheduler {
1520

16-
private static final String MAIN_THREAD_NAME_FORMAT = "scheduler-%s-main";
17-
private static final String ASYNC_THREAD_NAME_FORMAT = "scheduler-%s-async-%d";
21+
private final Logger logger;
22+
private final ScheduledExecutorService mainExecutor;
23+
private final boolean closeMainExecutorOnShutdown;
1824

19-
private final ThreadLocal<Boolean> isMainThread = ThreadLocal.withInitial(() -> false);
20-
21-
private final ExecutorService mainExecutor;
22-
private final ExecutorService asyncExecutor;
23-
private final ScheduledExecutorService scheduledExecutor = Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors());
25+
private final ScheduledExecutorService asyncExecutor;
26+
private final boolean closeAsyncExecutorOnShutdown;
2427

28+
@Deprecated
2529
public SchedulerExecutorPoolImpl(String name) {
2630
this(name, -1);
2731
}
2832

33+
@Deprecated
2934
public SchedulerExecutorPoolImpl(String name, int pool) {
35+
this.logger = Logger.getLogger("LiteCommands");
36+
3037
this.mainExecutor = Executors.newSingleThreadScheduledExecutor(runnable -> {
3138
Thread thread = new Thread(runnable);
32-
thread.setName(String.format(MAIN_THREAD_NAME_FORMAT, name));
33-
39+
thread.setName(String.format("scheduler-%s-main", name));
3440
return thread;
3541
});
36-
37-
this.mainExecutor.submit(() -> isMainThread.set(true));
42+
this.closeMainExecutorOnShutdown = true;
3843

3944
AtomicInteger asyncCount = new AtomicInteger();
4045
ThreadFactory factory = runnable -> {
4146
Thread thread = new Thread(runnable);
42-
thread.setName(String.format(ASYNC_THREAD_NAME_FORMAT, name, asyncCount.getAndIncrement()));
47+
thread.setName(String.format("scheduler-%s-async-%d", name, asyncCount.getAndIncrement()));
4348

4449
return thread;
4550
};
4651

47-
this.asyncExecutor = pool < 0 ? Executors.newCachedThreadPool(factory) : Executors.newFixedThreadPool(pool, factory);
52+
this.asyncExecutor = Executors.newScheduledThreadPool(
53+
Math.max(pool, Runtime.getRuntime().availableProcessors()) / 2,
54+
factory
55+
);
56+
this.closeAsyncExecutorOnShutdown = true;
4857
}
4958

50-
@Override
51-
public <T> CompletableFuture<T> supplyLater(SchedulerPoll type, Duration delay, ThrowingSupplier<T, Throwable> supplier) {
52-
SchedulerPoll resolve = type.resolve(SchedulerPoll.MAIN, SchedulerPoll.ASYNCHRONOUS);
53-
CompletableFuture<T> future = new CompletableFuture<>();
59+
/**
60+
* Internal usage only. Use {@link SchedulerExecutorPoolBuilder}.
61+
*/
62+
@ApiStatus.Internal
63+
public SchedulerExecutorPoolImpl(Logger logger, ScheduledExecutorService mainExecutor, boolean closeMainExecutorOnShutdown, ScheduledExecutorService asyncExecutor, boolean closeAsyncExecutorOnShutdown) {
64+
this.logger = logger;
65+
this.mainExecutor = mainExecutor;
66+
this.closeMainExecutorOnShutdown = closeMainExecutorOnShutdown;
67+
this.asyncExecutor = asyncExecutor;
68+
this.closeAsyncExecutorOnShutdown = closeAsyncExecutorOnShutdown;
69+
}
5470

55-
if (resolve.equals(SchedulerPoll.MAIN) && delay.isZero() && isMainThread.get()) {
56-
return tryRun(supplier, future);
57-
}
71+
@Override
72+
public CompletableFuture<Void> run(SchedulerPoll type, ThrowingRunnable<Throwable> runnable) {
73+
return CompletableFuture.runAsync(() -> {
74+
try {
75+
runnable.run();
76+
} catch (Throwable e) {
77+
throw new CompletionException(e);
78+
}
79+
}, getSchedulerByPollType(type));
80+
}
5881

59-
ExecutorService executor = resolve.equals(SchedulerPoll.MAIN) ? mainExecutor : asyncExecutor;
82+
@Override
83+
public CompletableFuture<Void> runLater(SchedulerPoll type, Duration delay, ThrowingRunnable<Throwable> runnable) {
84+
CompletableFuture<Void> future = new CompletableFuture<>();
85+
getSchedulerByPollType(type).schedule(() -> {
86+
try {
87+
runnable.run();
88+
future.complete(null);
89+
} catch (Throwable e) {
90+
future.completeExceptionally(e);
91+
}
92+
}, delay.toMillis(), TimeUnit.MILLISECONDS);
93+
return future;
94+
}
6095

61-
if (delay.isZero()) {
62-
executor.submit(() -> tryRun(supplier, future));
63-
}
64-
else {
65-
scheduledExecutor.schedule(() -> {
66-
executor.submit(() -> tryRun(supplier, future));
67-
}, delay.toMillis(), TimeUnit.MILLISECONDS);
68-
}
96+
@Override
97+
public <T> CompletableFuture<T> supply(SchedulerPoll type, ThrowingSupplier<T, Throwable> supplier) {
98+
return CompletableFuture.supplyAsync(() -> {
99+
try {
100+
return supplier.get();
101+
} catch (Throwable e) {
102+
throw new CompletionException(e);
103+
}
104+
}, getSchedulerByPollType(type));
105+
}
69106

107+
@Override
108+
public <T> CompletableFuture<T> supplyLater(SchedulerPoll type, Duration delay, ThrowingSupplier<T, Throwable> supplier) {
109+
CompletableFuture<T> future = new CompletableFuture<>();
110+
getSchedulerByPollType(type).schedule(() -> {
111+
try {
112+
future.complete(supplier.get());
113+
} catch (Throwable e) {
114+
future.completeExceptionally(e);
115+
}
116+
}, delay.toMillis(), TimeUnit.MILLISECONDS);
70117
return future;
71118
}
72119

73-
private static <T> CompletableFuture<T> tryRun(ThrowingSupplier<T, Throwable> supplier, CompletableFuture<T> future) {
74-
try {
75-
future.complete(supplier.get());
120+
@Override
121+
public void shutdown() {
122+
if (closeMainExecutorOnShutdown) {
123+
try {
124+
mainExecutor.close();
125+
} catch (Throwable e) {
126+
logger.severe("Error closing main executor: \n" + getStacktraceAsString(e));
127+
}
76128
}
77-
catch (Throwable throwable) {
78-
future.completeExceptionally(throwable);
129+
130+
if (closeAsyncExecutorOnShutdown) {
131+
try {
132+
asyncExecutor.close();
133+
} catch (Throwable e) {
134+
logger.severe("Error closing async executor: \n" + getStacktraceAsString(e));
135+
}
79136
}
137+
}
80138

81-
return future;
139+
private ScheduledExecutorService getSchedulerByPollType(SchedulerPoll type) {
140+
return type.resolve(SchedulerPoll.MAIN, SchedulerPoll.ASYNCHRONOUS).equals(SchedulerPoll.MAIN)
141+
? mainExecutor
142+
: asyncExecutor;
82143
}
83144

84-
@Override
85-
public void shutdown() {
86-
mainExecutor.shutdown();
87-
asyncExecutor.shutdown();
88-
isMainThread.remove();
145+
private String getStacktraceAsString(Throwable throwable) {
146+
StringWriter stringWriter = new StringWriter();
147+
throwable.printStackTrace(new PrintWriter(stringWriter));
148+
return stringWriter.toString();
89149
}
90150

91151
}

‎litecommands-framework/src/dev/rollczi/litecommands/LiteCommandsBaseBuilder.java

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,7 @@
4444
import dev.rollczi.litecommands.reflect.type.TypeRange;
4545
import dev.rollczi.litecommands.requirement.RequirementMatchService;
4646
import dev.rollczi.litecommands.scheduler.Scheduler;
47+
import dev.rollczi.litecommands.scheduler.SchedulerExecutorPoolBuilder;
4748
import dev.rollczi.litecommands.scheduler.SchedulerExecutorPoolImpl;
4849
import dev.rollczi.litecommands.scheduler.SchedulerReference;
4950
import dev.rollczi.litecommands.schematic.SchematicFastFormat;
@@ -162,7 +163,7 @@ public LiteCommandsBaseBuilder(
162163
this.messageRegistry = messageRegistry;
163164
this.strictService = strictService;
164165

165-
this.scheduler = new SchedulerReference(new SchedulerExecutorPoolImpl("litecommands"));
166+
this.scheduler = new SchedulerReference(new SchedulerExecutorPoolBuilder().build());
166167
this.eventPublisher = new SimpleEventPublisher(bindRegistry);
167168
this.schematicGenerator = new SchematicGeneratorReference<>(new SchematicFastGenerator<>(SchematicFormat.angleBrackets(), validatorService, parserRegistry));
168169
}

0 commit comments

Comments
 (0)