Flask-socketio: Emits are not sent before parent function return

[ypperlig]

Hi! :-)
It is very possible that I am doing something wrong, but I've tried numerous options and I am still unable to make it function properly.

Version: 2.7.1
Async mode: eventlet
socket.io version: 1.4.8 (https://cdnjs.cloudflare.com/ajax/libs/socket.io/1.4.8/socket.io.min.js)

In 5% of cases it seems to work correctly and I am receiving the emits on client side in real time.

Output with debug enabled:

127.0.0.1 - - [06/Oct/2016 21:59:38] "GET /socket.io/?EIO=3&transport=polling&t=LURb6uy&sid=7a4d7de7d49145ddad8195f807f5d1ec HTTP/1.1" 200 247 0.000201
7a4d7de7d49145ddad8195f807f5d1ec: Sending packet MESSAGE data 2/namespace,["update",{"eta":"N/A","percentage":"  0.0","download":true}]
7a4d7de7d49145ddad8195f807f5d1ec: Sending packet MESSAGE data 2/namespace,["update",{"eta":"N/A","percentage":"  0.1","download":true}]
7a4d7de7d49145ddad8195f807f5d1ec: Sending packet MESSAGE data 2/namespace,["update",{"eta":"N/A","percentage":"  0.2","download":true}]
7a4d7de7d49145ddad8195f807f5d1ec: Sending packet MESSAGE data 2/namespace,["update",{"eta":"N/A","percentage":"  0.4","download":true}]
7a4d7de7d49145ddad8195f807f5d1ec: Sending packet MESSAGE data 2/namespace,["update",{"eta":"N/A","percentage":"  0.9","download":true}]
7a4d7de7d49145ddad8195f807f5d1ec: Sending packet MESSAGE data 2/namespace,["update",{"eta":"N/A","percentage":"  1.8","download":true}]
7a4d7de7d49145ddad8195f807f5d1ec: Sending packet MESSAGE data 2/namespace,["update",{"eta":"N/A","percentage":"  3.6","download":true}]
7a4d7de7d49145ddad8195f807f5d1ec: Sending packet MESSAGE data 2/namespace,["update",{"eta":"N/A","percentage":"  7.2","download":true}]
7a4d7de7d49145ddad8195f807f5d1ec: Sending packet MESSAGE data 2/namespace,["update",{"eta":"N/A","percentage":" 14.4","download":true}]
7a4d7de7d49145ddad8195f807f5d1ec: Sending packet MESSAGE data 2/namespace,["update",{"eta":"N/A","percentage":" 28.7","download":true}]
7a4d7de7d49145ddad8195f807f5d1ec: Sending packet MESSAGE data 2/namespace,["update",{"eta":"N/A","percentage":" 57.5","download":true}]
7a4d7de7d49145ddad8195f807f5d1ec: Sending packet MESSAGE data 2/namespace,["update",{"eta":"N/A","percentage":"100.0","download":true}]
7a4d7de7d49145ddad8195f807f5d1ec: Sending packet MESSAGE data 2/namespace,["update",{"progress":{"seconds":0,"percentage":"0.0"}}]
7a4d7de7d49145ddad8195f807f5d1ec: Sending packet MESSAGE data 2/namespace,["update",{"progress":{"seconds":10,"percentage":"4.5662100456621"}}]
7a4d7de7d49145ddad8195f807f5d1ec: Sending packet MESSAGE data 2/namespace,["update",{"progress":{"seconds":20,"percentage":"9.1324200913242"}}]
7a4d7de7d49145ddad8195f807f5d1ec: Sending packet MESSAGE data 2/namespace,["update",{"progress":{"seconds":30,"percentage":"13.698630136986301"}}]
7a4d7de7d49145ddad8195f807f5d1ec: Sending packet MESSAGE data 2/namespace,["update",{"progress":{"seconds":40,"percentage":"18.2648401826484"}}]
7a4d7de7d49145ddad8195f807f5d1ec: Sending packet MESSAGE data 2/namespace,["update",{"progress":{"seconds":50,"percentage":"22.831050228310502"}}]
7a4d7de7d49145ddad8195f807f5d1ec: Sending packet MESSAGE data 2/namespace,["update",{"progress":{"seconds":60,"percentage":"27.397260273972602"}}]
7a4d7de7d49145ddad8195f807f5d1ec: Sending packet MESSAGE data 2/namespace,["update",{"progress":{"seconds":70,"percentage":"31.963470319634702"}}]
7a4d7de7d49145ddad8195f807f5d1ec: Sending packet MESSAGE data 2/namespace,["update",{"progress":{"seconds":80,"percentage":"36.5296803652968"}}]
7a4d7de7d49145ddad8195f807f5d1ec: Sending packet MESSAGE data 2/namespace,["update",{"progress":{"seconds":90,"percentage":"41.0958904109589"}}]
7a4d7de7d49145ddad8195f807f5d1ec: Sending packet MESSAGE data 2/namespace,["update",{"progress":{"seconds":100,"percentage":"45.662100456621005"}}]
7a4d7de7d49145ddad8195f807f5d1ec: Sending packet MESSAGE data 2/namespace,["update",{"progress":{"seconds":110,"percentage":"50.2283105022831"}}]
7a4d7de7d49145ddad8195f807f5d1ec: Sending packet MESSAGE data 2/namespace,["update",{"progress":{"seconds":120,"percentage":"54.794520547945204"}}]
7a4d7de7d49145ddad8195f807f5d1ec: Sending packet MESSAGE data 2/namespace,["update",{"progress":{"seconds":130,"percentage":"59.3607305936073"}}]
7a4d7de7d49145ddad8195f807f5d1ec: Sending packet MESSAGE data 2/namespace,["update",{"progress":{"seconds":140,"percentage":"63.926940639269404"}}]
7a4d7de7d49145ddad8195f807f5d1ec: Sending packet MESSAGE data 2/namespace,["update",{"progress":{"seconds":150,"percentage":"68.4931506849315"}}]
7a4d7de7d49145ddad8195f807f5d1ec: Sending packet MESSAGE data 2/namespace,["update",{"progress":{"seconds":160,"percentage":"73.0593607305936"}}]
7a4d7de7d49145ddad8195f807f5d1ec: Sending packet MESSAGE data 2/namespace,["update",{"progress":{"seconds":170,"percentage":"77.6255707762557"}}]
7a4d7de7d49145ddad8195f807f5d1ec: Sending packet MESSAGE data 2/namespace,["update",{"progress":{"seconds":180,"percentage":"82.1917808219178"}}]
7a4d7de7d49145ddad8195f807f5d1ec: Sending packet MESSAGE data 2/namespace,["update",{"progress":{"seconds":190,"percentage":"86.7579908675799"}}]
7a4d7de7d49145ddad8195f807f5d1ec: Sending packet MESSAGE data 2/namespace,["update",{"progress":{"seconds":200,"percentage":"91.32420091324201"}}]
7a4d7de7d49145ddad8195f807f5d1ec: Sending packet MESSAGE data 2/namespace,["update",{"progress":{"seconds":210,"percentage":"95.8904109589041"}}]

One function that is using emit:

def download_status(self, obj):
      push_obj = {'download': True, 'percentage': None, 'eta': None}
      socketio.emit('update', push_obj, namespace='/namespace')

JavaScript:

        var socket = io.connect('http://' + document.domain + ':' + location.port + '/namespace');
        var run_once = 0;
        socket.on('connect', function() {
            if (run_once === 0) {
                run_once = 1;
                init();
            }
        });

        socket.on('test', function(data) {
            console.log(timestamp(), data);
        });

        socket.on('update', function(data) {
            console.log(timestamp(), data);
        });
        socket.on('error', function(data) {
            console.log(timestamp(), data);
        });

        socket.on('success', function(data) {
            console.log(timestamp(), data);
            socket.disconnect();
        });

        socket.on('disconnect', function(data) {
            console.log(timestamp(), "Socket disconnected");
        });

Screenshot of console in Chrome:

Thank you for any help with this!

[miguelgrinberg]

Do those delayed messages ever arrive? How long does it take worst case?

It may be possible that you have the expectation that messages are sent synchronously. This is actually not the way it works, when you call emit the message goes into a queue that is serviced by a background thread. Your function needs to end or yield, so that other threads get a chance to do what they need to do and get that message sent. Depending on how busy things are you may see a variation in delivery times.

[ypperlig]

Hi,
Thanks for the quick response!

Seems like I have the misconception that messages should be sent synchronously.

The goal we want to accomplish with the library is to provide a progressbar client side when everything is being processed in the backend.

As for when they are usually sent, they are sent on the parent function return.

The parent function looks like this:

@socketio.on('identify', namespace='/namespace')
def app_socket(data):
        g.request_id = str(uuid4())
        return emit('success', {'success': class(params).function()}

This is all currently running on localhost and only one request running.

Our project usually takes 1-2 mins to complete, but can also take up to 20 minutes (Due to download/upload speed).

Sometimes the queue is sent out in real time, but in 90% of cases it returns the entire queue at the end which is what is puzzling me. I assume this depends on how busy things are as you mention above?

[miguelgrinberg]

Okay, great. I don't think this is a busy server problem, I actually think I have a solution for you. :)

If you want to make sure pending events are flushed to the client, what you can do is add a socketio.sleep(0) call at the appropriate place. This allows other threads to run. You can try adding it right after each emit call if you like. If you don't see a big performance drop (I doubt you will) then that's a perfectly valid solution.

The reason the sleep call helps is that when you use eventlet or gevent the thread switches are not automatic, they only happen when you make certain calls, such as anything to do with network I/O. The sleep(0) call is the standard way to force a task switch in these frameworks.

[ypperlig]

Ah, alright. I did previously use eventlet.sleep(0) after using emit (After reading through a couple of earlier issues), but that didn't seem to do the trick. The issue also still seems to be present with socketio.sleep(0) (I assume it uses the sleep option in the framework that is being used?).

Screenshot of console in Chrome:

And the console from debug:

app.py:

import eventlet, datetime, time, tempfile, os, requests
from flask import Flask, Blueprint, render_template
from flask_socketio import SocketIO, emit
app = Flask(__name__, template_folder="templates")
socketio = SocketIO(app)

def download_example():
    url = 'http://speedtest.london.linode.com/100MB-london.bin'
    req = requests.get(url, stream=True)

    with tempfile.TemporaryDirectory() as tmpdirname:
        filepath = os.path.join(tmpdirname, url.split('/')[-1])

        with open(filepath, 'wb') as file:
            for chunk in req.iter_content(chunk_size=1024):
                if chunk:
                    file.write(chunk)

        return '%s: Downloaded file with a size of: %s' % (timestamp(), str(os.path.getsize(filepath) >> 20) + 'MB')


def timestamp():
    return datetime.datetime.fromtimestamp(time.time()).strftime('%Y-%m-%d %H:%M:%S')

@app.route('/')
def hello():
    return render_template("index.html")

@socketio.on('test', namespace='/namespace')
def echo_socket(data):
    print('%s: Initialized' % timestamp())
    emit('test', 'Wobalabadubdub', namespace='/namespace')
    socketio.sleep(0)
    print('%s: This should theoretically not happen at the same time as the second emit?' % timestamp())
    emit('test', download_example(), namespace='/namespace')
    socketio.sleep(0)
    print('%s: Second emit finished' % timestamp())

    return 'foo'

if __name__ == "__main__":
    socketio.run(app, debug=True)

templates/index.html:

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Test</title>
    <script type="text/javascript" src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/1.4.8/socket.io.min.js"></script>
</head>
<body>
    <script type="text/javascript">

        function timestamp() {
            var time = new Date();

            var hours = time.getHours() < 10 ? "0" + time.getHours() : time.getHours();
            var minutes = time.getMinutes() < 10 ? "0" + time.getMinutes() : time.getMinutes();
            var seconds = time.getSeconds() < 10 ? "0" + time.getSeconds() : time.getSeconds();

            return hours + ':' + minutes + ':' + seconds
        }

        function init() {
            socket.emit('test', 'hello');
        }
        var socket = io.connect('http://' + document.domain + ':' + location.port + '/namespace');
        var run_once = 0;
        socket.on('connect', function() {
            if (run_once === 0) {
                run_once = 1;
                init();
            }
        });

        socket.on('test', function(data) {
            console.log(timestamp(), data);
        });
        socket.on('disconnect', function(data) {
            console.log(timestamp(), "Socket disconnected");
        });
    </script>
</body>
</html>

[ypperlig]

Hi,
I added socketio.sleep(0) inside the loop and it now seems to work as expected :-)

        with open(filepath, 'wb') as file:
            for chunk in req.iter_content(chunk_size=1024):
                if chunk:
                    file.write(chunk)
                    emit('test', 'test', namespace='/namespace') 

Thanks for your help!