fix: Replace asserts with None checks for graceful shutdown #1244

mukund-ananthu · 2024-09-09T00:41:49Z

There are two code paths that race to set / read the value of self._scheduler variable of streaming_pull_manager

Code path 1

future.cancel():

python-pubsub/google/cloud/pubsub_v1/subscriber/futures.py

Lines 56 to 76 in 5402b62

    
               def cancel(self) -> bool: 
        
                   """Stops pulling messages and shutdowns the background thread consuming 
        
                   messages. 
        
                   The method always returns ``True``, as the shutdown is always initiated. 
        
                   However, if the background stream is already being shut down or the shutdown 
        
                   has completed, this method is a no-op. 
        
                   .. versionchanged:: 2.4.1 
        
                      The method does not block anymore, it just triggers the shutdown and returns 
        
                      immediately. To block until the background stream is terminated, call 
        
                      :meth:`result()` after cancelling the future. 
        
                   .. versionchanged:: 2.10.0 
        
                      The method always returns ``True`` instead of ``None``. 
        
                   """ 
        
                   # NOTE: We circumvent the base future's self._state to track the cancellation 
        
                   # state, as this state has different meaning with streaming pull futures. 
        
                   self.__cancelled = True 
        
                   self.__manager.close() 
        
                   return True

calls

streaming_pull_manager.close():

python-pubsub/google/cloud/pubsub_v1/subscriber/futures.py

Line 75 in 5402b62

self.__manager.close()

which schedules the _shutdown method:

python-pubsub/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py

Lines 896 to 902 in 5402b62

    
           self._regular_shutdown_thread = threading.Thread( 
        
               name=_REGULAR_SHUTDOWN_THREAD_NAME, 
        
               daemon=True, 
        
               target=self._shutdown, 
        
               kwargs={"reason": reason}, 
        
           ) 
        
           self._regular_shutdown_thread.start()

that sets the self._scheduler variable to None using a lock self._closing:

python-pubsub/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py

Lines 912 to 929 in 5402b62

    
           with self._closing: 
        
               if self._closed: 
        
                   return 
        
               # Stop consuming messages. 
        
               if self.is_active: 
        
                   _LOGGER.debug("Stopping consumer.") 
        
                   assert self._consumer is not None 
        
                   self._consumer.stop() 
        
               self._consumer = None 
        
               # Shutdown all helper threads 
        
               _LOGGER.debug("Stopping scheduler.") 
        
               assert self._scheduler is not None 
        
               dropped_messages = self._scheduler.shutdown( 
        
                   await_msg_callbacks=self._await_callbacks_on_shutdown 
        
               ) 
        
               self._scheduler = None

Code Path 2

The streaming_pull_manager's _on_response method:

python-pubsub/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py

Lines 1106 to 1129 in 5402b62

    
           with self._pause_resume_lock: 
        
               assert self._scheduler is not None 
        
               assert self._leaser is not None 
        
               for received_message in received_messages: 
        
                   if ( 
        
                       not self._exactly_once_delivery_enabled() 
        
                       or received_message.ack_id not in expired_ack_ids 
        
                   ): 
        
                       message = google.cloud.pubsub_v1.subscriber.message.Message( 
        
                           received_message.message, 
        
                           received_message.ack_id, 
        
                           received_message.delivery_attempt, 
        
                           self._scheduler.queue, 
        
                           self._exactly_once_delivery_enabled, 
        
                       ) 
        
                       self._messages_on_hold.put(message) 
        
                       self._on_hold_bytes += message.size 
        
                       req = requests.LeaseRequest( 
        
                           ack_id=message.ack_id, 
        
                           byte_size=message.size, 
        
                           ordering_key=message.ordering_key, 
        
                       ) 
        
                       self._leaser.add([req])

asserts that the self._scheduler variable is not None before proceeding to use it for processing further.

Since the two code paths use different locks and are running on different threads, there are times when code path1 sets the self._scheduler value to None, while the _on_response method is still executing. The assert in the _on_response method fails causing an assertion error to be thrown. This happens when the library tries to process streaming pull responses when the library is also being shut down.

This change replaces the asserts with a None check that would return early and stop processing(same behavior as an assert minus the error being thrown), resulting in a graceful shutdown of the library without red herring errors being logged to the user.

Fixes #997 🦕

conventional-commit-lint-gcf · 2024-09-09T00:44:25Z

🤖 I detect that the PR title and the commit message differ and there's only one commit. To use the PR title for the commit history, you can use Github's automerge feature with squashing, or use automerge label. Good luck human!

-- conventional-commit-lint bot
https://conventionalcommits.org/

mukund-ananthu · 2024-09-09T01:12:58Z

@parthea PTAL.

mukund-ananthu · 2024-09-09T01:20:09Z

@hongalex

mukund-ananthu requested review from a team as code owners September 9, 2024 00:41

mukund-ananthu self-assigned this Sep 9, 2024

product-auto-label bot added size: m Pull request size is medium. api: pubsub Issues related to the googleapis/python-pubsub API. labels Sep 9, 2024

mukund-ananthu force-pushed the fix branch from b5a3c15 to 9c2b9e9 Compare September 9, 2024 00:44

mukund-ananthu added the kokoro:force-run Add this label to force Kokoro to re-run the tests. label Sep 9, 2024

yoshi-kokoro removed the kokoro:force-run Add this label to force Kokoro to re-run the tests. label Sep 9, 2024

mukund-ananthu force-pushed the fix branch from 9c2b9e9 to f8ee290 Compare September 9, 2024 01:02

fix: Replace asserts with None checks for graceful shutdown

16f6226

mukund-ananthu force-pushed the fix branch from f8ee290 to 16f6226 Compare September 9, 2024 01:07

mukund-ananthu requested a review from parthea September 9, 2024 01:09

mukund-ananthu requested a review from hongalex September 9, 2024 01:20

githubwua mentioned this pull request Sep 9, 2024

streaming_pull_manager.py throws TypeError: object of type 'NoneType' has no len() #1238

Closed

parthea approved these changes Sep 9, 2024

View reviewed changes

hongalex approved these changes Sep 9, 2024

View reviewed changes

Merge branch 'main' into fix

6d2b288

mukund-ananthu enabled auto-merge (squash) September 9, 2024 18:51

mukund-ananthu merged commit ced4f52 into main Sep 9, 2024
26 of 28 checks passed

mukund-ananthu deleted the fix branch September 9, 2024 19:35

release-please bot mentioned this pull request Sep 9, 2024

chore(main): release 2.23.1 #1246

Merged

githubwua mentioned this pull request Sep 10, 2024

Update streaming_pull_manager.py to address NoneType #1239

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

fix: Replace asserts with None checks for graceful shutdown #1244

fix: Replace asserts with None checks for graceful shutdown #1244

Uh oh!

mukund-ananthu commented Sep 9, 2024 •

edited

Loading

Uh oh!

conventional-commit-lint-gcf bot commented Sep 9, 2024 •

edited

Loading

Uh oh!

mukund-ananthu commented Sep 9, 2024

Uh oh!

mukund-ananthu commented Sep 9, 2024

Uh oh!

Uh oh!

Uh oh!

	def cancel(self) -> bool:
	"""Stops pulling messages and shutdowns the background thread consuming
	messages.

	The method always returns ``True``, as the shutdown is always initiated.
	However, if the background stream is already being shut down or the shutdown
	has completed, this method is a no-op.

	.. versionchanged:: 2.4.1
	The method does not block anymore, it just triggers the shutdown and returns
	immediately. To block until the background stream is terminated, call
	:meth:`result()` after cancelling the future.

	.. versionchanged:: 2.10.0
	The method always returns ``True`` instead of ``None``.
	"""
	# NOTE: We circumvent the base future's self._state to track the cancellation
	# state, as this state has different meaning with streaming pull futures.
	self.__cancelled = True
	self.__manager.close()
	return True

	self._regular_shutdown_thread = threading.Thread(
	name=_REGULAR_SHUTDOWN_THREAD_NAME,
	daemon=True,
	target=self._shutdown,
	kwargs={"reason": reason},
	)
	self._regular_shutdown_thread.start()

	with self._closing:
	if self._closed:
	return

	# Stop consuming messages.
	if self.is_active:
	_LOGGER.debug("Stopping consumer.")
	assert self._consumer is not None
	self._consumer.stop()
	self._consumer = None

	# Shutdown all helper threads
	_LOGGER.debug("Stopping scheduler.")
	assert self._scheduler is not None
	dropped_messages = self._scheduler.shutdown(
	await_msg_callbacks=self._await_callbacks_on_shutdown
	)
	self._scheduler = None

	with self._pause_resume_lock:
	assert self._scheduler is not None
	assert self._leaser is not None

	for received_message in received_messages:
	if (
	not self._exactly_once_delivery_enabled()
	or received_message.ack_id not in expired_ack_ids
	):
	message = google.cloud.pubsub_v1.subscriber.message.Message(
	received_message.message,
	received_message.ack_id,
	received_message.delivery_attempt,
	self._scheduler.queue,
	self._exactly_once_delivery_enabled,
	)
	self._messages_on_hold.put(message)
	self._on_hold_bytes += message.size
	req = requests.LeaseRequest(
	ack_id=message.ack_id,
	byte_size=message.size,
	ordering_key=message.ordering_key,
	)
	self._leaser.add([req])

fix: Replace asserts with None checks for graceful shutdown #1244

fix: Replace asserts with None checks for graceful shutdown #1244

Uh oh!

Conversation

mukund-ananthu commented Sep 9, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Code path 1

Code Path 2

Uh oh!

conventional-commit-lint-gcf bot commented Sep 9, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

mukund-ananthu commented Sep 9, 2024

Uh oh!

mukund-ananthu commented Sep 9, 2024

Uh oh!

Uh oh!

Uh oh!

mukund-ananthu commented Sep 9, 2024 •

edited

Loading

conventional-commit-lint-gcf bot commented Sep 9, 2024 •

edited

Loading