blob: 1526ca3c23bdecb21a8112eea15350e4417be05c (
plain)
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
44
45
46
47
|
#!/usr/bin/python
#
# A small test script to excercise the bgjobs stuff. We create a number of
# jobs, and make run them, and make sure the results are correct.
#
# By running this script a large number of times (say, 10 000) in a loop,
# it may be possible to expose race conditions. (As of the time of committing,
# no such are known.)
#
# If you want to run this in the source directory, do this:
#
# PYTHONPATH=. python scripts/test-bgjobs
import Queue
import time
import dimbola
N = 100000
class SlowJob(dimbola.BackgroundJob):
def __init__(self, value):
self.value = value
def run(self):
return self.value
manager = dimbola.BackgroundManager()
for i in range(N):
manager.add_job(SlowJob(i))
manager.start_jobs(maxproc=None)
results = 0
while manager.running:
try:
result = manager.results.get(block=False)
except Queue.Empty:
pass
else:
results += 1
manager.stop_jobs()
assert results == N, "results==%d (not %d)" % (results, N)
|
