From 9d72bfd656a8d74b5e9e54b287ba8fa72da6798c Mon Sep 17 00:00:00 2001 From: Pavel Nevezhin Date: Sat, 22 Aug 2026 23:29:24 +0300 Subject: [PATCH] core: add denominator-safe weight domains Unrestricted, pairwise-coprime weights can exhaust the exact-tag budget after only a few hundred admissions, which makes the full long range too broad as a generic production recommendation. Allow deployments to constrain registered weights to divisors of one common scale while preserving the unrestricted compatibility default. Document the remaining numerator and cancellation-debt limits and keep the bounded failure as an executable regression. --- README.md | 43 +++++++++- docs/FORMAL_SPEC.md | 54 +++++++++--- .../github/pzhin/sfqd/RegisterFlowResult.java | 2 + .../io/github/pzhin/sfqd/SchedulerConfig.java | 27 +++++- .../io/github/pzhin/sfqd/SfqdScheduler.java | 10 ++- .../io/github/pzhin/sfqd/WeightDomain.java | 86 +++++++++++++++++++ .../github/pzhin/sfqd/ApiVocabularyTest.java | 17 +++- .../pzhin/sfqd/PublicArtifactSurfaceTest.java | 3 +- .../pzhin/sfqd/ReferenceAdmissionTest.java | 14 +++ .../github/pzhin/sfqd/ReferenceScheduler.java | 3 + .../SfqdLongRunNumericAccumulationTest.java | 78 +++++++++++++++++ .../github/pzhin/sfqd/SfqdSchedulerTest.java | 20 +++++ .../sfqd/build/CoreArtifactVerifier.java | 3 +- .../sfqd/jcstress/SchedulerTestSupport.java | 1 + 14 files changed, 340 insertions(+), 21 deletions(-) create mode 100644 sfqd-core/src/main/java/io/github/pzhin/sfqd/WeightDomain.java diff --git a/README.md b/README.md index 950cfe0..1e7f87e 100644 --- a/README.md +++ b/README.md @@ -95,15 +95,22 @@ deterministic FIFO tie-break. No floating-point arithmetic is used. ## Minimal example ```java +import io.github.pzhin.sfqd.CancellationAccounting; import io.github.pzhin.sfqd.CompletionResult; import io.github.pzhin.sfqd.Dispatch; import io.github.pzhin.sfqd.EnqueueResult; import io.github.pzhin.sfqd.RegisterFlowResult; import io.github.pzhin.sfqd.SchedulerConfig; import io.github.pzhin.sfqd.SfqdScheduler; +import io.github.pzhin.sfqd.WeightDomain; var scheduler = new SfqdScheduler( - new SchedulerConfig(4, 1_000, 100_000)); + new SchedulerConfig( + 4, + 1_000, + 100_000, + CancellationAccounting.CHARGE_RESERVED_COST, + WeightDomain.divisorsOf(8))); var result = scheduler.registerFlow("tenant-a", 2); if (!(result instanceof RegisterFlowResult.Registered registered)) { @@ -164,9 +171,9 @@ whose later jobs could be stranded if an earlier pool submission throws. ### Register a flow `registerFlow(flowId, weight)` creates an opaque flow handle. A flow identifier -cannot be registered twice at the same time, and the configured flow limit is -enforced. A closed identifier may later be registered again, producing a new -handle. +cannot be registered twice at the same time, and the configured flow and weight +domain limits are enforced. A closed identifier may later be registered again, +producing a new handle. ### Enqueue a job @@ -326,6 +333,34 @@ new SchedulerConfig( No free-cancellation accounting policy is currently implemented. +The three- and four-argument forms preserve the unrestricted positive `long` +weight domain. For production configurations with a known common scale, the +five-argument form can reject weights outside a denominator-safe profile at +registration: + +```java +new SchedulerConfig( + issueDepth, + maxFlows, + maxLiveJobs, + CancellationAccounting.CHARGE_RESERVED_COST, + WeightDomain.divisorsOf(8)); +``` + +This profile accepts `8, 4, 2, 1, 1`: every weight divides `8`. Consequently, +every reduced `cost / weight` denominator divides `8`, and exact addition, +maximum, and rebase subtraction cannot introduce new denominator factors. +`WeightDomain.unrestricted()` remains available for workloads that need the +full `long` range. + +The divisor profile prevents denominator growth caused by mutually coprime +weights; it is not an unconditional promise that `NUMERIC_LIMIT` can never +occur. Numerators and accumulated cancellation debt still use the documented +finite exact-arithmetic budget. In unrestricted mode, a natural trace with an +anchor job and 69 consecutive prime weights above `2^60` reaches +`NUMERIC_LIMIT` after 273 successful admissions; this behavior is covered by a +regression test. + Weights and costs are positive `long` values. Choose `D` as the number of jobs your execution layer can have issued but not completed. For `N` identical non-preemptive resources, the usual direct mapping is `D = N`. diff --git a/docs/FORMAL_SPEC.md b/docs/FORMAL_SPEC.md index 6c92f1d..9f11873 100644 --- a/docs/FORMAL_SPEC.md +++ b/docs/FORMAL_SPEC.md @@ -47,6 +47,11 @@ The operation is not a notification that capacity has become available. - `cost` — the supplied job cost, an integer in `1..Long.MAX_VALUE`. - `weight` — the flow weight, an integer in `1..Long.MAX_VALUE`. +The configured `weightDomain` MAY retain that unrestricted syntactic domain or +restrict successful registration to positive divisors of one fixed common +scale `W` in `1..Long.MAX_VALUE`. A weight outside the configured domain is an +operational registration rejection, not an invalid Java argument. + `FlowId` and `JobId` MUST honor the Java `equals/hashCode` contract throughout the stated lifetime. Both methods MUST be deterministic, side-effect-free, and non-reentrant with respect to the scheduler; they MUST NOT invoke scheduler @@ -69,8 +74,10 @@ Fairness is defined in terms of `cost`, not unknown actual execution time. `D..Integer.MAX_VALUE`. - `cancellationAccounting` — the fixed policy `CancellationAccounting.CHARGE_RESERVED_COST`; no alternative policy exists. +- `weightDomain` — either unrestricted positive `long` weights or the positive + divisors of one fixed common scale `W`. -All four values are immutable after instance construction. Null values, +All five values are immutable after instance construction. Null values, out-of-range values, and `maxLiveJobs < D` are rejected before an observable instance exists. @@ -172,9 +179,9 @@ transaction. Silent overflow, rounding, partial rebase, and order changes are forbidden. When `V=0` and `lastFinish=0`, a registered flow MUST accept one job with any -`cost,weight` pair in `1..Long.MAX_VALUE`, including the maximum values, unless -an independent identity or live-item limit applies: a single reduced fraction -occupies at most 63 bits in either component. +`cost` in `1..Long.MAX_VALUE` and its already accepted registration weight, +unless an independent identity or live-item limit applies: a single reduced +fraction occupies at most 63 bits in either component. The number of successfully accepted jobs over an instance lifetime is limited to `Long.MAX_VALUE`. State stores `lastJobSequence` in @@ -206,6 +213,23 @@ The number of read-only and failed calls is not limited by scheduler state. The number of successful cancel, dispatch, and complete calls is limited by the number of accepted incarnations. +### 3.2.1 Divisor-constrained weight domain + +For `weightDomain = divisorsOf(W)`, registration accepts a weight `w` only when +`W mod w = 0`. For every accepted job, the reduced denominator of +`cost / weight` therefore divides `W`. The set of rational numbers whose +reduced denominators divide `W` is closed under exact addition, subtraction, +and maximum. It follows inductively that every stored `S`, `F`, `V`, +`lastFinish`, and every exact rebase result has a reduced denominator dividing +`W`; distinct registered weights cannot accumulate new prime denominator +factors. + +This is a denominator-safety property, not an infinite-lifetime numeric +guarantee. Numerators remain subject to the 4096-bit persistent budget, and the +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.3 Exact rebasing A rebase is a representational substitution of the same semantic state. Let @@ -504,16 +528,17 @@ consistently. Order of processing: 1. Validate a non-null `flowId` and `weight` in `1..Long.MAX_VALUE`. -2. If `flowId` is in `RegisteredById`, return +2. If `weight` is outside `weightDomain`, return `WEIGHT_OUTSIDE_DOMAIN`. +3. If `flowId` is in `RegisteredById`, return `DUPLICATE_REGISTERED_ID`. -3. If the registered count equals `maxFlows`, return `FLOW_LIMIT`. -4. If `lastFlowSequence == Long.MAX_VALUE`, return +4. If the registered count equals `maxFlows`, return `FLOW_LIMIT`. +5. If `lastFlowSequence == Long.MAX_VALUE`, return `FLOW_SEQUENCE_EXHAUSTED`. -5. Create an inert `FlowHandle(ownerToken,lastFlowSequence+1)` and FlowState +6. Create an inert `FlowHandle(ownerToken,lastFlowSequence+1)` and FlowState with `lastFinish=0`, zero counts, `acceptedCost=0`, `dispatchedCost=0`, `cancelledCost=0`, `runningSuppliedCost=0`, and the fixed weight; insert both registration indexes and update the sequence. -6. Return `REGISTERED(flowHandle)`. +7. Return `REGISTERED(flowHandle)`. Registration during a non-empty busy period is allowed: it does not affect scheduling before the first enqueue. A rejection leaves state unchanged and @@ -937,7 +962,7 @@ not transitively retain the scheduler. | Batch dispatch | Filling depth is described without API atomicity | One call selects a sequential SFQ(D) batch but linearizes as a whole | | Completion order | Black-box server | Any running handle may complete; the scheduler imposes no completion order | | Weight changes | Undefined | Weight is fixed for the registration lifetime; change requires safe close and a new registration | -| Numbers | Mathematical unbounded tags | Exact canonical rationals, fail-closed 4096-bit persistent and 8193-bit transient budgets, and transactional all-registration rebase | +| Numbers | Mathematical unbounded tags | Exact canonical rationals, fail-closed 4096-bit persistent and 8193-bit transient budgets, transactional all-registration rebase, and an optional common-scale divisor weight domain that prevents new denominator factors but does not bound numerators or cancellation debt | | Retention | Not considered | Payload release, no terminal metadata, and `maxLiveJobs` and `maxFlows` bounds | | Introspection | Not considered | Exact atomic aggregate and per-registration lifecycle snapshots without tags, identifiers, or a clock | | Executor rejection | Not considered | Dispatch is irrevocable; the caller must complete, and requeue is a new enqueue | @@ -957,6 +982,15 @@ budget is genuinely violated after the one permitted transactional rebase; the production state remains unchanged, and oracle state is rolled back so the shared trace can continue. +The production regression suite MUST include a public-operation-only trace in +the unrestricted domain with `D=2`, `maxFlows=70`, `maxLiveJobs=3`, one running +unit-weight anchor, and 69 consecutive prime weights above `2^60`. Two visits +of two unit-cost jobs per fractional flow MUST reproduce the bounded numeric +failure after 273 total successful admissions and MUST verify that the rejected +enqueue is an atomic no-op. The same prime weights MUST be rejected at +registration by a divisor-constrained domain whose common scale they do not +divide. + When no such expected rejection occurs, the reference model and production implementation MUST agree on: diff --git a/sfqd-core/src/main/java/io/github/pzhin/sfqd/RegisterFlowResult.java b/sfqd-core/src/main/java/io/github/pzhin/sfqd/RegisterFlowResult.java index a115e50..fe7ef50 100644 --- a/sfqd-core/src/main/java/io/github/pzhin/sfqd/RegisterFlowResult.java +++ b/sfqd-core/src/main/java/io/github/pzhin/sfqd/RegisterFlowResult.java @@ -28,6 +28,8 @@ enum Rejected implements RegisterFlowResult { DUPLICATE_REGISTERED_ID, /** The configured registration limit is full. */ FLOW_LIMIT, + /** The weight is outside the configured weight domain. */ + WEIGHT_OUTSIDE_DOMAIN, /** The lifetime flow sequence has been exhausted. */ FLOW_SEQUENCE_EXHAUSTED } 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 f60ba8b..e71c9b6 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 @@ -20,12 +20,14 @@ * @param maxLiveJobs maximum queued plus running jobs, in * {@code [depth, Integer.MAX_VALUE]} * @param cancellationAccounting virtual fairness accounting policy for cancelled queued jobs + * @param weightDomain registration policy for fixed flow weights */ public record SchedulerConfig( int depth, int maxFlows, int maxLiveJobs, - CancellationAccounting cancellationAccounting) { + CancellationAccounting cancellationAccounting, + WeightDomain weightDomain) { /** * Maximum representable and validated issue depth. * @@ -42,7 +44,24 @@ public record SchedulerConfig( * @param maxLiveJobs maximum queued plus running jobs, in {@code [depth, Integer.MAX_VALUE]} */ public SchedulerConfig(int depth, int maxFlows, int maxLiveJobs) { - this(depth, maxFlows, maxLiveJobs, CancellationAccounting.CHARGE_RESERVED_COST); + this(depth, maxFlows, maxLiveJobs, + CancellationAccounting.CHARGE_RESERVED_COST, WeightDomain.unrestricted()); + } + + /** + * Creates a configuration with an unrestricted weight domain. + * + * @param depth maximum outstanding 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 + */ + public SchedulerConfig( + int depth, + int maxFlows, + int maxLiveJobs, + CancellationAccounting cancellationAccounting) { + this(depth, maxFlows, maxLiveJobs, cancellationAccounting, WeightDomain.unrestricted()); } /** @@ -55,13 +74,15 @@ public SchedulerConfig(int depth, int maxFlows, int maxLiveJobs) { * @param maxLiveJobs maximum queued plus running jobs, in * {@code [depth, Integer.MAX_VALUE]} * @param cancellationAccounting virtual fairness accounting policy for cancelled queued jobs + * @param weightDomain registration policy for fixed flow weights * @throws IllegalArgumentException if depth is outside {@code [1, 1_000_000]}, * maxFlows is outside {@code [1, Integer.MAX_VALUE]}, or * maxLiveJobs is outside {@code [depth, Integer.MAX_VALUE]} - * @throws NullPointerException if cancellationAccounting is null + * @throws NullPointerException if cancellationAccounting or weightDomain is null */ public SchedulerConfig { Objects.requireNonNull(cancellationAccounting, "cancellationAccounting"); + Objects.requireNonNull(weightDomain, "weightDomain"); if (depth < 1 || depth > MAX_DEPTH) { throw new IllegalArgumentException("depth must be in [1, 1_000_000]"); } 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 3fe4096..7d28bf5 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 @@ -36,7 +36,9 @@ * {@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. * - *

Weights and costs are positive {@code long} values in {@code [1, Long.MAX_VALUE]}. Tags are exact reduced + *

Weights and costs are positive {@code long} values in {@code [1, Long.MAX_VALUE]}. A configured + * {@link WeightDomain} may further restrict registered weights to divisors of one common scale, preventing new + * denominator factors from accumulating across flows. Tags are exact reduced * non-negative rationals: each numerator and denominator retained in scheduler state has bit length at most 4096, * and each canonical raw or reduced component of an exact primitive has bit length at most 8193. No rounding or * floating-point fallback is permitted. If a newly computed start or finish tag first exceeds the persistent budget, @@ -86,6 +88,9 @@ public SfqdScheduler(SchedulerConfig config) { * *

A success linearizes at insertion into both registration indexes and sequence advancement. A rejection * linearizes at the first applicable check and does not mutate state or consume a sequence. + * A divisor-constrained {@link WeightDomain} returns + * {@link RegisterFlowResult.Rejected#WEIGHT_OUTSIDE_DOMAIN} before identity, capacity, and sequence checks when + * the positive weight does not divide the configured common scale. * * @param flowId stable non-null flow identifier * @param weight fixed registration weight in {@code [1, Long.MAX_VALUE]} @@ -98,6 +103,9 @@ public RegisterFlowResult registerFlow(F flowId, long weight) { requirePositive(weight, "weight"); lock.lock(); try { + if (!config.weightDomain().permits(weight)) { + return RegisterFlowResult.Rejected.WEIGHT_OUTSIDE_DOMAIN; + } if (registeredById.containsKey(flowId)) { return RegisterFlowResult.Rejected.DUPLICATE_REGISTERED_ID; } diff --git a/sfqd-core/src/main/java/io/github/pzhin/sfqd/WeightDomain.java b/sfqd-core/src/main/java/io/github/pzhin/sfqd/WeightDomain.java new file mode 100644 index 0000000..1e689d0 --- /dev/null +++ b/sfqd-core/src/main/java/io/github/pzhin/sfqd/WeightDomain.java @@ -0,0 +1,86 @@ +package io.github.pzhin.sfqd; + +import java.util.Objects; +import java.util.OptionalLong; + +/** + * Immutable registration policy for flow weights. + * + *

{@link #unrestricted()} preserves the full syntactic weight range in + * {@code [1, Long.MAX_VALUE]}. It does not constrain the combined arithmetic + * structure of registered weights, so an otherwise valid workload can reach + * the scheduler's exact-tag representation budget after a bounded number of + * admissions. + * + *

{@link #divisorsOf(long)} admits only weights that divide one fixed common + * scale. Every reduced {@code cost / weight} denominator then divides that + * scale, and addition, maximum, and rebase subtraction cannot introduce a new + * denominator factor. This prevents pairwise-coprime weights from causing + * denominator growth. It is not a promise that {@code NUMERIC_LIMIT} is + * impossible: numerators and accumulated cancellation debt remain bounded by + * the scheduler's documented representation budget. + */ +public final class WeightDomain { + private static final WeightDomain UNRESTRICTED = new WeightDomain(OptionalLong.empty()); + + private final OptionalLong commonScale; + + private WeightDomain(OptionalLong commonScale) { + this.commonScale = commonScale; + } + + /** + * Returns the compatibility policy accepting every positive {@code long} weight. + * + * @return unrestricted weight domain + */ + public static WeightDomain unrestricted() { + return UNRESTRICTED; + } + + /** + * Returns a domain accepting exactly the positive divisors of {@code commonScale}. + * + * @param commonScale fixed positive common weight scale + * @return divisor-constrained weight domain + * @throws IllegalArgumentException if commonScale is not positive + */ + public static WeightDomain divisorsOf(long commonScale) { + if (commonScale <= 0L) { + throw new IllegalArgumentException("commonScale must be positive"); + } + return new WeightDomain(OptionalLong.of(commonScale)); + } + + /** + * Returns the common scale when weights are divisor-constrained. + * + * @return common scale, or empty for the unrestricted domain + */ + public OptionalLong commonScale() { + return commonScale; + } + + boolean permits(long weight) { + return commonScale.isEmpty() || commonScale.getAsLong() % weight == 0L; + } + + @Override + public boolean equals(Object other) { + return this == other + || other instanceof WeightDomain domain + && commonScale.equals(domain.commonScale); + } + + @Override + public int hashCode() { + return Objects.hash(commonScale); + } + + @Override + public String toString() { + return commonScale.isEmpty() + ? "WeightDomain[unrestricted]" + : "WeightDomain[divisorsOf=" + commonScale.getAsLong() + "]"; + } +} 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 72aa06d..a460d64 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 @@ -44,10 +44,25 @@ void configMakesCancellationAccountingExplicit() { 8, 10, 20, CancellationAccounting.CHARGE_RESERVED_COST); assertEquals(CancellationAccounting.CHARGE_RESERVED_COST, defaultConfig.cancellationAccounting()); + assertEquals(WeightDomain.unrestricted(), defaultConfig.weightDomain()); assertEquals(defaultConfig, explicitConfig); assertEquals(1, CancellationAccounting.values().length); } + @Test + void configCanConstrainWeightsToOneCommonScale() { + WeightDomain divisorsOfEight = WeightDomain.divisorsOf(8L); + SchedulerConfig config = new SchedulerConfig( + 8, 10, 20, CancellationAccounting.CHARGE_RESERVED_COST, divisorsOfEight); + + assertEquals(divisorsOfEight, config.weightDomain()); + assertEquals(java.util.OptionalLong.of(8L), divisorsOfEight.commonScale()); + assertTrue(WeightDomain.unrestricted().commonScale().isEmpty()); + assertEquals(WeightDomain.divisorsOf(8L), divisorsOfEight); + assertEquals("WeightDomain[divisorsOf=8]", divisorsOfEight.toString()); + assertThrows(IllegalArgumentException.class, () -> WeightDomain.divisorsOf(0L)); + } + @Test void handlesAreOpaqueInertCapabilitiesWithOwnerScopedEquality() { OwnerToken firstOwner = new OwnerToken(); @@ -108,7 +123,7 @@ void valueOutcomesAndSnapshotCarryOnlySpecifiedData() { JobHandle job = new JobHandle(owner, 1L); assertSame(flow, new RegisterFlowResult.Registered(flow).flowHandle()); assertSame(job, new EnqueueResult.Accepted(job).jobHandle()); - assertEquals(3, RegisterFlowResult.Rejected.values().length); + assertEquals(4, RegisterFlowResult.Rejected.values().length); assertEquals(5, EnqueueResult.Rejected.values().length); assertEquals(4, CloseFlowResult.values().length); assertEquals(3, CancelResult.values().length); diff --git a/sfqd-core/src/test/java/io/github/pzhin/sfqd/PublicArtifactSurfaceTest.java b/sfqd-core/src/test/java/io/github/pzhin/sfqd/PublicArtifactSurfaceTest.java index 59a56f9..e8b9229 100644 --- a/sfqd-core/src/test/java/io/github/pzhin/sfqd/PublicArtifactSurfaceTest.java +++ b/sfqd-core/src/test/java/io/github/pzhin/sfqd/PublicArtifactSurfaceTest.java @@ -31,7 +31,8 @@ final class PublicArtifactSurfaceTest { "io.github.pzhin.sfqd.RegisterFlowResult$Rejected", "io.github.pzhin.sfqd.SchedulerConfig", "io.github.pzhin.sfqd.SchedulerSnapshot", - "io.github.pzhin.sfqd.SfqdScheduler"); + "io.github.pzhin.sfqd.SfqdScheduler", + "io.github.pzhin.sfqd.WeightDomain"); @Test void exposesOnlyTheSpecifiedPublicTypes() throws IOException, URISyntaxException { 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 6438b77..2018dfe 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 @@ -32,6 +32,20 @@ void registerAndCloseFollowIdentityCapacityAndFairnessDebtRules() { model.enqueue(foreignHandle, "foreign-job", new Object(), 1L)); } + @Test + void registrationEnforcesConfiguredWeightDomainBeforeOtherOperationalLimits() { + SchedulerConfig config = new SchedulerConfig( + 1, 1, 1, CancellationAccounting.CHARGE_RESERVED_COST, WeightDomain.divisorsOf(8L)); + ReferenceScheduler model = new ReferenceScheduler<>(config); + + assertEquals(RegisterFlowResult.Rejected.WEIGHT_OUTSIDE_DOMAIN, model.registerFlow("outside", 3L)); + FlowHandle flow = registered(model.registerFlow("inside", 4L)); + assertEquals(RegisterFlowResult.Rejected.WEIGHT_OUTSIDE_DOMAIN, model.registerFlow("inside", 3L)); + assertEquals(RegisterFlowResult.Rejected.FLOW_LIMIT, model.registerFlow("full", 2L)); + assertEquals(1, model.snapshot().registeredFlows()); + assertEquals(CloseFlowResult.CLOSED, model.closeFlow(flow)); + } + @Test void enqueueCalculatesExactTagsAndAppliesAdmissionChecksInOrder() { ReferenceScheduler model = model(1, 2, 2); 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 a6c3bec..6d571c4 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 @@ -43,6 +43,9 @@ static ReferenceScheduler withSequences( RegisterFlowResult registerFlow(F flowId, long weight) { Objects.requireNonNull(flowId, "flowId"); requirePositive(weight, "weight"); + if (!config.weightDomain().permits(weight)) { + return RegisterFlowResult.Rejected.WEIGHT_OUTSIDE_DOMAIN; + } if (registeredById.containsKey(flowId)) { return RegisterFlowResult.Rejected.DUPLICATE_REGISTERED_ID; } diff --git a/sfqd-core/src/test/java/io/github/pzhin/sfqd/SfqdLongRunNumericAccumulationTest.java b/sfqd-core/src/test/java/io/github/pzhin/sfqd/SfqdLongRunNumericAccumulationTest.java index fb3da49..253187e 100644 --- a/sfqd-core/src/test/java/io/github/pzhin/sfqd/SfqdLongRunNumericAccumulationTest.java +++ b/sfqd-core/src/test/java/io/github/pzhin/sfqd/SfqdLongRunNumericAccumulationTest.java @@ -21,6 +21,76 @@ class SfqdLongRunNumericAccumulationTest { private static final int PERSISTENT_TAG_BITS = 4096; private static final int NUMERIC_STRESS_FLOOR_BITS = 3500; + @Test + void pairwiseCoprimeLongWeightsReachNumericLimitAfterBoundedAdmissions() { + SfqdScheduler scheduler = + new SfqdScheduler<>(new SchedulerConfig(DEPTH, 70, 3)); + FlowHandle anchor = registered(scheduler.registerFlow("anchor", 1L)); + List flows = new ArrayList<>(69); + BigInteger nextWeight = BigInteger.ONE.shiftLeft(60).nextProbablePrime(); + for (int index = 0; index < 69; index++) { + flows.add(registered(scheduler.registerFlow("fractional-" + index, nextWeight.longValueExact()))); + nextWeight = nextWeight.nextProbablePrime(); + } + + JobHandle anchorJob = accepted(scheduler.enqueue(anchor, 1L, new Payload(1L), 1L)); + assertEquals(anchorJob, scheduler.dispatchUpTo(1).get(0).jobHandle()); + + long nextJobId = 2L; + int successfulAdmissions = 1; + for (int pass = 0; pass < 2; pass++) { + for (FlowHandle flow : flows) { + List visit = new ArrayList<>(2); + for (int visitJob = 0; visitJob < 2; visitJob++) { + SchedulerSnapshot before = scheduler.snapshot(); + EnqueueResult result = scheduler.enqueue( + flow, nextJobId, new Payload(nextJobId), 1L); + if (result == EnqueueResult.Rejected.NUMERIC_LIMIT) { + assertEquals(273, successfulAdmissions); + assertEquals(before, scheduler.snapshot()); + for (JobHandle accepted : visit) { + assertEquals(accepted, scheduler.dispatchUpTo(1).get(0).jobHandle()); + assertEquals(CompletionResult.COMPLETED, scheduler.complete(accepted)); + } + assertEquals(CompletionResult.COMPLETED, scheduler.complete(anchorJob)); + JobHandle afterIdleReset = accepted(scheduler.enqueue( + flow, nextJobId, new Payload(nextJobId), 1L)); + assertEquals(afterIdleReset, scheduler.dispatchUpTo(1).get(0).jobHandle()); + assertEquals(CompletionResult.COMPLETED, scheduler.complete(afterIdleReset)); + return; + } + JobHandle accepted = assertInstanceOf(EnqueueResult.Accepted.class, result).jobHandle(); + visit.add(accepted); + successfulAdmissions++; + nextJobId++; + } + for (JobHandle accepted : visit) { + assertEquals(accepted, scheduler.dispatchUpTo(1).get(0).jobHandle()); + assertEquals(CompletionResult.COMPLETED, scheduler.complete(accepted)); + } + } + } + throw new AssertionError("trace did not reach NUMERIC_LIMIT"); + } + + @Test + void divisorWeightDomainRejectsCounterexampleWeightAtRegistration() { + SfqdScheduler scheduler = new SfqdScheduler<>(new SchedulerConfig( + DEPTH, + 70, + 3, + CancellationAccounting.CHARGE_RESERVED_COST, + WeightDomain.divisorsOf(8L))); + SchedulerSnapshot before = scheduler.snapshot(); + BigInteger nextWeight = BigInteger.ONE.shiftLeft(60).nextProbablePrime(); + for (int index = 0; index < 69; index++) { + assertEquals(RegisterFlowResult.Rejected.WEIGHT_OUTSIDE_DOMAIN, + scheduler.registerFlow("fractional-" + index, nextWeight.longValueExact())); + assertEquals(before, scheduler.snapshot()); + nextWeight = nextWeight.nextProbablePrime(); + } + } + @Test void publicOperationsPreserveExactOrderAcrossLongFractionalAccumulationNearNumericBudget() { SchedulerConfig config = new SchedulerConfig(DEPTH, FLOW_COUNT + 1, 5); @@ -153,6 +223,14 @@ private static FlowPair registerPair( return new FlowPair(expected.flowHandle(), actual.flowHandle()); } + 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(); + } + private static JobPair enqueuePair( ReferenceScheduler reference, SfqdScheduler production, diff --git a/sfqd-core/src/test/java/io/github/pzhin/sfqd/SfqdSchedulerTest.java b/sfqd-core/src/test/java/io/github/pzhin/sfqd/SfqdSchedulerTest.java index 4105508..e6a9874 100644 --- a/sfqd-core/src/test/java/io/github/pzhin/sfqd/SfqdSchedulerTest.java +++ b/sfqd-core/src/test/java/io/github/pzhin/sfqd/SfqdSchedulerTest.java @@ -10,6 +10,26 @@ import org.junit.jupiter.api.Test; class SfqdSchedulerTest { + @Test + void divisorWeightDomainRejectsOutsideWeightsWithoutConsumingCapacity() { + SchedulerConfig config = new SchedulerConfig( + 2, 5, 5, CancellationAccounting.CHARGE_RESERVED_COST, WeightDomain.divisorsOf(8L)); + SfqdScheduler scheduler = new SfqdScheduler<>(config); + + SchedulerSnapshot before = scheduler.snapshot(); + assertEquals(RegisterFlowResult.Rejected.WEIGHT_OUTSIDE_DOMAIN, + scheduler.registerFlow("outside", 3L)); + assertEquals(before, scheduler.snapshot()); + List handles = List.of( + registered(scheduler.registerFlow("eight", 8L)), + registered(scheduler.registerFlow("four", 4L)), + registered(scheduler.registerFlow("two", 2L)), + registered(scheduler.registerFlow("one-a", 1L)), + registered(scheduler.registerFlow("one-b", 1L))); + assertEquals(5, handles.size()); + assertEquals(5, scheduler.snapshot().registeredFlows()); + } + @Test void schedulesByExactStartTagAndAccountsLifecycle() { SfqdScheduler scheduler = diff --git a/sfqd-core/src/test/java/io/github/pzhin/sfqd/build/CoreArtifactVerifier.java b/sfqd-core/src/test/java/io/github/pzhin/sfqd/build/CoreArtifactVerifier.java index a4b5472..4c5d6c5 100644 --- a/sfqd-core/src/test/java/io/github/pzhin/sfqd/build/CoreArtifactVerifier.java +++ b/sfqd-core/src/test/java/io/github/pzhin/sfqd/build/CoreArtifactVerifier.java @@ -52,7 +52,8 @@ public final class CoreArtifactVerifier { "io.github.pzhin.sfqd.RegisterFlowResult$Rejected", "io.github.pzhin.sfqd.SchedulerConfig", "io.github.pzhin.sfqd.SchedulerSnapshot", - "io.github.pzhin.sfqd.SfqdScheduler"); + "io.github.pzhin.sfqd.SfqdScheduler", + "io.github.pzhin.sfqd.WeightDomain"); private CoreArtifactVerifier() { } diff --git a/sfqd-jcstress/src/main/java/io/github/pzhin/sfqd/jcstress/SchedulerTestSupport.java b/sfqd-jcstress/src/main/java/io/github/pzhin/sfqd/jcstress/SchedulerTestSupport.java index 52ec365..5d4d3d2 100644 --- a/sfqd-jcstress/src/main/java/io/github/pzhin/sfqd/jcstress/SchedulerTestSupport.java +++ b/sfqd-jcstress/src/main/java/io/github/pzhin/sfqd/jcstress/SchedulerTestSupport.java @@ -65,6 +65,7 @@ static int registerCode(RegisterFlowResult result) { case FLOW_LIMIT -> FLOW_LIMIT; case DUPLICATE_REGISTERED_ID -> 3; case FLOW_SEQUENCE_EXHAUSTED -> 4; + case WEIGHT_OUTSIDE_DOMAIN -> 5; }; }