Task scheduler design

1import java.time.*;2import java.util.*;3import java.util.concurrent.*;4import java.util.concurrent.atomic.AtomicLong;5 6public final class TaskScheduler implements AutoCloseable {7    public enum Status { SCHEDULED, RUNNING, DONE, FAILED, CANCELLED }8 9    private final class Task implements Comparable<Task> {10        final long id = ids.incrementAndGet();11        final Runnable action;12        final Duration period;        // null for one-shot13        Instant nextRun;14        volatile Status status = Status.SCHEDULED;15        Task(Runnable action, Instant nextRun, Duration period) { this.action = action; this.nextRun = nextRun; this.period = period; }16        public int compareTo(Task o) { return nextRun.compareTo(o.nextRun); }17    }18 19    private final AtomicLong ids = new AtomicLong();20    private final PriorityQueue<Task> queue = new PriorityQueue<>();21    private final Map<Long, Task> byId = new ConcurrentHashMap<>();22    private final ExecutorService workers;23    private final Clock clock;24    private final Thread dispatcher;25    private volatile boolean running = true;26 27    public TaskScheduler(int workerCount, Clock clock) {28        this.workers = Executors.newFixedThreadPool(workerCount);29        this.clock = clock;30        this.dispatcher = new Thread(this::dispatchLoop, "scheduler-dispatcher");31        dispatcher.start();32    }33 34    public long schedule(Runnable action, Duration delay) { return add(new Task(action, clock.instant().plus(delay), null)); }35    public long scheduleAtFixedRate(Runnable action, Duration initialDelay, Duration period) {36        return add(new Task(action, clock.instant().plus(initialDelay), period));37    }38 39    public boolean cancel(long id) {40        Task t = byId.get(id);41        if (t == null) return false;42        synchronized (this) { queue.remove(t); notifyAll(); }43        t.status = Status.CANCELLED;44        return true;45    }46 47    public Status status(long id) { return byId.get(id).status; }48 49    private synchronized long add(Task t) {50        byId.put(t.id, t);51        queue.add(t);52        notifyAll();                                   // wake dispatcher: maybe an earlier task53        return t.id;54    }55 56    private void dispatchLoop() {57        while (running) {58            Task due;59            synchronized (this) {60                while (running && (queue.isEmpty() || queue.peek().nextRun.isAfter(clock.instant()))) {61                    long waitMs = queue.isEmpty() ? 1000 : Math.max(1, Duration.between(clock.instant(), queue.peek().nextRun).toMillis());62                    try { wait(waitMs); } catch (InterruptedException e) { return; }63                }64                if (!running) return;65                due = queue.poll();66            }67            workers.submit(() -> run(due));68        }69    }70 71    private void run(Task t) {72        if (t.status == Status.CANCELLED) return;73        t.status = Status.RUNNING;74        try {75            t.action.run();76            t.status = t.period == null ? Status.DONE : Status.SCHEDULED;77        } catch (RuntimeException e) {78            t.status = Status.FAILED;79        }80        if (t.period != null && t.status == Status.SCHEDULED) {81            t.nextRun = t.nextRun.plus(t.period);     // fixed rate82            add(t);83        }84    }85 86    @Override87    public void close() throws InterruptedException {88        running = false;89        synchronized (this) { notifyAll(); }90        dispatcher.join();91        workers.shutdown();92        workers.awaitTermination(10, TimeUnit.SECONDS);93    }94}