Module rsyscall.sys.timerfd

#include <sys/timerfd.h>

"`#include <sys/timerfd.h>`"
from __future__ import annotations
from rsyscall._raw import lib # type: ignore
from rsyscall.near.sysif import SyscallInterface
from rsyscall.sys.syscall import SYS
import enum
import rsyscall.near.types as near
import typing as t
from rsyscall.handle.fd import BaseFileDescriptor, FileDescriptorTask
from rsyscall.handle.pointer import Pointer, WrittenPointer

# re-exported
from rsyscall.time import Timespec, Itimerspec

__all__ = [
    "CLOCK",
    "TFD",
    "TFD_TIMER",
    "TimerFileDescriptor",
    "Timespec",
    "Itimerspec",
]

class CLOCK(enum.IntEnum):
    REALTIME = lib.CLOCK_REALTIME
    MONOTONIC = lib.CLOCK_MONOTONIC
    BOOTTIME = lib.CLOCK_BOOTTIME
    REALTIME_ALARM = lib.CLOCK_REALTIME_ALARM
    BOOTTIME_ALARM = lib.CLOCK_BOOTTIME_ALARM

class TFD(enum.IntFlag):
    NONE = 0
    CLOEXEC = lib.EFD_CLOEXEC
    NONBLOCK = lib.EFD_NONBLOCK

class TFD_TIMER(enum.IntFlag):
    NONE = 0
    ABSTIME = lib.TFD_TIMER_ABSTIME
    CANCEL_ON_SET = lib.TFD_TIMER_CANCEL_ON_SET

async def _timerfd_create(sysif: SyscallInterface, clockid: CLOCK, flags: TFD) -> near.FileDescriptor:
    return near.FileDescriptor(await sysif.syscall(SYS.timerfd_create, clockid, flags))

async def _timerfd_settime(sysif: SyscallInterface, fd: near.FileDescriptor,
                           flags: TFD_TIMER,
                           new_value: near.Address, old_value: t.Optional[near.Address]) -> None:
    if old_value is None:
        old_value = 0 # type: ignore
    await sysif.syscall(SYS.timerfd_settime, fd, flags, new_value, old_value)

async def _timerfd_gettime(sysif: SyscallInterface, fd: near.FileDescriptor,
                           curr_value: near.Address) -> None:
    await sysif.syscall(SYS.timerfd_gettime, fd, curr_value)


T_fd = t.TypeVar('T_fd', bound='TimerFileDescriptor')
class TimerFileDescriptor(BaseFileDescriptor):
    @t.overload
    async def timerfd_settime(
            self, flags: TFD_TIMER, new_value: WrittenPointer[Itimerspec]) -> None: ...

    @t.overload
    async def timerfd_settime(
            self, flags: TFD_TIMER, new_value: WrittenPointer[Itimerspec],
            old_value: Pointer[Itimerspec]) -> Pointer[Itimerspec]: ...

    async def timerfd_settime(
            self, flags: TFD_TIMER, new_value: WrittenPointer[Itimerspec],
            old_value: t.Optional[Pointer[Itimerspec]]=None) -> t.Optional[Pointer[Itimerspec]]:
        self._validate()
        with new_value.borrow(self.task):
            if old_value:
                with old_value.borrow(self.task):
                    await _timerfd_settime(
                        self.task.sysif, self.near, flags, new_value.near, old_value.near)
                    return old_value
            else:
                await _timerfd_settime(
                    self.task.sysif, self.near, flags, new_value.near, None)
                return None

    async def timerfd_gettime(self, curr_value: Pointer[Itimerspec]) -> Pointer[Itimerspec]:
        self._validate()
        with curr_value.borrow(self.task):
            await _timerfd_gettime(self.task.sysif, self.near, curr_value.near)
            return curr_value

class TimerfdTask(FileDescriptorTask[T_fd]):
    async def timerfd_create(self, clockid: CLOCK, flags: TFD=TFD.NONE) -> T_fd:
        return self.make_fd_handle(await _timerfd_create(self.sysif, clockid, flags|TFD.CLOEXEC))

Classes

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

An enumeration.

Expand source code Browse git

class CLOCK(enum.IntEnum):
    REALTIME = lib.CLOCK_REALTIME
    MONOTONIC = lib.CLOCK_MONOTONIC
    BOOTTIME = lib.CLOCK_BOOTTIME
    REALTIME_ALARM = lib.CLOCK_REALTIME_ALARM
    BOOTTIME_ALARM = lib.CLOCK_BOOTTIME_ALARM

Ancestors

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

Class variables

var REALTIME

var MONOTONIC

var BOOTTIME

var REALTIME_ALARM

var BOOTTIME_ALARM

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

An enumeration.

Expand source code Browse git

class TFD(enum.IntFlag):
    NONE = 0
    CLOEXEC = lib.EFD_CLOEXEC
    NONBLOCK = lib.EFD_NONBLOCK

Ancestors

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

Class variables

var NONE

var CLOEXEC

var NONBLOCK

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

An enumeration.

Expand source code Browse git

class TFD_TIMER(enum.IntFlag):
    NONE = 0
    ABSTIME = lib.TFD_TIMER_ABSTIME
    CANCEL_ON_SET = lib.TFD_TIMER_CANCEL_ON_SET

Ancestors

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

Class variables

var NONE

var ABSTIME

var CANCEL_ON_SET

class TimerFileDescriptor (task: FileDescriptorTask, near: FileDescriptor, valid: bool)

A file descriptor accessed through some Task

This is an rsyscall-internal base class, which other FileDescriptor objects inherit from for core lifecycle methods. See FileDescriptor for more information.

Expand source code Browse git

class TimerFileDescriptor(BaseFileDescriptor):
    @t.overload
    async def timerfd_settime(
            self, flags: TFD_TIMER, new_value: WrittenPointer[Itimerspec]) -> None: ...

    @t.overload
    async def timerfd_settime(
            self, flags: TFD_TIMER, new_value: WrittenPointer[Itimerspec],
            old_value: Pointer[Itimerspec]) -> Pointer[Itimerspec]: ...

    async def timerfd_settime(
            self, flags: TFD_TIMER, new_value: WrittenPointer[Itimerspec],
            old_value: t.Optional[Pointer[Itimerspec]]=None) -> t.Optional[Pointer[Itimerspec]]:
        self._validate()
        with new_value.borrow(self.task):
            if old_value:
                with old_value.borrow(self.task):
                    await _timerfd_settime(
                        self.task.sysif, self.near, flags, new_value.near, old_value.near)
                    return old_value
            else:
                await _timerfd_settime(
                    self.task.sysif, self.near, flags, new_value.near, None)
                return None

    async def timerfd_gettime(self, curr_value: Pointer[Itimerspec]) -> Pointer[Itimerspec]:
        self._validate()
        with curr_value.borrow(self.task):
            await _timerfd_gettime(self.task.sysif, self.near, curr_value.near)
            return curr_value

Ancestors

• BaseFileDescriptor

Subclasses

• FileDescriptor

Methods

async def timerfd_settime(self, flags: TFD_TIMER, new_value: WrittenPointer[Itimerspec], old_value: t.Optional[Pointer[Itimerspec]] = None) ‑> Optional[Pointer[Itimerspec]]

Expand source code Browse git

async def timerfd_settime(
        self, flags: TFD_TIMER, new_value: WrittenPointer[Itimerspec],
        old_value: t.Optional[Pointer[Itimerspec]]=None) -> t.Optional[Pointer[Itimerspec]]:
    self._validate()
    with new_value.borrow(self.task):
        if old_value:
            with old_value.borrow(self.task):
                await _timerfd_settime(
                    self.task.sysif, self.near, flags, new_value.near, old_value.near)
                return old_value
        else:
            await _timerfd_settime(
                self.task.sysif, self.near, flags, new_value.near, None)
            return None

async def timerfd_gettime(self, curr_value: Pointer[Itimerspec]) ‑> Pointer[Itimerspec]

Expand source code Browse git

async def timerfd_gettime(self, curr_value: Pointer[Itimerspec]) -> Pointer[Itimerspec]:
    self._validate()
    with curr_value.borrow(self.task):
        await _timerfd_gettime(self.task.sysif, self.near, curr_value.near)
        return curr_value

Inherited members

• BaseFileDescriptor:
  • borrow
  • close
  • for_task
  • inherit
  • invalidate
  • move
  • near
  • task
  • valid

class Timespec (sec: int, nsec: int)

struct timespec, as used by several time-related system-calls.

This struct specifies time with nanosecond precision, but there's no good standard way in Python to represent such times. The growing standard is to use an integer number of nanoseconds, but that's easy to confuse with an integer number of seconds, and most functions don't take number-of-nanoseconds, they take number-of-seconds.

So, this class supports conversion to and from a bunch of other formats.

See the proposal of using Decimal to represent nanosecond timestamps:

https://www.python.org/dev/peps/pep-0410/ https://www.python.org/dev/peps/pep-0564/ https://vstinner.github.io/python37-pep-564-nanoseconds.html

The rejection of that proposal by Guido:

https://mail.python.org/pipermail/python-dev/2012-February/116837.html https://bugs.python.org/issue23084

Expand source code Browse git

@dataclass
class Timespec:
    """struct timespec, as used by several time-related system-calls.

    This struct specifies time with nanosecond precision, but there's no good standard way
    in Python to represent such times. The growing standard is to use an integer number of
    nanoseconds, but that's easy to confuse with an integer number of seconds, and most
    functions don't take number-of-nanoseconds, they take number-of-seconds.

    So, this class supports conversion to and from a bunch of other formats.

    See the proposal of using Decimal to represent nanosecond timestamps:

    https://www.python.org/dev/peps/pep-0410/
    https://www.python.org/dev/peps/pep-0564/
    https://vstinner.github.io/python37-pep-564-nanoseconds.html

    The rejection of that proposal by Guido:

    https://mail.python.org/pipermail/python-dev/2012-February/116837.html
    https://bugs.python.org/issue23084

    """
    sec: int
    nsec: int

    def to_decimal(self) -> Decimal:
        raise NotImplementedError

    def to_nanos(self) -> int:
        raise NotImplementedError

    def _to_cffi_dict(self) -> t.Dict[str, int]:
        return {
            "tv_sec": self.sec,
            "tv_nsec": self.nsec,
        }

    def to_bytes(self) -> bytes:
        return bytes(ffi.buffer(ffi.new('struct timespec const*', self._to_cffi_dict())))

    T = t.TypeVar('T', bound='Timespec')
    @classmethod
    def from_float(cls: t.Type[T], value: t.Union[float, Decimal, Timespec]) -> T:
        if isinstance(value, Timespec):
            return cls(value.sec, value.nsec)
        elif isinstance(value, Decimal):
            frac, i = decimal.getcontext().divmod(value, Decimal(1))
            return cls(int(i), int(frac*NSEC_PER_SEC))
        else:
            fractional, integer = math.modf(value)
            return cls(int(integer), int(fractional*NSEC_PER_SEC))

    @classmethod
    def from_nanos(cls: t.Type[T], nsec: int) -> T:
        raise NotImplementedError

    @classmethod
    def from_cffi(cls: t.Type[T], cffi_value: t.Any) -> T:
        return cls(cffi_value.tv_sec, cffi_value.tv_nsec)

    @classmethod
    def from_bytes(cls: t.Type[T], data: bytes) -> T:
        struct = ffi.cast('struct timespec*', ffi.from_buffer(data))
        return cls.from_cffi(struct)

    @classmethod
    def sizeof(cls) -> int:
        return ffi.sizeof('struct timespec')

Class variables

var sec : int

var nsec : int

var T

Static methods

def from_float(value: t.Union[float, Decimal, Timespec]) ‑> T

Expand source code Browse git

@classmethod
def from_float(cls: t.Type[T], value: t.Union[float, Decimal, Timespec]) -> T:
    if isinstance(value, Timespec):
        return cls(value.sec, value.nsec)
    elif isinstance(value, Decimal):
        frac, i = decimal.getcontext().divmod(value, Decimal(1))
        return cls(int(i), int(frac*NSEC_PER_SEC))
    else:
        fractional, integer = math.modf(value)
        return cls(int(integer), int(fractional*NSEC_PER_SEC))

def from_nanos(nsec: int) ‑> T

Expand source code Browse git

@classmethod
def from_nanos(cls: t.Type[T], nsec: int) -> T:
    raise NotImplementedError

def from_cffi(cffi_value: t.Any) ‑> T

Expand source code Browse git

@classmethod
def from_cffi(cls: t.Type[T], cffi_value: t.Any) -> T:
    return cls(cffi_value.tv_sec, cffi_value.tv_nsec)

def from_bytes(data: bytes) ‑> T

Expand source code Browse git

@classmethod
def from_bytes(cls: t.Type[T], data: bytes) -> T:
    struct = ffi.cast('struct timespec*', ffi.from_buffer(data))
    return cls.from_cffi(struct)

def sizeof() ‑> int

Expand source code Browse git

@classmethod
def sizeof(cls) -> int:
    return ffi.sizeof('struct timespec')

Methods

def to_decimal(self) ‑> decimal.Decimal

Expand source code Browse git

def to_decimal(self) -> Decimal:
    raise NotImplementedError

def to_nanos(self) ‑> int

Expand source code Browse git

def to_nanos(self) -> int:
    raise NotImplementedError

def to_bytes(self) ‑> bytes

Expand source code Browse git

def to_bytes(self) -> bytes:
    return bytes(ffi.buffer(ffi.new('struct timespec const*', self._to_cffi_dict())))

class Itimerspec (interval: t.Union[float, Decimal, Timespec], value: t.Union[float, Decimal, Timespec])

Itimerspec(interval: 't.Union[float, Decimal, Timespec]', value: 't.Union[float, Decimal, Timespec]') -> 'None'

Expand source code Browse git

@dataclass
class Itimerspec(Struct):
    interval: Timespec
    value: Timespec

    def __init__(self, interval: t.Union[float, Decimal, Timespec],
                 value: t.Union[float, Decimal, Timespec]) -> None:
        self.interval = Timespec.from_float(interval)
        self.value = Timespec.from_float(interval)

    def _to_cffi_dict(self) -> t.Dict[str, t.Dict[str, int]]:
        return {
            "it_interval": self.interval._to_cffi_dict(),
            "it_value": self.value._to_cffi_dict(),
        }

    def to_bytes(self) -> bytes:
        return bytes(ffi.buffer(ffi.new('struct itimerspec const*', self._to_cffi_dict())))

    T = t.TypeVar('T', bound='Itimerspec')
    @classmethod
    def from_bytes(cls: t.Type[T], data: bytes) -> T:
        struct = ffi.cast('struct itimerspec*', ffi.from_buffer(data))
        return cls(Timespec.from_cffi(struct.it_interval), Timespec.from_cffi(struct.it_value))

    @classmethod
    def sizeof(cls) -> int:
        return ffi.sizeof('struct itimerspec')

Ancestors

• Struct
• Serializable
• FixedSize
• FixedSerializer
• HasSerializer

Class variables

var interval : Timespec

var value : Timespec

var T

Inherited members

• Struct:
  • from_bytes
  • get_self_serializer
  • get_serializer
  • sizeof
  • to_bytes