Server has multiple connections open but there's only one client

[mfoglio]

Hello,

I am designing the client-side of an application. I have access to the server code but I didn't write it myself so my understanding of it is limited. Both the client and the server use python-socketio==5.4.0. The client and the server are two docker containers which runs on the same machine.

The server is storing some credentials. The client sends a GET request to the server to request the credentials. Then, the client waits to receive the credentials on the publish handler. After waiting for 15 seconds, the client forces a reconnection: it disconnects, it initializes a new socketio.Client(), register the namespaces, and start the connection.
Here's the issue: the timeout occurs at least a few times a day. As a consequence, the connection is restarted a few times.
After this happens, the server seems to have multiple connections opened on its side. Is this an expected behavior? What may be happening here? Is it possible that the timeout happens because the server is unresponsive (or the connection has been lost) and as a consequence the client cannot effectively close the connection on the server? I believed the connection was managed by the library.

Here's my client code. Note that the only way to start the connection (using this code) is to call the reconnect method. The client connection is used by multiple threads. As you can see I implemented some mechanism to prevent multiple very-close-in-time reconnections (I originally believed there could be some race conditions). Also, to prevent race conditions, when emitting from the client, I always wait for socket_client.connected to be True (see emit method below).
socket_manager.py

import logging
import traceback
import time
from threading import Lock
from typing import List
from waiting import TimeoutExpired, wait

import socketio

import settings


logger = logging.getLogger(__name__)


class SocketManagerUser:

    def register_namespace(self) -> None:
        raise NotImplementedError


class SocketManager:

    def __init__(
            self,
            on_emit_timeout: float = settings.SOCKET_ON_EMIT_TIMEOUT,
            on_emit_wait_time_before_reconnecting: float = settings.SOCKET_ON_EMIT_WAIT_TIME_BEFORE_RECONNECTING
    ):
        self.on_emit_timeout = on_emit_timeout
        self.on_emit_wait_time_before_reconnecting = on_emit_wait_time_before_reconnecting
        # Initialise list of users
        self.users: List[SocketManagerUser] = list()
        # Initialize client
        self.socket_client = socketio.Client()
        self.last_reconnection_attempt = None
        # Thread lock
        self.lock = Lock()

    def reconnect(self, called_on_disconnect: bool = False):
        """
        :param called_on_disconnect: can be set to True to avoid disconnecting. This is useful when this function
            is called by disconnection handler, because otherwise we would trigger an infinite loop.
        :return:
        """
        # Use thread lock to prevent multiple connection attempts at the same time
        with self.lock:
            # Check if enough time has passed since last attempt
            if self.last_reconnection_attempt is not None and \
                    time.time() - self.last_reconnection_attempt < settings.SOCKET_RECONNECTION_MIN_WAIT_TIME:
                logger.warning(
                    f"Not attempting to reconnect socket because last attempt has been less than "
                    f"{settings.SOCKET_RECONNECTION_MIN_WAIT_TIME} seconds ago."
                )
                return
            # Log reconnection attempt
            self.last_reconnection_attempt = time.time()
        # Disconnect and reconnect
        if called_on_disconnect is False:
            self._disconnect()
        self._initialize_client()
        self._start_connection()

    def _start_connection(self):
        """ Namespaces must be registered before
        """
        # Start connection
        self.socket_client.connect(
            f"{settings.WEBSOCKET_URL}:{settings.WEBSOCKET_PORT}",
            headers={"CLIENT_ID": settings.AI_DETECTOR_CLIENT_ID},
            wait=True,
            wait_timeout=1
        )
        logger.info(f"Socket Client: namespaces registered after connecting: {self.socket_client.namespaces}")

    def _initialize_client(self):
        # Initialize client
        self.socket_client = socketio.Client()
        # Register namespaces
        for user in self.users:
            user.register_namespace()

    def _disconnect(self):
        try:
            if self.socket_client.connected is True:
                # Disconnect client
                self.socket_client.disconnect()
            else:
                logger.warning(f"Won't try to disconnect socket client cause it is not connected.")
        except:
            logger.error(
                f"Socket client disconnection failed:\n"
                f"{traceback.format_exc()}"
            )

    def register_user(self, user: SocketManagerUser):
        if user not in self.users:
            self.users.append(user)

    def emit(self, event, data=None, namespace=None, callback=None):
        """
        This function can raise a TimeoutExpired exception.
        :param event:
        :param data:
        :param namespace:
        :param callback:
        :return:
        """
        if self.socket_client.connected is False:
            # Wait for a short time before attempting to reconnect
            logger.warning(
                f"Attempting to emit on the socket client but the client is not connected: "
                f"attempting to reconnect and then wait {self.on_emit_wait_time_before_reconnecting}"
            )
            self.reconnect()
            time.sleep(self.on_emit_wait_time_before_reconnecting)
            # Wait for connection
            wait(
                predicate=lambda: self.socket_client.connected is True,
                timeout_seconds=self.on_emit_timeout,
                waiting_for="Socket client to be connected for emit event"
            )
        self.socket_client.emit(event, data, namespace, callback)

I didn't write the server code. I am doing my best to understand how it works. There's some hierarchy involved, and the server code uses: Flask-SocketIO==5.0.1 with python-socketio==5.4.0. The server code is a bit long, so I don't expect you to read it all. I am posting it here because I am not sure which part you would need. I am happy to provide more specific answers based on your suggestions.
The server code counts the number of clients connected by looking at the attribute _clients of the base class. Also, SocketIOServerBaseMessageHandler.
SocketIOServerCredentialsHandler extends SocketIOServerOnlyNamespaceMessageHandler which extends SocketIOServerBaseMessageHandler which extends Namespace (from flask_socketio import Namespace).

class SocketIOServerCredentialsHandler(SocketIOServerOnlyNamespaceMessageHandler):
    """
    class SocketIOServerOnlyNamespaceMessageHandler message handler for /aws-iam
    namespace to retrieve temporary credentials.
    """
    def __init__(self, credentials_manager, namespace=None, allowed_clients=None):
        super().__init__(namespace=namespace, allowed_clients=allowed_clients)
        self._credentials_manager = credentials_manager

    def on_publish(self, data, *args):
        """
        This namespace doesn't support publish.
        :param data: Data for publish event
        :param args: Args for publish event
        """
        client_id = self._get_client_id(*args)
        logger.info("Message Requested")
        self.reply(
            data={"ERROR": "PUBLISHED NOT SUPPORTED FOR THIS NAMESPACE"},
            client_id=client_id,
        )

    def on_get(self, data, *args):
        """
        Return Credentials for the GET event.
        :param data: Data for get event
        :param args: Args for get event
        """
        client_id = self._get_client_id(*args)
        if not self._validate(client_id):
            return

        logger.info(f"Requested GET message to {self.namespace} from {client_id}")

        # Get Credentials
        credentials = self._credentials_manager.get(client_id=client_id)
        if credentials is False:
            self.reply(
                data={"ERROR": "ERROR GETTING CREDENTIALS FROM AWS"},
                client_id=client_id,
            )
            return

        logger.info(f"Replying to {self.namespace} - {client_id} with Credentials")
        self.reply(credentials, client_id=client_id)

class SocketIOServerOnlyNamespaceMessageHandler(SocketIOServerBaseMessageHandler):
    """
    class SocketIOServerOnlyNamespaceMessageHandler message handler for
    socketio2socketio messages that are shadow.
    """

    def on_publish(self, data, *args):
        """
        Data is resent
        :param data: Data
        :param args: publish args
        """
        client_id = self._get_client_id(*args)
        data = self._validate(client_id, data)
        if not data:
            return

        self.publish(data, skip_sid=args[-1]["sid"])

    def on_get(self, data, *args):
        client_id = self._get_client_id(*args)
        if not self._validate(client_id):
            return

        logger.info(f"Requested GET message to {self.namespace} from {client_id}")
        self.publish(data, event=settings.SOCKET_IO_EVENT_GET, skip_sid=args[-1]["sid"])

class SocketIOServerBaseMessageHandler(Namespace):
    """
    https://flask-socketio.readthedocs.io/en/latest/#class-based-namespaces
    SocketIOServerBaseMessageHandler Base message handler based on flask_socketio
    class based namespaces
    """

    def __init__(self, namespace=None, allowed_clients=None):
        self._clients = {}
        self._sid = None
        self._allowed_clients = allowed_clients
        super().__init__(namespace=namespace)
        logger.info(f"Registering Namespace {self.namespace}")

    def _get_client_id(self, *args):
        """
        Util function to extract client from event args.
        :param args: Event args
        """
        client_id = self._clients.get(args[-1]["sid"])
        return client_id

    def _validate(self, client_id, data=None):
        """
        Validate if client is allow to that namespace or that event and data is
        valid.
        :param client_id: event client_id
        :param data: event data
        """
        if not client_id:
            self.reply({"ERROR": "CLIENT ID IS REQUIRED"})
            return False

        if self._allowed_clients and client_id not in self._allowed_clients:
            self.reply(
                {"ERROR": "CLIENT_ID IS NOT AUTHORIZED TO PUBLIC IN THIS TOPIC"},
                client_id=client_id,
            )
            return False

        if data is None:
            return True

        try:
            data = json.loads(data)

        except Exception:
            self.reply({"ERROR": "BODY IS INVALID"}, client_id=client_id)
            return False

        return data

    def trigger_event(self, event, *args):
        """
        Override super method to related sid with header HTTP_CLIENT_ID.
        :param event: event
        :param args: args of events
        """
        #
        args += ({"sid": args[0]},)
        if event == "connect":
            self._clients[args[0]] = args[1].get("HTTP_CLIENT_ID")
            if not self._clients[args[0]]:
                return "", 400

            else:
                logger.info(f"{self._clients[args[0]]} Connected")

        super().trigger_event(event, *args)

    def on_connect(self):
        pass

    def on_disconnect(self, *args):
        """
        on_disconnect event remove sid from list of clients
        :param args:
        """
        client_id = self._clients.pop(args[0]["sid"])
        if client_id:
            logger.info(f"{client_id} disconnected")

    def publish(self, data, event=settings.SOCKET_IO_EVENT_PUBLISH, skip_sid=None):
        """
        :param data: Data to publish. Must be a dict
        :param event: Event to emit data two possibles GET and PUBLISH
        :param skip_sid: Boolean to avoid resend the message to sender
        """
        logger.info(f"Publishing Message to Topic {self.namespace} from cloud")
        if event not in [
            settings.SOCKET_IO_EVENT_GET,
            settings.SOCKET_IO_EVENT_PUBLISH,
        ]:
            return

        if isinstance(data, dict):
            data = json.dumps(data)

        self.server.emit(
            event=event,
            data=data,
            namespace=self.namespace,
            skip_sid=skip_sid,
            broadcast=True,
        )

    def reply(self, data, client_id=None):
        """
        :param data: Data to publish
        :param client_id: Client Id to publish the message
        """
        if isinstance(data, dict):
            data = json.dumps(data)

        if not client_id:
            logger.info(f"Replying Message to Client from Cloud to {self.namespace}")
            self.server.emit(
                event=settings.SOCKET_IO_EVENT_PUBLISH,
                data=data,
                namespace=self.namespace,
                skip_sid=None,
                broadcast=True,
            )

        else:
            # Convert client_id to sids
            clients_sids = [key for key, value in self._clients.items() if value == client_id]
            logger.info(f"Replying Message to Client {client_id} from Cloud to {self.namespace}")
            for sid in clients_sids:
                self.emit(settings.SOCKET_IO_EVENT_PUBLISH, data, room=sid)

Could you provide any hint on what may be happening? Let me know if you need other information.

[miguelgrinberg]

The Socket.IO client includes its own reconnection support. My guess is that the multiple connections happen because your own reconnection attempts interfere with the internal ones. If you want to use your own reconnection logic I suggest you turn off the client's own by passing reconnection=False when you create a Client object.

[mfoglio]

Thank you, @miguelgrinberg . I will try that. I also noticed that socketio.Client has a parameter handle_sigint. According to the documentation:

handle_sigint – Set to True to automatically handle disconnection when the process is interrupted, or to False to leave interrupt handling to the calling application. Interrupt handling can only be enabled when the client instance is created in the main thread.

My socket connections are used by different threads. Upon errors, one of this thread will recreate the connection to be shared with the other threads. Could this be the cause of my problems? I am not sure how to interpret the documentation.
My threads never end unless the application is terminated.

I am wondering if always having the main thread recreating the connection may fix the issue.

[miguelgrinberg]

The handle_sigint option is for the interrupt handler, like when you press ctrl-c. It has nothing to do with reconnections.

[mfoglio]

Hi @miguelgrinberg , I noticed an error on the server side: ValueError: Too many packets in payload.
I found a possible solution here: miguelgrinberg/python-engineio#142 .
Could this be the reason why my client is requesting credentials to the server, but the server sometimes does not reply back? I am not entirely sure what the error means. If you could point me to some documents I'd be happy to read more about this.
Thank you

[miguelgrinberg]

I can't really diagnose your problem just from your high level description. Do you have logs to show me? See the troubleshooting section of the documentation to learn how to enable logging if you haven't done so already.