blob: d0ea8709e9704e65ff07d8f7e39c90919f36919c (
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
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
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
|
import threading
from kazoo.testing import KazooTestCase
class KazooBarrierTests(KazooTestCase):
def test_barrier_not_exist(self):
b = self.client.Barrier("/some/path")
assert b.wait()
def test_barrier_exists(self):
b = self.client.Barrier("/some/path")
b.create()
assert not b.wait(0)
b.remove()
assert b.wait()
def test_remove_nonexistent_barrier(self):
b = self.client.Barrier("/some/path")
assert not b.remove()
class KazooDoubleBarrierTests(KazooTestCase):
def test_basic_barrier(self):
b = self.client.DoubleBarrier("/some/path", 1)
assert not b.participating
b.enter()
assert b.participating
b.leave()
assert not b.participating
def test_two_barrier(self):
av = threading.Event()
ev = threading.Event()
bv = threading.Event()
release_all = threading.Event()
b1 = self.client.DoubleBarrier("/some/path", 2)
b2 = self.client.DoubleBarrier("/some/path", 2)
def make_barrier_one():
b1.enter()
ev.set()
release_all.wait()
b1.leave()
ev.set()
def make_barrier_two():
bv.wait()
b2.enter()
av.set()
release_all.wait()
b2.leave()
av.set()
# Spin up both of them
t1 = threading.Thread(target=make_barrier_one)
t1.start()
t2 = threading.Thread(target=make_barrier_two)
t2.start()
assert not b1.participating
assert not b2.participating
bv.set()
av.wait()
ev.wait()
assert b1.participating
assert b2.participating
av.clear()
ev.clear()
release_all.set()
av.wait()
ev.wait()
assert not b1.participating
assert not b2.participating
t1.join()
t2.join()
def test_three_barrier(self):
av = threading.Event()
ev = threading.Event()
bv = threading.Event()
release_all = threading.Event()
b1 = self.client.DoubleBarrier("/some/path", 3)
b2 = self.client.DoubleBarrier("/some/path", 3)
b3 = self.client.DoubleBarrier("/some/path", 3)
def make_barrier_one():
b1.enter()
ev.set()
release_all.wait()
b1.leave()
ev.set()
def make_barrier_two():
bv.wait()
b2.enter()
av.set()
release_all.wait()
b2.leave()
av.set()
# Spin up both of them
t1 = threading.Thread(target=make_barrier_one)
t1.start()
t2 = threading.Thread(target=make_barrier_two)
t2.start()
assert not b1.participating
assert not b2.participating
bv.set()
assert not b1.participating
assert not b2.participating
b3.enter()
ev.wait()
av.wait()
assert b1.participating
assert b2.participating
assert b3.participating
av.clear()
ev.clear()
release_all.set()
b3.leave()
av.wait()
ev.wait()
assert not b1.participating
assert not b2.participating
assert not b3.participating
t1.join()
t2.join()
def test_barrier_existing_parent_node(self):
b = self.client.DoubleBarrier('/some/path', 1)
assert b.participating is False
self.client.create('/some', ephemeral=True)
# the barrier cannot create children under an ephemeral node
b.enter()
assert b.participating is False
def test_barrier_existing_node(self):
b = self.client.DoubleBarrier('/some', 1)
assert b.participating is False
self.client.ensure_path(b.path)
self.client.create(b.create_path, ephemeral=True)
# the barrier will re-use an existing node
b.enter()
assert b.participating is True
b.leave()
|
