diff options
| -rw-r--r-- | Lib/test/test_signal.py | 65 |
1 files changed, 50 insertions, 15 deletions
diff --git a/Lib/test/test_signal.py b/Lib/test/test_signal.py index fc7725a..0ddfe36 100644 --- a/Lib/test/test_signal.py +++ b/Lib/test/test_signal.py @@ -2,6 +2,7 @@ import os import random import signal import socket +import statistics import subprocess import sys import time @@ -949,13 +950,55 @@ class StressTest(unittest.TestCase): previously tripped signal handlers. """ + def setsig(self, signum, handler): + old_handler = signal.signal(signum, handler) + self.addCleanup(signal.signal, signum, old_handler) + + def measure_itimer_resolution(self): + N = 20 + times = [] + + def handler(signum=None, frame=None): + if len(times) < N: + times.append(time.perf_counter()) + # 1 µs is the smallest possible timer interval, + # we want to measure what the concrete duration + # will be on this platform + signal.setitimer(signal.ITIMER_REAL, 1e-6) + + self.addCleanup(signal.setitimer, signal.ITIMER_REAL, 0) + self.setsig(signal.SIGALRM, handler) + handler() + while len(times) < N: + time.sleep(1e-3) + + durations = [times[i+1] - times[i] for i in range(len(times) - 1)] + med = statistics.median(durations) + if support.verbose: + print("detected median itimer() resolution: %.6f s." % (med,)) + return med + + def decide_itimer_count(self): + # Some systems have poor setitimer() resolution (for example + # measured around 20 ms. on FreeBSD 9), so decide on a reasonable + # number of sequential timers based on that. + reso = self.measure_itimer_resolution() + if reso <= 1e-4: + return 10000 + elif reso <= 1e-2: + return 100 + else: + self.skipTest("detected itimer resolution (%.3f s.) too high " + "(> 10 ms.) on this platform (or system too busy)" + % (reso,)) + @unittest.skipUnless(hasattr(signal, "setitimer"), "test needs setitimer()") def test_stress_delivery_dependent(self): """ This test uses dependent signal handlers. """ - N = 10000 + N = self.decide_itimer_count() sigs = [] def first_handler(signum, frame): @@ -969,16 +1012,12 @@ class StressTest(unittest.TestCase): def second_handler(signum=None, frame=None): sigs.append(signum) - def setsig(signum, handler): - old_handler = signal.signal(signum, handler) - self.addCleanup(signal.signal, signum, old_handler) - # Here on Linux, SIGPROF > SIGALRM > SIGUSR1. By using both # ascending and descending sequences (SIGUSR1 then SIGALRM, # SIGPROF then SIGALRM), we maximize chances of hitting a bug. - setsig(signal.SIGPROF, first_handler) - setsig(signal.SIGUSR1, first_handler) - setsig(signal.SIGALRM, second_handler) # for ITIMER_REAL + self.setsig(signal.SIGPROF, first_handler) + self.setsig(signal.SIGUSR1, first_handler) + self.setsig(signal.SIGALRM, second_handler) # for ITIMER_REAL expected_sigs = 0 deadline = time.time() + 15.0 @@ -1005,18 +1044,14 @@ class StressTest(unittest.TestCase): """ This test uses simultaneous signal handlers. """ - N = 10000 + N = self.decide_itimer_count() sigs = [] def handler(signum, frame): sigs.append(signum) - def setsig(signum, handler): - old_handler = signal.signal(signum, handler) - self.addCleanup(signal.signal, signum, old_handler) - - setsig(signal.SIGUSR1, handler) - setsig(signal.SIGALRM, handler) # for ITIMER_REAL + self.setsig(signal.SIGUSR1, handler) + self.setsig(signal.SIGALRM, handler) # for ITIMER_REAL expected_sigs = 0 deadline = time.time() + 15.0 |