unclose session after disconnect

[wangjiancn]

server.py

sio = socketio.AsyncServer(
    async_mode='asgi', client_manager=mgr, cors_allowed_origins="*")


@sio.event
async def connect(sid, environ):
    claims = get_jwt_claims_from_environ(environ)
    ga = get_ga_from_environ(environ)
    async with sio.session(sid) as session:
        await Lock().wait()
        if claims:
            user_ID = claims.get('user_ID')
            session['user_ID'] = user_ID
            ret = await RedisClient.add_to_set(user_ID)
        if not ret:
            raise ConnectionRefusedError('Duplicate connect')


@sio.event
async def disconnect(sid):
    async with sio.session(sid) as session:
        user_ID = session.get('user_ID')
        if user_ID is not None:
            await RedisClient.remove_from_set(session.get('user_ID', ''))

version:

# python 3.6.9

aioredis        1.3.1  
async-timeout   3.0.1  
click           7.1.1  
h11             0.9.0  
hiredis         1.0.1  
httptools       0.1.1  
pip             19.3.1 
python-engineio 3.12.1 
python-socketio 4.5.0  
setuptools      41.6.0 
six             1.14.0 
uvicorn         0.11.3 
uvloop          0.14.0 
websockets      8.1    
wheel           0.33.6 

start:

uvicorn server:app --host 0.0.0.0 --workers 1 --log-level error --use-colors

nginx conf:

    location /socketio/ {
        proxy_pass http://localhost:8888/;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection $connection_upgrade;
    }

When connect, I store user_ID in session and redis sets. If user_ID exists in redis sets, i will refuse connect.
When disconnect, remove user_ID from redis sets.
I find some user_ID not exists in redis sets but in session, I get session use follow code:

def get_online_user(sio):
    for s in sio.eio.sockets.values():
        session = s.session.get('/', {})
        if session.get('user_ID') is not None:
            online_user.append(session['user_ID'])
    return online_user

Sometimes, it has duplicate user_ID in session.

[miguelgrinberg]

I can't really comment on the state of your Redis set since that is controlled by your application, but if you think the connect/disconnect handlers are invoked more times than expected please provide a complete example app I can use to verify the problem.

[wangjiancn]

Thanks for the quick response!

I will provite a demo we use.

We tried to reproduce the problem, but it failed. It noly appear after run long time (24 hour or more).

[wangjiancn]

Pipfile:

[[source]]
name = "pypi"
url = "https://pypi.org/simple"
verify_ssl = true

[dev-packages]

[packages]
uvicorn = "*"
python-socketio = "*"
aioredis = "*"

[requires]
python_version = "3.6"

server.py:

import json

import socketio
import uvicorn

SOCKET_IO_PORT = 5006

# 调试端口,服务端口在docker-compose.yml
sio = socketio.AsyncServer(async_mode='asgi', cors_allowed_origins="*")


def get_jwt_claims_from_environ(environ):
    bearer_token = environ.get('HTTP_AUTHORIZATION', '')  # type:str
    return bearer_token


connect_count = 0


@sio.event
async def connect(sid, environ):
    global connect_count
    connect_count += 1
    print('connect count:', connect_count)
    token = get_jwt_claims_from_environ(environ)
    async with sio.session(sid) as session:
        if token:
            session['token'] = token
    # raise ConnectionRefusedError(f'Duplicate connect {sid}')
    return False


disconnect_count = 0


@sio.event
async def disconnect(sid):
    global connect_count
    connect_count -= 1
    print('disconnect count:', connect_count)
    async with sio.session(sid) as session:
        print(session)


def get_online_user(sio):
    online_user = []
    for s in sio.eio.sockets.values():
        session = s.session.get('/', {})
        if session.get('token') is not None:
            if session['token']:
                online_user.append(session['token'])
    return online_user


async def other_asgi_app(scope, receive, send):
    assert scope['type'] == 'http'
    status = 200
    online_user_from_session = get_online_user(sio)
    body = {
        'online_user_from_session_count': len(online_user_from_session),
        'online_user_from_session': online_user_from_session,
        'online_user_from_session_set_count': len(
            set(online_user_from_session)),
        'online_user_from_session_set': list(set(online_user_from_session)),
    }

    await send({
        'type': 'http.response.start',
        'status': status,
        'headers': [
            [b'content-type', b'application/json'],
        ]
    })
    await send({
        'type': 'http.response.body',
        'body': json.dumps(body).encode(),
    })


app = socketio.ASGIApp(sio, other_asgi_app=other_asgi_app)

if __name__ == "__main__":
    uvicorn.run(
        app,
        host="0.0.0.0",
        port=SOCKET_IO_PORT,
        log_level="info",
        access_log=True,
        use_colors=True
    )

clinet.html:

<!DOCTYPE html>
<html lang="en">


<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Document</title>
    <script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/2.3.0/socket.io.slim.dev.js"></script>
</head>

<body>
    <script>
        const token = new Date().toISOString()
        const socket = io('http://localhost:5006', {
            transportOptions: {
                polling: {
                    extraHeaders: {
                        'Authorization': token
                    }
                }
            },
        });
        console.log('token', token)
    </script>

</body>

</html>

visit http://localhost:5006/ can see session like this:

when open client.html and then close it, server log like this:

INFO:     127.0.0.1:41904 - "OPTIONS /socket.io/?EIO=3&transport=polling&t=N4Cmn9r HTTP/1.1" 200 OK
connect count: 1
INFO:     127.0.0.1:41908 - "GET /socket.io/?EIO=3&transport=polling&t=N4Cmn9r HTTP/1.1" 200 OK
INFO:     ('127.0.0.1', 41912) - "WebSocket /socket.io/" [accepted]
INFO:     127.0.0.1:41904 - "OPTIONS /socket.io/?EIO=3&transport=polling&t=N4CmnAX&sid=0bf68241da724a92a3d97fba73aec1d2 HTTP/1.1" 200 OK
INFO:     127.0.0.1:41908 - "GET /socket.io/?EIO=3&transport=polling&t=N4CmnAX&sid=0bf68241da724a92a3d97fba73aec1d2 HTTP/1.1" 200 

It still have session when connect return False, and when html close disconnect event not trigger. It is expected like this ?

[miguelgrinberg]

Did you wait a minute after closing the browser? You are using polling, where disconnections cannot be detected immediately, it takes up to a minute.

[wangjiancn]

In server, connect event return False :

1. It still store session in sio.eio.sockets.values()
2. When close client, disconnect event not trigger, but it can remove session from sio.eio.sockets.

I think, when connect event return False, it should not store session.

[miguelgrinberg]

Okay, I think I understand now. I'll look into it.

[wangjiancn]

Some error message in log, may be useful to position bug.

Traceback (most recent call last):
  File "/usr/local/lib/python3.6/site-packages/uvicorn/protocols/websockets/websockets_impl.py", line 153, in run_asgi
    result = await self.app(self.scope, self.asgi_receive, self.asgi_send)
  File "/usr/local/lib/python3.6/site-packages/uvicorn/middleware/proxy_headers.py", line 45, in __call__
    return await self.app(scope, receive, send)
  File "/usr/local/lib/python3.6/site-packages/engineio/async_drivers/asgi.py", line 46, in __call__
    await self.engineio_server.handle_request(scope, receive, send)
  File "/usr/local/lib/python3.6/site-packages/socketio/asyncio_server.py", line 348, in handle_request
    return await self.eio.handle_request(*args, **kwargs)
  File "/usr/local/lib/python3.6/site-packages/engineio/asyncio_server.py", line 303, in handle_request
    r['response'], environ)
  File "/usr/local/lib/python3.6/site-packages/engineio/async_drivers/asgi.py", line 168, in make_response
    'headers': headers})
  File "/usr/local/lib/python3.6/site-packages/uvicorn/protocols/websockets/websockets_impl.py", line 220, in asgi_send
    raise RuntimeError(msg % message_type)
RuntimeError: Expected ASGI message 'websocket.send' or 'websocket.close', but got 'http.response.start'.

[viktorvorobev]

Hello!
I've noticed the same issue (I guess) with using SocketIO as a server to run. For some reason, the closed sockets are not being deleted from within the EngineIO, and they remain in memory, despite the fact that they are closed.
I've been tracking the memory usage of my app, and it appears to be that when I refresh the browser page, I'm getting some memory leaks. I've logged out the sockets via simple script:

f.write("Opened sockets: {}".format(len(socketio.server.eio.sockets)))
f.write("\n")
for sid, socket in socketio.server.eio.sockets.iteritems():
    f.write("SID: {}\n".format(sid))
    f.write("State: connected {} closing {}, closed {}\n".format(socket.connected, socket.closing, socket.closed))
    f.write("Socket Queue: size {} \n\n".format(socket.queue.qsize()))

And the result looked quite sad, like so (don't mind the queue size, I'm broadcasting quite a lot of data, so no surprises here):

Opened sockets: 30                                                                                   
SID: 68737ede0359448e8084c0e3c7b25b2e                                                                
State: connected True, upgrading False, upgraded False, closing True, closed True                    
Socket Queue: full False, empty False, size 169                                                      

SID: dc47bce25482424aac9a3a521fff30cf                                                                
State: connected True, upgrading False, upgraded False, closing True, closed True                    
Socket Queue: full False, empty False, size 135                                                      

SID: 3abd0a4f07dc491ea7f836d1dd4d65e1                                                                
State: connected True, upgrading False, upgraded False, closing True, closed True                    
Socket Queue: full False, empty False, size 137                                                      

SID: 24289a8c680449599600b9da6512b7e1                                                                
State: connected True, upgrading False, upgraded False, closing True, closed True                    
Socket Queue: full False, empty False, size 154                                                      

SID: 95c1437242cd485a9d9a922310a2a60d                                                                
State: connected True, upgrading False, upgraded False, closing True, closed True                    
Socket Queue: full False, empty False, size 139       

...

The worst part is that even when the selenium script is stopped and all the browsers closed, those sockets are not going anywhere.

I can, of course, clean them up manually via this code:

sessions = socketio.server.eio.sockets.keys()
for sid in sessions:
    try:
        socketio.server.eio._get_socket(sid)
    except KeyError:
        pass

But this just don't seem to be right thing to do.

My setup is straight forward SocketIO(http_compression=True).run() with no Gunicorn neither Nginx to mess up with.

python-engineio == 3.12.1
python-socketio == 4.3.0

[Kazhuu]

I've been looking into why my application code doesn't close old connections. It seems this is the very same issue I'm facing. Any update on this?