Skip to content

Fix race condition in bidi.BackgroundConsumer #5256

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
theacodes merged 5 commits into from
Apr 27, 2018
Merged

Fix race condition in bidi.BackgroundConsumer #5256

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
a167f71
Fix race condition in bidi.BackgroundConsumer
Apr 27, 2018
4be7066
Address review comments
Apr 27, 2018
11d0fbe
Tiny thing
Apr 27, 2018
ac00d77
Tiny thing
Apr 27, 2018
6a8630e
Remove unnecessary if
Apr 27, 2018
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
26 changes: 17 additions & 9 deletions pubsub/google/cloud/pubsub_v1/subscriber/_protocol/bidi.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -421,16 +421,24 @@ def _thread_main(self):
self._bidi_rpc.open()

while self._bidi_rpc.is_active:
if not self.is_paused:
_LOGGER.debug('waiting for recv.')
response = self._bidi_rpc.recv()
_LOGGER.debug('recved response.')
self._on_response(response)
else:
_LOGGER.debug('paused, waiting for waking.')
with self._wake:
# Do not allow the paused status to change at all during this
# section. There is a condition where we could be resumed
# between checking if we are paused and calling wake.wait(),
# which means that we will miss the notification to wake up
# (oops!) and wait for a notification that will never come.
# Keeping the lock throughout avoids that.
# In the future, we could use `Condition.wait_for` if we drop
# Python 2.7.
with self._wake:
if self._paused:
_LOGGER.debug('paused, waiting for waking.')
self._wake.wait()
_LOGGER.debug('woken.')
_LOGGER.debug('woken.')

_LOGGER.debug('waiting for recv.')
response = self._bidi_rpc.recv()
_LOGGER.debug('recved response.')
self._on_response(response)

except exceptions.GoogleAPICallError as exc:
_LOGGER.debug(
Expand Down
2 changes: 2 additions & 0 deletions pubsub/tests/unit/pubsub_v1/subscriber/test_bidi.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -504,7 +504,9 @@ def on_response(response):
# consumer.
should_continue.set()
consumer.resume()

responses_and_events[mock.sentinel.response_2].wait()

assert recved_responses == [
mock.sentinel.response_1, mock.sentinel.response_2]

Expand Down