Skip to content

fix(core): do not allow responses to choke request and ping processing #633

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Dec 13, 2020
Merged

fix(core): do not allow responses to choke request and ping processing #633

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 21 additions & 10 deletions kazoo/protocol/connection.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -561,7 +561,6 @@ def _connect_loop(self, retry):
def _connect_attempt(self, host, hostip, port, retry):
client = self.client
KazooTimeoutError = self.handler.timeout_exception
close_connection = False

self._socket = None

Expand All @@ -582,13 +581,14 @@ def _connect_attempt(self, host, hostip, port, retry):
connect_timeout = connect_timeout / 1000.0
retry.reset()
self.ping_outstanding.clear()
last_send = time.time()
with self._socket_error_handling():
while not close_connection:
while True:
# Watch for something to read or send
jitter_time = random.randint(0, 40) / 100.0
jitter_time = random.randint(1, 40) / 100.0
deadline = last_send + read_timeout / 2.0 - jitter_time
# Ensure our timeout is positive
timeout = max([read_timeout / 2.0 - jitter_time,
jitter_time])
timeout = max([deadline - time.time(), jitter_time])
s = self.handler.select([self._socket, self._read_sock],
[], [], timeout)[0]

Expand All @@ -597,12 +597,23 @@ def _connect_attempt(self, host, hostip, port, retry):
self.ping_outstanding.clear()
raise ConnectionDropped(
"outstanding heartbeat ping not received")
self._send_ping(connect_timeout)
elif s[0] == self._socket:
response = self._read_socket(read_timeout)
close_connection = response == CLOSE_RESPONSE
else:
self._send_request(read_timeout, connect_timeout)
if self._socket in s:
response = self._read_socket(read_timeout)
if response == CLOSE_RESPONSE:
break
# Check if any requests need sending before proceeding
# to process more responses. Otherwise the responses
# may choke out the requests. See PR#633.
if self._read_sock in s:
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You may want to add a comment here saying something like:

Check if any requests need sending before proceeding to process more responses. Otherwise the responses may choke out the requests.

That way if someone refactors this code, they'll be aware of the issue... providing that "why" context is super important IMO for tricky code like this. Ideally we'd add a test, but this kind of perf issue is super hard to test reliably so I completely understand why there isn't one.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thank you for the review! Comment added.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

FWIW, you may already know this, but most modern fonts are designed so you only need one space after the period, not two. Not something that should hold up this PR at all, but just an FYI.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hi @jeffwidman,

Right. Old habits die hard… In any case, I should have paid closer attention to surrounding comments; sorry about that.

self._send_request(read_timeout, connect_timeout)
# Requests act as implicit pings.
last_send = time.time()
continue

if time.time() >= deadline:
self._send_ping(connect_timeout)
last_send = time.time()
self.logger.info('Closing connection to %s:%s', host, port)
client._session_callback(KeeperState.CLOSED)
return STOP_CONNECTING
Expand Down