Client doesn't receive messages from server when sent from separate thread

[Heanthor]

Hello,

I've done a lot of digging on the issue of using different threads with Flask-SocketIO. I feel like my problem is incredibly simple, although it's possible I'm missing something conceptually important.

I'm trying to send a message to the client using a background thread, which reads from an event queue, like this:

@app.route("/all_sims/", methods=["GET"])
def all_sims():
   ...

   t = threading.Thread(target=check_sim_status, args=(sb.event_queue,), daemon=True)
   t.start()

# thread target
def check_sim_status(queue):
    while True:
        message = queue.get()
        print("MESSAGE IN SITE: " + str(message))
        # with app.app_context():
        socketio.emit("progressbar", json.dumps(message))
        socketio.sleep(0)

        queue.task_done()

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

Socket.io seems to ignore the emit call, no debug line is printed and the program goes on its way. Other calls within decorators communicate with the client site just fine.

Originally I was just using standard python threads for this event queue, but other posts (such as miguelgrinberg/python-socketio#16) talked about monkey patching with eventlet, so I tried that as well. I got the infinite recursion bug (as seen here eventlet/eventlet#371), so I couldn't proceed.

Another post recommended adding socketio.sleep(0) after my emit, but the message does not show up on the client, and it additionally throws a greenlet exception:

Traceback (most recent call last):
  File "C:\Users\reedt\AppData\Local\Programs\Python\Python36\lib\site-packages\eventlet\queue.py", line 118, in switch
    self.greenlet.switch(value)
greenlet.error: cannot switch to a different thread

Is there something obvious I'm doing wrong? I don't have any performance requirements right now, I just want to send a message from a background thread.

[miguelgrinberg]

greenlet.error: cannot switch to a different thread

This happens because you are mixing greenlets with threads. You need to follow that advice on monkey patching, or else use the eventlet API to start your background tasks, so that they are also greenlets. Regarding the infinite recursion error, have you tried downgrading eventlet to 0.17.4 like others have done?

Lastly, if you don't see the MESSAGE IN SITE: in your console, then that could be an indication of yet another problem. You should add more prints, like one before the queue.get() call, to check if the thread is even starting.

[Heanthor]

Thanks for the quick response. I do see the MESSAGE IN SITE: line, so the thread is running properly, at least.

I tried spawning an eventlet thread like

eventlet.spawn(check_sim_status, sb.event_queue)

and now the MESSAGE IN SITE: line is not printed. I tried monkey patching in addition, and now I'm seeing exceptions in the main thread:

Exception in thread Thread-4:
Traceback (most recent call last):
  File "C:\Users\reedt\AppData\Local\Programs\Python\Python36\lib\threading.py", line 916, in _bootstrap_inner
    self.run()
  File "C:\Users\reedt\AppData\Local\Programs\Python\Python36\lib\threading.py", line 864, in run
    self._target(*self._args, **self._kwargs)
  File "C:\Users\reedt\AppData\Local\Programs\Python\Python36\lib\multiprocessing\pool.py", line 365, in _handle_workers
    while thread._state == RUN or (pool._cache and thread._state != TERMINATE):
AttributeError: '_MainThread' object has no attribute '_state'

Exception in thread Thread-5:
Traceback (most recent call last):
  File "C:\Users\reedt\AppData\Local\Programs\Python\Python36\lib\multiprocessing\pool.py", line 381, in _handle_tasks
    if thread._state:
AttributeError: '_MainThread' object has no attribute '_state'

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "C:\Users\reedt\AppData\Local\Programs\Python\Python36\lib\threading.py", line 916, in _bootstrap_inner
    self.run()
  File "C:\Users\reedt\AppData\Local\Programs\Python\Python36\lib\threading.py", line 864, in run
    self._target(*self._args, **self._kwargs)
  File "C:\Users\reedt\AppData\Local\Programs\Python\Python36\lib\multiprocessing\pool.py", line 401, in _handle_tasks
    cache[job]._set(ind + 1, (False, ex))
TypeError: unsupported operand type(s) for +: 'NoneType' and 'int'

It occurred to me that I'm using the multiprocessing module somewhere in my program, with multiprocessing.ThreadPool, apply_async and subprocess.check_output. I know this isn't the place to ask questions about how eventlet interacts with these modules, but this monkey_patch seems to be less magic than I've read.

Now, other requests calls are blocking indefinitely, maybe I've run out of ports? Going to restart..

[Heanthor]

This behavior is so strange. Without monkey patching, my main thread hangs on the request if I launch my background thread with eventlet.spawn. It doesn't hang if I use the standard threading module. The background thread is not interacting with these requests at all. This downgraded version doesn't seem usable.

Jeez. I finally just uninstalled eventlet and gevent, and now my program is working perfectly well (so, without Websockets?). The performance is definitely slow.

[miguelgrinberg]

and now my program is working perfectly well.

But you lost the ability to use WebSocket.

I think the problem is that you are expecting a multi-threaded app to translate automatically to eventlet, and it isn't that easy. There are many functions from the standard library that aren't compatible, you are probably using some of those. Monkey patching makes most of the library compatible, but there are still some things that eventlet cannot do.

The multiprocessing stuff that you are apparently using is definitely a suspect. I think if/when you decide to switch to eventlet, you have to convert that into something that is more appropriate for an async framework.

[Heanthor]

This is definitely true. I just don't understand why importing and using eventlet (without monkey patching) on a background thread that just interacts with a queue would change how the parts of my program function that do not interact with eventlet at all.

I reinstalled eventlet to the latest version and spawned my background thread using eventlet.spawn; nothing broke and the debug line gets printed from the BG thread, but again, no messages are sent to the server. I feel like I've come full circle, what would cause socketio to silently skip this emit? The code that I'm running is almost entirely in my first post, it's not any more complicated than that. The message is sent when I'm not using eventlet, and is skipped when I am using it.

[Heanthor]

I'm going to do some more work porting some of my underlying code over to eventlets once I do some more reading on them, now that I know I have to go all-or-nothing with this library. I'm pretty sure my previous comment is false, that code is not working properly just replacing the call to thread with eventlet.

[miguelgrinberg]

Well, for starters, eventlet is a single-threaded framework. Anything that interacts in any way with eventlet must be all on the same thread. You can use other threads, I guess, but there must be no interaction between the eventlet thread (usually the main thread) and your other threads, unless you manage that interaction exclusively through eventlet.

The reason for all these conditions is that for eventlet to function only non-blocking functions must be used. A lot of the standard library functions are blocking, so any blocking call that you make will prevent eventlet from running. And this, can be the reason why your emits are not working.

[Heanthor]

I don't have any terribly blocking I/O on my main thread. My program uses subprocess and thread pools to do expensive work in another program, and wait on the results, but this processing does not touch socket.io.

Now, I've stripped out all manually created threads and eventlet entirely from my reporter. It now calls the socketio.emit function directly, I've verified the method is called correctly, but I still don't see any messages being sent or coming through. The handshake is done just fine. I'm not sure there's a simpler use case than this:

def report_sim_update(message):
    print("MESSAGE IN SITE: " + str(message))
    socketio.emit("progressbar", json.dumps(message))

and internally

self._notifier_function({
                    "player": player["name"],
                    "done:": True
                })

I'm sorry I'm posting a lot on this thread, I'm just surprised by the complexity using this library. It really doesn't seem that nuanced of a problem on the surface.