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
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
|
"""Stress test diskcache.persistent.Index."""
import itertools as it
import multiprocessing as mp
import random
import time
import diskcache as dc
KEYS = 100
OPERATIONS = 10000
SEED = 0
functions = []
def register(function):
functions.append(function)
return function
@register
def stress_get(index):
key = random.randrange(KEYS)
index.get(key, None)
@register
def stress_set(index):
key = random.randrange(KEYS)
value = random.random()
index[key] = value
register(stress_set)
register(stress_set)
register(stress_set)
@register
def stress_del(index):
key = random.randrange(KEYS)
try:
del index[key]
except KeyError:
pass
@register
def stress_pop(index):
key = random.randrange(KEYS)
index.pop(key, None)
@register
def stress_popitem(index):
try:
if random.randrange(2):
index.popitem()
else:
index.popitem(last=False)
except KeyError:
pass
@register
def stress_iter(index):
iterator = it.islice(index, 5)
for key in iterator:
pass
@register
def stress_reversed(index):
iterator = it.islice(reversed(index), 5)
for key in iterator:
pass
@register
def stress_len(index):
len(index)
def stress(seed, index):
random.seed(seed)
for count in range(OPERATIONS):
function = random.choice(functions)
function(index)
def test(status=False):
random.seed(SEED)
index = dc.Index(enumerate(range(KEYS)))
processes = []
for count in range(8):
process = mp.Process(target=stress, args=(SEED + count, index))
process.start()
processes.append(process)
for value in it.count():
time.sleep(1)
if status:
print('\r', value, 's', len(index), 'keys', ' ' * 20, end='')
if all(not process.is_alive() for process in processes):
break
if status:
print('')
assert all(process.exitcode == 0 for process in processes)
if __name__ == '__main__':
test(status=True)
|
