René's URL Explorer Experiment

{"@context":"https://schema.org","@type":"DiscussionForumPosting","headline":"Possible race condition between `_on_response` and `close`","articleBody":"\r\n#### Environment details\r\n\r\n  - OS type and version: Ubuntu 22.04\r\n  - Python version: 3.11\r\n  - pip version: 23.0.1\r\n  - `google-cloud-pubsub` version: 2.18.4\r\n\r\n#### Steps to reproduce\r\n\r\n  1. Run code sample indefinitely\r\n  2. Sometime it shows Assertion error in `_on_response` function.\r\n\r\n#### Code example\r\n\r\n```python\r\ndef on_subscribe(subscription, until=None):\r\n    \"\"\"Decorator factory that provides subscribed messages to function.\r\n    It handle decorated function as callback. So message should be acked/nacked\r\n    inside decorated function.\r\n\r\n    Args:\r\n        subscription (str): Subscription ID. \r\n            Should be `projects/{PROJECT_ID}/subscriptions/{SUBSCRIPTION_ID}`\r\n        until (datetime.datetime): This function will subscribe messages\r\n            published until this timestamp.\r\n    \"\"\"\r\n\r\n    def _callback_factory(func, finished, subscribe_until, **kwargs):\r\n\r\n        def _callback(message):\r\n            \"\"\"Callback function.\r\n                \r\n            It sends signal if subscribed all messages.\r\n            \"\"\"\r\n            publish_time = datetime.fromtimestamp(\r\n                message.publish_time.timestamp())\r\n            if subscribe_until and publish_time \u003c= subscribe_until:\r\n                return func(message, **kwargs)\r\n            if subscribe_until and not finished.is_set():\r\n                logging.info('Subscribed all messages published until %s',\r\n                             subscribe_until)\r\n                finished.set()\r\n                message.nack()\r\n\r\n        return _callback\r\n\r\n    def _wrapper(func):\r\n\r\n        @functools.wraps(func)\r\n        def _inner_wrapper(**kwargs):\r\n            # Event variable that is triggered when all messages are subscribed\r\n            all_subscribed = Event()\r\n\r\n            callback = _callback_factory(func=func,\r\n                                         finished=all_subscribed,\r\n                                         subscribe_until=subscribe_until,\r\n                                         **kwargs)\r\n\r\n            # Ensure closing subscriber for memory leak prevention.\r\n            with pubsub_v1.SubscriberClient() as subscriber:\r\n                future = subscriber.subscribe(\r\n                    subscription=subscription,\r\n                    callback=callback,\r\n                    await_callbacks_on_shutdown=True,\r\n                    flow_control=pubsub_v1.types.FlowControl(max_messages=5000),\r\n                )\r\n\r\n                all_subscribed.wait(timeout=60)\r\n\r\n                # Wait until future is finished when it's cancelled.\r\n                # If it cancelled by timeout or keyboard interrupt, ignore it.\r\n                try:\r\n                    future.cancel()\r\n                    future.result(timeout=60)\r\n                except (KeyboardInterrupt, TimeoutError):\r\n                    pass\r\n                except Exception as e:\r\n                    logging.error(\"Error occurs during subscription to %s: %s\",\r\n                                  subscription, str(e))\r\n\r\n        return _inner_wrapper\r\n\r\n    return _wrapper\r\n\r\n@on_subscribe(subscription=\"SUBSCRIPTION\")\r\ndef callback(message):\r\n    # Do something with message\r\n```\r\n\r\n#### Stack trace\r\n```\r\nTraceback (most recent call last):\r\n  File \"/layers/google.python.pip/pip/lib/python3.11/site-packages/google/api_core/bidi.py\", line 657, in _thread_main\r\n    self._on_response(response)\r\n  File \"/layers/google.python.pip/pip/lib/python3.11/site-packages/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py\", line 1107, in _on_response\r\n    assert self._scheduler is not None\r\n```\r\n\r\n#### Explanation\r\nIt's because `future.cancel()` executes `manager.close()` which makes `_scheduler` as `None` and it makes `_on_response` raise AssertionError.\r\n\r\nMaybe it has to be protected by threading lock somehow.","author":{"url":"https://github.com/hson2","@type":"Person","name":"hson2"},"datePublished":"2023-09-21T05:43:50.000Z","interactionStatistic":{"@type":"InteractionCounter","interactionType":"https://schema.org/CommentAction","userInteractionCount":3},"url":"https://github.com/997/python-pubsub/issues/997"}