Skip to content

Commit 774afb2

Browse files
aboudreaultaboudreault
authored
Merge pull request apache#1085 from datastax/python-1248
PYTHON-1248: tcp flow control for libevreactor
2 parents 3317f42 + bbdefc7 commit 774afb2

10 files changed

Lines changed: 377 additions & 100 deletions

File tree

‎CHANGELOG.rst‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ Features
66
--------
77
* Make geomet an optional dependency at runtime (PYTHON-1237)
88
* Add use_default_tempdir cloud config options (PYTHON-1245)
9+
* Tcp flow control for libevreactor (PYTHON-1248)
910

1011
Bug Fixes
1112
---------

‎Jenkinsfile‎

Lines changed: 44 additions & 38 deletions
Large diffs are not rendered by default.

‎cassandra/cluster.py‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -48,7 +48,7 @@
4848
from cassandra.connection import (ConnectionException, ConnectionShutdown,
4949
ConnectionHeartbeat, ProtocolVersionUnsupported,
5050
EndPoint, DefaultEndPoint, DefaultEndPointFactory,
51-
ContinuousPagingState, SniEndPointFactory)
51+
ContinuousPagingState, SniEndPointFactory, ConnectionBusy)
5252
from cassandra.cqltypes import UserType
5353
from cassandra.encoder import Encoder
5454
from cassandra.protocol import (QueryMessage, ResultMessage,
@@ -4445,15 +4445,18 @@ def _query(self, host, message=None, cb=None):
44454445
except NoConnectionsAvailable as exc:
44464446
log.debug("All connections for host %s are at capacity, moving to the next host", host)
44474447
self._errors[host] = exc
4448-
return None
4448+
except ConnectionBusy as exc:
4449+
log.debug("Connection for host %s is busy, moving to the next host", host)
4450+
self._errors[host] = exc
44494451
except Exception as exc:
44504452
log.debug("Error querying host %s", host, exc_info=True)
44514453
self._errors[host] = exc
44524454
if self._metrics is not None:
44534455
self._metrics.on_connection_error()
44544456
if connection:
44554457
pool.return_connection(connection)
4456-
return None
4458+
4459+
return None
44574460

44584461
@property
44594462
def has_more_pages(self):

‎cassandra/connection.py‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -692,6 +692,7 @@ def __init__(self, host='127.0.0.1', port=9042, authenticator=None,
692692
self._requests = {}
693693
self._iobuf = io.BytesIO()
694694
self._continuous_paging_sessions = {}
695+
self._socket_writable = True
695696

696697
if ssl_options:
697698
self._check_hostname = bool(self.ssl_options.pop('check_hostname', False))
@@ -926,6 +927,8 @@ def send_msg(self, msg, request_id, cb, encoder=ProtocolHandler.encode_message,
926927
raise ConnectionShutdown("Connection to %s is defunct" % self.endpoint)
927928
elif self.is_closed:
928929
raise ConnectionShutdown("Connection to %s is closed" % self.endpoint)
930+
elif not self._socket_writable:
931+
raise ConnectionBusy("Connection %s is overloaded" % self.endpoint)
929932

930933
# queue the decoder function with the request
931934
# this allows us to inject custom functions per request to encode, decode messages
@@ -1443,7 +1446,7 @@ def __init__(self, connection, owner):
14431446
log.debug("Sending options message heartbeat on idle connection (%s) %s",
14441447
id(connection), connection.endpoint)
14451448
with connection.lock:
1446-
if connection.in_flight <= connection.max_request_id:
1449+
if connection.in_flight < connection.max_request_id:
14471450
connection.in_flight += 1
14481451
connection.send_msg(OptionsMessage(), connection.get_request_id(), self._options_callback)
14491452
else:

‎cassandra/io/libevreactor.py‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -310,13 +310,17 @@ def handle_write(self, watcher, revents, errno=None):
310310
with self._deque_lock:
311311
next_msg = self.deque.popleft()
312312
except IndexError:
313+
if not self._socket_writable:
314+
self._socket_writable = True
313315
return
314316

315317
try:
316318
sent = self._socket.send(next_msg)
317319
except socket.error as err:
318320
if (err.args[0] in NONBLOCKING or
319321
err.args[0] in (ssl.SSL_ERROR_WANT_READ, ssl.SSL_ERROR_WANT_WRITE)):
322+
if err.args[0] in NONBLOCKING:
323+
self._socket_writable = False
320324
with self._deque_lock:
321325
self.deque.appendleft(next_msg)
322326
else:

‎cassandra/pool.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -418,7 +418,7 @@ def borrow_connection(self, timeout):
418418
remaining = timeout
419419
while True:
420420
with conn.lock:
421-
if conn.in_flight <= conn.max_request_id:
421+
if conn.in_flight < conn.max_request_id:
422422
conn.in_flight += 1
423423
return conn, conn.get_request_id()
424424
if timeout is not None:

‎tests/integration/simulacron/test_backpressure.py‎

Lines changed: 179 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,179 @@
1+
# Copyright DataStax, Inc.
2+
#
3+
# Licensed under the Apache License, Version 2.0 (the "License");
4+
# you may not use this file except in compliance with the License.
5+
# You may obtain a copy of the License at
6+
#
7+
# http://www.apache.org/licenses/LICENSE-2.0
8+
#
9+
# Unless required by applicable law or agreed to in writing, software
10+
# distributed under the License is distributed on an "AS IS" BASIS,
11+
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
# See the License for the specific language governing permissions and
13+
# limitations under the License.
14+
import time
15+
16+
from cassandra import OperationTimedOut
17+
from cassandra.cluster import Cluster, ExecutionProfile, EXEC_PROFILE_DEFAULT, NoHostAvailable
18+
from cassandra.policies import RoundRobinPolicy, WhiteListRoundRobinPolicy
19+
from tests.integration import requiressimulacron, libevtest
20+
from tests.integration.simulacron import SimulacronBase, PROTOCOL_VERSION
21+
from tests.integration.simulacron.utils import ResumeReads, PauseReads, prime_request, start_and_prime_singledc
22+
23+
24+
@requiressimulacron
25+
@libevtest
26+
class TCPBackpressureTests(SimulacronBase):
27+
def setUp(self):
28+
self.callback_successes = 0
29+
self.callback_errors = 0
30+
31+
def callback_success(self, results):
32+
self.callback_successes += 1
33+
34+
def callback_error(self, results):
35+
self.callback_errors += 1
36+
37+
def _fill_buffers(self, session, query, expected_blocked=3, **execute_kwargs):
38+
futures = []
39+
buffer = '1' * 50000
40+
for _ in range(100000):
41+
future = session.execute_async(query, [buffer], **execute_kwargs)
42+
futures.append(future)
43+
44+
total_blocked = 0
45+
for pool in session.get_pools():
46+
if not pool._connection._socket_writable:
47+
total_blocked += 1
48+
if total_blocked >= expected_blocked:
49+
break
50+
else:
51+
raise Exception("Unable to fill TCP send buffer on expected number of nodes")
52+
return futures
53+
54+
def test_paused_connections(self):
55+
""" Verify all requests come back as expected if node resumes within query timeout """
56+
start_and_prime_singledc()
57+
profile = ExecutionProfile(request_timeout=500, load_balancing_policy=RoundRobinPolicy())
58+
cluster = Cluster(
59+
protocol_version=PROTOCOL_VERSION,
60+
compression=False,
61+
execution_profiles={EXEC_PROFILE_DEFAULT: profile},
62+
)
63+
session = cluster.connect(wait_for_all_pools=True)
64+
self.addCleanup(cluster.shutdown)
65+
66+
query = session.prepare("INSERT INTO table1 (id) VALUES (?)")
67+
68+
prime_request(PauseReads())
69+
futures = self._fill_buffers(session, query)
70+
71+
# Make sure we actually have some stuck in-flight requests
72+
for in_flight in [pool._connection.in_flight for pool in session.get_pools()]:
73+
self.assertGreater(in_flight, 100)
74+
time.sleep(.5)
75+
for in_flight in [pool._connection.in_flight for pool in session.get_pools()]:
76+
self.assertGreater(in_flight, 100)
77+
78+
prime_request(ResumeReads())
79+
80+
for future in futures:
81+
try:
82+
future.result()
83+
except NoHostAvailable as e:
84+
# We shouldn't have any timeouts here, but all of the queries beyond what can fit
85+
# in the tcp buffer will have returned with a ConnectionBusy exception
86+
self.assertIn("ConnectionBusy", str(e))
87+
88+
# Verify that we can continue sending queries without any problems
89+
for host in session.cluster.metadata.all_hosts():
90+
session.execute(query, ["a"], host=host)
91+
92+
def test_queued_requests_timeout(self):
93+
""" Verify that queued requests timeout as expected """
94+
start_and_prime_singledc()
95+
profile = ExecutionProfile(request_timeout=.1, load_balancing_policy=RoundRobinPolicy())
96+
cluster = Cluster(
97+
protocol_version=PROTOCOL_VERSION,
98+
compression=False,
99+
execution_profiles={EXEC_PROFILE_DEFAULT: profile},
100+
)
101+
session = cluster.connect(wait_for_all_pools=True)
102+
self.addCleanup(cluster.shutdown)
103+
104+
query = session.prepare("INSERT INTO table1 (id) VALUES (?)")
105+
106+
prime_request(PauseReads())
107+
108+
futures = []
109+
for i in range(1000):
110+
future = session.execute_async(query, [str(i)])
111+
future.add_callbacks(callback=self.callback_success, errback=self.callback_error)
112+
futures.append(future)
113+
114+
successes = 0
115+
for future in futures:
116+
try:
117+
future.result()
118+
successes += 1
119+
except OperationTimedOut:
120+
pass
121+
122+
# Simulacron will respond to a couple queries before cutting off reads, so we'll just verify
123+
# that only "a few" successes happened here
124+
self.assertLess(successes, 50)
125+
self.assertLess(self.callback_successes, 50)
126+
self.assertEqual(self.callback_errors, len(futures) - self.callback_successes)
127+
128+
def test_cluster_busy(self):
129+
""" Verify that once TCP buffer is full we get busy exceptions rather than timeouts """
130+
start_and_prime_singledc()
131+
profile = ExecutionProfile(load_balancing_policy=RoundRobinPolicy())
132+
cluster = Cluster(
133+
protocol_version=PROTOCOL_VERSION,
134+
compression=False,
135+
execution_profiles={EXEC_PROFILE_DEFAULT: profile},
136+
)
137+
session = cluster.connect(wait_for_all_pools=True)
138+
self.addCleanup(cluster.shutdown)
139+
140+
query = session.prepare("INSERT INTO table1 (id) VALUES (?)")
141+
142+
prime_request(PauseReads())
143+
144+
# These requests will get stuck in the TCP buffer and we have no choice but to let them time out
145+
self._fill_buffers(session, query, expected_blocked=3)
146+
147+
# Now that our send buffer is completely full, verify we immediately get busy exceptions rather than timing out
148+
for i in range(1000):
149+
with self.assertRaises(NoHostAvailable) as e:
150+
session.execute(query, [str(i)])
151+
self.assertIn("ConnectionBusy", str(e.exception))
152+
153+
def test_node_busy(self):
154+
""" Verify that once TCP buffer is full, queries continue to get re-routed to other nodes """
155+
start_and_prime_singledc()
156+
profile = ExecutionProfile(load_balancing_policy=RoundRobinPolicy())
157+
cluster = Cluster(
158+
protocol_version=PROTOCOL_VERSION,
159+
compression=False,
160+
execution_profiles={EXEC_PROFILE_DEFAULT: profile},
161+
)
162+
session = cluster.connect(wait_for_all_pools=True)
163+
self.addCleanup(cluster.shutdown)
164+
165+
query = session.prepare("INSERT INTO table1 (id) VALUES (?)")
166+
167+
prime_request(PauseReads(dc_id=0, node_id=0))
168+
169+
blocked_profile = ExecutionProfile(load_balancing_policy=WhiteListRoundRobinPolicy(["127.0.0.1"]))
170+
cluster.add_execution_profile('blocked_profile', blocked_profile)
171+
172+
# Fill our blocked node's tcp buffer until we get a busy exception
173+
self._fill_buffers(session, query, expected_blocked=1, execution_profile='blocked_profile')
174+
175+
# Now that our send buffer is completely full on one node,
176+
# verify queries get re-routed to other nodes and queries complete successfully
177+
for i in range(1000):
178+
session.execute(query, [str(i)])
179+

‎tests/integration/simulacron/test_connection.py‎

Lines changed: 39 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -24,22 +24,22 @@
2424
from cassandra import OperationTimedOut
2525
from cassandra.cluster import (EXEC_PROFILE_DEFAULT, Cluster, ExecutionProfile,
2626
_Scheduler, NoHostAvailable)
27-
from cassandra.policies import HostStateListener, RoundRobinPolicy
27+
from cassandra.policies import HostStateListener, RoundRobinPolicy, WhiteListRoundRobinPolicy
2828

2929
from tests import connection_class, thread_pool_executor_class
3030
from tests.util import late
3131
from tests.integration import requiressimulacron, libevtest
3232
from tests.integration.util import assert_quiescent_pool_state
3333
# important to import the patch PROTOCOL_VERSION from the simulacron module
3434
from tests.integration.simulacron import SimulacronBase, PROTOCOL_VERSION
35-
from cassandra.connection import DEFAULT_CQL_VERSION
35+
from cassandra.connection import DEFAULT_CQL_VERSION, Connection
3636
from tests.unit.cython.utils import cythontest
3737
from tests.integration.simulacron.utils import (NO_THEN, PrimeOptions,
3838
prime_query, prime_request,
3939
start_and_prime_cluster_defaults,
4040
start_and_prime_singledc,
4141
clear_queries, RejectConnections,
42-
RejectType, AcceptConnections)
42+
RejectType, AcceptConnections, PauseReads, ResumeReads)
4343

4444

4545
class TrackDownListener(HostStateListener):
@@ -475,3 +475,39 @@ def test_driver_recovers_nework_isolation(self):
475475
time.sleep(idle_heartbeat_timeout + idle_heartbeat_interval + 2)
476476

477477
self.assertIsNotNone(session.execute("SELECT * from system.local"))
478+
479+
def test_max_in_flight(self):
480+
""" Verify we don't exceed max_in_flight when borrowing connections or sending heartbeats """
481+
Connection.max_in_flight = 50
482+
start_and_prime_singledc()
483+
profile = ExecutionProfile(request_timeout=1, load_balancing_policy=WhiteListRoundRobinPolicy(['127.0.0.1']))
484+
cluster = Cluster(
485+
protocol_version=PROTOCOL_VERSION,
486+
compression=False,
487+
execution_profiles={EXEC_PROFILE_DEFAULT: profile},
488+
idle_heartbeat_interval=.1,
489+
idle_heartbeat_timeout=.1,
490+
)
491+
session = cluster.connect(wait_for_all_pools=True)
492+
self.addCleanup(cluster.shutdown)
493+
494+
query = session.prepare("INSERT INTO table1 (id) VALUES (?)")
495+
496+
prime_request(PauseReads())
497+
498+
futures = []
499+
# + 50 because simulacron doesn't immediately block all queries
500+
for i in range(Connection.max_in_flight + 50):
501+
futures.append(session.execute_async(query, ['a']))
502+
503+
prime_request(ResumeReads())
504+
505+
for future in futures:
506+
# We're veryfing we don't get an assertion error from Connection.get_request_id,
507+
# so skip any valid errors
508+
try:
509+
future.result()
510+
except OperationTimedOut:
511+
pass
512+
except NoHostAvailable:
513+
pass

‎tests/integration/simulacron/utils.py‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -338,6 +338,33 @@ def method(self):
338338
return "DELETE"
339339

340340

341+
class _PauseOrResumeReads(SimulacronRequest):
342+
def __init__(self, cluster_name=DEFAULT_CLUSTER, dc_id=None, node_id=None):
343+
self.path = "pause-reads/{}".format(cluster_name)
344+
if dc_id is not None:
345+
self.path += "/{}".format(dc_id)
346+
if node_id is not None:
347+
self.path += "/{}".format(node_id)
348+
elif node_id:
349+
raise Exception("Can't set node_id without dc_id")
350+
351+
@property
352+
def method(self):
353+
raise NotImplementedError()
354+
355+
356+
class PauseReads(_PauseOrResumeReads):
357+
@property
358+
def method(self):
359+
return "PUT"
360+
361+
362+
class ResumeReads(_PauseOrResumeReads):
363+
@property
364+
def method(self):
365+
return "DELETE"
366+
367+
341368
def prime_driver_defaults():
342369
"""
343370
Function to prime the necessary queries so the test harness can run

0 commit comments

Comments
 (0)