/usr/lib/python3/dist-packages/plainbox/vendor/pyglibc/_pthread_sigmask.py is in python3-plainbox 0.25-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 | # Copyright (c) 2014 Canonical Ltd.
#
# Author: Zygmunt Krynicki <zygmunt.krynicki@canonical.com>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
"""
:mod:`pyglibc._pthread_sigmask` -- python wrapper around pthread_sigmask
========================================================================
.. warning::
``pthread_sigmask(2)`` operates on the mask of flags associated with the
calling thread. Therefore there are no thread safety considerations as the
whole object is inherently unsafe (or thread-specific).
"""
from __future__ import absolute_import
from plainbox.vendor.glibc import (
NSIG, SIG_BLOCK, SIG_UNBLOCK, SIG_SETMASK, sigset_t, sigemptyset,
sigaddset, sigismember, pthread_sigmask as _pthread_sigmask)
__all__ = ['pthread_sigmask']
class pthread_sigmask(object):
"""
Pythonic wrapper around the ``pthread_sigmask(2)``
"""
__slots__ = ('_signals', '_setmask', '_mask', '_old_mask', '_is_active')
def __init__(self, signals=None, setmask=False):
"""
Initialize a new pthread_sigmask object.
:param signals:
List of signals to block
:param setmask:
A flag that controls if ``SIG_SETMASK`` should be used over
``SIG_BLOCK`` and ``SIG_UNBLOCK``. For details see :meth:`block()`
and :meth:`unblock()`.
.. note::
``pthread_sigmask(2)`` is not called until :meth:`block()` is
called.
"""
if signals is None:
self._signals = frozenset()
else:
self._signals = frozenset(signals)
self._setmask = setmask
self._mask = sigset_t()
self._old_mask = None # old mask is only used for SIG_SETMASK
self._is_active = False
sigemptyset(self._mask)
for signal in self.signals:
sigaddset(self._mask, signal)
def __repr__(self):
return "<pthread_sigmask signals:{} mode:{} active:{}>".format(
self._signals,
"SIG_SETMASK" if self._setmask else "SIG_BLOCK",
"yes" if self.is_active else "no")
def __enter__(self):
"""
Part of the context manager protocol.
This method calls :meth:`block()`.
:returns:
self
"""
self.block()
return self
def __exit__(self, *args):
"""
Part of the context manager protocol.
This method calls :meth:`unblock()`.
"""
self.unblock()
@property
def is_active(self):
"""
Flag that remebers if the signals are blocked.
If is_active is True, modifications to :meth:`signals` are applied
instantly.
"""
return self._is_active
@property
def signals(self):
"""
associated set of blocked signals
:returns:
The frozenset of signals associated with this pthread_sigmask.
.. note::
Whether the signals returned by this method are currently blocked
or not depends on the circumstances. They can be assumed to be
blocked after :meth:`block()` returns but any code running after
that may alter the effective mask thus rendering this value stale.
This property can be assigned to. Doing so while :meth:`is_active` is
True will update the effective mask on the fly. Otherwise modifications
are buffered until :meth:`enter()` is first called.
"""
return self._signals
@signals.setter
def signals(self, new_signals):
# Convert signals to frozendict as we depend on that below
new_signals = frozenset(new_signals)
# Reset the mask to what signals describes
sigemptyset(self._mask)
for signal in new_signals:
sigaddset(self._mask, signal)
# If we're active, re-apply the changes
if self.is_active:
# In setmask mode we can just overwrite the old values directly
if self._setmask:
# NOTE: we're not updating self._old_mask here. This way
# unblock will trully restore everything despite modifications
# to signals that happened after the call to block()
_pthread_sigmask(SIG_SETMASK, self._mask, None)
else:
# in the non-setmask mode, let's just apply the delta
delta_mask = sigset_t()
# Let's start blocking the new signals first
added_signals = new_signals - self._signals
if added_signals:
sigemptyset(delta_mask)
for signal in added_signals:
sigaddset(delta_mask, signal)
_pthread_sigmask(SIG_BLOCK, delta_mask, None)
# Let's unblock signals next
removed_signals = self._signals - new_signals
if removed_signals:
sigemptyset(delta_mask)
for signal in removed_signals:
sigaddset(delta_mask, signal)
_pthread_sigmask(SIG_UNBLOCK, delta_mask, None)
# Reset signals to the new value
self._signals = new_signals
def block(self):
"""
Use ``pthread_sigmask(2)`` to block signals.
This method uses either ``SIG_SETMASK`` or ``SIG_BLOCK``, depending on
how the object was constructed. After this method is called, the
subsequent call to :meth:`unblock()` will undo its effects.
.. note::
This method is a no-op if signal blocking is currently active (as
determined by :meth:`is_active` returning True).
"""
if self._is_active:
return
if self._setmask:
self._old_mask = sigset_t()
sigemptyset(self._old_mask)
_pthread_sigmask(SIG_SETMASK, self._mask, self._old_mask)
else:
_pthread_sigmask(SIG_BLOCK, self._mask, None)
self._is_active = True
def unblock(self):
"""
Use ``pthread_sigmask(2)`` to unblock signals.
:raises ValueError:
If the old mask is not obtained yet. This only happens in when
``setmask=True`` was passed to the initializer and ``unblock()`` is
called before ``block()`` was called.
This method uses either ``SIG_SETMASK`` or ``SIG_UNBLOCK``, depending
on how the object was constructed. Actual behavior differs as explained
below. In both cases the term *old mask* refers to the effective mask
that was obtained at the time :meth:`block()` was called.
- In the ``SIG_SETMASK`` mode the old mask is restored (overwrite)
- In the ``SIG_UNBLOCK`` mode the old mask is ignored and the desired
signals are unblocked (incremental change)
.. note::
This method is a no-op if signal blocking is currently inactive (as
determined by :meth:`is_active` returning False).
"""
if not self._is_active:
return
if self._setmask:
if self._old_mask is None:
raise ValueError("block() wasn't called yet!")
_pthread_sigmask(SIG_SETMASK, self._old_mask, None)
self._old_mask = None
else:
_pthread_sigmask(SIG_UNBLOCK, self._mask, None)
self._is_active = False
@classmethod
def get(cls):
"""
Use ``pthread_sigmask(2)`` to obtain the mask of blocked signals
:returns:
A fresh :class:`pthread_sigmask` object.
The returned object behaves as it was constructed with the list of
currently blocked signals, ``setmask=False`` and as if the
:meth:`block()` was immediately called.
That is, calling :meth:`unblock()` will will cause those signals not to
be blocked anymore while calling :meth:`block()` will re-block them (if
they were unblocked after this method returns).
"""
mask = sigset_t()
sigemptyset(mask)
_pthread_sigmask(0, None, mask)
signals = []
for sig_num in range(1, NSIG):
if sigismember(mask, sig_num):
signals.append(sig_num)
self = cls(signals)
self._is_active = True
self._old_mask = mask
return self
|