Skip to content
Closed
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
65 changes: 65 additions & 0 deletions Packages/CmuxGit/Sources/CmuxGit/GitProbeConcurrencyLimiter.swift
Original file line number Diff line number Diff line change
@@ -0,0 +1,65 @@
import Foundation

/// Bounds how many git-metadata probes parse index files at the same time.
///
/// Each tracked workspace panel schedules a detached probe that reads and
/// parses the entire `.git/index`. A burst — a workspace restore, the periodic
/// fallback refresh, or a flurry of file-system watch events — can schedule one
/// probe per panel at once. With dozens of panels that spawns dozens of
/// parallel parses and saturates every core, producing the multi-core CPU
/// spikes tracked in https://github.com/manaflow-ai/cmux/issues/4639.
///
/// This actor caps how many probes run concurrently. Callers wrap the parse in
/// ``run(_:)``; work beyond the cap suspends until a slot frees and resumes in
/// FIFO order. Construct one and share it across every probe:
///
/// ```swift
/// let limiter = GitProbeConcurrencyLimiter(maxConcurrent: 4)
/// let snapshot = await limiter.run { await parseIndex() }
/// ```
public actor GitProbeConcurrencyLimiter {
private let maxConcurrent: Int
private var available: Int
private var waiters: [CheckedContinuation<Void, Never>] = []

/// Creates a limiter allowing at most `maxConcurrent` probes to run at once.
///
/// - Parameter maxConcurrent: The concurrency ceiling. Values below 1 are
/// clamped to 1 so the limiter always makes forward progress.
public init(maxConcurrent: Int) {
let ceiling = max(1, maxConcurrent)
self.maxConcurrent = ceiling
self.available = ceiling
}
Comment on lines +20 to +33

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 maxConcurrent is stored as an instance property but is never read after init. The live ceiling is tracked entirely by available; maxConcurrent is dead state. It can be removed.

Suggested change
public actor GitProbeConcurrencyLimiter {
private let maxConcurrent: Int
private var available: Int
private var waiters: [CheckedContinuation<Void, Never>] = []
/// Creates a limiter allowing at most `maxConcurrent` probes to run at once.
///
/// - Parameter maxConcurrent: The concurrency ceiling. Values below 1 are
/// clamped to 1 so the limiter always makes forward progress.
public init(maxConcurrent: Int) {
let ceiling = max(1, maxConcurrent)
self.maxConcurrent = ceiling
self.available = ceiling
}
public actor GitProbeConcurrencyLimiter {
private var available: Int
private var waiters: [CheckedContinuation<Void, Never>] = []
/// Creates a limiter allowing at most `maxConcurrent` probes to run at once.
///
/// - Parameter maxConcurrent: The concurrency ceiling. Values below 1 are
/// clamped to 1 so the limiter always makes forward progress.
public init(maxConcurrent: Int) {
self.available = max(1, maxConcurrent)
}

Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!


/// Runs `body` once a slot is free, releasing the slot when it returns.
///
/// While `body` is suspended on its own `await`s the actor is free to admit
/// other callers up to the ceiling; only the slot accounting is serialized.
///
/// - Parameter body: The probe work to run while holding a slot.
/// - Returns: Whatever `body` returns.
public func run<T: Sendable>(_ body: @Sendable () async -> T) async -> T {
await acquire()
defer { release() }
return await body()
}

private func acquire() async {
if available > 0 {
available -= 1
return
}
// Resumed by `release()`, which hands off its slot directly without
// bumping `available`, so the slot count stays balanced.
await withCheckedContinuation { waiters.append($0) }

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1: Handle cancellation while waiting in acquire(): a cancelled task currently stays queued until resumed and then still runs body(), which wastes a slot and can perform unnecessary git probe I/O.

Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At Packages/CmuxGit/Sources/CmuxGit/GitProbeConcurrencyLimiter.swift, line 55:

<comment>Handle cancellation while waiting in `acquire()`: a cancelled task currently stays queued until resumed and then still runs `body()`, which wastes a slot and can perform unnecessary git probe I/O.</comment>

<file context>
@@ -0,0 +1,65 @@
+        }
+        // Resumed by `release()`, which hands off its slot directly without
+        // bumping `available`, so the slot count stays balanced.
+        await withCheckedContinuation { waiters.append($0) }
+    }
+
</file context>

}
Comment on lines +48 to +56

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Cancellation hole: queued probes run after their task is cancelled

withCheckedContinuation does not respond to Swift cooperative cancellation — a task whose Task.isCancelled flag is set while it is waiting in waiters stays parked until a slot is released and release() resumes it. When finally dequeued, body() runs to completion anyway (including the Data(contentsOf:) read inside workspaceMetadata). In a burst where several workspaces close while ~36 probes are queued, cancelled probes still absorb a slot, perform unnecessary disk I/O, and delay the legitimate probes behind them.

The fix is to wrap the suspension in withTaskCancellationHandler, removing the cancelled continuation from waiters and releasing the slot without running body().

Comment on lines +48 to +56

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🧹 Nitpick | 🔵 Trivial | ⚖️ Poor tradeoff

Optional enhancement: task cancellation support.

Cancelled tasks currently remain in the waiters queue until their slot is granted, potentially delaying other waiters. Consider adding withTaskCancellationHandler around the continuation to remove cancelled waiters from the queue:

await withTaskCancellationHandler {
    await withCheckedContinuation { waiters.append($0) }
} onCancel: {
    Task { await removeCancelledWaiter() }
}

However, this adds complexity (tracking which continuation to remove, ensuring atomicity). The current simple design is acceptable given that git probe work is lightweight and short-lived.

🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@Packages/CmuxGit/Sources/CmuxGit/GitProbeConcurrencyLimiter.swift` around
lines 48 - 56, Update acquire() to support task cancellation by wrapping the
withCheckedContinuation wait in a withTaskCancellationHandler so cancelled tasks
are removed from the waiters queue; specifically, when calling
withCheckedContinuation in GitProbeConcurrencyLimiter.acquire(), register the
continuation object in waiters, and in the onCancel handler remove that specific
continuation (ensure removal and append are atomic relative to release()) so
cancelled continuations don't linger, and adjust release() if necessary to
handle the case where a continuation was removed. Use the existing symbols
acquire(), waiters, and release() to locate where to add the cancellation
handler and removal logic.


private func release() {
if waiters.isEmpty {
available += 1
} else {
waiters.removeFirst().resume()
}
}
}
Original file line number Diff line number Diff line change
@@ -1,5 +1,27 @@
import Foundation

/// Lowercase-hex digit table used by ``lowercaseHexString(_:)``.
private let lowercaseHexDigits: [UInt8] = Array("0123456789abcdef".utf8)

/// Lowercase hex encoding of a byte collection, without `String(format:)`.
///
/// `String(format: "%02x", byte)` acquires a process-global locale lock
/// (`localeconv_l` / `os_unfair_lock`) on every call. Encoding a 20-byte
/// object ID for every index entry, across thousands of entries and many git
/// probes running in parallel, turned that lock into a multi-core CPU storm
/// (https://github.com/manaflow-ai/cmux/issues/4639). This table-based encoder
/// produces byte-identical output, allocation-light and lock-free.
func lowercaseHexString<Bytes: Collection>(_ bytes: Bytes) -> String
where Bytes.Element == UInt8 {
var ascii = [UInt8]()
ascii.reserveCapacity(bytes.count * 2)
for byte in bytes {
ascii.append(lowercaseHexDigits[Int(byte >> 4)])
ascii.append(lowercaseHexDigits[Int(byte & 0x0f)])
}
return String(decoding: ascii, as: UTF8.self)
}

extension GitMetadataService {
/// Compares the working tree against the parsed index to decide dirtiness,
/// returning the index signatures alongside.
Expand Down Expand Up @@ -86,9 +108,7 @@ extension GitMetadataService {
let mtimeNanoseconds = readBigEndianUInt32(bytes, at: offset + 12)
let mode = readBigEndianUInt32(bytes, at: offset + 24)
let size = readBigEndianUInt32(bytes, at: offset + 36)
let objectID = bytes[(offset + 40)..<(offset + 60)].map {
String(format: "%02x", $0)
}.joined()
let objectID = lowercaseHexString(bytes[(offset + 40)..<(offset + 60)])
let flags = readBigEndianUInt16(bytes, at: offset + 60)
let pathLength = Int(flags & 0x0fff)
let hasExtendedFlags = version >= 3 && (flags & 0x4000) != 0
Expand Down Expand Up @@ -157,7 +177,7 @@ extension GitMetadataService {
}
}

let checksum = bytes[(bytes.count - 20)..<bytes.count].map { String(format: "%02x", $0) }.joined()
let checksum = lowercaseHexString(bytes[(bytes.count - 20)..<bytes.count])
return GitIndexSnapshot(
entries: entries,
signature: checksum,
Expand Down Expand Up @@ -198,7 +218,8 @@ extension GitMetadataService {
appendString(entry.objectID)
appendByte(0)
}
return String(format: "%016llx", CUnsignedLongLong(hash))
let hashBytes = (0..<8).map { UInt8(truncatingIfNeeded: hash >> (8 * (7 - $0))) }
return lowercaseHexString(hashBytes)
}

/// The submodule's currently checked-out commit for a parent gitlink entry,
Expand Down Expand Up @@ -251,7 +272,7 @@ extension GitMetadataService {
guard let data = try? Data(contentsOf: indexURL), data.count >= 20 else {
return nil
}
return data.suffix(20).map { String(format: "%02x", $0) }.joined()
return lowercaseHexString(data.suffix(20))
}

/// Decodes a git index v4 path strip-length varint, advancing `offset`.
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,85 @@
import Foundation
import Testing
@testable import CmuxGit

/// Records how many bodies run between ``enter()`` and ``leave()`` at once so a
/// test can assert the limiter never exceeds its ceiling.
private actor ConcurrencyTracker {
private(set) var current = 0
private(set) var peak = 0
private(set) var completed = 0

func enter() {
current += 1
peak = max(peak, current)
}

func leave() {
current -= 1
completed += 1
}
}

@Suite struct GitProbeConcurrencyLimiterTests {
/// 60 tasks through a ceiling of 3 must never run more than 3 bodies at
/// once, and every task must still complete.
@Test func capsConcurrencyAndCompletesEveryTask() async {
let limiter = GitProbeConcurrencyLimiter(maxConcurrent: 3)
let tracker = ConcurrencyTracker()
await withTaskGroup(of: Void.self) { group in
for _ in 0..<60 {
group.addTask {
await limiter.run {
await tracker.enter()
await Task.yield()
await Task.yield()
await tracker.leave()
}
}
}
}
#expect(await tracker.peak <= 3)
#expect(await tracker.peak >= 1)
#expect(await tracker.completed == 60)
}

/// A ceiling of 1 strictly serializes: despite yields, no two bodies are
/// ever in flight together.
@Test func ceilingOfOneSerializes() async {
let limiter = GitProbeConcurrencyLimiter(maxConcurrent: 1)
let tracker = ConcurrencyTracker()
await withTaskGroup(of: Void.self) { group in
for _ in 0..<25 {
group.addTask {
await limiter.run {
await tracker.enter()
await Task.yield()
await tracker.leave()
}
}
}
}
#expect(await tracker.peak == 1)
#expect(await tracker.completed == 25)
}

/// A non-positive ceiling is clamped to 1 rather than deadlocking on zero
/// available slots.
@Test func clampsNonPositiveCeilingToOne() async {
let limiter = GitProbeConcurrencyLimiter(maxConcurrent: 0)
let tracker = ConcurrencyTracker()
await withTaskGroup(of: Void.self) { group in
for _ in 0..<10 {
group.addTask {
await limiter.run {
await tracker.enter()
await Task.yield()
await tracker.leave()
}
}
}
}
#expect(await tracker.peak == 1)
#expect(await tracker.completed == 10)
}
}
51 changes: 51 additions & 0 deletions Packages/CmuxGit/Tests/CmuxGitTests/HexEncodingTests.swift
Original file line number Diff line number Diff line change
@@ -0,0 +1,51 @@
import Foundation
import Testing
@testable import CmuxGit

/// Verifies the table-based hex encoder that replaced per-byte
/// `String(format: "%02x")` in the git index parser produces byte-identical
/// output. The replacement removed a process-global locale lock that turned
/// parallel git probes into a multi-core CPU storm
/// (https://github.com/manaflow-ai/cmux/issues/4639).
@Suite struct HexEncodingTests {
@Test func matchesPrintfFormatForEveryByteValue() {
for value in UInt8.min...UInt8.max {
#expect(lowercaseHexString([value]) == String(format: "%02x", value))
}
}

@Test func matchesPrintfFormatForObjectIDWidthSequence() {
// 20 bytes is the SHA-1 object-ID width formatted per index entry,
// the hot path in gitIndexSnapshot.
let bytes: [UInt8] = (0..<20).map { UInt8(($0 * 37 + 5) & 0xff) }
#expect(lowercaseHexString(bytes) == bytes.map { String(format: "%02x", $0) }.joined())
#expect(lowercaseHexString(bytes).count == 40)
}

@Test func encodesEmptyCollectionAsEmptyString() {
#expect(lowercaseHexString([UInt8]()) == "")
}

@Test func matchesPrintfFormatForArraySliceAndDataSuffix() {
// The real call sites pass an ArraySlice (object ID / checksum) and a
// Data.suffix (file signature); both must encode identically.
let bytes: [UInt8] = [0x00, 0x0f, 0xa0, 0xff, 0x10, 0x01, 0x7e]
let slice = bytes[1..<6]
#expect(lowercaseHexString(slice) == slice.map { String(format: "%02x", $0) }.joined())
let data = Data(bytes)
#expect(
lowercaseHexString(data.suffix(4))
== data.suffix(4).map { String(format: "%02x", $0) }.joined()
)
}

@Test func matchesPrintfFormatForUInt64BigEndianHash() {
// The FNV content signature was %016llx of a UInt64; the replacement
// encodes the 8 big-endian bytes and must match for full-width values.
let values: [UInt64] = [0, 1, 0x0123_4567_89ab_cdef, .max, 0xff]
for value in values {
let bytes = (0..<8).map { UInt8(truncatingIfNeeded: value >> (8 * (7 - $0))) }
#expect(lowercaseHexString(bytes) == String(format: "%016llx", CUnsignedLongLong(value)))
}
}
}
46 changes: 26 additions & 20 deletions Sources/TabManager.swift
Original file line number Diff line number Diff line change
Expand Up @@ -1200,6 +1200,10 @@ class TabManager: ObservableObject {
// slugs) off the main actor. Stateless; the reads are pure functions of the
// directory argument.
private let gitMetadataService: GitMetadataService
// Caps how many git-index parses run at once so a burst of per-panel probes
// (workspace restore, fallback timer, file-watch flurry) can't saturate
// every core (https://github.com/manaflow-ai/cmux/issues/4639).
private let gitProbeLimiter = GitProbeConcurrencyLimiter(maxConcurrent: 4)

// Resolves GitHub PR badges (slug resolution, REST fetch, candidate
// matching). Stateless; the repo cache stays here in
Expand Down Expand Up @@ -3005,29 +3009,31 @@ class TabManager: ObservableObject {
private nonisolated func initialWorkspaceGitMetadataSnapshot(
for directory: String
) async -> InitialWorkspaceGitMetadataSnapshot {
let metadata = await gitMetadataService.workspaceMetadata(for: directory)
guard metadata.isRepository else {
await gitProbeLimiter.run {
let metadata = await self.gitMetadataService.workspaceMetadata(for: directory)
guard metadata.isRepository else {
return InitialWorkspaceGitMetadataSnapshot(
isRepository: false,
branch: nil,
isDirty: false,
indexSignature: nil,
indexContentSignature: nil,
headSignature: nil,
pullRequest: .notFound
)
}

let branch = GitMetadataService.normalizedBranchName(metadata.branch)
return InitialWorkspaceGitMetadataSnapshot(
isRepository: false,
branch: nil,
isDirty: false,
indexSignature: nil,
indexContentSignature: nil,
headSignature: nil,
pullRequest: .notFound
isRepository: true,
branch: branch,
isDirty: metadata.isDirty,
indexSignature: metadata.indexSignature,
indexContentSignature: metadata.indexContentSignature,
headSignature: metadata.headSignature,
pullRequest: branch == nil ? .notFound : .deferred
)
}

let branch = GitMetadataService.normalizedBranchName(metadata.branch)
return InitialWorkspaceGitMetadataSnapshot(
isRepository: true,
branch: branch,
isDirty: metadata.isDirty,
indexSignature: metadata.indexSignature,
indexContentSignature: metadata.indexContentSignature,
headSignature: metadata.headSignature,
pullRequest: branch == nil ? .notFound : .deferred
)
}

func requestBackgroundWorkspaceLoad(for workspaceId: UUID) {
Expand Down
Loading