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

Jump to bottom

Make sure inactive request generator is stopped before spawning new one in Pub / Sub consumer. #4503

Merged

dhermes merged 9 commits into googleapis:master from dhermes:pubsub-zombie-1

Dec 1, 2017

Contributor

dhermes commented Nov 30, 2017 •

edited

Loading

~~This makes the request_queue used by Consumer._request_generator_thread() local to the caller. This way on recovery a new queue can be used for the new thread.~~

dhermes added the api: pubsub label

dhermes assigned lukesneeringer, tseaver and theacodes

dhermes requested a review from lukesneeringer as a code owner

November 30, 2017 17:22

googlebot added the cla: yes label

dhermes force-pushed the pubsub-zombie-1 branch from da0510c to f33ac4b Compare

November 30, 2017 17:33

dhermes added a commit to dhermes/google-cloud-pubsub-performance that referenced this pull request


          Adding support for running from a local checkout.

c8bbda6

Then running `message-after-recover` against PR
googleapis/google-cloud-python#4503

lukesneeringer suggested changes

View reviewed changes

Contributor

lukesneeringer left a comment

One item that I believe to be wrong, and one question.

pubsub/google/cloud/pubsub_v1/subscriber/_consumer.py Outdated

@@ @@ -244,7 +252,19 @@ def _blocking_consume(self): @@
                               break
                           except Exception as exc:
                               recover = self._policy.on_exception(exc)
-                              if not recover:
+                              if recover:
+                                  with threading.Lock():

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

pubsub/google/cloud/pubsub_v1/subscriber/_helper_threads.py

@@ @@ -60,12 +60,13 @@ def __init__(self): @@
                   def __contains__(self, needle):
                       return needle in self._helper_threads
-                  def start(self, name, queue, target):
+                  def start(self, name, queue_put, target):

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

dhermes commented

View reviewed changes

pubsub/google/cloud/pubsub_v1/subscriber/_consumer.py Outdated

+                                  with threading.Lock():
+                                      # Must lock so that ``self.send_request()`` cannot add
+                                      # more items to the queue while it is being replaced.
+                                      previous_queue = self._request_queue

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

dhermes mentioned this pull request

Send STOP to defunct consumer threads without swapping request queue. #4504

Closed

Contributor Author

dhermes commented Dec 1, 2017

@jonparrott @lukesneeringer PTAL

dhermes commented

View reviewed changes

pubsub/google/cloud/pubsub_v1/subscriber/_consumer.py Outdated

+                          try:
+                              request_generator.close()
+                          except ValueError as exc:
+                              if exc.args != ('generator already executing',):

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

pubsub/google/cloud/pubsub_v1/subscriber/_consumer.py Outdated

+                      """
+                      with self._put_lock:
+                          try:
+                              request_generator.close()

This comment was marked as spam.

Sign in to view

pubsub/google/cloud/pubsub_v1/subscriber/_helper_threads.py

@@ @@ -60,12 +60,13 @@ def __init__(self): @@
                   def __contains__(self, needle):
                       return needle in self._helper_threads
-                  def start(self, name, queue, target):
+                  def start(self, name, queue_put, target):

This comment was marked as spam.

Sign in to view

dhermes changed the title ~~Possible solution for zombie threads in Pub / Sub consumer.~~ Make sure inactive request generator is stopped before spawning new one in Pub / Sub consumer.

lukesneeringer approved these changes

View reviewed changes

theacodes reviewed

View reviewed changes

pubsub/google/cloud/pubsub_v1/subscriber/_consumer.py Outdated

+                              # queue **was** and remains empty.
+                              self._request_queue.put(_helper_threads.STOP)
+                              # Wait for the request generator to ``.get()`` the ``STOP``.
+                              while not self._request_queue.empty():

This comment was marked as spam.

Sign in to view

pubsub/google/cloud/pubsub_v1/subscriber/_consumer.py Outdated

+                          except ValueError as exc:
+                              if exc.args != ('generator already executing',):
+                                  raise
+                              if not self._request_queue.empty():

This comment was marked as spam.

Sign in to view

dhermes force-pushed the pubsub-zombie-1 branch 2 times, most recently from 640524c to bc44c1e Compare

December 1, 2017 20:06

dhermes commented

View reviewed changes

pubsub/google/cloud/pubsub_v1/subscriber/_consumer.py Outdated

+                                  'Waiting for active request generator to receive STOP')
+                              while not self._request_queue.empty():
+                                  pass
+                              request_generator.close()

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

dhermes added 7 commits

December 1, 2017 13:40


          Possible solution for zombie threads in Pub / Sub consumer.

b40e878


          Dropping the need to swap the request queue, using a "put()" lock.

92590a5


          Missing newline.

29090b4


          Unit test hygiene borrowed from googleapis#4504.

713ce4e


          Updating "stop generator" unit tests.

63f2b17

- Made the "already closed" test use an actual generator instead
  of a mock.
- Removed data race from `test_stop_request_generator_running()`
- Added tracking of the values received from the queue in
  test generator


          Re-factoring _stop_request_generator() a slight bit.

c85ee41

- Dropping the "extra" check
- Making it return `True/False` rather than throwing an exception,
  this way the recovery can just be negated instead of a panic
- Added logging statements to indicate if it was stopped and when
  the helper is actually waiting on `Queue.empty()`


          Dropping the second request_generator.close().

e78f9d2

dhermes force-pushed the pubsub-zombie-1 branch from bc44c1e to e78f9d2 Compare

December 1, 2017 21:40

Contributor Author

dhermes commented Dec 1, 2017

@lukesneeringer @jonparrott PTAL. Phew I think it's finally merge-able.

theacodes reviewed

View reviewed changes

pubsub/google/cloud/pubsub_v1/subscriber/_consumer.py Outdated

+                                      'request queue is not empty.')
+                                  return False
+                              # If we **cannot** close the request generator,
+                              # then there is no blocking get on the queue. Since

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

theacodes reviewed

View reviewed changes

pubsub/tests/unit/pubsub_v1/subscriber/test_consumer.py

		yield value


		def test_stop_request_generator_not_running():

This comment was marked as spam.

Sign in to view

theacodes approved these changes

View reviewed changes

dhermes added 2 commits

December 1, 2017 13:58


          Amending comment.

e4c1876


          Adding comments for _stop_request_generator() unit tests.

a0c3278

Also updating `test_stop_request_generator_queue_non_empty()` to
use a real generator rather than a mock.

Contributor Author

dhermes commented Dec 1, 2017

I added explanations for each unit test @jonparrott. Also updated one of them to use a real generator instead of a mock. Will merge once CI is green.

This code is no fun.

dhermes merged commit 5c010e3 into googleapis:master

dhermes deleted the pubsub-zombie-1 branch

December 1, 2017 22:39

dhermes mentioned this pull request

PubSub with Python 3.5 spawns threads that don't die, eventually bringing the process to consume all CPU available #4563

Closed

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

api: pubsub cla: yes