.. % thread.interrupt_main() (used from IDLE), so it's still needed.
+.. cfunction:: int PySignal_SetWakeupFd(int fd)
+
+ This utility function specifies a file descriptor to which a ``'\0'`` byte will
+ be written whenever a signal is received. It returns the previous such file
+ descriptor. The value ``-1`` disables the feature; this is the initial state.
+ This is equivalent to :func:`signal.set_wakeup_fd` in Python, but without any
+ error checking. *fd* should be a valid file descriptor. The function should
+ only be called from the main thread.
+
+
.. cfunction:: PyObject* PyErr_NewException(char *name, PyObject *base, PyObject *dict)
This utility function creates and returns a new exception object. The *name*
:manpage:`signal(2)`.)
+.. function:: set_wakeup_fd(fd)
+
+ Set the wakeup fd to *fd*. When a signal is received, a ``'\0'`` byte is
+ written to the fd. This can be used by a library to wakeup a poll or select
+ call, allowing the signal to be fully processed.
+
+ The old wakeup fd is returned. *fd* must be non-blocking. It is up to the
+ library to remove any bytes before calling poll or select again.
+
+ When threads are enabled, this function can only be called from the main thread;
+ attempting to call it from other threads will cause a :exc:`ValueError`
+ exception to be raised.
+
+
.. function:: signal(signalnum, handler)
Set the handler for signal *signalnum* to the function *handler*. *handler* can
PyAPI_FUNC(int) PyErr_CheckSignals(void);
PyAPI_FUNC(void) PyErr_SetInterrupt(void);
+/* In signalmodule.c */
+int PySignal_SetWakeupFd(int fd);
+
/* Support for adding program text to SyntaxErrors */
PyAPI_FUNC(void) PyErr_SyntaxLocation(const char *, int);
PyAPI_FUNC(PyObject *) PyErr_ProgramText(const char *, int);
self.assertRaises(TypeError, signal.signal,
signal.SIGUSR1, None)
+class WakeupSignalTests(unittest.TestCase):
+ TIMEOUT_FULL = 10
+ TIMEOUT_HALF = 5
+
+ def test_wakeup_fd_early(self):
+ import select
+
+ signal.alarm(1)
+ before_time = time.time()
+ # We attempt to get a signal during the sleep,
+ # before select is called
+ time.sleep(self.TIMEOUT_FULL)
+ mid_time = time.time()
+ self.assert_(mid_time - before_time < self.TIMEOUT_HALF)
+ select.select([self.read], [], [], self.TIMEOUT_FULL)
+ after_time = time.time()
+ self.assert_(after_time - mid_time < self.TIMEOUT_HALF)
+
+ def test_wakeup_fd_during(self):
+ import select
+
+ signal.alarm(1)
+ before_time = time.time()
+ # We attempt to get a signal during the select call
+ self.assertRaises(select.error, select.select,
+ [self.read], [], [], self.TIMEOUT_FULL)
+ after_time = time.time()
+ self.assert_(after_time - before_time < self.TIMEOUT_HALF)
+
+ def setUp(self):
+ import fcntl
+
+ self.alrm = signal.signal(signal.SIGALRM, lambda x,y:None)
+ self.read, self.write = os.pipe()
+ flags = fcntl.fcntl(self.write, fcntl.F_GETFL, 0)
+ flags = flags | os.O_NONBLOCK
+ fcntl.fcntl(self.write, fcntl.F_SETFL, flags)
+ self.old_wakeup = signal.set_wakeup_fd(self.write)
+
+ def tearDown(self):
+ signal.set_wakeup_fd(self.old_wakeup)
+ os.close(self.read)
+ os.close(self.write)
+ signal.signal(signal.SIGALRM, self.alrm)
+
+
def test_main():
if sys.platform[:3] in ('win', 'os2') or sys.platform == 'riscos':
raise test_support.TestSkipped("Can't test signal on %s" % \
sys.platform)
- test_support.run_unittest(BasicSignalTests, InterProcessSignalTests)
+ test_support.run_unittest(BasicSignalTests, InterProcessSignalTests,
+ WakeupSignalTests)
if __name__ == "__main__":
#include <signal.h>
+#include <sys/stat.h>
+
#ifndef SIG_ERR
#define SIG_ERR ((PyOS_sighandler_t)(-1))
#endif
PyObject *func;
} Handlers[NSIG];
+static sig_atomic_t wakeup_fd = -1;
+
/* Speed up sigcheck() when none tripped */
static volatile sig_atomic_t is_tripped = 0;
cleared in PyErr_CheckSignals() before .tripped. */
is_tripped = 1;
Py_AddPendingCall(checksignals_witharg, NULL);
+ if (wakeup_fd != -1)
+ write(wakeup_fd, "\0", 1);
#ifdef WITH_THREAD
}
#endif
anything else -- the callable Python object used as a handler");
+static PyObject *
+signal_set_wakeup_fd(PyObject *self, PyObject *args)
+{
+ struct stat buf;
+ int fd, old_fd;
+ if (!PyArg_ParseTuple(args, "i:set_wakeup_fd", &fd))
+ return NULL;
+#ifdef WITH_THREAD
+ if (PyThread_get_thread_ident() != main_thread) {
+ PyErr_SetString(PyExc_ValueError,
+ "set_wakeup_fd only works in main thread");
+ return NULL;
+ }
+#endif
+ if (fd != -1 && fstat(fd, &buf) != 0) {
+ PyErr_SetString(PyExc_ValueError, "invalid fd");
+ return NULL;
+ }
+ old_fd = wakeup_fd;
+ wakeup_fd = fd;
+ return PyLong_FromLong(old_fd);
+}
+
+PyDoc_STRVAR(set_wakeup_fd_doc,
+"set_wakeup_fd(fd) -> fd\n\
+\n\
+Sets the fd to be written to (with '\\0') when a signal\n\
+comes in. A library can use this to wakeup select or poll.\n\
+The previous fd is returned.\n\
+\n\
+The fd must be non-blocking.");
+
+/* C API for the same, without all the error checking */
+int
+PySignal_SetWakeupFd(int fd)
+{
+ int old_fd = wakeup_fd;
+ if (fd < 0)
+ fd = -1;
+ wakeup_fd = fd;
+ return old_fd;
+}
+
+
/* List of functions defined in the module */
static PyMethodDef signal_methods[] = {
#ifdef HAVE_ALARM
#endif
{"signal", signal_signal, METH_VARARGS, signal_doc},
{"getsignal", signal_getsignal, METH_VARARGS, getsignal_doc},
+ {"set_wakeup_fd", signal_set_wakeup_fd, METH_VARARGS, set_wakeup_fd_doc},
#ifdef HAVE_PAUSE
{"pause", (PyCFunction)signal_pause,
METH_NOARGS,pause_doc},
#endif
- {"default_int_handler", signal_default_int_handler,
+ {"default_int_handler", signal_default_int_handler,
METH_VARARGS, default_int_handler_doc},
{NULL, NULL} /* sentinel */
};
projects / python / commitdiff