Repository navigation
fix(windows): serve over loopback ZMQ and a selector event loop - #580
YevheniiKotyrlo wants to merge 7 commits into
Conversation
0e50062 to
1b0ee44
Compare
1b0ee44 to
effa99e
Compare
What a serve's ZMQ links meet on WindowsThe links are the ZMQ sockets between the API server, the tokenizer and detokenizer workers and the scheduler. Measured on Windows 11 Pro (build 26300), Python 3.14, pyzmq 27.2.0 with libzmq 4.3.5:
A TCP link brings a hazard of its own. A connect to a port nothing listens on yet is refused and retried until the port is bound, every retry from a new ephemeral source port. If that source port equals the destination, TCP's simultaneous open connects the socket to itself:
Before the last two commits the scheduler bound its backend link only after the engine had loaded its weights, so a tokenizer retried for the whole load, and a server whose tokenizer met itself answered To rerun: ipc and event-loop scriptimport asyncio
import os
import sys
import zmq
import zmq.asyncio
context = zmq.Context()
socket = context.socket(zmq.PULL)
try:
socket.bind(f"ipc:///tmp/freetoken_probe.pid={os.getpid()}")
print(f'zmq.has("ipc") is {zmq.has("ipc")}; an ipc link binds')
except zmq.ZMQError as error:
print(f'zmq.has("ipc") is {zmq.has("ipc")}; an ipc link bind raises ZMQError: {error}')
socket.close(linger=0)
context.term()
async def await_a_poll() -> str:
pull = zmq.asyncio.Context.instance().socket(zmq.PULL)
pull.bind("tcp://127.0.0.1:*")
try:
return f"poll returns {await pull.poll(100)}"
except RuntimeError as error:
return f"RuntimeError: {error}"
print(f"selector loop: {asyncio.SelectorEventLoop().run_until_complete(await_a_poll())}")
if sys.platform == "win32":
print(f"Proactor loop: {asyncio.ProactorEventLoop().run_until_complete(await_a_poll())}")
sys.stdout.flush()
os._exit(0) # a zmq.asyncio socket under the Proactor loop raises again when it is closedSelf-connect scriptimport select
import socket
import time
with socket.socket() as probe:
probe.bind(("127.0.0.1", 0))
port = probe.getsockname()[1]
attempts = itself = 0
source_ports = set()
for _ in range(48):
batch = []
for _ in range(512):
connection = socket.socket()
connection.setblocking(False)
connection.connect_ex(("127.0.0.1", port))
batch.append(connection)
pending, settle_by = list(batch), time.monotonic() + 5
while pending and time.monotonic() < settle_by:
_, writable, failed = select.select([], pending, pending, 0.5)
done = set(writable) | set(failed)
pending = [connection for connection in pending if connection not in done]
for connection in batch:
attempts += 1
source_ports.add(connection.getsockname()[1])
if connection not in pending and connection.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR) == 0:
try:
itself += connection.getpeername() == connection.getsockname()
except OSError:
pass
connection.close()
print(f"port {port}: {attempts} connects from {len(source_ports)} source ports, {itself} connected to themselves")ZMQ peer-type scriptimport socket
import time
import zmq
from zmq.utils.monitor import recv_monitor_message
for stranger_type in (zmq.PUSH, zmq.PUB):
with socket.socket() as probe:
probe.bind(("127.0.0.1", 0))
port = probe.getsockname()[1]
context = zmq.Context()
stranger = context.socket(stranger_type)
stranger.setsockopt(zmq.LINGER, 0)
stranger.bind(f"tcp://127.0.0.1:{port}")
push = context.socket(zmq.PUSH)
push.setsockopt(zmq.SNDTIMEO, 5000)
push.setsockopt(zmq.LINGER, 0)
monitor = push.get_monitor_socket()
push.connect(f"tcp://127.0.0.1:{port}")
time.sleep(1.0)
stranger.close()
time.sleep(0.5)
pull = context.socket(zmq.PULL)
pull.setsockopt(zmq.LINGER, 0)
pull.bind(f"tcp://127.0.0.1:{port}")
try:
push.send(b"request")
verdict = "delivered" if pull.poll(5000) else "sent into a queue nothing drains"
except zmq.Again:
verdict = "send blocked for 5 s: no peer"
events = []
while monitor.poll(0):
events.append(zmq.Event(recv_monitor_message(monitor)["event"]).name)
print(f"stranger {zmq.SocketType(stranger_type).name}: {verdict}; monitor {events}")
push.disable_monitor()
for closing in (push, pull, monitor):
closing.close(linger=0)
context.term() |
The two transport tests already assert five distinct links each, and the platform test restated the branch it ran through.
…an be its own peer
8d5199e to
f75278c
Compare
Part of #384
Problem
With the build fixed (#575),
ft serveon Windows starts and then answers nothing:ipc:///tmp/freetoken_N..., and the libzmq pyzmq ships for Windows has no ipc transport:zmq.has("ipc")isFalseand a bind fails withProtocol not supported(pyzmq 27.2.0, libzmq 4.3.5).zmq.asyncioneedsadd_reader, which that loop does not implement:RuntimeError: Proactor event loop does not implement add_reader family of methods required for zmq._create_listener_oncecreates and drops, so nothing observes it: the server stays up, accepts every request and answers none of them.os.kill(os.getpid(), signal.SIGTERM). On Windows that isTerminateProcess: no handler runs, so uvicorn's lifespan never terminates the remaining workers and they outlive the server. A Flash-Next serve whose scheduler died on a CUDA error left its detokenizer worker running with 1.9 GB./healthand none of its requests.Separately, every load requests expandable segments through
torch.cuda.memory._set_allocator_settings, which torch has deprecated since 2.11, so every start prints a FutureWarning, on Linux too.Solution
SchedulerConfigchooses its five links once, in the parent (_choose_zmq_links): the sameipcpaths as today where libzmq has ipc, loopback TCP ports where it does not. The links travel with the config to every spawned worker.tp_cpu_grouponce the engine has formed it, so its backend port is held for the whole load rather than left free for another program to bind.asyncio:SelectorEventLoopon Windows (UVICORN_LOOP), in both the serve and the shell path.fatal_error,maintenance_state = "failed") and exits the way it already does when the backend dies. This part is not Windows-specific - it is how the failure in (2) surfaced as silence.signal.raise_signal) instead ofos.kill-ing our own pid, so uvicorn's handler runs the lifespan shutdown that terminates the workers. On Linux the two are the same; on Windows only the first runs a handler._ensure_expandable_segmentscallstorch._C._accelerator_setAllocatorSettings, the binding the deprecated wrapper forwards to and the warning names.pyzmq's own message suggests installing tornado to keep the Proactor loop; that adds a dependency and a selector thread per loop, where the selector loop needs neither.
Tests
tests/scheduler/test_zmq_links.py: without ipc every link is a distinct loopback port, and a connect to one leaves from another address than the link's; with ipc the links are the per-process socket paths they were; a pickled config keeps its links on either transport (what a spawned worker receives); the scheduler's backend port is held while the engine loads.tests/server/test_supervisor.py: a backend death stops the server through its own SIGTERM handler - a child process installs one and waits for it on a wakeup socket.tests/engine/test_cache_budget.py: expandable segments are requested without a deprecation warning; without this change the check fails ontorch.cuda._set_allocator_settings is deprecated. Use torch._C._accelerator_setAllocatorSettings instead.Verification
Windows 11, Python 3.14, pyzmq 27.2.0, torch 2.14.1, RTX 3090 Ti, and WSL2 Ubuntu 26.04 with the same Python, torch and pyzmq:
tests/scheduler/test_zmq_links.py6 passed on both; the two new cases fail without the last two changes, onassert '127.0.0.1' != '127.0.0.1'and on a stranger binding the backend port during the load.tests/server/test_supervisor.py13 passed on both; without this change its new test fails on Windows (the child exits 15, killed before its handler runs) and passes on Linux, whereos.killreaches the handler.tests/engine/test_cache_budget.py29 passed and 1 skipped on Linux (WSL's pinning cap); on Windows 28 passed and its twofifixtures fail without flashinfer, as they do onmain(#575 moves them totriton). The evidence comment below has the loopback self-connect and ZMQ's behaviour on it, measured on Windows.Before the links moved to 127.0.0.2, with #575 and #581 on
mainon Windows:python -m freetoken --model-path Qwen3-0.6B --host 127.0.0.1 --port <port> --num-pages 4000was ready in 14-30 s and answered a chat, an Anthropic and a Responses request; its log hadEnabled expandable_segmentsand neitherexpandable_segments not supported on this platformnor the deprecation warning, and Ctrl+C stopped it with exit code 0 and every worker gone. Killing the scheduler of a servingft serve: without this change the server exited 10 s later with code 15 and one worker kept running (1.8 GB); with it uvicorn ran its shutdown (Application shutdown complete.) 11 s after the kill and no worker was left. FreeToken's own queues connecting to a link 0 s and 3 s before it is bound deliver on Windows and Linux.Over the 127.0.0.2 links, with #575 and #581 on
mainat4503c42on Windows: the same command served in 15 s and answered a chat, an Anthropic and a Responses request; the server and its two workers each listened on 127.0.0.2, every connection to them came from 127.0.0.1, and Ctrl+Break stopped it in under a second with no worker left.Not in this PR
A link's port is chosen in the parent by binding it and releasing it, since every worker must know it before any of them binds; a process that takes the port in between makes the link's bind fail, and the start with it.
A hard kill of the server process on Windows (
TerminateProcess, which is whatPopen.terminate()does there) runs no handler, so its two worker processes outlive it:tests/e2e/test_cache_rebuild.pyleaves them behind on every run here, about 1.4 GB and 0.7 GB. A job object with kill-on-close would tie them to the server, a separate change: #869.