Trouble getting real time messages with gevent & gunicorn

[N-Jay-Dee]

I want to write a program that runs a shell script on the server side and emits the output to a client in real time. I'm attempting to do so with subprocess, gevent, and gunicorn.

my event handler looks something like this:

    process = Popen(file_path, stdout=PIPE, stderr=PIPE, shell = True, universal_newlines=True)

    while True:
        out = process.stdout.readline()
        err = process.stderr.readline()

        if out == '' and err == '' and process.poll() is not None:
            break
        if out != '':
            socketio.emit('stdout_chunk', {'data': out})
        if err != '':
            socketio.emit('stderr_chunk', {'data': err})

        socketio.sleep(0)

I have a shell script that looks like this:

echo "before sleep"
sleep 10
echo "after sleep1"
sleep 10
echo "after sleep2"

And I start my server like so:

gunicorn -k geventwebsocket.gunicorn.workers.GeventWebSocketWorker -w 1 server:app

I would expect to receive the first message, wait ten seconds, receive the next, wait another ten, and then receive the last. What actually happens is 20 seconds goes by and all messages come through. This happens regardless of what I set the sleep time to (for example if sleep was 2, 20 seconds would go by and then the messages would come through 2 seconds apart). I'm sure I'm misunderstanding either something about the monkey patched subprocess or how the emits are queued but nothing I've tried seems to change this behavior.

[miguelgrinberg]

Assuming you have monkey-patched the subprocess module, and that the monkey-patched implementation in gevent does not have bugs, then I'd say the problem here has nothing to do with Socket.IO and has everything to do with buffering on the pipe streams.

What I would do is remove gevent and see if the problem persists. That should tell you if you need to look into gevent's monkey-patched subprocess module or (more likely) into buffering in your piped process.

[N-Jay-Dee]

Thank you for the swift reply, you were absolutely correct. I was attempting to use a method for real time reading of the standard subprocess module that doesn't work with the monkeypatched gevent version.

For anyone with a similar issue I was able to get it to work with the following:

def emit_stream(stream, event):

    while not stream.closed:
        message = stream.readline()

        if not message: 
            break
        
        socketio.emit(event, { 'data': message.strip() })

process = Popen(file_path, stdout=PIPE, stderr=PIPE, shell = True, universal_newlines=True)

gevent.spawn(emit_stream, process.stdout, 'stdout_chunk')
gevent.spawn(emit_stream, process.stderr, 'stderr_chunk')

Note that if you don't want one stream to wait for another you should seperate the greenlets, and that using process.poll() in your loop will cause it to block. for more details see here