[mirror_edk2.git] / AppPkg / Applications / Python / Python-2.7.2 / Lib / test / test_thread.py

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