Skip to content

Parallelize independent worktree admission #1311

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
chubes4 merged 1 commit into from
Aug 30, 2026
Merged

Parallelize independent worktree admission #1311

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
56 changes: 44 additions & 12 deletions inc/Workspace/WorkspaceWorktreeLifecycle.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1231,8 +1231,8 @@ public function worktree_add_request( WorktreeAllocationRequest $request ): arra

// Fetch and demand planning only touch this primary. Keep them out of the
// global capacity critical section so unrelated repositories can prepare in
// parallel; capacity-changing checkout remains globally fenced. Bootstrap
// demand is reserved durably before its child processes run without locks.
// parallel. Admission reserves bounded demand atomically, then releases the
// global lock so repository-scoped checkout can proceed independently.
$this->worktree_add_progress($progress_callback, 'repo_preflight');
$preflight = WorkspaceMutationLock::with_repo(
$this->workspace_path,
Expand Down Expand Up @@ -1473,10 +1473,10 @@ private function worktree_capacity_dry_run( string $repo, string $branch, ?strin
/**
* Resolve the explicit global-capacity lock wait budget.
*
* Lock order is always global capacity first, then the repository lock. The
* global lock remains held through checkout and durable bootstrap reservation.
* Later admissions include that reservation while dependency children run
* without inheriting this lock descriptor.
* Lock order is global capacity admission first, then the repository lock.
* The global lock serializes inspect-and-reserve only; checkout remains
* repository-scoped. Later admissions include durable reservations while
* dependency children run without inheriting this lock descriptor.
*/
public static function worktree_capacity_wait_timeout_seconds( bool $bootstrap = true ): int {
$timeout = self::worktree_capacity_operation_timeout_seconds($bootstrap) + 60;
Expand Down Expand Up @@ -1521,9 +1521,10 @@ public static function worktree_capacity_aggregate_timeout_seconds( bool $bootst
}

/**
* Inspect, create, and reserve bootstrap demand while holding the workspace-
* wide capacity lock. A later admission includes the durable reservation while
* the dependency process runs outside mutation lock boundaries.
* Inspect and reserve demand under the workspace capacity lock, then create
* the worktree under the repository lock. Later admissions include the durable
* reservation while checkout and dependency processes run without holding the
* global lock.
*/
private function worktree_add_with_capacity_lock(
string $repo,
Expand Down Expand Up @@ -1865,6 +1866,21 @@ static function ( $row ): string {
);
}

$demand_plan['reservation_handle'] = $wt_handle;
$reserved = WorktreeContextInjector::reserve_capacity($this->workspace_path, $wt_handle, $demand_plan);
if ( is_wp_error($reserved) ) {
return $reserved;
}
if ( $capacity_lock instanceof WorkspaceMutationLock ) {
$released = $capacity_lock->release();
if ( is_wp_error($released) ) {
WorktreeContextInjector::release_capacity_reservation($this->workspace_path, $wt_handle);
return $released;
}
$capacity_lock = null;
}

try {
$repo_timeout = $this->worktree_operation_remaining_seconds($operation_deadline);
if ( $repo_timeout <= 0 ) {
return $this->worktree_operation_timeout('repo_lock_wait', $operation_timeout, $operation_started);
Expand Down Expand Up @@ -1955,6 +1971,7 @@ static function ( $row ): string {
$measurement_plan = $post_rebase_demand;
$post_rebase_demand = WorktreeBootstrapper::remaining_demand_after_materialization($post_rebase_demand);
$post_rebase_demand['allow_percentage_byte_floor_exception'] = $allow_percentage_byte_floor_exception;
$post_rebase_demand['reservation_handle'] = $wt_handle;
$this->worktree_add_progress($progress_callback, 'post_rebase_capacity_inspection');
$post_rebase_budget = $this->inspect_worktree_capacity($repo, $branch, $force, $post_rebase_demand);
$this->worktree_add_progress($progress_callback, 'post_rebase_artifact_reclamation');
Expand Down Expand Up @@ -1987,7 +2004,7 @@ static function ( $row ): string {
}

if ( $bootstrap ) {
$bootstrap_before_capacity = $this->inspect_worktree_capacity($repo, $branch, false, array());
$bootstrap_before_capacity = $this->inspect_worktree_capacity($repo, $branch, false, array( 'reservation_handle' => $wt_handle ));
$remaining_seconds = $this->worktree_operation_remaining_seconds($operation_deadline);
if ( $remaining_seconds <= 0 ) {
$recorded = $this->record_bootstrap_outcome($wt_handle, 'failed', array(), 'operation_timeout');
Expand All @@ -2003,7 +2020,7 @@ static function ( $row ): string {
if ( is_wp_error($response) ) {
return $response;
}
$after_capacity = $this->inspect_worktree_capacity($repo, $branch, false, array());
$after_capacity = $this->inspect_worktree_capacity($repo, $branch, false, array( 'reservation_handle' => $wt_handle ));
$response['capacity_evidence'] = WorktreeDemandCalibration::record_bootstrap($repo, $measurement_plan, $bootstrap_before_capacity, $after_capacity, true);
$response['bootstrap_noop_completed'] = true;
} else {
Expand Down Expand Up @@ -2075,6 +2092,9 @@ static function ( $row ): string {
$this->emit_workspace_changed('worktree_add', $repo, $wt_handle, $wt_path);

return $response;
} finally {
WorktreeContextInjector::release_capacity_reservation($this->workspace_path, $wt_handle);
}
}

/** Whether target-tree planning proves bootstrap has no dependency work. */
Expand Down Expand Up @@ -5769,7 +5789,19 @@ public function worktree_prune( bool $dry_run = false, mixed $until_budget = nul
* @return array<string,mixed>
*/
protected function inspect_worktree_capacity( string $repo, string $branch, bool $force, array $demand_plan ): array {
$reservations = WorktreeContextInjector::bootstrap_capacity_reservations();
$reservations = WorktreeContextInjector::capacity_reservations($this->workspace_path);
$exclude = (string) ( $demand_plan['reservation_handle'] ?? '' );
if ( '' !== $exclude && isset($reservations['by_handle'][ $exclude ]) ) {
$reservations['bytes'] = max(0, (int) $reservations['bytes'] - (int) $reservations['by_handle'][ $exclude ]['bytes']);
$reservations['inodes'] = max(0, (int) $reservations['inodes'] - (int) $reservations['by_handle'][ $exclude ]['inodes']);
$reservations['handles'] = array_values(
array_filter(
(array) $reservations['handles'],
static fn( string $handle ): bool => $handle !== $exclude
)
);
unset($reservations['by_handle'][ $exclude ]);
}
$demand_plan['bytes'] = max(0, (int) ( $demand_plan['bytes'] ?? 0 )) + (int) $reservations['bytes'];
$demand_plan['inodes'] = max(0, (int) ( $demand_plan['inodes'] ?? 0 )) + (int) $reservations['inodes'];
$demand_plan['capacity_reservations'] = $reservations;
Expand Down
186 changes: 182 additions & 4 deletions inc/Workspace/WorktreeContextInjector.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -271,6 +271,9 @@ private static function worktree_add_isolation_command( array $request, bool $te
*/
public const METADATA_OPTION = 'datamachine_worktree_metadata';

/** Workspace-local demand reserved after admission and before checkout completes. */
public const CAPACITY_RESERVATION_DIR = 'capacity-reservations';

/** Journal-only record written before a Git worktree mutation. */
public const CREATION_INTENT_KEY = 'creation_intent';

Expand Down Expand Up @@ -1194,7 +1197,7 @@ public static function bootstrap_readiness( ?array $metadata ): array {
/** Return dependency demand reserved by materialized worktrees still bootstrapping. */
public static function bootstrap_capacity_reservations(): array {
if ( ! function_exists('get_option') ) {
return array( 'bytes' => 0, 'inodes' => 0, 'handles' => array() );
return array( 'bytes' => 0, 'inodes' => 0, 'handles' => array(), 'by_handle' => array() );
}
// Capacity admission must not reuse an earlier request's option snapshot
// after another process has committed a reservation.
Expand All @@ -1205,6 +1208,7 @@ public static function bootstrap_capacity_reservations(): array {
$bytes = 0;
$inodes = 0;
$handles = array();
$by_handle = array();
foreach ( is_array($all) ? $all : array() as $handle => $metadata ) {
$bootstrap = (array) ($metadata['provisioning']['bootstrap'] ?? array());
$reservation = is_array($bootstrap['capacity_reservation'] ?? null) ? $bootstrap['capacity_reservation'] : null;
Expand All @@ -1213,11 +1217,185 @@ public static function bootstrap_capacity_reservations(): array {
if ( ! is_array($reservation) || 'running' !== ($bootstrap['outcome'] ?? null) || ( 'stale' === $coordinator['state'] && 'stale' === $child['state'] ) ) {
continue;
}
$bytes += max(0, (int) ($reservation['bytes'] ?? 0));
$inodes += max(0, (int) ($reservation['inodes'] ?? 0));
$demand = array(
'bytes' => max(0, (int) ($reservation['bytes'] ?? 0)),
'inodes' => max(0, (int) ($reservation['inodes'] ?? 0)),
);
$bytes += $demand['bytes'];
$inodes += $demand['inodes'];
$handles[] = (string) $handle;
$by_handle[ (string) $handle ] = $demand;
}
return array( 'bytes' => $bytes, 'inodes' => $inodes, 'handles' => $handles, 'by_handle' => $by_handle );
}

/**
* Return live bootstrap plus admitted-but-not-yet-materialized demand.
*
* @return array{bytes:int,inodes:int,handles:array<int,string>,by_handle:array<string,array{bytes:int,inodes:int}>}
*/
public static function capacity_reservations( string $workspace_path = '' ): array {
return self::merge_capacity_reservations(
self::bootstrap_capacity_reservations(),
self::admission_capacity_reservations($workspace_path)
);
}

/**
* Atomically reserve this handle's bounded demand so later admissions include it
* after the global capacity lock is released.
*
* @param array<string,mixed> $demand
*/
public static function reserve_capacity( string $workspace_path, string $handle, array $demand ): bool|\WP_Error {
$dir = self::capacity_reservation_dir($workspace_path);
if ( '' === $dir || '' === $handle ) {
return new \WP_Error(
'workspace_capacity_reservation_invalid_target',
'Capacity reservation requires a workspace path and worktree handle.',
array( 'status' => 400 )
);
}
if ( ! is_dir($dir) ) {
$created = function_exists('wp_mkdir_p')
? wp_mkdir_p($dir)
: @mkdir($dir, 0755, true); // phpcs:ignore WordPress.WP.AlternativeFunctions.file_system_operations_mkdir,WordPress.PHP.NoSilencedErrors.Discouraged -- Atomic local reservation setup rechecks the directory.
if ( ! $created && ! is_dir($dir) ) {
return new \WP_Error(
'workspace_capacity_reservation_create_failed',
sprintf('Failed to create capacity reservation directory: %s', $dir),
array( 'status' => 500 )
);
}
}

$path = $dir . '/' . self::capacity_reservation_filename($handle) . '.json';
$payload = array(
'handle' => $handle,
'bytes' => max(0, (int) ( $demand['bytes'] ?? 0 )),
'inodes' => max(0, (int) ( $demand['inodes'] ?? 0 )),
'coordinator' => self::bootstrap_owner(),
'reserved_at' => gmdate('c'),
);
$json = function_exists('wp_json_encode') ? wp_json_encode($payload) : json_encode($payload); // phpcs:ignore WordPress.WP.AlternativeFunctions.json_encode_json_encode -- Reservation files also run outside WordPress bootstrap.
$temporary = $path . '.' . bin2hex(random_bytes(6)) . '.tmp';
if ( false === file_put_contents($temporary, false === $json ? '{}' : (string) $json, LOCK_EX) ) { // phpcs:ignore WordPress.WP.AlternativeFunctions.file_system_operations_file_put_contents
return new \WP_Error(
'workspace_capacity_reservation_persist_failed',
sprintf('Failed to persist capacity reservation for "%s".', $handle),
array( 'status' => 500 )
);
}
if ( ! rename($temporary, $path) ) { // phpcs:ignore WordPress.WP.AlternativeFunctions.rename_rename
unlink($temporary); // phpcs:ignore WordPress.WP.AlternativeFunctions.unlink_unlink
return new \WP_Error(
'workspace_capacity_reservation_persist_failed',
sprintf('Failed to commit capacity reservation for "%s".', $handle),
array( 'status' => 500 )
);
}

return true;
}

/** Drop an admitted reservation after checkout, bootstrap handoff, or failure. */
public static function release_capacity_reservation( string $workspace_path, string $handle ): void {
$path = self::capacity_reservation_path($workspace_path, $handle);
if ( '' !== $path && is_file($path) ) {
unlink($path); // phpcs:ignore WordPress.WP.AlternativeFunctions.unlink_unlink
}
}

/**
* Return admitted demand that has not yet been converted to bootstrap reservation.
*
* @return array{bytes:int,inodes:int,handles:array<int,string>,by_handle:array<string,array{bytes:int,inodes:int}>}
*/
public static function admission_capacity_reservations( string $workspace_path ): array {
$dir = self::capacity_reservation_dir($workspace_path);
if ( '' === $dir || ! is_dir($dir) ) {
return array( 'bytes' => 0, 'inodes' => 0, 'handles' => array(), 'by_handle' => array() );
}
$files = glob($dir . '/*.json');
$bytes = 0;
$inodes = 0;
$handles = array();
$by_handle = array();
foreach ( false === $files ? array() : $files as $file ) {
// phpcs:ignore WordPress.WP.AlternativeFunctions.file_get_contents_file_get_contents,WordPress.PHP.NoSilencedErrors.Discouraged -- A vanished reservation file is an expected concurrent release race.
$data = json_decode((string) @file_get_contents($file), true);
if ( ! is_array($data) ) {
continue;
}
$coordinator = self::bootstrap_owner_state($data['coordinator'] ?? null);
if ( 'stale' === ( $coordinator['state'] ?? '' ) ) {
continue;
}
$handle = (string) ( $data['handle'] ?? '' );
if ( '' === $handle ) {
continue;
}
$demand = array(
'bytes' => max(0, (int) ( $data['bytes'] ?? 0 )),
'inodes' => max(0, (int) ( $data['inodes'] ?? 0 )),
);
$bytes += $demand['bytes'];
$inodes += $demand['inodes'];
$handles[] = $handle;
$by_handle[ $handle ] = $demand;
}

return array( 'bytes' => $bytes, 'inodes' => $inodes, 'handles' => $handles, 'by_handle' => $by_handle );
}

/**
* @param array{bytes?:int,inodes?:int,handles?:array<int,string>,by_handle?:array<string,array{bytes:int,inodes:int}>} $left
* @param array{bytes?:int,inodes?:int,handles?:array<int,string>,by_handle?:array<string,array{bytes:int,inodes:int}>} $right
* @return array{bytes:int,inodes:int,handles:array<int,string>,by_handle:array<string,array{bytes:int,inodes:int}>}
*/
private static function merge_capacity_reservations( array $left, array $right ): array {
$by_handle = array();
foreach ( array( $left, $right ) as $set ) {
foreach ( (array) ( $set['by_handle'] ?? array() ) as $handle => $demand ) {
$handle = (string) $handle;
if ( '' === $handle || ! is_array($demand) ) {
continue;
}
$existing = $by_handle[ $handle ] ?? array( 'bytes' => 0, 'inodes' => 0 );
$by_handle[ $handle ] = array(
'bytes' => max( (int) $existing['bytes'], max(0, (int) ( $demand['bytes'] ?? 0 )) ),
'inodes' => max( (int) $existing['inodes'], max(0, (int) ( $demand['inodes'] ?? 0 )) ),
);
}
}
$bytes = 0;
$inodes = 0;
foreach ( $by_handle as $demand ) {
$bytes += (int) $demand['bytes'];
$inodes += (int) $demand['inodes'];
}
return array( 'bytes' => $bytes, 'inodes' => $inodes, 'handles' => $handles );

return array(
'bytes' => $bytes,
'inodes' => $inodes,
'handles' => array_keys($by_handle),
'by_handle' => $by_handle,
);
}

private static function capacity_reservation_dir( string $workspace_path ): string {
$workspace_path = rtrim($workspace_path, '/');
return '' === $workspace_path ? '' : $workspace_path . '/.locks/' . self::CAPACITY_RESERVATION_DIR;
}

private static function capacity_reservation_path( string $workspace_path, string $handle ): string {
$dir = self::capacity_reservation_dir($workspace_path);
return '' === $dir || '' === $handle ? '' : $dir . '/' . self::capacity_reservation_filename($handle) . '.json';
}

private static function capacity_reservation_filename( string $handle ): string {
$handle = preg_replace('/[^a-zA-Z0-9._@-]/', '', $handle);
return trim( (string) $handle, '-.');
}

/** Capture a PID and OS-issued process identity for bootstrap ownership. */
Expand Down
5 changes: 5 additions & 0 deletions tests/workspace-capacity-lock-concurrency.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -800,6 +800,11 @@ static function () use ( $state, $ready ): string {
};
$lifecycle_source = (string) file_get_contents(dirname(__DIR__) . '/inc/Workspace/WorkspaceWorktreeLifecycle.php');
capacity_lock_assert(str_contains($lifecycle_source, "'workspace-capacity-admission', \$reuse"), 'Bootstrap resume must acquire global capacity admission before its repository lock.');
$capacity_fn = strpos($lifecycle_source, 'function worktree_add_with_capacity_lock');
$reserve_at = strpos($lifecycle_source, 'WorktreeContextInjector::reserve_capacity');
$release_at = strpos($lifecycle_source, '$capacity_lock->release()');
$create_at = strpos($lifecycle_source, '$this->worktree_add_locked(');
capacity_lock_assert(false !== $capacity_fn && false !== $reserve_at && false !== $release_at && false !== $create_at && $capacity_fn < $reserve_at && $reserve_at < $release_at && $release_at < $create_at, 'New worktree checkout must run only after demand is reserved and the global capacity lock is released.');
capacity_lock_assert(2400 === $policy::worktree_capacity_wait_timeout_seconds(true), 'Bootstrap admission wait must exceed the complete bounded operation lifecycle.');

$state = $workspace . '/capacity-state';
Expand Down
Loading
Loading