1375 lines
48 KiB
Nim
1375 lines
48 KiB
Nim
#
|
|
# Chronos' asynchronous process management
|
|
#
|
|
# (c) Copyright 2022-Present Status Research & Development GmbH
|
|
#
|
|
# Licensed under either of
|
|
# Apache License, version 2.0, (LICENSE-APACHEv2)
|
|
# MIT license (LICENSE-MIT)
|
|
|
|
{.push raises: [].}
|
|
{.pragma: apforward, gcsafe, raises: [].}
|
|
|
|
import std/strtabs
|
|
import "."/[config, asyncloop, handles, osdefs, osutils, oserrno],
|
|
streams/asyncstream
|
|
import stew/[byteutils], results
|
|
from std/os import quoteShell, quoteShellWindows, quoteShellPosix, envPairs
|
|
|
|
export strtabs, results
|
|
export quoteShell, quoteShellWindows, quoteShellPosix, envPairs
|
|
|
|
const
|
|
AsyncProcessTrackerName* = "async.process"
|
|
## AsyncProcess leaks tracker name
|
|
|
|
type
|
|
AsyncProcessError* = object of AsyncError
|
|
AsyncProcessTimeoutError* = object of AsyncProcessError
|
|
|
|
AsyncProcessResult*[T] = Result[T, OSErrorCode]
|
|
|
|
AsyncProcessOption* {.pure.} = enum
|
|
UsePath,
|
|
EvalCommand,
|
|
StdErrToStdOut,
|
|
ProcessGroup
|
|
|
|
StandardKind {.pure.} = enum
|
|
Stdin, Stdout, Stderr
|
|
|
|
ProcessFlag {.pure.} = enum
|
|
UserStdin, UserStdout, UserStderr,
|
|
AutoStdin, AutoStdout, AutoStderr,
|
|
NoStdin, NoStdout, NoStderr,
|
|
CopyStdout
|
|
|
|
ProcessStreamHandleKind {.pure.} = enum
|
|
None, Auto, ProcHandle, Transport, StreamReader, StreamWriter
|
|
|
|
ProcessStreamHandle* = object
|
|
case kind: ProcessStreamHandleKind
|
|
of ProcessStreamHandleKind.None:
|
|
discard
|
|
of ProcessStreamHandleKind.Auto:
|
|
discard
|
|
of ProcessStreamHandleKind.ProcHandle:
|
|
handle: AsyncFD
|
|
of ProcessStreamHandleKind.Transport:
|
|
transp: StreamTransport
|
|
of ProcessStreamHandleKind.StreamReader:
|
|
reader: AsyncStreamReader
|
|
of ProcessStreamHandleKind.StreamWriter:
|
|
writer: AsyncStreamWriter
|
|
|
|
StreamHolderFlag {.pure.} = enum
|
|
Transport, Stream
|
|
|
|
StreamKind {.pure.} = enum
|
|
None, Reader, Writer
|
|
|
|
AsyncStreamHolder = object
|
|
case kind: StreamKind
|
|
of StreamKind.Reader:
|
|
reader: AsyncStreamReader
|
|
of StreamKind.Writer:
|
|
writer: AsyncStreamWriter
|
|
of StreamKind.None:
|
|
discard
|
|
flags: set[StreamHolderFlag]
|
|
|
|
AsyncProcessPipes = object
|
|
flags: set[ProcessFlag]
|
|
stdinHolder: AsyncStreamHolder
|
|
stdoutHolder: AsyncStreamHolder
|
|
stderrHolder: AsyncStreamHolder
|
|
stdinHandle: AsyncFD
|
|
stdoutHandle: AsyncFD
|
|
stderrHandle: AsyncFD
|
|
|
|
AsyncProcess* = object
|
|
|
|
AsyncProcessImpl = object of RootObj
|
|
when defined(windows):
|
|
processHandle: HANDLE
|
|
threadHandle: HANDLE
|
|
processId: DWORD
|
|
else:
|
|
processId: Pid
|
|
pipes: AsyncProcessPipes
|
|
exitStatus: Opt[int]
|
|
flags: set[ProcessFlag]
|
|
options: set[AsyncProcessOption]
|
|
|
|
AsyncProcessRef* = ref AsyncProcessImpl
|
|
|
|
CommandExResponse* = object
|
|
stdOutput*: string
|
|
stdError*: string
|
|
status*: int
|
|
|
|
WaitOperation {.pure.} = enum
|
|
Kill, Terminate
|
|
|
|
template Pipe*(t: typedesc[AsyncProcess]): ProcessStreamHandle =
|
|
ProcessStreamHandle(kind: ProcessStreamHandleKind.Auto)
|
|
|
|
proc init*(t: typedesc[AsyncFD], handle: ProcessStreamHandle): AsyncFD =
|
|
case handle.kind
|
|
of ProcessStreamHandleKind.ProcHandle:
|
|
handle.handle
|
|
of ProcessStreamHandleKind.Transport:
|
|
handle.transp.fd
|
|
of ProcessStreamHandleKind.StreamReader:
|
|
doAssert(not(isNil(handle.reader.tsource)))
|
|
handle.reader.tsource.fd
|
|
of ProcessStreamHandleKind.StreamWriter:
|
|
doAssert(not(isNil(handle.writer.tsource)))
|
|
handle.writer.tsource.fd
|
|
of ProcessStreamHandleKind.Auto:
|
|
raiseAssert "ProcessStreamHandle could not be auto at this moment"
|
|
of ProcessStreamHandleKind.None:
|
|
raiseAssert "ProcessStreamHandle could not be empty at this moment"
|
|
|
|
proc init*(t: typedesc[AsyncStreamHolder], handle: AsyncStreamReader,
|
|
baseFlags: set[StreamHolderFlag] = {}): AsyncStreamHolder =
|
|
AsyncStreamHolder(kind: StreamKind.Reader, reader: handle, flags: baseFlags)
|
|
|
|
proc init*(t: typedesc[AsyncStreamHolder], handle: AsyncStreamWriter,
|
|
baseFlags: set[StreamHolderFlag] = {}): AsyncStreamHolder =
|
|
AsyncStreamHolder(kind: StreamKind.Writer, writer: handle, flags: baseFlags)
|
|
|
|
proc init*(t: typedesc[AsyncStreamHolder]): AsyncStreamHolder =
|
|
AsyncStreamHolder(kind: StreamKind.None)
|
|
|
|
proc init*(t: typedesc[AsyncStreamHolder], handle: ProcessStreamHandle,
|
|
kind: StreamKind, baseFlags: set[StreamHolderFlag] = {}
|
|
): AsyncProcessResult[AsyncStreamHolder] =
|
|
case handle.kind
|
|
of ProcessStreamHandleKind.ProcHandle:
|
|
case kind
|
|
of StreamKind.Reader:
|
|
let
|
|
transp = ? fromPipe2(handle.handle)
|
|
reader = newAsyncStreamReader(transp)
|
|
flags = baseFlags + {StreamHolderFlag.Stream,
|
|
StreamHolderFlag.Transport}
|
|
ok(AsyncStreamHolder(kind: StreamKind.Reader, reader: reader,
|
|
flags: flags))
|
|
of StreamKind.Writer:
|
|
let
|
|
transp = ? fromPipe2(handle.handle)
|
|
writer = newAsyncStreamWriter(transp)
|
|
flags = baseFlags + {StreamHolderFlag.Stream,
|
|
StreamHolderFlag.Transport}
|
|
ok(AsyncStreamHolder(kind: StreamKind.Writer, writer: writer,
|
|
flags: flags))
|
|
of StreamKind.None:
|
|
ok(AsyncStreamHolder(kind: StreamKind.None))
|
|
of ProcessStreamHandleKind.Transport:
|
|
case kind
|
|
of StreamKind.Reader:
|
|
let
|
|
reader = newAsyncStreamReader(handle.transp)
|
|
flags = baseFlags + {StreamHolderFlag.Stream}
|
|
ok(AsyncStreamHolder(kind: StreamKind.Reader, reader: reader,
|
|
flags: flags))
|
|
of StreamKind.Writer:
|
|
let
|
|
writer = newAsyncStreamWriter(handle.transp)
|
|
flags = baseFlags + {StreamHolderFlag.Stream}
|
|
ok(AsyncStreamHolder(kind: StreamKind.Writer, writer: writer,
|
|
flags: flags))
|
|
of StreamKind.None:
|
|
ok(AsyncStreamHolder(kind: StreamKind.None))
|
|
of ProcessStreamHandleKind.StreamReader:
|
|
ok(AsyncStreamHolder(kind: StreamKind.Reader, reader: handle.reader,
|
|
flags: baseFlags))
|
|
of ProcessStreamHandleKind.StreamWriter:
|
|
ok(AsyncStreamHolder(kind: StreamKind.Writer, writer: handle.writer,
|
|
flags: baseFlags))
|
|
of ProcessStreamHandleKind.None, ProcessStreamHandleKind.Auto:
|
|
ok(AsyncStreamHolder(kind: StreamKind.None))
|
|
|
|
proc init*(t: typedesc[ProcessStreamHandle]): ProcessStreamHandle =
|
|
ProcessStreamHandle(kind: ProcessStreamHandleKind.None)
|
|
|
|
proc init*(t: typedesc[ProcessStreamHandle],
|
|
handle: AsyncFD): ProcessStreamHandle =
|
|
ProcessStreamHandle(kind: ProcessStreamHandleKind.ProcHandle, handle: handle)
|
|
|
|
proc init*(t: typedesc[ProcessStreamHandle],
|
|
transp: StreamTransport): ProcessStreamHandle =
|
|
doAssert(transp.kind == TransportKind.Pipe,
|
|
"Only pipe transports can be used as process streams")
|
|
ProcessStreamHandle(kind: ProcessStreamHandleKind.ProcHandle, transp: transp)
|
|
|
|
proc init*(t: typedesc[ProcessStreamHandle],
|
|
reader: AsyncStreamReader): ProcessStreamHandle =
|
|
ProcessStreamHandle(kind: ProcessStreamHandleKind.StreamReader,
|
|
reader: reader)
|
|
|
|
proc init*(t: typedesc[ProcessStreamHandle],
|
|
writer: AsyncStreamWriter): ProcessStreamHandle =
|
|
ProcessStreamHandle(kind: ProcessStreamHandleKind.StreamWriter,
|
|
writer: writer)
|
|
|
|
proc isEmpty*(handle: ProcessStreamHandle): bool =
|
|
handle.kind == ProcessStreamHandleKind.None
|
|
|
|
proc suspend*(p: AsyncProcessRef): AsyncProcessResult[void] {.apforward.}
|
|
proc resume*(p: AsyncProcessRef): AsyncProcessResult[void] {.apforward.}
|
|
proc terminate*(p: AsyncProcessRef): AsyncProcessResult[void] {.apforward.}
|
|
proc kill*(p: AsyncProcessRef): AsyncProcessResult[void] {.apforward.}
|
|
proc running*(p: AsyncProcessRef): AsyncProcessResult[bool] {.apforward.}
|
|
proc peekExitCode*(p: AsyncProcessRef): AsyncProcessResult[int] {.apforward.}
|
|
proc preparePipes(options: set[AsyncProcessOption],
|
|
stdinHandle, stdoutHandle, stderrHandle: ProcessStreamHandle
|
|
): AsyncProcessResult[AsyncProcessPipes] {.apforward.}
|
|
proc closeProcessHandles(pipes: var AsyncProcessPipes,
|
|
options: set[AsyncProcessOption],
|
|
lastError: OSErrorCode): OSErrorCode {.apforward.}
|
|
proc closeProcessStreams(pipes: AsyncProcessPipes,
|
|
options: set[AsyncProcessOption]): Future[void] {.
|
|
async: (raises: []).}
|
|
proc closeWait(holder: AsyncStreamHolder): Future[void] {.
|
|
async: (raises: []).}
|
|
|
|
template isOk(code: OSErrorCode): bool =
|
|
when defined(windows):
|
|
code == ERROR_SUCCESS
|
|
else:
|
|
code == OSErrorCode(0)
|
|
|
|
template closePipe(handle: AsyncFD): bool =
|
|
let fd =
|
|
when defined(windows):
|
|
osdefs.HANDLE(handle)
|
|
else:
|
|
cint(handle)
|
|
closeFd(fd) != -1
|
|
|
|
proc closeProcessHandles(pipes: var AsyncProcessPipes,
|
|
options: set[AsyncProcessOption],
|
|
lastError: OSErrorCode): OSErrorCode =
|
|
# We trying to preserve error code of last failed operation.
|
|
var currentError = lastError
|
|
|
|
if ProcessFlag.AutoStdin in pipes.flags:
|
|
if pipes.stdinHandle != asyncInvalidPipe:
|
|
if currentError.isOk():
|
|
if not(closePipe(pipes.stdinHandle)):
|
|
currentError = osLastError()
|
|
else:
|
|
discard closePipe(pipes.stdinHandle)
|
|
pipes.stdinHandle = asyncInvalidPipe
|
|
|
|
if ProcessFlag.AutoStdout in pipes.flags:
|
|
if pipes.stdoutHandle != asyncInvalidPipe:
|
|
if currentError.isOk():
|
|
if not(closePipe(pipes.stdoutHandle)):
|
|
currentError = osLastError()
|
|
else:
|
|
discard closePipe(pipes.stdoutHandle)
|
|
pipes.stdoutHandle = asyncInvalidPipe
|
|
|
|
if ProcessFlag.AutoStderr in pipes.flags:
|
|
if pipes.stderrHandle != asyncInvalidPipe:
|
|
if currentError.isOk():
|
|
if not(closePipe(pipes.stderrHandle)):
|
|
currentError = osLastError()
|
|
else:
|
|
discard closePipe(pipes.stderrHandle)
|
|
pipes.stderrHandle = asyncInvalidPipe
|
|
|
|
currentError
|
|
|
|
template pipesPresent*(pipes: AsyncProcessPipes): bool =
|
|
let mask = {ProcessFlag.AutoStdin, ProcessFlag.AutoStdout,
|
|
ProcessFlag.AutoStderr,ProcessFlag.UserStdin,
|
|
ProcessFlag.UserStdout, ProcessFlag.UserStderr}
|
|
pipes.flags * mask != {}
|
|
|
|
proc raiseAsyncProcessError(msg: string, exc: ref CatchableError = nil) {.
|
|
noreturn, noinit, noinline, raises: [AsyncProcessError].} =
|
|
let message =
|
|
if isNil(exc):
|
|
msg
|
|
else:
|
|
msg & " ([" & $exc.name & "]: " & $exc.msg & ")"
|
|
raise newException(AsyncProcessError, message)
|
|
|
|
proc raiseAsyncProcessTimeoutError() {.
|
|
noreturn, noinit, noinline, raises: [AsyncProcessTimeoutError].} =
|
|
let message = "Operation timed out"
|
|
raise newException(AsyncProcessTimeoutError, message)
|
|
|
|
proc raiseAsyncProcessError(msg: string, error: OSErrorCode|cint) {.
|
|
noreturn, noinit, noinline, raises: [AsyncProcessError].} =
|
|
when error is OSErrorCode:
|
|
let message = msg & " ([OSError]: " & osErrorMsg(error) & ")"
|
|
else:
|
|
let message = msg & " ([OSError]: " & osErrorMsg(OSErrorCode(error)) & ")"
|
|
raise newException(AsyncProcessError, message)
|
|
|
|
when defined(windows):
|
|
|
|
proc getStdinHandle(pipes: AsyncProcessPipes): HANDLE =
|
|
if pipes.flags * {ProcessFlag.AutoStdin, ProcessFlag.UserStdin} != {}:
|
|
HANDLE(pipes.stdinHandle)
|
|
else:
|
|
getStdHandle(STD_INPUT_HANDLE)
|
|
|
|
proc getStdoutHandle(pipes: AsyncProcessPipes): HANDLE =
|
|
if pipes.flags * {ProcessFlag.AutoStdout, ProcessFlag.UserStdout} != {}:
|
|
HANDLE(pipes.stdoutHandle)
|
|
else:
|
|
getStdHandle(STD_OUTPUT_HANDLE)
|
|
|
|
proc getStderrHandle(pipes: AsyncProcessPipes): HANDLE =
|
|
if pipes.flags * {ProcessFlag.AutoStderr, ProcessFlag.UserStderr,
|
|
ProcessFlag.CopyStdout} != {}:
|
|
HANDLE(pipes.stderrHandle)
|
|
else:
|
|
getStdHandle(STD_ERROR_HANDLE)
|
|
|
|
proc getProcessEnvironment*(): StringTableRef =
|
|
var res = newStringTable(modeCaseInsensitive)
|
|
var env = getEnvironmentStringsW()
|
|
if isNil(env):
|
|
return res
|
|
var slider = env
|
|
while int(slider[]) != 0:
|
|
let pos = wcschr(slider, WCHAR(0x0000))
|
|
let line = slider.toString().valueOr("")
|
|
slider = cast[LPWSTR](cast[uint](pos) + uint(sizeof(WCHAR)))
|
|
if len(line) > 0:
|
|
let delim = line.find('=')
|
|
if delim > 0:
|
|
res[substr(line, 0, delim - 1)] = substr(line, delim + 1)
|
|
discard freeEnvironmentStringsW(env)
|
|
res
|
|
|
|
proc buildCommandLine(a: string, args: openArray[string]): string =
|
|
# TODO: Procedures quoteShell/(Windows, Posix)() needs security and bug review
|
|
# or reimplementation, for example quoteShellWindows() do not handle `\`
|
|
# properly.
|
|
# https://docs.microsoft.com/en-us/cpp/cpp/main-function-command-line-args?redirectedfrom=MSDN&view=msvc-170#parsing-c-command-line-arguments
|
|
var res = quoteShell(a)
|
|
for i in 0 ..< len(args):
|
|
res.add(' ')
|
|
res.add(quoteShell(args[i]))
|
|
res
|
|
|
|
proc buildEnvironment(env: StringTableRef): Result[LPWSTR, OSErrorCode] =
|
|
var str: string
|
|
for key, value in pairs(env):
|
|
doAssert('=' notin key, "`=` must not be present in key name")
|
|
str.add(key)
|
|
str.add('=')
|
|
str.add(value)
|
|
str.add('\x00')
|
|
str.add("\x00\x00")
|
|
toWideString(str)
|
|
|
|
proc closeThreadAndProcessHandle(p: AsyncProcessRef
|
|
): AsyncProcessResult[void] =
|
|
if p.threadHandle != HANDLE(0):
|
|
if closeHandle(p.threadHandle) == FALSE:
|
|
discard closeHandle(p.processHandle)
|
|
return err(osLastError())
|
|
p.threadHandle = HANDLE(0)
|
|
|
|
if p.processHandle != HANDLE(0):
|
|
if closeHandle(p.processHandle) == FALSE:
|
|
return err(osLastError())
|
|
p.processHandle = HANDLE(0)
|
|
|
|
proc startProcess*(command: string, workingDir: string = "",
|
|
arguments: seq[string] = @[],
|
|
environment: StringTableRef = nil,
|
|
options: set[AsyncProcessOption] = {},
|
|
stdinHandle = ProcessStreamHandle(),
|
|
stdoutHandle = ProcessStreamHandle(),
|
|
stderrHandle = ProcessStreamHandle(),
|
|
): Future[AsyncProcessRef] {.
|
|
async: (raises: [AsyncProcessError, CancelledError]).} =
|
|
var
|
|
pipes = preparePipes(options, stdinHandle, stdoutHandle,
|
|
stderrHandle).valueOr:
|
|
raiseAsyncProcessError("Unable to initialze process pipes", error)
|
|
|
|
let
|
|
commandLine =
|
|
if AsyncProcessOption.EvalCommand in options:
|
|
chronosProcShell & " /C " & command
|
|
else:
|
|
buildCommandLine(command, arguments)
|
|
workingDirectory =
|
|
if len(workingDir) > 0:
|
|
workingDir.toWideString().valueOr:
|
|
raiseAsyncProcessError("Unable to proceed working directory path",
|
|
error)
|
|
else:
|
|
nil
|
|
environment =
|
|
if not(isNil(environment)):
|
|
buildEnvironment(environment).valueOr:
|
|
raiseAsyncProcessError("Unable to build child process environment",
|
|
error)
|
|
else:
|
|
nil
|
|
flags = CREATE_UNICODE_ENVIRONMENT
|
|
var
|
|
psa = getSecurityAttributes(false)
|
|
tsa = getSecurityAttributes(false)
|
|
startupInfo =
|
|
block:
|
|
var res = STARTUPINFO(cb: DWORD(sizeof(STARTUPINFO)))
|
|
if pipes.pipesPresent():
|
|
res.dwFlags = STARTF_USESTDHANDLES
|
|
res.hStdInput = pipes.getStdinHandle()
|
|
res.hStdOutput = pipes.getStdoutHandle()
|
|
res.hStdError = pipes.getStderrHandle()
|
|
res
|
|
procInfo = PROCESS_INFORMATION()
|
|
|
|
let wideCommandLine = commandLine.toWideString().valueOr:
|
|
raiseAsyncProcessError("Unable to proceed command line", error)
|
|
|
|
let res = createProcess(
|
|
nil,
|
|
wideCommandLine,
|
|
addr psa, addr tsa,
|
|
TRUE, # NOTE: This is very important flag and MUST not be modified.
|
|
# All overloaded pipe handles will not work if this flag will be
|
|
# set to FALSE.
|
|
flags,
|
|
environment,
|
|
workingDirectory,
|
|
startupInfo, procInfo
|
|
)
|
|
|
|
if(not(isNil(environment))):
|
|
free(environment)
|
|
free(wideCommandLine)
|
|
|
|
var currentError = osLastError()
|
|
if res == FALSE:
|
|
await pipes.closeProcessStreams(options)
|
|
currentError = closeProcessHandles(pipes, options, currentError)
|
|
|
|
if res == FALSE:
|
|
raiseAsyncProcessError("Unable to spawn process", currentError)
|
|
|
|
let process = AsyncProcessRef(
|
|
processHandle: procInfo.hProcess,
|
|
threadHandle: procInfo.hThread,
|
|
processId: procInfo.dwProcessId,
|
|
pipes: pipes,
|
|
options: options,
|
|
flags: pipes.flags
|
|
)
|
|
|
|
trackCounter(AsyncProcessTrackerName)
|
|
return process
|
|
|
|
proc peekProcessExitCode(p: AsyncProcessRef): AsyncProcessResult[int] =
|
|
var wstatus: DWORD = 0
|
|
if p.exitStatus.isSome():
|
|
return ok(p.exitStatus.get())
|
|
|
|
let res = getExitCodeProcess(p.processHandle, wstatus)
|
|
if res == TRUE:
|
|
if wstatus != STILL_ACTIVE:
|
|
let status = int(wstatus)
|
|
p.exitStatus = Opt.some(status)
|
|
ok(status)
|
|
else:
|
|
ok(-1)
|
|
else:
|
|
err(osLastError())
|
|
|
|
proc suspend(p: AsyncProcessRef): AsyncProcessResult[void] =
|
|
if suspendThread(p.threadHandle) != 0xFFFF_FFFF'u32:
|
|
ok()
|
|
else:
|
|
err(osLastError())
|
|
|
|
proc resume(p: AsyncProcessRef): AsyncProcessResult[void] =
|
|
if resumeThread(p.threadHandle) != 0xFFFF_FFFF'u32:
|
|
ok()
|
|
else:
|
|
err(osLastError())
|
|
|
|
proc terminate(p: AsyncProcessRef): AsyncProcessResult[void] =
|
|
if terminateProcess(p.processHandle, 0) != 0'u32:
|
|
ok()
|
|
else:
|
|
err(osLastError())
|
|
|
|
proc kill(p: AsyncProcessRef): AsyncProcessResult[void] =
|
|
p.terminate()
|
|
|
|
proc running(p: AsyncProcessRef): AsyncProcessResult[bool] =
|
|
let res = ? p.peekExitCode()
|
|
if res == -1:
|
|
ok(true)
|
|
else:
|
|
ok(false)
|
|
|
|
proc waitForExit*(p: AsyncProcessRef,
|
|
timeout = InfiniteDuration): Future[int] {.
|
|
async: (raises: [AsyncProcessError, AsyncProcessTimeoutError,
|
|
CancelledError]).} =
|
|
if p.exitStatus.isSome():
|
|
return p.exitStatus.get()
|
|
|
|
let wres =
|
|
try:
|
|
await waitForSingleObject(p.processHandle, timeout)
|
|
except AsyncError as exc:
|
|
raiseAsyncProcessError("Unable to wait for process handle", exc)
|
|
|
|
if wres == WaitableResult.Timeout:
|
|
let res = p.kill()
|
|
if res.isErr():
|
|
raiseAsyncProcessError("Unable to terminate process", res.error())
|
|
|
|
let exitCode = p.peekProcessExitCode().valueOr:
|
|
raiseAsyncProcessError("Unable to peek process exit code", error)
|
|
|
|
if exitCode >= 0:
|
|
p.exitStatus = Opt.some(exitCode)
|
|
|
|
exitCode
|
|
|
|
proc peekExitCode(p: AsyncProcessRef): AsyncProcessResult[int] =
|
|
if p.exitStatus.isSome():
|
|
return ok(p.exitStatus.get())
|
|
let res = waitForSingleObject(p.processHandle, DWORD(0))
|
|
if res != WAIT_TIMEOUT:
|
|
let exitCode = ? p.peekProcessExitCode()
|
|
ok(exitCode)
|
|
else:
|
|
ok(-1)
|
|
else:
|
|
import std/strutils
|
|
|
|
type
|
|
SpawnAttr = object
|
|
attrs: PosixSpawnAttr
|
|
actions: PosixSpawnFileActions
|
|
|
|
proc fd(h: AsyncStreamHolder): cint =
|
|
case h.kind
|
|
of StreamKind.Reader:
|
|
cint(h.reader.tsource.fd)
|
|
of StreamKind.Writer:
|
|
cint(h.writer.tsource.fd)
|
|
of StreamKind.None:
|
|
raiseAssert "Incorrect stream holder"
|
|
|
|
proc isEmpty(h: AsyncStreamHolder): bool =
|
|
h.kind == StreamKind.None
|
|
|
|
proc initSpawn(pipes: AsyncProcessPipes, options: set[AsyncProcessOption]
|
|
): Result[SpawnAttr, OSErrorCode] =
|
|
template doCheck(body: untyped): untyped =
|
|
let res = body
|
|
if res != 0:
|
|
return err(OSErrorCode(res))
|
|
|
|
var
|
|
attrs =
|
|
block:
|
|
var value: PosixSpawnAttr
|
|
let res = posixSpawnAttrInit(value)
|
|
if res != 0:
|
|
return err(OSErrorCode(res))
|
|
value
|
|
actions =
|
|
block:
|
|
var value: PosixSpawnFileActions
|
|
let res = posixSpawnFileActionsInit(value)
|
|
if res != 0:
|
|
discard posixSpawnAttrDestroy(attrs)
|
|
return err(OSErrorCode(res))
|
|
value
|
|
mask =
|
|
block:
|
|
var res: Sigset
|
|
discard sigemptyset(res)
|
|
res
|
|
|
|
doCheck(posixSpawnAttrSetSigMask(attrs, mask))
|
|
if AsyncProcessOption.ProcessGroup in options:
|
|
doCheck(posixSpawnAttrSetPgroup(attrs, 0))
|
|
doCheck(posixSpawnAttrSetFlags(attrs, osdefs.POSIX_SPAWN_USEVFORK or
|
|
osdefs.POSIX_SPAWN_SETSIGMASK or
|
|
osdefs.POSIX_SPAWN_SETPGROUP))
|
|
else:
|
|
doCheck(posixSpawnAttrSetFlags(attrs, osdefs.POSIX_SPAWN_USEVFORK or
|
|
osdefs.POSIX_SPAWN_SETSIGMASK))
|
|
|
|
if pipes.flags * {ProcessFlag.AutoStdin, ProcessFlag.UserStdin} != {}:
|
|
# Close child process STDIN.
|
|
doCheck(posixSpawnFileActionsAddClose(actions, cint(0)))
|
|
# Make a duplicate of `stdinHandle` as child process STDIN.
|
|
doCheck(posixSpawnFileActionsAddDup2(actions, cint(pipes.stdinHandle),
|
|
cint(0)))
|
|
# Close child process side of `stdinHandle`.
|
|
doCheck(posixSpawnFileActionsAddClose(actions,
|
|
cint(pipes.stdinHandle)))
|
|
# Close parent process side of `stdinHandle`.
|
|
if not(pipes.stdinHolder.isEmpty()):
|
|
let fd = cint(pipes.stdinHolder.fd())
|
|
doCheck(posixSpawnFileActionsAddClose(actions, fd))
|
|
|
|
if pipes.flags * {ProcessFlag.AutoStdout, ProcessFlag.UserStdout} != {}:
|
|
# Close child process STDOUT.
|
|
doCheck(posixSpawnFileActionsAddClose(actions, cint(1)))
|
|
# Make a duplicate of `stdoutHandle` as child process STDOUT.
|
|
doCheck(posixSpawnFileActionsAddDup2(actions, cint(pipes.stdoutHandle),
|
|
cint(1)))
|
|
if AsyncProcessOption.StdErrToStdOut notin options:
|
|
# Close child process side of `stdoutHandle`.
|
|
doCheck(posixSpawnFileActionsAddClose(actions,
|
|
cint(pipes.stdoutHandle)))
|
|
# Close parent process side of `stdoutHandle`.
|
|
if not(pipes.stdoutHolder.isEmpty()):
|
|
let fd = cint(pipes.stdoutHolder.fd())
|
|
doCheck(posixSpawnFileActionsAddClose(actions, fd))
|
|
|
|
if pipes.flags * {ProcessFlag.AutoStderr, ProcessFlag.UserStderr} != {}:
|
|
# Close child process STDERR.
|
|
doCheck(posixSpawnFileActionsAddClose(actions, cint(2)))
|
|
# Make a duplicate of `stderrHandle` as child process STDERR.
|
|
doCheck(posixSpawnFileActionsAddDup2(actions, cint(pipes.stderrHandle),
|
|
cint(2)))
|
|
# Close child process side of `stderrHandle`.
|
|
doCheck(posixSpawnFileActionsAddClose(actions,
|
|
cint(pipes.stderrHandle)))
|
|
# Close parent process side of `stderrHandle`.
|
|
if not(pipes.stderrHolder.isEmpty()):
|
|
let fd = cint(pipes.stderrHolder.fd())
|
|
doCheck(posixSpawnFileActionsAddClose(actions, fd))
|
|
else:
|
|
if AsyncProcessOption.StdErrToStdOut in options:
|
|
# Close child process STDERR.
|
|
doCheck(posixSpawnFileActionsAddClose(actions, cint(2)))
|
|
# Make a duplicate of `stdoutHandle` as child process STDERR.
|
|
doCheck(posixSpawnFileActionsAddDup2(actions, cint(pipes.stdoutHandle),
|
|
cint(2)))
|
|
# Close child process side of `stdoutHandle`.
|
|
doCheck(posixSpawnFileActionsAddClose(actions,
|
|
cint(pipes.stdoutHandle)))
|
|
# Close parent process side of `stdoutHandle`.
|
|
if not(pipes.stdoutHolder.isEmpty()):
|
|
let fd = cint(pipes.stdoutHolder.fd())
|
|
doCheck(posixSpawnFileActionsAddClose(actions, fd))
|
|
ok(SpawnAttr(attrs: attrs, actions: actions))
|
|
|
|
proc free(v: var SpawnAttr): Result[void, OSErrorCode] =
|
|
block:
|
|
let res = posixSpawnAttrDestroy(v.attrs)
|
|
if res != 0:
|
|
discard posixSpawnFileActionsDestroy(v.actions)
|
|
return err(OSErrorCode(res))
|
|
block:
|
|
let res = posixSpawnFileActionsDestroy(v.actions)
|
|
if res != 0:
|
|
return err(OSErrorCode(res))
|
|
ok()
|
|
|
|
proc getKeyValueItem(key: string, value: string): cstring =
|
|
var p = cast[cstring](alloc(len(key) + len(value) + 1 + 1))
|
|
var offset = 0
|
|
if len(key) > 0:
|
|
copyMem(addr p[offset], unsafeAddr(key[0]), len(key))
|
|
inc(offset, len(key))
|
|
p[offset] = '='
|
|
inc(offset)
|
|
if len(value) > 0:
|
|
copyMem(addr p[offset], unsafeAddr(value[0]), len(value))
|
|
inc(offset, len(value))
|
|
p[offset] = '\x00'
|
|
p
|
|
|
|
proc envToCStringArray(t: StringTableRef): cstringArray =
|
|
let itemsCount = len(t)
|
|
var
|
|
res = cast[cstringArray](alloc((itemsCount + 1) * sizeof(cstring)))
|
|
i = 0
|
|
for key, value in pairs(t):
|
|
res[i] = getKeyValueItem(key, value)
|
|
inc(i)
|
|
res[i] = nil # Last item in CStringArray should be `nil`.
|
|
res
|
|
|
|
proc envToCStringArray(): cstringArray =
|
|
let itemsCount =
|
|
block:
|
|
var res = 0
|
|
for key, value in envPairs(): inc(res)
|
|
res
|
|
var
|
|
res = cast[cstringArray](alloc((itemsCount + 1) * sizeof(cstring)))
|
|
i = 0
|
|
for key, value in envPairs():
|
|
res[i] = getKeyValueItem(key, value)
|
|
inc(i)
|
|
res[i] = nil # Last item in CStringArray should be `nil`.
|
|
res
|
|
|
|
when defined(macosx) or defined(macos) or defined(ios):
|
|
proc getEnvironment(): ptr cstringArray {.
|
|
importc: "_NSGetEnviron", header: "<crt_externs.h>".}
|
|
else:
|
|
var globalEnv {.importc: "environ", header: "<unistd.h>".}: cstringArray
|
|
|
|
proc getProcessEnvironment*(): StringTableRef =
|
|
var res = newStringTable(modeCaseInsensitive)
|
|
let env =
|
|
when defined(macosx) or defined(macos) or defined(ios):
|
|
getEnvironment()[]
|
|
else:
|
|
globalEnv
|
|
var i = 0
|
|
while not(isNil(env[i])):
|
|
let line = $env[i]
|
|
if len(line) > 0:
|
|
let delim = line.find('=')
|
|
if delim > 0:
|
|
res[substr(line, 0, delim - 1)] = substr(line, delim + 1)
|
|
inc(i)
|
|
res
|
|
|
|
func exitStatusLikeShell(status: int): int =
|
|
if WAITIFSIGNALED(cint(status)):
|
|
# like the shell!
|
|
128 + WAITTERMSIG(cint(status))
|
|
else:
|
|
WAITEXITSTATUS(cint(status))
|
|
|
|
proc getCurrentDirectory(): AsyncProcessResult[string] =
|
|
var bufsize = 1024
|
|
var res = newString(bufsize)
|
|
|
|
proc strLength(a: string): int {.nimcall.} =
|
|
for i in 0 ..< len(a):
|
|
if a[i] == '\x00':
|
|
return i
|
|
len(a)
|
|
|
|
while true:
|
|
if osdefs.getcwd(cstring(res), bufsize) != nil:
|
|
setLen(res, strLength(res))
|
|
return ok(res)
|
|
else:
|
|
let errorCode = osLastError()
|
|
if errorCode == oserrno.ERANGE:
|
|
bufsize = bufsize shl 1
|
|
doAssert(bufsize >= 0)
|
|
res = newString(bufsize)
|
|
else:
|
|
return err(errorCode)
|
|
|
|
proc setCurrentDirectory(dir: string): AsyncProcessResult[void] =
|
|
let res = osdefs.chdir(cstring(dir))
|
|
if res == -1:
|
|
return err(osLastError())
|
|
ok()
|
|
|
|
proc closeThreadAndProcessHandle(p: AsyncProcessRef
|
|
): AsyncProcessResult[void] =
|
|
discard
|
|
|
|
proc startProcess*(command: string, workingDir: string = "",
|
|
arguments: seq[string] = @[],
|
|
environment: StringTableRef = nil,
|
|
options: set[AsyncProcessOption] = {},
|
|
stdinHandle = ProcessStreamHandle(),
|
|
stdoutHandle = ProcessStreamHandle(),
|
|
stderrHandle = ProcessStreamHandle(),
|
|
): Future[AsyncProcessRef] {.
|
|
async: (raises: [AsyncProcessError, CancelledError]).} =
|
|
var
|
|
pid: Pid
|
|
pipes = preparePipes(options, stdinHandle, stdoutHandle,
|
|
stderrHandle).valueOr:
|
|
raiseAsyncProcessError("Unable to initialze process pipes",
|
|
error)
|
|
sa = pipes.initSpawn(options).valueOr:
|
|
discard closeProcessHandles(pipes, options, OSErrorCode(0))
|
|
await pipes.closeProcessStreams(options)
|
|
raiseAsyncProcessError("Unable to initalize spawn attributes", 0)
|
|
|
|
let
|
|
(commandLine, commandArguments) =
|
|
if AsyncProcessOption.EvalCommand in options:
|
|
let args = @[chronosProcShell, "-c", command]
|
|
(chronosProcShell, allocCStringArray(args))
|
|
else:
|
|
var res = @[command]
|
|
for arg in arguments.items():
|
|
res.add(arg)
|
|
(command, allocCStringArray(res))
|
|
commandEnv =
|
|
if isNil(environment):
|
|
envToCStringArray()
|
|
else:
|
|
envToCStringArray(environment)
|
|
|
|
var currentError: OSErrorCode
|
|
var currentDir: string
|
|
|
|
try:
|
|
currentDir =
|
|
if len(workingDir) > 0:
|
|
# Save current working directory and change it to `workingDir`.
|
|
let cres = getCurrentDirectory()
|
|
if cres.isErr():
|
|
raiseAsyncProcessError("Unable to obtain current directory",
|
|
cres.error())
|
|
let sres = setCurrentDirectory(workingDir)
|
|
if sres.isErr():
|
|
raiseAsyncProcessError("Unable to change current directory",
|
|
sres.error())
|
|
cres.get()
|
|
else:
|
|
""
|
|
|
|
let res =
|
|
if AsyncProcessOption.UsePath in options:
|
|
posixSpawnp(pid, cstring(commandLine), sa.actions, sa.attrs,
|
|
commandArguments, commandEnv)
|
|
else:
|
|
posixSpawn(pid, cstring(commandLine), sa.actions, sa.attrs,
|
|
commandArguments, commandEnv)
|
|
|
|
if res != 0:
|
|
await pipes.closeProcessStreams(options)
|
|
currentError = closeProcessHandles(pipes, options, OSErrorCode(res))
|
|
|
|
finally:
|
|
# Restore working directory
|
|
if (len(workingDir) > 0) and (len(currentDir) > 0):
|
|
# Restore working directory.
|
|
let cres = getCurrentDirectory()
|
|
if cres.isErr():
|
|
# On error we still try to restore original working directory.
|
|
if currentError.isOk():
|
|
currentError = cres.error()
|
|
discard setCurrentDirectory(currentDir)
|
|
else:
|
|
if cres.get() != currentDir:
|
|
let sres = setCurrentDirectory(currentDir)
|
|
if sres.isErr():
|
|
if currentError.isOk():
|
|
currentError = sres.error()
|
|
|
|
# Cleanup allocated memory
|
|
deallocCStringArray(commandArguments)
|
|
deallocCStringArray(commandEnv)
|
|
|
|
# Cleanup posix_spawn attributes and file operations
|
|
if not(currentError.isOk()):
|
|
discard sa.free()
|
|
else:
|
|
let res = sa.free()
|
|
if res.isErr():
|
|
currentError = res.error()
|
|
|
|
# If currentError has been set, raising an exception.
|
|
if not(currentError.isOk()):
|
|
raiseAsyncProcessError("Unable to spawn process", currentError)
|
|
|
|
let process = AsyncProcessRef(
|
|
processId: pid,
|
|
pipes: pipes,
|
|
options: options,
|
|
flags: pipes.flags
|
|
)
|
|
|
|
trackCounter(AsyncProcessTrackerName)
|
|
process
|
|
|
|
proc peekProcessExitCode(p: AsyncProcessRef,
|
|
reap = false): AsyncProcessResult[int] =
|
|
var wstatus: cint = 0
|
|
if p.exitStatus.isSome():
|
|
return ok(p.exitStatus.get())
|
|
let
|
|
flags = if reap: cint(0) else: osdefs.WNOHANG
|
|
waitRes =
|
|
block:
|
|
var res: cint = 0
|
|
while true:
|
|
res = osdefs.waitpid(p.processId, wstatus, flags)
|
|
if not((res == -1) and (osLastError() == oserrno.EINTR)):
|
|
break
|
|
res
|
|
if waitRes == p.processId:
|
|
if WAITIFEXITED(wstatus) or WAITIFSIGNALED(wstatus):
|
|
let status = int(wstatus)
|
|
p.exitStatus = Opt.some(status)
|
|
ok(status)
|
|
else:
|
|
ok(-1)
|
|
elif waitRes == 0:
|
|
ok(-1)
|
|
else:
|
|
err(osLastError())
|
|
|
|
proc suspend(p: AsyncProcessRef): AsyncProcessResult[void] =
|
|
if osdefs.kill(p.processId, osdefs.SIGSTOP) == 0:
|
|
ok()
|
|
else:
|
|
err(osLastError())
|
|
|
|
proc resume(p: AsyncProcessRef): AsyncProcessResult[void] =
|
|
if osdefs.kill(p.processId, osdefs.SIGCONT) == 0:
|
|
ok()
|
|
else:
|
|
err(osLastError())
|
|
|
|
proc terminate(p: AsyncProcessRef): AsyncProcessResult[void] =
|
|
if osdefs.kill(p.processId, osdefs.SIGTERM) == 0:
|
|
ok()
|
|
else:
|
|
err(osLastError())
|
|
|
|
proc kill(p: AsyncProcessRef): AsyncProcessResult[void] =
|
|
if osdefs.kill(p.processId, osdefs.SIGKILL) == 0:
|
|
ok()
|
|
else:
|
|
err(osLastError())
|
|
|
|
proc running(p: AsyncProcessRef): AsyncProcessResult[bool] =
|
|
let res = ? p.peekProcessExitCode()
|
|
if res == -1:
|
|
ok(true)
|
|
else:
|
|
ok(false)
|
|
|
|
proc waitForExit*(p: AsyncProcessRef,
|
|
timeout = InfiniteDuration): Future[int] {.
|
|
async: (raw: true, raises: [
|
|
AsyncProcessError, AsyncProcessTimeoutError, CancelledError]).} =
|
|
var
|
|
retFuture = newFuture[int]("chronos.waitForExit()")
|
|
processHandle: ProcessHandle
|
|
timer: TimerCallback = nil
|
|
|
|
if p.exitStatus.isSome():
|
|
retFuture.complete(p.exitStatus.get())
|
|
return retFuture
|
|
|
|
if timeout == ZeroDuration:
|
|
let res = p.kill()
|
|
if res.isErr():
|
|
retFuture.fail(newException(AsyncProcessError, osErrorMsg(res.error())))
|
|
return retFuture
|
|
|
|
block:
|
|
let exitCode = p.peekProcessExitCode().valueOr:
|
|
retFuture.fail(newException(AsyncProcessError, osErrorMsg(error)))
|
|
return retFuture
|
|
if exitCode != -1:
|
|
retFuture.complete(exitStatusLikeShell(exitCode))
|
|
return retFuture
|
|
|
|
if timeout == ZeroDuration:
|
|
retFuture.complete(-1)
|
|
return retFuture
|
|
|
|
proc continuation(udata: pointer) {.gcsafe.} =
|
|
let source = cast[int](udata)
|
|
if not(retFuture.finished()):
|
|
if source == 1:
|
|
# Process exited.
|
|
let res = removeProcess2(processHandle)
|
|
if res.isErr():
|
|
retFuture.fail(newException(AsyncProcessError,
|
|
osErrorMsg(res.error())))
|
|
return
|
|
if not(isNil(timer)):
|
|
clearTimer(timer)
|
|
let exitCode = p.peekProcessExitCode().valueOr:
|
|
retFuture.fail(newException(AsyncProcessError, osErrorMsg(error)))
|
|
return
|
|
if exitCode == -1:
|
|
retFuture.complete(-1)
|
|
else:
|
|
retFuture.complete(exitStatusLikeShell(exitCode))
|
|
else:
|
|
# Timeout exceeded.
|
|
let res = p.kill()
|
|
if res.isErr():
|
|
retFuture.fail(newException(AsyncProcessError,
|
|
osErrorMsg(res.error())))
|
|
|
|
timer = nil
|
|
|
|
proc cancellation(udata: pointer) {.gcsafe.} =
|
|
if not(isNil(timer)):
|
|
clearTimer(timer)
|
|
timer = nil
|
|
# Ignore any errors because of cancellation.
|
|
discard removeProcess2(processHandle)
|
|
|
|
if timeout != InfiniteDuration:
|
|
timer = setTimer(Moment.fromNow(timeout), continuation, cast[pointer](2))
|
|
|
|
processHandle = addProcess2(int(p.processId), continuation,
|
|
cast[pointer](1)).valueOr:
|
|
if error == oserrno.ESRCH:
|
|
# "zombie death race" problem.
|
|
# If process exited right after `waitpid()` - `kqueue` call
|
|
# could return ESRCH error. So we need to handle it properly and
|
|
# try to reap process code from exiting process.
|
|
let exitCode = p.peekProcessExitCode(true).valueOr:
|
|
retFuture.fail(newException(AsyncProcessError, osErrorMsg(error)))
|
|
return retFuture
|
|
if exitCode == -1:
|
|
# This should not be happens one more time, so we just report
|
|
# original error.
|
|
retFuture.fail(newException(AsyncProcessError,
|
|
osErrorMsg(oserrno.ESRCH)))
|
|
else:
|
|
retFuture.complete(exitStatusLikeShell(exitCode))
|
|
else:
|
|
retFuture.fail(newException(AsyncProcessError, osErrorMsg(error)))
|
|
return retFuture
|
|
|
|
# addProcess2() has race condition problem inside. Its possible that child
|
|
# process (we going to wait) sends SIGCHLD right after addProcess2() blocks
|
|
# signals and before it starts monitoring for signal (`signalfd` or
|
|
# `kqueue`). To avoid this problem we going to check process for completion
|
|
# one more time.
|
|
block:
|
|
let exitCode = p.peekProcessExitCode().valueOr:
|
|
discard removeProcess2(processHandle)
|
|
retFuture.fail(newException(AsyncProcessError, osErrorMsg(error)))
|
|
return retFuture
|
|
if exitCode != -1:
|
|
discard removeProcess2(processHandle)
|
|
retFuture.complete(exitStatusLikeShell(exitCode))
|
|
return retFuture
|
|
|
|
# Process is still running, so we going to wait for SIGCHLD.
|
|
retFuture.cancelCallback = cancellation
|
|
retFuture
|
|
|
|
proc peekExitCode(p: AsyncProcessRef): AsyncProcessResult[int] =
|
|
let res = ? p.peekProcessExitCode()
|
|
ok(exitStatusLikeShell(res))
|
|
|
|
proc createPipe(kind: StandardKind
|
|
): Result[tuple[read: AsyncFD, write: AsyncFD], OSErrorCode] =
|
|
case kind
|
|
of StandardKind.Stdin:
|
|
let pipes =
|
|
when defined(windows):
|
|
let
|
|
readFlags: set[DescriptorFlag] = {DescriptorFlag.NonBlock}
|
|
writeFlags: set[DescriptorFlag] = {DescriptorFlag.NonBlock}
|
|
? createOsPipe(readFlags, writeFlags)
|
|
else:
|
|
let
|
|
readFlags: set[DescriptorFlag] = {}
|
|
writeFlags: set[DescriptorFlag] = {DescriptorFlag.NonBlock}
|
|
? createOsPipe(readFlags, writeFlags)
|
|
ok((read: AsyncFD(pipes.read), write: AsyncFD(pipes.write)))
|
|
of StandardKind.Stdout, StandardKind.Stderr:
|
|
let pipes =
|
|
when defined(windows):
|
|
let
|
|
readFlags: set[DescriptorFlag] = {DescriptorFlag.NonBlock}
|
|
writeFlags: set[DescriptorFlag] = {DescriptorFlag.NonBlock}
|
|
? createOsPipe(readFlags, writeFlags)
|
|
else:
|
|
let
|
|
readFlags: set[DescriptorFlag] = {DescriptorFlag.NonBlock}
|
|
writeFlags: set[DescriptorFlag] = {}
|
|
? createOsPipe(readFlags, writeFlags)
|
|
ok((read: AsyncFD(pipes.read), write: AsyncFD(pipes.write)))
|
|
|
|
proc preparePipes(options: set[AsyncProcessOption],
|
|
stdinHandle, stdoutHandle,
|
|
stderrHandle: ProcessStreamHandle
|
|
): AsyncProcessResult[AsyncProcessPipes] =
|
|
|
|
let
|
|
(stdinFlags, localStdin, remoteStdin) =
|
|
case stdinHandle.kind
|
|
of ProcessStreamHandleKind.None:
|
|
({ProcessFlag.NoStdin}, AsyncStreamHolder.init(),
|
|
asyncInvalidPipe)
|
|
of ProcessStreamHandleKind.Auto:
|
|
let (pipeIn, pipeOut) = ? createPipe(StandardKind.Stdin)
|
|
let holder = ? AsyncStreamHolder.init(
|
|
ProcessStreamHandle.init(pipeOut), StreamKind.Writer, {})
|
|
({ProcessFlag.AutoStdin}, holder, pipeIn)
|
|
else:
|
|
({ProcessFlag.UserStdin},
|
|
AsyncStreamHolder.init(), AsyncFD.init(stdinHandle))
|
|
(stdoutFlags, localStdout, remoteStdout) =
|
|
case stdoutHandle.kind
|
|
of ProcessStreamHandleKind.None:
|
|
({ProcessFlag.NoStdout}, AsyncStreamHolder.init(),
|
|
asyncInvalidPipe)
|
|
of ProcessStreamHandleKind.Auto:
|
|
let (pipeIn, pipeOut) = ? createPipe(StandardKind.Stdout)
|
|
let holder = ? AsyncStreamHolder.init(
|
|
ProcessStreamHandle.init(pipeIn), StreamKind.Reader, {})
|
|
({ProcessFlag.AutoStdout}, holder, pipeOut)
|
|
else:
|
|
({ProcessFlag.UserStdout},
|
|
AsyncStreamHolder.init(), AsyncFD.init(stdoutHandle))
|
|
(stderrFlags, localStderr, remoteStderr) =
|
|
if AsyncProcessOption.StdErrToStdOut in options:
|
|
doAssert(stderrHandle.isEmpty(),
|
|
"`stderrHandle` argument must not be set, when" &
|
|
"`AsyncProcessOption.StdErrToStdOut` flag is used")
|
|
case stdoutHandle.kind
|
|
of ProcessStreamHandleKind.None:
|
|
raiseAssert "`stdoutHandle` argument must be present, when " &
|
|
"`AsyncProcessOption.StdErrToStdOut` flag is used"
|
|
of ProcessStreamHandleKind.Auto:
|
|
({ProcessFlag.CopyStdout}, localStdout, remoteStdout)
|
|
else:
|
|
({ProcessFlag.CopyStdout}, localStdout, remoteStdout)
|
|
else:
|
|
case stderrHandle.kind
|
|
of ProcessStreamHandleKind.None:
|
|
({ProcessFlag.NoStderr}, AsyncStreamHolder.init(),
|
|
asyncInvalidPipe)
|
|
of ProcessStreamHandleKind.Auto:
|
|
let (pipeIn, pipeOut) = ? createPipe(StandardKind.Stderr)
|
|
let holder = ? AsyncStreamHolder.init(
|
|
ProcessStreamHandle.init(pipeIn), StreamKind.Reader, {})
|
|
({ProcessFlag.AutoStderr}, holder, pipeOut)
|
|
else:
|
|
({ProcessFlag.UserStderr},
|
|
AsyncStreamHolder.init(), AsyncFD.init(stderrHandle))
|
|
|
|
ok(AsyncProcessPipes(
|
|
flags: stdinFlags + stdoutFlags + stderrFlags,
|
|
stdinHolder: localStdin,
|
|
stdoutHolder: localStdout,
|
|
stderrHolder: localStderr,
|
|
stdinHandle: remoteStdin,
|
|
stdoutHandle: remoteStdout,
|
|
stderrHandle: remoteStderr
|
|
))
|
|
|
|
proc closeWait(holder: AsyncStreamHolder) {.async: (raises: []).} =
|
|
let (future, transp) =
|
|
case holder.kind
|
|
of StreamKind.None:
|
|
(nil, nil)
|
|
of StreamKind.Reader:
|
|
if StreamHolderFlag.Stream in holder.flags:
|
|
(holder.reader.closeWait(), holder.reader.tsource)
|
|
else:
|
|
(nil, holder.reader.tsource)
|
|
of StreamKind.Writer:
|
|
if StreamHolderFlag.Stream in holder.flags:
|
|
(holder.writer.closeWait(), holder.writer.tsource)
|
|
else:
|
|
(nil, holder.writer.tsource)
|
|
|
|
let pending =
|
|
block:
|
|
var res: seq[Future[void]]
|
|
if not(isNil(future)):
|
|
res.add(future)
|
|
if not(isNil(transp)):
|
|
if StreamHolderFlag.Transport in holder.flags:
|
|
res.add(transp.closeWait())
|
|
res
|
|
|
|
if len(pending) > 0:
|
|
await noCancel allFutures(pending)
|
|
|
|
proc closeProcessStreams(pipes: AsyncProcessPipes,
|
|
options: set[AsyncProcessOption]): Future[void] {.
|
|
async: (raw: true, raises: []).} =
|
|
let pending =
|
|
block:
|
|
var res: seq[Future[void]]
|
|
if ProcessFlag.AutoStdin in pipes.flags:
|
|
res.add(pipes.stdinHolder.closeWait())
|
|
if ProcessFlag.AutoStdout in pipes.flags:
|
|
res.add(pipes.stdoutHolder.closeWait())
|
|
if ProcessFlag.AutoStderr in pipes.flags:
|
|
res.add(pipes.stderrHolder.closeWait())
|
|
res
|
|
noCancel allFutures(pending)
|
|
|
|
proc opAndWaitForExit(p: AsyncProcessRef, op: WaitOperation,
|
|
timeout = InfiniteDuration): Future[int] {.
|
|
async: (raises: [
|
|
AsyncProcessError, AsyncProcessTimeoutError, CancelledError]).} =
|
|
let timerFut =
|
|
if timeout == InfiniteDuration:
|
|
newFuture[void]("chronos.killAndwaitForExit")
|
|
else:
|
|
sleepAsync(timeout)
|
|
|
|
while true:
|
|
if p.running().get(true):
|
|
# We ignore operation errors because we going to repeat calling
|
|
# operation until process will not exit.
|
|
case op
|
|
of WaitOperation.Kill:
|
|
discard p.kill()
|
|
of WaitOperation.Terminate:
|
|
discard p.terminate()
|
|
else:
|
|
let exitCode = p.peekExitCode().valueOr:
|
|
raiseAsyncProcessError("Unable to peek process exit code", error)
|
|
if not(timerFut.finished()):
|
|
await cancelAndWait(timerFut)
|
|
return exitCode
|
|
|
|
let waitFut = p.waitForExit().wait(100.milliseconds)
|
|
try:
|
|
discard await race(FutureBase(waitFut), FutureBase(timerFut))
|
|
except ValueError:
|
|
raiseAssert "This should not be happened!"
|
|
|
|
if waitFut.finished() and not(waitFut.failed()):
|
|
let res = p.peekExitCode()
|
|
if res.isOk():
|
|
if not(timerFut.finished()):
|
|
await cancelAndWait(timerFut)
|
|
return res.get()
|
|
|
|
if timerFut.finished():
|
|
if not(waitFut.finished()):
|
|
await waitFut.cancelAndWait()
|
|
raiseAsyncProcessTimeoutError()
|
|
|
|
proc closeWait*(p: AsyncProcessRef) {.async: (raises: []).} =
|
|
# Here we ignore all possible errrors, because we do not want to raise
|
|
# exceptions.
|
|
discard closeProcessHandles(p.pipes, p.options, OSErrorCode(0))
|
|
await p.pipes.closeProcessStreams(p.options)
|
|
discard p.closeThreadAndProcessHandle()
|
|
untrackCounter(AsyncProcessTrackerName)
|
|
|
|
proc stdinStream*(p: AsyncProcessRef): AsyncStreamWriter =
|
|
## Returns STDIN async stream associated with process `p`.
|
|
doAssert(p.pipes.stdinHolder.kind == StreamKind.Writer,
|
|
"StdinStreamWriter is not available")
|
|
p.pipes.stdinHolder.writer
|
|
|
|
proc stdoutStream*(p: AsyncProcessRef): AsyncStreamReader =
|
|
## Returns STDOUT async stream associated with process `p`.
|
|
doAssert(p.pipes.stdoutHolder.kind == StreamKind.Reader,
|
|
"StdoutStreamReader is not available")
|
|
p.pipes.stdoutHolder.reader
|
|
|
|
proc stderrStream*(p: AsyncProcessRef): AsyncStreamReader =
|
|
## Returns STDERR async stream associated with process `p`.
|
|
doAssert(p.pipes.stderrHolder.kind == StreamKind.Reader,
|
|
"StderrStreamReader is not available")
|
|
p.pipes.stderrHolder.reader
|
|
|
|
proc execCommand*(command: string,
|
|
options = {AsyncProcessOption.EvalCommand},
|
|
timeout = InfiniteDuration
|
|
): Future[int] {.
|
|
async: (raises: [
|
|
AsyncProcessError, AsyncProcessTimeoutError, CancelledError]).} =
|
|
let
|
|
poptions = options + {AsyncProcessOption.EvalCommand}
|
|
process = await startProcess(command, options = poptions)
|
|
res =
|
|
try:
|
|
await process.waitForExit(timeout)
|
|
finally:
|
|
await process.closeWait()
|
|
res
|
|
|
|
proc execCommandEx*(command: string,
|
|
options = {AsyncProcessOption.EvalCommand},
|
|
timeout = InfiniteDuration
|
|
): Future[CommandExResponse] {.
|
|
async: (raises: [
|
|
AsyncProcessError, AsyncProcessTimeoutError, CancelledError]).} =
|
|
let
|
|
process = await startProcess(command, options = options,
|
|
stdoutHandle = AsyncProcess.Pipe,
|
|
stderrHandle = AsyncProcess.Pipe)
|
|
outputReader = process.stdoutStream.read()
|
|
errorReader = process.stderrStream.read()
|
|
res =
|
|
try:
|
|
await allFutures(outputReader, errorReader)
|
|
let
|
|
status = await process.waitForExit(timeout)
|
|
output =
|
|
try:
|
|
string.fromBytes(await outputReader)
|
|
except AsyncStreamError as exc:
|
|
raiseAsyncProcessError("Unable to read process' stdout channel",
|
|
exc)
|
|
error =
|
|
try:
|
|
string.fromBytes(await errorReader)
|
|
except AsyncStreamError as exc:
|
|
raiseAsyncProcessError("Unable to read process' stderr channel",
|
|
exc)
|
|
CommandExResponse(status: status, stdOutput: output, stdError: error)
|
|
finally:
|
|
await process.closeWait()
|
|
|
|
res
|
|
|
|
proc pid*(p: AsyncProcessRef): int =
|
|
## Returns process ``p`` unique process identifier.
|
|
int(p.processId)
|
|
|
|
template processId*(p: AsyncProcessRef): int = pid(p)
|
|
|
|
proc killAndWaitForExit*(p: AsyncProcessRef,
|
|
timeout = InfiniteDuration): Future[int] {.
|
|
async: (raw: true, raises: [
|
|
AsyncProcessError, AsyncProcessTimeoutError, CancelledError]).} =
|
|
## Perform continuous attempts to kill the ``p`` process for specified period
|
|
## of time ``timeout``.
|
|
##
|
|
## On Posix systems, killing means sending ``SIGKILL`` to the process ``p``,
|
|
## On Windows, it uses ``TerminateProcess`` to kill the process ``p``.
|
|
##
|
|
## If the process ``p`` fails to be killed within the ``timeout`` time, it
|
|
## will raise ``AsyncProcessTimeoutError``.
|
|
##
|
|
## In case of error this it will raise ``AsyncProcessError``.
|
|
##
|
|
## Returns process ``p`` exit code.
|
|
opAndWaitForExit(p, WaitOperation.Kill, timeout)
|
|
|
|
proc terminateAndWaitForExit*(p: AsyncProcessRef,
|
|
timeout = InfiniteDuration): Future[int] {.
|
|
async: (raw: true, raises: [
|
|
AsyncProcessError, AsyncProcessTimeoutError, CancelledError]).} =
|
|
## Perform continuous attempts to terminate the ``p`` process for specified
|
|
## period of time ``timeout``.
|
|
##
|
|
## On Posix systems, terminating means sending ``SIGTERM`` to the process
|
|
## ``p``, on Windows, it uses ``TerminateProcess`` to terminate the process
|
|
## ``p``.
|
|
##
|
|
## If the process ``p`` fails to be terminated within the ``timeout`` time, it
|
|
## will raise ``AsyncProcessTimeoutError``.
|
|
##
|
|
## In case of error this it will raise ``AsyncProcessError``.
|
|
##
|
|
## Returns process ``p`` exit code.
|
|
opAndWaitForExit(p, WaitOperation.Terminate, timeout)
|