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
2 changes: 1 addition & 1 deletion cli/cli/src/main.rs
Original file line number Diff line number Diff line change
Expand Up @@ -327,7 +327,7 @@ fn run() -> i32 {
}
}
};
dx_process::configure_real(&workspace);
dx_process::configure_real_in(&workspace, dx_plan::slots::slot_idle_secs(&workspace));
dx_plan::plan::set_rules_dx_repo(&workspace);
if let Some(note) = dx_plan::args::bare_scope_note(&invocation, &workspace, &cwd) {
eprintln!("{note}");
Expand Down
94 changes: 89 additions & 5 deletions cli/plan/src/slots.rs
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,45 @@ pub const BUILD_SLOTS_DISABLED_CODE: &str = "build_slots_disabled";
/// Notice code for the disk reclaim a build slot runs before a build.
pub const DISK_LOW_CODE: &str = "disk_low";

/// Seconds a `dx worktree` slot's Bazel server may sit idle before it exits.
pub const SLOT_MAX_IDLE_SECS: u64 = 300;

/// The `--max_idle_secs` for the server of `workspace`: short inside a `dx worktree` slot, `None` elsewhere.
pub fn slot_idle_secs(workspace: &Path) -> Option<u64> {
dx_worktree::lease_for_dir(workspace).map(|_| SLOT_MAX_IDLE_SECS)
}

/// Sets the memory one build slot is planned to use: MiB, or a count with an `m` or `g` suffix.
pub const SLOT_MEMORY_ENV: &str = "DX_BUILD_SLOT_MEMORY";

/// Reads a slot memory size in bytes; `None` for zero or anything unreadable.
pub fn parse_slot_memory(value: &str) -> Option<u64> {
let value = value.trim().to_ascii_lowercase();
let (digits, unit) = match value.chars().last()? {
'g' => (&value[..value.len() - 1], 30),
'm' => (&value[..value.len() - 1], 20),
_ => (value.as_str(), 20),
};
if digits.is_empty() || !digits.bytes().all(|b| b.is_ascii_digit()) {
return None;
}
let bytes = digits.parse::<u64>().ok()?.checked_mul(1 << unit)?;
(bytes > 0).then_some(bytes)
}

/// Memory one local build slot is planned to use: `DX_BUILD_SLOT_MEMORY` when set and valid.
pub fn slot_bytes(env_get: &dyn Fn(&str) -> Option<String>) -> u64 {
env_get(SLOT_MEMORY_ENV)
.and_then(|value| parse_slot_memory(&value))
.unwrap_or(BYTES_PER_SLOT)
}

/// `bytes` rescaled so that dividing by the default slot size counts slots of `per_slot` bytes.
fn per_slot_scaled(bytes: u64, per_slot: u64) -> u64 {
let scaled = u128::from(bytes) * u128::from(BYTES_PER_SLOT) / u128::from(per_slot);
u64::try_from(scaled).unwrap_or(u64::MAX)
}

pub mod pressure;

use pressure::{HostPressure, NoSignal, PRESSURE_WAIT, Pressure, PressurePause};
Expand Down Expand Up @@ -345,7 +384,12 @@ pub fn share_report(
let dir = slots_dir(env_get)?;
let memory = available_memory()?;
let reserved = nested_reserve(workspace, env_get);
let size = sizing(setting, memory, reserved)?;
let per_slot = slot_bytes(env_get);
let size = sizing(
setting,
per_slot_scaled(memory, per_slot),
per_slot_scaled(reserved, per_slot),
)?;
let cpus = std::thread::available_parallelism().map_or(1, |cpus| cpus.get());
let cpus = u64::try_from(cpus).unwrap_or(1);
// `usize::MAX` is no slot, so `held` counts every held slot plus the new run.
Expand Down Expand Up @@ -1125,6 +1169,7 @@ pub struct Gate {
reclaimer: Option<Reclaimer>,
reaper: Option<Reaper>,
wait: Option<Duration>,
slot_bytes: u64,
}

/// The time source a gate paces its pressure pause by.
Expand Down Expand Up @@ -1186,6 +1231,15 @@ impl Gate {
reclaimer: None,
reaper: None,
wait: None,
slot_bytes: BYTES_PER_SLOT,
}
}

/// Plans each local slot to use `bytes` of memory.
pub fn slot_memory(self, bytes: u64) -> Self {
Gate {
slot_bytes: bytes,
..self
}
}

Expand Down Expand Up @@ -1214,7 +1268,7 @@ impl Gate {
let target = if self.remote {
BYTES_PER_REMOTE_SLOT
} else {
BYTES_PER_SLOT
self.slot_bytes
};
Some((reaper.0)(&keep, target))
}
Expand Down Expand Up @@ -1309,11 +1363,16 @@ impl Gate {
let dir = slots_dir(env_get)?;
let memory = available_memory()?;
let reserved = nested_reserve(workspace, env_get);
let per_slot = slot_bytes(env_get);
let Sizing {
count,
pool,
extras,
} = sizing(setting, memory, reserved)?;
} = sizing(
setting,
per_slot_scaled(memory, per_slot),
per_slot_scaled(reserved, per_slot),
)?;
let read = dx_process::rc::read_file;
let read_user_rc = dx_process::RcPolicy::from_env(env_get, workspace, &read).read_user_rc;
let startup = own_startup_options(args);
Expand Down Expand Up @@ -1348,11 +1407,12 @@ impl Gate {
warm,
..Gate::new(dir, count, workspace.display().to_string(), json)
.reserving(pool)
.slot_memory(per_slot)
.extra(extras, remote)
.with_pressure(Arc::new(HostPressure::new(if remote {
BYTES_PER_REMOTE_SLOT
} else {
BYTES_PER_SLOT
per_slot
})))
.sharing(cpus, memory, reserved)
.reclaiming(move || dx_worktree::reclaim_for_build(&reclaim_in))
Expand Down Expand Up @@ -1412,7 +1472,9 @@ impl Gate {
remote: self.remote,
pause: PressurePause::new(PRESSURE_WAIT),
now: &*self.clock.0,
deadline: self.wait.and_then(|wait| (self.clock.0)().checked_add(wait)),
deadline: self
.wait
.and_then(|wait| (self.clock.0)().checked_add(wait)),
};
let result = acquire_reserving(
&self.dir,
Expand Down Expand Up @@ -2627,6 +2689,28 @@ mod tests {
}
}

#[test]
fn slot_memory_parses_sizes_and_scales_counts() {
assert_eq!(parse_slot_memory("4g"), Some(4 << 30));
assert_eq!(parse_slot_memory(" 512M "), Some(512 << 20));
assert_eq!(parse_slot_memory("2048"), Some(2048 << 20));
for bad in ["", "0", "g", "-1g", "1.5g", "4x"] {
assert_eq!(parse_slot_memory(bad), None, "{bad}");
}
let env = |value: &'static str| {
move |name: &str| (name == SLOT_MEMORY_ENV).then(|| value.to_owned())
};
assert_eq!(slot_bytes(&env("6g")), 6 << 30);
assert_eq!(slot_bytes(&env("nope")), BYTES_PER_SLOT);
assert_eq!(slot_bytes(&|_: &str| None), BYTES_PER_SLOT);
let memory = per_slot_scaled(24 << 30, 6 << 30);
assert_eq!(slot_count(memory, 0), 4);
assert_eq!(
slot_count(per_slot_scaled(24 << 30, 1 << 30), 0),
MAX_SLOTS.min(24)
);
}

#[test]
fn available_memory_text_parses() {
let text = "MemTotal: 4096 kB\nMemAvailable: 2048 kB\n";
Expand Down
9 changes: 6 additions & 3 deletions cli/process/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -33,8 +33,8 @@ mod stderr_tail;
pub use rc::{
DISK_CACHE_GC_MAX_AGE, DISK_CACHE_GC_MAX_SIZE, LocalShare, PRESET_LOCAL_RESOURCES, RcColor,
RcPolicy, ShareToken, StartupPaths, VerbMask, build_verb_options, clear_share,
clear_share_source, configure, configure_real, current, default_disk_cache, repository_caches,
set_share, set_share_source, share, startup_output_paths, user_cache_dir,
clear_share_source, configure, configure_real, configure_real_in, current, default_disk_cache,
repository_caches, set_share, set_share_source, share, startup_output_paths, user_cache_dir,
};
pub use runner::{BinaryRunner, Holding};
pub use stderr_tail::last_bazel_stderr;
Expand Down Expand Up @@ -397,11 +397,14 @@ pub fn head_settings(repo: &str) -> Vec<String> {

fn launcher_head(policy: &RcPolicy, startup_options: &[String]) -> Vec<String> {
let opts = policy.startup_opts();
let mut argv = Vec::with_capacity(1 + opts.len() + startup_options.len());
let mut argv = Vec::with_capacity(2 + opts.len() + startup_options.len());
argv.push(launcher_argv0().to_owned());
for opt in opts {
argv.push((*opt).to_owned());
}
if let Some(secs) = policy.slot_idle_secs {
argv.push(format!("--max_idle_secs={secs}"));
}
argv.extend(startup_options.iter().cloned());
argv
}
Expand Down
30 changes: 28 additions & 2 deletions cli/process/src/rc.rs
Original file line number Diff line number Diff line change
Expand Up @@ -175,6 +175,8 @@ pub struct RcPolicy {
pub disk_cache_gc: bool,
/// The gc idle delay dx adds so cleanup can run before an idle server exits.
pub gc_idle_delay_secs: Option<u64>,
/// The `--max_idle_secs` dx starts a worktree slot's server with.
pub slot_idle_secs: Option<u64>,
/// Where an rc file sets `--color`, which then wins over `NO_COLOR`.
pub rc_color: RcColor,
}
Expand All @@ -188,6 +190,7 @@ static HERMETIC: RcPolicy = RcPolicy {
local_memory: VerbMask::ALL,
disk_cache_gc: false,
gc_idle_delay_secs: None,
slot_idle_secs: None,
rc_color: RcColor::NONE,
};

Expand Down Expand Up @@ -288,6 +291,16 @@ impl RcPolicy {
env_get: &dyn Fn(&str) -> Option<String>,
workspace: &Path,
read_file: &dyn Fn(&Path) -> Option<String>,
) -> RcPolicy {
Self::from_env_in(env_get, workspace, read_file, None)
}

/// `from_env` for a workspace whose server idles out after `slot_idle_secs`, when given.
pub fn from_env_in(
env_get: &dyn Fn(&str) -> Option<String>,
workspace: &Path,
read_file: &dyn Fn(&Path) -> Option<String>,
slot_idle_secs: Option<u64>,
) -> RcPolicy {
let in_bazel_test = env_get("BAZEL_TEST").is_some_and(|value| !value.is_empty());
if in_bazel_test {
Expand Down Expand Up @@ -318,11 +331,17 @@ impl RcPolicy {
};
let disk_cache_gc =
disk_cache.is_some() && !rc_sets_disk_cache_gc(&roots, workspace, read_file);
let slot_idle_secs = slot_idle_secs.filter(|secs| *secs > 0 && !ci);
let gc_idle_delay_secs =
if ci || rc_sets_flag(&roots, workspace, read_file, GC_IDLE_DELAY_FLAG) {
None
} else {
rc_min_max_idle_secs(&roots, workspace, read_file).and_then(gc_idle_delay_for)
let rc_idle = rc_min_max_idle_secs(&roots, workspace, read_file);
let idle = match (rc_idle, slot_idle_secs) {
(Some(rc), Some(slot)) => Some(rc.min(slot)),
(rc, slot) => rc.or(slot),
};
idle.and_then(gc_idle_delay_for)
};
RcPolicy {
read_user_rc: true,
Expand All @@ -332,6 +351,7 @@ impl RcPolicy {
local_memory: VerbMask::where_rc_leaves_free(&lines, "memory"),
disk_cache_gc,
gc_idle_delay_secs,
slot_idle_secs,
rc_color: RcColor::from_lines(&lines),
}
}
Expand Down Expand Up @@ -438,10 +458,16 @@ pub fn configure(policy: RcPolicy) {

/// Sets the policy for this process from the real environment and files.
pub fn configure_real(workspace: &Path) {
configure(RcPolicy::from_env(
configure_real_in(workspace, None);
}

/// `configure_real` for a workspace whose server idles out after `slot_idle_secs`, when given.
pub fn configure_real_in(workspace: &Path, slot_idle_secs: Option<u64>) {
configure(RcPolicy::from_env_in(
&|name| std::env::var(name).ok(),
workspace,
&read_file,
slot_idle_secs,
));
}

Expand Down
22 changes: 22 additions & 0 deletions cli/process/src/rc_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -338,6 +338,7 @@ fn local_policy() -> RcPolicy {
local_memory: VerbMask::ALL,
disk_cache_gc: true,
gc_idle_delay_secs: None,
slot_idle_secs: None,
rc_color: RcColor::NONE,
}
}
Expand Down Expand Up @@ -1186,3 +1187,24 @@ fn an_rc_color_choice_counts_only_when_its_section_and_config_are_active() {
assert!(chosen(nested, &["bazel", "test", "--config=outer"]));
assert!(!chosen("test --color=yes\n", &["bazel", "build", "//..."]));
}

#[test]
fn a_slot_idle_adds_a_startup_flag_and_an_idle_delay() {
let policy = RcPolicy::from_env_in(
&|name| (name == "HOME").then(|| "/home/u".to_owned()),
Path::new("/ws"),
&|_| None,
Some(300),
);
assert_eq!(policy.slot_idle_secs, Some(300));
assert_eq!(policy.gc_idle_delay_secs, Some(100));
let head = startup_argv_with(&policy, &[]);
assert_eq!(head.last().map(String::as_str), Some("--max_idle_secs=300"));
let plain = RcPolicy::from_env_in(&|_| None, Path::new("/ws"), &|_| None, None);
assert_eq!(plain.slot_idle_secs, None);
assert!(
!startup_argv_with(&plain, &[])
.iter()
.any(|a| a.starts_with("--max_idle_secs"))
);
}
8 changes: 6 additions & 2 deletions cli/worktree/src/hook.rs
Original file line number Diff line number Diff line change
Expand Up @@ -315,7 +315,7 @@ pub fn ignored(kind: HookKind, input: &Value) -> bool {
&& Dialect::of(input) != Dialect::Claude
}

/// Marks the slots of a finished subagent as held by a gone holder.
/// Marks the slots of a finished subagent as held by a gone holder and stops their Bazel servers.
pub fn stop(
pool: &Pool,
host: &dyn Host,
Expand All @@ -327,7 +327,7 @@ pub fn stop(
}
let agent = field(input, "agent_id").ok_or_else(|| missing(HookKind::Stop, "agent_id"))?;
let session = who.session.clone().or_else(|| field(input, "session_id"));
let _lock = pool.lock()?;
let lock = pool.lock()?;
let mut marked = Vec::new();
for mut record in pool.records() {
if record.held
Expand All @@ -341,6 +341,10 @@ pub fn stop(
marked.push(record.slot);
}
}
drop(lock);
for slot in &marked {
ops::stop_server(host, &pool.slot_path(*slot));
}
Ok(marked)
}

Expand Down
2 changes: 2 additions & 0 deletions cli/worktree/src/tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -3628,10 +3628,12 @@ mod hooks {
.is_empty()
);
let stop = serde_json::json!({"agent_id": "a1"});
host.serve(900, &pool.slot_path(1));
assert_eq!(
hook::stop(&pool, &host, &first, &stop).expect("stop"),
vec![1]
);
assert_eq!(host.shutdown_paths(), vec![pool.slot_path(1)]);
let row = ops::list(&pool, &host, false).expect("list").rows.remove(0);
assert_eq!(
row.state,
Expand Down
4 changes: 3 additions & 1 deletion docs/cli/commands/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -100,7 +100,8 @@ When `CLAUDECODE`, `CODEX_THREAD_ID`, `CURSOR_AGENT`, `GEMINI_CLI`, `AI_AGENT`,
Commands that run Bazel wait for a build slot, so parallel runs on one machine
do not run out of memory. Slots are shared by every worktree on the machine.
The count is the machine memory, or the cgroup limit if lower, divided by 3 GiB.
It is at least 1 and at most 16.
It is at least 1 and at most 16. `DX_BUILD_SLOT_MEMORY` replaces the 3 GiB:
MiB, or a number ending in `m` or `g`. An unreadable or zero value is ignored.

```sh
tools/dx build //... --build-slots 4
Expand Down Expand Up @@ -150,6 +151,7 @@ A flag on the command line wins over these defaults.
| `DX_OUTPUT=<mode>` | Default for `--output`. |
| `DX_BUILD_SLOTS=auto\|<n>\|off` | Default for `--build-slots`. |
| `DX_BUILD_SLOTS_WAIT=<duration>` | Default for `--build-slots-wait`. |
| `DX_BUILD_SLOT_MEMORY=<size>` | Memory planned per build slot, default `3g`. |
| `NO_COLOR=<non-empty>` | Disable color in `auto` mode. |
| `BUILD_WORKSPACE_DIRECTORY=<dir>` | Workspace start under `bazel run`. |
| `BUILD_WORKING_DIRECTORY=<dir>` | Current directory under `bazel run`, for `--here` and a relative `worktree release <path>`. |
Expand Down
7 changes: 6 additions & 1 deletion docs/cli/commands/worktree.md
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,11 @@ Another holder's slot is freed only when it is reclaimable or with `--force`. It
`slot_busy` while a `dx` command runs in the slot. On an external worktree of
this repository it removes the checkout and keeps the branch.

Commands that run Bazel inside a slot start its server with
`--max_idle_secs=300`, so an idle slot's server exits after five minutes.
`--build-slots <n>` limits how many run at once and `DX_BUILD_SLOT_MEMORY` sets
the memory each is planned to use.

The working-directory check reads `/proc` on Linux only. It does not see other
users' processes, processes hidden by `hidepid` or processes in other PID
namespaces, and each check of a held slot scans every process. Any process whose
Expand Down Expand Up @@ -168,7 +173,7 @@ Cursor or Gemini CLI. They ignore `DX_*` variables and print text.
| --- | --- | --- |
| `create` | `name`, `cwd` | Leases a slot on branch `worktree-<name>` and prints the path. |
| `remove` | `worktree_path` | Releases that slot. Paths outside the pool are ignored. |
| `stop` | `agent_id` | Marks the slots that agent holds as held by a gone holder. |
| `stop` | `agent_id` | Marks the slots that agent holds as held by a gone holder and stops their Bazel servers. |
| `guard` | `tool_input.command` | Exits `2` for `git worktree add`, `remove` or `prune`. Exits `2` for `git checkout`, `switch`, `reset`, `rebase`, `fetch`, `pull`, `stash` or `commit`, `bazel`, and `dx build`, `test` or `check` in a slot another agent holds. Exits `2` for the same commands in the primary checkout when the caller holds a slot; the message names the slot. Read-only git, `gh` and `dx worktree add`, `release` and `list` stay allowed. |

`create`, `remove` and `stop` exit `1` on any failure, including empty or
Expand Down
Loading