[BUG] Router mapping updates orphan thread-local clients and leak TCP sockets

[llyycchhee]

Describe the bug

Router.set_mapping(), Router.add_router(), and
Router.remove_router() replace self._cache_local with a new
threading.local() object.

When Router clients have already been created in other threads, replacing the
threading.local() object makes those per-thread client caches unreachable.

However, ActorCallerThreadLocal._clients and its listener tasks still retain
the old clients. Since the remote server remains alive, these TCP connections
stay in ESTABLISHED state and are not closed.

The next RPC cannot find the old client in the Router cache, so it creates a
new TCP connection. Repeated route mapping updates therefore cause file
descriptors and TCP sockets to grow continuously.

Relevant code:

• python/xoscar/backends/router.py
  • Router.set_mapping
  • Router.add_router
  • Router.remove_router
• python/xoscar/backends/core.py
  • ActorCallerThreadLocal._listen_client
  • ActorCallerThreadLocal._listen

To Reproduce

Environment:

1. Python: 3.12
2. Xoscar: 0.9.8
3. OS: Linux container
4. Distributed topology:
   • main/local node: 10.1.0.45
   • worker node: 10.1.0.44

The issue was reproduced through Xinference, which uses Xoscar actor pools:

1. Start a local/supervisor process on 10.1.0.45.
2. Start a worker process on 10.1.0.44.
3. Repeatedly create and remove a sub-pool, or repeatedly trigger a model
   deployment failure followed by cleanup.
4. Observe the FD count of the Xoscar main-pool process.

Example:

PID=<xoscar-main-pool-pid>

watch -n 1 '
printf "fds: "
ls /proc/'"$PID"'/fd | wc -l
printf "sockets: "
find /proc/'"$PID"'/fd -lname "socket:*" | wc -l
'

Each route configuration synchronization invokes Router mapping updates.
Before applying a workaround, every failed deployment created an additional
TCP connection such as:
10.1.0.45: -> 10.1.0.44:57265 ESTABLISHED
The destination port belonged to the Xoscar worker process.
Waiting for 60 seconds did not release these sockets. Repeating the operation
caused monotonically increasing socket FDs.
A minimal cache-level reproduction is:

import threading

from xoscar.backends.router import Router


router = Router(["main"], None)
ready = threading.Event()
updated = threading.Event()
result = {}


def use_router():
    cache = router._cache
    cache[("worker", None, None, None)] = object()
    result["before"] = cache
    ready.set()

    updated.wait()
    result["after"] = router._cache


thread = threading.Thread(target=use_router)
thread.start()
ready.wait()

router.set_mapping({"worker": "10.1.0.44:1234"})
updated.set()
thread.join()

assert result["after"] is result["before"]

The assertion fails because set_mapping() replaces the entire
threading.local() object. In the actual actor caller, the object stored in
the old cache is a live TCP client that remains referenced by
ActorCallerThreadLocal.

Expected behavior

Updating Router mappings should not orphan live clients owned by other
threads.
Xoscar should either:

1. selectively invalidate clients whose resolved destination changed and
   explicitly close them;
2. retain unaffected clients so they can be reused; or
3. maintain a registry of per-thread caches and safely close/evict clients
   from every affected cache.
   Existing in-flight RPCs should not be interrupted unnecessarily.

Additional context

As a temporary Xinference-side workaround, we preserve the original
_cache_local object around calls to:

• Router.set_mapping
• Router.add_router
• Router.remove_router
  After applying this workaround, repeated failed deployments no longer
  increase the number of TCP socket FDs.
  This workaround demonstrates the source of the leak, but it may not be the
  best general fix for Xoscar because a mapping can legitimately change its
  destination. A proper upstream fix should selectively invalidate stale
  entries and close their clients without discarding unaffected per-thread
  caches.