Running external emit first and connecting to client giving error

[arun6894]

Considering a real-time scenario connections can happen when emits are already happening.

So, I have tried to emit through external emit in a separate process. Running servers in 2 processes. I have tried connecting from Asynclient. I could see the disconnections.

Server.py

from gevent import monkey
monkey.patch_all()
import socketio
import gevent
from gevent import pywsgi
from geventwebsocket.handler import WebSocketHandler
from datetime import datetime



url=['kafka://0.0.0.0:9092', 'kafka://0.0.0.0:9092', 'kafka://0.0.0.0:9092']
mgr = socketio.KafkaManager(url=url, channel='check')
sio = socketio.Server(async_mode='gevent',client_manager=mgr, logger=True, engineio_logger=True)
app = socketio.WSGIApp(sio)

user_to_sid = {}
sid_to_user = {}
client_count = 0


def authenticate_user(username):
    """Create user id with the username """

    global user_to_sid
    user_id = 1
    if not user_to_sid:
        user_to_sid = {user_id :username}
    if username not in list(user_to_sid.values()):
        user_id = max(list(user_to_sid.keys()))+1
        user_to_sid[user_id]= username
    return user_id


@sio.event
def connect(sid, environ):
    """Triggers event during client connections """
    global client_count


    username = environ.get('HTTP_X_USERNAME')
    print('username:', username)
    if not username:
        return False
    user_id = authenticate_user(username)
    user_to_sid[user_id] = sid
    sid_to_user[sid] = user_id

    with sio.session(sid) as session:
        session['username'] = username
    sio.save_session(sid, {'username': username})
    client_count += 1

    with open("message_connect_a.log", mode="a") as log:
        now = datetime.now()
        log.write("Client {} connect at {}\n".format(username, now))
    sio.enter_room(sid, 'a')

@sio.event
def one(sid, message):
    session = sio.get_session(sid)
    print('message from {} recieved with {}'.format(session['username'], message))

@sio.event
def custom_server(sid, message):
    print('pubsub_message',message)


@sio.event
def all(sid, message):
    session = sio.get_session(sid)
    print('message from {} all of recieved with {}'.format(session['username'], message))

@sio.event
def disconnect(sid):
    global client_count
    user_id = sid_to_user[sid]
    del user_to_sid[user_id]
    del sid_to_user[sid]
    client_count -= 1
    
    with open("message_disconnect_a.log", mode="a") as log:
        now = datetime.now()
        log.write("Client {} shutdown at {} with {} count\n".format(sid, now, client_count))

    print(sid, 'disconnected')
    #await sio.emit('client_count', client_count)
        

if __name__=='__main__':
    pywsgi.WSGIServer(('0.0.0.0', 61431), app, handler_class=WebSocketHandler).serve_forever()

Client.py

import asyncio
import socketio
from datetime import datetime
from random_username.generate import generate_username
import random


lst = []
total_emit_count = 0
run_seconds = 0
start =  datetime.now()

class Check:
    def __init__(self):
        self.sio = socketio.AsyncClient(handle_sigint=False, logger=False, engineio_logger=False)
        self.sio.on('connect', self.connect, namespace='/')
        self.sio.on('mult', self.mult, namespace='/')
        self.sio.on('mult1', self.mult1, namespace='/')
        self.sio.on('connect_error', self.connect_error, namespace='/')
        self.sio.on('disconnect', self.disconnect, namespace='/')
        self.sio.on('client_count', self.client_count, namespace='/')
        self.sio.on('user_left', self.user_left, namespace='/')
        self.sio.on('custom_client', self.custom_client, namespace='/')



    async def main_connect(self):
        username = generate_username(1)[0]
        await self.sio.connect('http://0.0.0.0:61432/?socket_qa2=True&session_id='+str(random.randint(1,100000)), wait_timeout = 10, headers={'X-Username': username})

        
    async def connect(self):
        print('connected')


    async def connect_error(self,e):
        print(e)

    async def mult(self,data):
        print(data)

        
    async def mult1(self,data):
        pass
    
    async def custom_client(self, data):
        print('pubsub_message',data)

    async def disconnect(self):
        print('disconnected')
        


    async def client_count(self,count):
        print('There are', count, 'connected clients.')

    
    async def user_left(self,username):
        print('User', username, 'has left.')

    async def run(self):
        await self.main_connect()   
 





async def main():
    async def loop(i):
        await asyncio.sleep(i/10)   # <--- allow a small delay between connections here
        obj1 = Check()
        await obj1.run()
        await obj1.sio.wait()
    x = [loop(i) for i in range(1000)]
    await asyncio.gather(*x)


loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.run_until_complete(main())

external_emit.py

import socketio
import gevent
url=['kafka://0.0.0.0:9092', 'kafka://0.0.0.0:9092', 'kafka://0.0.0.0:9092']
external_sio = socketio.KafkaManager(url=url, channel='check', write_only=True)

data1= {"hi": "test"}


try:
   count = 0
   while True:
       count+=1
       print('message sent:',count)
       for i in range(10):
            gevent.sleep(0.1)
            data1['id_']= 'room_A'
            result1 = external_sio.emit('mult', data1, room='a')
            data1['id_']= 'room_B'
            result2 = external_sio.emit('mult', data1, room='b')
       gevent.sleep(0.1)

except Exception as error:
   print(error)

From the external emit I am trying to emit 10 emits for 1 second.
Later trying to run client.py for establishing a client connection.
Here client connections and emits are happening parallelly. Seeing the error from client.py. Server is seeing disconnections and attempts polling.

Sending packet PONG data
Received packet PING data
Sending packet PONG data
Attempting polling connection to http://0.0.0.0:61431/socket.io/?socket_qa2=True&session_id=7436&transport=polling&EIO=4
Sending packet CLOSE data None
Engine.IO connection dropped
Write loop: WebSocket connection was closed, aborting
Exiting write loop task
Waiting for write loop task to end
Exiting read loop task
Traceback (most recent call last):
  File "client.py", line 104, in <module>
    loop.run_until_complete(main())
  File "/usr/lib64/python3.6/asyncio/base_events.py", line 484, in run_until_complete
    return future.result()
  File "client.py", line 92, in main
    await asyncio.gather(*x)
  File "client.py", line 89, in loop
    await obj1.run()
  File "client.py", line 78, in run
    await self.main_connect()
  File "client.py", line 29, in main_connect
    await self.sio.connect('http://0.0.0.0:61431/?socket_qa2=True&session_id='+str(random.randint(1,100000)), wait_timeout = 50, headers={'X-Username': username})
  File "/websocket_test/test_websocket/lib64/python3.6/site-packages/socketio/asyncio_client.py", line 164, in connect
    'One or more namespaces failed to connect')
socketio.exceptions.ConnectionError: One or more namespaces failed to connect

When trying to increase emit time and also connection time for external_emit and client file. Could see sometime there is no disconnection. Later I am seeing the above error.

[miguelgrinberg]

What does the server say about these errors?

[arun6894]

Today when I ran it again i get the following error from client.py

packet queue is empty, aborting
packet queue is empty, aborting
packet queue is empty, aborting
packet queue is empty, aborting
packet queue is empty, aborting
packet queue is empty, aborting
Connection refused by the server
Connection refused by the server
Connection refused by the server
Connection refused by the server
Connection refused by the server
Connection refused by the server
Traceback (most recent call last):
  File "client.py", line 110, in <module>
    loop.run_until_complete(main())
  File "/usr/lib64/python3.6/asyncio/base_events.py", line 484, in run_until_complete
    return future.result()
  File "client.py", line 98, in main
    await asyncio.gather(*x)
  File "client.py", line 95, in loop
    await obj1.run()
  File "client.py", line 84, in run
    await self.main_connect()
  File "client.py", line 30, in main_connect
    await self.sio.connect('http://0.0.0.0:61431/?socket_qa2=True&session_id='+str(random.randint(1,100000)), wait_timeout = 10, headers={'X-Username': username})
  File "/root/websocket_test/test_websocket/lib/python3.6/site-packages/socketio/asyncio_client.py", line 149, in connect
    raise exceptions.ConnectionError(exc.args[0]) from None
socketio.exceptions.ConnectionError: Connection refused by the server

From server.py:

annot send to sid Fp0cx1UrC6bMhFmUAACg

Cannot send to sid fnaOOJheVwkRKho6AAJD
Cannot send to sid pSTOPnQFrHU5eMIjAAJF
Cannot send to sid IihYas9o8FN6mBmGAAJI
Cannot send to sid oGtQ-8UvUmrnEo03AAJK
Cannot send to sid AgaR2LnTYvwq8y8-AAJL
Cannot send to sid t0pnlsOJbsJCqx2RAAJN

[miguelgrinberg]

I'm not really sure what you expect me to do or tell you with this information, as you seem to be asking your system to do more than it can handle. This package does not have a limitation on number of connections, but your hardware does. You can't just create as many accounts as you like and expect things will work, you need to find what your system can manage and keep within the range where things work well. You also appear to be running all the clients and the server on the same computer, which is a really bad way to measure performance.

[arun6894]

@miguelgrinberg I was running a server in a single machine with 2 cores and a client in other machines with 4 cores. Externally emitting is happening in the same server machine in a different process.

external emit is doing 10 emits per second. While the client is trying to connect only 1000 connections. In this process I see failures. Do you think the above hardware is not sufficient to handle this ? Just want to know your thoughts on this.

[miguelgrinberg]

What I think or don't think is really not the point here, as I have really no idea what your hardware is capable of doing. You are testing this. Find out what your hardware can do and stay within those limits and you should be fine.

Also, those 0.0.0.0 URLs you are connecting to are curious. Not sure how your client is connecting to a different server machine if you don't use actual IP addresses.

Finally, how many cores you have on your system is irrelevant, since you seem to be running a single process server. So you are always using one core.

[arun6894]

@miguelgrinberg Apologies, those actual IP's are removed while posting here for security reasons. I work on Linux machines. Client and server machines were able to communicate by adding machine details in /etc/hosts. so that they can recognize the ports and establish connections.

I understand it still takes 1 core. will try to check what the single core can handle with the test. Thanks for the inputs.