Won't self.pubsub.subscribe in RedisManager._listen() raise exception?

[griiid]

Hello, thanks for building such a good tool.

I want to know whether the code self.pubsub.subscribe(self.channel) in RedisManager._listen() will raise exception or not in any siuation.

python-socketio/src/socketio/redis_manager.py

Lines 110 to 117 in de4d5b5

	def _listen(self):
	channel = self.channel.encode('utf-8')
	self.pubsub.subscribe(self.channel)
	for message in self._redis_listen_with_retries():
	if message['channel'] == channel and \
	message['type'] == 'message' and 'data' in message:
	yield message['data']
	self.pubsub.unsubscribe(self.channel)

If it may raise exception, will the thread (PubSubManager._thread) be killed and we will never receive any data anymore?

The problem I met is: once the server log shows Cannot publish to redis... retrying, the client side will never receive socket data anymore. (I don't see the log Cannot publish to redis... giving up)

I wondered It's because the publisher reconnected successfully, but the subscriber (which uses RedisManager._listen()) was died, so after the log Cannot publish to redis... retrying, the server can successfully emit message (I see the emitting logs), but never receive data from Redis. (I don't see the pubsub messgae emit log)

[miguelgrinberg]

Looking at this I think I agree that it is possible for the listening thread to die. But you should see a stack trace in your log if that was the case.

But yes, this thread needs a handler to catch all exceptions and prevents the thread from dying. I will improve this.

[griiid]

Thank you.

I can't see any stack trace due to exceptions occurred when the socket server is not working. I'm not sure whether it will show in the Logging Explorer on GCP (Google Cloud Platform) or not.

I've added log to monitor this: socketio.server.manager.thread.dead where socketio is created by socketio = SocketIO() with some parameters. I think this might help me to see whether the thread is dead or not as I assumed.

[dongzeli95]

@griiid Did you happen to find some interesting logs? I wonder if there is some issue with the downstream Redis pkg as well? Thanks!

[griiid]

nope, sorry.

[dongzeli95]

I am using flask-socketio==5.2.0 and encounter the same issue. Is this issue confirmed with the root cause of thread dying? Do we have a way to gracefully handle this yet?

[miguelgrinberg]

@dongzeli95 please provide a description of the issue, including stack trace(s) that you get.

[dongzeli95]

{
    "textPayload": "[2023-07-10 15:24:55,649] [10] [ERROR] Cannot publish to redis... retrying",
    "insertId": "64ac22c70009e844b6ec9af9",
    "resource": {
      "type": "cloud_run_revision",
      "labels": {
        "project_id": "biji-390501",
        "configuration_name": "biji-ai",
        "revision_name": "biji-ai-00037-zaz",
        "location": "us-west1",
        "service_name": "biji-ai"
      }
    },
    "timestamp": "2023-07-10T15:24:55.649284Z",
    "labels": {
      "instanceId": "001bdd422e079b3509f7566a627b9da2412272cdead206241c402ed6557f94954909ba199f6b980340ef6fa0e6716c14c19c8ee3e9952f89bcf8374c7a6bec09"
    },
    "logName": "projects/biji-390501/logs/run.googleapis.com%2Fstderr",
    "receiveTimestamp": "2023-07-10T15:24:55.986028775Z"
  },
  {
    "textPayload": "emitting event \"gpt_stream\" to 93ecc9fc-db28-46c4-9740-9360c3610fa7 [/]",
    "insertId": "64ac22c7000d74faece31bca",
    "resource": {
      "type": "cloud_run_revision",
      "labels": {
        "revision_name": "biji-ai-00037-zaz",
        "location": "us-west1",
        "project_id": "biji-390501",
        "configuration_name": "biji-ai",
        "service_name": "biji-ai"
      }
    },
    "timestamp": "2023-07-10T15:24:55.881914Z",
    "labels": {
      "instanceId": "001bdd422e079b3509f7566a627b9da2412272cdead206241c402ed6557f94954909ba199f6b980340ef6fa0e6716c14c19c8ee3e9952f89bcf8374c7a6bec09"
    },
    "logName": "projects/biji-390501/logs/run.googleapis.com%2Fstderr",
    "receiveTimestamp": "2023-07-10T15:24:55.986028775Z"
  },
  {
    "textPayload": "[2023-07-10 15:24:55,881] [10] [INFO] emitting event \"gpt_stream\" to 93ecc9fc-db28-46c4-9740-9360c3610fa7 [/]",
    "insertId": "64ac22c7000d7522a9c4eab1",
    "resource": {
      "type": "cloud_run_revision",
      "labels": {
        "configuration_name": "biji-ai",
        "revision_name": "biji-ai-00037-zaz",
        "location": "us-west1",
        "project_id": "biji-390501",
        "service_name": "biji-ai"
      }
    },
    "timestamp": "2023-07-10T15:24:55.881954Z",
    "labels": {
      "instanceId": "001bdd422e079b3509f7566a627b9da2412272cdead206241c402ed6557f94954909ba199f6b980340ef6fa0e6716c14c19c8ee3e9952f89bcf8374c7a6bec09"
    },
    "logName": "projects/biji-390501/logs/run.googleapis.com%2Fstderr",
    "receiveTimestamp": "2023-07-10T15:24:55.986028775Z"
  },

I upgrade my flask-socketio to 5.3.4(latest). But the issue happened again this morning, I would have to redeploy my application on cloud run to make it work again.

Essentially every time I got this message: [ERROR] Cannot publish to redis... retrying",. The subsequent messages cannot be received by client anymore. Although it looks like the socket is working properly by emitting messages, it won't be received by client.

On my server log, I can only see [ERROR] Cannot publish to redis... retrying", without further details, is there any way we can debug this? I think at this point, the issue is highly related to the RedisManager pkg.

Note: I am also using python 3.8, so not sure if that affect anything.

[dongzeli95]

Context about my current setups:

// requirement.txt

openai==0.27.2
requests==2.28.2
langchain==0.0.202
redis==4.5.5
tiktoken==0.4.0

# Socket IO.
Werkzeug==2.0.1
flask-socketio==5.3.4
gunicorn==20.1.0
eventlet==0.30.2

// socketio_init.py

from flask_socketio import SocketIO

# Create a SocketIO instance.
# We will initialize it with the app later in the create_app() function.
socketio = SocketIO(logger=True, engineio_logger=True)

// my_app.py

import eventlet
eventlet.monkey_patch()

from app.api.socketio_init import socketio

def create_app(environ=None, start_response=None):
    app = Flask(__name__)
    # Redis url for streaming.
    redis_url = os.environ.get('REDIS_URL')
    app.config["REDIS_URL"] = redis_url

    # Initialize SocketIO with the application and Redis for message queue
    socketio.init_app(app, message_queue=redis_url, async_mode='eventlet')

    # Register routes.
    app.register_blueprint(routes_bp.bp)

    # Configure logging
    configure_logging()

    return app

// Dockerfile
CMD ["gunicorn", "-k", "eventlet", "-w", "1", "my_app:app", "-b", "0.0.0.0:8080"]

[dongzeli95]

{
insertId: "64ad754d0009607251bc4b9e"
labels: {1}
logName: "projects/biji-390501/logs/run.googleapis.com%2Fstderr"
receiveTimestamp: "2023-07-11T15:29:17.864299041Z"
resource: {2}
textPayload: "[2023-07-11 15:29:17,614] [4] [ERROR] Cannot publish to redis... retrying"
timestamp: "2023-07-11T15:29:17.614514Z"
}
{
insertId: "64ad754d000960886e81d174"
labels: {1}
logName: "projects/biji-390501/logs/run.googleapis.com%2Fstderr"
receiveTimestamp: "2023-07-11T15:29:17.864299041Z"
resource: {2}
textPayload: "[2023-07-11 15:29:17,614] [4] [ERROR] Connection closed by server."
timestamp: "2023-07-11T15:29:17.614536Z"
}

Added some more logs and this is what I got. Any emit events won't received after that, still trying to pinpoint the issue.

[miguelgrinberg]

@dongzeli95 questions:

• Why is your redis suddenly failing to publish?
• Does the retrying work? Does publishing start working again?
• Any exceptions in the log that may suggest that your Redis listening thread has died (the topic of this issue)?

[dongzeli95]

@miguelgrinberg

• I don't know exactly why it disconnected in the first place but this is flaky so it could be server hiccup.
• Yes the retrying certainly worked, publishing start working after that. Because I don't see the retrying giving up logs ever from server log.
• Good question, I haven't forked PubSubManager so I can't see if there is any thread exception going on. I am planning to add some.

Considering something like this:

class ExceptionLoggingThread(threading.Thread):
    def run(self):
        try:
            if self._target:
                self._target(*self._args, **self._kwargs)
        except Exception:
            logging.exception('Exception in thread: {}'.format(self.name))

class PubSubManager:
    def initialize(self):
        super(PubSubManager, self).initialize()
        if not self.write_only:
            self.thread = ExceptionLoggingThread(target=self._thread)
            self.thread.start()
        self._get_logger().info(self.name + ' backend initialized.')

[miguelgrinberg]

If you are capturing standard output and standard error then you should have a stack trace in your log without having to do anything special to capture crashes. Python does this automatically.

I think it is necessary to understand what's happening to your redis in more detail than just "server hiccup". Based on what you are saying the publish side errors cleanly and works again after a reconnect, while the listening side either hangs or dies, but we don't know which or how or why.

[dongzeli95]

@miguelgrinberg Yes contacting my cloud provider to see they have any insights on this occasional disconnection.

But for the temp solution, re-initializing the thread fixed my problem. This is my new server log:

2023-07-15 08:36:57.108 PDT
[2023-07-15 15:36:57,108] [6] [ERROR] Cannot publish to redis... retrying
2023-07-15 08:36:57.108 PDT
[2023-07-15 15:36:57,108] [6] [ERROR] Connection closed by server.
2023-07-15 08:36:57.110 PDT
redis backend initialized.
2023-07-15 08:36:57.110 PDT
[2023-07-15 15:36:57,110] [6] [INFO] redis backend initialized.
2023-07-15 08:36:57.232 PDT
emitting event "gpt_stream" to 0b46b2e4-e661-4b32-9b7e-19230b645d05 [/]
2023-07-15 08:36:57.232 PDT
[2023-07-15 15:36:57,232] [6] [INFO] emitting event "gpt_stream" to 0b46b2e4-e661-4b32-9b7e-19230b645d05 [/]
2023-07-15 08:36:57.232 PDT
[2023-07-15 15:36:57,232] [6] [ERROR] yield message here

After I re-init my redis backend, the subscribe thread works again by yielding messages.

[miguelgrinberg]

This solution would be an okay workaround if the thread died. If the thread did not die, though, you have a hung thread, and now you started a second thread to take its place. If you get several failures, the hung threads are going to accumulate and consume resources. So while this may be a short-term workaround for you, I don't see it as a permanent fix.