Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 6 additions & 1 deletion dag/extdeps/systemd/systemd.dag
Original file line number Diff line number Diff line change
Expand Up @@ -274,9 +274,14 @@ fn parse_systemd_cpu_quota_per_sec_show(raw: String) -> SystemdCpuQuotaPerSec {

// COMPARABLE ARMS HAVE AN IDENTITY STRING; UNDECODABLE HAS NONE. Desired never produces Undecodable:
// it renders through systemd_cpu_quota_per_sec_magnitude_identity of of_threads, which is total.
// THE UNBOUNDED IDENTITY IS A NAMED CONSTANT because a DESIRED side that declares no ceiling has to
// spell it the same way the observed side does, and reaching that spelling through the identity
// function would hand the desired side an Optional it can never see Absent on.
data systemd_cpu_quota_unbounded_identity: String = "infinity"

fn systemd_cpu_quota_per_sec_identity(q: SystemdCpuQuotaPerSec) -> String? {
match q {
CpuQuotaUnbounded => Present { value: "infinity" }
CpuQuotaUnbounded => Present { value: systemd_cpu_quota_unbounded_identity }
CpuQuotaPerSec { usec: u } => Present { value: systemd_cpu_quota_per_sec_magnitude_identity(usec: u) }
CpuQuotaAbsent => Present { value: "" }
CpuQuotaUndecodable { raw: _ } => none
Expand Down
17 changes: 17 additions & 0 deletions dag/extdeps/systemd/unit_file.dag
Original file line number Diff line number Diff line change
Expand Up @@ -449,6 +449,7 @@ type SystemdDirectiveValue
| DirectiveWeight { weight: Int }
| DirectiveCpuTime { threads: HardwareThreadCount }
| DirectiveUnlimited
| DirectiveReset

fn systemd_directive_value_wire(value: SystemdDirectiveValue) -> String {
match value {
Expand All @@ -457,9 +458,18 @@ fn systemd_directive_value_wire(value: SystemdDirectiveValue) -> String {
DirectiveWeight { weight: w } => to_string(w)
DirectiveCpuTime { threads: t } => join([to_string(hardware_thread_count_value(t: t) * 100), "%"], "")
DirectiveUnlimited => systemd_directive_unlimited_wire
DirectiveReset => systemd_directive_reset_wire
}
}

// AN EMPTY ASSIGNMENT RESETS THE SETTING TO ITS DEFAULT. systemd.syntax(7) states it for the unit
// file, and systemd.resource-control(5) restates it for the one knob that has no `infinity`
// spelling: "Setting CPUQuota= to an empty value unsets the quota" -- `CPUQuota=infinity` is not a
// value that directive's parser admits, so `DirectiveUnlimited` cannot serve it. `systemctl
// set-property` reads the same empty assignment as unset. `DirectiveReset` is the write-side arm
// for that wire; the read side already carries the state it produces as `CpuQuotaUnbounded`.
data systemd_directive_reset_wire: String = ""

// systemd's sentinel for "no limit", spelled once. `max` is the cgroup v2 kernel interface's word;
// `infinity` is systemd's, and a unit file is read by systemd rather than by the kernel.
data systemd_directive_unlimited_wire: String = "infinity"
Expand Down Expand Up @@ -521,6 +531,13 @@ fn slice_cpu_quota(threads: HardwareThreadCount) -> SystemdSliceDirective {
SystemdSliceDirective { property: CPUQuota, value: DirectiveCpuTime { threads: threads } }
}

// NO CEILING ON THE CPU AXIS IS AN ASSIGNMENT, NOT AN OMISSION. A slice that was once given a
// quota keeps it until something unsets it, so a desired state of "unbounded" that emitted nothing
// would converge only on cells that never had one. This mint is the affirmative unset.
fn slice_cpu_quota_unbounded() -> SystemdSliceDirective {
SystemdSliceDirective { property: CPUQuota, value: DirectiveReset }
}

// The directive names are systemd's, spelled once here. A consumer never writes the literal
// "ExecStart"; it constructs `ExecStart { command }` and this module decides how that renders.

Expand Down
89 changes: 48 additions & 41 deletions dag/gunbc/ci/ci_runner_placement.dag
Original file line number Diff line number Diff line change
Expand Up @@ -11,11 +11,11 @@ import std.disposition { Disposition, Scaffold, SingleAuthority }
import std.decl_ref { DeclarationRef, WholeDeclaration }
import gunbc.runner_spec_from_offer { runner_spec_from_offer }
import gunbc.runner_slot_allocation {
CoresPerSlotResolved,
CoresPerSlotUnresolved,
RunnerSlotCpuQuota,
CpuQuotaResolved,
SlotCpuQuotaUnbounded,
gunbc_runner_slot_desired,
gunbc_runner_slots_per_host,
gunbc_runner_cores_per_slot,
}
import gunbc.session_reservation_policy {
SessionReservationCharged, SessionReservationUnsized,
Expand Down Expand Up @@ -109,17 +109,24 @@ fn workload_class_cpu_weight(wc: WorkloadClass) -> Int {
// amount of anything. CPUQuota= is an ABSOLUTE ceiling on CPU time per wall second. AllowedCPUs=
// is a third thing again -- a pinning of the unit to named CPUs.
//
// CapacityQuota is the no-overcommit contract this fleet adopts: each slot may consume at most
// cores_per_slot threads' worth of CPU time, and the sum of the runner, compile-pool and host
// reserve ceilings must fit the machine. Pinning each slot to a pairwise-disjoint SET of named CPUs
// is a stronger contract again, and this fleet does not make it; the block below records why it has
// no constructor rather than a variant.
// CapacityQuota is a no-overcommit contract: each slot may consume at most that many threads'
// worth of CPU time, and the sum of the runner, compile-pool and host reserve ceilings must fit
// the machine. NoCpuCeiling is the contract this fleet ADOPTS as of the 2026-09-19 operator ruling
// ("there is no CPU demand - the processes we run are single core"): a slot carries CPUWeight only
// and no CPUQuota, which is the state the hosts run in today, and a 100% quota at the policy of
// one core per slot would throttle the parallel link/codegen phases of a job that is otherwise
// single-core. The arm is read from gunbc.runner_slot_allocation's cpu_quota, never derived from
// cores_per_slot here: width admission and the per-slot ceiling are two facts of one row, and
// deriving one from the other is what bound the fleet at a tenfold over-reservation.
// Pinning each slot to a pairwise-disjoint SET of named CPUs is a stronger contract again, and
// this fleet does not make it; the block below records why it has no constructor rather than a
// variant.
//
// THE ARM THIS TYPE REPLACES WAS NEITHER OF THEM. cores_per_slot was consumed to decide how many
// units to create and then dropped, so the width was CPU-derived while the units it created carried
// no CPU directive at all -- srv1 and srv2 show MemoryMax as their only resource line, CPUWeight
// unset and CPUQuota infinity. Admission without confinement: the arithmetic said six threads per
// slot and the host was never told.
// NoCpuCeiling IS NOT THE "ADMISSION WITHOUT CONFINEMENT" ARM THIS TYPE ONCE REPLACED. That arm
// derived a width from cores_per_slot and then told the host nothing, so the units ran unconfined
// while the model believed them confined. NoCpuCeiling realizes as an affirmative `CPUQuota=`
// reset on every unit (gunbc.host_converge runner_cpu_boundary_knobs), so the model and the host
// agree that there is no ceiling; a unit that once carried one is converged back to none.
// THE PINNING CONTRACT HAS NO CONSTRUCTOR, AND THAT IS THE POINT RATHER THAN AN OMISSION.
//
// An earlier revision carried a DisjointCpuSet arm so the promise this fleet does NOT make would have
Expand All @@ -144,6 +151,7 @@ fn workload_class_cpu_weight(wc: WorkloadClass) -> Int {
// this deletion removes.
type RunnerCpuBoundary
= CapacityQuota { threads: HardwareThreadCount }
| NoCpuCeiling

type RunnerCpuBoundaryResolution
= RunnerCpuBoundaryDerived { boundary: RunnerCpuBoundary }
Expand All @@ -167,9 +175,15 @@ type RunnerCpuBoundaryResolution
//
// NEXT-RUNG TRIGGER: a compile-pool CPU charge in the host allocation, at which point this arm
// derives a rebalanced width and ceiling instead of refusing.
fn runner_cpu_boundary_for(placement: CompilePoolPlacement, per_slot: HardwareThreadCount) -> RunnerCpuBoundaryResolution {
fn runner_cpu_boundary_for(placement: CompilePoolPlacement, quota: RunnerSlotCpuQuota) -> RunnerCpuBoundaryResolution {
match placement {
CompilePoolInRunnerSlots => RunnerCpuBoundaryDerived { boundary: CapacityQuota { threads: per_slot } }
CompilePoolInRunnerSlots =>
RunnerCpuBoundaryDerived {
boundary: match quota {
CpuQuotaResolved { threads: t } => CapacityQuota { threads: t }
SlotCpuQuotaUnbounded => NoCpuCeiling
}
}
CompilePoolSliceDeclared { slice_unit: u } =>
RunnerCpuBoundaryUnchargeable {
reason: concat("compile pool is a declared host slice and carries no modeled CPU charge, so a per-slot CPU ceiling derived from cores_per_slot alone would commit the host past its threads once that pool runs: rederive runner width and the per-slot ceiling together with the pool's CPU charge. Slice: ", u)
Expand Down Expand Up @@ -215,7 +229,6 @@ type HostRunnerDeploymentRefusal
| ZeroRunnerWidth
| JobserverCitationAbsent { cause: String }
| HostCpuAllocationUnchargeable { cause: String }
| RunnerCoreRatioUnresolved { cause: String }
| HostNotEnrolledInFleet

type RefusedHostDeployment {
Expand Down Expand Up @@ -511,7 +524,6 @@ fn host_deployment_refusal_text(c: HostRunnerDeploymentRefusal) -> String {
ZeroRunnerWidth => "zero_runner_width"
JobserverCitationAbsent { cause: t } => concat("jobserver_citation_absent ", t)
HostCpuAllocationUnchargeable { cause: t } => concat("host_cpu_allocation_unchargeable ", t)
RunnerCoreRatioUnresolved { cause: t } => concat("runner_core_ratio_unresolved ", t)
HostNotEnrolledInFleet => "host_not_enrolled_in_fleet"
}
}
Expand Down Expand Up @@ -602,30 +614,25 @@ fn host_standing_with_session_slice(
cause: concat("no cited jobserver configuration, so the compile-token budget has no derivation and the host cannot be told a bound it did not compute: ", host_id)
})
JobserverCited { config: jcfg, provenance: _ } =>
match gunbc_runner_cores_per_slot() {
CoresPerSlotUnresolved { cause: rc } =>
refuse_host(host: host_id, cause: RunnerCoreRatioUnresolved { cause: rc })
CoresPerSlotResolved { threads: per_slot_threads } =>
match runner_cpu_boundary_for(
placement: gunbc_compile_pool_placement,
per_slot: per_slot_threads
) {
RunnerCpuBoundaryUnchargeable { reason: cr } =>
refuse_host(host: host_id, cause: HostCpuAllocationUnchargeable { cause: cr })
RunnerCpuBoundaryDerived { boundary: cpub } =>
admit_or_refuse_zero_width(
dep: conservation_host_deployment(
offer: offer,
build_tokens: derive_jobserver_tokens(
nproc: cpu_facts_threads(cpu: cpu),
mem_total: host_ram,
cfg: jcfg
),
cpu_boundary: cpub
),
host_id: host_id
)
}
match runner_cpu_boundary_for(
placement: gunbc_compile_pool_placement,
quota: gunbc_runner_slot_desired().cpu_quota
) {
RunnerCpuBoundaryUnchargeable { reason: cr } =>
refuse_host(host: host_id, cause: HostCpuAllocationUnchargeable { cause: cr })
RunnerCpuBoundaryDerived { boundary: cpub } =>
admit_or_refuse_zero_width(
dep: conservation_host_deployment(
offer: offer,
build_tokens: derive_jobserver_tokens(
nproc: cpu_facts_threads(cpu: cpu),
mem_total: host_ram,
cfg: jcfg
),
cpu_boundary: cpub
),
host_id: host_id
)
}
}
}
Expand Down
91 changes: 36 additions & 55 deletions dag/gunbc/fabric/fabric_cell_converge.dag
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ import std.measure { byte_size_count }
import extdeps.systemd {
systemd_cpu_quota_per_sec_of_threads,
systemd_cpu_quota_per_sec_magnitude_identity,
systemd_cpu_quota_unbounded_identity,
}
import product.placement_supply { HostIdentity }
import product.fabric.sanitation { CellId }
Expand All @@ -18,7 +19,7 @@ import gunbc.build_cache_instance { RunnerSlotIdentity, runner_slot_instance_nam
import gunbc.runner_slot_allocation {
fabric_execution_slot_identities,
gunbc_runner_slot_desired,
RunnerSlotCpuQuota, CpuQuotaResolved, CpuQuotaRefused,
RunnerSlotCpuQuota, CpuQuotaResolved, SlotCpuQuotaUnbounded,
runner_slot_membership,
CommittedSlotMember,
SlotOutsideCommittedPopulation,
Expand Down Expand Up @@ -204,31 +205,33 @@ fn fabric_cell_member_ownership(member: FabricCellMember) -> Optional<Ownership>
// address instead of drifting silently. The digest is over the values the boundary will carry.
fn fabric_cell_resource_boundary_state_digest() -> NonEmptyStr {
let desired = gunbc_runner_slot_desired()
match desired.cpu_quota {
CpuQuotaRefused { cause: c } =>
join(["cpu_quota_survey_refused=", c], "") as NonEmptyStr
fabric_cell_boundary_state_digest_of(
memory_max: to_string(byte_size_count(b: desired.memory_max)),
memory_high: to_string(byte_size_count(b: desired.memory_high)),
memory_swap_max: to_string(byte_size_count(b: desired.memory_swap_max)),
tasks_max: to_string(desired.tasks_max),
cpu_weight: to_string(desired.cpu_weight),
cpu_quota_per_sec: fabric_cell_desired_cpu_quota_identity(quota: desired.cpu_quota),
)
}

// THE QUOTA MEMBER IS THE IDENTITY THE OBSERVED SIDE PRODUCES, NOT A SECOND RENDERER AND NOT A
// SENTINEL. A resolved quota is the magnitude identity of systemd_cpu_quota_per_sec_of_threads:
// desired authors a thread count; observed reads CPUQuotaPerSecUSec as a formatted span, parses it,
// and supplies the same identity, so a six-thread cell and a host that printed `6s` compare equal
// as 6000000 microseconds. An UNBOUNDED quota is spelled by extdeps.systemd's
// systemd_cpu_quota_unbounded_identity -- the same constant the read-side identity returns for the
// CpuQuotaUnbounded arm a host that printed `infinity` parses to -- so a cell with no ceiling and a
// desired row with no ceiling compare equal through one spelling rather than a literal authored
// here beside the parser's.
fn fabric_cell_desired_cpu_quota_identity(quota: RunnerSlotCpuQuota) -> String {
match quota {
CpuQuotaResolved { threads: t } =>
fabric_cell_boundary_state_digest_of(
memory_max: to_string(byte_size_count(b: desired.memory_max)),
memory_high: to_string(byte_size_count(b: desired.memory_high)),
memory_swap_max: to_string(byte_size_count(b: desired.memory_swap_max)),
tasks_max: to_string(desired.tasks_max),
cpu_weight: to_string(desired.cpu_weight),
cpu_quota_per_sec: systemd_cpu_quota_per_sec_magnitude_identity(
usec: systemd_cpu_quota_per_sec_of_threads(threads: t),
),
)
systemd_cpu_quota_per_sec_magnitude_identity(usec: systemd_cpu_quota_per_sec_of_threads(threads: t))
SlotCpuQuotaUnbounded => systemd_cpu_quota_unbounded_identity
}
}

// THE QUOTA MEMBER IS THE MAGNITUDE IDENTITY OF systemd_cpu_quota_per_sec_of_threads, NOT A SECOND
// RENDERER AND NOT A SENTINEL. Desired authors a thread count; observed reads CPUQuotaPerSecUSec as
// a formatted span, parses it, and supplies the same identity. A six-thread cell and a host that
// printed `6s` compare equal as 6000000 microseconds. A refused survey has no magnitude and is not
// stuffed into this field: fabric_cell_resource_boundary_state_digest names the refusal on its own
// arm, and fabric_cell_address_standing_for_quota blocks every disposition so reconcile cannot
// treat that string as drift-to-repair.

// THE SPELLING IS A FUNCTION, BECAUSE THE OBSERVED SIDE HAS TO PRODUCE THE SAME ONE. The desired
// digest above is compared against a digest an OBSERVER builds out of what a host reported, and if
// each side spells the record itself then the two spellings are a fork: a reordered key or a
Expand Down Expand Up @@ -684,46 +687,24 @@ fn fabric_cell_address_disposition_ref(d: FabricCellAddressDisposition) -> Fabri
}
}

fn fabric_cell_address_standing_for_quota(
fn fabric_cell_address_standing(
slot: RunnerSlotIdentity,
d: FabricCellAddressDisposition,
quota: RunnerSlotCpuQuota,
) -> FabricCellAddressStanding {
match quota {
CpuQuotaRefused { cause: c } =>
StandingBlocked {
disposition: AddressUnreadable {
ref: fabric_cell_address_disposition_ref(d: d),
cause: join(["desired CPU quota refused: ", c], ""),
},
}
CpuQuotaResolved { threads: _ } =>
match d {
AddressUnreadable { ref: _, cause: _ } => StandingBlocked { disposition: d }
AddressContradicted { ref: _ } => StandingBlocked { disposition: d }
AddressMissingOnHost { ref: r } => StandingExplicitlyAbsent { ref: r }
AddressUnobserved { ref: _ } => StandingBlocked { disposition: d }
AddressPresent { ref: r, member: m } =>
if (m.state_digest as String) == (fabric_cell_desired_state_digest(slot: r.slot, address: r.address) as String) {
StandingSatisfied { ref: r }
} else {
StandingObservedDrift { ref: r, observed_state_digest: m.state_digest }
}
match d {
AddressUnreadable { ref: _, cause: _ } => StandingBlocked { disposition: d }
AddressContradicted { ref: _ } => StandingBlocked { disposition: d }
AddressMissingOnHost { ref: r } => StandingExplicitlyAbsent { ref: r }
AddressUnobserved { ref: _ } => StandingBlocked { disposition: d }
AddressPresent { ref: r, member: m } =>
if (m.state_digest as String) == (fabric_cell_desired_state_digest(slot: r.slot, address: r.address) as String) {
StandingSatisfied { ref: r }
} else {
StandingObservedDrift { ref: r, observed_state_digest: m.state_digest }
}
}
}

fn fabric_cell_address_standing(
slot: RunnerSlotIdentity,
d: FabricCellAddressDisposition,
) -> FabricCellAddressStanding {
fabric_cell_address_standing_for_quota(
slot: slot,
d: d,
quota: gunbc_runner_slot_desired().cpu_quota,
)
}

fn fabric_cell_address_standings(
readback: FabricCellReadback,
slot: RunnerSlotIdentity,
Expand Down
Loading