Flask SocketIO + external process : some messages lost

[raphael-bresson]

Hi all

First thing first muchas gracias Miguel para su increible trabajo !

My question is related to many other questions that I read but I can't find a solution for my particular case.

I'm using Flask + Flask_SocketIO + Flask_APScheduler + Redis on my server.
I want to be able to launch several background jobs on my server that can communicate their results to my front via SocketIO.
It works perfectly fine when I use a pool of thread workers.
I instantiate a second SocketIO in my tasks module, with a message queue shared with my main SocketIO(app) instance.

However, when switching to a pool of process workers (APScheduler, for some heavy work) it does not work as expected :

• I can see that all emited messages from the external process are stored in the redis database. So far so good.
• but they are not all sent to my front. Some are, others not.
• the websocket connection seems to be lost sometimes. I sometimes see 2 connections messages on my front console.
• After I've submitted one job on an external process, after modifying some server files the auto reloading feature of Flask gets stuck (ok that may be another issue) because of the process not correctly freed ?

Here is a exerpt of my server code.

Thank you for your help if possible.

app.py

from gevent import monkey
monkey.patch_all()

from flask  import              Flask, jsonify, request
from flask_socketio    import   SocketIO, send
from flask_apscheduler import   APScheduler

from src.serveur.tasks import heavy_work

app = Flask(__name__)
socketio = SocketIO(app, message_queue='redis://')
class Config:
    SCHEDULER_EXECUTORS = {"default": {"type": "threadpool", "max_workers": 20},
                           "processpool": {"type": "processpool", "max_workers": 3}}
    SCHEDULER_API_ENABLED = True

app.config.from_object(Config())
scheduler = APScheduler()
scheduler.init_app(app)
scheduler.start()

@socketio.on('connect')
def handle_connect():
    sid = request.sid
    send(f"Connexion SocketIO {sid} etablie avec succès !")


@app.route('/api/compute', methods=['POST'])
def computeRejets():
    data = request.get_json(force=True)
    username = data['username']
    site = data['site']
    rejets = data['rejets']
    sid = data['sid']
    socketio.send("compute", to=sid)
    scheduler.add_job(  'compute',
                        heavy_work,
                        args=[username, site, rejets, sid],
                        executor='processpool'
                        )
    return jsonify({"msg": "Simulations lancées."})

tasks.py

from flask_socketio    import   SocketIO, emit
socketio = SocketIO(message_queue='redis://')

def heavy_work(user, site, rejets, sid):
    time.sleep(2)
    message = f"{user} a soumis de nouveaux calculs pour le site {site}"
    socketio.send(message, to=sid)
    return message

[miguelgrinberg]

You should enable Socket.IO logs on your server and external processes to see if there are any clues there.

[raphael-bresson]

If I leave the client idle I actually have way more than 2 pings

In send
Freltmi58a-lePo3AAAC: Sending packet4 PING data None
Freltmi58a-lePo3AAAC: Received packet PONG data
In send
Freltmi58a-lePo3AAAC: Sending packet4 PING data None
Freltmi58a-lePo3AAAC: Received packet PONG data
In send
Freltmi58a-lePo3AAAC: Sending packet4 PING data None
Freltmi58a-lePo3AAAC: Received packet PONG data
In send
Freltmi58a-lePo3AAAC: Sending packet4 PING data None
Freltmi58a-lePo3AAAC: Received packet PONG data
In send
Freltmi58a-lePo3AAAC: Sending packet4 PING data None
Freltmi58a-lePo3AAAC: Received packet PONG data
In send
Freltmi58a-lePo3AAAC: Sending packet4 PING data None
Freltmi58a-lePo3AAAC: Received packet PONG data
In send
Freltmi58a-lePo3AAAC: Sending packet4 PING data None
Freltmi58a-lePo3AAAC: Received packet PONG data

(The Send is me printing in the code to understand). So you say this is not normal ?

Maybe I'm doing it all wrong, instantiating the second SocketIO instance at the top level of the module tasks.py which contains the functions to submit on external processes. But I tried to instantiate SocketIO in the functions themselves, same results. And I have several background tasks and callbacks. So should I instantiate SocketIO in each of them ?

After reading the part of you mega tutorial on background jobs, I also tried to use rq instead (what I really need is not scheduling tasks but submitting background tasks and dealing with dependency between them). And in that case SocketIO emits properly to the client with the way I'm doing things.
(But what is lacking with that solution is emitting from success/failure callbacks since their signature do not accept custom parameters like for example the room to emit to, or am I missing something ?)

Thanks

[miguelgrinberg]

This is what's wrong wrt pings.

zV_wu0aWsBajat-OAAAA: Sending packet PING data None
zV_wu0aWsBajat-OAAAA: Sending packet PING data None
zV_wu0aWsBajat-OAAAA: Received packet PONG data
zV_wu0aWsBajat-OAAAA: Received packet PONG data

There should always be one outstanding ping, never two or more.

You keep asking me about how to make this work with APScheduler and I have already told you that I do not know, and that I do not recommend APScheduler mixed with Socket.IO because that leads to problems. The Celery/RQ model makes a lot more sense for background tasks in a web application, APScheduler does not. I'm not saying that it is impossible to make it work, just that you are not the first to have problems. Given that I think it is the wrong tool for the task, I do not consider it a priority to support it.

But what is lacking with that solution is emitting from success/failure callbacks since their signature do not accept custom parameters like for example the room to emit to, or am I missing something ?

You are missing a few things, yes. First of all, you could catch all exceptions and create your own error handler within the task, without having to do it on a callback.

But if you want to use their callbacks, then functools.partial should allow you to add any custom argument that you want.

[raphael-bresson]

Ok so it seems I have something wrong in my config regardless the background tasks.
As soon the websocket connection is etablished between my client and the server, even while leaving the client idle, I get regular pings emitted as shown previously.
On the front side I use VueJS with a dev server and configured the proxy to reroute the websockets request to the Flask server. Something like this (in case you have an idea)

export default defineConfig({
  plugins: [
    vue(),
    Components({
      resolvers: [BootstrapVueNextResolver(), IconsResolve()],
      dts: true,
    }),
    Icons({
      compiler: "vue3",
      autoInstall: true,
    }),
  ],
  resolve: {
    alias: {
      "@": fileURLToPath(new URL("./src", import.meta.url)),
    },
  },
  build: { outDir: "../serveur/static" },
  server: {
    proxy: {
      "/static": { target: "http://0.0.0.0:5008", changeOrigin: true },
      "/api": { target: "http://0.0.0.0:5008", changeOrigin: true },
      "/socket.io": {
        target: "ws://localhost:5008",
        ws: true,
      },
    },
    host: true,
  },
});

[raphael-bresson]

Thank you for your last answer (that I hadn't seen when I last wrote to you, sorry).

Ok I get it regarding APScheduler. I'll try to move to RQ then.

Of course I can catch exceptions within the task as you suggest but I did not want to mess the socketio call, which I use to notify about errors that occured, with the task itself which is pure standard python and could be run outside of the webApp.
As for partials, they don't work either since the callback has to be defined at the module level, and can not be built inside a function that would provide the additional arguments (room).

As for PING what you meant is that their should never be 2 consecutive PINGS with no PONG response in between, is that right ? But the succession of PING/PONG pairs is ok ,right ?
Since, looking at engineio codebase, I see that a PING task is scheduled to run regularly.

    def _handle_connect(self, environ, start_response, transport, jsonp_index=None):
        """Handle a client connection request."""
        if self.start_service_task:
            # start the service task to monitor connected clients
            self.start_service_task = False
            self.service_task_handle = self.start_background_task(self._service_task)

        sid = self.generate_id()
        s = socket.Socket(self, sid)
        self.sockets[sid] = s

        pkt = packet.Packet(
            packet.OPEN,
            {
                "sid": sid,
                "upgrades": self._upgrades(sid, transport),
                "pingTimeout": int(self.ping_timeout * 1000),
                "pingInterval": int(
                    self.ping_interval + self.ping_interval_grace_period
                )
                * 1000,
            },
        )
        s.send(pkt)
        s.schedule_ping()

[raphael-bresson]

Yes, of course I can always call my task function from a wrapper function dedicated to exception catching and client notification.