CODEWITHSUNDEEP
pythonsignals
Velkan
Velkan

Reputation: 7602

Block SIGUSR1 while handling SIGTERM in Python (sa_mask of sigaction(2))

The sigaction(2) call has a mask parameter that allows to block other signals of different type while handling a signal.

While handling SIGTERM I want SIGUSR1 to be blocked. The Python signal module doesn't seem to have the sigaction call, there is only a function to block signals (that will cause a race condition). How to set this block mask for the handler?

Upvotes: 1

Views: 241

Answers (1)

Velkan
Velkan

Reputation: 7602

A Python signal handler does not get executed inside the low-level (C) signal handler. Instead, the low-level signal handler sets a flag which tells the virtual machine to execute the corresponding Python signal handler at a later point(for example at the next bytecode instruction).

So it doesn't make sense to try blocking. Although it kinda keeps it blocked:

from ctypes import Structure, POINTER, CFUNCTYPE, CDLL, byref, get_errno, \
    sizeof, c_int, c_long, c_ulong, c_void_p
from ctypes.util import find_library
from typing import Callable


class SIGSET_T(Structure):
    _fields_ = [
        ("__val", c_ulong * (1024 // (8 * sizeof(c_long)))),
    ]

SA_HANDLER_FUNCTYPE = CFUNCTYPE(None, c_int, use_errno=True)

class SIGACTION(Structure):
    _fields_ = [
        ("sa_handler", SA_HANDLER_FUNCTYPE),
        ("sa_mask", SIGSET_T),
        ("sa_flags", c_int),
        ("sa_restorer", c_void_p),
    ]


libc = CDLL(find_library('c'), use_errno=True)


def sigaction(signal, sighandler: Callable[[int], None], masked_signals=None):
    """Change the action taken by a process on receipt
    of a specific signal.

    A partial Python wrapper for the sigaction(2) function.

    :param signal: a signal value (from the signal module)
    :param sighandler: handler to install
    :param masked_signals: list of other signals to block while handling

    :raises: OSError if a library call fails
    """
    action = SIGACTION(sa_handler=SA_HANDLER_FUNCTYPE(sighandler))

    for maksed_signal in (masked_signals or []):
        if libc.sigaddset(byref(action.sa_mask), maksed_signal.value) != 0:
            if get_errno() != 0:
                raise OSError(get_errno())

    if libc.sigaction(signal.value, byref(action), None) != 0:
        if get_errno() != 0:
            raise OSError(get_errno())

Upvotes: 1

Related Questions