Charles-François Natali <neolo...@free.fr> added the comment:
Patch attached:
- a new process is spawned (using assert_python_ok()) to avoid
undefined behavior (and crash on FreeBSD)
- I've kept SIGUSR1 default handler, because, as noted in
http://bugs.python.org/issue8407#msg138066 , there can be subtle
differences between default handlers and user-installed ones which can
mask bugs
- I've fixed a comment in test_sigwait
----------
keywords: +patch
Added file: http://bugs.python.org/file22341/test_sigwait_thread.diff
_______________________________________
Python tracker <rep...@bugs.python.org>
<http://bugs.python.org/issue12316>
_______________________________________
diff -r 0e22c47b47a3 Lib/test/test_signal.py
--- a/Lib/test/test_signal.py Sun Jun 12 23:02:57 2011 +0200
+++ b/Lib/test/test_signal.py Mon Jun 13 15:53:57 2011 +0200
@@ -9,6 +9,7 @@
import subprocess
import traceback
import sys, os, time, errno
+from test.script_helper import assert_python_ok
try:
import threading
except ImportError:
@@ -627,8 +628,7 @@
else:
os._exit(0)
else:
- # parent: let the child some time to wait, send him the signal, and
- # check it correcty received it
+ # parent: check that the child correcty received the signal
self.assertEqual(os.waitpid(pid, 0), (pid, 0))
@unittest.skipUnless(hasattr(signal, 'sigwait'),
@@ -649,24 +649,36 @@
@unittest.skipUnless(hasattr(signal, 'sigwait'),
'need signal.sigwait()')
@unittest.skipIf(threading is None, "test needs threading module")
- @unittest.skipUnless(hasattr(os, 'fork'), 'need os.fork()')
def test_sigwait_thread(self):
- def kill_later(signum):
- # wait until the main thread is waiting in sigwait()
- time.sleep(1)
- os.kill(os.getpid(), signum)
+ # Check that calling sigwait() from a thread doesn't suspend the whole
+ # process.
+ # A new interpreter is spawned to avoid problems when mixing threads
and
+ # fork() (only async-safe functions are allowed between fork() and
+ # exec()).
+ assert_python_ok("-c", """if True:
+ import os, threading, sys, time, signal
- def test(signum):
- killer = threading.Thread(target=kill_later, args=(signum,))
+ # the default handler terminates the process
+ signum = signal.SIGUSR1
+
+ def kill_later():
+ # wait until the main thread is waiting in sigwait()
+ time.sleep(1)
+ os.kill(os.getpid(), signum)
+
+ # the signal must be blocked by all the threads
+ signal.pthread_sigmask(signal.SIG_BLOCK, [signum])
+ killer = threading.Thread(target=kill_later)
killer.start()
received = signal.sigwait([signum])
if received != signum:
print("sigwait() received %s, not %s" % (received, signum),
file=sys.stderr)
- os._exit(1)
+ sys.exit(1)
killer.join()
-
- self.check_sigwait(test, signal.SIGUSR1)
+ # unblock the signal, which should have been cleared by sigwait()
+ signal.pthread_sigmask(signal.SIG_UNBLOCK, [signum])
+ """)
@unittest.skipUnless(hasattr(signal, 'pthread_sigmask'),
'need signal.pthread_sigmask()')
_______________________________________________
Python-bugs-list mailing list
Unsubscribe:
http://mail.python.org/mailman/options/python-bugs-list/archive%40mail-archive.com
