Refactor HMBIRD scheduler queues & threading

Replace raw concurrent deques/queues with priority queues ordered by scheduled/enqueue time and id, adding synchronization around queue mutations to ensure correct timed ordering. Improve HMBIRDLatencySampler to track actual sample count and return trimmed, sorted snapshots. Expand HMBIRDMetrics with new counters (coalesced notifications, intermediate requeues, soft-watermark activations, autoscale events), rename latency samplers, and broaden the snapshot API to include worker/queue/soft-reject/autoscale fields. Major rewrite of HMBIRDSchedulerThreadPool: introduce dynamic worker lifecycle and selection, separate timer handling that enqueues due ticks, strengthened cancellation/coalescing logic for task notifications, soft-watermark backpressure with autoscaling, safer shutdown/start semantics, and various concurrency/locking improvements (task enqueue tracking, defer/requeue handling). Also adjust local queue steal/offer semantics, polling logic, and logging. Update localization strings to reflect soft-watermark wording.

Author: Bacteriawa

mint-server/src/main/java/dev/bacteriawa/mint/scheduler/HMBIRDDispatchQueue.java

@@ -1,20 +1,33 @@
 package dev.bacteriawa.mint.scheduler;
 
-import java.util.concurrent.ConcurrentLinkedQueue;
+import ca.spottedleaf.concurrentutil.util.TimeUtil;
+import java.util.Comparator;
+import java.util.PriorityQueue;
 import java.util.concurrent.atomic.LongAdder;
 
 public final class HMBIRDDispatchQueue {
 
     // Global queues per deadline level (higher levels are polled first).
-    private final ConcurrentLinkedQueue<HMBIRDTask>[] byDeadlineLevel;
+    private final PriorityQueue<HMBIRDTask>[] byDeadlineLevel;
     // Approximate size across all levels.
     private final LongAdder size = new LongAdder();
+    private static final Comparator<HMBIRDTask> TASK_ORDER = (a, b) -> {
+        int cmp = TimeUtil.compareTimes(a.scheduledNanos(), b.scheduledNanos());
+        if (cmp != 0) {
+            return cmp;
+        }
+        cmp = TimeUtil.compareTimes(a.enqueueNanos(), b.enqueueNanos());
+        if (cmp != 0) {
+            return cmp;
+        }
+        return Long.compare(a.id(), b.id());
+    };
 
     @SuppressWarnings("unchecked")
     public HMBIRDDispatchQueue() {
-        this.byDeadlineLevel = (ConcurrentLinkedQueue<HMBIRDTask>[])new ConcurrentLinkedQueue[4];
+        this.byDeadlineLevel = (PriorityQueue<HMBIRDTask>[])new PriorityQueue[4];
         for (int i = 0; i < this.byDeadlineLevel.length; ++i) {
-            this.byDeadlineLevel[i] = new ConcurrentLinkedQueue<>();
+            this.byDeadlineLevel[i] = new PriorityQueue<>(TASK_ORDER);
         }
     }
 
@@ -26,44 +39,36 @@ public void offer(final HMBIRDTask task) {
         if (!task.tryMarkQueued()) {
             return;
         }
-        this.byDeadlineLevel[task.prop().deadlineLevel()].offer(task);
-        this.size.increment();
+        final PriorityQueue<HMBIRDTask> queue = this.byDeadlineLevel[task.prop().deadlineLevel()];
+        synchronized (queue) {
+            queue.offer(task);
+            this.size.increment();
+        }
     }
 
     public HMBIRDTask pollAny() {
-        // Prefer higher deadline levels.
-        for (int level = this.byDeadlineLevel.length - 1; level >= 0; --level) {
-            final ConcurrentLinkedQueue<HMBIRDTask> queue = this.byDeadlineLevel[level];
-            final HMBIRDTask task = queue.poll();
-            if (task == null) {
-                continue;
-            }
-            this.size.decrement();
-            if (task.tryMarkDispatching()) {
-                return task;
-            }
-        }
-        return null;
+        return this.pollReady(Long.MAX_VALUE);
     }
 
     public HMBIRDTask pollReady(final long nowNanos) {
         // Only return tasks whose scheduled time has passed.
         for (int level = this.byDeadlineLevel.length - 1; level >= 0; --level) {
-            final ConcurrentLinkedQueue<HMBIRDTask> queue = this.byDeadlineLevel[level];
-            final HMBIRDTask task = queue.peek();
-            if (task == null) {
-                continue;
-            }
-            if (task.scheduledNanos() > nowNanos) {
-                continue;
-            }
-            final HMBIRDTask polled = queue.poll();
-            if (polled == null) {
-                continue;
-            }
-            this.size.decrement();
-            if (polled.tryMarkDispatching()) {
-                return polled;
+            final PriorityQueue<HMBIRDTask> queue = this.byDeadlineLevel[level];
+            synchronized (queue) {
+                while (true) {
+                    final HMBIRDTask task = queue.peek();
+                    if (task == null) {
+                        break;
+                    }
+                    if (TimeUtil.compareTimes(task.scheduledNanos(), nowNanos) > 0) {
+                        break;
+                    }
+                    final HMBIRDTask polled = queue.poll();
+                    this.size.decrement();
+                    if (polled.tryMarkDispatching()) {
+                        return polled;
+                    }
+                }
             }
         }
         return null;

mint-server/src/main/java/dev/bacteriawa/mint/scheduler/HMBIRDLatencySampler.java

@@ -8,6 +8,7 @@ public final class HMBIRDLatencySampler {
     // Ring buffer of recent samples (nanoseconds).
     private final long[] samples;
     private final AtomicInteger writeIndex = new AtomicInteger();
+    private final AtomicInteger sampleCount = new AtomicInteger();
 
     public HMBIRDLatencySampler(final int capacity) {
         if (capacity <= 0) {
@@ -20,10 +21,12 @@ public void record(final long value) {
         // Overwrites oldest samples in a ring.
         final int idx = Math.floorMod(this.writeIndex.getAndIncrement(), this.samples.length);
         this.samples[idx] = value;
+        this.sampleCount.updateAndGet(count -> Math.min(this.samples.length, count + 1));
     }
 
     public Snapshot snapshot() {
-        final long[] copy = Arrays.copyOf(this.samples, this.samples.length);
+        final int count = Math.min(this.sampleCount.get(), this.samples.length);
+        final long[] copy = Arrays.copyOf(this.samples, count);
         Arrays.sort(copy);
         return new Snapshot(copy);
     }

mint-server/src/main/java/dev/bacteriawa/mint/scheduler/HMBIRDLocalQueue.java

@@ -1,62 +1,109 @@
 package dev.bacteriawa.mint.scheduler;
 
-import java.util.concurrent.ConcurrentLinkedDeque;
+import ca.spottedleaf.concurrentutil.util.TimeUtil;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+import java.util.PriorityQueue;
 import java.util.concurrent.atomic.LongAdder;
 
 public final class HMBIRDLocalQueue {
 
-    // Per-worker queues, LIFO for locality; higher levels are polled first.
-    private final ConcurrentLinkedDeque<HMBIRDTask>[] byDeadlineLevel;
+    // Per-worker queues; higher levels are polled first, then older due/enqueued tasks.
+    private final PriorityQueue<HMBIRDTask>[] byDeadlineLevel;
     // Approximate size across all levels.
     private final LongAdder size = new LongAdder();
+    private static final Comparator<HMBIRDTask> TASK_ORDER = (a, b) -> {
+        int cmp = TimeUtil.compareTimes(a.scheduledNanos(), b.scheduledNanos());
+        if (cmp != 0) {
+            return cmp;
+        }
+        cmp = TimeUtil.compareTimes(a.enqueueNanos(), b.enqueueNanos());
+        if (cmp != 0) {
+            return cmp;
+        }
+        return Long.compare(a.id(), b.id());
+    };
 
     @SuppressWarnings("unchecked")
     public HMBIRDLocalQueue() {
-        this.byDeadlineLevel = (ConcurrentLinkedDeque<HMBIRDTask>[])new ConcurrentLinkedDeque[4];
+        this.byDeadlineLevel = (PriorityQueue<HMBIRDTask>[])new PriorityQueue[4];
         for (int i = 0; i < this.byDeadlineLevel.length; ++i) {
-            this.byDeadlineLevel[i] = new ConcurrentLinkedDeque<>();
+            this.byDeadlineLevel[i] = new PriorityQueue<>(TASK_ORDER);
         }
     }
 
     public long size() {
         return this.size.sum();
     }
 
+    public boolean isEmpty() {
+        return this.size() <= 0L;
+    }
+
     public void offer(final HMBIRDTask task) {
         if (!task.tryMarkQueued()) {
             return;
         }
-        this.byDeadlineLevel[task.prop().deadlineLevel()].offerFirst(task);
-        this.size.increment();
+        final PriorityQueue<HMBIRDTask> queue = this.byDeadlineLevel[task.prop().deadlineLevel()];
+        synchronized (queue) {
+            queue.offer(task);
+            this.size.increment();
+        }
     }
 
     public HMBIRDTask pollAny() {
         // Prefer higher deadline levels.
         for (int level = this.byDeadlineLevel.length - 1; level >= 0; --level) {
-            final ConcurrentLinkedDeque<HMBIRDTask> queue = this.byDeadlineLevel[level];
-            final HMBIRDTask task = queue.pollFirst();
-            if (task == null) {
-                continue;
-            }
-            this.size.decrement();
-            if (task.tryMarkDispatching()) {
-                return task;
+            final PriorityQueue<HMBIRDTask> queue = this.byDeadlineLevel[level];
+            synchronized (queue) {
+                while (true) {
+                    final HMBIRDTask task = queue.poll();
+                    if (task == null) {
+                        break;
+                    }
+                    this.size.decrement();
+                    if (task.tryMarkDispatching()) {
+                        return task;
+                    }
+                }
             }
         }
         return null;
     }
 
     public HMBIRDTask stealAny() {
-        // Steal from tail to reduce contention and preserve locality.
+        // Steal the oldest stealable task at the highest available deadline level.
         for (int level = this.byDeadlineLevel.length - 1; level >= 0; --level) {
-            final ConcurrentLinkedDeque<HMBIRDTask> queue = this.byDeadlineLevel[level];
-            final HMBIRDTask task = queue.pollLast();
-            if (task == null) {
-                continue;
-            }
-            this.size.decrement();
-            if (task.prop().allowSteal() && task.tryMarkDispatching()) {
-                return task;
+            final PriorityQueue<HMBIRDTask> queue = this.byDeadlineLevel[level];
+            synchronized (queue) {
+                List<HMBIRDTask> skipped = null;
+                try {
+                    while (true) {
+                        final HMBIRDTask task = queue.poll();
+                        if (task == null) {
+                            break;
+                        }
+                        this.size.decrement();
+                        if (!task.prop().allowSteal()) {
+                            if (skipped == null) {
+                                skipped = new ArrayList<>();
+                            }
+                            skipped.add(task);
+                            continue;
+                        }
+                        if (task.tryMarkDispatching()) {
+                            return task;
+                        }
+                    }
+                } finally {
+                    if (skipped != null) {
+                        for (final HMBIRDTask task : skipped) {
+                            queue.offer(task);
+                            this.size.increment();
+                        }
+                    }
+                }
             }
         }
         return null;