Skip to content
Navigation Menu
{{ message }}
forked from apache/cassandra-python-driver
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_concurrent.py
More file actions
260 lines (218 loc) · 10.4 KB
/
Copy pathtest_concurrent.py
File metadata and controls
260 lines (218 loc) · 10.4 KB
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
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
# Copyright DataStax, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
try:
import unittest2 as unittest
except ImportError:
import unittest # noqa
from itertools import cycle
from mock import Mock
import time
import threading
from six.moves.queue import PriorityQueue
import sys
import platform
from cassandra.cluster import Cluster, Session
from cassandra.concurrent import execute_concurrent, execute_concurrent_with_args
from cassandra.pool import Host
from cassandra.policies import SimpleConvictionPolicy
from tests.unit.utils import mock_session_pools
class MockResponseResponseFuture():
"""
This is a mock ResponseFuture. It is used to allow us to hook into the underlying session
and invoke callback with various timing.
"""
_query_trace = None
_col_names = None
_col_types = None
# a list pending callbacks, these will be prioritized in reverse or normal orderd
pending_callbacks = PriorityQueue()
def __init__(self, reverse):
# if this is true invoke callback in the reverse order then what they were insert
self.reverse = reverse
# hardcoded to avoid paging logic
self.has_more_pages = False
if(reverse):
self.priority = 100
else:
self.priority = 0
def add_callback(self, fn, *args, **kwargs):
"""
This is used to add a callback our pending list of callbacks.
If reverse is specified we will invoke the callback in the opposite order that we added it
"""
time_added = time.time()
self.pending_callbacks.put((self.priority, (fn, args, kwargs, time_added)))
if not reversed:
self.priority += 1
else:
self.priority -= 1
def add_callbacks(self, callback, errback,
callback_args=(), callback_kwargs=None,
errback_args=(), errback_kwargs=None):
self.add_callback(callback, *callback_args, **(callback_kwargs or {}))
def get_next_callback(self):
return self.pending_callbacks.get()
def has_next_callback(self):
return not self.pending_callbacks.empty()
def has_more_pages(self):
return False
def clear_callbacks(self):
return
class TimedCallableInvoker(threading.Thread):
"""
This is a local thread which is runs and invokes all the callbacks on the pending callback queue.
The slowdown flag can used to invoke random slowdowns in our simulate queries.
"""
def __init__(self, handler, slowdown=False):
super(TimedCallableInvoker, self).__init__()
self.slowdown = slowdown
self._stopper = threading.Event()
self.handler = handler
def stop(self):
self._stopper.set()
def stopped(self):
return self._stopper.isSet()
def run(self):
while(not self.stopped()):
if(self.handler.has_next_callback()):
pending_callback = self.handler.get_next_callback()
priority_num = pending_callback[0]
if (priority_num % 10) == 0 and self.slowdown:
self._stopper.wait(.1)
callback_args = pending_callback[1]
fn, args, kwargs, time_added = callback_args
fn([time_added], *args, **kwargs)
self._stopper.wait(.001)
return
class ConcurrencyTest((unittest.TestCase)):
def test_results_ordering_forward(self):
"""
This tests the ordering of our various concurrent generator class ConcurrentExecutorListResults
when queries complete in the order they were executed.
"""
self.insert_and_validate_list_results(False, False)
def test_results_ordering_reverse(self):
"""
This tests the ordering of our various concurrent generator class ConcurrentExecutorListResults
when queries complete in the reverse order they were executed.
"""
self.insert_and_validate_list_results(True, False)
def test_results_ordering_forward_slowdown(self):
"""
This tests the ordering of our various concurrent generator class ConcurrentExecutorListResults
when queries complete in the order they were executed, with slow queries mixed in.
"""
self.insert_and_validate_list_results(False, True)
def test_results_ordering_reverse_slowdown(self):
"""
This tests the ordering of our various concurrent generator class ConcurrentExecutorListResults
when queries complete in the reverse order they were executed, with slow queries mixed in.
"""
self.insert_and_validate_list_results(True, True)
def test_results_ordering_forward_generator(self):
"""
This tests the ordering of our various concurrent generator class ConcurrentExecutorGenResults
when queries complete in the order they were executed.
"""
self.insert_and_validate_list_generator(False, False)
def test_results_ordering_reverse_generator(self):
"""
This tests the ordering of our various concurrent generator class ConcurrentExecutorGenResults
when queries complete in the reverse order they were executed.
"""
self.insert_and_validate_list_generator(True, False)
def test_results_ordering_forward_generator_slowdown(self):
"""
This tests the ordering of our various concurrent generator class ConcurrentExecutorGenResults
when queries complete in the order they were executed, with slow queries mixed in.
"""
self.insert_and_validate_list_generator(False, True)
def test_results_ordering_reverse_generator_slowdown(self):
"""
This tests the ordering of our various concurrent generator class ConcurrentExecutorGenResults
when queries complete in the reverse order they were executed, with slow queries mixed in.
"""
self.insert_and_validate_list_generator(True, True)
def insert_and_validate_list_results(self, reverse, slowdown):
"""
This utility method will execute submit various statements for execution using the ConcurrentExecutorListResults,
then invoke a separate thread to execute the callback associated with the futures registered
for those statements. The parameters will toggle various timing, and ordering changes.
Finally it will validate that the results were returned in the order they were submitted
:param reverse: Execute the callbacks in the opposite order that they were submitted
:param slowdown: Cause intermittent queries to perform slowly
"""
our_handler = MockResponseResponseFuture(reverse=reverse)
mock_session = Mock()
statements_and_params = zip(cycle(["INSERT INTO test3rf.test (k, v) VALUES (%s, 0)"]),
[(i, ) for i in range(100)])
mock_session.execute_async.return_value = our_handler
t = TimedCallableInvoker(our_handler, slowdown=slowdown)
t.start()
results = execute_concurrent(mock_session, statements_and_params)
while(not our_handler.pending_callbacks.empty()):
time.sleep(.01)
t.stop()
self.validate_result_ordering(results)
def insert_and_validate_list_generator(self, reverse, slowdown):
"""
This utility method will execute submit various statements for execution using the ConcurrentExecutorGenResults,
then invoke a separate thread to execute the callback associated with the futures registered
for those statements. The parameters will toggle various timing, and ordering changes.
Finally it will validate that the results were returned in the order they were submitted
:param reverse: Execute the callbacks in the opposite order that they were submitted
:param slowdown: Cause intermittent queries to perform slowly
"""
our_handler = MockResponseResponseFuture(reverse=reverse)
mock_session = Mock()
statements_and_params = zip(cycle(["INSERT INTO test3rf.test (k, v) VALUES (%s, 0)"]),
[(i, ) for i in range(100)])
mock_session.execute_async.return_value = our_handler
t = TimedCallableInvoker(our_handler, slowdown=slowdown)
t.start()
try:
results = execute_concurrent(mock_session, statements_and_params, results_generator=True)
self.validate_result_ordering(results)
finally:
t.stop()
def validate_result_ordering(self, results):
"""
This method will validate that the timestamps returned from the result are in order. This indicates that the
results were returned in the order they were submitted for execution
:param results:
"""
last_time_added = 0
for success, result in results:
self.assertTrue(success)
current_time_added = list(result)[0]
#Windows clock granularity makes this equal most of the times
if "Windows" in platform.system():
self.assertLessEqual(last_time_added, current_time_added)
else:
self.assertLess(last_time_added, current_time_added)
last_time_added = current_time_added
@mock_session_pools
def test_recursion_limited(self):
"""
Verify that recursion is controlled when raise_on_first_error=False and something is wrong with the query.
PYTHON-585
"""
max_recursion = sys.getrecursionlimit()
s = Session(Cluster(), [Host("127.0.0.1", SimpleConvictionPolicy)])
self.assertRaises(TypeError, execute_concurrent_with_args, s, "doesn't matter", [('param',)] * max_recursion, raise_on_first_error=True)
results = execute_concurrent_with_args(s, "doesn't matter", [('param',)] * max_recursion, raise_on_first_error=False) # previously
self.assertEqual(len(results), max_recursion)
for r in results:
self.assertFalse(r[0])
self.assertIsInstance(r[1], TypeError)
You can’t perform that action at this time.