LinuxContainer: Support pause/resume (#217)

This allows pausing the (really VM) container and any processes running
inside.

Sources/Containerization/LinuxContainer.swift

@@ -154,6 +154,12 @@ public final class LinuxContainer: Container, Sendable {
         case stopped
         /// An error occurred during the lifetime of this class.
         case errored(Swift.Error)
+        /// The container is being paused.
+        case pausing(PausingState)
+        /// The container is paused.
+        case paused(PausedState)
+        /// The container is being resumed.
+        case resuming(ResumingState)
 
         struct CreatingState: Sendable {}
 
@@ -172,6 +178,42 @@ public final class LinuxContainer: Container, Sendable {
             }
         }
 
+        struct PausingState: Sendable {
+            let vm: any VirtualMachineInstance
+            let relayManager: UnixSocketRelayManager
+            let process: LinuxProcess
+
+            init(_ state: StartedState) {
+                self.vm = state.vm
+                self.relayManager = state.relayManager
+                self.process = state.process
+            }
+        }
+
+        struct PausedState: Sendable {
+            let vm: any VirtualMachineInstance
+            let relayManager: UnixSocketRelayManager
+            let process: LinuxProcess
+
+            init(_ state: PausingState) {
+                self.vm = state.vm
+                self.relayManager = state.relayManager
+                self.process = state.process
+            }
+        }
+
+        struct ResumingState: Sendable {
+            let vm: any VirtualMachineInstance
+            let relayManager: UnixSocketRelayManager
+            let process: LinuxProcess
+
+            init(_ state: PausedState) {
+                self.vm = state.vm
+                self.relayManager = state.relayManager
+                self.process = state.process
+            }
+        }
+
         struct StartedState: Sendable {
             let vm: any VirtualMachineInstance
             let process: LinuxProcess
@@ -182,6 +224,12 @@ public final class LinuxContainer: Container, Sendable {
                 self.relayManager = state.relayManager
                 self.process = process
             }
+
+            init(_ state: ResumingState) {
+                self.vm = state.vm
+                self.relayManager = state.relayManager
+                self.process = state.process
+            }
         }
 
         struct StoppingState: Sendable {
@@ -244,6 +292,18 @@ public final class LinuxContainer: Container, Sendable {
             }
         }
 
+        mutating func setResumed() throws {
+            switch self {
+            case .resuming(let state):
+                self = .started(.init(state))
+            default:
+                throw ContainerizationError(
+                    .invalidState,
+                    message: "container must be in resuming state before being resumed"
+                )
+            }
+        }
+
         mutating func stopping() throws -> StartedState {
             switch self {
             case .started(let state):
@@ -269,6 +329,44 @@ public final class LinuxContainer: Container, Sendable {
             }
         }
 
+        mutating func setPausing() throws -> StartedState {
+            switch self {
+            case .started(let state):
+                self = .pausing(.init(state))
+                return state
+            default:
+                throw ContainerizationError(
+                    .invalidState,
+                    message: "failed to pause: container must be running"
+                )
+            }
+        }
+
+        mutating func setPaused() throws {
+            switch self {
+            case .pausing(let state):
+                self = .paused(.init(state))
+            default:
+                throw ContainerizationError(
+                    .invalidState,
+                    message: "failed to pause: container must be running"
+                )
+            }
+        }
+
+        mutating func setResuming() throws -> PausedState {
+            switch self {
+            case .paused(let state):
+                self = .resuming(.init(state))
+                return state
+            default:
+                throw ContainerizationError(
+                    .invalidState,
+                    message: "failed to resume: container must be paused"
+                )
+            }
+        }
+
         mutating func stopped() throws {
             switch self {
             case .stopping(_):
@@ -574,6 +672,28 @@ extension LinuxContainer {
         try self.state.withLock { try $0.stopped() }
     }
 
+    /// Pause the container.
+    public func pause() async throws {
+        do {
+            let state = try self.state.withLock { try $0.setPausing() }
+            try await state.vm.pause()
+            try self.state.withLock { try $0.setPaused() }
+        } catch {
+            self.state.withLock { $0.errored(error: error) }
+        }
+    }
+
+    /// Resume the container.
+    public func resume() async throws {
+        do {
+            let state = try self.state.withLock { try $0.setResuming() }
+            try await state.vm.resume()
+            try self.state.withLock { try $0.setResumed() }
+        } catch {
+            self.state.withLock { $0.errored(error: error) }
+        }
+    }
+
     /// Send a signal to the container.
     public func kill(_ signal: Int32) async throws {
         let state = try self.state.withLock { try $0.startedState("kill") }

Sources/Containerization/TimeSyncer.swift

@@ -20,9 +20,11 @@ import Logging
 actor TimeSyncer {
     private var task: Task<Void, Never>?
     private var context: Vminitd?
+    private var paused: Bool
     private let logger: Logger?
 
     init(logger: Logger?) {
+        self.paused = false
         self.logger = logger
     }
 
@@ -38,6 +40,10 @@ actor TimeSyncer {
                         return
                     }
 
+                    guard !paused else {
+                        continue
+                    }
+
                     var timeval = timeval()
                     guard gettimeofday(&timeval, nil) == 0 else {
                         throw POSIXError.fromErrno()
@@ -54,6 +60,14 @@ actor TimeSyncer {
         }
     }
 
+    func pause() async {
+        self.paused = true
+    }
+
+    func resume() async {
+        self.paused = false
+    }
+
     func close() async throws {
         guard let task else {
             preconditionFailure("time syncer was already closed")

Sources/Containerization/VZVirtualMachine+Helpers.swift

@@ -91,6 +91,34 @@ extension VZVirtualMachine {
             }
         }
     }
+
+    func pause(queue: DispatchQueue) async throws {
+        try await withCheckedThrowingContinuation { (cont: CheckedContinuation<Void, Error>) in
+            queue.sync {
+                self.pause { result in
+                    if case .failure(let error) = result {
+                        cont.resume(throwing: error)
+                        return
+                    }
+                    cont.resume()
+                }
+            }
+        }
+    }
+
+    func resume(queue: DispatchQueue) async throws {
+        try await withCheckedThrowingContinuation { (cont: CheckedContinuation<Void, Error>) in
+            queue.sync {
+                self.resume { result in
+                    if case .failure(let error) = result {
+                        cont.resume(throwing: error)
+                        return
+                    }
+                    cont.resume()
+                }
+            }
+        }
+    }
 }
 
 extension VZVirtualMachine {

Sources/Containerization/VZVirtualMachineInstance.swift

@@ -172,6 +172,20 @@ extension VZVirtualMachineInstance {
         }
     }
 
+    func pause() async throws {
+        try await lock.withLock { _ in
+            await self.timeSyncer.pause()
+            try await self.vm.pause(queue: self.queue)
+        }
+    }
+
+    func resume() async throws {
+        try await lock.withLock { _ in
+            try await self.vm.resume(queue: self.queue)
+            await self.timeSyncer.resume()
+        }
+    }
+
     public func dialAgent() async throws -> Vminitd {
         let conn = try await dial(Vminitd.port)
         return Vminitd(connection: conn, group: self.group)

Sources/Containerization/VirtualMachineInstance.swift

@@ -14,6 +14,7 @@
 // limitations under the License.
 //===----------------------------------------------------------------------===//
 
+import ContainerizationError
 import Foundation
 
 /// The runtime state of the virtual machine instance.
@@ -46,4 +47,17 @@ public protocol VirtualMachineInstance: Sendable {
     func start() async throws
     /// Stop the virtual machine.
     func stop() async throws
+    /// Pause the virtual machine.
+    func pause() async throws
+    /// Resume the virtual machine.
+    func resume() async throws
+}
+
+extension VirtualMachineInstance {
+    func pause() async throws {
+        throw ContainerizationError(.unsupported, message: "pause")
+    }
+    func resume() async throws {
+        throw ContainerizationError(.unsupported, message: "resume")
+    }
 }

Sources/Integration/Suite.swift

@@ -209,6 +209,9 @@ struct IntegrationSuite: AsyncParsableCommand {
             "container hostname": testHostname,
             "container hosts": testHostsFile,
             "container mount": testMounts,
+            "container pause and resume": testPauseResume,
+            "container pause, resume and wait": testPauseResumeWait,
+            "container pause, resume and verify io": testPauseResumeIO,
             "nested virt": testNestedVirtualizationEnabled,
             "container manager": testContainerManagerCreate,
             "container reuse": testContainerReuse,

Sources/Integration/VMTests.swift

@@ -52,6 +52,91 @@ extension IntegrationSuite {
         }
     }
 
+    func testPauseResume() async throws {
+        let id = "test-pause-resume"
+
+        let bs = try await bootstrap()
+        let container = try LinuxContainer(id, rootfs: bs.rootfs, vmm: bs.vmm) { config in
+            config.process.arguments = ["sleep", "infinity"]
+        }
+
+        try await container.create()
+        try await container.start()
+
+        // Very simple test of can we perform actions on the container after pause/resume.
+        try await container.pause()
+        try await Task.sleep(for: .milliseconds(500))
+        try await container.resume()
+
+        try await container.kill(SIGKILL)
+        try await container.wait()
+        try await container.stop()
+    }
+
+    func testPauseResumeWait() async throws {
+        let id = "test-pause-resume-wait"
+
+        let bs = try await bootstrap()
+        let container = try LinuxContainer(id, rootfs: bs.rootfs, vmm: bs.vmm) { config in
+            config.process.arguments = ["sleep", "2"]
+        }
+
+        try await container.create()
+        try await container.start()
+
+        let t = Task {
+            try await container.wait(timeoutInSeconds: 5)
+        }
+
+        try await Task.sleep(for: .milliseconds(25))
+
+        try await container.pause()
+        try await Task.sleep(for: .milliseconds(500))
+        try await container.resume()
+
+        let status = try await t.value
+
+        guard status == 0 else {
+            throw IntegrationError.assert(msg: "process status \(status) != 0")
+        }
+
+        try await container.stop()
+    }
+
+    func testPauseResumeIO() async throws {
+        let id = "test-pause-resume-io"
+
+        let bs = try await bootstrap()
+        let buffer = BufferWriter()
+        let container = try LinuxContainer(id, rootfs: bs.rootfs, vmm: bs.vmm) { config in
+            config.process.arguments = ["ping", "-c", "5", "localhost"]
+            config.process.stdout = buffer
+        }
+
+        try await container.create()
+        try await container.start()
+
+        try await container.pause()
+        try await Task.sleep(for: .seconds(2))
+        try await container.resume()
+
+        try await container.wait()
+
+        guard let str = String(data: buffer.data, encoding: .utf8) else {
+            throw IntegrationError.assert(msg: "failed to convert stdout to utf8")
+        }
+
+        // Should be 10 lines long. 5 of "filler" and 5 of actual
+        // output, however one of the lines is a blank newline.
+        let expectedLines = 9
+        let lines = str.split(separator: "\n")
+        guard lines.count == expectedLines else {
+            throw IntegrationError.assert(msg: "expected \(expectedLines), got \(lines.count)")
+        }
+
+        try await container.stop()
+    }
+
     func testNestedVirtualizationEnabled() async throws {
         let id = "test-nested-virt"