From 87859e3a22e76f77a58f176a8dbce9951923a7ff Mon Sep 17 00:00:00 2001 From: rogerneal Date: Sun, 2 Aug 2026 20:47:00 -0400 Subject: [PATCH] Reap completed exec host state in the Linux runtime helper. Drop ExitMonitor bookkeeping after exit and remove per-exec process and waiter entries so long-lived container-runtime-linux helpers do not grow with every finished command. --- Package.swift | 6 ++ .../Runtime/RuntimeClient/ExitMonitor.swift | 11 ++- .../RuntimeLinux/Server/RuntimeService.swift | 90 ++++++++++++++----- .../ExitMonitorTests.swift | 89 ++++++++++++++++++ 4 files changed, 170 insertions(+), 26 deletions(-) create mode 100644 Tests/ContainerRuntimeClientTests/ExitMonitorTests.swift diff --git a/Package.swift b/Package.swift index d46548cae..ec7b3eed2 100644 --- a/Package.swift +++ b/Package.swift @@ -412,6 +412,12 @@ let package = Package( ], path: "Sources/Services/Runtime/RuntimeClient" ), + .testTarget( + name: "ContainerRuntimeClientTests", + dependencies: [ + "ContainerRuntimeClient", + ] + ), .target( name: "ContainerResource", dependencies: [ diff --git a/Sources/Services/Runtime/RuntimeClient/ExitMonitor.swift b/Sources/Services/Runtime/RuntimeClient/ExitMonitor.swift index 6ac2fa9e6..4462f5cc0 100644 --- a/Sources/Services/Runtime/RuntimeClient/ExitMonitor.swift +++ b/Sources/Services/Runtime/RuntimeClient/ExitMonitor.swift @@ -48,8 +48,7 @@ public actor ExitMonitor { if let task = self.runningTasks[id] { task.cancel() } - exitCallbacks.removeValue(forKey: id) - runningTasks.removeValue(forKey: id) + finishTracking(id: id) } /// Register long running work so that the monitor invokes @@ -86,6 +85,14 @@ public actor ExitMonitor { self.log?.error("WaitHandler for \(id) threw error \(String(describing: error))") try? await onExit(id, ExitStatus(exitCode: -1)) } + // Drop callback/task entries once the exit path finishes so completed + // work does not retain host state indefinitely (apple/container#2057). + await self.finishTracking(id: id) } } + + private func finishTracking(id: String) { + exitCallbacks.removeValue(forKey: id) + runningTasks.removeValue(forKey: id) + } } diff --git a/Sources/Services/RuntimeLinux/Server/RuntimeService.swift b/Sources/Services/RuntimeLinux/Server/RuntimeService.swift index 948a65603..e2bc239ac 100644 --- a/Sources/Services/RuntimeLinux/Server/RuntimeService.swift +++ b/Sources/Services/RuntimeLinux/Server/RuntimeService.swift @@ -45,6 +45,7 @@ public actor RuntimeService { private let monitor: ExitMonitor private let eventLoopGroup: any EventLoopGroup private var waiters: [String: ExitWaiter] = [:] + private var completedExits: [String: ExitStatus] = [:] private let lock: AsyncLock = AsyncLock() private let log: Logging.Logger private var state: State = .created @@ -58,10 +59,13 @@ public actor RuntimeService { class ExitWaiter { public var exitStatus: ExitStatus? = nil public var continuations: [CheckedContinuation] = [] + /// True once at least one waiter has received the exit status. + public private(set) var didDeliverExit = false public func wait(_ cc: CheckedContinuation) { if let exitStatus = exitStatus { // `doExit` has already been called for this waiter + didDeliverExit = true cc.resume(returning: exitStatus) return } @@ -69,10 +73,13 @@ public actor RuntimeService { } public func doExit(exitStatus: ExitStatus) { + if !continuations.isEmpty { + didDeliverExit = true + } for cc in continuations { cc.resume(returning: exitStatus) } - + continuations.removeAll() self.exitStatus = exitStatus } } @@ -440,18 +447,17 @@ public actor RuntimeService { onExit: { id, exitStatus in await self.releaseWaiters(for: id, status: exitStatus) - guard let process = await self.processes[id]?.process else { - throw ContainerizationError( - .invalidState, - message: "ProcessInfo missing for process \(id)" - ) + if let process = await self.processes[id]?.process { + try await process.delete() } - try await process.delete() - try await self.setProcessState(id: id, state: .stopped) + // Reap host-side exec state so completed processes do not + // accumulate in a long-lived runtime helper (#2057). + await self.reapExecProcess(id: id) } ) } catch { await self.releaseWaiters(for: id, status: ExitStatus(exitCode: -1)) + await self.reapExecProcess(id: id) throw error } @@ -932,22 +938,31 @@ public actor RuntimeService { containerConfig: containerInfo.config, ) - let process = try await container.exec(id, configuration: czConfig) - try self.setUnderlyingProcess(id, process) + do { + let process = try await container.exec(id, configuration: czConfig) + try self.setUnderlyingProcess(id, process) - try await process.start() + try await process.start() - let waitFunc: ExitMonitor.WaitHandler = { - let code = try await process.wait() - if let out = processInfo.io[1] { - try self.closeHandle(out.fileDescriptor) + let waitFunc: ExitMonitor.WaitHandler = { + let code = try await process.wait() + if let out = processInfo.io[1] { + try self.closeHandle(out.fileDescriptor) + } + if let err = processInfo.io[2] { + try self.closeHandle(err.fileDescriptor) + } + return code } - if let err = processInfo.io[2] { - try self.closeHandle(err.fileDescriptor) + try await self.monitor.track(id: id, waitingOn: waitFunc) + } catch { + if let process = self.processes[id]?.process { + try? await process.delete() } - return code + self.releaseWaiters(for: id, status: ExitStatus(exitCode: -1)) + await self.reapExecProcess(id: id) + throw error } - try await self.monitor.track(id: id, waitingOn: waitFunc) } private func startSocketForwarders(attachment: Attachment, publishedPorts: [PublishPort]) async throws { @@ -1357,6 +1372,17 @@ public actor RuntimeService { let status = exitStatus ?? ExitStatus(exitCode: 255) self.releaseWaiters(for: id, status: status) + + // Reap any ad hoc exec processes that outlived the container. + for processId in Array(self.processes.keys) { + if let process = self.processes[processId]?.process { + try? await process.delete() + } + await self.reapExecProcess(id: processId) + } + await self.monitor.stopTracking(id: id) + self.waiters.removeValue(forKey: id) + self.completedExits.removeValue(forKey: id) } } @@ -1550,19 +1576,35 @@ extension RuntimeService { } private func waitForExit(id: String, cont: CheckedContinuation) { - guard let waiter = waiters[id] else { - // No waiter was initialized at all, resume immediately - cont.resume(returning: ExitStatus(exitCode: -1)) + if let waiter = waiters[id] { + waiter.wait(cont) return } - - waiter.wait(cont) + if let status = completedExits.removeValue(forKey: id) { + cont.resume(returning: status) + return + } + // No waiter was initialized at all, or the process was already reaped + // without a recorded exit status (failed create / never started). + cont.resume(returning: ExitStatus(exitCode: -1)) } private func releaseWaiters(for id: String, status: ExitStatus) { waiters[id]?.doExit(exitStatus: status) } + /// Drop completed exec bookkeeping. Callers must already have notified waiters + /// and deleted any underlying guest process. + private func reapExecProcess(id: String) async { + if let waiter = waiters[id], let status = waiter.exitStatus, !waiter.didDeliverExit { + // Preserve the status for a late `wait` that arrives after reaping. + completedExits[id] = status + } + waiters.removeValue(forKey: id) + processes.removeValue(forKey: id) + await self.monitor.stopTracking(id: id) + } + private func setUnderlyingProcess(_ id: String, _ process: LinuxProcess) throws { guard var info = self.processes[id] else { throw ContainerizationError(.invalidState, message: "process \(id) not found") diff --git a/Tests/ContainerRuntimeClientTests/ExitMonitorTests.swift b/Tests/ContainerRuntimeClientTests/ExitMonitorTests.swift new file mode 100644 index 000000000..207257f75 --- /dev/null +++ b/Tests/ContainerRuntimeClientTests/ExitMonitorTests.swift @@ -0,0 +1,89 @@ +//===----------------------------------------------------------------------===// +// Copyright © 2025-2026 Apple Inc. and the container project authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +//===----------------------------------------------------------------------===// + +import ContainerRuntimeClient +import Containerization +import Foundation +import Testing + +struct ExitMonitorTests { + @Test + func trackReleasesEntriesAfterExit() async throws { + let monitor = ExitMonitor() + let id = "exec-\(UUID().uuidString)" + + try await monitor.registerProcess(id: id) { _, _ in } + try await monitor.track(id: id) { + ExitStatus(exitCode: 0) + } + + // Allow the monitor task to finish and drop its bookkeeping. + for _ in 0..<50 { + do { + try await monitor.registerProcess(id: id) { _, _ in } + return + } catch { + try await Task.sleep(for: .milliseconds(10)) + } + } + Issue.record("ExitMonitor did not release tracking for \(id) after exit") + } + + @Test + func trackReleasesEntriesAfterWaitHandlerFailure() async throws { + let monitor = ExitMonitor() + let id = "exec-fail-\(UUID().uuidString)" + + try await monitor.registerProcess(id: id) { _, status in + #expect(status.exitCode == -1) + } + try await monitor.track(id: id) { + struct Boom: Error {} + throw Boom() + } + + for _ in 0..<50 { + do { + try await monitor.registerProcess(id: id) { _, _ in } + return + } catch { + try await Task.sleep(for: .milliseconds(10)) + } + } + Issue.record("ExitMonitor did not release tracking after WaitHandler failure") + } + + @Test + func manyCompletedTracksDoNotBlockReregistration() async throws { + let monitor = ExitMonitor() + + for i in 0..<200 { + let id = "batch-\(i)" + try await monitor.registerProcess(id: id) { _, _ in } + try await monitor.track(id: id) { + ExitStatus(exitCode: 0) + } + } + + // Give tasks a moment to settle, then confirm IDs can be reused. + try await Task.sleep(for: .milliseconds(200)) + for i in 0..<200 { + let id = "batch-\(i)" + try await monitor.registerProcess(id: id) { _, _ in } + await monitor.stopTracking(id: id) + } + } +}