Flask-socketio: I am confuse using asynchronous socketio

[FahmiAyil]

I have a complex calculation in my program, iam using socket io to display the results to webpage, the results are continously until the condition is done.
How can i do that?

Maybe if i give some example is i have infinity loop to generate number, how can i display the results continously on webpage?
Thanks

[mrwillis]

A common way to do this is as per here in a background thread if you look at this these lines:

app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret!'
socketio = SocketIO(app, async_mode=async_mode)
thread = None
thread_lock = Lock()

def background_thread():
    """Example of how to send server generated events to clients."""
    count = 0
    while True:
        socketio.sleep(10)
        count += 1
        socketio.emit('my_response',
                      {'data': 'Server generated event', 'count': count},
                      namespace='/test')

With this style, when you create the thread in the below method, it's important to use a lock so that multiple threads are not created:

@utils.authenticated_only
@socketio.on('connect')
def emit_market_data():
    global thread
    with thread_lock:
        if thread is None:
            thread = socketio.start_background_task(target=background_thread)

You can also choose to use a message_queue as in the docs and instead emit from an external process. Hope this helps.

[felixhao28]

How does this work when multiple connections are active?

Will all active clients receive the results from the background task started by the first client?

Or will the clients other than the first simply receive nothing?

Or (what I actually want) every client gets a background thread to do its work and receive results separately.

Instead of a single global thread lock, is there a built-in thread pool to avoid too many jobs running at the same time?

[miguelgrinberg]

Will all active clients receive the results from the background task started by the first client?

It depends on how you implement your thread. If you make it broadcast then yes, if you make it address a specific client, then no.

Or (what I actually want) every client gets a background thread to do its work and receive results separately.

This is different. For this you need one task per client, not one for the whole server.

Instead of a single global thread lock, is there a built-in thread pool to avoid too many jobs running at the same time?

You need to use the facilities of the asynchronous framework that you are using for that. Both eventlet and gevent have thread pools. Check out their documentation.

[felixhao28]

Thank you for the quick response!

If you make it broadcast then yes, if you make it address a specific client, then no.

I basically copied the code above and the log says emitting event "json" to all [/]. After testing, I can confirm that other active clients are receiving the broadcast message. So I think I need to somehow call the context-aware emit function instead of socketio.emit. I don't see anything helpful in the document. Any hint where to start?

[miguelgrinberg]

To address a specific client you need to add a room argument set to the sid of the addressee.

[felixhao28]

Ah yes. You are very helpful.

I am going to paste my solution here for anyone who googled to here.

# What I expected to work
@socketio.on('json')
def handle_json(json):
    job_done = False
    results = None
    while not job_done:
        # do_heavy_job takes ~ 1.0 seconds
        job_done, results = do_heavy_job(json['X1'], json['X2'], continue_from=results)
        emit('json', results)
    emit('json', {'done': True})


# Should be written like this
def bg_job(sid, X1, X2):
    job_done = False
    results = None
    while not job_done:
        # do_heavy_job takes ~ 1.0 seconds
        job_done, results = do_heavy_job(X1, X2, continue_from=results)
        socketio.emit('json', results, room=sid)
        eventlet.sleep() # this is somehow required to flush the results, which is unintuitive
    socketio.emit('json', {'done': True}, room=sid)


@socketio.on('json')
def handle_json(json):
    sid = flask.request.sid
    thread = socketio.start_background_task(bg_job, sid, json['X1'], json['X2'])

I think an async multi-threaded service is a very common use case for websocket. The implementation I just made works but is not very straightforward. So either I have done it sub-optimally or there is space for improvements on the API designs or at least documentation.

[miguelgrinberg]

The sleep() call is the standard way in which you release the CPU in Python asynchronous programs. You may be used to traditional multi-tasking, in which the OS does all the hard work. When using async, the CPU is not managed by the OS, each task needs to explicitly release it to give other tasks a chance to run.

What does the do_heavy_job() function do? How long does a background task run for? Not completely sure you need it, your original version could probably work as well, if you add the sleep() call.

[felixhao28]

Yes you are right. I don't need the background job. A sleep() is all I need. do_heavy_job() calls from the python hook of tensorflow into some native code. It is a synchronized call.

Then what does start_background_task really do?

[miguelgrinberg]

it runs the function as a separate task, but as long as you sleep often that may not be necessary. Depends on the situation.

[naveenreddysg]

Iam trying to implement socketio chat app, whichi is better practice to save chat history in database