Module rsyscall.sys.wait

#include <sys/wait.h>

"`#include <sys/wait.h>`"
from __future__ import annotations
import typing as t
from dataclasses import dataclass
from rsyscall._raw import ffi, lib # type: ignore
from rsyscall.signal import Siginfo, SIG
from rsyscall.command import Command
import enum

class IdType(enum.IntEnum):
    PID = lib.P_PID # Wait for the child whose process ID matches id.
    PGID = lib.P_PGID # Wait for any child whose process group ID matches id.
    ALL = lib.P_ALL # Wait for any child; id is ignored.

class CLD(enum.IntEnum):
    EXITED = lib.CLD_EXITED # child called _exit(2)
    KILLED = lib.CLD_KILLED # child killed by signal
    DUMPED = lib.CLD_DUMPED # child killed by signal, and dumped core
    STOPPED = lib.CLD_STOPPED # child stopped by signal
    TRAPPED = lib.CLD_TRAPPED # traced child has trapped
    CONTINUED = lib.CLD_CONTINUED # child continued by SIGCONT

class W(enum.IntFlag):
    # wait for different statuses
    EXITED = lib.WEXITED
    STOPPED = lib.WSTOPPED
    CONTINUED = lib.WCONTINUED
    # additional options
    NOHANG = lib.WNOHANG
    NOWAIT = lib.WNOWAIT
    # wait for different kinds of children
    # Note, these are only supported in waitid after Linux 4.7;
    # before that, they were only supported in wait4.
    CLONE = lib._WCLONE
    ALL = lib._WALL
    NOTHREAD = lib._WNOTHREAD


class UncleanExit(Exception):
    "Thrown when a process exits uncleanly"
    state: ChildState
    "State of the child at exit"
    command: t.Optional[Command]
    "Optionally attached to UncleanExit as useful information for debugging"

    def __init__(self, state: ChildState, command: Command=None) -> None:
        super().__init__(state, command)
        self.state = state
        self.command = command

@dataclass
class ChildState:
    code: CLD
    pid: int
    uid: int
    exit_status: t.Optional[int]
    sig: t.Optional[SIG]

    @staticmethod
    def make(code: CLD, pid: int, uid: int, status: int) -> ChildState:
        if code is CLD.EXITED:
            return ChildState(code, pid, uid, status, None)
        else:
            return ChildState(code, pid, uid, None, SIG(status))

    @staticmethod
    def make_from_siginfo(siginfo: Siginfo) -> ChildState:
        return ChildState.make(CLD(siginfo.code),
                               pid=siginfo.pid, uid=siginfo.uid,
                               status=siginfo.status)

    def state(self, options: W) -> bool:
        """Return true if this W option would have returned this state change

        Mainly useful for categorizing state changes into EXITED,
        STOPPED or CONTINUED.

        """
        return bool(options & {
            CLD.EXITED: W.EXITED,
            CLD.KILLED: W.EXITED,
            CLD.DUMPED: W.EXITED,
            CLD.STOPPED: W.STOPPED,
            CLD.TRAPPED: W.STOPPED,
            CLD.CONTINUED: W.CONTINUED,
        }[self.code])

    def died(self) -> bool:
        return self.code in {CLD.EXITED, CLD.KILLED, CLD.DUMPED}

    def clean(self) -> bool:
        return self.code == CLD.EXITED and self.exit_status == 0

    def check(self) -> None:
        if self.clean():
            return None
        else:
            raise UncleanExit(self)

    def killed_with(self) -> SIG:
        """What signal was the child killed with?

        Throws if the child was not killed with a signal.

        """
        if not self.died():
            raise Exception("Child isn't dead")
        if self.sig is None:
            raise Exception("Child wasn't killed with a signal")
        return self.sig


#### Raw syscalls ####
from rsyscall.near.sysif import SyscallInterface
from rsyscall.sys.syscall import SYS
from rsyscall.near.types import (
    Address,
    Process,
    ProcessGroup,
)

async def _waitid(sysif: SyscallInterface,
                  id: t.Union[Process, ProcessGroup, None], infop: t.Optional[Address], options: int,
                  rusage: t.Optional[Address]) -> int:
    if isinstance(id, Process):
        idtype = IdType.PID
    elif isinstance(id, ProcessGroup):
        idtype = IdType.PGID
    elif id is None:
        idtype = IdType.ALL
        id = 0 # type: ignore
    else:
        raise ValueError("unknown id type", id)
    if infop is None:
        infop = 0 # type: ignore
    if rusage is None:
        rusage = 0 # type: ignore
    return (await sysif.syscall(SYS.waitid, idtype, id, infop, options, rusage))


#### Tests ####
from unittest import TestCase

class TestWait(TestCase):
    def test_child_state(self) -> None:
        state = ChildState.make_from_siginfo(Siginfo(code=CLD.EXITED, pid=1, uid=13, status=1))
        self.assertFalse(state.clean())

Classes

class IdType (value, names=None, *, module=None, qualname=None, type=None, start=1)

An enumeration.

Expand source code Browse git

class IdType(enum.IntEnum):
    PID = lib.P_PID # Wait for the child whose process ID matches id.
    PGID = lib.P_PGID # Wait for any child whose process group ID matches id.
    ALL = lib.P_ALL # Wait for any child; id is ignored.

Ancestors

• enum.IntEnum
• builtins.int
• enum.Enum

Class variables

var PID

var PGID

var ALL

class CLD (value, names=None, *, module=None, qualname=None, type=None, start=1)

An enumeration.

Expand source code Browse git

class CLD(enum.IntEnum):
    EXITED = lib.CLD_EXITED # child called _exit(2)
    KILLED = lib.CLD_KILLED # child killed by signal
    DUMPED = lib.CLD_DUMPED # child killed by signal, and dumped core
    STOPPED = lib.CLD_STOPPED # child stopped by signal
    TRAPPED = lib.CLD_TRAPPED # traced child has trapped
    CONTINUED = lib.CLD_CONTINUED # child continued by SIGCONT

Ancestors

• enum.IntEnum
• builtins.int
• enum.Enum

Class variables

var EXITED

var KILLED

var DUMPED

var STOPPED

var TRAPPED

var CONTINUED

class W (value, names=None, *, module=None, qualname=None, type=None, start=1)

An enumeration.

Expand source code Browse git

class W(enum.IntFlag):
    # wait for different statuses
    EXITED = lib.WEXITED
    STOPPED = lib.WSTOPPED
    CONTINUED = lib.WCONTINUED
    # additional options
    NOHANG = lib.WNOHANG
    NOWAIT = lib.WNOWAIT
    # wait for different kinds of children
    # Note, these are only supported in waitid after Linux 4.7;
    # before that, they were only supported in wait4.
    CLONE = lib._WCLONE
    ALL = lib._WALL
    NOTHREAD = lib._WNOTHREAD

Ancestors

• enum.IntFlag
• builtins.int
• enum.Flag
• enum.Enum

Class variables

var EXITED

var STOPPED

var CONTINUED

var NOHANG

var NOWAIT

var CLONE

var ALL

var NOTHREAD

class UncleanExit (state: ChildState, command: Command = None)

Thrown when a process exits uncleanly

Expand source code Browse git

class UncleanExit(Exception):
    "Thrown when a process exits uncleanly"
    state: ChildState
    "State of the child at exit"
    command: t.Optional[Command]
    "Optionally attached to UncleanExit as useful information for debugging"

    def __init__(self, state: ChildState, command: Command=None) -> None:
        super().__init__(state, command)
        self.state = state
        self.command = command

Ancestors

• builtins.Exception
• builtins.BaseException

Class variables

var state : ChildState

State of the child at exit

var command : Optional[Command]

Optionally attached to UncleanExit as useful information for debugging

class ChildState (code: CLD, pid: int, uid: int, exit_status: t.Optional[int], sig: t.Optional[SIG])

ChildState(code: 'CLD', pid: 'int', uid: 'int', exit_status: 't.Optional[int]', sig: 't.Optional[SIG]')

Expand source code Browse git

@dataclass
class ChildState:
    code: CLD
    pid: int
    uid: int
    exit_status: t.Optional[int]
    sig: t.Optional[SIG]

    @staticmethod
    def make(code: CLD, pid: int, uid: int, status: int) -> ChildState:
        if code is CLD.EXITED:
            return ChildState(code, pid, uid, status, None)
        else:
            return ChildState(code, pid, uid, None, SIG(status))

    @staticmethod
    def make_from_siginfo(siginfo: Siginfo) -> ChildState:
        return ChildState.make(CLD(siginfo.code),
                               pid=siginfo.pid, uid=siginfo.uid,
                               status=siginfo.status)

    def state(self, options: W) -> bool:
        """Return true if this W option would have returned this state change

        Mainly useful for categorizing state changes into EXITED,
        STOPPED or CONTINUED.

        """
        return bool(options & {
            CLD.EXITED: W.EXITED,
            CLD.KILLED: W.EXITED,
            CLD.DUMPED: W.EXITED,
            CLD.STOPPED: W.STOPPED,
            CLD.TRAPPED: W.STOPPED,
            CLD.CONTINUED: W.CONTINUED,
        }[self.code])

    def died(self) -> bool:
        return self.code in {CLD.EXITED, CLD.KILLED, CLD.DUMPED}

    def clean(self) -> bool:
        return self.code == CLD.EXITED and self.exit_status == 0

    def check(self) -> None:
        if self.clean():
            return None
        else:
            raise UncleanExit(self)

    def killed_with(self) -> SIG:
        """What signal was the child killed with?

        Throws if the child was not killed with a signal.

        """
        if not self.died():
            raise Exception("Child isn't dead")
        if self.sig is None:
            raise Exception("Child wasn't killed with a signal")
        return self.sig

Class variables

var code : CLD

var pid : int

var uid : int

var exit_status : Optional[int]

var sig : Optional[SIG]

Static methods

def make(code: CLD, pid: int, uid: int, status: int) ‑> ChildState

Expand source code Browse git

@staticmethod
def make(code: CLD, pid: int, uid: int, status: int) -> ChildState:
    if code is CLD.EXITED:
        return ChildState(code, pid, uid, status, None)
    else:
        return ChildState(code, pid, uid, None, SIG(status))

def make_from_siginfo(siginfo: Siginfo) ‑> ChildState

Expand source code Browse git

@staticmethod
def make_from_siginfo(siginfo: Siginfo) -> ChildState:
    return ChildState.make(CLD(siginfo.code),
                           pid=siginfo.pid, uid=siginfo.uid,
                           status=siginfo.status)

Methods

def state(self, options: W) ‑> bool

Return true if this W option would have returned this state change

Mainly useful for categorizing state changes into EXITED, STOPPED or CONTINUED.

Expand source code Browse git

def state(self, options: W) -> bool:
    """Return true if this W option would have returned this state change

    Mainly useful for categorizing state changes into EXITED,
    STOPPED or CONTINUED.

    """
    return bool(options & {
        CLD.EXITED: W.EXITED,
        CLD.KILLED: W.EXITED,
        CLD.DUMPED: W.EXITED,
        CLD.STOPPED: W.STOPPED,
        CLD.TRAPPED: W.STOPPED,
        CLD.CONTINUED: W.CONTINUED,
    }[self.code])

def died(self) ‑> bool

Expand source code Browse git

def died(self) -> bool:
    return self.code in {CLD.EXITED, CLD.KILLED, CLD.DUMPED}

def clean(self) ‑> bool

Expand source code Browse git

def clean(self) -> bool:
    return self.code == CLD.EXITED and self.exit_status == 0

def check(self) ‑> NoneType

Expand source code Browse git

def check(self) -> None:
    if self.clean():
        return None
    else:
        raise UncleanExit(self)

def killed_with(self) ‑> SIG

What signal was the child killed with?

Throws if the child was not killed with a signal.

Expand source code Browse git

def killed_with(self) -> SIG:
    """What signal was the child killed with?

    Throws if the child was not killed with a signal.

    """
    if not self.died():
        raise Exception("Child isn't dead")
    if self.sig is None:
        raise Exception("Child wasn't killed with a signal")
    return self.sig

class TestWait (methodName='runTest')

A class whose instances are single test cases.

By default, the test code itself should be placed in a method named 'runTest'.

If the fixture may be used for many test cases, create as many test methods as are needed. When instantiating such a TestCase subclass, specify in the constructor arguments the name of the test method that the instance is to execute.

Test authors should subclass TestCase for their own tests. Construction and deconstruction of the test's environment ('fixture') can be implemented by overriding the 'setUp' and 'tearDown' methods respectively.

If it is necessary to override the init method, the base class init method must always be called. It is important that subclasses should not change the signature of their init method, since instances of the classes are instantiated automatically by parts of the framework in order to be run.

When subclassing TestCase, you can set these attributes: * failureException: determines which exception will be raised when the instance's assertion methods fail; test methods raising this exception will be deemed to have 'failed' rather than 'errored'. * longMessage: determines whether long messages (including repr of objects used in assert methods) will be printed on failure in addition to any explicit message passed. * maxDiff: sets the maximum length of a diff in failure messages by assert methods using difflib. It is looked up as an instance attribute so can be configured by individual tests if required.

Create an instance of the class that will use the named test method when executed. Raises a ValueError if the instance does not have a method with the specified name.

Expand source code Browse git

class TestWait(TestCase):
    def test_child_state(self) -> None:
        state = ChildState.make_from_siginfo(Siginfo(code=CLD.EXITED, pid=1, uid=13, status=1))
        self.assertFalse(state.clean())

Ancestors

• unittest.case.TestCase

Methods

def test_child_state(self) ‑> NoneType

Expand source code Browse git

def test_child_state(self) -> None:
    state = ChildState.make_from_siginfo(Siginfo(code=CLD.EXITED, pid=1, uid=13, status=1))
    self.assertFalse(state.clean())