fix warnings
This commit is contained in:
Ian Gulliver
+98
-140
@@ -38,7 +38,9 @@ final class Remote {
|
||||
stopped = true
|
||||
ws?.cancel(with: .normalClosure, reason: nil)
|
||||
ws = nil
|
||||
hands.stop()
|
||||
Task { [hands] in
|
||||
await hands.stop()
|
||||
}
|
||||
}
|
||||
|
||||
private func receive(_ task: URLSessionWebSocketTask) {
|
||||
@@ -209,79 +211,57 @@ private func writeFile(_ path: String, _ data: Data) throws {
|
||||
try data.write(to: target)
|
||||
}
|
||||
|
||||
private final class Flag: @unchecked Sendable {
|
||||
private let lock = NSLock()
|
||||
private var value = false
|
||||
|
||||
func set() -> Bool {
|
||||
lock.withLock {
|
||||
let was = value
|
||||
value = true
|
||||
return !was
|
||||
}
|
||||
private func start(_ p: Process) throws -> AsyncStream<Void> {
|
||||
let (exit, exited) = AsyncStream.makeStream(of: Void.self)
|
||||
p.terminationHandler = { _ in
|
||||
exited.finish()
|
||||
}
|
||||
try p.run()
|
||||
return exit
|
||||
}
|
||||
|
||||
private func spawn(_ argv: [String], in dir: String, timeoutMs: Int) async throws -> (Int, String) {
|
||||
let out = FileManager.default.temporaryDirectory.appending(path: "byre-\(UUID().uuidString).log")
|
||||
let fm = FileManager.default
|
||||
let out = fm.temporaryDirectory.appending(path: "byre-\(UUID().uuidString).log")
|
||||
defer {
|
||||
try? FileManager.default.removeItem(at: out)
|
||||
try? fm.removeItem(at: out)
|
||||
}
|
||||
fm.createFile(atPath: out.path, contents: nil, attributes: [.posixPermissions: 0o600])
|
||||
let sink = try FileHandle(forWritingTo: out)
|
||||
defer {
|
||||
try? sink.close()
|
||||
}
|
||||
|
||||
var attr: posix_spawnattr_t?
|
||||
unsafe posix_spawnattr_init(&attr)
|
||||
defer {
|
||||
unsafe posix_spawnattr_destroy(&attr)
|
||||
let p = Process()
|
||||
p.executableURL = URL(filePath: argv[0])
|
||||
p.arguments = Array(argv.dropFirst())
|
||||
p.currentDirectoryURL = URL(filePath: dir)
|
||||
p.standardInput = FileHandle.nullDevice
|
||||
p.standardOutput = sink
|
||||
p.standardError = sink
|
||||
let exit: AsyncStream<Void>
|
||||
do {
|
||||
exit = try start(p)
|
||||
} catch {
|
||||
throw ApiError(status: 0, message: "start \(argv[0]): \(error.localizedDescription)")
|
||||
}
|
||||
unsafe posix_spawnattr_setflags(&attr, Int16(POSIX_SPAWN_SETPGROUP))
|
||||
unsafe posix_spawnattr_setpgroup(&attr, 0)
|
||||
|
||||
var actions: posix_spawn_file_actions_t?
|
||||
unsafe posix_spawn_file_actions_init(&actions)
|
||||
defer {
|
||||
unsafe posix_spawn_file_actions_destroy(&actions)
|
||||
let pid = p.processIdentifier
|
||||
let timer = Task {
|
||||
try await Task.sleep(for: .milliseconds(timeoutMs))
|
||||
kill(-pid, SIGKILL)
|
||||
}
|
||||
unsafe posix_spawn_file_actions_addopen(&actions, 0, "/dev/null", O_RDONLY, 0)
|
||||
unsafe posix_spawn_file_actions_addopen(&actions, 1, out.path, O_WRONLY | O_CREAT | O_TRUNC, 0o600)
|
||||
unsafe posix_spawn_file_actions_adddup2(&actions, 1, 2)
|
||||
unsafe posix_spawn_file_actions_addchdir(&actions, dir)
|
||||
|
||||
var cargs = unsafe argv.map { unsafe strdup($0) } + [nil]
|
||||
defer {
|
||||
for unsafe p in unsafe cargs {
|
||||
unsafe free(p)
|
||||
}
|
||||
}
|
||||
var spawned: pid_t = 0
|
||||
let rc = unsafe posix_spawn(&spawned, argv[0], &actions, &attr, &cargs, environ)
|
||||
if rc != 0 {
|
||||
throw ApiError(status: 0, message: "start \(argv[0]): \(unsafe String(cString: strerror(rc)))")
|
||||
}
|
||||
let pid = spawned
|
||||
|
||||
let done = Flag()
|
||||
DispatchQueue.global().asyncAfter(deadline: .now() + .milliseconds(timeoutMs)) {
|
||||
if done.set() {
|
||||
kill(-pid, SIGKILL)
|
||||
}
|
||||
}
|
||||
let status: Int32 = await withCheckedContinuation { c in
|
||||
DispatchQueue.global().async {
|
||||
var s: Int32 = 0
|
||||
while unsafe waitpid(pid, &s, 0) == -1 && errno == EINTR {}
|
||||
c.resume(returning: s)
|
||||
}
|
||||
}
|
||||
let timedOut = !done.set()
|
||||
for await _ in exit {}
|
||||
timer.cancel()
|
||||
let timedOut = (try? await timer.value) != nil
|
||||
|
||||
let output = tail(out)
|
||||
if timedOut {
|
||||
return (-1, output)
|
||||
}
|
||||
if status & 0x7f == 0 {
|
||||
return (Int((status >> 8) & 0xff), output)
|
||||
if p.terminationReason == .exit {
|
||||
return (Int(p.terminationStatus), output)
|
||||
}
|
||||
return (128 + Int(status & 0x7f), output)
|
||||
return (128 + Int(p.terminationStatus), output)
|
||||
}
|
||||
|
||||
private func tail(_ file: URL) -> String {
|
||||
@@ -318,113 +298,91 @@ private struct HandsResponse: Decodable {
|
||||
let error: RPCError?
|
||||
}
|
||||
|
||||
private final class Helper: @unchecked Sendable {
|
||||
let process = Process()
|
||||
let input: FileHandle
|
||||
var buffer = Data()
|
||||
var pending: [Int: CheckedContinuation<OpenAPIObjectContainer, any Error>] = [:]
|
||||
|
||||
init(input: FileHandle) {
|
||||
self.input = input
|
||||
}
|
||||
}
|
||||
|
||||
private final class Hands: @unchecked Sendable {
|
||||
private let lock = NSLock()
|
||||
private var helper: Helper?
|
||||
private actor Hands {
|
||||
private var process: Process?
|
||||
private var input: FileHandle?
|
||||
private var next = 0
|
||||
private var pending: [Int: CheckedContinuation<OpenAPIObjectContainer, any Error>] = [:]
|
||||
|
||||
func call(_ tool: String, _ arguments: OpenAPIObjectContainer) async throws -> OpenAPIObjectContainer {
|
||||
try await withCheckedThrowingContinuation { c in
|
||||
lock.withLock {
|
||||
next += 1
|
||||
let id = next
|
||||
var target: Helper?
|
||||
do {
|
||||
let h = try running()
|
||||
target = h
|
||||
var line = try JSONEncoder().encode(HandsRequest(id: id, params: .init(name: tool, arguments: arguments)))
|
||||
line.append(0x0a)
|
||||
h.pending[id] = c
|
||||
try h.input.write(contentsOf: line)
|
||||
} catch {
|
||||
target?.pending[id] = nil
|
||||
c.resume(throwing: error)
|
||||
}
|
||||
next += 1
|
||||
let id = next
|
||||
var line = try JSONEncoder().encode(HandsRequest(id: id, params: .init(name: tool, arguments: arguments)))
|
||||
line.append(0x0a)
|
||||
let input = try running()
|
||||
return try await withCheckedThrowingContinuation { c in
|
||||
pending[id] = c
|
||||
do {
|
||||
try input.write(contentsOf: line)
|
||||
} catch {
|
||||
pending[id] = nil
|
||||
c.resume(throwing: error)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func stop() {
|
||||
lock.withLock {
|
||||
helper?.process.terminate()
|
||||
}
|
||||
process?.terminate()
|
||||
}
|
||||
|
||||
private func running() throws -> Helper {
|
||||
if let helper, helper.process.isRunning {
|
||||
return helper
|
||||
private func running() throws -> FileHandle {
|
||||
if let input {
|
||||
return input
|
||||
}
|
||||
signal(SIGPIPE, SIG_IGN)
|
||||
let stdin = Pipe()
|
||||
let stdout = Pipe()
|
||||
let h = Helper(input: stdin.fileHandleForWriting)
|
||||
let p = h.process
|
||||
let p = Process()
|
||||
p.executableURL = Bundle.main.executableURL!.deletingLastPathComponent().appending(path: "byre-hands")
|
||||
p.standardInput = stdin
|
||||
p.standardOutput = stdout
|
||||
p.standardError = FileHandle.nullDevice
|
||||
stdout.fileHandleForReading.readabilityHandler = { [weak self] fh in
|
||||
self?.read(h, fh.availableData)
|
||||
}
|
||||
p.terminationHandler = { [weak self] _ in
|
||||
self?.exited(h)
|
||||
}
|
||||
try p.run()
|
||||
let exit = try start(p)
|
||||
log.info("byre-hands started: pid \(p.processIdentifier)")
|
||||
helper = h
|
||||
return h
|
||||
process = p
|
||||
input = stdin.fileHandleForWriting
|
||||
Task {
|
||||
await serve(p, stdout.fileHandleForReading, exit)
|
||||
}
|
||||
return stdin.fileHandleForWriting
|
||||
}
|
||||
|
||||
private func read(_ h: Helper, _ data: Data) {
|
||||
let lines: [Data] = lock.withLock {
|
||||
h.buffer.append(data)
|
||||
var out: [Data] = []
|
||||
while let nl = h.buffer.firstIndex(of: 0x0a) {
|
||||
out.append(h.buffer[h.buffer.startIndex..<nl])
|
||||
h.buffer = Data(h.buffer[(nl + 1)...])
|
||||
private func serve(_ p: Process, _ output: FileHandle, _ exit: AsyncStream<Void>) async {
|
||||
var line: [UInt8] = []
|
||||
do {
|
||||
for try await byte in output.bytes {
|
||||
if byte != 0x0a {
|
||||
line.append(byte)
|
||||
continue
|
||||
}
|
||||
reply(Data(line))
|
||||
line.removeAll(keepingCapacity: true)
|
||||
}
|
||||
return out
|
||||
} catch {
|
||||
log.error("read byre-hands: \(errorText(error), privacy: .public)")
|
||||
}
|
||||
for line in lines {
|
||||
guard let resp = try? JSONDecoder().decode(HandsResponse.self, from: line) else {
|
||||
log.error("byre-hands sent an unreadable line")
|
||||
continue
|
||||
}
|
||||
let c = lock.withLock {
|
||||
h.pending.removeValue(forKey: resp.id)
|
||||
}
|
||||
if let result = resp.result {
|
||||
c?.resume(returning: result)
|
||||
} else {
|
||||
c?.resume(throwing: ApiError(status: 0, message: "byre-hands: \(resp.error?.message ?? "no result")"))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private func exited(_ h: Helper) {
|
||||
let waiting = lock.withLock {
|
||||
let all = h.pending
|
||||
h.pending = [:]
|
||||
if helper === h {
|
||||
helper = nil
|
||||
}
|
||||
return all
|
||||
}
|
||||
let status = h.process.terminationStatus
|
||||
for await _ in exit {}
|
||||
let status = p.terminationStatus
|
||||
log.info("byre-hands exited: status \(status)")
|
||||
process = nil
|
||||
input = nil
|
||||
let waiting = pending
|
||||
pending = [:]
|
||||
for c in waiting.values {
|
||||
c.resume(throwing: ApiError(status: 0, message: "byre-hands exited with status \(status) before replying"))
|
||||
}
|
||||
}
|
||||
|
||||
private func reply(_ line: Data) {
|
||||
guard let resp = try? JSONDecoder().decode(HandsResponse.self, from: line) else {
|
||||
log.error("byre-hands sent an unreadable line")
|
||||
return
|
||||
}
|
||||
let c = pending.removeValue(forKey: resp.id)
|
||||
if let result = resp.result {
|
||||
c?.resume(returning: result)
|
||||
} else {
|
||||
c?.resume(throwing: ApiError(status: 0, message: "byre-hands: \(resp.error?.message ?? "no result")"))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user