From 31a395054e746fe3c887eefb02841ac1b3401224 Mon Sep 17 00:00:00 2001 From: Pavel Nevezhin Date: Wed, 30 Sep 2026 01:03:22 +0300 Subject: [PATCH] sfqd-core: fix refundable queue scaling for release 1.2.0 Keep long cancellation-heavy queues from monopolizing the scheduler lock. Preserve exact refund admission and dispatch behavior while bounding ordinary queue updates logarithmically. --- CHANGELOG.md | 25 +++ README.md | 29 ++-- docs/FORMAL_SPEC.md | 18 ++- docs/THEORY.md | 5 +- docs/TOOLING.md | 2 +- pom.xml | 6 +- sfqd-benchmarks/README.md | 6 +- sfqd-benchmarks/pom.xml | 2 +- .../CancellationCycleBenchmark.java | 6 +- sfqd-core/pom.xml | 2 +- .../io/github/pzhin/sfqd/RefundCosts.java | 143 +++++++++++++++++ .../io/github/pzhin/sfqd/SfqdScheduler.java | 148 ++++++++---------- .../io/github/pzhin/sfqd/RefundCostsTest.java | 44 ++++++ .../pzhin/sfqd/SfqdNumericBoundaryTest.java | 33 ++++ .../pzhin/sfqd/SfqdRefundLongQueueTest.java | 74 +++++++++ sfqd-coverage/pom.xml | 2 +- sfqd-examples/pom.xml | 2 +- sfqd-jcstress/pom.xml | 2 +- tools/verify-publication-topology.sh | 16 +- 19 files changed, 452 insertions(+), 113 deletions(-) create mode 100644 sfqd-core/src/main/java/io/github/pzhin/sfqd/RefundCosts.java create mode 100644 sfqd-core/src/test/java/io/github/pzhin/sfqd/RefundCostsTest.java create mode 100644 sfqd-core/src/test/java/io/github/pzhin/sfqd/SfqdRefundLongQueueTest.java diff --git a/CHANGELOG.md b/CHANGELOG.md index 95b3879..1fdcc0a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,31 @@ All notable changes to this project are documented in this file. +## 1.2.0 - 2026-09-30 + +Removes per-flow queue scans from ordinary refundable admission and queued +cancellation, preventing long queues from multiplying their lock-held work. + +### Changed + +- refund-mode enqueue and cancellation now update an AVL multiset of live + costs in `O(log K)` time, plus head-index maintenance where applicable; +- exact queued tags are materialized on head promotion; dispatch adds + `O(log K)` per selected job to maintain the cost aggregate; +- incremental sum/gcd admission preserves the exact existing refund-closure + domain, including denominator-factor removal after cancellation or dispatch; +- JMH cancellation cycles now expose per-flow queue length as a parameter. + +### Verification and compatibility + +- adds long-queue differential, cost-aggregate, and cancellation-before-rebase + regression coverage; +- preserves public signatures, exact scheduling order, numeric budgets, atomic + rejection, and default charge-reserved accounting; +- canonical rebasing remains transactional and linear in queued jobs plus + registered flows; the scheduler still uses one lock; +- no application-specific throughput or deadline guarantee is introduced. + ## 1.1.0 - 2026-08-30 Adds opt-in virtual-cost refunds for queued cancellation while preserving the diff --git a/README.md b/README.md index 97cd3cb..f7f448c 100644 --- a/README.md +++ b/README.md @@ -10,8 +10,8 @@ weights. ## Status -Version 1.1.0 is the current stable release. The library is published to Maven -Central as `io.github.pzhin:sfqd-core:1.1.0` and uses the Java package +Version 1.2.0 is the current stable release. The library is published to Maven +Central as `io.github.pzhin:sfqd-core:1.2.0` and uses the Java package `io.github.pzhin.sfqd`. The benchmark harness is an executable measurement protocol, not a benchmark @@ -57,7 +57,7 @@ Then add it to a local consumer: io.github.pzhin sfqd-core - 1.1.0 + 1.2.0 ``` @@ -248,7 +248,8 @@ job's cost from future virtual debt. Every later queued job of the same flow is recomputed in enqueue order from the cancelled job's start tag, and the flow's finish history moves to the end of that recomputed suffix. Other flows do not change. The recomputation is one atomic transition under the scheduler's common -lock; callers cannot observe a partial suffix. +lock; callers cannot observe a partial suffix. The implementation represents +this transition lazily and materializes tags when a job becomes the flow head. Refund admission is intentionally narrower. Before accepting a job, enqueue checks that the prospective flow queue has one common exact denominator and a @@ -415,7 +416,7 @@ recorded run is reviewed for the target hardware and workload. ## Complexity Let `R` be registered flows, `Q` queued jobs, `B` backlogged flows, `K` jobs in -the affected per-flow queue or suffix, and `m` the number of jobs returned by +the affected per-flow queue, and `m` the number of jobs returned by one capacity call. | Operation | Expected or worst-case time | @@ -423,18 +424,26 @@ one capacity call. | register or close flow | expected `O(1)` | | charge-reserved enqueue to a backlogged flow | expected `O(1)` | | charge-reserved enqueue that makes a flow backlogged | `O(log B)` | -| refund enqueue to a backlogged flow | `O(K)` | -| refund enqueue that makes a flow backlogged | `O(K + log B)` | +| refund enqueue to a backlogged flow | `O(log K)` | +| refund enqueue that makes a flow backlogged | `O(log K + log B)` | | charge-reserved cancel of a non-head queued job | expected `O(1)` | | charge-reserved cancel of a flow head | `O(log B)` | -| refund cancel without changing the indexed head | `O(K)` | -| refund cancel that changes the indexed head | `O(K + log B)` | -| dispatch `m` jobs | `O(m log B + m)` | +| refund cancel without changing the indexed head | `O(log K)` | +| refund cancel that changes the indexed head | `O(log K + log B)` | +| charge-reserved dispatch `m` jobs | `O(m log B + m)` | +| refund dispatch `m` jobs | `O(m (log B + log K) + m)` | | ordinary completion | expected `O(1)` | | aggregate or per-flow snapshot | expected `O(1)` | | transition to global idle | `O(R)` | | rare exact-tag normalization | `O(Q + R)` time and temporary space | +Refund accounting materializes tags only when a job becomes its flow's head. +An AVL multiset of live queued costs maintains their exact sum and gcd, so +ordinary admission and cancellation never scan the queue. Numeric admission +and scheduling semantics are unchanged; rare canonical rebases still visit +all queued jobs. Both snapshot methods take the scheduler lock but do not +scan jobs. + Retained state is `O(Q + running jobs + R)`. Terminal jobs and payloads are not retained as tombstones. These are algorithmic complexity bounds, not measured performance claims; no throughput or latency claim is made for refund diff --git a/docs/FORMAL_SPEC.md b/docs/FORMAL_SPEC.md index b0ad38b..3de645f 100644 --- a/docs/FORMAL_SPEC.md +++ b/docs/FORMAL_SPEC.md @@ -687,7 +687,7 @@ cancellation. At the successful cancel LP: 6. if the indexed flow head changed, insert the new head into the global priority index only after its recomputed tag is installed. -The complete removal and suffix recomputation are one atomic transition under +The complete removal and logical suffix recomputation are one atomic transition under the scheduler's serialization boundary. No partially recomputed suffix is observable. The refund-closure invariant established at enqueue makes every exact cancellation computation fit the existing budgets, so `cancel` gains no @@ -695,6 +695,22 @@ numeric rejection or exception. Other flows and their tags do not change, intra-flow enqueue order is preserved, every queued `S` remains at least `V`, and no past `dispatchUpTo` result is reconsidered. +An implementation may represent this exact transition lazily. While a flow is +backlogged, every successor starts at its predecessor's finish: `V` cannot +pass the indexed head. Therefore cancellation subtracts `cost(c)/weight` +from `lastFinish`, preserves the head start (including head cancellation), +and removes the cost from the live chain. Head promotion materializes the +next exact tag without consulting the then-current `V`. This is a representation +optimization, not a change to the above transition or admission domain. + +For incremental admission, with total queued cost `C` and gcd `G` of the +individual costs, the increment-denominator lcm is `weight/gcd(weight,G)`. +Combining this with the head denominator yields exactly the `L` in ยง3.2.2; +the final numerator is `baseStart * L + C * L / weight` using exact rational +scaling. The gcd must include individual costs, not only their sum. Deletion +must remove obsolete denominator factors. A rebase still validates every +logical queued tag and the transformed closure before any state is committed. + ### 7.5 `dispatchUpTo(k)` / `dispatch(k)` `k` is an integer in `0..D`; a negative value or `k>D` is an invalid argument. diff --git a/docs/THEORY.md b/docs/THEORY.md index f6b5caf..1be5717 100644 --- a/docs/THEORY.md +++ b/docs/THEORY.md @@ -96,9 +96,10 @@ work while preserving the ordering within each busy period. Cancellation is a library extension rather than a result supplied by the SFQ or SFQ(D) papers. The default policy keeps the tags assigned at admission, so a cancelled queued cost remains virtual debt until global idle. The opt-in refund -policy instead recomputes the later queued suffix of that flow at the +policy instead logically recomputes the later queued suffix of that flow at the cancellation linearization point. It does not undo an earlier dispatch or -rewrite virtual-time history. +rewrite virtual-time history. The implementation may defer tag materialization +until head promotion while preserving this exact transition. Exact refund must not turn cancellation into a fallible cleanup operation. The refund policy therefore narrows admission: before accepting a job, the library diff --git a/docs/TOOLING.md b/docs/TOOLING.md index 6a68f8d..9ff4011 100644 --- a/docs/TOOLING.md +++ b/docs/TOOLING.md @@ -153,7 +153,7 @@ Before running the profile, the release engineer must have: - an unlocked GnuPG agent, or `MAVEN_GPG_PASSPHRASE` supplied through a secure environment mechanism. -Publish the immutable 1.1.0 coordinates from the commit tagged `v1.1.0`: +Publish the immutable 1.2.0 coordinates from the commit tagged `v1.2.0`: ```shell ./mvnw --batch-mode --no-transfer-progress -Prelease clean deploy diff --git a/pom.xml b/pom.xml index 22b2cbe..5f31ca1 100644 --- a/pom.xml +++ b/pom.xml @@ -6,7 +6,7 @@ 4.0.0 io.github.pzhin sfqd-java-parent - 1.1.0 + 1.2.0 pom SFQ(D) Java Parent Build parent for the SFQ(D) Java library. @@ -35,7 +35,7 @@ scm:git:https://github.com/pzhin/sfqd-java.git scm:git:ssh://git@github.com/pzhin/sfqd-java.git https://github.com/pzhin/sfqd-java - v1.1.0 + v1.2.0 @@ -289,7 +289,7 @@ true published - sfqd-java 1.1.0 + sfqd-java 1.2.0 diff --git a/sfqd-benchmarks/README.md b/sfqd-benchmarks/README.md index 19ec483..0082953 100644 --- a/sfqd-benchmarks/README.md +++ b/sfqd-benchmarks/README.md @@ -75,7 +75,11 @@ metric and the normalization. ### Cancellation cycles `CancellationCycleBenchmark` measures `cancel -> enqueue replacement`. The -score belongs to the combined cycle, not cancellation alone. The +score belongs to the combined cycle, not cancellation alone. `queueLength` +selects 2, 250, 2,500, or 20,000 jobs per active flow. For a bounded long-queue +comparison use `-p flowCount=1 -p depth=16 -p scenario=UNIFORM -p target=HEAD` +and vary `queueLength` under both policies; do not combine every large flow +count with every large per-flow queue on a memory-limited machine. The `cancellationAccounting` parameter selects `CHARGE_RESERVED_COST` or `REFUND_CANCELLED_COST`; the same policy dimension is present in isolated head/non-head cancellation latency, terminal idle-reset cancellation, and diff --git a/sfqd-benchmarks/pom.xml b/sfqd-benchmarks/pom.xml index 36a783b..fb80080 100644 --- a/sfqd-benchmarks/pom.xml +++ b/sfqd-benchmarks/pom.xml @@ -6,7 +6,7 @@ io.github.pzhin sfqd-java-parent - 1.1.0 + 1.2.0 sfqd-benchmarks SFQ(D) JMH Harness diff --git a/sfqd-benchmarks/src/main/java/io/github/pzhin/sfqd/benchmarks/CancellationCycleBenchmark.java b/sfqd-benchmarks/src/main/java/io/github/pzhin/sfqd/benchmarks/CancellationCycleBenchmark.java index 3ef95fa..09381b4 100644 --- a/sfqd-benchmarks/src/main/java/io/github/pzhin/sfqd/benchmarks/CancellationCycleBenchmark.java +++ b/sfqd-benchmarks/src/main/java/io/github/pzhin/sfqd/benchmarks/CancellationCycleBenchmark.java @@ -47,6 +47,10 @@ public static class CancellationState { @Param({"1", "8", "64", "256"}) private int depth; + /** Queued jobs per active flow. */ + @Param({"2", "250", "2500", "20000"}) + private int queueLength = 2; + /** Deterministic workload distribution. */ @Param private Scenario scenario; @@ -72,7 +76,7 @@ public static class CancellationState { /** Creates the bounded caller and scheduler queues outside measurements. */ @Setup(Level.Trial) public void setupTrial() { - fixture = new Fixture(flowCount, depth, scenario, 0, 2, cancellationAccounting); + fixture = new Fixture(flowCount, depth, scenario, 0, queueLength, cancellationAccounting); } /** Selects and removes caller-side target bookkeeping outside the timed cycle. */ diff --git a/sfqd-core/pom.xml b/sfqd-core/pom.xml index a112a27..b0ff4f7 100644 --- a/sfqd-core/pom.xml +++ b/sfqd-core/pom.xml @@ -6,7 +6,7 @@ io.github.pzhin sfqd-java-parent - 1.1.0 + 1.2.0 sfqd-core SFQ(D) Core diff --git a/sfqd-core/src/main/java/io/github/pzhin/sfqd/RefundCosts.java b/sfqd-core/src/main/java/io/github/pzhin/sfqd/RefundCosts.java new file mode 100644 index 0000000..e681fe7 --- /dev/null +++ b/sfqd-core/src/main/java/io/github/pzhin/sfqd/RefundCosts.java @@ -0,0 +1,143 @@ +package io.github.pzhin.sfqd; + +import java.math.BigInteger; +import java.util.Objects; + +/** AVL multiset of queued costs; retains only live costs, their sum, and their gcd. */ +final class RefundCosts { + private Node root; + private BigInteger sum = BigInteger.ZERO; + + BigInteger sum() { + return sum; + } + + BigInteger gcd() { + return BigInteger.valueOf(gcd(root)); + } + + void add(long cost) { + root = add(root, cost); + sum = sum.add(BigInteger.valueOf(cost)); + } + + void remove(long cost) { + root = remove(root, cost, false); + sum = sum.subtract(BigInteger.valueOf(cost)); + } + + private static Node add(Node node, long cost) { + if (node == null) { + return new Node(cost); + } + if (cost < node.cost) { + node.left = add(node.left, cost); + } else if (cost > node.cost) { + node.right = add(node.right, cost); + } else { + node.count++; + } + return balance(node); + } + + private static Node remove(Node node, long cost, boolean all) { + if (node == null) { + throw new IllegalStateException("queued cost missing"); + } + if (cost < node.cost) { + node.left = remove(node.left, cost, all); + } else if (cost > node.cost) { + node.right = remove(node.right, cost, all); + } else if (!all && node.count > 1) { + node.count--; + } else { + if (node.left == null) { + return node.right; + } + if (node.right == null) { + return node.left; + } + Node successor = node.right; + while (successor.left != null) { + successor = successor.left; + } + node.cost = successor.cost; + node.count = successor.count; + node.right = remove(node.right, successor.cost, true); + } + return balance(node); + } + + private static Node balance(Node node) { + update(node); + if (height(node.left) - height(node.right) > 1) { + Node left = Objects.requireNonNull(node.left); + if (height(left.left) < height(left.right)) { + node.left = rotateLeft(node.left); + } + return rotateRight(node); + } + if (height(node.right) - height(node.left) > 1) { + Node right = Objects.requireNonNull(node.right); + if (height(right.right) < height(right.left)) { + node.right = rotateRight(node.right); + } + return rotateLeft(node); + } + return node; + } + + private static Node rotateLeft(Node node) { + Node promoted = Objects.requireNonNull(node.right); + node.right = promoted.left; + promoted.left = node; + update(node); + update(promoted); + return promoted; + } + + private static Node rotateRight(Node node) { + Node promoted = Objects.requireNonNull(node.left); + node.left = promoted.right; + promoted.right = node; + update(node); + update(promoted); + return promoted; + } + + private static void update(Node node) { + node.height = 1 + Math.max(height(node.left), height(node.right)); + node.gcd = gcd(node.cost, gcd(gcd(node.left), gcd(node.right))); + } + + private static int height(Node node) { + return node == null ? 0 : node.height; + } + + private static long gcd(Node node) { + return node == null ? 0L : node.gcd; + } + + private static long gcd(long first, long second) { + while (second != 0L) { + long remainder = first % second; + first = second; + second = remainder; + } + return first; + } + + private static final class Node { + private long cost; + private int count = 1; + private int height = 1; + private long gcd; + private Node left; + private Node right; + + private Node(long cost) { + this.cost = cost; + gcd = cost; + } + } +} diff --git a/sfqd-core/src/main/java/io/github/pzhin/sfqd/SfqdScheduler.java b/sfqd-core/src/main/java/io/github/pzhin/sfqd/SfqdScheduler.java index 1faaf63..f4173b0 100644 --- a/sfqd-core/src/main/java/io/github/pzhin/sfqd/SfqdScheduler.java +++ b/sfqd-core/src/main/java/io/github/pzhin/sfqd/SfqdScheduler.java @@ -33,11 +33,12 @@ * *

Queued jobs are linked per flow and only each flow head is globally ordered. Under charge-reserved accounting, * enqueue and cancellation of a flow head are {@code O(log B)} for {@code B} backlogged flows; cancellation of a - * non-head is expected {@code O(1)}. Under refund accounting, enqueue scans the {@code K} queued jobs of its flow to - * reserve numeric budget and costs {@code O(K + log B)} when it creates a backlogged head or {@code O(K)} otherwise. - * Refund cancellation costs {@code O(K + log B)} when the indexed head changes and {@code O(K)} otherwise, for the - * suffix of {@code K} later jobs. Completion is expected {@code O(1)}. A batch selecting {@code m} jobs is - * {@code O(m log B + m)}. Snapshot is {@code O(1)}. The normative idle reset is + * non-head is expected {@code O(1)}. Under refund accounting, an AVL multiset maintains queued cost aggregates in + * {@code O(log K)} for {@code K} queued jobs in the flow. Enqueue and cancellation cost {@code O(log K + log B)} + * when the indexed head changes and {@code O(log K)} otherwise. Later tags are materialized only at head promotion. + * Completion is expected {@code O(1)}. A batch selecting {@code m} jobs under charge-reserved accounting costs + * {@code O(m log B + m)}; refund accounting adds {@code O(log K)} per selected job. Snapshot is {@code O(1)}. + * The normative idle reset is * {@code O(registeredFlows)}. A canonically triggered exact-tag rebase is * {@code O(queuedJobs + registeredFlows)} and is computed transactionally before it becomes observable. Internal * records are bounded by configured live-job and registration limits; terminal tombstones are not retained. @@ -179,7 +180,7 @@ public CloseFlowResult closeFlow(FlowHandle flowHandle) { * transition only when all transient and persistent results fit. Otherwise {@code NUMERIC_LIMIT} discards the * entire temporary computation: no tag, index, counter, sequence, or other observable state changes. * Under {@link CancellationAccounting#REFUND_CANCELLED_COST}, admission additionally verifies that every tag - * reachable by later queued cancellations fits the same budget. This check scans the prospective flow queue and + * reachable by later queued cancellations fits the same budget. This check uses exact queued-cost aggregates and * can return {@code NUMERIC_LIMIT} even when the candidate's immediate start and finish tags fit. A planned rebase * must preserve this property for every affected queued flow or the whole enqueue remains a no-op. * @@ -258,9 +259,10 @@ public EnqueueResult enqueue(FlowHandle flowHandle, J jobId, P payload, long cos *

Fairness accounting warning: under * {@link CancellationAccounting#CHARGE_RESERVED_COST}, cancellation does not reduce the flow's finish history or * recompute later tags, so the cancelled cost remains charged until global idle. Under - * {@link CancellationAccounting#REFUND_CANCELLED_COST}, cancellation recomputes every later queued job of the same - * flow from the cancelled job's start and updates the flow's finish history. Admission has already reserved the - * exact-arithmetic budget, so a queued cancellation has no numeric rejection or fallback. Recalculation occurs + * {@link CancellationAccounting#REFUND_CANCELLED_COST}, cancellation logically recomputes later queued tags + * from the cancelled job's start and updates the flow's finish history. Tags are materialized at head promotion. + * Admission has already reserved the + * exact-arithmetic budget, so a queued cancellation has no numeric rejection or fallback. The transition occurs * under the scheduler lock, does not change other flows, and does not revise earlier dispatch decisions. * Completed-work fairness guarantees do not apply to cancellation traces under either policy. * @@ -274,10 +276,7 @@ public CancelResult cancel(JobHandle handle) { try { QueuedJob job = queued.get(handle); if (job != null) { - RefundPlan refund = config.cancellationAccounting() - == CancellationAccounting.REFUND_CANCELLED_COST - ? prepareRefund(job) : null; - removeQueued(job, refund); + removeQueued(job); queued.remove(handle); liveById.remove(job.jobId); job.flow.cancelledCost = job.flow.cancelledCost.add(BigInteger.valueOf(job.cost)); @@ -448,9 +447,16 @@ private RebasePlan prepareRebase() throws NumericLimitException { flowTags.put(flow, transformed); } Map, TagPair> jobTags = new IdentityHashMap<>(); - for (QueuedJob job : queued.values()) { - jobTags.put(job, new TagPair( - job.start.subtractNonNegative(virtualTime), job.finish.subtractNonNegative(virtualTime))); + for (FlowState flow : registeredFlows.values()) { + ExactTag nextStart = flow.head == null ? null : flow.head.start; + for (QueuedJob job = flow.head; job != null; job = job.next) { + ExactTag start = refunds() ? nextStart : job.start; + ExactTag finish = refunds() ? start.add(ExactTag.fromCostAndWeight(job.cost, flow.weight)) + : job.finish; + jobTags.put(job, new TagPair( + start.subtractNonNegative(virtualTime), finish.subtractNonNegative(virtualTime))); + nextStart = finish; + } } return new RebasePlan<>(flowTags, jobTags); } @@ -487,51 +493,27 @@ private boolean refundClosureFits( if (base == null) { throw new AssertionError("refund closure requires a non-empty prospective chain"); } - BigInteger commonDenominator = base.denominator(); - for (QueuedJob job = flow.head; job != null; job = job.next) { - commonDenominator = lcm(commonDenominator, - reducedIncrementDenominator(job.cost, flow.weight)); - if (commonDenominator.bitLength() > ExactTag.MAX_PERSISTENT_BITS) { - return false; - } - } + // lcm(w/gcd(w,c_i)) = w/gcd(w,c_1,...,c_n). Using the sum alone would + // incorrectly accept chains whose reduced total hides individual denominator factors. + BigInteger weight = BigInteger.valueOf(flow.weight); + BigInteger costGcd = flow.refundCosts.gcd(); + BigInteger total = flow.refundCosts.sum(); if (candidateCost != 0L) { - commonDenominator = lcm(commonDenominator, - reducedIncrementDenominator(candidateCost, flow.weight)); - if (commonDenominator.bitLength() > ExactTag.MAX_PERSISTENT_BITS) { - return false; - } + costGcd = costGcd.gcd(BigInteger.valueOf(candidateCost)); + total = total.add(BigInteger.valueOf(candidateCost)); } - - BigInteger accumulatedNumerator = base.numerator() - .multiply(commonDenominator.divide(base.denominator())); - for (QueuedJob job = flow.head; job != null; job = job.next) { - accumulatedNumerator = addIncrementNumerator( - accumulatedNumerator, commonDenominator, job.cost, flow.weight); - } - if (candidateCost != 0L) { - accumulatedNumerator = addIncrementNumerator( - accumulatedNumerator, commonDenominator, candidateCost, flow.weight); + BigInteger incrementDenominator = weight.divide(weight.gcd(costGcd)); + BigInteger commonDenominator = lcm(base.denominator(), incrementDenominator); + if (commonDenominator.bitLength() > ExactTag.MAX_PERSISTENT_BITS) { + return false; } + BigInteger divisor = total.gcd(weight); + BigInteger accumulatedNumerator = base.numerator() + .multiply(commonDenominator.divide(base.denominator())) + .add(total.divide(divisor).multiply(commonDenominator.divide(weight.divide(divisor)))); return accumulatedNumerator.bitLength() <= ExactTag.MAX_PERSISTENT_BITS; } - private static BigInteger reducedIncrementDenominator(long cost, long weight) { - BigInteger numerator = BigInteger.valueOf(cost); - BigInteger denominator = BigInteger.valueOf(weight); - return denominator.divide(numerator.gcd(denominator)); - } - - private static BigInteger addIncrementNumerator( - BigInteger accumulated, BigInteger commonDenominator, long cost, long weight) { - BigInteger numerator = BigInteger.valueOf(cost); - BigInteger denominator = BigInteger.valueOf(weight); - BigInteger divisor = numerator.gcd(denominator); - BigInteger reducedNumerator = numerator.divide(divisor); - BigInteger reducedDenominator = denominator.divide(divisor); - return accumulated.add(reducedNumerator.multiply(commonDenominator.divide(reducedDenominator))); - } - private static BigInteger lcm(BigInteger first, BigInteger second) { return first.divide(first.gcd(second)).multiply(second); } @@ -547,18 +529,16 @@ private static TagPair plannedTags(QueuedJob job, RebasePlan r return tags; } - private RefundPlan prepareRefund(QueuedJob cancelledJob) { + private boolean refunds() { + return config.cancellationAccounting() == CancellationAccounting.REFUND_CANCELLED_COST; + } + + private static void setHeadTags(QueuedJob head, ExactTag start) { try { - ExactTag nextFinish = virtualTime.max(cancelledJob.start); - Map, TagPair> suffixTags = new IdentityHashMap<>(); - for (QueuedJob job = cancelledJob.next; job != null; job = job.next) { - ExactTag finish = nextFinish.add(ExactTag.fromCostAndWeight(job.cost, job.flow.weight)); - suffixTags.put(job, new TagPair(nextFinish, finish)); - nextFinish = finish; - } - return new RefundPlan<>(nextFinish, suffixTags); + head.start = start; + head.finish = start.add(ExactTag.fromCostAndWeight(head.cost, head.flow.weight)); } catch (NumericLimitException impossible) { - throw new AssertionError("refund-closed queued chain exceeded its reserved numeric budget", impossible); + throw new AssertionError("refund-closed head exceeded its reserved numeric budget", impossible); } } @@ -591,9 +571,12 @@ private void appendQueued(FlowState flow, QueuedJob job) { flow.tail = job; } flow.queuedCount++; + if (refunds()) { + flow.refundCosts.add(job.cost); + } } - private void removeQueued(QueuedJob job, RefundPlan refund) { + private void removeQueued(QueuedJob job) { FlowState flow = job.flow; boolean removesHead = job.previous == null; if (removesHead) { @@ -610,12 +593,17 @@ private void removeQueued(QueuedJob job, RefundPlan refund) { job.next.previous = job.previous; } flow.queuedCount--; - if (refund != null) { - for (Map.Entry, TagPair> entry : refund.suffixTags.entrySet()) { - entry.getKey().start = entry.getValue().start; - entry.getKey().finish = entry.getValue().finish; + if (refunds()) { + try { + flow.lastFinish = flow.lastFinish.subtractNonNegative( + ExactTag.fromCostAndWeight(job.cost, flow.weight)); + } catch (NumericLimitException impossible) { + throw new AssertionError("refund-closed finish exceeded its reserved numeric budget", impossible); + } + flow.refundCosts.remove(job.cost); + if (removesHead && flow.head != null) { + setHeadTags(flow.head, job.start); } - flow.lastFinish = refund.lastFinish; } if (removesHead && flow.head != null) { requireIndexChange(backlogged.add(flow.head), "promoted flow head was already indexed"); @@ -634,9 +622,15 @@ private void unlinkHead(FlowState flow, QueuedJob job) { flow.tail = null; } else { flow.head.previous = null; + if (refunds()) { + setHeadTags(flow.head, job.finish); + } requireIndexChange(backlogged.add(flow.head), "promoted flow head was already indexed"); } flow.queuedCount--; + if (refunds()) { + flow.refundCosts.remove(job.cost); + } } private void resetIfIdle() { @@ -682,6 +676,7 @@ private static final class FlowState { private final F flowId; private final long weight; private ExactTag lastFinish = ExactTag.zero(); + private final RefundCosts refundCosts = new RefundCosts(); private QueuedJob head; private QueuedJob tail; private int queuedCount; @@ -705,6 +700,7 @@ private static final class QueuedJob { private final P payload; private final long cost; private final long sequence; + // In refund mode only head tags are current; successors are derived from the head and live costs. private ExactTag start; private ExactTag finish; private QueuedJob previous; @@ -766,16 +762,6 @@ private RebasePlan( } } - private static final class RefundPlan { - private final ExactTag lastFinish; - private final Map, TagPair> suffixTags; - - private RefundPlan(ExactTag lastFinish, Map, TagPair> suffixTags) { - this.lastFinish = lastFinish; - this.suffixTags = suffixTags; - } - } - private static final class TagPair { private final ExactTag start; private final ExactTag finish; diff --git a/sfqd-core/src/test/java/io/github/pzhin/sfqd/RefundCostsTest.java b/sfqd-core/src/test/java/io/github/pzhin/sfqd/RefundCostsTest.java new file mode 100644 index 0000000..4088ea2 --- /dev/null +++ b/sfqd-core/src/test/java/io/github/pzhin/sfqd/RefundCostsTest.java @@ -0,0 +1,44 @@ +package io.github.pzhin.sfqd; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import java.math.BigInteger; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Random; +import org.junit.jupiter.api.Test; + +final class RefundCostsTest { + @Test + void aggregateMatchesIndependentScanThroughRotationsDuplicatesAndDeletion() { + RefundCosts costs = new RefundCosts(); + List remaining = new ArrayList<>(); + Random random = new Random(781923L); + for (int index = 1; index <= 2000; index++) { + long cost = index % 3 == 0 ? Long.MAX_VALUE : index; + costs.add(cost); + remaining.add(cost); + assertAggregate(costs, remaining); + } + Collections.shuffle(remaining, random); + while (!remaining.isEmpty()) { + costs.remove(remaining.remove(remaining.size() - 1)); + assertAggregate(costs, remaining); + } + assertThrows(IllegalStateException.class, () -> costs.remove(1)); + assertAggregate(costs, remaining); + } + + private static void assertAggregate(RefundCosts actual, List expected) { + BigInteger sum = BigInteger.ZERO; + BigInteger gcd = BigInteger.ZERO; + for (long cost : expected) { + sum = sum.add(BigInteger.valueOf(cost)); + gcd = gcd.gcd(BigInteger.valueOf(cost)); + } + assertEquals(sum, actual.sum()); + assertEquals(gcd, actual.gcd()); + } +} diff --git a/sfqd-core/src/test/java/io/github/pzhin/sfqd/SfqdNumericBoundaryTest.java b/sfqd-core/src/test/java/io/github/pzhin/sfqd/SfqdNumericBoundaryTest.java index 7ce6997..92284bc 100644 --- a/sfqd-core/src/test/java/io/github/pzhin/sfqd/SfqdNumericBoundaryTest.java +++ b/sfqd-core/src/test/java/io/github/pzhin/sfqd/SfqdNumericBoundaryTest.java @@ -89,6 +89,39 @@ void canonicalRebaseTransformsAllStateAndMatchesLogicalReference() assertEquals(reference.snapshot(), scheduler.snapshot()); } + @Test + void refundRebaseMaterializesLogicalSuffixAfterMiddleCancellation() + throws NumericLimitException, ReflectiveOperationException { + SchedulerConfig config = new SchedulerConfig(2, 4, 8, CancellationAccounting.REFUND_CANCELLED_COST); + SfqdScheduler scheduler = new SfqdScheduler<>(config); + ReferenceScheduler reference = new ReferenceScheduler<>(config); + FlowHandles flows = registerFour(scheduler); + FlowHandles oracle = registerFour(reference); + assertInstanceOf(EnqueueResult.Accepted.class, scheduler.enqueue(flows.first, "a1", "p", 1L)); + JobHandle middle = accepted(scheduler.enqueue(flows.first, "a2", "p", 7L)); + assertInstanceOf(EnqueueResult.Accepted.class, scheduler.enqueue(flows.first, "a3", "p", 2L)); + assertInstanceOf(EnqueueResult.Accepted.class, scheduler.enqueue(flows.second, "b1", "p", 1L)); + assertInstanceOf(EnqueueResult.Accepted.class, reference.enqueue(oracle.first, "a1", "p", 1L)); + JobHandle oracleMiddle = accepted(reference.enqueue(oracle.first, "a2", "p", 7L)); + assertInstanceOf(EnqueueResult.Accepted.class, reference.enqueue(oracle.first, "a3", "p", 2L)); + assertInstanceOf(EnqueueResult.Accepted.class, reference.enqueue(oracle.second, "b1", "p", 1L)); + assertEquals(reference.cancel(oracleMiddle), scheduler.cancel(middle)); + BigInteger baseInteger = BigInteger.ONE.shiftLeft(4095); + ExactTag base = ExactTag.fromComponents(baseInteger, BigInteger.ONE); + ExactTag targetFinish = ExactTag.fromComponents( + BigInteger.ONE.shiftLeft(4096).subtract(BigInteger.ONE), BigInteger.ONE); + NumericProbe.shiftQueuedAndFlowState(scheduler, base, flows, targetFinish); + NumericProbe.setReferenceLastFinish(reference, oracle.target, + ExactRational.of(baseInteger.subtract(BigInteger.ONE), BigInteger.ONE)); + assertInstanceOf(EnqueueResult.Accepted.class, scheduler.enqueue(flows.target, "target", "p", 1L)); + assertInstanceOf(EnqueueResult.Accepted.class, reference.enqueue(oracle.target, "target", "p", 1L)); + assertEquals(ExactTag.zero(), NumericProbe.capture(scheduler).virtualTime); + while (scheduler.snapshot().queuedJobs() != 0) { + complete(reference, scheduler, assertDispatches(reference.dispatchUpTo(2), scheduler.dispatchUpTo(2))); + } + assertEquals(reference.snapshot(), scheduler.snapshot()); + } + @Test void failedRebaseLeavesEveryHiddenNumericAndOrderingFieldUnchanged() throws NumericLimitException, ReflectiveOperationException { diff --git a/sfqd-core/src/test/java/io/github/pzhin/sfqd/SfqdRefundLongQueueTest.java b/sfqd-core/src/test/java/io/github/pzhin/sfqd/SfqdRefundLongQueueTest.java new file mode 100644 index 0000000..e929b0a --- /dev/null +++ b/sfqd-core/src/test/java/io/github/pzhin/sfqd/SfqdRefundLongQueueTest.java @@ -0,0 +1,74 @@ +package io.github.pzhin.sfqd; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; + +import java.util.ArrayList; +import java.util.List; +import java.util.SplittableRandom; +import org.junit.jupiter.api.Test; + +final class SfqdRefundLongQueueTest { + @Test + void arbitraryLongQueueCancellationsAndReplacementsMatchEagerOracle() { + SchedulerConfig config = new SchedulerConfig(16, 3, 4000, + CancellationAccounting.REFUND_CANCELLED_COST); + SfqdScheduler actual = new SfqdScheduler<>(config); + ReferenceScheduler expected = new ReferenceScheduler<>(config); + FlowHandle[] flows = new FlowHandle[3]; + FlowHandle[] oracleFlows = new FlowHandle[3]; + long[] weights = {6, 35, 16}; + for (int index = 0; index < flows.length; index++) { + flows[index] = registered(actual.registerFlow(index, weights[index])); + oracleFlows[index] = registered(expected.registerFlow(index, weights[index])); + } + List handles = new ArrayList<>(); + List oracleHandles = new ArrayList<>(); + SplittableRandom random = new SplittableRandom(793117L); + for (int id = 0; id < 3000; id++) { + int flow = id < 2500 ? 0 : id % 3; + long cost = id % 11 == 0 ? Long.MAX_VALUE : 1 + random.nextInt(1000); + handles.add(accepted(actual.enqueue(flows[flow], id, "payload", cost))); + oracleHandles.add(accepted(expected.enqueue(oracleFlows[flow], id, "payload", cost))); + } + for (int step = 0; step < 3000; step++) { + int index = step % 7 == 0 ? step / 7 : random.nextInt(handles.size()); + assertEquals(expected.cancel(oracleHandles.get(index)), actual.cancel(handles.get(index))); + if (step % 5 == 0) { + List> wanted = expected.dispatchUpTo(16); + List> got = actual.dispatchUpTo(16); + assertEquals(wanted.stream().map(Dispatch::jobId).toList(), + got.stream().map(Dispatch::jobId).toList()); + for (int job = 0; job < got.size(); job++) { + assertEquals(expected.complete(wanted.get(job).jobHandle()), + actual.complete(got.get(job).jobHandle())); + } + } + int flow = step % 3; + int id = 3000 + step; + handles.add(accepted(actual.enqueue(flows[flow], id, "replacement", 1 + step % 37))); + oracleHandles.add(accepted(expected.enqueue(oracleFlows[flow], id, "replacement", 1 + step % 37))); + assertEquals(expected.snapshot(), actual.snapshot()); + assertEquals(expected.snapshot(oracleFlows[flow]), actual.snapshot(flows[flow])); + } + while (actual.snapshot().queuedJobs() > 0) { + List> wanted = expected.dispatchUpTo(16); + List> got = actual.dispatchUpTo(16); + assertEquals(wanted.stream().map(Dispatch::jobId).toList(), + got.stream().map(Dispatch::jobId).toList()); + for (int job = 0; job < got.size(); job++) { + assertEquals(expected.complete(wanted.get(job).jobHandle()), + actual.complete(got.get(job).jobHandle())); + } + } + assertEquals(expected.snapshot(), actual.snapshot()); + } + + private static FlowHandle registered(RegisterFlowResult result) { + return assertInstanceOf(RegisterFlowResult.Registered.class, result).flowHandle(); + } + + private static JobHandle accepted(EnqueueResult result) { + return assertInstanceOf(EnqueueResult.Accepted.class, result).jobHandle(); + } +} diff --git a/sfqd-coverage/pom.xml b/sfqd-coverage/pom.xml index a2d09ff..6240009 100644 --- a/sfqd-coverage/pom.xml +++ b/sfqd-coverage/pom.xml @@ -6,7 +6,7 @@ io.github.pzhin sfqd-java-parent - 1.1.0 + 1.2.0 sfqd-coverage pom diff --git a/sfqd-examples/pom.xml b/sfqd-examples/pom.xml index 8d03342..3b8b129 100644 --- a/sfqd-examples/pom.xml +++ b/sfqd-examples/pom.xml @@ -6,7 +6,7 @@ io.github.pzhin sfqd-java-parent - 1.1.0 + 1.2.0 sfqd-examples SFQ(D) Examples diff --git a/sfqd-jcstress/pom.xml b/sfqd-jcstress/pom.xml index 48c2b50..ff16bc8 100644 --- a/sfqd-jcstress/pom.xml +++ b/sfqd-jcstress/pom.xml @@ -6,7 +6,7 @@ io.github.pzhin sfqd-java-parent - 1.1.0 + 1.2.0 sfqd-jcstress SFQ(D) jcstress Harness diff --git a/tools/verify-publication-topology.sh b/tools/verify-publication-topology.sh index e761a5b..8344769 100755 --- a/tools/verify-publication-topology.sh +++ b/tools/verify-publication-topology.sh @@ -100,8 +100,8 @@ PY ) readonly GROUP_DIRECTORY="${DEPLOY_REPOSITORY}/io/github/pzhin" -readonly PARENT_DIRECTORY="${GROUP_DIRECTORY}/sfqd-java-parent/1.1.0" -readonly CORE_DIRECTORY="${GROUP_DIRECTORY}/sfqd-core/1.1.0" +readonly PARENT_DIRECTORY="${GROUP_DIRECTORY}/sfqd-java-parent/1.2.0" +readonly CORE_DIRECTORY="${GROUP_DIRECTORY}/sfqd-core/1.2.0" shopt -s nullglob parent_poms=("${PARENT_DIRECTORY}"/*.pom) @@ -143,7 +143,7 @@ import sys import xml.etree.ElementTree as ET namespace = {"m": "http://maven.apache.org/POM/4.0.0"} -expected_parent = ("io.github.pzhin", "sfqd-java-parent", "1.1.0") +expected_parent = ("io.github.pzhin", "sfqd-java-parent", "1.2.0") parent = ET.parse(sys.argv[1]).getroot() parent_coordinates = tuple( @@ -209,7 +209,7 @@ expected_scm = ( "scm:git:https://github.com/pzhin/sfqd-java.git", "scm:git:ssh://git@github.com/pzhin/sfqd-java.git", "https://github.com/pzhin/sfqd-java", - "v1.1.0", + "v1.2.0", ) if scm_metadata != expected_scm: raise SystemExit( @@ -235,7 +235,7 @@ effective_core_coordinates = tuple( effective_core.findtext(f"m:{name}", namespaces=namespace) for name in ("groupId", "artifactId", "version") ) -expected_core = ("io.github.pzhin", "sfqd-core", "1.1.0") +expected_core = ("io.github.pzhin", "sfqd-core", "1.2.0") if effective_core_coordinates != expected_core: raise SystemExit( "ERROR: effective core coordinates are " @@ -313,7 +313,7 @@ cat >"${CONSUMER_ROOT}/pom.xml" < io.github.pzhin sfqd-core - 1.1.0 + 1.2.0 @@ -343,8 +343,8 @@ LC_ALL=C LANG=C TZ=UTC "${REPOSITORY_ROOT}/mvnw" \ -Dmaven.repo.local="${CONSUMER_LOCAL_REPOSITORY}" \ -f "${CONSUMER_ROOT}/pom.xml" clean compile -readonly RESOLVED_PARENT="${CONSUMER_LOCAL_REPOSITORY}/io/github/pzhin/sfqd-java-parent/1.1.0" -readonly RESOLVED_CORE="${CONSUMER_LOCAL_REPOSITORY}/io/github/pzhin/sfqd-core/1.1.0" +readonly RESOLVED_PARENT="${CONSUMER_LOCAL_REPOSITORY}/io/github/pzhin/sfqd-java-parent/1.2.0" +readonly RESOLVED_CORE="${CONSUMER_LOCAL_REPOSITORY}/io/github/pzhin/sfqd-core/1.2.0" if [[ ! -f "${RESOLVED_PARENT}/_remote.repositories" ]]; then fail "fresh consumer did not resolve the published parent" fi