diff --git a/README.md b/README.md index ade3916..f20c743 100644 --- a/README.md +++ b/README.md @@ -223,23 +223,52 @@ dispatched, it must be completed. #### Cancellation accounting -The only supported policy is -`CancellationAccounting.CHARGE_RESERVED_COST`. Cancelling a queued job removes -it from the queue and live-job indexes and releases its payload, but it does -not roll back the job's reserved virtual cost: +`CancellationAccounting.CHARGE_RESERVED_COST` remains the default for every +constructor that does not receive an explicit policy. Cancelling a queued job +removes it from the queue and live-job indexes and releases its payload, but it +does not roll back the job's reserved virtual cost: - the flow's `lastFinish` tag is not reduced; - tags already assigned to later jobs of the flow are not recomputed; - the charge disappears only when the scheduler becomes globally idle and ends the current busy period. -Consequently, completed-work fairness guarantees do not apply to any trace -containing cancellation. Treat frequent deadline or timeout cancellations as -a release blocker for an integration unless the resulting virtual charge and -dispatch delay are acceptable for that workload. +The opt-in alternative is: -For example, consider two equal-weight flows in a new busy period. Keeping a -job from B live prevents an idle reset: +```java +new SchedulerConfig( + issueDepth, + maxFlows, + maxLiveJobs, + CancellationAccounting.REFUND_CANCELLED_COST); +``` + +With `REFUND_CANCELLED_COST`, a successful queued cancellation removes that +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. + +Refund admission is intentionally narrower. Before accepting a job, enqueue +checks that the prospective flow queue has one common exact denominator and a +maximum accumulated numerator within the 4096-bit persistent budget. This +reserves enough numeric space for every later subset of queued cancellations. +An otherwise representable enqueue can therefore return `NUMERIC_LIMIT` under +the refund policy while the same enqueue remains accepted under the default +charge-reserved policy. If enqueue attempts a canonical rebase, every affected +queued flow must remain refund-closed or the complete enqueue is an atomic +no-op. An already accepted queued job can always be cancelled without a numeric +failure. + +Refund is prospective: dispatch decisions that linearized before cancellation +are never revised. Consequently, the published completed-work fairness bound is +not claimed for traces containing cancellation under either policy without a +separate proof. + +For example, under the default charge-reserved policy, consider two +equal-weight flows in a new busy period. Keeping a job from B live prevents an +idle reset: ```text A: enqueue cost=1_000_000 @@ -327,7 +356,9 @@ new SchedulerConfig( CancellationAccounting.CHARGE_RESERVED_COST); ``` -No free-cancellation accounting policy is currently implemented. +`REFUND_CANCELLED_COST` must always be selected explicitly. Existing +constructors and configurations using `CHARGE_RESERVED_COST` retain their +previous behavior and admission domain. The three- and four-argument forms preserve the unrestricted positive `long` weight domain. For production configurations with a known common scale, the @@ -370,7 +401,9 @@ The scheduler rejects work rather than allocating without bound: An enqueue may perform one transactional normalization when exact tags approach their numeric budget. If the result still cannot fit, it returns -`NUMERIC_LIMIT` without changing scheduler state. +`NUMERIC_LIMIT` without changing scheduler state. Refund accounting additionally +reserves numeric space for every later queued cancellation, so it may reject a +larger portion of the otherwise valid admission domain. `1_000_000` is a representation and validation limit for `issueDepth`, not a practically tested scale. One `dispatchUpTo(k)` call is atomic, holds the @@ -381,16 +414,21 @@ recorded run is reviewed for the target hardware and workload. ## Complexity -Let `R` be registered flows, `Q` queued jobs, `B` backlogged flows, and `m` the -number of jobs returned by one capacity call. +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 +one capacity call. | Operation | Expected or worst-case time | | --- | ---: | | register or close flow | expected `O(1)` | -| enqueue to a backlogged flow | expected `O(1)` | -| enqueue that makes a flow backlogged | `O(log B)` | -| cancel a non-head queued job | expected `O(1)` | -| cancel a flow head | `O(log B)` | +| 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)` | +| 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)` | | ordinary completion | expected `O(1)` | | aggregate or per-flow snapshot | expected `O(1)` | @@ -398,7 +436,9 @@ number of jobs returned by one capacity call. | rare exact-tag normalization | `O(Q + R)` time and temporary space | Retained state is `O(Q + running jobs + R)`. Terminal jobs and payloads are -not retained as tombstones. +not retained as tombstones. These are algorithmic complexity bounds, not +measured performance claims; no throughput or latency claim is made for refund +accounting without a preserved benchmark run. ## Verification and benchmarks diff --git a/docs/FORMAL_SPEC.md b/docs/FORMAL_SPEC.md index 9f11873..b0ad38b 100644 --- a/docs/FORMAL_SPEC.md +++ b/docs/FORMAL_SPEC.md @@ -72,8 +72,10 @@ Fairness is defined in terms of `cost`, not unknown actual execution time. integer in `1..Integer.MAX_VALUE`. - `maxLiveJobs` — the explicit `queued + dispatched` limit, an integer in `D..Integer.MAX_VALUE`. -- `cancellationAccounting` — the fixed policy - `CancellationAccounting.CHARGE_RESERVED_COST`; no alternative policy exists. +- `cancellationAccounting` — either + `CancellationAccounting.CHARGE_RESERVED_COST` or the opt-in + `CancellationAccounting.REFUND_CANCELLED_COST`. Constructors that do not + receive this value explicitly select `CHARGE_RESERVED_COST`. - `weightDomain` — either unrestricted positive `long` weights or the positive divisors of one fixed common scale `W`. @@ -230,6 +232,53 @@ charge-reserved cancellation policy can accumulate finish-tag debt during a continuous busy period. `NUMERIC_LIMIT` therefore remains a permitted enqueue result under a divisor-constrained domain. +### 3.2.2 Refund-closed queued chains + +**Cancellation accounting design decision.** Under +`REFUND_CANCELLED_COST`, successful admission MUST reserve enough of the +existing persistent numeric budget for every future subset of queued-job +cancellations to remain representable. The admission domain of this opt-in +policy is therefore narrower than the admission domain of +`CHARGE_RESERVED_COST`. + +For a non-empty queued chain of one flow, let `baseStart` be the start tag of +its head and let `r_i = cost_i/weight` be each queued job's reduced exact +increment in enqueue order. Define + +```text +L = lcm(denominator(baseStart), + denominator(r_1), ..., denominator(r_n)) + +A_0 = numerator(baseStart) * (L / denominator(baseStart)) +A_k = A_0 + sum(i=1..k, + numerator(r_i) * (L / denominator(r_i))) +``` + +The chain is **refund-closed** exactly when: + +```text +bitLength(L) <= 4096 +bitLength(A_n) <= 4096 +``` + +All increments are positive, so `A_n` is the maximum unreduced accumulated +numerator. Removing any subset of queued increments can only decrease that +numerator, and every resulting reduced denominator divides `L`. Consequently, +every tag required by any later sequence of successful queued cancellations +fits the persistent budget. Arithmetic over two such persistent values also +fits the 8193-bit transient budget. + +For `REFUND_CANCELLED_COST`, `enqueue` MUST test the prospective complete chain, +including the candidate job, before commit. Failure returns `NUMERIC_LIMIT` as +an atomic no-op. A canonical rebase attempted for the ordinary trigger in §3.2 +MUST also leave every transformed queued chain refund-closed, including the +candidate chain; otherwise the complete rebase-and-enqueue transaction returns +`NUMERIC_LIMIT` without mutation. Failure of the additional refund-closure +test alone does not trigger a rebase. + +`CHARGE_RESERVED_COST` does not perform this additional admission test and +retains the pre-existing numeric admission domain. + ### 3.3 Exact rebasing A rebase is a representational substitution of the same semantic state. Let @@ -275,7 +324,8 @@ plain SFQ(D) described under Scheduler state is the tuple: ```text -Config = (D, maxFlows, maxLiveJobs) +Config = (D, maxFlows, maxLiveJobs, cancellationAccounting, + weightDomain) ownerToken = inert identity token of this instance V = virtual-time tag lastJobSequence = last issued job long sequence, initially 0 @@ -389,6 +439,8 @@ In every observable state: runningSuppliedCost = sum(cost(job) for running jobs of this flow) ``` +16. Under `REFUND_CANCELLED_COST`, every non-empty queued flow chain is + refund-closed as defined in §3.2.2. ## 5. Lifecycles and identity @@ -478,14 +530,21 @@ create a registration and does not accept a weight. A dormant registered flow uses its stored `lastFinish`, not zero; an inactive interval within a busy period therefore does not reset fairness history. -Tags are fixed at enqueue. Later cancellation of another queued job MUST NOT -recalculate the tags of remaining jobs or roll back `lastFinish`. +Under `CHARGE_RESERVED_COST`, tags are fixed at enqueue. Later cancellation of +another queued job MUST NOT recalculate the tags of remaining jobs or roll back +`lastFinish`; the cancelled supplied cost remains a virtual charge until the +end of the current global busy period. -This last rule is a **cancellation design decision** absent from Jin04. It -prevents retroactive changes to accepted scheduling decisions. Its cost is that -a cancelled supplied cost remains a virtual charge until the end of the current -global busy period; the published completed-work fairness bound is not claimed -for intervals containing cancellation. +Under `REFUND_CANCELLED_COST`, enqueue assigns the same initial tags, but §7.4 +prospectively recomputes later queued tags of the same flow after a successful +cancellation. The refund changes only future scheduling state at the cancel +linearization point. Earlier dispatch results and virtual-time observations are +irreversible, so the result is not required to equal a history in which the +cancelled job was never accepted. + +Both policies are **cancellation design decisions** absent from Jin04. The +published completed-work fairness bound is not claimed for intervals containing +cancellation under either policy without a separate proof. ### 6.2 Deterministic tie-breaking @@ -575,7 +634,10 @@ Order of processing: 5. If the job sequence is exhausted, return `SEQUENCE_EXHAUSTED`. 6. Using the fixed registered weight and `lastFinish`, calculate exact `S` and `F`; when required, apply §3.3 transactionally to the complete necessary - state copy. If the budget remains violated, return `NUMERIC_LIMIT`. + state copy. Under `REFUND_CANCELLED_COST`, also apply the prospective + refund-closure checks from §3.2.2, including checks for every chain affected + by a planned rebase. If any applicable budget remains violated, return + `NUMERIC_LIMIT`. 7. Create the inert JobHandle and queued record, insert it into all job indexes, increment the registered flow's `acceptedCost` by `cost`, and update the registered flow counts, `lastFinish`, scheduler-wide counters, and job @@ -590,8 +652,11 @@ not consume a sequence. A null handle is an invalid argument. An opaque handle from another scheduler instance is treated as `NOT_LIVE`. -- If the handle is in `Queued`, atomically remove the job from the queue, - priority, and `LiveById`; decrement the flow count, increment the registered +- If the handle is in `Queued`, cancellation always succeeds. Under + `CHARGE_RESERVED_COST`, atomically remove the job from the queue, priority, + and `LiveById` without changing remaining tags or `lastFinish`. Under + `REFUND_CANCELLED_COST`, apply the prospective refund transition below. + Under both policies, decrement the flow count, increment the registered flow's `cancelledCost` by `cost(job)`, increment the scheduler-wide `cancelled` counter, release the payload, and return `CANCELLED`. - If the handle is in `Running`, change nothing and return @@ -608,6 +673,28 @@ scheduler's last live job, the same transition performs the §3.4 reset for all registrations. Cancellation does not return capacity because a queued job did not occupy any. +For `REFUND_CANCELLED_COST`, let `c` be the queued job selected for +cancellation. At the successful cancel LP: + +1. set `nextFinish := max(V, S(c))`; +2. for every later queued job of the same flow in enqueue order, set + `S(job) := nextFinish`, then set + `F(job) := S(job) + cost(job)/weight(flow)`, and advance + `nextFinish := F(job)`; +3. set `lastFinish(flow) := nextFinish`; +4. remove `c` from the per-flow queue, queued/live indexes, and priority index; +5. release its payload and update lifecycle and supplied-cost counters; +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 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 +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. + ### 7.5 `dispatchUpTo(k)` / `dispatch(k)` `k` is an integer in `0..D`; a negative value or `k>D` is an invalid argument. @@ -851,9 +938,12 @@ FlowHandle in this busy period uses `max(V,lastFinish)`. While identity. Once `V` reaches `lastFinish`, the registration can be closed safely without changing the start tag of the next possible job. -The virtual charge of a cancelled job persists even after deactivation until -the global idle reset. This is the intentional non-retroactive semantics of -§6.1. +Under `CHARGE_RESERVED_COST`, the virtual charge of a cancelled job persists +even after deactivation until the global idle reset. Under +`REFUND_CANCELLED_COST`, `lastFinish` is instead moved to the exact end of the +remaining queued suffix at cancellation, while debt represented by already +dispatched work is retained through the suffix base. Both behaviors follow +§6.1 and §7.4. ### 9.4 Registered → closed @@ -887,8 +977,11 @@ weights, finite per-flow maximum costs, and a publication-compatible trace: ``` The units are supplied cost. This document does not claim the bound for -intervals containing cancellation because a non-retroactive virtual charge is -not completed work. +intervals containing cancellation under either accounting policy. Reserved +cost is not completed work under `CHARGE_RESERVED_COST`, while prospective +refund changes future tags without revising earlier dispatch decisions under +`REFUND_CANCELLED_COST`; neither extension inherits the published bound without +a separate proof. No-starvation is claimed only under the preconditions in [Starvation and progress](THEORY.md#starvation-and-progress): a bounded diff --git a/docs/THEORY.md b/docs/THEORY.md index eb201d0..f6b5caf 100644 --- a/docs/THEORY.md +++ b/docs/THEORY.md @@ -79,6 +79,8 @@ The Java library makes the unspecified parts explicit: - every public operation is linearizable; - equal start tags use admission order as a stable FIFO tie-break; - cancellation succeeds only before dispatch; +- charge-reserved cancellation is the default, while opt-in refund cancellation + prospectively removes queued virtual cost from later work of the same flow; - handles are opaque and scheduler-specific; - counts, sequences, and exact-number sizes are bounded and fail closed; - rejected operations are atomic no-ops; @@ -89,6 +91,28 @@ The last rule is an equivalent busy-period normalization described for SFQ. It prevents old virtual debt from growing forever across periods with no live work while preserving the ordering within each busy period. +### Cancellation policies and claims + +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 +cancellation linearization point. It does not undo an earlier dispatch or +rewrite virtual-time history. + +Exact refund must not turn cancellation into a fallible cleanup operation. The +refund policy therefore narrows admission: before accepting a job, the library +proves that all tags reachable by removing any subset of the queued costs fit +the fixed exact-number budget. Positivity makes the full queued sum the largest +candidate numerator, and a common denominator covers every subset. Numeric +failure is reported by enqueue, while every accepted queued job remains safely +cancellable. + +Neither cancellation extension automatically inherits the papers' +completed-work fairness bound. Charge-reserved debt is not completed service, +and prospective refund changes future tags without revising past scheduling +decisions. A claim for cancellation intervals would require a separate proof. + ## Starvation and progress The scheduler cannot make work finish and cannot guarantee progress when the diff --git a/sfqd-benchmarks/README.md b/sfqd-benchmarks/README.md index 9da43e3..19ec483 100644 --- a/sfqd-benchmarks/README.md +++ b/sfqd-benchmarks/README.md @@ -12,8 +12,9 @@ establish production performance by existing in the repository. java -jar sfqd-benchmarks/target/sfqd-benchmarks.jar -l ``` -The Maven `verify` phase also runs a bounded 60-case idle-reset fixture smoke -and three scale-fixture cases. The largest scale fixture uses `B=10_000`, +The Maven `verify` phase also runs a bounded 90-case idle-reset fixture smoke +covering both cancellation-accounting policies, and three scale-fixture cases. +The largest scale fixture uses `B=10_000`, `Q=100_000`, and `depth=1_024`. These checks validate scheduler state and workload restoration; their execution time is not a benchmark result. @@ -74,7 +75,11 @@ metric and the normalization. ### Cancellation cycles `CancellationCycleBenchmark` measures `cancel -> enqueue replacement`. The -score belongs to the combined cycle, not cancellation alone. +score belongs to the combined cycle, not cancellation alone. 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 +first-busy-period cancellation-cycle workloads. ### First busy-period cycles 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 50441cd..3ef95fa 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 @@ -4,6 +4,7 @@ import static io.github.pzhin.sfqd.benchmarks.SchedulerBenchmarkSupport.requireCancelled; import io.github.pzhin.sfqd.CancelResult; +import io.github.pzhin.sfqd.CancellationAccounting; import io.github.pzhin.sfqd.EnqueueResult; import io.github.pzhin.sfqd.SchedulerSnapshot; import io.github.pzhin.sfqd.benchmarks.SchedulerBenchmarkSupport.Fixture; @@ -54,6 +55,10 @@ public static class CancellationState { @Param private Target target; + /** Cancellation accounting policy under measurement. */ + @Param({"CHARGE_RESERVED_COST", "REFUND_CANCELLED_COST"}) + private CancellationAccounting cancellationAccounting; + private Fixture fixture; private boolean nextHead; private int nextFlow; @@ -67,7 +72,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); + fixture = new Fixture(flowCount, depth, scenario, 0, 2, cancellationAccounting); } /** Selects and removes caller-side target bookkeeping outside the timed cycle. */ diff --git a/sfqd-benchmarks/src/main/java/io/github/pzhin/sfqd/benchmarks/FirstBusyPeriodCycleBenchmark.java b/sfqd-benchmarks/src/main/java/io/github/pzhin/sfqd/benchmarks/FirstBusyPeriodCycleBenchmark.java index d704c14..c82257b 100644 --- a/sfqd-benchmarks/src/main/java/io/github/pzhin/sfqd/benchmarks/FirstBusyPeriodCycleBenchmark.java +++ b/sfqd-benchmarks/src/main/java/io/github/pzhin/sfqd/benchmarks/FirstBusyPeriodCycleBenchmark.java @@ -1,5 +1,6 @@ package io.github.pzhin.sfqd.benchmarks; +import io.github.pzhin.sfqd.CancellationAccounting; import io.github.pzhin.sfqd.benchmarks.IdleResetBenchmarkSupport.FirstBusyPeriodFixture; import java.util.Objects; import java.util.concurrent.TimeUnit; @@ -30,12 +31,16 @@ public static class CycleState { @Param({"1", "256"}) private int depth; + /** Cancellation accounting policy under measurement. */ + @Param({"CHARGE_RESERVED_COST", "REFUND_CANCELLED_COST"}) + private CancellationAccounting cancellationAccounting; + private FirstBusyPeriodFixture fixture; /** Registers all flows and establishes the initial globally idle boundary. */ @Setup(Level.Trial) public void setupTrial() { - fixture = new FirstBusyPeriodFixture(flowCount, depth); + fixture = new FirstBusyPeriodFixture(flowCount, depth, cancellationAccounting); } /** Verifies that every measured transaction conserved bounded idle state. */ diff --git a/sfqd-benchmarks/src/main/java/io/github/pzhin/sfqd/benchmarks/IdleResetBenchmarkSupport.java b/sfqd-benchmarks/src/main/java/io/github/pzhin/sfqd/benchmarks/IdleResetBenchmarkSupport.java index 0ad1986..8e81ecb 100644 --- a/sfqd-benchmarks/src/main/java/io/github/pzhin/sfqd/benchmarks/IdleResetBenchmarkSupport.java +++ b/sfqd-benchmarks/src/main/java/io/github/pzhin/sfqd/benchmarks/IdleResetBenchmarkSupport.java @@ -7,6 +7,7 @@ import static io.github.pzhin.sfqd.benchmarks.SchedulerBenchmarkSupport.requireRegistered; import io.github.pzhin.sfqd.CancelResult; +import io.github.pzhin.sfqd.CancellationAccounting; import io.github.pzhin.sfqd.CompletionResult; import io.github.pzhin.sfqd.Dispatch; import io.github.pzhin.sfqd.EnqueueResult; @@ -45,6 +46,15 @@ static final class TerminalFixture { private JobHandle terminalHandle; TerminalFixture(int flowCount, int depth, boolean allTagged, TerminalOperation operation) { + this(flowCount, depth, allTagged, operation, CancellationAccounting.CHARGE_RESERVED_COST); + } + + TerminalFixture( + int flowCount, + int depth, + boolean allTagged, + TerminalOperation operation, + CancellationAccounting cancellationAccounting) { if (flowCount < 1 || depth < 1) { throw new IllegalArgumentException("flowCount and depth must be positive"); } @@ -52,7 +62,8 @@ static final class TerminalFixture { this.depth = depth; this.allTagged = allTagged; this.operation = operation; - this.scheduler = new SfqdScheduler<>(new SchedulerConfig(depth, flowCount, Math.max(depth, flowCount))); + this.scheduler = new SfqdScheduler<>(new SchedulerConfig( + depth, flowCount, Math.max(depth, flowCount), cancellationAccounting)); this.flows = new ArrayList<>(flowCount); this.jobs = new ArrayList<>(flowCount); for (int index = 0; index < flowCount; index++) { @@ -187,12 +198,18 @@ static final class FirstBusyPeriodFixture { private final JobKey job = new JobKey(1L); FirstBusyPeriodFixture(int flowCount, int depth) { + this(flowCount, depth, CancellationAccounting.CHARGE_RESERVED_COST); + } + + FirstBusyPeriodFixture( + int flowCount, int depth, CancellationAccounting cancellationAccounting) { if (flowCount < 1 || depth < 1) { throw new IllegalArgumentException("flowCount and depth must be positive"); } this.flowCount = flowCount; this.depth = depth; - this.scheduler = new SfqdScheduler<>(new SchedulerConfig(depth, flowCount, depth)); + this.scheduler = new SfqdScheduler<>(new SchedulerConfig( + depth, flowCount, depth, cancellationAccounting)); FlowHandle first = null; for (int index = 0; index < flowCount; index++) { FlowHandle registered = requireRegistered(scheduler.registerFlow(new FlowKey(index), 1L)); diff --git a/sfqd-benchmarks/src/main/java/io/github/pzhin/sfqd/benchmarks/IdleResetWorkloadSmoke.java b/sfqd-benchmarks/src/main/java/io/github/pzhin/sfqd/benchmarks/IdleResetWorkloadSmoke.java index fb9675f..05902c0 100644 --- a/sfqd-benchmarks/src/main/java/io/github/pzhin/sfqd/benchmarks/IdleResetWorkloadSmoke.java +++ b/sfqd-benchmarks/src/main/java/io/github/pzhin/sfqd/benchmarks/IdleResetWorkloadSmoke.java @@ -1,6 +1,7 @@ package io.github.pzhin.sfqd.benchmarks; import io.github.pzhin.sfqd.CancelResult; +import io.github.pzhin.sfqd.CancellationAccounting; import io.github.pzhin.sfqd.CompletionResult; import io.github.pzhin.sfqd.EnqueueResult; import io.github.pzhin.sfqd.benchmarks.IdleResetBenchmarkSupport.FirstBusyPeriodFixture; @@ -14,6 +15,10 @@ public final class IdleResetWorkloadSmoke { private static final int[] ALL_TAGGED_FLOW_COUNTS = {1, 100, 10_000}; private static final int[] TARGET_DEPTHS = {1, 256}; private static final int[] FIRST_BUSY_FLOW_COUNTS = {1, 10_000}; + private static final CancellationAccounting[] CANCELLATION_POLICIES = { + CancellationAccounting.CHARGE_RESERVED_COST, + CancellationAccounting.REFUND_CANCELLED_COST + }; private IdleResetWorkloadSmoke() { } @@ -31,25 +36,34 @@ public static void main(String[] arguments) { for (int flowCount : FULL_FLOW_COUNTS) { for (int depth : FULL_DEPTHS) { exerciseCompletion(flowCount, depth, false); - exerciseCancellation(flowCount, depth, false); - cases += 2; + cases++; + for (CancellationAccounting policy : CANCELLATION_POLICIES) { + exerciseCancellation(flowCount, depth, false, policy); + cases++; + } } } for (int flowCount : ALL_TAGGED_FLOW_COUNTS) { for (int depth : TARGET_DEPTHS) { exerciseCompletion(flowCount, depth, true); - exerciseCancellation(flowCount, depth, true); - cases += 2; + cases++; + for (CancellationAccounting policy : CANCELLATION_POLICIES) { + exerciseCancellation(flowCount, depth, true, policy); + cases++; + } } } for (int flowCount : FIRST_BUSY_FLOW_COUNTS) { for (int depth : TARGET_DEPTHS) { exerciseFirstAdmission(flowCount, depth); - exerciseCycle(flowCount, depth); - cases += 2; + cases++; + for (CancellationAccounting policy : CANCELLATION_POLICIES) { + exerciseCycle(flowCount, depth, policy); + cases++; + } } } - if (cases != 60) { + if (cases != 90) { throw new IllegalStateException("unexpected idle-reset smoke matrix size: " + cases); } System.out.println("IDLE_RESET_WORKLOAD_SMOKE PASS cases=" + cases); @@ -63,9 +77,10 @@ private static void exerciseCompletion(int flowCount, int depth, boolean allTagg fixture.verifyPrepared(); } - private static void exerciseCancellation(int flowCount, int depth, boolean allTagged) { + private static void exerciseCancellation( + int flowCount, int depth, boolean allTagged, CancellationAccounting policy) { TerminalFixture fixture = new TerminalFixture( - flowCount, depth, allTagged, TerminalOperation.CANCEL); + flowCount, depth, allTagged, TerminalOperation.CANCEL, policy); CancelResult result = fixture.cancelLastQueued(); fixture.restoreAfterCancellation(result); fixture.verifyPrepared(); @@ -78,8 +93,8 @@ private static void exerciseFirstAdmission(int flowCount, int depth) { fixture.verifyIdle(); } - private static void exerciseCycle(int flowCount, int depth) { - FirstBusyPeriodFixture fixture = new FirstBusyPeriodFixture(flowCount, depth); + private static void exerciseCycle(int flowCount, int depth, CancellationAccounting policy) { + FirstBusyPeriodFixture fixture = new FirstBusyPeriodFixture(flowCount, depth, policy); if (fixture.enqueueCancelCycle() != 1) { throw new IllegalStateException("first-busy-period cycle count diverged"); } diff --git a/sfqd-benchmarks/src/main/java/io/github/pzhin/sfqd/benchmarks/OperationLatencyBenchmark.java b/sfqd-benchmarks/src/main/java/io/github/pzhin/sfqd/benchmarks/OperationLatencyBenchmark.java index dfad629..71e62ef 100644 --- a/sfqd-benchmarks/src/main/java/io/github/pzhin/sfqd/benchmarks/OperationLatencyBenchmark.java +++ b/sfqd-benchmarks/src/main/java/io/github/pzhin/sfqd/benchmarks/OperationLatencyBenchmark.java @@ -6,6 +6,7 @@ import static io.github.pzhin.sfqd.benchmarks.SchedulerBenchmarkSupport.requireCompleted; import io.github.pzhin.sfqd.CancelResult; +import io.github.pzhin.sfqd.CancellationAccounting; import io.github.pzhin.sfqd.CompletionResult; import io.github.pzhin.sfqd.Dispatch; import io.github.pzhin.sfqd.EnqueueResult; @@ -248,6 +249,10 @@ public static class CancellationState extends MatrixState { @Param({"HEAD", "NON_HEAD"}) private CancelPosition position; + /** Cancellation accounting policy under measurement. */ + @Param({"CHARGE_RESERVED_COST", "REFUND_CANCELLED_COST"}) + private CancellationAccounting cancellationAccounting; + private Fixture fixture; private JobRecord target; private CancelResult result; @@ -258,7 +263,7 @@ public static class CancellationState extends MatrixState { /** Builds per-flow queues with at least two entries. */ @Setup(Level.Trial) public void setupTrial() { - fixture = new Fixture(flowCount(), depth(), scenario(), 0, 2); + fixture = new Fixture(flowCount(), depth(), scenario(), 0, 2, cancellationAccounting); } /** Selects a stable queued target without charging caller lookup to scheduler latency. */ @@ -429,13 +434,18 @@ public static class OneTaggedCancellationState { @Param({"1", "8", "64", "256"}) private int depth; + /** Cancellation accounting policy under measurement. */ + @Param({"CHARGE_RESERVED_COST", "REFUND_CANCELLED_COST"}) + private CancellationAccounting cancellationAccounting; + private TerminalFixture fixture; private CancelResult result; /** Establishes exactly one queued job while every other registered flow remains untagged. */ @Setup(Level.Trial) public void setupTrial() { - fixture = new TerminalFixture(flowCount, depth, false, TerminalOperation.CANCEL); + fixture = new TerminalFixture( + flowCount, depth, false, TerminalOperation.CANCEL, cancellationAccounting); } /** Validates the idle transition and restores the exact one-tagged boundary. */ @@ -523,13 +533,18 @@ public static class AllTaggedCancellationState { @Param({"1", "256"}) private int depth; + /** Cancellation accounting policy under measurement. */ + @Param({"CHARGE_RESERVED_COST", "REFUND_CANCELLED_COST"}) + private CancellationAccounting cancellationAccounting; + private TerminalFixture fixture; private CancelResult result; /** Tags every flow without ending the busy period and leaves the final job queued. */ @Setup(Level.Trial) public void setupTrial() { - fixture = new TerminalFixture(flowCount, depth, true, TerminalOperation.CANCEL); + fixture = new TerminalFixture( + flowCount, depth, true, TerminalOperation.CANCEL, cancellationAccounting); } /** Validates global idle and re-establishes the all-tagged boundary. */ diff --git a/sfqd-benchmarks/src/main/java/io/github/pzhin/sfqd/benchmarks/SchedulerBenchmarkSupport.java b/sfqd-benchmarks/src/main/java/io/github/pzhin/sfqd/benchmarks/SchedulerBenchmarkSupport.java index 7766803..74b4dc6 100644 --- a/sfqd-benchmarks/src/main/java/io/github/pzhin/sfqd/benchmarks/SchedulerBenchmarkSupport.java +++ b/sfqd-benchmarks/src/main/java/io/github/pzhin/sfqd/benchmarks/SchedulerBenchmarkSupport.java @@ -1,6 +1,7 @@ package io.github.pzhin.sfqd.benchmarks; import io.github.pzhin.sfqd.CancelResult; +import io.github.pzhin.sfqd.CancellationAccounting; import io.github.pzhin.sfqd.CompletionResult; import io.github.pzhin.sfqd.Dispatch; import io.github.pzhin.sfqd.EnqueueResult; @@ -119,6 +120,17 @@ static final class Fixture { } Fixture(int flowCount, int depth, Scenario scenario, int inactiveRegistrations, int minimumAnchors) { + this(flowCount, depth, scenario, inactiveRegistrations, minimumAnchors, + CancellationAccounting.CHARGE_RESERVED_COST); + } + + Fixture( + int flowCount, + int depth, + Scenario scenario, + int inactiveRegistrations, + int minimumAnchors, + CancellationAccounting cancellationAccounting) { if (flowCount < 1 || inactiveRegistrations < 0) { throw new IllegalArgumentException("flow counts must be valid"); } @@ -129,7 +141,7 @@ static final class Fixture { int initiallyActive = expectedActiveFlows(flowCount, scenario); int plannedJobs = Math.addExact(Math.multiplyExact(initiallyActive, anchors), reserve); this.scheduler = new SfqdScheduler<>(new SchedulerConfig( - depth, registered, Math.addExact(plannedJobs, depth + 64))); + depth, registered, Math.addExact(plannedJobs, depth + 64), cancellationAccounting)); this.flowIds = new ArrayList<>(registered); this.flowHandles = new ArrayList<>(registered); this.queuedByFlow = new ArrayList<>(registered); diff --git a/sfqd-core/src/main/api/public-api.txt b/sfqd-core/src/main/api/public-api.txt index 11a7894..a55932d 100644 --- a/sfqd-core/src/main/api/public-api.txt +++ b/sfqd-core/src/main/api/public-api.txt @@ -29,6 +29,7 @@ FIELD owner=io.github.pzhin.sfqd.CancelResult name=CANCELLED modifiers=[public, FIELD owner=io.github.pzhin.sfqd.CancelResult name=NOT_LIVE modifiers=[public, static, final] type=io.github.pzhin.sfqd.CancelResult enumConstant=true synthetic=false FIELD owner=io.github.pzhin.sfqd.CancelResult name=TOO_LATE_ALREADY_DISPATCHED modifiers=[public, static, final] type=io.github.pzhin.sfqd.CancelResult enumConstant=true synthetic=false FIELD owner=io.github.pzhin.sfqd.CancellationAccounting name=CHARGE_RESERVED_COST modifiers=[public, static, final] type=io.github.pzhin.sfqd.CancellationAccounting enumConstant=true synthetic=false +FIELD owner=io.github.pzhin.sfqd.CancellationAccounting name=REFUND_CANCELLED_COST modifiers=[public, static, final] type=io.github.pzhin.sfqd.CancellationAccounting enumConstant=true synthetic=false FIELD owner=io.github.pzhin.sfqd.CloseFlowResult name=CLOSED modifiers=[public, static, final] type=io.github.pzhin.sfqd.CloseFlowResult enumConstant=true synthetic=false FIELD owner=io.github.pzhin.sfqd.CloseFlowResult name=FAIRNESS_DEBT_ACTIVE modifiers=[public, static, final] type=io.github.pzhin.sfqd.CloseFlowResult enumConstant=true synthetic=false FIELD owner=io.github.pzhin.sfqd.CloseFlowResult name=FLOW_ACTIVE modifiers=[public, static, final] type=io.github.pzhin.sfqd.CloseFlowResult enumConstant=true synthetic=false diff --git a/sfqd-core/src/main/java/io/github/pzhin/sfqd/CancellationAccounting.java b/sfqd-core/src/main/java/io/github/pzhin/sfqd/CancellationAccounting.java index 0398258..41dd386 100644 --- a/sfqd-core/src/main/java/io/github/pzhin/sfqd/CancellationAccounting.java +++ b/sfqd-core/src/main/java/io/github/pzhin/sfqd/CancellationAccounting.java @@ -3,8 +3,8 @@ /** * Defines how a successful queued-job cancellation affects virtual fairness accounting. * - *

The current implementation supports only {@link #CHARGE_RESERVED_COST}. The explicit policy prevents a queued - * cancellation from being mistaken for a free rollback of the job's scheduling tags. + *

{@link #CHARGE_RESERVED_COST} preserves accepted virtual debt and remains the default. + * {@link #REFUND_CANCELLED_COST} is opt-in and prospectively recomputes later queued work of the same flow. */ public enum CancellationAccounting { /** @@ -15,5 +15,14 @@ public enum CancellationAccounting { * performs its global idle reset. Completed-work fairness guarantees do not apply to traces containing a * cancellation. */ - CHARGE_RESERVED_COST + CHARGE_RESERVED_COST, + + /** + * Refunds a cancelled queued job's virtual cost from later queued work of the same flow. + * + *

Admission reserves sufficient exact-arithmetic budget so every later queued cancellation can recompute the + * affected suffix without a numeric failure. Cancellation does not revise earlier dispatch decisions, and + * completed-work fairness guarantees do not apply to traces containing cancellation. + */ + REFUND_CANCELLED_COST } diff --git a/sfqd-core/src/main/java/io/github/pzhin/sfqd/SchedulerConfig.java b/sfqd-core/src/main/java/io/github/pzhin/sfqd/SchedulerConfig.java index e71c9b6..4e0ef1e 100644 --- a/sfqd-core/src/main/java/io/github/pzhin/sfqd/SchedulerConfig.java +++ b/sfqd-core/src/main/java/io/github/pzhin/sfqd/SchedulerConfig.java @@ -13,13 +13,18 @@ * {@code [0, depth]}, registered flows are in {@code [0, maxFlows]}, and * queued plus running jobs are in {@code [0, maxLiveJobs]}. * + *

{@link CancellationAccounting#CHARGE_RESERVED_COST} retains the historical admission domain and is selected by + * the three-argument constructor. The opt-in {@link CancellationAccounting#REFUND_CANCELLED_COST} policy may reject + * an otherwise representable enqueue with {@link EnqueueResult.Rejected#NUMERIC_LIMIT} when the complete prospective + * flow queue would not reserve enough exact-arithmetic budget for every later cancellation. + * * @param depth maximum dispatched-but-not-completed issue depth, in * {@code [1, 1_000_000]} * @param maxFlows maximum simultaneously registered flows, in * {@code [1, Integer.MAX_VALUE]} * @param maxLiveJobs maximum queued plus running jobs, in * {@code [depth, Integer.MAX_VALUE]} - * @param cancellationAccounting virtual fairness accounting policy for cancelled queued jobs + * @param cancellationAccounting immutable virtual fairness accounting and admission policy for cancelled queued jobs * @param weightDomain registration policy for fixed flow weights */ public record SchedulerConfig( 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 7d28bf5..1faaf63 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 @@ -18,8 +18,10 @@ *

The scheduler only makes admission and dispatch decisions. It neither executes jobs nor owns an executor, * thread pool, resource pool, or completion callback. Fairness is measured against the caller-supplied cost and the * registered flow weight, not against unknown actual execution time. - * Completed-work fairness guarantees do not apply to traces containing cancellation because the supported - * {@link CancellationAccounting#CHARGE_RESERVED_COST} policy retains cancelled jobs' virtual cost until global idle. + * Completed-work fairness guarantees do not apply to traces containing cancellation. The default + * {@link CancellationAccounting#CHARGE_RESERVED_COST} policy retains cancelled jobs' virtual cost until global idle; + * the opt-in {@link CancellationAccounting#REFUND_CANCELLED_COST} policy changes only future queued tags and does not + * revise earlier dispatch decisions. * *

All public operations are linearizable and may be invoked concurrently without external synchronization. This * baseline uses one private lock: a successful mutating operation linearizes when its complete state transition is @@ -29,10 +31,14 @@ * {@link CancelResult#NOT_LIVE} alone intentionally does not reveal the terminal cause. Completion and cancellation * successes are exactly once. * - *

Queued jobs are linked per flow and only each flow head is globally ordered. Enqueue and cancellation of a flow - * head are {@code O(log B)} for {@code B} backlogged flows; cancellation of a non-head and completion are 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 {@code O(registeredFlows)}. A canonically triggered exact-tag rebase is + *

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 + * {@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. * @@ -172,6 +178,10 @@ public CloseFlowResult closeFlow(FlowHandle flowHandle) { * tag, and every registered flow finish tag, then admission is retried. The rebase and accepted job commit as one * 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 + * 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. * * @param flowHandle registered flow capability * @param jobId stable non-null identifier, unique among live jobs @@ -211,6 +221,10 @@ public EnqueueResult enqueue(FlowHandle flowHandle, J jobId, P payload, long cos if (computation == null) { return EnqueueResult.Rejected.NUMERIC_LIMIT; } + if (config.cancellationAccounting() == CancellationAccounting.REFUND_CANCELLED_COST + && !refundClosureFitsAfterEnqueue(flow, computation.start, cost, computation.rebase)) { + return EnqueueResult.Rejected.NUMERIC_LIMIT; + } if (computation.rebase != null) { commitRebase(computation.rebase); } @@ -242,11 +256,13 @@ public EnqueueResult enqueue(FlowHandle flowHandle, J jobId, P payload, long cos * {@code NOT_LIVE} alone does not identify the winner or terminal cause. * *

Fairness accounting warning: under - * {@link CancellationAccounting#CHARGE_RESERVED_COST}, cancellation does not reduce the flow's finish history and - * does not recompute tags of its later jobs. The cancelled supplied cost remains a virtual charge until all live - * jobs leave the scheduler and the global busy period ends. Completed-work fairness guarantees therefore do not - * apply to traces containing cancellation. Workloads with frequent deadline or timeout cancellations must account - * for the resulting dispatch delay. + * {@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 + * 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. * * @param handle opaque job capability * @return cancellation outcome at the operation's linearization point @@ -258,7 +274,10 @@ public CancelResult cancel(JobHandle handle) { try { QueuedJob job = queued.get(handle); if (job != null) { - removeQueued(job); + RefundPlan refund = config.cancellationAccounting() + == CancellationAccounting.REFUND_CANCELLED_COST + ? prepareRefund(job) : null; + removeQueued(job, refund); queued.remove(handle); liveById.remove(job.jobId); job.flow.cancelledCost = job.flow.cancelledCost.add(BigInteger.valueOf(job.cost)); @@ -436,6 +455,113 @@ private RebasePlan prepareRebase() throws NumericLimitException { return new RebasePlan<>(flowTags, jobTags); } + private boolean refundClosureFitsAfterEnqueue( + FlowState target, + ExactTag candidateStart, + long candidateCost, + RebasePlan rebase) { + if (rebase == null) { + return refundClosureFits(target, candidateStart, candidateCost, null); + } + for (FlowState flow : registeredFlows.values()) { + if (flow.head == null && flow != target) { + continue; + } + if (!refundClosureFits( + flow, + flow == target ? candidateStart : null, + flow == target ? candidateCost : 0L, + rebase)) { + return false; + } + } + return true; + } + + private boolean refundClosureFits( + FlowState flow, + ExactTag candidateStart, + long candidateCost, + RebasePlan rebase) { + ExactTag base = flow.head == null ? candidateStart : plannedTags(flow.head, rebase).start; + 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; + } + } + if (candidateCost != 0L) { + commonDenominator = lcm(commonDenominator, + reducedIncrementDenominator(candidateCost, flow.weight)); + if (commonDenominator.bitLength() > ExactTag.MAX_PERSISTENT_BITS) { + return false; + } + } + + 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); + } + 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); + } + + private static TagPair plannedTags(QueuedJob job, RebasePlan rebase) { + if (rebase == null) { + return new TagPair(job.start, job.finish); + } + TagPair tags = rebase.jobTags.get(job); + if (tags == null) { + throw new AssertionError("rebase plan must contain every queued job"); + } + return tags; + } + + private RefundPlan prepareRefund(QueuedJob cancelledJob) { + 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); + } catch (NumericLimitException impossible) { + throw new AssertionError("refund-closed queued chain exceeded its reserved numeric budget", impossible); + } + } + private void commitRebase(RebasePlan rebase) { // TreeSet's SortedSet copy path is linear. Every queued start tag receives the same subtraction, so the // existing (start, sequence) order remains valid after the prevalidated tag replacements below. @@ -467,7 +593,7 @@ private void appendQueued(FlowState flow, QueuedJob job) { flow.queuedCount++; } - private void removeQueued(QueuedJob job) { + private void removeQueued(QueuedJob job, RefundPlan refund) { FlowState flow = job.flow; boolean removesHead = job.previous == null; if (removesHead) { @@ -484,6 +610,13 @@ private void removeQueued(QueuedJob job) { 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; + } + flow.lastFinish = refund.lastFinish; + } if (removesHead && flow.head != null) { requireIndexChange(backlogged.add(flow.head), "promoted flow head was already indexed"); } @@ -633,6 +766,16 @@ 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/ApiVocabularyTest.java b/sfqd-core/src/test/java/io/github/pzhin/sfqd/ApiVocabularyTest.java index a460d64..e280ab6 100644 --- a/sfqd-core/src/test/java/io/github/pzhin/sfqd/ApiVocabularyTest.java +++ b/sfqd-core/src/test/java/io/github/pzhin/sfqd/ApiVocabularyTest.java @@ -42,11 +42,14 @@ void configMakesCancellationAccountingExplicit() { SchedulerConfig defaultConfig = new SchedulerConfig(8, 10, 20); SchedulerConfig explicitConfig = new SchedulerConfig( 8, 10, 20, CancellationAccounting.CHARGE_RESERVED_COST); + SchedulerConfig refundConfig = new SchedulerConfig( + 8, 10, 20, CancellationAccounting.REFUND_CANCELLED_COST); assertEquals(CancellationAccounting.CHARGE_RESERVED_COST, defaultConfig.cancellationAccounting()); assertEquals(WeightDomain.unrestricted(), defaultConfig.weightDomain()); assertEquals(defaultConfig, explicitConfig); - assertEquals(1, CancellationAccounting.values().length); + assertEquals(CancellationAccounting.REFUND_CANCELLED_COST, refundConfig.cancellationAccounting()); + assertEquals(2, CancellationAccounting.values().length); } @Test diff --git a/sfqd-core/src/test/java/io/github/pzhin/sfqd/ReferenceAdmissionTest.java b/sfqd-core/src/test/java/io/github/pzhin/sfqd/ReferenceAdmissionTest.java index 2018dfe..ab5e662 100644 --- a/sfqd-core/src/test/java/io/github/pzhin/sfqd/ReferenceAdmissionTest.java +++ b/sfqd-core/src/test/java/io/github/pzhin/sfqd/ReferenceAdmissionTest.java @@ -6,6 +6,7 @@ import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertThrows; +import java.util.List; import org.junit.jupiter.api.Test; final class ReferenceAdmissionTest { @@ -119,6 +120,49 @@ void closedAndReregisteredFlowGetsANewCapability() { model.enqueue(oldHandle, "stale", new Object(), 1L)); } + @Test + void reachableRefundClosureTraceRejectsProspectiveCancellationOverflowWithoutMutation() { + ReferenceScheduler model = + new ReferenceScheduler<>(RefundNumericRegressionTrace.config()); + FlowHandle anchor = registered(model.registerFlow("anchor", 1L)); + JobHandle anchorJob = accepted(model.enqueue(anchor, "anchor", new Object(), 1L)); + assertSame(anchorJob, model.dispatchUpTo(1).get(0).jobHandle()); + int enqueueAttempts = 1; + + int builderId = 0; + for (long weight : RefundNumericRegressionTrace.builderWeights()) { + FlowHandle builder = registered(model.registerFlow("builder-" + builderId, weight)); + JobHandle firstBuilder = accepted(model.enqueue( + builder, "builder-" + builderId + "-first", new Object(), 1L)); + JobHandle secondBuilder = accepted(model.enqueue( + builder, "builder-" + builderId + "-second", new Object(), 1L)); + enqueueAttempts += 2; + assertSame(firstBuilder, model.dispatchUpTo(1).get(0).jobHandle()); + assertEquals(CompletionResult.COMPLETED, model.complete(firstBuilder)); + assertSame(secondBuilder, model.dispatchUpTo(1).get(0).jobHandle()); + assertEquals(CompletionResult.COMPLETED, model.complete(secondBuilder)); + builderId++; + } + + FlowHandle target = registered(model.registerFlow("target", 6L)); + JobHandle first = accepted(model.enqueue(target, "first", new Object(), 3L)); + enqueueAttempts++; + SchedulerSnapshot before = model.snapshot(); + List queuedBefore = model.queuedHandles(); + assertEquals(1, before.runningJobs()); + assertEquals(1, before.queuedJobs()); + assertEquals(132L, before.acceptedTotal()); + + assertEquals(EnqueueResult.Rejected.NUMERIC_LIMIT, + model.enqueue(target, "rejected", new Object(), 1L)); + enqueueAttempts++; + assertEquals(RefundNumericRegressionTrace.ENQUEUE_ATTEMPTS, enqueueAttempts); + assertEquals(before, model.snapshot()); + assertEquals(queuedBefore, model.queuedHandles()); + assertEquals(List.of(first), queuedBefore); + assertEquals(CancelResult.CANCELLED, model.cancel(first)); + } + private static ReferenceScheduler model(int depth, int maxFlows, int maxJobs) { return new ReferenceScheduler<>(new SchedulerConfig(depth, maxFlows, maxJobs)); } diff --git a/sfqd-core/src/test/java/io/github/pzhin/sfqd/ReferenceCancellationTest.java b/sfqd-core/src/test/java/io/github/pzhin/sfqd/ReferenceCancellationTest.java index 03132db..2e3be21 100644 --- a/sfqd-core/src/test/java/io/github/pzhin/sfqd/ReferenceCancellationTest.java +++ b/sfqd-core/src/test/java/io/github/pzhin/sfqd/ReferenceCancellationTest.java @@ -4,6 +4,7 @@ import static org.junit.jupiter.api.Assertions.assertInstanceOf; import static org.junit.jupiter.api.Assertions.assertSame; +import java.util.List; import org.junit.jupiter.api.Test; final class ReferenceCancellationTest { @@ -113,10 +114,113 @@ void completionCounterPreservesBothConservationEquations() { assertEquals(0, snapshot.runningJobs()); } + @Test + void refundRecomputesHeadSuffixAndCanChangeGlobalOrderWithStableTies() { + ReferenceScheduler refund = refundModel(3, 2, 5); + FlowHandle refundA = registered(refund.registerFlow("a", 2L)); + FlowHandle refundB = registered(refund.registerFlow("b", 1L)); + JobHandle cancelled = accepted(refund.enqueue(refundA, "cancelled", new Object(), 4L)); + JobHandle successor = accepted(refund.enqueue(refundA, "successor", new Object(), 2L)); + JobHandle secondSuccessor = accepted(refund.enqueue(refundA, "second-successor", new Object(), 2L)); + JobHandle competitor = accepted(refund.enqueue(refundB, "competitor", new Object(), 1L)); + + assertEquals(CancelResult.CANCELLED, refund.cancel(cancelled)); + assertEquals(ExactRational.ZERO, refund.startTag(successor)); + assertEquals(ExactRational.ONE, refund.finishTag(successor)); + assertEquals(ExactRational.ONE, refund.startTag(secondSuccessor)); + assertEquals(ExactRational.of(2L, 1L), refund.finishTag(secondSuccessor)); + assertEquals(List.of(successor, competitor, secondSuccessor), refund.queuedHandles()); + + ReferenceScheduler charged = model(3, 2, 4); + FlowHandle chargedA = registered(charged.registerFlow("a", 2L)); + FlowHandle chargedB = registered(charged.registerFlow("b", 1L)); + JobHandle chargedCancelled = accepted(charged.enqueue(chargedA, "cancelled", new Object(), 4L)); + JobHandle chargedSuccessor = accepted(charged.enqueue(chargedA, "successor", new Object(), 2L)); + JobHandle chargedCompetitor = accepted(charged.enqueue(chargedB, "competitor", new Object(), 1L)); + + assertEquals(CancelResult.CANCELLED, charged.cancel(chargedCancelled)); + assertEquals(ExactRational.of(2L, 1L), charged.startTag(chargedSuccessor)); + assertEquals(List.of(chargedCompetitor, chargedSuccessor), charged.queuedHandles()); + } + + @Test + void refundOfMiddleAndTailPreservesFractionalOrderAndOtherFlows() { + ReferenceScheduler model = refundModel(3, 2, 6); + FlowHandle target = registered(model.registerFlow("target", 6L)); + FlowHandle other = registered(model.registerFlow("other", 5L)); + JobHandle first = accepted(model.enqueue(target, "first", new Object(), 1L)); + JobHandle middle = accepted(model.enqueue(target, "middle", new Object(), 2L)); + JobHandle tail = accepted(model.enqueue(target, "tail", new Object(), 3L)); + JobHandle untouched = accepted(model.enqueue(other, "untouched", new Object(), 2L)); + ExactRational otherStart = model.startTag(untouched); + ExactRational otherFinish = model.finishTag(untouched); + + assertEquals(CancelResult.CANCELLED, model.cancel(middle)); + assertEquals(ExactRational.of(1L, 6L), model.startTag(tail)); + assertEquals(ExactRational.of(2L, 3L), model.finishTag(tail)); + assertEquals(CancelResult.CANCELLED, model.cancel(tail)); + JobHandle next = accepted(model.enqueue(target, "next", new Object(), 1L)); + + assertEquals(ExactRational.of(1L, 6L), model.startTag(next)); + assertEquals(otherStart, model.startTag(untouched)); + assertEquals(otherFinish, model.finishTag(untouched)); + assertEquals(List.of(first, next), model.queuedHandles().stream() + .filter(handle -> handle.equals(first) || handle.equals(next)).toList()); + assertEquals(new FlowSnapshot(2, 0, java.math.BigInteger.valueOf(7L), + java.math.BigInteger.ZERO, java.math.BigInteger.valueOf(5L), java.math.BigInteger.ZERO), + model.snapshot(target).orElseThrow()); + assertEquals(5L, model.snapshot().acceptedTotal()); + assertEquals(2L, model.snapshot().cancelledTotal()); + } + + @Test + void refundKeepsDispatchedDebtAndGlobalIdleStillResetsHistory() { + ReferenceScheduler model = refundModel(2, 2, 5); + FlowHandle target = registered(model.registerFlow("target", 2L)); + FlowHandle progress = registered(model.registerFlow("progress", 1L)); + JobHandle running = accepted(model.enqueue(target, "running", new Object(), 2L)); + JobHandle cancelled = accepted(model.enqueue(target, "cancelled", new Object(), 4L)); + assertSame(running, model.dispatchUpTo(1).get(0).jobHandle()); + + assertEquals(CancelResult.CANCELLED, model.cancel(cancelled)); + JobHandle afterRefund = accepted(model.enqueue(target, "after-refund", new Object(), 2L)); + assertEquals(ExactRational.ONE, model.startTag(afterRefund)); + assertEquals(CancelResult.CANCELLED, model.cancel(afterRefund)); + assertEquals(CompletionResult.COMPLETED, model.complete(running)); + + JobHandle reset = accepted(model.enqueue(progress, "reset", new Object(), 1L)); + assertEquals(ExactRational.ZERO, model.startTag(reset)); + } + + @Test + void refundOfOnlyQueuedJobKeepsDispatchedDebtForCloseFlow() { + ReferenceScheduler model = refundModel(2, 2, 4); + FlowHandle target = registered(model.registerFlow("target", 1L)); + FlowHandle progress = registered(model.registerFlow("progress", 1L)); + JobHandle running = accepted(model.enqueue(target, "running", new Object(), 1L)); + JobHandle cancelled = accepted(model.enqueue(target, "cancelled", new Object(), 5L)); + assertSame(running, model.dispatchUpTo(1).get(0).jobHandle()); + JobHandle progressFirst = accepted(model.enqueue(progress, "progress-1", new Object(), 1L)); + JobHandle progressSecond = accepted(model.enqueue(progress, "progress-2", new Object(), 1L)); + + assertEquals(CancelResult.CANCELLED, model.cancel(cancelled)); + assertEquals(CompletionResult.COMPLETED, model.complete(running)); + assertEquals(CloseFlowResult.FAIRNESS_DEBT_ACTIVE, model.closeFlow(target)); + assertSame(progressFirst, model.dispatchUpTo(1).get(0).jobHandle()); + assertEquals(CompletionResult.COMPLETED, model.complete(progressFirst)); + assertSame(progressSecond, model.dispatchUpTo(1).get(0).jobHandle()); + assertEquals(CloseFlowResult.CLOSED, model.closeFlow(target)); + } + private static ReferenceScheduler model(int depth, int maxFlows, int maxJobs) { return new ReferenceScheduler<>(new SchedulerConfig(depth, maxFlows, maxJobs)); } + private static ReferenceScheduler refundModel(int depth, int maxFlows, int maxJobs) { + return new ReferenceScheduler<>(new SchedulerConfig( + depth, maxFlows, maxJobs, CancellationAccounting.REFUND_CANCELLED_COST)); + } + private static FlowHandle registered(RegisterFlowResult result) { return assertInstanceOf(RegisterFlowResult.Registered.class, result).flowHandle(); } diff --git a/sfqd-core/src/test/java/io/github/pzhin/sfqd/ReferenceLifecycleProperties.java b/sfqd-core/src/test/java/io/github/pzhin/sfqd/ReferenceLifecycleProperties.java index 44dda10..36f5123 100644 --- a/sfqd-core/src/test/java/io/github/pzhin/sfqd/ReferenceLifecycleProperties.java +++ b/sfqd-core/src/test/java/io/github/pzhin/sfqd/ReferenceLifecycleProperties.java @@ -22,9 +22,10 @@ final class ReferenceLifecycleProperties { void oneFlowDispatchIsWorkConservingAndConservesLifecycleCounts( @ForAll @IntRange(min = 1, max = 16) int depth, @ForAll @IntRange(min = 1, max = 64) int jobs, - @ForAll @IntRange(min = 0, max = 16) int requested) { + @ForAll @IntRange(min = 0, max = 16) int requested, + @ForAll("cancellationPolicies") CancellationAccounting policy) { ReferenceScheduler model = - new ReferenceScheduler<>(new SchedulerConfig(depth, 1, Math.max(depth, jobs))); + new ReferenceScheduler<>(new SchedulerConfig(depth, 1, Math.max(depth, jobs), policy)); FlowHandle flow = assertInstanceOf( RegisterFlowResult.Registered.class, model.registerFlow("flow", 1L)).flowHandle(); for (int index = 0; index < jobs; index++) { @@ -44,8 +45,9 @@ void oneFlowDispatchIsWorkConservingAndConservesLifecycleCounts( @Property(tries = 200) void mixedCommandTraceConservesStateAfterEveryEvent( - @ForAll("commandTraces") List commands) { - LifecycleHarness harness = new LifecycleHarness(); + @ForAll("commandTraces") List commands, + @ForAll("cancellationPolicies") CancellationAccounting policy) { + LifecycleHarness harness = new LifecycleHarness(policy); for (int command : commands) { harness.apply(command); harness.assertConservation(); @@ -65,6 +67,11 @@ Arbitrary> commandTraces() { }); } + @Provide + Arbitrary cancellationPolicies() { + return Arbitraries.of(CancellationAccounting.values()); + } + private enum JobPhase { QUEUED, RUNNING @@ -85,8 +92,7 @@ private static final class LifecycleHarness { private static final String FIRST_FLOW = "first"; private static final String SECOND_FLOW = "second"; - private final ReferenceScheduler model = - new ReferenceScheduler<>(new SchedulerConfig(2, 2, 8)); + private final ReferenceScheduler model; private final FlowHandle foreignFlow = new FlowHandle(new OwnerToken(), 1L); private final JobHandle foreignJob = new JobHandle(new OwnerToken(), 1L); private final Map flows = new LinkedHashMap<>(); @@ -98,6 +104,10 @@ private static final class LifecycleHarness { private long cancelled; private long completed; + private LifecycleHarness(CancellationAccounting policy) { + model = new ReferenceScheduler<>(new SchedulerConfig(2, 2, 8, policy)); + } + private void apply(int command) { switch (command) { case 0 -> register(FIRST_FLOW); diff --git a/sfqd-core/src/test/java/io/github/pzhin/sfqd/ReferenceScheduler.java b/sfqd-core/src/test/java/io/github/pzhin/sfqd/ReferenceScheduler.java index 6d571c4..089a0a5 100644 --- a/sfqd-core/src/test/java/io/github/pzhin/sfqd/ReferenceScheduler.java +++ b/sfqd-core/src/test/java/io/github/pzhin/sfqd/ReferenceScheduler.java @@ -100,6 +100,10 @@ EnqueueResult enqueue(FlowHandle flowHandle, J jobId, P payload, long cost) { } ExactRational start = virtualTime.max(flow.lastFinish); ExactRational finish = start.add(ExactRational.of(cost, flow.weight)); + if (config.cancellationAccounting() == CancellationAccounting.REFUND_CANCELLED_COST + && !refundClosed(flowHandle, flow, start, cost)) { + return EnqueueResult.Rejected.NUMERIC_LIMIT; + } long sequence = lastJobSequence + 1L; JobHandle handle = new JobHandle(ownerToken, sequence); QueuedJob job = new QueuedJob<>( @@ -141,11 +145,24 @@ List> dispatchUpTo(int capacity) { CancelResult cancel(JobHandle handle) { Objects.requireNonNull(handle, "handle"); - QueuedJob job = queued.remove(handle); + QueuedJob job = queued.get(handle); if (job != null) { priority.remove(handle); - liveById.remove(job.jobId); FlowState flow = registeredFlows.get(job.flowHandle); + if (config.cancellationAccounting() == CancellationAccounting.REFUND_CANCELLED_COST) { + ExactRational nextFinish = virtualTime.max(job.start); + for (QueuedJob later : queued.values()) { + if (later.flowHandle.equals(job.flowHandle) && later.sequence > job.sequence) { + later.start = nextFinish; + later.finish = nextFinish.add(ExactRational.of(later.cost, flow.weight)); + nextFinish = later.finish; + } + } + flow.lastFinish = nextFinish; + priority.sort(this::compareQueued); + } + queued.remove(handle); + liveById.remove(job.jobId); flow.queuedCount--; flow.cancelledCost = flow.cancelledCost.add(BigInteger.valueOf(job.cost)); cancelled++; @@ -228,6 +245,63 @@ private QueuedJob queuedJob(JobHandle handle) { return Objects.requireNonNull(queued.get(handle), "queued job"); } + private boolean refundClosed( + FlowHandle flowHandle, FlowState flow, ExactRational candidateStart, long candidateCost) { + ExactRational base = candidateStart; + for (QueuedJob job : queued.values()) { + if (job.flowHandle.equals(flowHandle)) { + base = job.start; + break; + } + } + BigInteger commonDenominator = base.denominator(); + for (QueuedJob job : queued.values()) { + if (job.flowHandle.equals(flowHandle)) { + commonDenominator = lcm(commonDenominator, + reducedIncrementDenominator(job.cost, flow.weight)); + if (commonDenominator.bitLength() > ExactTag.MAX_PERSISTENT_BITS) { + return false; + } + } + } + commonDenominator = lcm(commonDenominator, + reducedIncrementDenominator(candidateCost, flow.weight)); + if (commonDenominator.bitLength() > ExactTag.MAX_PERSISTENT_BITS) { + return false; + } + BigInteger accumulatedNumerator = base.numerator() + .multiply(commonDenominator.divide(base.denominator())); + for (QueuedJob job : queued.values()) { + if (job.flowHandle.equals(flowHandle)) { + accumulatedNumerator = addIncrementNumerator( + accumulatedNumerator, commonDenominator, job.cost, flow.weight); + } + } + accumulatedNumerator = addIncrementNumerator( + accumulatedNumerator, commonDenominator, candidateCost, flow.weight); + 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); + } + private int compareQueued(JobHandle first, JobHandle second) { QueuedJob firstJob = queuedJob(first); QueuedJob secondJob = queuedJob(second); @@ -296,8 +370,8 @@ private static final class QueuedJob { private final F flowId; private final P payload; private final long cost; - private final ExactRational start; - private final ExactRational finish; + private ExactRational start; + private ExactRational finish; private final long sequence; private QueuedJob( diff --git a/sfqd-core/src/test/java/io/github/pzhin/sfqd/RefundNumericRegressionTrace.java b/sfqd-core/src/test/java/io/github/pzhin/sfqd/RefundNumericRegressionTrace.java new file mode 100644 index 0000000..86386fd --- /dev/null +++ b/sfqd-core/src/test/java/io/github/pzhin/sfqd/RefundNumericRegressionTrace.java @@ -0,0 +1,47 @@ +package io.github.pzhin.sfqd; + +final class RefundNumericRegressionTrace { + static final int ENQUEUE_ATTEMPTS = 133; + + // This frozen public-operation trace grows the busy-period virtual-time + // denominator to the refund admission boundary without seeding internals. + private static final long[] BUILDER_WEIGHTS = { + 9223372036854775783L, 9223372036854775643L, 9223372036854775549L, + 9223372036854775507L, 9223372036854775433L, 9223372036854775421L, + 9223372036854775417L, 9223372036854775399L, 9223372036854775351L, + 9223372036854775337L, 9223372036854775291L, 9223372036854775279L, + 9223372036854775259L, 9223372036854775181L, 9223372036854775159L, + 9223372036854775139L, 9223372036854775097L, 9223372036854775073L, + 9223372036854775057L, 9223372036854774959L, 9223372036854774937L, + 9223372036854774917L, 9223372036854774893L, 9223372036854774797L, + 9223372036854774739L, 9223372036854774713L, 9223372036854774679L, + 9223372036854774629L, 9223372036854774587L, 9223372036854774571L, + 9223372036854774559L, 9223372036854774511L, 9223372036854774509L, + 9223372036854774499L, 9223372036854774451L, 9223372036854774413L, + 9223372036854774341L, 9223372036854774319L, 9223372036854774307L, + 9223372036854774277L, 9223372036854774257L, 9223372036854774247L, + 9223372036854774233L, 9223372036854774199L, 9223372036854774179L, + 9223372036854774173L, 9223372036854774053L, 9223372036854773999L, + 9223372036854773977L, 9223372036854773953L, 9223372036854773899L, + 9223372036854773867L, 9223372036854773783L, 9223372036854773639L, + 9223372036854773561L, 9223372036854773557L, 9223372036854773519L, + 9223372036854773507L, 9223372036854773489L, 9223372036854773477L, + 9223372036854773443L, 9223372036854773429L, 9223372036854773407L, + 9223372036854773353L, 3074457345618286127L + }; + + private RefundNumericRegressionTrace() { + } + + static SchedulerConfig config() { + return new SchedulerConfig( + 2, + BUILDER_WEIGHTS.length + 2, + 4, + CancellationAccounting.REFUND_CANCELLED_COST); + } + + static long[] builderWeights() { + return BUILDER_WEIGHTS.clone(); + } +} diff --git a/sfqd-core/src/test/java/io/github/pzhin/sfqd/SfqdDeterministicConcurrencyTest.java b/sfqd-core/src/test/java/io/github/pzhin/sfqd/SfqdDeterministicConcurrencyTest.java index 0378e45..7ff69d4 100644 --- a/sfqd-core/src/test/java/io/github/pzhin/sfqd/SfqdDeterministicConcurrencyTest.java +++ b/sfqd-core/src/test/java/io/github/pzhin/sfqd/SfqdDeterministicConcurrencyTest.java @@ -240,7 +240,17 @@ void depthTwoConcurrentSingleDispatchesUseDisjointCapacity() { @Test void selectedCancelDispatchRaceReportsTheWinnerInCombinedHistory() { - SfqdScheduler scheduler = scheduler(1, 1, 1); + assertSelectedCancelDispatchRace(CancellationAccounting.CHARGE_RESERVED_COST); + } + + @Test + void selectedRefundCancelDispatchRaceReportsTheWinnerInCombinedHistory() { + assertSelectedCancelDispatchRace(CancellationAccounting.REFUND_CANCELLED_COST); + } + + private static void assertSelectedCancelDispatchRace(CancellationAccounting policy) { + SfqdScheduler scheduler = + new SfqdScheduler<>(new SchedulerConfig(1, 1, 1, policy)); FlowHandle flow = registered(scheduler.registerFlow("flow", 1L)); JobHandle victim = accepted(scheduler.enqueue(flow, "victim", "p", 1L)); diff --git a/sfqd-core/src/test/java/io/github/pzhin/sfqd/SfqdDifferentialProperties.java b/sfqd-core/src/test/java/io/github/pzhin/sfqd/SfqdDifferentialProperties.java index 9a0ec17..e20cef4 100644 --- a/sfqd-core/src/test/java/io/github/pzhin/sfqd/SfqdDifferentialProperties.java +++ b/sfqd-core/src/test/java/io/github/pzhin/sfqd/SfqdDifferentialProperties.java @@ -22,12 +22,14 @@ class SfqdDifferentialProperties { void boundedProductionMatchesUnboundedReferenceForEverySmallTrace( @ForAll("depths") int depth, @ForAll @Size(min = 50, max = 180) List<@IntRange(min = -10_000, max = 10_000) Integer> commands) { - Harness harness = new Harness(depth); + for (CancellationAccounting policy : CancellationAccounting.values()) { + Harness harness = new Harness(depth, policy); - for (int event = 0; event < commands.size(); event++) { - harness.apply(commands.get(event), event); - assertEquals(harness.reference.snapshot(), harness.production.snapshot()); - harness.assertFlowSnapshotsEqual(); + for (int event = 0; event < commands.size(); event++) { + harness.apply(commands.get(event), event); + assertEquals(harness.reference.snapshot(), harness.production.snapshot()); + harness.assertFlowSnapshotsEqual(); + } } } @@ -69,8 +71,8 @@ private static final class Harness { private final List jobs = new ArrayList<>(); private final Map referenceToProduction = new HashMap<>(); - private Harness(int depth) { - config = new SchedulerConfig(depth, 8, 24); + private Harness(int depth, CancellationAccounting policy) { + config = new SchedulerConfig(depth, 8, 24, policy); reference = new ReferenceScheduler<>(config); production = new SfqdScheduler<>(config); ReferenceScheduler foreignReference = new ReferenceScheduler<>(config); 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 f27818e..7ce6997 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 @@ -104,6 +104,32 @@ void failedRebaseLeavesEveryHiddenNumericAndOrderingFieldUnchanged() assertEquals(before, NumericProbe.capture(fixture.scheduler)); } + @Test + void rebaseThatWouldBreakRefundClosureRejectsCompleteEnqueueWithoutMutation() + throws NumericLimitException, ReflectiveOperationException { + SfqdScheduler scheduler = new SfqdScheduler<>(new SchedulerConfig( + 1, 2, 3, CancellationAccounting.REFUND_CANCELLED_COST)); + FlowHandle queuedFlow = registered(scheduler.registerFlow("queued", 6L)); + FlowHandle targetFlow = registered(scheduler.registerFlow("target", 1L)); + scheduler.enqueue(queuedFlow, "first", "p", 3L); + scheduler.enqueue(queuedFlow, "second", "p", 1L); + + BigInteger limit = BigInteger.ONE.shiftLeft(ExactTag.MAX_PERSISTENT_BITS); + BigInteger denominator = limit.divide(BigInteger.valueOf(6L)).nextProbablePrime(); + BigInteger targetNumerator = limit.subtract(denominator.divide(BigInteger.TWO)); + BigInteger virtualNumerator = targetNumerator.subtract(BigInteger.ONE); + ExactTag virtualTime = ExactTag.fromComponents(virtualNumerator, denominator); + ExactTag targetFinish = ExactTag.fromComponents(targetNumerator, denominator); + NumericProbe.seedRefundRebaseFixture( + scheduler, queuedFlow, targetFlow, virtualTime, targetFinish, + ExactTag.fromCostAndWeight(10L, 1L)); + NumericState before = NumericProbe.capture(scheduler); + + assertEquals(EnqueueResult.Rejected.NUMERIC_LIMIT, + scheduler.enqueue(targetFlow, "rejected", "p", 1L)); + assertEquals(before, NumericProbe.capture(scheduler)); + } + @Test void sequenceExhaustionPrecedesNumericLimitForTheSameCandidate() throws NumericLimitException, ReflectiveOperationException { @@ -353,6 +379,29 @@ static void setLongForDeepState(Object target, String name, long value) setLong(target, name, value); } + private static void seedRefundRebaseFixture( + SfqdScheduler scheduler, + FlowHandle queuedFlow, + FlowHandle targetFlow, + ExactTag virtualTime, + ExactTag targetFinish, + ExactTag queuedBase) throws ReflectiveOperationException, NumericLimitException { + set(scheduler, "virtualTime", virtualTime); + Map registered = map(scheduler, "registeredFlows"); + Object queued = registered.get(queuedFlow); + Object first = get(queued, "head"); + Object second = get(first, "next"); + ExactTag half = ExactTag.fromCostAndWeight(3L, 6L); + ExactTag sixth = ExactTag.fromCostAndWeight(1L, 6L); + ExactTag firstFinish = queuedBase.add(half); + set(first, "start", queuedBase); + set(first, "finish", firstFinish); + set(second, "start", firstFinish); + set(second, "finish", firstFinish.add(sixth)); + set(queued, "lastFinish", firstFinish.add(sixth)); + set(registered.get(targetFlow), "lastFinish", targetFinish); + } + private static void shiftQueuedAndFlowState( SfqdScheduler scheduler, ExactTag base, diff --git a/sfqd-core/src/test/java/io/github/pzhin/sfqd/SfqdRefundCancellationTest.java b/sfqd-core/src/test/java/io/github/pzhin/sfqd/SfqdRefundCancellationTest.java new file mode 100644 index 0000000..f1d42e1 --- /dev/null +++ b/sfqd-core/src/test/java/io/github/pzhin/sfqd/SfqdRefundCancellationTest.java @@ -0,0 +1,180 @@ +package io.github.pzhin.sfqd; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertSame; + +import java.math.BigInteger; +import java.util.List; +import org.junit.jupiter.api.Test; + +final class SfqdRefundCancellationTest { + @Test + void refundOfHeadChangesGlobalOrderWhileDefaultRemainsChargeReserved() { + SfqdScheduler refund = scheduler(3, 2, 5, + CancellationAccounting.REFUND_CANCELLED_COST); + FlowHandle refundA = registered(refund.registerFlow("a", 2L)); + FlowHandle refundB = registered(refund.registerFlow("b", 1L)); + JobHandle cancelled = accepted(refund.enqueue(refundA, "cancelled", "p", 4L)); + refund.enqueue(refundA, "successor", "p", 2L); + refund.enqueue(refundA, "second-successor", "p", 2L); + refund.enqueue(refundB, "competitor", "p", 1L); + + assertEquals(CancelResult.CANCELLED, refund.cancel(cancelled)); + assertEquals(List.of("successor", "competitor", "second-successor"), ids(refund.dispatchUpTo(3))); + + SfqdScheduler charged = new SfqdScheduler<>(new SchedulerConfig(3, 2, 4)); + FlowHandle chargedA = registered(charged.registerFlow("a", 2L)); + FlowHandle chargedB = registered(charged.registerFlow("b", 1L)); + JobHandle chargedCancellation = accepted(charged.enqueue(chargedA, "cancelled", "p", 4L)); + charged.enqueue(chargedA, "successor", "p", 2L); + charged.enqueue(chargedB, "competitor", "p", 1L); + + assertEquals(CancelResult.CANCELLED, charged.cancel(chargedCancellation)); + assertEquals(List.of("competitor", "successor"), ids(charged.dispatchUpTo(3))); + assertEquals(CancellationAccounting.CHARGE_RESERVED_COST, + new SchedulerConfig(1, 1, 1).cancellationAccounting()); + } + + @Test + void refundOfMiddleAndTailPreservesOtherFlowAndCostCounters() { + SfqdScheduler scheduler = scheduler(3, 2, 6, + CancellationAccounting.REFUND_CANCELLED_COST); + FlowHandle target = registered(scheduler.registerFlow("target", 6L)); + FlowHandle other = registered(scheduler.registerFlow("other", 5L)); + scheduler.enqueue(target, "first", "p", 1L); + JobHandle middle = accepted(scheduler.enqueue(target, "middle", "p", 2L)); + JobHandle tail = accepted(scheduler.enqueue(target, "tail", "p", 3L)); + scheduler.enqueue(other, "other", "p", 2L); + + assertEquals(CancelResult.CANCELLED, scheduler.cancel(middle)); + assertEquals(CancelResult.CANCELLED, scheduler.cancel(tail)); + scheduler.enqueue(target, "next", "p", 1L); + + assertEquals(List.of("first", "other", "next"), ids(scheduler.dispatchUpTo(3))); + assertEquals(new FlowSnapshot( + 0, + 2, + BigInteger.valueOf(7L), + BigInteger.valueOf(2L), + BigInteger.valueOf(5L), + BigInteger.valueOf(2L)), scheduler.snapshot(target).orElseThrow()); + assertEquals(new FlowSnapshot( + 0, + 1, + BigInteger.valueOf(2L), + BigInteger.valueOf(2L), + BigInteger.ZERO, + BigInteger.valueOf(2L)), scheduler.snapshot(other).orElseThrow()); + assertEquals(5L, scheduler.snapshot().acceptedTotal()); + assertEquals(2L, scheduler.snapshot().cancelledTotal()); + } + + @Test + void refundPreservesRunningDebtAndIdleReset() { + SfqdScheduler scheduler = scheduler(2, 2, 5, + CancellationAccounting.REFUND_CANCELLED_COST); + FlowHandle target = registered(scheduler.registerFlow("target", 2L)); + FlowHandle progress = registered(scheduler.registerFlow("progress", 1L)); + JobHandle running = accepted(scheduler.enqueue(target, "running", "p", 2L)); + JobHandle cancelled = accepted(scheduler.enqueue(target, "cancelled", "p", 4L)); + assertSame(running, scheduler.dispatchUpTo(1).get(0).jobHandle()); + + assertEquals(CancelResult.CANCELLED, scheduler.cancel(cancelled)); + JobHandle afterRefund = accepted(scheduler.enqueue(target, "after-refund", "p", 2L)); + JobHandle progressJob = accepted(scheduler.enqueue(progress, "progress", "p", 1L)); + assertEquals(List.of("progress"), ids(scheduler.dispatchUpTo(1))); + assertEquals(CancelResult.CANCELLED, scheduler.cancel(afterRefund)); + assertEquals(CompletionResult.COMPLETED, scheduler.complete(running)); + assertEquals(CompletionResult.COMPLETED, scheduler.complete(progressJob)); + + scheduler.enqueue(target, "after-idle", "p", 1L); + scheduler.enqueue(progress, "tie", "p", 1L); + assertEquals(List.of("after-idle", "tie"), ids(scheduler.dispatchUpTo(2))); + } + + @Test + void reachableRefundClosureTraceRejectsSecondAdmissionAsAtomicNoOp() + throws ReflectiveOperationException { + SfqdScheduler scheduler = + new SfqdScheduler<>(RefundNumericRegressionTrace.config()); + FlowHandle anchor = registered(scheduler.registerFlow("anchor", 1L)); + JobHandle anchorJob = accepted(scheduler.enqueue(anchor, "anchor", "p", 1L)); + assertSame(anchorJob, scheduler.dispatchUpTo(1).get(0).jobHandle()); + int enqueueAttempts = 1; + + int builderId = 0; + for (long weight : RefundNumericRegressionTrace.builderWeights()) { + FlowHandle builder = registered(scheduler.registerFlow("builder-" + builderId, weight)); + JobHandle firstBuilder = accepted(scheduler.enqueue( + builder, "builder-" + builderId + "-first", "p", 1L)); + JobHandle secondBuilder = accepted(scheduler.enqueue( + builder, "builder-" + builderId + "-second", "p", 1L)); + enqueueAttempts += 2; + assertSame(firstBuilder, scheduler.dispatchUpTo(1).get(0).jobHandle()); + assertEquals(CompletionResult.COMPLETED, scheduler.complete(firstBuilder)); + assertSame(secondBuilder, scheduler.dispatchUpTo(1).get(0).jobHandle()); + assertEquals(CompletionResult.COMPLETED, scheduler.complete(secondBuilder)); + builderId++; + } + + FlowHandle target = registered(scheduler.registerFlow("target", 6L)); + JobHandle first = accepted(scheduler.enqueue(target, "first", "p", 3L)); + enqueueAttempts++; + SchedulerSnapshot publicBefore = scheduler.snapshot(); + Object before = SfqdNumericBoundaryTest.NumericProbe.captureDeepState(scheduler); + assertEquals(1, publicBefore.runningJobs()); + assertEquals(1, publicBefore.queuedJobs()); + assertEquals(132L, publicBefore.acceptedTotal()); + + assertEquals(EnqueueResult.Rejected.NUMERIC_LIMIT, + scheduler.enqueue(target, "rejected", "p", 1L)); + enqueueAttempts++; + assertEquals(RefundNumericRegressionTrace.ENQUEUE_ATTEMPTS, enqueueAttempts); + assertEquals(publicBefore, scheduler.snapshot()); + assertEquals(before, SfqdNumericBoundaryTest.NumericProbe.captureDeepState(scheduler)); + assertEquals(CancelResult.CANCELLED, scheduler.cancel(first)); + } + + @Test + void failedRefundCancellationsAreAtomicNoOps() throws ReflectiveOperationException { + SfqdScheduler scheduler = scheduler(1, 1, 2, + CancellationAccounting.REFUND_CANCELLED_COST); + FlowHandle flow = registered(scheduler.registerFlow("flow", 1L)); + JobHandle running = accepted(scheduler.enqueue(flow, "running", "p", 1L)); + scheduler.dispatchUpTo(1); + JobHandle queued = accepted(scheduler.enqueue(flow, "queued", "p", 1L)); + SfqdScheduler foreign = scheduler(1, 1, 1, + CancellationAccounting.REFUND_CANCELLED_COST); + FlowHandle foreignFlow = registered(foreign.registerFlow("foreign", 1L)); + JobHandle foreignJob = accepted(foreign.enqueue(foreignFlow, "foreign", "p", 1L)); + + Object beforeRunning = SfqdNumericBoundaryTest.NumericProbe.captureDeepState(scheduler); + assertEquals(CancelResult.TOO_LATE_ALREADY_DISPATCHED, scheduler.cancel(running)); + assertEquals(beforeRunning, SfqdNumericBoundaryTest.NumericProbe.captureDeepState(scheduler)); + assertEquals(CancelResult.NOT_LIVE, scheduler.cancel(foreignJob)); + assertEquals(beforeRunning, SfqdNumericBoundaryTest.NumericProbe.captureDeepState(scheduler)); + + assertEquals(CancelResult.CANCELLED, scheduler.cancel(queued)); + Object afterSuccess = SfqdNumericBoundaryTest.NumericProbe.captureDeepState(scheduler); + assertEquals(CancelResult.NOT_LIVE, scheduler.cancel(queued)); + assertEquals(afterSuccess, SfqdNumericBoundaryTest.NumericProbe.captureDeepState(scheduler)); + } + + private static SfqdScheduler scheduler( + int depth, int flows, int jobs, CancellationAccounting policy) { + return new SfqdScheduler<>(new SchedulerConfig(depth, flows, jobs, policy)); + } + + private static List ids(List> dispatches) { + return dispatches.stream().map(Dispatch::jobId).toList(); + } + + 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-jcstress/src/main/java/io/github/pzhin/sfqd/jcstress/RefundUnselectedCancelDispatchStress.java b/sfqd-jcstress/src/main/java/io/github/pzhin/sfqd/jcstress/RefundUnselectedCancelDispatchStress.java new file mode 100644 index 0000000..baa6797 --- /dev/null +++ b/sfqd-jcstress/src/main/java/io/github/pzhin/sfqd/jcstress/RefundUnselectedCancelDispatchStress.java @@ -0,0 +1,69 @@ +package io.github.pzhin.sfqd.jcstress; + +import io.github.pzhin.sfqd.CancellationAccounting; +import io.github.pzhin.sfqd.FlowHandle; +import io.github.pzhin.sfqd.JobHandle; +import io.github.pzhin.sfqd.SchedulerConfig; +import io.github.pzhin.sfqd.SchedulerSnapshot; +import io.github.pzhin.sfqd.SfqdScheduler; +import org.openjdk.jcstress.annotations.Actor; +import org.openjdk.jcstress.annotations.Arbiter; +import org.openjdk.jcstress.annotations.Expect; +import org.openjdk.jcstress.annotations.JCStressTest; +import org.openjdk.jcstress.annotations.Outcome; +import org.openjdk.jcstress.annotations.State; +import org.openjdk.jcstress.infra.results.IIIIII_Result; + +/** Proves refund cancellation remains atomic against dispatch of the preceding flow head. */ +@JCStressTest +@Outcome(id = "257, 1, 0, 1, 1, 1", expect = Expect.ACCEPTABLE, + desc = "One head dispatched and its queued successor refunded and cancelled.") +@Outcome(expect = Expect.FORBIDDEN, desc = "Refund cancellation and dispatch did not form one legal history.") +@State +public class RefundUnselectedCancelDispatchStress { + private final SfqdScheduler scheduler = + new SfqdScheduler<>(new SchedulerConfig( + 1, 1, 2, CancellationAccounting.REFUND_CANCELLED_COST)); + private final JobHandle victim; + + /** Builds an ordered head and queued refund victim. */ + public RefundUnselectedCancelDispatchStress() { + FlowHandle flow = SchedulerTestSupport.register(scheduler, "flow"); + SchedulerTestSupport.enqueue(scheduler, flow, 1, 1L); + victim = SchedulerTestSupport.enqueue(scheduler, flow, 2, 1L); + } + + /** + * Dispatches only the deterministic head. + * + * @param result actor result carrier + */ + @Actor + public void dispatch(IIIIII_Result result) { + result.r1 = SchedulerTestSupport.dispatchCode(scheduler.dispatchUpTo(1)); + } + + /** + * Cancels and refunds the queued successor. + * + * @param result actor result carrier + */ + @Actor + public void cancel(IIIIII_Result result) { + result.r2 = SchedulerTestSupport.cancelCode(scheduler.cancel(victim)); + } + + /** + * Reports final state and lifecycle counters. + * + * @param result actor and arbiter result carrier + */ + @Arbiter + public void report(IIIIII_Result result) { + SchedulerSnapshot snapshot = scheduler.snapshot(); + result.r3 = snapshot.queuedJobs(); + result.r4 = snapshot.runningJobs(); + result.r5 = (int) snapshot.cancelledTotal(); + result.r6 = (int) snapshot.dispatchedTotal(); + } +}