codex-storage
/
nim-chronos
mirror of https://github.com/codex-storage/nim-chronos.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Make asyncproc use asyncraises. (#497) 

* Make asyncproc use asyncraises.

* Fix missing asyncraises for waitForExit().
This commit is contained in:
Eugene Kabanov 2024-01-23 09:34:10 +02:00 committed by GitHub
parent e296ae30c8
commit 09a0b11719
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 50 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

73
chronos/asyncproc.nim
View File

@ -231,8 +231,9 @@ proc closeProcessHandles(pipes: var AsyncProcessPipes,
lastError: OSErrorCode): OSErrorCode {.apforward.} lastError: OSErrorCode): OSErrorCode {.apforward.}
proc closeProcessStreams(pipes: AsyncProcessPipes, proc closeProcessStreams(pipes: AsyncProcessPipes,
options: set[AsyncProcessOption]): Future[void] {. options: set[AsyncProcessOption]): Future[void] {.
apforward.} async: (raises: []).}
proc closeWait(holder: AsyncStreamHolder): Future[void] {.apforward.} proc closeWait(holder: AsyncStreamHolder): Future[void] {.
async: (raises: []).}
template isOk(code: OSErrorCode): bool = template isOk(code: OSErrorCode): bool =
when defined(windows): when defined(windows):
@ -391,7 +392,8 @@ when defined(windows):
stdinHandle = ProcessStreamHandle(), stdinHandle = ProcessStreamHandle(),
stdoutHandle = ProcessStreamHandle(), stdoutHandle = ProcessStreamHandle(),
stderrHandle = ProcessStreamHandle(), stderrHandle = ProcessStreamHandle(),
): Future[AsyncProcessRef] {.async.} = ): Future[AsyncProcessRef] {.
async: (raises: [AsyncProcessError, CancelledError]).} =
var var
pipes = preparePipes(options, stdinHandle, stdoutHandle, pipes = preparePipes(options, stdinHandle, stdoutHandle,
stderrHandle).valueOr: stderrHandle).valueOr:
@ -517,14 +519,16 @@ when defined(windows):
ok(false) ok(false)
proc waitForExit*(p: AsyncProcessRef, proc waitForExit*(p: AsyncProcessRef,
timeout = InfiniteDuration): Future[int] {.async.} = timeout = InfiniteDuration): Future[int] {.
async: (raises: [AsyncProcessError, AsyncProcessTimeoutError,
CancelledError]).} =
if p.exitStatus.isSome(): if p.exitStatus.isSome():
return p.exitStatus.get() return p.exitStatus.get()
let wres = let wres =
try: try:
await waitForSingleObject(p.processHandle, timeout) await waitForSingleObject(p.processHandle, timeout)
except ValueError as exc: except AsyncError as exc:
raiseAsyncProcessError("Unable to wait for process handle", exc) raiseAsyncProcessError("Unable to wait for process handle", exc)
if wres == WaitableResult.Timeout: if wres == WaitableResult.Timeout:
@ -537,7 +541,8 @@ when defined(windows):
if exitCode >= 0: if exitCode >= 0:
p.exitStatus = Opt.some(exitCode) p.exitStatus = Opt.some(exitCode)
return exitCode
exitCode
proc peekExitCode(p: AsyncProcessRef): AsyncProcessResult[int] = proc peekExitCode(p: AsyncProcessRef): AsyncProcessResult[int] =
if p.exitStatus.isSome(): if p.exitStatus.isSome():
@ -787,7 +792,8 @@ else:
stdinHandle = ProcessStreamHandle(), stdinHandle = ProcessStreamHandle(),
stdoutHandle = ProcessStreamHandle(), stdoutHandle = ProcessStreamHandle(),
stderrHandle = ProcessStreamHandle(), stderrHandle = ProcessStreamHandle(),
): Future[AsyncProcessRef] {.async.} = ): Future[AsyncProcessRef] {.
async: (raises: [AsyncProcessError, CancelledError]).} =
var var
pid: Pid pid: Pid
pipes = preparePipes(options, stdinHandle, stdoutHandle, pipes = preparePipes(options, stdinHandle, stdoutHandle,
@ -887,7 +893,7 @@ else:
) )
trackCounter(AsyncProcessTrackerName) trackCounter(AsyncProcessTrackerName)
return process process
proc peekProcessExitCode(p: AsyncProcessRef, proc peekProcessExitCode(p: AsyncProcessRef,
reap = false): AsyncProcessResult[int] = reap = false): AsyncProcessResult[int] =
@ -948,7 +954,9 @@ else:
ok(false) ok(false)
proc waitForExit*(p: AsyncProcessRef, proc waitForExit*(p: AsyncProcessRef,
timeout = InfiniteDuration): Future[int] = timeout = InfiniteDuration): Future[int] {.
async: (raw: true, raises: [
AsyncProcessError, AsyncProcessTimeoutError, CancelledError]).} =
var var
retFuture = newFuture[int]("chronos.waitForExit()") retFuture = newFuture[int]("chronos.waitForExit()")
processHandle: ProcessHandle processHandle: ProcessHandle
@ -1050,7 +1058,7 @@ else:
# Process is still running, so we going to wait for SIGCHLD. # Process is still running, so we going to wait for SIGCHLD.
retFuture.cancelCallback = cancellation retFuture.cancelCallback = cancellation
return retFuture retFuture
proc peekExitCode(p: AsyncProcessRef): AsyncProcessResult[int] = proc peekExitCode(p: AsyncProcessRef): AsyncProcessResult[int] =
let res = ? p.peekProcessExitCode() let res = ? p.peekProcessExitCode()
@ -1155,7 +1163,7 @@ proc preparePipes(options: set[AsyncProcessOption],
stderrHandle: remoteStderr stderrHandle: remoteStderr
)) ))
proc closeWait(holder: AsyncStreamHolder) {.async.} = proc closeWait(holder: AsyncStreamHolder) {.async: (raises: []).} =
let (future, transp) = let (future, transp) =
case holder.kind case holder.kind
of StreamKind.None: of StreamKind.None:
@ -1182,10 +1190,11 @@ proc closeWait(holder: AsyncStreamHolder) {.async.} =
res res
if len(pending) > 0: if len(pending) > 0:
await allFutures(pending) await noCancel allFutures(pending)
proc closeProcessStreams(pipes: AsyncProcessPipes, proc closeProcessStreams(pipes: AsyncProcessPipes,
options: set[AsyncProcessOption]): Future[void] = options: set[AsyncProcessOption]): Future[void] {.
async: (raw: true, raises: []).} =
let pending = let pending =
block: block:
var res: seq[Future[void]] var res: seq[Future[void]]
@ -1196,10 +1205,12 @@ proc closeProcessStreams(pipes: AsyncProcessPipes,
if ProcessFlag.AutoStderr in pipes.flags: if ProcessFlag.AutoStderr in pipes.flags:
res.add(pipes.stderrHolder.closeWait()) res.add(pipes.stderrHolder.closeWait())
res res
allFutures(pending) noCancel allFutures(pending)
proc opAndWaitForExit(p: AsyncProcessRef, op: WaitOperation, proc opAndWaitForExit(p: AsyncProcessRef, op: WaitOperation,
timeout = InfiniteDuration): Future[int] {.async.} = timeout = InfiniteDuration): Future[int] {.
async: (raises: [
AsyncProcessError, AsyncProcessTimeoutError, CancelledError]).} =
let timerFut = let timerFut =
if timeout == InfiniteDuration: if timeout == InfiniteDuration:
newFuture[void]("chronos.killAndwaitForExit") newFuture[void]("chronos.killAndwaitForExit")
@ -1223,7 +1234,10 @@ proc opAndWaitForExit(p: AsyncProcessRef, op: WaitOperation,
return exitCode return exitCode
let waitFut = p.waitForExit().wait(100.milliseconds) let waitFut = p.waitForExit().wait(100.milliseconds)
try:
discard await race(FutureBase(waitFut), FutureBase(timerFut)) discard await race(FutureBase(waitFut), FutureBase(timerFut))
except ValueError:
raiseAssert "This should not be happened!"
if waitFut.finished() and not(waitFut.failed()): if waitFut.finished() and not(waitFut.failed()):
let res = p.peekExitCode() let res = p.peekExitCode()
@ -1237,25 +1251,28 @@ proc opAndWaitForExit(p: AsyncProcessRef, op: WaitOperation,
await waitFut.cancelAndWait() await waitFut.cancelAndWait()
raiseAsyncProcessTimeoutError() raiseAsyncProcessTimeoutError()
proc closeWait*(p: AsyncProcessRef) {.async.} = proc closeWait*(p: AsyncProcessRef) {.async: (raises: []).} =
# Here we ignore all possible errrors, because we do not want to raise # Here we ignore all possible errrors, because we do not want to raise
# exceptions. # exceptions.
discard closeProcessHandles(p.pipes, p.options, OSErrorCode(0)) discard closeProcessHandles(p.pipes, p.options, OSErrorCode(0))
await noCancel(p.pipes.closeProcessStreams(p.options)) await p.pipes.closeProcessStreams(p.options)
discard p.closeThreadAndProcessHandle() discard p.closeThreadAndProcessHandle()
untrackCounter(AsyncProcessTrackerName) untrackCounter(AsyncProcessTrackerName)
proc stdinStream*(p: AsyncProcessRef): AsyncStreamWriter = proc stdinStream*(p: AsyncProcessRef): AsyncStreamWriter =
## Returns STDIN async stream associated with process `p`.
doAssert(p.pipes.stdinHolder.kind == StreamKind.Writer, doAssert(p.pipes.stdinHolder.kind == StreamKind.Writer,
"StdinStreamWriter is not available") "StdinStreamWriter is not available")
p.pipes.stdinHolder.writer p.pipes.stdinHolder.writer
proc stdoutStream*(p: AsyncProcessRef): AsyncStreamReader = proc stdoutStream*(p: AsyncProcessRef): AsyncStreamReader =
## Returns STDOUT async stream associated with process `p`.
doAssert(p.pipes.stdoutHolder.kind == StreamKind.Reader, doAssert(p.pipes.stdoutHolder.kind == StreamKind.Reader,
"StdoutStreamReader is not available") "StdoutStreamReader is not available")
p.pipes.stdoutHolder.reader p.pipes.stdoutHolder.reader
proc stderrStream*(p: AsyncProcessRef): AsyncStreamReader = proc stderrStream*(p: AsyncProcessRef): AsyncStreamReader =
## Returns STDERR async stream associated with process `p`.
doAssert(p.pipes.stderrHolder.kind == StreamKind.Reader, doAssert(p.pipes.stderrHolder.kind == StreamKind.Reader,
"StderrStreamReader is not available") "StderrStreamReader is not available")
p.pipes.stderrHolder.reader p.pipes.stderrHolder.reader
@ -1263,7 +1280,9 @@ proc stderrStream*(p: AsyncProcessRef): AsyncStreamReader =
proc execCommand*(command: string, proc execCommand*(command: string,
options = {AsyncProcessOption.EvalCommand}, options = {AsyncProcessOption.EvalCommand},
timeout = InfiniteDuration timeout = InfiniteDuration
): Future[int] {.async.} = ): Future[int] {.
async: (raises: [
AsyncProcessError, AsyncProcessTimeoutError, CancelledError]).} =
let let
poptions = options + {AsyncProcessOption.EvalCommand} poptions = options + {AsyncProcessOption.EvalCommand}
process = await startProcess(command, options = poptions) process = await startProcess(command, options = poptions)
@ -1277,7 +1296,9 @@ proc execCommand*(command: string,
proc execCommandEx*(command: string, proc execCommandEx*(command: string,
options = {AsyncProcessOption.EvalCommand}, options = {AsyncProcessOption.EvalCommand},
timeout = InfiniteDuration timeout = InfiniteDuration
): Future[CommandExResponse] {.async.} = ): Future[CommandExResponse] {.
async: (raises: [
AsyncProcessError, AsyncProcessTimeoutError, CancelledError]).} =
let let
process = await startProcess(command, options = options, process = await startProcess(command, options = options,
stdoutHandle = AsyncProcess.Pipe, stdoutHandle = AsyncProcess.Pipe,
@ -1291,13 +1312,13 @@ proc execCommandEx*(command: string,
status = await process.waitForExit(timeout) status = await process.waitForExit(timeout)
output = output =
try: try:
string.fromBytes(outputReader.read()) string.fromBytes(await outputReader)
except AsyncStreamError as exc: except AsyncStreamError as exc:
raiseAsyncProcessError("Unable to read process' stdout channel", raiseAsyncProcessError("Unable to read process' stdout channel",
exc) exc)
error = error =
try: try:
string.fromBytes(errorReader.read()) string.fromBytes(await errorReader)
except AsyncStreamError as exc: except AsyncStreamError as exc:
raiseAsyncProcessError("Unable to read process' stderr channel", raiseAsyncProcessError("Unable to read process' stderr channel",
exc) exc)
@ -1308,13 +1329,15 @@ proc execCommandEx*(command: string,
res res
proc pid*(p: AsyncProcessRef): int = proc pid*(p: AsyncProcessRef): int =
## Returns process ``p`` identifier. ## Returns process ``p`` unique process identifier.
int(p.processId) int(p.processId)
template processId*(p: AsyncProcessRef): int = pid(p) template processId*(p: AsyncProcessRef): int = pid(p)
proc killAndWaitForExit*(p: AsyncProcessRef, proc killAndWaitForExit*(p: AsyncProcessRef,
timeout = InfiniteDuration): Future[int] = timeout = InfiniteDuration): Future[int] {.
async: (raw: true, raises: [
AsyncProcessError, AsyncProcessTimeoutError, CancelledError]).} =
## Perform continuous attempts to kill the ``p`` process for specified period ## Perform continuous attempts to kill the ``p`` process for specified period
## of time ``timeout``. ## of time ``timeout``.
## ##
@ -1330,7 +1353,9 @@ proc killAndWaitForExit*(p: AsyncProcessRef,
opAndWaitForExit(p, WaitOperation.Kill, timeout) opAndWaitForExit(p, WaitOperation.Kill, timeout)
proc terminateAndWaitForExit*(p: AsyncProcessRef, proc terminateAndWaitForExit*(p: AsyncProcessRef,
timeout = InfiniteDuration): Future[int] = timeout = InfiniteDuration): Future[int] {.
async: (raw: true, raises: [
AsyncProcessError, AsyncProcessTimeoutError, CancelledError]).} =
## Perform continuous attempts to terminate the ``p`` process for specified ## Perform continuous attempts to terminate the ``p`` process for specified
## period of time ``timeout``. ## period of time ``timeout``.
## ##