From 0293aebb1724d40777c09d753e805ab7dddf3f8f Mon Sep 17 00:00:00 2001 From: Swung0x48 Date: Tue, 22 Sep 2026 21:41:24 -0400 Subject: [PATCH] [MG_Remote, ci] (P7): a bound session closes its data-connection source, so a later DataBind is refused by name instead of queued into a child that stopped reading, and the single-session listener closes with it - ServerMain.cpp: RunSession takes the descriptor its data-connection source reads from (the child's hand-off end under --serve, the listener itself without it) and closes it the moment Accept returns; the source object goes with it - the supervisor's hand-off send is MSG_DONTWAIT (FdPassing::SendFd dontWait), so a peer that stopped reading answers at once; "peer gone" on a datagram pair (ECONNREFUSED, ENOTCONN) is an answer the caller names, not a logged fault - before: every DataBind after the bind sat in the child's receive queue with no reply, its TCP connection held open until the session ended, and past the queue limit (net.unix.max_dgram_qlen 10 on the phone; ~288 datagrams of sender buffer on this host) the supervisor blocked in sendmsg for the rest of the session - no accepts, no Busy, no reaping - before, without --serve: the listener stayed open after the data connection bound, so a second client or a retry completed a TCP handshake nobody would accept and waited out its 20 s connect budget instead of the ECONNREFUSED it got before PH-7 (4) - tcp_supervisor_smoke.py: data_bind_after_bound (12 extra DataBinds on a bound session each get Refuse{Authentication} "data connection could not be handed to the live session", and a fresh control connection is still Busy) and single_session_listener (a connect after `ready` is ECONNREFUSED); connect_control retries ECONNREFUSED at startup for pair() and held_session() alike - red-once at da313010: smoke TimeoutError on the first extra's receive; probe with qlen+2 fire-and-forget DataBinds: 288 queued silently, then the supervisor stopped accepting (fresh control: connect timed out) while still alive; single-session connect after ready succeeded --- MobileGL/MG_Remote/Server/ServerMain.cpp | 46 ++++++++-- MobileGL/MG_Remote/Transport/FdPassing.cpp | 18 +++- MobileGL/MG_Remote/Transport/FdPassing.h | 10 +- scripts/ci/tcp_supervisor_smoke.py | 101 ++++++++++++++++++--- 4 files changed, 150 insertions(+), 25 deletions(-) diff --git a/MobileGL/MG_Remote/Server/ServerMain.cpp b/MobileGL/MG_Remote/Server/ServerMain.cpp index 76cd7674..ed5e5b43 100644 --- a/MobileGL/MG_Remote/Server/ServerMain.cpp +++ b/MobileGL/MG_Remote/Server/ServerMain.cpp @@ -74,6 +74,18 @@ void RefuseBusy(SocketTransport& transport, const char* detail) { // the child that minted it. This replaces AcceptPair's rule that the second connection to // arrive inside the window IS the data plane: over a Windows `adb forward` the two arrived // reordered, the data connection was read as control, and every child exited 67. +// +// THE HAND-OFF HAS ONE READER, AND IT STOPS READING (F fix round, review of slices 1-3). The +// child reads the hand-off only inside BindDataConnection; once its data connection is bound +// it never reads it again. The first shape of this function kept forwarding every later +// DataBind regardless: each descriptor sat in the child's receive queue with no reply, its TCP +// connection was held open until the session ended, and once the queue reached +// net.unix.max_dgram_qlen (10 on the phone's kernel, 512 on a desktop's) the supervisor BLOCKED +// in sendmsg for the rest of the session - no accepts, no Busy refusals, no reaping. Two halves +// close it: the child closes its end of the pair the moment Accept returns (RunSession), so the +// send fails with "peer gone" and the refusal below fires by name; and the send is MSG_DONTWAIT, +// so a queue that is full for any other reason refuses too rather than stalling the one process +// that answers the port. constexpr std::uint32_t kFirstFrameWaitMs = 2000; constexpr std::uint64_t kFirstFrameMaxBytes = 1024 * 1024; @@ -83,7 +95,8 @@ void RouteWhileBusy(SocketTransport& connection, int handoff) { const auto read = SocketTransport::ReceiveOneFrame(fd, kFirstFrameWaitMs, kFirstFrameMaxBytes, &frame); std::uint8_t nonce[kDataNonceBytes] = {}; if (read == MOBILEGL_OK && DecodeDataBind(frame, nonce)) { - if (handoff >= 0 && FdPassing::SendFd(handoff, fd, MobileGLByteSpan{nonce, sizeof(nonce)}) == MOBILEGL_OK) { + if (handoff >= 0 && + FdPassing::SendFd(handoff, fd, MobileGLByteSpan{nonce, sizeof(nonce)}, /*dontWait=*/true) == MOBILEGL_OK) { // The child holds its own descriptor for this connection now; ours goes without a // shutdown(2), which would disconnect the child's too. connection.CloseLocalCopy(); @@ -172,9 +185,11 @@ void SendLogAck(void* pointer) { // Called only after fork: the supervisor never creates a backend or EGL context. // `dataSource` is set on TCP (PH-7 (4)) and empty on a unix endpoint, whose second connection -// is the SCM_RIGHTS socket AcceptPair already paired. +// is the SCM_RIGHTS socket AcceptPair already paired. `sourceFd` is the descriptor that source +// reads from - the child's end of the hand-off socketpair under --serve, the listener itself in +// the single-session shape - and -1 when there is none; it is closed the moment Accept returns. [[noreturn]] void RunSession(std::unique_ptr control, - Server::ServerSession::DataConnectionSource dataSource) { + Server::ServerSession::DataConnectionSource dataSource, int sourceFd) { const int selfPid = static_cast(::getpid()); const bool tcp = control->IsTcp(); std::vector helloFrame; @@ -237,6 +252,21 @@ void SendLogAck(void* pointer) { } if (tcp) session.SetDataConnectionSource(std::move(dataSource)); const auto accepted = session.Accept(*control, &helloFrame); + // THE SOURCE IS DONE WHEN ACCEPT IS, whichever way Accept went (F fix round). Under --serve + // this is the child's end of the hand-off pair: closing it turns every later DataBind the + // supervisor would have queued here into a send failure the supervisor refuses by name on + // that connection ("data connection could not be handed to the live session"), instead of a + // descriptor nobody will read holding a peer's connection open and, past max_dgram_qlen, + // holding the supervisor itself. In the single-session shape it is the listener: kept open + // only until the data connection arrived on it, and closed here so a second client or a + // retry is refused at connect (ECONNREFUSED, as before PH-7 (4)) rather than completing a + // TCP handshake this process would never accept and waiting out its whole connect budget. + // The source object goes with it, so no later call can poll a descriptor number the kernel + // may have reused. + if (sourceFd >= 0) { + ::close(sourceFd); + session.SetDataConnectionSource({}); + } if (accepted != MOBILEGL_OK) { WireLogError("MG_Remote server: pid=%d Accept failed (rc=%d)", selfPid, static_cast(accepted)); std::fflush(nullptr); @@ -498,13 +528,14 @@ extern "C" __attribute__((visibility("default"))) int mobilegl_server_main(int a if (accepted != MOBILEGL_OK) { LogSupervisorSummary(sessionsFaulted); ::close(listener); return 73; } if (!serve) { if (tcpEndpoint) { - // The listener stays open: this session's data connection has yet to arrive on it. + // The listener stays open until this session's data connection has arrived on + // it; RunSession closes it the moment Accept returns. const int controlFd = control->StreamFd(); - RunSession(std::move(control), ListenerSource(listener, controlFd)); + RunSession(std::move(control), ListenerSource(listener, controlFd), listener); } ::close(listener); if (endpoint[0] != '@') ::unlink(endpoint.c_str()); - RunSession(std::move(control), {}); + RunSession(std::move(control), {}, -1); } reap(); // exit_group closes files just before waitpid can observe the exit. @@ -531,7 +562,8 @@ extern "C" __attribute__((visibility("default"))) int mobilegl_server_main(int a if (pair[0] >= 0) ::close(pair[0]); const int controlFd = control->StreamFd(); RunSession(std::move(control), tcpEndpoint ? HandoffSource(pair[1], controlFd) - : Server::ServerSession::DataConnectionSource{}); + : Server::ServerSession::DataConnectionSource{}, + pair[1]); } if (child < 0) { if (pair[0] >= 0) ::close(pair[0]); diff --git a/MobileGL/MG_Remote/Transport/FdPassing.cpp b/MobileGL/MG_Remote/Transport/FdPassing.cpp index a2dff4c4..578bd2ad 100644 --- a/MobileGL/MG_Remote/Transport/FdPassing.cpp +++ b/MobileGL/MG_Remote/Transport/FdPassing.cpp @@ -38,7 +38,7 @@ namespace MobileGL::MG_Remote::Transport::FdPassing { MobileGLResult CreateSocketPair(int[2]) { return MOBILEGL_ERR_UNSUPPORTED; } - MobileGLResult SendFd(int, int, MobileGLByteSpan) { return MOBILEGL_ERR_UNSUPPORTED; } + MobileGLResult SendFd(int, int, MobileGLByteSpan, bool) { return MOBILEGL_ERR_UNSUPPORTED; } MobileGLResult ReceiveFd(int, int*, MobileGLMutableByteSpan, std::uint64_t*, std::uint32_t) { return MOBILEGL_ERR_UNSUPPORTED; @@ -105,7 +105,7 @@ namespace MobileGL::MG_Remote::Transport::FdPassing { return MOBILEGL_OK; } - MobileGLResult SendFd(int socket, int fd, MobileGLByteSpan sideband) { + MobileGLResult SendFd(int socket, int fd, MobileGLByteSpan sideband, bool dontWait) { if (socket < 0 || fd < 0) { return MOBILEGL_ERR_INVALID_ARGUMENT; } @@ -148,8 +148,9 @@ namespace MobileGL::MG_Remote::Transport::FdPassing { cmsg->cmsg_len = CMSG_LEN(sizeof(int)); std::memcpy(CMSG_DATA(cmsg), &fd, sizeof(fd)); + const int flags = MSG_NOSIGNAL | (dontWait ? MSG_DONTWAIT : 0); for (;;) { - const ssize_t sent = ::sendmsg(socket, &msg, MSG_NOSIGNAL); + const ssize_t sent = ::sendmsg(socket, &msg, flags); if (sent >= 0) { if (static_cast(sent) != payloadSize) { // A datagram socket sends all or nothing. @@ -162,9 +163,18 @@ namespace MobileGL::MG_Remote::Transport::FdPassing { if (errno == EINTR) { continue; } - if (errno == EPIPE || errno == ECONNRESET) { + // A connected AF_UNIX datagram socket whose peer closed answers ECONNREFUSED on the + // first send after the close and ENOTCONN on every later one; a stream peer answers + // EPIPE / ECONNRESET. All four are "the peer is gone", which is an answer and not a + // fault, so none of them is logged here - the caller names what it does about it. + if (errno == EPIPE || errno == ECONNRESET || errno == ECONNREFUSED || errno == ENOTCONN) { return MOBILEGL_ERR_TRANSPORT_CLOSED; } + if (dontWait && (errno == EAGAIN || errno == EWOULDBLOCK)) { + MGLOG_E("MG_Remote fd passing: the peer's receive queue is full and it is not " + "reading; the descriptor was not queued"); + return MOBILEGL_ERR_TIMEOUT; + } MGLOG_E("MG_Remote fd passing: sendmsg failed (errno=%d)", errno); return MOBILEGL_ERR_TRANSPORT_CLOSED; } diff --git a/MobileGL/MG_Remote/Transport/FdPassing.h b/MobileGL/MG_Remote/Transport/FdPassing.h index d3ccb5c4..66989e52 100644 --- a/MobileGL/MG_Remote/Transport/FdPassing.h +++ b/MobileGL/MG_Remote/Transport/FdPassing.h @@ -48,7 +48,15 @@ namespace MobileGL::MG_Remote::Transport::FdPassing { // Sends `fd` with `sideband` attached. The caller keeps ownership of `fd` // (the peer gets its own descriptor for the same open file description). // sideband.size must be <= kMaxSidebandBytes. - MobileGLResult SendFd(int socket, int fd, MobileGLByteSpan sideband); + // + // `dontWait` sends with MSG_DONTWAIT: a receive queue the peer is not draining + // (net.unix.max_dgram_qlen datagrams, 10 on an Android kernel) answers + // MOBILEGL_ERR_TIMEOUT at once instead of blocking the sender until the peer + // reads. A peer that has closed its end answers MOBILEGL_ERR_TRANSPORT_CLOSED + // either way. The TCP supervisor's hand-off (ServerMain.cpp RouteWhileBusy) + // sends this way because the session child stops reading the hand-off once its + // data connection is bound, and a supervisor blocked in sendmsg serves nobody. + MobileGLResult SendFd(int socket, int fd, MobileGLByteSpan sideband, bool dontWait = false); // Receives one descriptor and its sideband bytes. // diff --git a/scripts/ci/tcp_supervisor_smoke.py b/scripts/ci/tcp_supervisor_smoke.py index 981b8654..214f9471 100644 --- a/scripts/ci/tcp_supervisor_smoke.py +++ b/scripts/ci/tcp_supervisor_smoke.py @@ -222,13 +222,16 @@ def refused_listen(server, log, endpoint, token): @contextmanager -def supervisor(server, log, same_build=False): +def supervisor(server, log, same_build=False, serve=True): + """A supervisor on its own free loopback port. `serve=False` is the single-session shape + (no fork; the process IS the session and exits with it), which `single_session_listener` + below needs and nothing else does.""" port = free_loopback_port() base = log.with_suffix('.library.log') env = supervisor_environment(base, kToken, same_build=same_build) with log.open('wb') as output: - child = subprocess.Popen([server, f'tcp://127.0.0.1:{port}', '--serve'], env=env, - stdout=output, stderr=output, start_new_session=True) + child = subprocess.Popen([server, f'tcp://127.0.0.1:{port}'] + (['--serve'] if serve else []), + env=env, stdout=output, stderr=output, start_new_session=True) try: # Log.cpp's RoleLogPath: the base name is a BASE, and the server role writes # `.server`. Yielded rather than recomputed at the call site, so the one @@ -242,6 +245,20 @@ def supervisor(server, log, same_build=False): child.wait(timeout=10) +def connect_control(port, seconds=5): + """A connection to the supervisor's port, retried on ECONNREFUSED for `seconds`: a supervisor + just started may not have reached listen() yet, and a refused connect before it has is + startup, not a finding.""" + deadline = time.monotonic() + seconds + while True: + try: + return socket.create_connection(('127.0.0.1', port), timeout=3) + except ConnectionRefusedError: + if time.monotonic() >= deadline: + raise + time.sleep(.02) + + @contextmanager def held_session(port, schema, flatbuffers, message): """A welcomed session whose sockets STAY OPEN, so its child is still alive to be killed. @@ -258,7 +275,7 @@ def held_session(port, schema, flatbuffers, message): data = None reply = None for _ in range(30): - control = socket.create_connection(('127.0.0.1', port), timeout=3) + control = connect_control(port) control.sendall(message) reply = receive(control, schema) if reply.get('code') != 7: @@ -305,15 +322,7 @@ def pair(port): two connections paired by arrival order, and this opened both. The data connection is now opened after Welcome with the nonce it carries (open_data), so a control that stops at the handshake - every one but held_session's - never opens one.""" - deadline = time.monotonic() + 5 - while True: - try: - control = socket.create_connection(('127.0.0.1', port), timeout=3) - break - except ConnectionRefusedError: - if time.monotonic() >= deadline: - raise - time.sleep(.02) + control = connect_control(port) try: yield control finally: @@ -543,10 +552,76 @@ def main(): assert peak < ask, f'session pid={welcome["welcome"]} reached VmPeak={peak} for a 64 GiB ask' evidence['hello_asks_64_gib'] = {'asked': ask, 'terms': terms, 'childVmPeakBytes': peak} + # ---- F fix round: A BOUND SESSION'S SUPERVISOR REFUSES EVERY LATER DataBind BY NAME -- + # + # The session child reads its hand-off socketpair only while binding; after that it + # never reads it again. Until this control the supervisor kept forwarding every later + # DataBind into it anyway: each descriptor sat in the child's receive queue with NO + # reply, its TCP connection was held open until the session ended, and once the queue + # reached net.unix.max_dgram_qlen (10 on the phone's kernel, 512 on this host's) the + # supervisor blocked in sendmsg for the rest of the session - no accepts, no Busy, no + # reaping. Now the child closes its end the moment Accept returns and the send is + # non-blocking, so each of these twelve is answered on its own connection with + # Refuse{Authentication} "data connection could not be handed to the live session", + # and a fresh control connection afterwards is still answered Busy. RED before the fix + # at the very first `receive`: a 3 s socket timeout, because nothing ever answered. + with held_session(port, schema, flatbuffers, + hello(schema, flatbuffers, fingerprint=fingerprint)) as (control, welcome): + assert welcome.get('welcome', 0), welcome + # Bound, not merely welcomed: an extra DataBind that arrived while the child was + # still in BindDataConnection would be forwarded and refused as a nonce mismatch, + # which is the other control's line, not this one's. + ready = wait_for_log(serverLog, f'pid={welcome["welcome"]} transport=spawn role=server ready') + assert ready is not None, f'the held session never came up; see {serverLog}' + answered = [] + for _ in range(12): + extra = open_data(port, schema, flatbuffers, os.urandom(16)) + try: + answered.append(receive(extra, schema)) + finally: + extra.close() + for answer in answered: + assert answer.get('code') == 4 and answer.get('detail') == ( + 'data connection could not be handed to the live session'), answer + with pair(port) as fresh: + fresh.sendall(hello(schema, flatbuffers, fingerprint=fingerprint)) + still_busy = receive(fresh, schema) + assert still_busy.get('code') == 7, still_busy + evidence['data_bind_after_bound'] = {'refused': len(answered), 'last': answered[-1], + 'fresh_control': still_busy} + assert process.poll() is None, 'the refused DataBinds killed the supervisor' + with supervisor(args.server, args.out / 'same-build-supervisor.log', same_build=True) as (port, _, _log): strict = exchange(port, schema, hello(schema, flatbuffers, fingerprint=fingerprint)) assert strict.get('code') == 3, strict evidence['same_build_required'] = strict + + # ---- F fix round: THE SINGLE-SESSION SERVER CLOSES ITS LISTENER ONCE THE SESSION IS BOUND + # + # Without --serve the process is the session; it keeps the listener open only so the + # session's data connection can arrive on it (ListenerSource). Until this control nothing + # closed it afterwards, so the kernel went on completing TCP handshakes into a backlog no + # accept() would ever drain: a second client, or a retry, hung for its whole connect budget + # (kSpawnConnectTimeoutMs, 20 s) instead of the ECONNREFUSED it got before PH-7 (4). Now + # RunSession closes the listener the moment Accept returns, and a connect after `ready` + # is refused at once. RED before the fix: the connect below succeeds. + with supervisor(args.server, args.out / 'single-session.log', serve=False) as (port, process, serverLog): + with held_session(port, schema, flatbuffers, + hello(schema, flatbuffers, fingerprint=fingerprint)) as (control, welcome): + assert welcome.get('welcome', 0), welcome + ready = wait_for_log(serverLog, f'pid={welcome["welcome"]} transport=spawn role=server ready') + assert ready is not None, f'the single session never came up; see {serverLog}' + try: + second = socket.create_connection(('127.0.0.1', port), timeout=3) + except ConnectionRefusedError as refused: + second = None + at_connect = repr(refused) + else: + second.close() + assert second is None, ( + 'the single-session server still completed a TCP handshake after its data ' + 'connection bound; the listener was not closed') + evidence['single_session_listener'] = {'second_connect': at_connect} (args.out / 'protocol-controls.json').write_text(json.dumps(evidence, indent=2)) print(json.dumps(evidence, indent=2)) return 0