Update to 0.0.4-SNAPSHOT

- Add AreaDependentQueue
  AreaDependentQueue is capable of dynamically ordering tasks
  scheduled (possibly) onto overlapping x/y coordinates. For
  a given coordinate, only one task at most is being executed.
- Add stream id (long) to PrioritisedExecutor
  This will be later used by thread pools that may perform different
  scheduling depending on the stream id. For example, it may be
  possible that tasks with a lower stream id are executed before
  other tasks, or that tasks with different stream ids are shared
  equally. At this stage, I've not yet decided.
- Add BalancedPrioritisedThreadPool
  This thread pool will share time equally between groups but will
  not share time equally between queues within the groups. The task
  execution order is determined by priority then by suborder. This
  results is more fair task execution as determined by queue size
  (i.e more tasks in one queue should generally receive more time).
- Rewrite spinwait logic in PrioritisedQueueExecutorThread
  1. Do not yield and park immediately upon emptying the queue
     Instead, set the parked field. This allows wakeup to occur if
     a task is queued while short parking.
  2. Only spinwait if configured to
- Fix PrioritisedQueueExecutorThread#close(wait=true, ...) not waiting
  Need to only break when the thread dies. Forgot to invert the alive
  check.
- Add COWArrayList
  The underlying array is not Object[] but is actually of the
  element type. The list also works only through reference equality
  as well. The array is retrievable for faster iteration.
- Change coordinates from x/z to x/y coordinates in ReentrantAreaLock
- Use AtomicLong for size is ConcurrentLong2ReferenceChainedHashTable
  We really don't see any benefit from using LongAdder since on
  additions we need to sum over the entire adder (defeating the
  purpose of splitting the additions for parallelism). There should
  be a minor improvement in put/remove performance, even in contended
  scenarios.
- Remove ConcurrentUtil#rethrow
  Use ThrowUtil#throwUnchecked
- Remove PrioritisedThreadPool
  Use BalancedPrioritisedThreadPool, or in the future even
  StreamOrderedThreadPool

Author: Spottedleaf

gradle/wrapper/gradle-wrapper.jar

@@ -1,6 +1,6 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-8.11.1-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-8.14.3-bin.zip
 networkTimeout=10000
 validateDistributionUrl=true
 zipStoreBase=GRADLE_USER_HOME

gradle/wrapper/gradle-wrapper.properties

@@ -1,8 +1,8 @@
 package ca.spottedleaf.concurrentutil.collection;
 
 import ca.spottedleaf.concurrentutil.util.ConcurrentUtil;
+import ca.spottedleaf.concurrentutil.util.ThrowUtil;
 import ca.spottedleaf.concurrentutil.util.Validate;
-
 import java.lang.invoke.VarHandle;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -1134,7 +1134,7 @@ protected final E removeHead() {
      * @return The total number of elements drained.
      */
     public int drain(final Consumer<E> consumer) {
-        return this.drain(consumer, false, ConcurrentUtil::rethrow);
+        return this.drain(consumer, false, ThrowUtil::throwUnchecked);
     }
 
     /**
@@ -1154,7 +1154,7 @@ public int drain(final Consumer<E> consumer) {
      * @return The total number of elements drained.
      */
     public int drain(final Consumer<E> consumer, final boolean preventAdds) {
-        return this.drain(consumer, preventAdds, ConcurrentUtil::rethrow);
+        return this.drain(consumer, preventAdds, ThrowUtil::throwUnchecked);
     }
 
     /**

gradlew

@@ -92,14 +92,16 @@ public interface PrioritisedExecutor {
      * @param task The task to run.
      * @param priority The priority for the task.
      * @param subOrder The task's suborder.
+     * @param stream The task's stream id.
      *
      * @throws IllegalStateException If this executor has shutdown.
      * @throws NullPointerException If the task is null
      * @throws IllegalArgumentException If the priority is invalid.
      * @return {@code null} if the current thread immediately executed the task, else returns the prioritised task
      *         associated with the parameter
      */
-    public PrioritisedTask queueTask(final Runnable task, final Priority priority, final long subOrder);
+    public PrioritisedTask queueTask(final Runnable task, final Priority priority, final long subOrder,
+                                     final long stream);
 
     /**
      * Creates, but does not queue or execute, a task at {@link Priority#NORMAL} priority.
@@ -130,13 +132,15 @@ public interface PrioritisedExecutor {
      * @param task The task to run.
      * @param priority The priority for the task.
      * @param subOrder The task's suborder.
+     * @param stream The task's stream.
      *
      * @throws NullPointerException If the task is null
      * @throws IllegalArgumentException If the priority is invalid.
      * @return {@code null} if the current thread immediately executed the task, else returns the prioritised task
      *         associated with the parameter
      */
-    public PrioritisedTask createTask(final Runnable task, final Priority priority, final long subOrder);
+    public PrioritisedTask createTask(final Runnable task, final Priority priority, final long subOrder,
+                                      final long stream);
 
     public static interface PrioritisedTask extends Cancellable {
 
@@ -220,7 +224,7 @@ public static interface PrioritisedTask extends Cancellable {
         public boolean lowerPriority(final Priority priority);
 
         /**
-         * Returns the suborder id associated with this task.
+         * Returns the suborder id associated with this task, or 0 if completing.
          * @return The suborder id associated with this task.
          */
         public long getSubOrder();
@@ -232,7 +236,7 @@ public static interface PrioritisedTask extends Cancellable {
          * @param subOrder Specified new sub order.
          *
          * @return {@code true} if successful, {@code false} if this task is completing or has completed or the queue
-         *         this task was scheduled on was shutdown, or if the current suborder is the same as the new sub order.
+         *         this task was scheduled on was shutdown, or if the current suborder is the same as the new suborder.
          */
         public boolean setSubOrder(final long subOrder);
 
@@ -257,15 +261,54 @@ public static interface PrioritisedTask extends Cancellable {
         public boolean lowerSubOrder(final long subOrder);
 
         /**
-         * Sets the priority and suborder id associated with this task. Ths function has no effect when this task
+         * Returns the stream id associated with this task, or 0 if completing.
+         * @return The stream id associated with this task.
+         */
+        public long getStream();
+
+        /**
+         * Sets the stream id associated with this task. Ths function has no effect when this task
          * is completing or is completed.
          *
-         * @param priority Priority specified
+         * @param stream Specified new stream.
+         *
+         * @return {@code true} if successful, {@code false} if this task is completing or has completed or the queue
+         *         this task was scheduled on was shutdown, or if the current stream is the same as the new stream.
+         */
+        public boolean setStream(final long stream);
+
+        /**
+         * Sets the priority, suborder id, and stream id associated with this task. Ths function has no effect when
+         * this task is completing or is completed.
+         *
+         * @param priority Specified new priority.
          * @param subOrder Specified new sub order.
+         * @param stream Specified new stream.
          * @return {@code true} if successful, {@code false} if this task is completing or has completed or the queue
-         *         this task was scheduled on was shutdown, or if the current priority and suborder are the same as
-         *         the parameters.
+         *         this task was scheduled on was shutdown, or if the current priority. suborder, and stream are the same
+         *         as the parameters.
          */
-        public boolean setPriorityAndSubOrder(final Priority priority, final long subOrder);
+        public boolean setPrioritySubOrderStream(final Priority priority, final long subOrder,
+                                                 final long stream);
+
+        /**
+         * Atomically retrieves the priority, suborder, and stream for this task. Returns {@code null} if the task
+         * is completing or cancelled.
+         * @return The current priority state, or {@code null} if completing or cancelled.
+         */
+        public PriorityState getPriorityState();
+    }
+
+    public static record PriorityState(Priority priority, long subOrder, long stream) implements Comparable<PriorityState> {
+
+        @Override
+        public int compareTo(final PriorityState other) {
+            final int priorityCompare = this.priority.priority - other.priority.priority;
+            if (priorityCompare != 0) {
+                return priorityCompare;
+            }
+
+            return Long.compare(this.subOrder, other.subOrder);
+        }
     }
 }