Strange behaviour: TimeoutError when invoking socketio.call() from Celery Worker #2079
Answered
by
miguelgrinberg
lucarillosi
asked this question in
Q&A
-
|
Good morning Miguel, first of all thanks for all the great work you have done with this project. I am very grateful for this and hope to get some help on an issue that i've been troubleshooting for a couple of days. Describe the setup
I am trying to send events from the celery worker and wait for the client acknowledge response. Describe the issue
Expected behaviour
Code snippets In app.py initialize the flask app. I use the # app.py
from gevent import monkey
if __name__ == "__main__":
monkey.patch_all()
import logging
import os
# ...
from app.gateway.server import socketio
from flask import Flask
from flask_cors import CORS
def create_app(environ: str = None, main: bool = True):
"""Creates a Flask app loaded with config based on the FLASK_ENV environment variable"""
app = Flask(__name__)
CORS(
app,
origins=[
"http://localhost",
"http://localhost:5000"
],
)
# Load config from object passed
config = Config(environ=environ).get()
app.config.from_mapping(config)
# ...
if main:
# Configure for Flask server
socketio.init_app(
app,
cors_allowed_origins="*",
logger=logging.getLogger("socketio"),
engineio_logger=logging.getLogger("engineio"),
message_queue="redis://redis",
async_mode="gevent",
)
else:
# Configure for Celery worker
socketio.init_app(
None,
cors_allowed_origins="*",
logger=logging.getLogger("socketio"),
engineio_logger=logging.getLogger("engineio"),
message_queue="redis://redis",
async_mode="threading",
)
return app
if __name__ == "__main__":
app = create_app()
socketio.run(app, host="0.0.0.0", port=5000, use_reloader=False)In server.py I set up socketio instance and connect/disconnect events. # app/gateway/server.py
import logging
from app.config import Config
from flask import request
from flask_socketio import SocketIO
logger = logging.getLogger(__name__)
config = Config()
socketio = SocketIO()
# This callback works from Flask. The result is correctly received.
def request_mycallback(session_id):
logger.info("Requiring mycallback")
result = socketio.call(
"mycallback", "Hello from the server!", to=session_id, timeout=10
)
logger.info("mycallback result: %s", result)
@socketio.on("connect")
def handle_connect():
# ...
print("-----------------------------CONNECTED-----------------------------")
socketio.start_background_task(request_mycallback, request.sid)
return True
@socketio.on("disconnect")
def handle_disconnect():
# ...
print("-----------------------------DIS--CONNECTED-----------------------------")In Task.py I define celery worker tasks. #task.py
import logging
from app.gateway.server import socketio
from celery import shared_task
logger = logging.getLogger(__name__)
# ...
@shared_task(ignore_result=False, name="mycallback_task")
def mycallback_task(organization_id: int) -> None:
"""
This task sends an event to the gateway via websocket, and waits for a response from the gateway.
To test this, run the Flask server, Celery worker, and Gateway client, then run:
$ celery -A app.tasks.celery_app call mycallback_task --args='[1]'
"""
# Get the session id
org_session_id = Organization.query.get(organization_id).gateway_sid
# Send event and wait for response
logger.info("Sending mycallback to %s", org_session_id)
try:
result = socketio.call(
"mycallback", "Hello from the server!", to=org_session_id, timeout=10
)
except TimeoutError as e:
logger.error("mycallback timeout: %s", e)
return None
# Done
logger.info("mycallback result: %s", result)Client.py defines the socketio client, which in some parts of the code is referred to as "Gateway" # client.py
import socketio
# ...
sio = socketio.Client()
# ...
@sio.event
def connect():
"""Called when the client is connected to the server."""
logger.info("Connected to server, sid: %s", sio.sid)
@sio.event
def disconnect():
"""Called when the client is disconnected from the server."""
logger.info("Disconnected from server")
# ...
@sio.on("mycallback")
def on_callb(data):
logger.info("Received mycallback event: %s", data)
return "Hello from client!"
# ...Celery_app.py is the entrypoint for the celery worker. # celery_app.py
"""
This module is the entry point for the Celery worker
It can be run with the following command:
celery -A "app.tasks.celery_app" worker --loglevel=info
It extracts the Celery worker present in the Flask extensions,
so the Flask app factory function must add the extension via celery_init_app().
"""
from app import create_app
# ...
flask_app = create_app(main=False)
celery_app = flask_app.extensions["celery"]Logs
Here you can see the client receives the event and responds: here you can see the Flask socketio instance receives both responses from the client: Here you can see the celery worker throwing TimeoutError as it never gets the response back: I attach a screenshot of Redis Insight, showing that in the Pub/Sub pane the values are received, even though the Celery worker does not seem to get it: Thank you for your patience and thank you again for this package. Any assistance would be greatly appreciated. I look forward to finding a solution. |
Beta Was this translation helpful? Give feedback.
Answered by
miguelgrinberg
Jul 15, 2024
Replies: 1 comment 1 reply
-
|
The external process can only send, it cannot receive anything. Replies will never reach, so you will always get a timeout. If you need more flexibility you should consider using a Socket.IO client in your celery worker. |
Beta Was this translation helpful? Give feedback.
1 reply
-
|
Thank you for your prompt response. I appreciate your clarification and your suggestion, which I hadn't considered. In the end, I opted for a quite simple solution: maintaining the WebSocket connection between the client and the Flask server, and reverting to plain HTTP(s) requests from the Celery worker to retrieve data from the client via the Flask server. This approach has proven to be clean, simple, and effective. Thank you again for your time and assistance. I hope you have a great day! |
Beta Was this translation helpful? Give feedback.
Answer selected by
lucarillosi
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
The external process can only send, it cannot receive anything. Replies will never reach, so you will always get a timeout. If you need more flexibility you should consider using a Socket.IO client in your celery worker.