1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
|
import unittest
from test.support import threading_helper
import cProfile
import pstats
NTHREADS = 10
INSERT_PER_THREAD = 1000
@threading_helper.requires_working_threading()
class TestCProfile(unittest.TestCase):
def test_cprofile_racing_list_insert(self):
def list_insert(lst):
for i in range(INSERT_PER_THREAD):
lst.insert(0, i)
lst = []
with cProfile.Profile() as pr:
threading_helper.run_concurrently(
worker_func=list_insert, nthreads=NTHREADS, args=(lst,)
)
pr.create_stats()
ps = pstats.Stats(pr)
stats_profile = ps.get_stats_profile()
list_insert_profile = stats_profile.func_profiles[
"<method 'insert' of 'list' objects>"
]
# Even though there is no explicit recursive call to insert,
# cProfile may record some calls as recursive due to limitations
# in its handling of multithreaded programs. This issue is not
# directly related to FT Python itself; however, it tends to be
# more noticeable when using FT Python. Therefore, consider only
# the calls section and disregard the recursive part.
list_insert_ncalls = list_insert_profile.ncalls.split("/")[0]
self.assertEqual(
int(list_insert_ncalls), NTHREADS * INSERT_PER_THREAD
)
self.assertEqual(len(lst), NTHREADS * INSERT_PER_THREAD)
|
