From 691e9496919590f93b9016c44ab8296d41e88566 Mon Sep 17 00:00:00 2001 From: Disy Date: Tue, 6 Oct 2026 16:44:11 +0800 Subject: [PATCH] feat: add bounded cancellable payload work queue --- build.gradle | 21 ++++- .../core/compat/PayloadWorkQueue.java | 92 +++++++++++++++++++ tests/PayloadWorkQueueRegression.java | 81 ++++++++++++++++ 3 files changed, 192 insertions(+), 2 deletions(-) create mode 100644 src/main/java/dev/modzuozhi/core/compat/PayloadWorkQueue.java create mode 100644 tests/PayloadWorkQueueRegression.java diff --git a/build.gradle b/build.gradle index 84aed20..32efb72 100644 --- a/build.gradle +++ b/build.gradle @@ -81,6 +81,22 @@ if (providers.gradleProperty('interactionRegression').isPresent()) { sourceSets.main.java.srcDir('tests/integration') sourceSets.main.java.exclude('**/ChunkHandoffRegression.java') } +if (providers.gradleProperty('ownershipRegression').isPresent()) { + if (providers.gradleProperty('interactionRegression').isPresent()) { + throw new GradleException('Run ownershipRegression and interactionRegression separately.') + } + sourceSets.main.java.srcDir('tests/ownership') + dependencies { runtimeOnly files(providers.gradleProperty('ccTestJar').get()) } + tasks.matching { it.name == 'runServer' }.configureEach { + doLast { + def output = file('runs/server/logs/latest.log').getText('UTF-8') + ['OWNERSHIP REGRESSION PASS', 'OWNERSHIP EDGE PASS', 'OWNERSHIP CALLBACK PASS', 'OWNERSHIP RESUME PASS'].each { marker -> + if (!output.contains(marker)) throw new GradleException("Missing runtime test result: ${marker}") + } + if (output.contains('Encountered an unexpected exception')) throw new GradleException('Runtime regression server crashed.') + } + } +} tasks.withType(ProcessResources).configureEach { var replaceProperties = [ @@ -121,11 +137,12 @@ tasks.register('compileConcurrencyRegression', JavaCompile) { 'src/main/java/dev/modzuozhi/core/compat/TaskBarrier.java', 'src/main/java/dev/modzuozhi/core/compat/ManagedFutureAwait.java', 'src/main/java/dev/modzuozhi/core/compat/PauseOwnership.java', 'tests/PauseOwnershipRegression.java', - 'src/main/java/dev/modzuozhi/core/compat/TickProgress.java', 'tests/TickProgressRegression.java') + 'src/main/java/dev/modzuozhi/core/compat/TickProgress.java', 'tests/TickProgressRegression.java', + 'src/main/java/dev/modzuozhi/core/compat/PayloadWorkQueue.java', 'tests/PayloadWorkQueueRegression.java') classpath = files() destinationDirectory = regressionClasses } -['TaskBarrierRegression', 'ManagedFutureAwaitRegression', 'PauseOwnershipRegression', 'TickProgressRegression'].each { regressionMain -> +['TaskBarrierRegression', 'ManagedFutureAwaitRegression', 'PauseOwnershipRegression', 'TickProgressRegression', 'PayloadWorkQueueRegression'].each { regressionMain -> def regressionTask = tasks.register("run${regressionMain}", JavaExec) { dependsOn 'compileConcurrencyRegression' classpath = files(regressionClasses) diff --git a/src/main/java/dev/modzuozhi/core/compat/PayloadWorkQueue.java b/src/main/java/dev/modzuozhi/core/compat/PayloadWorkQueue.java new file mode 100644 index 0000000..3df67a2 --- /dev/null +++ b/src/main/java/dev/modzuozhi/core/compat/PayloadWorkQueue.java @@ -0,0 +1,92 @@ +package dev.modzuozhi.core.compat; + +import java.util.*; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.RejectedExecutionException; +import java.util.function.LongSupplier; +import java.util.function.Supplier; + +/** Bounded FIFO. Queue locks never encompass handlers or future completion callbacks. */ +public final class PayloadWorkQueue { + private final LongSupplier clock; + private final LinkedHashSet> pending = new LinkedHashSet<>(); + private final Map counts = new HashMap<>(); + private Throwable closed; + private boolean draining; + + public PayloadWorkQueue(LongSupplier nanoTime) { clock = Objects.requireNonNull(nanoTime); } + + public CompletableFuture submit(K connection, Supplier action) { + Objects.requireNonNull(connection); + Objects.requireNonNull(action); + Entry entry = new Entry<>(connection, action); + Throwable rejected; + synchronized (this) { + rejected = closed; + if (rejected == null && (pending.size() >= 4096 || counts.getOrDefault(connection, 0) >= 256)) + rejected = new RejectedExecutionException("Payload queue capacity exceeded"); + if (rejected == null) { + pending.add(entry); + counts.merge(connection, 1, Integer::sum); + } + } + if (rejected != null) entry.result.completeExceptionally(rejected); + else entry.result.whenComplete((value, error) -> { + if (entry.result.isCancelled()) synchronized (this) { remove(entry); } + }); + return entry.result; + } + + private void remove(Entry entry) { + if (pending.remove(entry)) counts.compute(entry.connection, (key, count) -> count == 1 ? null : count - 1); + } + + /** Single consumer; recursive/concurrent drains are no-ops. Already running work is never interrupted. */ + public int drainBatch() { + synchronized (this) { + if (draining || closed != null) return 0; + draining = true; + } + int ran = 0; + long start = clock.getAsLong(); + try { + while (ran < 64 && clock.getAsLong() - start < 2_000_000L) { + Entry entry; + synchronized (this) { + if (closed != null || pending.isEmpty()) break; + entry = pending.iterator().next(); + remove(entry); + } + entry.run(); + ran++; + } + return ran; + } finally { synchronized (this) { draining = false; } } + } + + public void close(Throwable reason) { + Objects.requireNonNull(reason); + List> abandoned; + synchronized (this) { + if (closed != null) return; + closed = reason; + abandoned = new ArrayList<>(pending); + pending.clear(); counts.clear(); + } + abandoned.forEach(entry -> entry.result.completeExceptionally(reason)); + } + + public synchronized int size() { return pending.size(); } + + private final class Entry { + final K connection; + final Supplier action; + final CompletableFuture result = new CompletableFuture<>(); + Entry(K connection, Supplier action) { this.connection = connection; this.action = action; } + void run() { + if (result.isCancelled()) return; + try { result.complete(action.get()); } + catch (Throwable error) { result.completeExceptionally(error); } + } + } +} diff --git a/tests/PayloadWorkQueueRegression.java b/tests/PayloadWorkQueueRegression.java new file mode 100644 index 0000000..17a4ccb --- /dev/null +++ b/tests/PayloadWorkQueueRegression.java @@ -0,0 +1,81 @@ +import dev.modzuozhi.core.compat.PayloadWorkQueue; +import java.util.*; +import java.util.concurrent.*; +import java.util.concurrent.atomic.*; + +public class PayloadWorkQueueRegression { + static void check(boolean ok, String why) { if (!ok) throw new AssertionError(why); } + public static void main(String[] args) throws Exception { + fifoAndExactlyOnce(); capacityAndCancellation(); batchCountAndTime(); + closeAndReentrantCompletion(); concurrentProducers(); + System.out.println("PAYLOAD QUEUE PASS"); + } + static void fifoAndExactlyOnce() { + var q = new PayloadWorkQueue(() -> 0L); + var seen = new ArrayList(); + var futures = new ArrayList>(); + for (int i=0;i<256;i++) { int n=i; futures.add(q.submit("a", () -> { seen.add(n); return n; })); } + while(q.size()>0) q.drainBatch(); + check(seen.size()==256, "exactly once"); + for(int i=0;i<256;i++) check(seen.get(i)==i && futures.get(i).join()==i, "FIFO/results"); + check(q.drainBatch()==0, "no replay"); + } + static void capacityAndCancellation() { + var q = new PayloadWorkQueue(() -> 0L); + var fs = new ArrayList>(); + for(int c=0;c<16;c++) for(int i=0;i<256;i++) fs.add(q.submit(c, () -> 1)); + check(q.size()==4096, "server capacity"); + check(q.submit(99, () -> 2).isCompletedExceptionally(), "reject server overflow"); + fs.getFirst().cancel(false); + check(q.size()==4095, "cancel returns capacity immediately"); + check(!q.submit(0, () -> 3).isDone(), "cancel frees connection slot"); + while(q.size()>0) q.drainBatch(); + for(int i=0;i<256;i++) q.submit(7, () -> 1); + check(q.submit(7, () -> 9).isCompletedExceptionally(), "reject connection overflow"); + check(!q.submit(8, () -> 1).isDone(), "other connection still allowed"); + q.close(new IllegalStateException("stop")); + } + static void batchCountAndTime() { + var clock = new AtomicLong(); var q = new PayloadWorkQueue(clock::get); + for(int i=0;i<65;i++) q.submit("a", () -> 1); + check(q.drainBatch()==64 && q.size()==1, "64 budget"); q.drainBatch(); + var done = new AtomicInteger(); + q.submit("a", () -> { clock.addAndGet(2_000_000); return done.incrementAndGet(); }); + q.submit("a", done::incrementAndGet); + check(q.drainBatch()==1 && done.get()==1 && q.size()==1, "2ms boundary"); + q.drainBatch(); + var slow = q.submit("a", () -> { clock.addAndGet(100_000_000); return 42; }); + check(q.drainBatch()==1 && slow.join()==42, "slow task not interrupted"); + } + static void closeAndReentrantCompletion() { + var q = new PayloadWorkQueue(() -> 0L); + var count = new AtomicInteger(); var f=q.submit("a", count::incrementAndGet); + f.thenRun(() -> q.submit("a", count::incrementAndGet)); q.drainBatch(); + check(count.get()==2, "completion may enqueue"); + var boom = new IllegalArgumentException("handler failure"); + var bad=q.submit("a", () -> { throw boom; }); var good=q.submit("a", () -> 42); + q.drainBatch(); check(bad.isCompletedExceptionally() && good.join()==42, "isolate exceptions"); + var pending=q.submit("a", count::incrementAndGet); + q.close(boom); check(pending.isCompletedExceptionally() && q.size()==0, "close settles pending"); + check(q.submit("a", count::incrementAndGet).isCompletedExceptionally(), "close terminal"); + q.drainBatch(); check(count.get()==2, "closed work never executes"); + var q2=new PayloadWorkQueue(() -> 0L); + var cancel=q2.submit("a", count::incrementAndGet); cancel.cancel(false); q2.drainBatch(); + check(count.get()==2, "cancelled action skipped"); + } + static void concurrentProducers() throws Exception { + var q = new PayloadWorkQueue(() -> 0L); + var executed=new AtomicInteger(); var futures=new ConcurrentLinkedQueue>(); + var pool=Executors.newFixedThreadPool(8); + try { + var jobs=new ArrayList>(); + for(int c=0;c<8;c++) { int key=c; jobs.add(pool.submit(() -> { + for(int i=0;i<250;i++) { var f=q.submit(key, executed::incrementAndGet); futures.add(f); if(i%2==0) f.cancel(false); } + })); } + for(var job:jobs) job.get(5,TimeUnit.SECONDS); + while(q.size()>0) q.drainBatch(); + check(executed.get()==1000 && q.size()==0, "concurrent cancel/submit accounting"); + check(futures.stream().allMatch(CompletableFuture::isDone), "no pending futures"); + } finally { pool.shutdownNow(); } + } +}