Is it possible to store wsgi environ and eio packets pickled at Redis?

[zwerg44]

Hello!

I'm currently trying to configure my monolith service, using:

• nginx
• uwsgi (4 workers, one service per app in systemd)
• Flask-SocketIO (Redis broker)
• Flask-Session (Redis broker)
• gevent
• some rest api

For now, I don't want to split my socketio communication to separate service. I haven’t configured sticky sessions at nginx/uwsgi level, and I have some feeling that it is not necessary.

Now, I have issues only with polling transport. I'm getting 400 Bad requests while my polling requests hops over different workers (with 25% chance of success). I have tried to force not using polling transport at client side. And it actually looks working well.

var socket = io(
    'https://myhost.local',
    {
        transports: ['websocket'],
        upgrade: false
    }
);

So I looked through the python-socketio sources, and find out for myself, that it is statefull. server.py#L502 server.py#L527 It looks like state, that needs to be shared, is used only during the polling communications. Can it be (optionally) moved to somewhere like Redis/Memcached/Session? Does it written now like this for performance reason?

I think most clients will have ability to use websocket transport, and my service can sacrifice some performance with polling transport for having no need of sticky sessions. Or am I somewhere wrong, and I should split socketio to separate service with sticky sessions, balanced with nginx iphash?

Note: I understand that session gets forked at socket connection, and I use it only for identification/authentication. After my user logins via rest, the client will reconnet socketio with new socketio sid.

Note2: looks like at master branch of uwsgi, there is out of box support of iphash load balance, but no support of it at 2.0.17.1 subscription.c#L1396 With --subscription_algo iphash option, haven't try to compile it though.

[zwerg44]

Here is, how i use Flask-SockeIO:

import config
from flask import request, session, current_app
from flask_socketio import SocketIO, join_room, leave_room, rooms, disconnect
from db import models as m
from utils.logging import logger

# request.sid - sid of socket
# session.sid - sid of flask-session
# session.get('session_id') - id of auth session

def create_socket_app(app):
    socketio = SocketIO(
        app,
        message_queue=config.REDIS_WEBSOCKETS,
        manage_session=False,
    )

    @socketio.on('connect')
    def on_connect():
        logger.debug('Client connected %s %s %s' % (request.sid, session.sid, session))
        session_id = session.get('session_id')
        user_id = session.get('user_id')
        if not session_id or not user_id:
            disconnect(request.sid)
        user = m.User.query.get(user_id) if user_id else None
        company_id = user.company_id if user else None
        __join_rooms(request.sid, session_id, user_id, company_id)

    @socketio.on('disconnect')
    def on_disconnect():
        logger.debug('Client disconnected %s %s %s' % (request.sid, session.sid, session))

    @socketio.on('ping')
    def on_ping():
        user_id = session.get('user_id')
        notify_user(user_id, 'pong', {'socket_sid': request.sid, 'session_sid': session.sid})

    return socketio

def __join_rooms(socket_sid, session_id, user_id, company_id):
    if session_id:
        join_room('session_%s' % session_id, sid=socket_sid)
    if user_id:
        join_room('user_%s' % user_id, sid=socket_sid)
    if company_id:
        join_room('company_%s' % company_id, sid=socket_sid)

def __leave_rooms(socket_sid):
    _rooms = rooms(sid=socket_sid)
    for r in _rooms:
        if r.startswith('session_') or r.startswith('user_') or r.startswith('company_'):
            leave_room(r, sid=socket_sid)

def notify_sid(sid, event, data):
    current_app.socket_app.emit(event, data, room=sid)

def notify_session(session_id, event, data):
    current_app.socket_app.emit(event, data, room='session_%s' % session_id)

def notify_user(user_id, event, data):
    current_app.socket_app.emit(event, data, room='user_%s' % user_id)

def notify_company(company_id, event, data):
    current_app.socket_app.emit(event, data, room='company_%s' % company_id)

[miguelgrinberg]

Can it be (optionally) moved to somewhere like Redis/Memcached/Session? Does it written now like this for performance reason?

Mainly for performance reasons, yes. The state kept for each client is not only data, there are also queues and background tasks associated with clients, and these are harder to share.

May I ask what is the reason you don't want to turn on sticky sessions in your nginx config? That's all you need to do.

[zwerg44]

Thank you for so quick answer.

I am just trying to understand the right way, and some puzzle in the picture looks missing)

Before implementing socketio in my service, I was using gunicorn, but its multiworker configuration is not compatible with socketio, now I trying to move to uwsgi.

I looked through the example of Flask-SocketIO ngix config. There are 3 separate services at different ports, I suppose uwsgi services, and the rest api is routed to the first instance.

So:

• I don't understand why I should pick uwsgi then, if I have separate services (1 worker each) balanced with nginx
• it is more difficult to manage multiple systemd services, suppose I should read more about emperor uwsgi config
• multiple services will consume a bit more resources
• I suppose rest api should be also balanced with iphash, and I'm not sure about the performance then
• and of course load balancing without sticky sessions is more fair
• my service is written functional way and is, more or less stateless, everything else is at database or redis, and it just feels nicer to keep it more stateless)

[zwerg44]

If that is useful feature, and if it sounds for you that it wouldn’t break something internal in python-socketio, I could try to fork and PR then. I just wanted to consult, is it possible and does it make sense to do this.

[miguelgrinberg]

@zwerg44 I think it might be possible, but it is not an easy task and very likely have an effect on performance. You will need to use a database, maybe Redis or similar, to hold the state and to manage the send queue for each client. Most (if not all) of this change will go in the python-engineio package, which is the lower level protocol on top of which Socket.IO is built.

I would be open to accept it as a contribution as long as it is optional and the default remains the same.

I don't understand why I should pick uwsgi then, if I have separate services (1 worker each) balanced with nginx

I don't think you should pick uwsgi, both uwsgi and gunicorn are valid choices and have the same sticky session constraints. If uwsgi starts supporting sticky sessions then maybe it will have an advantage over gunicorn, though I believe gunicorn is also looking at implementing sticky sessions, per my request.

it is more difficult to manage multiple systemd services, suppose I should read more about emperor uwsgi config

Sure, there is a bit of an increase in maintenance, but consider that this allows you to scale beyond a single server.

I suppose rest api should be also balanced with iphash, and I'm not sure about the performance then

This is incorrect. You can have the ip_hash rule apply only to your Socket.IO connections in nginx, while leaving other endpoints free of it. Nginx allows you to create configurations based on the path, so you can add ip_hash only for /socket.io/ endpoints

[zwerg44]

I have tried to implement this using redis. The current solution looks working, but there is still much stuff to fix, clean up, and discuss. I haven't looked at the unit-tests and haven't compared performance yet. I think, I will do clean up during the week and prepare the PR then.

I can confirm that with my described configuration, this is working:

• I've added 2 options state_storage, socket_queue_storage
• without options set, the default remains the same
• with options set, polling transport works
• with options set, websocket transport works
• with options set, transport upgrade polling=>websocket works

Here is the affected code: python-engineio python-socketio Flask-SocketIO

@miguelgrinberg Please look through my changes, and comment which are reasonable and which are not.

[miguelgrinberg]

The two main problems I see:

• the environ dictionary isn't always pickle friendly. Sometimes there are objects there, or file handles. I don't have a solution for this problem, unfortunately. Maybe it is necessary to review all these objects and see which ones are needed and which can be removed from the dict.
• I'm not sure the two concurrent long-polling requests are safe from race conditions. Both will be accessing the state, potentially at about the same time.

Going into the smaller issues:

• I'll have to think about how to consolidate this with the existing support for message queues. It is odd that you have to provide the queue URL three times for three different things.
• It's also odd to only support Redis, when existing queue support works on many other queues through Kombu.
• I think you did not create an asyncio compatible redis dictionary.

There's probably more things to note, but I'll have to install the code and run it to get a more complete list. In any case, I'm surprised that it isn't as big a change as I expected!

[zwerg44]

the environ dictionary isn't always pickle friendly. Sometimes there are objects there, or file handles. I don't have a solution for this problem, unfortunately. Maybe it is necessary to review all these objects and see which ones are needed and which can be removed from the dict.

I'm not sure, but I suppose that unpicklable objects in the environ are always the same for all requests to the same flask instance so, I temporary saved the last appeared instances of unpicklable stuff to the state of socketio __last_environ. Maybe that environ parts, shared between different request, can be setted up during initialisation.

I'm not sure the two concurrent long-polling requests are safe from race conditions. Both will be accessing the state, potentially at about the same time.

I'm not sure here too, I think redis itself is single threaded, so by default it will act the same as GIL in python.

I'll have to think about how to consolidate this with the existing support for message queues. It is odd that you have to provide the queue URL three times for three different things.

This is a temporary decision, I separated them to be sure, I'm not affecting the current pubsub at redis manager. At first I thought redis based queue would also require some pubsub.