/
_pthread_sigmask.py
279 lines (237 loc) · 9.93 KB
/
_pthread_sigmask.py
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
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
# 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(3)`` 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 glibc import (
NSIG, SIG_BLOCK, SIG_UNBLOCK, SIG_SETMASK, sigset_t, sigemptyset,
sigaddset, sigismember, pthread_sigmask as _pthread_sigmask,
sigprocmask as _sigprocmask)
__all__ = ['pthread_sigmask', 'sigprocmask']
class _sigxxxmask_base(object):
"""
Pythonic wrapper around ``sigprocmask(2)`` and ``pthread_sigmask(3)``
This base class is defined to share the bulk of the code between
thread and process-wide version of the mask class. The actual
difference is in the implementation of :meth:`_do_mask()`
"""
__slots__ = ('_signals', '_setmask', '_mask', '_old_mask', '_is_active')
def __init__(self, signals=None, setmask=False):
"""
Initialize a new mask 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::
The masking function (``sigprocmask(2)`` or ``pthread_sigmask(3)``)
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 "<{} signals:{} mode:{} active:{}>".format(
self.__class__.__name__, 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 mask object.
.. 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()
self._do_mask(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)
self._do_mask(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)
self._do_mask(SIG_UNBLOCK, delta_mask, None)
# Reset signals to the new value
self._signals = new_signals
def block(self):
"""
Use the masking function (``sigprocmask(2)`` or ``pthread_sigmask(3)``)
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)
self._do_mask(SIG_SETMASK, self._mask, self._old_mask)
else:
self._do_mask(SIG_BLOCK, self._mask, None)
self._is_active = True
def unblock(self):
"""
Use the masking function (``sigprocmask(2)`` or ``pthread_sigmask(3)``)
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!")
self._do_mask(SIG_SETMASK, self._old_mask, None)
self._old_mask = None
else:
self._do_mask(SIG_UNBLOCK, self._mask, None)
self._is_active = False
@classmethod
def get(cls):
"""
Use the masking function (``sigprocmask(2)`` or ``pthread_sigmask(3)``)
to obtain the mask of blocked signals
:returns:
A fresh :class:`sigprocmask` 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)
cls._do_mask(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
@classmethod
def _do_mask(cls, how, set, oldset):
"""
Internal method that calls ``sigprocmask(2) or ``pthread_sigmask(3)``,
depending on the subclass (:class:`pthread_sigmask` or
:class:`sigprocmask` respectively)
"""
raise NotImplementedError
class pthread_sigmask(_sigxxxmask_base):
"""
Pythonic wrapper around the ``pthread_sigmask(3)``
"""
@classmethod
def _do_mask(cls, how, set, oldset):
return _pthread_sigmask(how, set, oldset)
class sigprocmask(_sigxxxmask_base):
"""
Pythonic wrapper around the ``sigprocmask(2)``
"""
@classmethod
def _do_mask(cls, how, set, oldset):
return _sigprocmask(how, set, oldset)