Universal Task Scheduling
This commit is contained in:
Molzonas
parent
1ce7188931
commit
ead567fd94
@ -0,0 +1,13 @@
|
||||
package fr.molzonas.mzcore.exception;
|
||||
|
||||
public class MZTaskException extends RuntimeException {
|
||||
public MZTaskException(String message) {
|
||||
super(message);
|
||||
}
|
||||
public MZTaskException(Exception e) {
|
||||
super(e);
|
||||
}
|
||||
public MZTaskException(Throwable t) {
|
||||
super(t);
|
||||
}
|
||||
}
|
||||
13
src/main/java/fr/molzonas/mzcore/tasks/MZTaskHandle.java
Normal file
13
src/main/java/fr/molzonas/mzcore/tasks/MZTaskHandle.java
Normal file
@ -0,0 +1,13 @@
|
||||
package fr.molzonas.mzcore.tasks;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.UUID;
|
||||
|
||||
public interface MZTaskHandle {
|
||||
UUID getId();
|
||||
boolean cancel();
|
||||
boolean isDone();
|
||||
boolean isCancelled();
|
||||
Duration getDuration();
|
||||
Throwable getCause();
|
||||
}
|
||||
25
src/main/java/fr/molzonas/mzcore/tasks/MZTaskInfo.java
Normal file
25
src/main/java/fr/molzonas/mzcore/tasks/MZTaskInfo.java
Normal file
@ -0,0 +1,25 @@
|
||||
package fr.molzonas.mzcore.tasks;
|
||||
|
||||
import fr.molzonas.mzcore.tasks.enums.MZTaskKind;
|
||||
import fr.molzonas.mzcore.tasks.enums.MZTaskStatus;
|
||||
import fr.molzonas.mzcore.tasks.enums.MZTaskType;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.UUID;
|
||||
|
||||
public record MZTaskInfo(
|
||||
UUID id,
|
||||
MZTaskType type,
|
||||
MZTaskKind kind,
|
||||
MZTaskStatus status,
|
||||
Long start,
|
||||
Long end,
|
||||
int runCount,
|
||||
String note,
|
||||
Throwable error
|
||||
) {
|
||||
public Duration duration() {
|
||||
long end = (end() != null) ? end() : System.nanoTime();
|
||||
return Duration.ofNanos(end - start);
|
||||
}
|
||||
}
|
||||
316
src/main/java/fr/molzonas/mzcore/tasks/MZTaskScheduler.java
Normal file
316
src/main/java/fr/molzonas/mzcore/tasks/MZTaskScheduler.java
Normal file
@ -0,0 +1,316 @@
|
||||
package fr.molzonas.mzcore.tasks;
|
||||
|
||||
import fr.molzonas.mzcore.exception.MZTaskException;
|
||||
import fr.molzonas.mzcore.tasks.enums.MZTaskKind;
|
||||
import fr.molzonas.mzcore.tasks.enums.MZTaskStatus;
|
||||
import fr.molzonas.mzcore.tasks.enums.MZTaskType;
|
||||
import org.bukkit.Bukkit;
|
||||
import org.bukkit.plugin.Plugin;
|
||||
import org.bukkit.scheduler.BukkitTask;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.time.temporal.ChronoUnit;
|
||||
import java.util.Collection;
|
||||
import java.util.List;
|
||||
import java.util.Objects;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.*;
|
||||
import java.util.function.Consumer;
|
||||
import java.util.function.Function;
|
||||
import java.util.function.Supplier;
|
||||
import java.util.logging.Level;
|
||||
|
||||
public class MZTaskScheduler implements MZTaskSchedulerInterface {
|
||||
private final Plugin plugin;
|
||||
private final ExecutorService asyncExecutor;
|
||||
private final ScheduledExecutorService asyncScheduler;
|
||||
private final ConcurrentHashMap<UUID, MZTaskInfo> tasks = new ConcurrentHashMap<>();
|
||||
|
||||
public MZTaskScheduler(Plugin plugin, int asyncThreads) {
|
||||
this.plugin = plugin;
|
||||
this.asyncExecutor = Executors.newFixedThreadPool(Math.max(1, asyncThreads),
|
||||
r -> {
|
||||
Thread t = new Thread(r, "MZCore-Async");
|
||||
t.setDaemon(true);
|
||||
return t;
|
||||
}
|
||||
);
|
||||
this.asyncScheduler = Executors.newScheduledThreadPool(1,
|
||||
r -> {
|
||||
Thread t = new Thread(r, "MZCore-AsyncScheduled");
|
||||
t.setDaemon(true);
|
||||
return t;
|
||||
}
|
||||
);
|
||||
}
|
||||
|
||||
@Override
|
||||
public MZTaskHandle runSync(Runnable task) {
|
||||
UUID id = newId(MZTaskType.SYNC, MZTaskKind.RUN, "runSync");
|
||||
BukkitTask bt = Bukkit.getScheduler().runTask(plugin, () -> wrapOnce(id, task));
|
||||
return new BukkitHandle(id, bt, this::lookupInfo);
|
||||
}
|
||||
|
||||
@Override
|
||||
public MZTaskHandle runAsync(Runnable task) {
|
||||
UUID id = newId(MZTaskType.ASYNC, MZTaskKind.RUN, "runAsync");
|
||||
Future<?> f = asyncExecutor.submit(() -> wrapOnce(id, task));
|
||||
return new FutureHandle(id, f, this::lookupInfo);
|
||||
}
|
||||
|
||||
@Override
|
||||
public MZTaskHandle scheduleSync(Runnable task, long delay) {
|
||||
UUID id = newId(MZTaskType.SYNC, MZTaskKind.SCHEDULE, "scheduleSync");
|
||||
BukkitTask bt = Bukkit.getScheduler().runTaskLater(plugin, () -> wrapOnce(id, task), delay);
|
||||
return new BukkitHandle(id, bt, this::lookupInfo);
|
||||
}
|
||||
|
||||
@Override
|
||||
public MZTaskHandle scheduleAsync(Runnable task, Duration delay) {
|
||||
UUID id = newId(MZTaskType.ASYNC, MZTaskKind.SCHEDULE, "scheduleAsync");
|
||||
ScheduledFuture<?> f = asyncScheduler.schedule(() -> wrapOnce(id, task), delay.toMillis(), TimeUnit.MILLISECONDS);
|
||||
return new FutureHandle(id, f, this::lookupInfo);
|
||||
}
|
||||
|
||||
@Override
|
||||
public MZTaskHandle repeatSync(Runnable task, long initialDelay, long period) {
|
||||
UUID id = newId(MZTaskType.SYNC, MZTaskKind.REPEAT, "repeatSync");
|
||||
BukkitTask bt = Bukkit.getScheduler().runTaskTimer(plugin, () -> wrapRepeatTick(id, task), initialDelay, period);
|
||||
return new BukkitHandle(id, bt, this::lookupInfo, true);
|
||||
}
|
||||
|
||||
@Override
|
||||
public MZTaskHandle repeatAsync(Runnable task, Duration initialDelay, Duration period) {
|
||||
UUID id = newId(MZTaskType.ASYNC, MZTaskKind.REPEAT, "repeatAsync");
|
||||
ScheduledFuture<?> f = asyncScheduler.scheduleAtFixedRate(() -> wrapRepeatTick(id, task),
|
||||
initialDelay.toMillis(), period.toMillis(), TimeUnit.MILLISECONDS);
|
||||
return new FutureHandle(id, f, this::lookupInfo, true);
|
||||
}
|
||||
|
||||
@Override
|
||||
public <T> CompletableFuture<T> supplyAsync(Supplier<T> supplier) {
|
||||
UUID id = newId(MZTaskType.ASYNC, MZTaskKind.SUPPLY, "supplyAsync");
|
||||
return CompletableFuture.supplyAsync(() -> {
|
||||
markRunning(id);
|
||||
try {
|
||||
T val = supplier.get();
|
||||
markDone(id);
|
||||
return val;
|
||||
} catch (Exception e) {
|
||||
markFailed(id, e);
|
||||
throw new MZTaskException(e);
|
||||
}
|
||||
}, asyncExecutor);
|
||||
}
|
||||
|
||||
@Override
|
||||
public <T> void thenSync(CompletionStage<T> completionStage, Consumer<T> consumer) {
|
||||
completionStage.whenComplete((val, err) ->
|
||||
Bukkit.getScheduler().runTask(plugin, () -> {
|
||||
if (err != null) throwUnchecked(unwrapCompletionException(err));
|
||||
consumer.accept(val);
|
||||
})
|
||||
);
|
||||
}
|
||||
|
||||
@Override
|
||||
public <T> CompletionStage<T> withTimeout(CompletionStage<T> stage, Duration timeout) {
|
||||
Objects.requireNonNull(stage, "stage");
|
||||
Objects.requireNonNull(timeout, "timeout");
|
||||
CompletableFuture<T> result = new CompletableFuture<>();
|
||||
stage.whenComplete((val, err) -> {
|
||||
if (err != null) result.completeExceptionally(unwrapCompletionException(err));
|
||||
else result.complete(val);
|
||||
});
|
||||
ScheduledFuture<?> killer = asyncScheduler.schedule(
|
||||
() -> result.completeExceptionally(new TimeoutException("Task timeout after " + timeout)),
|
||||
timeout.toMillis(), TimeUnit.MILLISECONDS
|
||||
);
|
||||
result.whenComplete((v, e) -> killer.cancel(false));
|
||||
return result;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Collection<MZTaskInfo> snapshot() {
|
||||
return List.copyOf(tasks.values());
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
try {
|
||||
Bukkit.getScheduler().cancelTasks(plugin);
|
||||
} finally {
|
||||
asyncScheduler.shutdownNow();
|
||||
asyncExecutor.shutdownNow();
|
||||
tasks.clear();
|
||||
}
|
||||
}
|
||||
|
||||
private UUID newId(MZTaskType type, MZTaskKind kind, String note) {
|
||||
UUID id = UUID.randomUUID();
|
||||
long now = System.nanoTime();
|
||||
tasks.put(id, new MZTaskInfo(id, type, kind, MZTaskStatus.PENDING, now, null, 0, note, null));
|
||||
return id;
|
||||
}
|
||||
|
||||
private void wrapOnce(UUID id, Runnable r) {
|
||||
markRunning(id);
|
||||
try {
|
||||
r.run();
|
||||
markDone(id);
|
||||
} catch (Exception e) {
|
||||
markFailed(id, e);
|
||||
throwUnchecked(e);
|
||||
}
|
||||
}
|
||||
|
||||
private void wrapRepeatTick(UUID id, Runnable r) {
|
||||
markRunning(id);
|
||||
try {
|
||||
r.run();
|
||||
} catch (Exception e) {
|
||||
markFailed(id, e);
|
||||
plugin.getLogger().log(Level.WARNING, "[MZTaskScheduler] Task " + id + " failed", e);
|
||||
} finally {
|
||||
// entre deux exécutions
|
||||
markWaiting(id);
|
||||
}
|
||||
}
|
||||
private void markRunning(UUID id) {
|
||||
tasks.computeIfPresent(id, (k, v) -> new MZTaskInfo(id, v.type(), v.kind(), MZTaskStatus.RUNNING,
|
||||
v.start(), v.end(), v.runCount() + 1, v.note(), null));
|
||||
}
|
||||
|
||||
private void markWaiting(UUID id) {
|
||||
tasks.computeIfPresent(id, (k, v) -> new MZTaskInfo(id, v.type(), v.kind(), MZTaskStatus.WAITING,
|
||||
v.start(), v.end(), v.runCount(), v.note(), v.error()));
|
||||
}
|
||||
|
||||
private void markDone(UUID id) {
|
||||
long end = System.nanoTime();
|
||||
tasks.computeIfPresent(id, (k, v) -> new MZTaskInfo(id, v.type(), v.kind(), MZTaskStatus.COMPLETED,
|
||||
v.start(), end, v.runCount(), v.note(), v.error()));
|
||||
tasks.remove(id);
|
||||
}
|
||||
|
||||
private void markCancelled(UUID id) {
|
||||
tasks.computeIfPresent(id, (k, v) -> new MZTaskInfo(id, v.type(), v.kind(), MZTaskStatus.CANCELLED,
|
||||
v.start(), v.end(), v.runCount(), v.note(), v.error()));
|
||||
}
|
||||
|
||||
private void markFailed(UUID id, Exception e) {
|
||||
long end = System.nanoTime();
|
||||
tasks.computeIfPresent(id, (k, v) -> new MZTaskInfo(id, v.type(), v.kind(), MZTaskStatus.FAILED,
|
||||
v.start(), end, v.runCount(), v.note(), e));
|
||||
}
|
||||
|
||||
private static void throwUnchecked(Throwable e) {
|
||||
if (e instanceof RuntimeException re) throw re;
|
||||
if (e instanceof Error er) throw er;
|
||||
throw new MZTaskException(e);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Plugin getPlugin() {
|
||||
return this.plugin;
|
||||
}
|
||||
|
||||
private MZTaskInfo lookupInfo(UUID id) {
|
||||
return tasks.get(id);
|
||||
}
|
||||
|
||||
private final class FutureHandle implements MZTaskHandle {
|
||||
private final UUID id;
|
||||
private final Future<?> future;
|
||||
private final Function<UUID, MZTaskInfo> infoFn;
|
||||
private final boolean repeating;
|
||||
|
||||
FutureHandle(UUID id, Future<?> future, Function<UUID, MZTaskInfo> infoFn) {
|
||||
this(id, future, infoFn, false);
|
||||
}
|
||||
FutureHandle(UUID id, Future<?> future, Function<UUID, MZTaskInfo> infoFn, boolean repeating) {
|
||||
this.id = id; this.future = future; this.infoFn = infoFn; this.repeating = repeating;
|
||||
}
|
||||
|
||||
@Override public UUID getId() { return id; }
|
||||
|
||||
@Override public boolean cancel() {
|
||||
boolean ok = future.cancel(true);
|
||||
markCancelled(id);
|
||||
return ok;
|
||||
}
|
||||
|
||||
@Override public boolean isDone() {
|
||||
MZTaskInfo ti = infoFn.apply(id);
|
||||
if (ti == null) return future.isDone();
|
||||
return switch (ti.status()) {
|
||||
case COMPLETED, FAILED, CANCELLED -> true;
|
||||
default -> !repeating && future.isDone();
|
||||
};
|
||||
}
|
||||
|
||||
@Override public boolean isCancelled() {
|
||||
MZTaskInfo ti = infoFn.apply(id);
|
||||
return ti != null && ti.status() == MZTaskStatus.CANCELLED || future.isCancelled();
|
||||
}
|
||||
|
||||
@Override public Duration getDuration() {
|
||||
MZTaskInfo ti = infoFn.apply(id);
|
||||
return ti != null ? ti.duration() : Duration.ZERO;
|
||||
}
|
||||
|
||||
@Override public Throwable getCause() {
|
||||
MZTaskInfo ti = infoFn.apply(id);
|
||||
return ti != null ? ti.error() : null;
|
||||
}
|
||||
}
|
||||
|
||||
private final class BukkitHandle implements MZTaskHandle {
|
||||
private final UUID id;
|
||||
private final BukkitTask task;
|
||||
private final Function<UUID, MZTaskInfo> infoFn;
|
||||
private final boolean repeating;
|
||||
|
||||
BukkitHandle(UUID id, BukkitTask task, Function<UUID, MZTaskInfo> infoFn) {
|
||||
this(id, task, infoFn, false);
|
||||
}
|
||||
BukkitHandle(UUID id, BukkitTask task, Function<UUID, MZTaskInfo> infoFn, boolean repeating) {
|
||||
this.id = id; this.task = task; this.infoFn = infoFn; this.repeating = repeating;
|
||||
}
|
||||
|
||||
@Override public UUID getId() { return id; }
|
||||
|
||||
@Override public boolean cancel() {
|
||||
try {
|
||||
task.cancel();
|
||||
return true;
|
||||
} finally {
|
||||
markCancelled(id);
|
||||
}
|
||||
}
|
||||
|
||||
@Override public boolean isDone() {
|
||||
MZTaskInfo ti = infoFn.apply(id);
|
||||
if (ti == null) return task.isCancelled();
|
||||
return switch (ti.status()) {
|
||||
case COMPLETED, FAILED, CANCELLED -> true;
|
||||
default -> !repeating && (ti.end() != null);
|
||||
};
|
||||
}
|
||||
|
||||
@Override public boolean isCancelled() {
|
||||
MZTaskInfo ti = infoFn.apply(id);
|
||||
return (ti != null && ti.status() == MZTaskStatus.CANCELLED) || task.isCancelled();
|
||||
}
|
||||
|
||||
@Override public Duration getDuration() {
|
||||
MZTaskInfo ti = infoFn.apply(id);
|
||||
return ti != null ? ti.duration() : Duration.ZERO;
|
||||
}
|
||||
|
||||
@Override public Throwable getCause() {
|
||||
MZTaskInfo ti = infoFn.apply(id);
|
||||
return ti != null ? ti.error() : null;
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,41 @@
|
||||
package fr.molzonas.mzcore.tasks;
|
||||
|
||||
import org.bukkit.Bukkit;
|
||||
import org.bukkit.plugin.Plugin;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.Collection;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.CompletionException;
|
||||
import java.util.concurrent.CompletionStage;
|
||||
import java.util.concurrent.ExecutionException;
|
||||
import java.util.function.Consumer;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
public interface MZTaskSchedulerInterface extends AutoCloseable {
|
||||
MZTaskHandle runSync(Runnable task);
|
||||
MZTaskHandle runAsync(Runnable task);
|
||||
MZTaskHandle scheduleSync(Runnable task, long delay);
|
||||
MZTaskHandle scheduleAsync(Runnable task, Duration delay);
|
||||
MZTaskHandle repeatSync(Runnable task, long initialDelay, long period);
|
||||
MZTaskHandle repeatAsync(Runnable task, Duration delay, Duration period);
|
||||
<T> CompletableFuture<T> supplyAsync(Supplier<T> supplier);
|
||||
<T> void thenSync(CompletionStage<T> completionStage, Consumer<T> consumer);
|
||||
default <T> void thenSync(CompletionStage<T> stage, Consumer<T> ok, Consumer<Throwable> ko) {
|
||||
stage.whenComplete((val, err) ->
|
||||
Bukkit.getScheduler().runTask(getPlugin(), () -> {
|
||||
if (err != null) ko.accept(unwrapCompletionException(err));
|
||||
else ok.accept(val);
|
||||
})
|
||||
);
|
||||
}
|
||||
default Throwable unwrapCompletionException(Throwable e) {
|
||||
if (e instanceof CompletionException ce && ce.getCause() != null) return ce.getCause();
|
||||
if (e instanceof ExecutionException ee && ee.getCause() != null) return ee.getCause();
|
||||
return e;
|
||||
}
|
||||
<T> CompletionStage<T> withTimeout(CompletionStage<T> stage, Duration timeout);
|
||||
Collection<MZTaskInfo> snapshot();
|
||||
Plugin getPlugin();
|
||||
@Override void close();
|
||||
}
|
||||
@ -0,0 +1,5 @@
|
||||
package fr.molzonas.mzcore.tasks.enums;
|
||||
|
||||
public enum MZTaskKind {
|
||||
RUN, SCHEDULE, REPEAT, SUPPLY;
|
||||
}
|
||||
@ -0,0 +1,5 @@
|
||||
package fr.molzonas.mzcore.tasks.enums;
|
||||
|
||||
public enum MZTaskStatus {
|
||||
PENDING, RUNNING, WAITING, COMPLETED, CANCELLED, FAILED, UNKNOWN;
|
||||
}
|
||||
@ -0,0 +1,5 @@
|
||||
package fr.molzonas.mzcore.tasks.enums;
|
||||
|
||||
public enum MZTaskType {
|
||||
SYNC, ASYNC;
|
||||
}
|
||||
Loading…
x
Reference in New Issue
Block a user