If i emit a event in a Thread, the client can't get the event.

[Abbyyan]

I'm trying to do as following.

(1) send message from client to server, add the message into a Queue on the server side.
(2) the server process the data in the queue in a Thread. 
(3) The server emit a response to the client after processing data

The server.py is as follows:

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import eventlet
import socketio
from time import sleep
import threading
import queue
q = queue.Queue()


sio = socketio.Server(logger=True, engineio_logger=True)
app = socketio.WSGIApp(sio, static_files={
    '/': {'content_type': 'text/html', 'filename': 'index.html'}
})


@sio.event
def connect(sid, environ):
    print('new client connect ', sid)


@sio.event
def request(sid, data):
    sio.emit('client_response', data)
    print("(3)"+str(data))


@sio.event
def response(sid, data):
    print("(2) "+str(data))
    info = {"sid": sid, "data": data}
    q.put(info)


def process_data():
    while True:
        if not q.empty():
            item = q.get()
            data = item.get('data', "")
            sid = item.get('sid', None)
            print("data here: %s" % data)
            sleep(1)
            data = "data send from server "+str(data)
            request(sid, data)



@sio.event
def disconnect(sid):
    print('disconnect ', sid)


threading.Thread(target=process_data).start()

if __name__ == '__main__':
    eventlet.wsgi.server(eventlet.listen(('x.x.x.x', 8081)), app)

The client.py is as follows:

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import socketio
sio = socketio.Client(logger=True)
sio.connect('http://x.x.x.x:8081')


@sio.event
def connect():
    print('connection established')


@sio.event
def client_request(data):
    print("(1) %s" % data)
    sio.emit('response', data)


@sio.event
def client_response(data):
    print("(4) %s" % data)




if __name__ == '__main__':
    for i in range(0, 2):
                client_request("id: "+str(i))
    sio.wait()

The server can get the message send from client, and can process data. But after the server do sio.emit('client_response', data) in the Thread, the client can't receive the processed data from server (If i do the emit in the main thread, it works well ). Could you please give me some suggestions about this please? Thanks a lot.
The log is as follows:
On server side:

% python server.py
Server initialized for eventlet.
(617) wsgi starting up on http://x.x.x.x:8081
(617) accepted ('x.x.x.x', 44067)
a721b61ce62345d399db3006ccf2a769: Sending packet OPEN data {'sid': 'a721b61ce62345d399db3006ccf2a769', 'upgrades': ['websocket'], 'pingTimeout': 60000, 'pingInterval': 25000}
new client connect  a721b61ce62345d399db3006ccf2a769
a721b61ce62345d399db3006ccf2a769: Sending packet MESSAGE data 0
x.x.x.x - - [22/Sep/2020 16:47:36] "GET /socket.io/?transport=polling&EIO=3&t=1600764456.5817175 HTTP/1.1" 200 371 0.026230
(617) accepted ('x.x.x.x', 44068)
a721b61ce62345d399db3006ccf2a769: Received request to upgrade to websocket
a721b61ce62345d399db3006ccf2a769: Upgrade to websocket successful
a721b61ce62345d399db3006ccf2a769: Received packet PING data None
a721b61ce62345d399db3006ccf2a769: Sending packet PONG data None
a721b61ce62345d399db3006ccf2a769: Received packet MESSAGE data 2["response","id: 0"]
received event "response" from a721b61ce62345d399db3006ccf2a769 [/]
(2) id: 0
a721b61ce62345d399db3006ccf2a769: Received packet MESSAGE data 2["response","id: 1"]
data here: id: 0
received event "response" from a721b61ce62345d399db3006ccf2a769 [/]
(2) id: 1
emitting event "client_response" to all [/]
a721b61ce62345d399db3006ccf2a769: Sending packet MESSAGE data 2["client_response","data send from server id: 0"]
(3)data send from server id: 0
data here: id: 1
emitting event "client_response" to all [/]
a721b61ce62345d399db3006ccf2a769: Sending packet MESSAGE data 2["client_response","data send from server id: 1"]
(3)data send from server id: 1

On client side:

% python client.py
Engine.IO connection established
Namespace / is connected
(1) id: 0
Emitting event "response" [/]
(1) id: 1
Emitting event "response" [/]
Engine.IO connection dropped
Connection failed, new attempt in 1.03 seconds
Engine.IO connection established
Namespace / is connected
connection established
Reconnection successful
Engine.IO connection dropped

[miguelgrinberg]

You can't use threading and queues with eventlet, they are incompatible. See https://eventlet.net/doc/patching.html for how to monkey patch them to become compatible.

[Abbyyan]

You can't use threading and queues with eventlet, they are incompatible. See https://eventlet.net/doc/patching.html for how to monkey patch them to become compatible.

Thanks for your help. I've tried to use green eventlet as follows

eventlet.monkey_patch(thread=True)
import threading

But it will not return from the threading. So after the threading.Thread , if the client try to connect to the server, it will refused by the server. If i just want to use a Queue to process the data, I should try to use kombu or AioPika , is that right?

[miguelgrinberg]

@Abbyyan I really don't know how to help, this isn't a problem with this package, you need to work around the limitations eventlet imposes in your application. Monkey patching helps in most cases, but not always.

[Abbyyan]

@Abbyyan I really don't know how to help, this isn't a problem with this package, you need to work around the limitations eventlet imposes in your application. Monkey patching helps in most cases, but not always.

Thanks a lot. According to https://python-socketio.readthedocs.io/en/latest/, I've changed to use gevent as follows

from gevent import pywsgi
from geventwebsocket.handler import WebSocketHandler
app = socketio.WSGIApp(sio)
pywsgi.WSGIServer(('', 8000), app,
                  handler_class=WebSocketHandler).serve_forever()

It can send message from server to client even in a sub thread now. But it's wired that it will take a long time receiving message from server. The log is as follows. After the server print("(3)"+str(data)) , it will take a long time to get message on the client side.
The log is as follows:
On the server side:

% python server.py 
Server initialized for gevent.
59b2423ec2a7474dba22a5b789de6666: Sending packet OPEN data {'sid': '59b2423ec2a7474dba22a5b789de6666', 'upgrades': ['websocket'], 'pingTimeout': 60000, 'pingInterval': 25000}
new client connect  59b2423ec2a7474dba22a5b789de6666
59b2423ec2a7474dba22a5b789de6666: Sending packet MESSAGE data 0
59b2423ec2a7474dba22a5b789de6666: Received request to upgrade to websocket
59b2423ec2a7474dba22a5b789de6666: Upgrade to websocket successful
59b2423ec2a7474dba22a5b789de6666: Received packet PING data None
59b2423ec2a7474dba22a5b789de6666: Sending packet PONG data None
59b2423ec2a7474dba22a5b789de6666: Received packet MESSAGE data 2["response","id: 0"]
received event "response" from 59b2423ec2a7474dba22a5b789de6666 [/]
59b2423ec2a7474dba22a5b789de6666: Received packet MESSAGE data 2["response","id: 1"]
received event "response" from 59b2423ec2a7474dba22a5b789de6666 [/]
(2) id: 0
(2) id: 1
data here: id: 0
emitting event "client_response" to all [/]
59b2423ec2a7474dba22a5b789de6666: Sending packet MESSAGE data 2["client_response","data send from server id: 0"]
(3)data send from server id: 0
data here: id: 1
emitting event "client_response" to all [/]
59b2423ec2a7474dba22a5b789de6666: Sending packet MESSAGE data 2["client_response","data send from server id: 1"]
(3)data send from server id: 1  # it will take a long time after this log
59b2423ec2a7474dba22a5b789de6666: Received packet PING data None
59b2423ec2a7474dba22a5b789de6666: Sending packet PONG data None

Only after the server print the last two lines log, the client side can receive the message send from server. My expectation is the
client will receive the message after the server emit. They are on the same machine, so the network latency should be relatively small . Hope for your help. Thanks a lot.

59b2423ec2a7474dba22a5b789de6666: Received packet PING data None
59b2423ec2a7474dba22a5b789de6666: Sending packet PONG data None

[miguelgrinberg]

Gevent is exactly the same as eventlet with regards to blocking and patching. As I said a few times already, you need to learn how to work with these platforms. You are not using standard Python anymore, these are asynchronous frameworks. The rules are different.

[Abbyyan]

Gevent is exactly the same as eventlet with regards to blocking and patching. As I said a few times already, you need to learn how to work with these platforms. You are not using standard Python anymore, these are asynchronous frameworks. The rules are different.

It seems like the eventlet are not compatible with multiprocessing , am i right?

[miguelgrinberg]

@Abbyyan without monkey patching it will not work. With monkey-patching I believe eventlet still has some issues/bugs that haven't been solved in the implementation of multiprocessing.

[stansotn]

I ran into the same issue. My application is interacting with threading/queue dependent hardware. It works well with http polling werkzeug + python-socketio. I have been working on upgrading my application to websockets and I have tried eventlet and gevent with no luck. As I understand, monkey-patching attempts to convert pythonic threads into async tasks, which is not feasible when interacting with hardware (you need to keep the drivers consistently awake). Any suggestions on how would I implement a websocket interaction in a multithreading fashion?

[miguelgrinberg]

You can't use greenlet frameworks with code that hasn't been written specifically for them. Monkey patching replaces a big portion of the standard library with compatible code. If you have other code that is blocking you have to fix it so that it is greenlet friendly and does not block. If you rely on third party libraries that is unlikely to have a solution unless you find another library that supports greenlets or write one yourself.

[stansotn]

I see, I guess I would have to run my blocking code in a different process feeding a multiprocessing message queue.

On the concurrent side of things (socketio application) I would have to run a background task start_background_task() which contains an infinite while loop that processes messages from the multiprocessing queue. Then each loop iteration I'd call socketio.sleep() to yield control to the other tasks managing the web stuff. I am going to try this out tomorrow, I would much appreciate if you let me know if this is a terrible idea..

[miguelgrinberg]

@stansotn not a terrible idea, this is actually a good compromise solution. I'm not 100% sure multiprocessing can be used under eventlet or gevent though, you may want to confirm that before you spend any significant amount of time on this.

[stansotn]

Confirming that multiprocessing works well with socketio and eventlet when monkeypatched. I wonder why does it need monkeypatching at all.. I am able to run all my cpu bound code in the child process and communicate via multiprocessing queues.

[miguelgrinberg]

I wonder why does it need monkeypatching at all

To be able to communicate with the child process without blocking.

[stansotn]

Right, I made sure all calls to the multiprocessing queue are non-blocking, now everything runs without monkeypatching!

# A non-blocking multiprocessing queue get example.
m = None
try:
    m = tx_queue.get(False)
except queue.Empty:
    pass
return m

ps: I noticed monkeypatching breaks the ability to generate coverage.