[SQS] Stop blocking Queue.get() call on LocalStack shutdown #12214

gregfurman · 2025-01-31T17:28:20Z

Motivation

Blocking calls in the SQS provider's internal queue implementation cause LocalStack shutdown to hang, and in some cases time-out.

This PR allows for these blocking Queue.get calls to exit early if LocalStack is shutting down and the do_shutdown method is called.

Changes

Add an InterruptibleQueue class that overwrites a Queue implementations get method to leverage a shutdown Threading.Event.
InterruptibleQueue also has a do_shutdown method that a queue to be shutdown early if it is currently undergoing a blocking call.
Add queue closing to the SQS provider's on_before_stop lifecycle hook (which happens after persistence) to ensure no hanging threads prevent LS from shutting down.

github-actions · 2025-01-31T18:15:43Z

LocalStack Community integration with Pro

2 files ± 0 2 suites ±0 1h 12m 5s ⏱️ - 41m 23s
2 990 tests - 1 085 2 860 ✅ - 900 130 💤 - 185 0 ❌ ±0
2 992 runs - 1 085 2 860 ✅ - 900 132 💤 - 185 0 ❌ ±0

Results for commit ca3b342. ± Comparison against base commit ca9275b.

This pull request removes 1087 and adds 2 tests. Note that renamed tests count towards both.

tests.aws.scenario.bookstore.test_bookstore.TestBookstoreApplication ‑ test_lambda_dynamodb
tests.aws.scenario.bookstore.test_bookstore.TestBookstoreApplication ‑ test_opensearch_crud
tests.aws.scenario.bookstore.test_bookstore.TestBookstoreApplication ‑ test_search_books
tests.aws.scenario.bookstore.test_bookstore.TestBookstoreApplication ‑ test_setup
tests.aws.scenario.kinesis_firehose.test_kinesis_firehose.TestKinesisFirehoseScenario ‑ test_kinesis_firehose_s3
tests.aws.scenario.lambda_destination.test_lambda_destination_scenario.TestLambdaDestinationScenario ‑ test_destination_sns
tests.aws.scenario.lambda_destination.test_lambda_destination_scenario.TestLambdaDestinationScenario ‑ test_infra
tests.aws.scenario.loan_broker.test_loan_broker.TestLoanBrokerScenario ‑ test_prefill_dynamodb_table
tests.aws.scenario.loan_broker.test_loan_broker.TestLoanBrokerScenario ‑ test_stepfunctions_input_recipient_list[step_function_input0-SUCCEEDED]
tests.aws.scenario.loan_broker.test_loan_broker.TestLoanBrokerScenario ‑ test_stepfunctions_input_recipient_list[step_function_input1-SUCCEEDED]
…

tests.aws.services.sqs.test_sqs.TestSqsProvider ‑ test_receive_empty_queue[sqs]
tests.aws.services.sqs.test_sqs.TestSqsProvider ‑ test_receive_empty_queue[sqs_query]

♻️ This comment has been updated with latest results.

dfangl

Already looks really nice, just some minor comments.
Could we test the shutdown behavior with a bootstrap test perhaps? Do you think that makes sense?

dfangl · 2025-02-03T10:12:27Z

localstack-core/localstack/services/sqs/models.py

@@ -300,6 +301,9 @@ def __init__(self, name: str, region: str, account_id: str, attributes=None, tag
        self.permissions = set()
        self.mutex = threading.RLock()

+    def do_shutdown(self):


Usually we call do_shutdown from a method like shutdown, especially when inheriting from an implementation which has a more generic shutdown method.

Why did you choose this name here?

Later versions of the python Queue already have a shutdown implementation so thought to distinguish this method from that. Also saw some other classes using it so assumed it was standard naming practice.

Will rename it to just shutdown -- but we should be cognizant that Queue.shutdown will be introduced in python 3.13+.

This one is the sqs queue implementation though, not our inherited one, right?

This method is the SQS model yeah (which contains an InterruptibleQueue attribute)

Then I do not understand the implementation of the shutdown method in python 3.13 for a standard queue influencing this? The argument would fit for the InterruptibleQueue implementation, not this one, or do I misunderstand something?

Since the shutdown signal is made to the SqsQueue object, I was wanting to distinguish the SqsQueue.shutdown from the SqsQueue.internal_queue.shutdown since a shutdown call would otherwise look like:

# StandardQueue(SqsQueue) def shutdown(self): self.queue.shutdown()

So the shutdown method is just us calling another interal shutdown method. Was hoping to distinguish these methods with the higher level abstraction calling a do_shutdown that (in turn) triggers the internal queue's shutdown.

Fair enough, but I think it is fine in this case!

Changed it to shutdown 🙂

dfangl · 2025-02-03T10:12:38Z

localstack-core/localstack/services/sqs/queue.py

+class InterruptiblePriorityQueue(PriorityQueue, InterruptibleQueue):
+    pass


dfangl · 2025-02-03T10:13:21Z

localstack-core/localstack/services/sqs/queue.py

+    def get(self, block=True, timeout=None):
+        with self.not_empty:
+            if not block:
+                if not self._qsize():
+                    raise Empty
+            elif timeout is None:
+                while not self._qsize() and not self.shutdown_event.is_set():
+                    self.not_empty.wait()
+            elif timeout < 0:
+                raise ValueError("'timeout' must be a non-negative number")
+            else:
+                endtime = time.time() + timeout
+                while not self._qsize() and not self.shutdown_event.is_set():
+                    remaining = endtime - time.time()
+                    if remaining <= 0.0:
+                        raise Empty
+                    self.not_empty.wait(remaining)
+            if self.shutdown_event.is_set():
+                raise Empty
+            item = self._get()
+            self.not_full.notify()
+            return item


Would love a comment here to highlight the changes we made from the actual queue implementation, so basically highlighting the event changes.

dfangl · 2025-02-03T10:13:38Z

localstack-core/localstack/services/sqs/queue.py

+
+    def __init__(self, maxsize=0):
+        super().__init__(maxsize)
+        self.shutdown_event = threading.Event()


Come to think about it, this could be a boolean as well, we never wait for it. WDYT?

The python 3.13 implementation actually has a boolean flag instead of a thread (see Queue.get). Suppose since we are locking on entry in anycase using a flag makes sense. Will change!

dfangl

LGTM, the changes are relatively minimal, but we can now "abort" connections on shutdown!
Would love an -ext run though (especially of the sqs persistence tests) before merging this, just to make sure!

gregfurman · 2025-02-03T14:56:25Z

@dfangl Am currently running an -ext test just targeting the SQS persistence tests. Also, will add a boostrap test in a follow-up

gregfurman added aws:sqs Amazon Simple Queue Service semver: patch Non-breaking changes which can be included in patch releases labels Jan 31, 2025

gregfurman requested a review from dfangl January 31, 2025 17:28

gregfurman self-assigned this Jan 31, 2025

gregfurman marked this pull request as ready for review January 31, 2025 20:42

gregfurman requested review from thrau and baermat as code owners January 31, 2025 20:43

dfangl reviewed Feb 3, 2025

View reviewed changes

gregfurman requested a review from dfangl February 3, 2025 14:50

dfangl approved these changes Feb 3, 2025

View reviewed changes

gregfurman added 2 commits February 3, 2025 22:43

[SQS] Stop blocking threads on LocalStack shutdown

e86bff5

Rename shutdown method and highlight custom implementation

ca3b342

gregfurman force-pushed the fix/sqs/long-polling-shutdown branch from 3b034a3 to ca3b342 Compare February 3, 2025 20:44

gregfurman merged commit fd3d900 into master Feb 4, 2025
32 checks passed

gregfurman deleted the fix/sqs/long-polling-shutdown branch February 4, 2025 08:50

		class InterruptiblePriorityQueue(PriorityQueue, InterruptibleQueue):
		pass

Uh oh!

[SQS] Stop blocking Queue.get() call on LocalStack shutdown #12214

[SQS] Stop blocking Queue.get() call on LocalStack shutdown #12214

Uh oh!

Conversation

gregfurman commented Jan 31, 2025

Motivation

Changes

Uh oh!

github-actions bot commented Jan 31, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

LocalStack Community integration with Pro

Uh oh!

dfangl left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

gregfurman Feb 3, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dfangl left a comment

Choose a reason for hiding this comment

Uh oh!

gregfurman commented Feb 3, 2025

Uh oh!

Uh oh!

Uh oh!

github-actions bot commented Jan 31, 2025 •

edited

Loading

gregfurman Feb 3, 2025 •

edited

Loading