Python Flask Server: AssertionError("write() before start_response()")

Question

I found the work of Miguel Grinberg very inspiring and I chose, for a University project, to use Flask along with the FlaskSocketIO library to make some Java instrumentation (pretty complicated stuff but for the curious I'm working on this project).

Basically the python part is just creating a command, executing it in a background thread and pushing every log to the screen, thanks to this great Flask SocketIO library.

All what is described works pretty good BUT I have an issue when I want to relaunch the task by calling a second time the same URL, I get this error and I found no thread on it so I assume it is trivial or very nasty (but nothing in the middle!). Does someone have an idea?

File "/Library/Python/2.7/site-packages/eventlet/wsgi.py", line 485, in handle_one_response
write(b'')
File "/Library/Python/2.7/site-packages/eventlet/wsgi.py", line 380, in write
raise AssertionError("write() before start_response()")
AssertionError: write() before start_response()

And, of course, here is my code (inspired from the example of Miguel because I was unable to set it up myself...)

#!/usr/bin/env python

async_mode = None

if async_mode is None:
    try:
        import eventlet
        async_mode = 'eventlet'
    except ImportError:
        pass

    if async_mode is None:
        try:
            from gevent import monkey
            async_mode = 'gevent'
        except ImportError:
            pass

    if async_mode is None:
        async_mode = 'threading'

    print('async_mode is ' + async_mode)

# monkey patching is necessary because this application uses a background
# thread
if async_mode == 'eventlet':
    import eventlet
    eventlet.monkey_patch()
elif async_mode == 'gevent':
    from gevent import monkey
    monkey.patch_all()

import time
from threading import Thread
import subprocess
from os import chdir

from flask import Flask, render_template
from flask_socketio import SocketIO, emit

import InstrumentationScripts as IS


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


def background_thread():
    time.sleep(1)
    socketio.emit('my response',
                  {'data': "Thread Started", 'count': 0},
                  namespace='/test')
    cb = IS.CommandBuilder()
    args = cb.createCommand().split()
    chdir(cb.InstrumentationPepDirectory)
    process = subprocess.Popen(args, stdout=subprocess.PIPE)
    for out in iter(process.stdout.readline, b""):
        out = '<pre>' + out + '</pre>'
        socketio.emit('my response', {'data': out, 'count':0}, namespace='/test')
        time.sleep(0.001)
    socketio.emit('my response',
                  {'data': "Thread Finished", 'count': 0},
                  namespace='/test')


@app.route('/')
def index():
    global thread
    if thread is None:
        thread = Thread(target=background_thread)
        thread.daemon = True
        thread.start()
    return render_template('index.html')


@socketio.on('connect', namespace='/test')
def test_connect():
    emit('my response', {'data': 'Connected', 'count': 0})


if __name__ == '__main__':
    socketio.run(app)

In case you need to have the whole code, here is the repo

Answer 1

So, thanks to Miguel comments, I came up with a solution and I think it can be reused for other similar problems:

I created a Button to start the process. This Button sends via WebSocket value to the server (a Boolean). When the server receives this signal, it starts the thread and adds a True value to a global Dictionary which stores all running threads (with their status: running (True) or not (False)).

To distinguish all sessions, I use this request.sid as mentioned in the example (be carefull, this request.sid is not retrievable from the not socketio methods (at least, I didn't success)).

To have a better understanding, the dictionary looks like something like this:

dict = {"sessionId1":True, "sessionId2": False}

It indicates that the thread of session 1 is running and the one of session 2 not (pretty obvious, right?).

Then, I pass the request.sid to the Thread as argument in order that it can check if the Thread has been cut from outside. Then, in the loop where I display the inputs, I check for the Boolean and if it's set to False I cut the process and the thread (by killing the process and ending the method).

So that was the theory, here is the working code:

#!/usr/bin/env python

async_mode = None

if async_mode is None:
    try:
        import eventlet
        async_mode = 'eventlet'
    except ImportError:
        pass

    if async_mode is None:
        try:
            from gevent import monkey
            async_mode = 'gevent'
        except ImportError:
            pass

    if async_mode is None:
        async_mode = 'threading'

    print('async_mode is ' + async_mode)

# monkey patching is necessary because this application uses a background
# thread
if async_mode == 'eventlet':
    import eventlet
    eventlet.monkey_patch()
elif async_mode == 'gevent':
    from gevent import monkey
    monkey.patch_all()

import time
from threading import Thread
import subprocess
from os import chdir
import os
import signal

from flask import Flask, render_template, send_file, request, session
from flask_socketio import SocketIO, emit

import InstrumentationScripts as IS


#TODO Emit only to the client who asked for instrumentation

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

thread_map = {}


def background_thread(sid):
    time.sleep(1)
    socketio.emit('my response',
                  {'data': "Thread Started", 'count': 0},
                  namespace='/test')
    cb = IS.CommandBuilder()
    args = cb.createCommand().split()
    chdir(cb.InstrumentationPepDirectory)
    process = subprocess.Popen(args, stdout=subprocess.PIPE)
    for out in iter(process.stdout.readline, b""):
        if (thread_map[sid]):
            out = '<pre>' + out + '</pre>'
            socketio.emit('my response', {'data': out, 'count':0}, namespace='/test')
            time.sleep(0.001)
        else:
            os.kill(process.pid, signal.SIGUSR1)
            print 'Java Process was killed'
            break
    socketio.emit('my response',
                  {'data': "Thread Finished", 'count': 0},
                  namespace='/test')
    thread_map[sid] = False



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


@socketio.on('connect', namespace='/test')
def test_connect():
    emit('my response', {'data': 'Connected\n', 'count': 0})

@socketio.on('disconnect', namespace='/test')
def test_disconnect():
    thread_map[request.sid] = False
    print('Client disconnected')

@socketio.on('startInstrumentation', namespace='/test')
def test_start_stop(message):
    if message['data']:
        thread = Thread(target=background_thread, args=(request.sid,))
        thread.daemon = True
        thread.setName(request.sid)
        thread_map[request.sid] = True
        thread.start()
    else:
        thread_map[request.sid] = False

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

And here the minimal HTML:

<!DOCTYPE HTML>
<html>
<head>
    <title>Flask-SocketIO Test</title>
    <script type="text/javascript" src="https://code.jquery.com/jquery-2.1.4.min.js"></script>
    <script type="text/javascript" src="https://cdn.socket.io/socket.io-1.3.7.js"></script>
    <script type="text/javascript" charset="utf-8">
        $(document).ready(function(){
            namespace = '/test'; // change to an empty string to use the global namespace

            // the socket.io documentation recommends sending an explicit package upon connection
            // this is specially important when using the global namespace
            var socket = io.connect('http://' + document.domain + ':' + location.port + namespace);

            // event handler for server sent data
            // the data is displayed in the "Received" section of the page
            socket.on('my response', function(msg) {
                var log = $('#log');
                log.append(msg.data);
                window.scrollTo(0,document.body.scrollHeight);
            });

            // event handler for new connections
            socket.on('connect', function() {
                socket.emit('my event', {data: 'I\'m connected!'});
            });
            var clicked = false;
           $('form#go').submit(function(event) {
               clicked = !clicked;
                socket.emit('startInstrumentation',  {data: clicked});
               if (clicked){
                   while ($('#log').firstChild) {
                       $('#log').removeChild($('#log').firstChild);
                   }
               }
                return false;
            });
        });
    </script>
</head>
<body>
    <h2>Receive:</h2>
    <form id="go" method="POST" action='#'>
        <input type="submit" value="Go!">
    </form>
    <div id="log"></div>
</body>
</html>

Hope it helps.