ยปCore Development>Code coverage>Lib/test/test_thread.py

Python code coverage for Lib/test/test_thread.py

#countcontent
1n/aimport os
2n/aimport unittest
3n/aimport random
4n/afrom test import support
5n/athread = support.import_module('_thread')
6n/aimport time
7n/aimport sys
8n/aimport weakref
9n/a
10n/afrom test import lock_tests
11n/a
12n/aNUMTASKS = 10
13n/aNUMTRIPS = 3
14n/a
15n/a_print_mutex = thread.allocate_lock()
16n/a
17n/adef verbose_print(arg):
18n/a """Helper function for printing out debugging output."""
19n/a if support.verbose:
20n/a with _print_mutex:
21n/a print(arg)
22n/a
23n/aclass BasicThreadTest(unittest.TestCase):
24n/a
25n/a def setUp(self):
26n/a self.done_mutex = thread.allocate_lock()
27n/a self.done_mutex.acquire()
28n/a self.running_mutex = thread.allocate_lock()
29n/a self.random_mutex = thread.allocate_lock()
30n/a self.created = 0
31n/a self.running = 0
32n/a self.next_ident = 0
33n/a
34n/a
35n/aclass ThreadRunningTests(BasicThreadTest):
36n/a
37n/a def newtask(self):
38n/a with self.running_mutex:
39n/a self.next_ident += 1
40n/a verbose_print("creating task %s" % self.next_ident)
41n/a thread.start_new_thread(self.task, (self.next_ident,))
42n/a self.created += 1
43n/a self.running += 1
44n/a
45n/a def task(self, ident):
46n/a with self.random_mutex:
47n/a delay = random.random() / 10000.0
48n/a verbose_print("task %s will run for %sus" % (ident, round(delay*1e6)))
49n/a time.sleep(delay)
50n/a verbose_print("task %s done" % ident)
51n/a with self.running_mutex:
52n/a self.running -= 1
53n/a if self.created == NUMTASKS and self.running == 0:
54n/a self.done_mutex.release()
55n/a
56n/a def test_starting_threads(self):
57n/a # Basic test for thread creation.
58n/a for i in range(NUMTASKS):
59n/a self.newtask()
60n/a verbose_print("waiting for tasks to complete...")
61n/a self.done_mutex.acquire()
62n/a verbose_print("all tasks done")
63n/a
64n/a def test_stack_size(self):
65n/a # Various stack size tests.
66n/a self.assertEqual(thread.stack_size(), 0, "initial stack size is not 0")
67n/a
68n/a thread.stack_size(0)
69n/a self.assertEqual(thread.stack_size(), 0, "stack_size not reset to default")
70n/a
71n/a @unittest.skipIf(os.name not in ("nt", "posix"), 'test meant for nt and posix')
72n/a def test_nt_and_posix_stack_size(self):
73n/a try:
74n/a thread.stack_size(4096)
75n/a except ValueError:
76n/a verbose_print("caught expected ValueError setting "
77n/a "stack_size(4096)")
78n/a except thread.error:
79n/a self.skipTest("platform does not support changing thread stack "
80n/a "size")
81n/a
82n/a fail_msg = "stack_size(%d) failed - should succeed"
83n/a for tss in (262144, 0x100000, 0):
84n/a thread.stack_size(tss)
85n/a self.assertEqual(thread.stack_size(), tss, fail_msg % tss)
86n/a verbose_print("successfully set stack_size(%d)" % tss)
87n/a
88n/a for tss in (262144, 0x100000):
89n/a verbose_print("trying stack_size = (%d)" % tss)
90n/a self.next_ident = 0
91n/a self.created = 0
92n/a for i in range(NUMTASKS):
93n/a self.newtask()
94n/a
95n/a verbose_print("waiting for all tasks to complete")
96n/a self.done_mutex.acquire()
97n/a verbose_print("all tasks done")
98n/a
99n/a thread.stack_size(0)
100n/a
101n/a def test__count(self):
102n/a # Test the _count() function.
103n/a orig = thread._count()
104n/a mut = thread.allocate_lock()
105n/a mut.acquire()
106n/a started = []
107n/a def task():
108n/a started.append(None)
109n/a mut.acquire()
110n/a mut.release()
111n/a thread.start_new_thread(task, ())
112n/a while not started:
113n/a time.sleep(0.01)
114n/a self.assertEqual(thread._count(), orig + 1)
115n/a # Allow the task to finish.
116n/a mut.release()
117n/a # The only reliable way to be sure that the thread ended from the
118n/a # interpreter's point of view is to wait for the function object to be
119n/a # destroyed.
120n/a done = []
121n/a wr = weakref.ref(task, lambda _: done.append(None))
122n/a del task
123n/a while not done:
124n/a time.sleep(0.01)
125n/a self.assertEqual(thread._count(), orig)
126n/a
127n/a def test_save_exception_state_on_error(self):
128n/a # See issue #14474
129n/a def task():
130n/a started.release()
131n/a raise SyntaxError
132n/a def mywrite(self, *args):
133n/a try:
134n/a raise ValueError
135n/a except ValueError:
136n/a pass
137n/a real_write(self, *args)
138n/a c = thread._count()
139n/a started = thread.allocate_lock()
140n/a with support.captured_output("stderr") as stderr:
141n/a real_write = stderr.write
142n/a stderr.write = mywrite
143n/a started.acquire()
144n/a thread.start_new_thread(task, ())
145n/a started.acquire()
146n/a while thread._count() > c:
147n/a time.sleep(0.01)
148n/a self.assertIn("Traceback", stderr.getvalue())
149n/a
150n/a
151n/aclass Barrier:
152n/a def __init__(self, num_threads):
153n/a self.num_threads = num_threads
154n/a self.waiting = 0
155n/a self.checkin_mutex = thread.allocate_lock()
156n/a self.checkout_mutex = thread.allocate_lock()
157n/a self.checkout_mutex.acquire()
158n/a
159n/a def enter(self):
160n/a self.checkin_mutex.acquire()
161n/a self.waiting = self.waiting + 1
162n/a if self.waiting == self.num_threads:
163n/a self.waiting = self.num_threads - 1
164n/a self.checkout_mutex.release()
165n/a return
166n/a self.checkin_mutex.release()
167n/a
168n/a self.checkout_mutex.acquire()
169n/a self.waiting = self.waiting - 1
170n/a if self.waiting == 0:
171n/a self.checkin_mutex.release()
172n/a return
173n/a self.checkout_mutex.release()
174n/a
175n/a
176n/aclass BarrierTest(BasicThreadTest):
177n/a
178n/a def test_barrier(self):
179n/a self.bar = Barrier(NUMTASKS)
180n/a self.running = NUMTASKS
181n/a for i in range(NUMTASKS):
182n/a thread.start_new_thread(self.task2, (i,))
183n/a verbose_print("waiting for tasks to end")
184n/a self.done_mutex.acquire()
185n/a verbose_print("tasks done")
186n/a
187n/a def task2(self, ident):
188n/a for i in range(NUMTRIPS):
189n/a if ident == 0:
190n/a # give it a good chance to enter the next
191n/a # barrier before the others are all out
192n/a # of the current one
193n/a delay = 0
194n/a else:
195n/a with self.random_mutex:
196n/a delay = random.random() / 10000.0
197n/a verbose_print("task %s will run for %sus" %
198n/a (ident, round(delay * 1e6)))
199n/a time.sleep(delay)
200n/a verbose_print("task %s entering %s" % (ident, i))
201n/a self.bar.enter()
202n/a verbose_print("task %s leaving barrier" % ident)
203n/a with self.running_mutex:
204n/a self.running -= 1
205n/a # Must release mutex before releasing done, else the main thread can
206n/a # exit and set mutex to None as part of global teardown; then
207n/a # mutex.release() raises AttributeError.
208n/a finished = self.running == 0
209n/a if finished:
210n/a self.done_mutex.release()
211n/a
212n/aclass LockTests(lock_tests.LockTests):
213n/a locktype = thread.allocate_lock
214n/a
215n/a
216n/aclass TestForkInThread(unittest.TestCase):
217n/a def setUp(self):
218n/a self.read_fd, self.write_fd = os.pipe()
219n/a
220n/a @unittest.skipIf(sys.platform.startswith('win'),
221n/a "This test is only appropriate for POSIX-like systems.")
222n/a @support.reap_threads
223n/a def test_forkinthread(self):
224n/a def thread1():
225n/a try:
226n/a pid = os.fork() # fork in a thread
227n/a except RuntimeError:
228n/a os._exit(1) # exit the child
229n/a
230n/a if pid == 0: # child
231n/a try:
232n/a os.close(self.read_fd)
233n/a os.write(self.write_fd, b"OK")
234n/a finally:
235n/a os._exit(0)
236n/a else: # parent
237n/a os.close(self.write_fd)
238n/a
239n/a thread.start_new_thread(thread1, ())
240n/a self.assertEqual(os.read(self.read_fd, 2), b"OK",
241n/a "Unable to fork() in thread")
242n/a
243n/a def tearDown(self):
244n/a try:
245n/a os.close(self.read_fd)
246n/a except OSError:
247n/a pass
248n/a
249n/a try:
250n/a os.close(self.write_fd)
251n/a except OSError:
252n/a pass
253n/a
254n/a
255n/aif __name__ == "__main__":
256n/a unittest.main()