diff options
| -rw-r--r-- | CLAUDE.md | 53 | ||||
| -rw-r--r-- | docs/MESHBAY_DESIGN.md | 57 | ||||
| -rw-r--r-- | docs/USERGUIDE.md | 13 | ||||
| -rw-r--r-- | packages/meshbay-cli/tests/test_e2e_local.py | 32 | ||||
| -rw-r--r-- | packages/meshbay-client/build/appx-extensions.xml | 7 | ||||
| -rw-r--r-- | packages/meshbay-client/build/appx-manifest.xml | 21 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc/core.py | 1 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc/proactor_udp.py | 139 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc/worker.py | 5 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_packaging_win.py | 21 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_proactor_udp.py | 150 | ||||
| -rw-r--r-- | packaging/win/README.md | 2 |
12 files changed, 469 insertions, 32 deletions
@@ -761,6 +761,59 @@ do. Read them before writing anything that touches the same mechanism. socket buffer gave 19.5 MB/s (`udp_buffer.py`); it needs `net.core.rmem_max` raised on a default machine +- **A transfer capped with a core busy: profile that process before tuning + the network.** The command-line client downloaded at 19 MB/s on loopback. + Its 4 MB receive buffer removed 3245 kernel drops and changed nothing else; + `top` then showed the client, not the node, at 100 %, and cProfile (a + wrapper script as `MESHBAY_CLI_E2E_BIN`, the e2e harness does the rest) put + two thirds of its time in aiortc's `InboundStream.pop_messages`: 153 million + steps for 400 MB, the reassembly walked from its start after every fragment. + Next, a SACK per packet cost the node more than sending. Both fixed in + `sctp_fix.py`: 39 MB/s, the node at 49. No amount of reading our code finds + a quadratic loop in a dependency; one profile does + +- **A ceiling that falls as the loop gets busier is not a CPU ceiling.** + Windows downloads stayed slow with no core saturated: the node at 45 % of a + core, 9 MB/s over eight connections, and an SCTP round trip of 745 ms on + loopback, which is a queue, not a wire. The ProactorEventLoop's datagram + transport sends one datagram per turn of the loop and reads one, each through + an overlapped operation (42 % of a connection process was + `IocpProactor.sendto`), so every other thing the loop does slows every + connection. Linux sends straight from the socket. `proactor_udp.py` does the + same on Windows; a benchmark only showed it once its loop had other work to + do, because an idle loop turns fast enough to hide it. When a transfer is slow, compare the round trip SCTP reports with the + real one before profiling: a large gap is a queue in this process + +- **A default of "off" ships a feature nobody has.** `connection_processes` + defaulted to 1, which starts no process: the multi-core path was measured at + 131 MB/s and reached by no node whose operator had not found the setting, so + on Windows a download over eight lanes landed on the node's one core and ran + at the speed of one connection. Worse, the Node page saved **every** setting + on each Save, so any operator who changed one thing had pinned all the others + at that day's defaults, in the roster and in `node.toml`. It sends only what + changed now, and `init` leaves the line commented. A test node that saved + its settings before this keeps `connection_processes = 1` until it is set + again on the Node page + +- **A reconnect let every message through, not just its handshake.** + `_inReconnectAttempt` opened `_sendAndWait` to all callers so the handshake + would not wait on itself; a chunk retry landing between the new channel's + open and the proof got "Handshake required", which no retry treats as a + dropped connection, and the download died with that text in the transfers + panel. Only the opening messages pass now (`SESSION_OPENING`, held to + `dispatch.py` by `test_reconnect_race.py`), and a replaced channel's late + `close` no longer clobbers the new one. `harness/reconnect_race_harness.mjs` + runs the real `connect()`, `_reconnectLoop` and pipeline against a node that + refuses before the proof — the first harness here that does; the earlier + ones stub `connect`, which is exactly where this lived + +- **A push handler that re-fetches the world.** The interactive CLI answered + every `index_delta` by fetching and re-parsing the whole index, on the event + loop that was receiving the download: 67 full indexes in 80 s while a node + enriched a library (one delta a second). It applies the delta in place + (`Tree.apply_delta`), and only a delta that adds or removes files asks again, + once per burst + ### More of the same, from four rounds of live testing - **A protocol harness cannot test the SPA.** Any second implementation of the diff --git a/docs/MESHBAY_DESIGN.md b/docs/MESHBAY_DESIGN.md index 4e99994..4db06c4 100644 --- a/docs/MESHBAY_DESIGN.md +++ b/docs/MESHBAY_DESIGN.md @@ -1486,9 +1486,12 @@ download land on different processes without the client doing anything. | 8 | 150 MB/s, the receiving browser being the limit | 21 % | Measured on loopback, a 1.5 GB file through the interface in a headless browser, -byte for byte. A process costs about 85 MB of memory. The default is 1, which -starts none and leaves the node exactly as it was; the operator raises it. A -change applies to new connections, and a connection already open stays where it +byte for byte. A process costs about 85 MB of memory. The default is one per +core, up to four: past four a single download gains nothing, and a process is +memory held for the life of the node. One starts none, and leaves every +connection, every lane of a download included, on the node's own core — the +case where Python's per-packet cost is the whole ceiling, and highest on Windows. +A change applies to new connections, and a connection already open stays where it is until it closes, in a process asked to leave or in the node. A process that dies takes its own connections with it, which their clients see as a dropped connection and reopen, and is replaced at the next offer. The loopback socket is @@ -1512,6 +1515,52 @@ machine nothing changes until the operator raises it; the node says so at startu The limit that remains is the node's own receive path, about 5 % of a core per MB/s. +**A receiver on aiortc reassembles in linear time and acknowledges every second +packet.** That is the command-line client in a download and the node in an +upload. aiortc 1.15 walks the reassembly from its start after every DATA chunk, +so a message of n fragments costs n²/2 steps — a 1 MiB `file_chunk` is about 870 +— and it acknowledges every packet, where RFC 4960 §6.2 asks for one SACK every +second packet; each SACK costs the sender a DTLS decryption and a parse on the +core that also sends. `sctp_fix.py` (the node's, and the client's copy) skips the +walk until a last fragment is in, and acknowledges every second packet of data, +at once on a gap, a duplicate or a packet carrying anything else, and within +200 ms otherwise. The client asks for a 4 MB receive buffer, as the node does. +Measured on loopback, 400 MB to the command-line client over one connection: 19 +MB/s with 3245 datagrams dropped by its kernel; none dropped with the buffer, and +still 20, the client at a full core; 28 with the reassembly fixed; 39 with the +acknowledgements, 49 by the node's own count of the transfer. An upload from the +client went from 25 to 29. + +**On Windows a datagram goes straight through the socket, in both directions.** +The node and its connection processes run on the ProactorEventLoop, the one that +can start ffmpeg there, and its datagram transport keeps one overlapped operation +per direction, starting the next from the completion callback of the one before, +on a later turn of the loop. A socket therefore sent one datagram per turn and +received one, and a turn lasts as long as everything else the loop does: the +other connections, the file, the hub. That is a ceiling per connection that +falls as the node gets busier, with cores to spare. With the loop otherwise busy +0.5 ms a turn: 1,928 datagrams sent a second, and 1,954 received with 87 % of +what arrived dropped by the kernel. In a node holding its connections itself, a +browser's download over eight of them ran at 9 MB/s with the node at 45 % of a +core and an SCTP round trip of 745 ms on loopback, and one of the eight sessions +was lost on the way. `proactor_udp.py` sends with a non-blocking `sendto` when +nothing is queued, and after a completed read takes whatever else the socket +holds before posting the next one; refusals and errors take the proactor's own +path. Measured on loopback on a two-core machine, 400 MB, with it: the +command-line client's download 27 to 33 MB/s, its upload 18 to 22 (23 with two +connection processes), the browser's download over eight connections 9 to 14 +before and 17 to 19 after, where the browser takes the other core. + +**A session is open from the end of its handshake, and a reconnect does not +shorten that.** Until connect() returns, a client sends only what the node takes +before the proof — `handshake`, `handshake_response`, `keypair_bundle_fetch`, +`gek_bundle_fetch`, `join_request`, `device_add_request` — and `device_hello`; +anything else waits for the session, or fails as a dropped connection, which a +download waits out. The node answers anything else with "Handshake required", a +refusal and not a dropped connection, so a chunk let through while a reconnect's +handshake runs ends the download the reconnect is for. The close event of a +channel a reconnect has replaced concerns nothing on the new one. + The lease module is free of asyncio and of the transport: it decides, and the caller does the I/O. A queue that reveals itself only through a DataChannel is a queue nobody can prove things about. @@ -2163,7 +2212,7 @@ Node page: | `max_concurrent_downloads` | 8 | node-wide download leases (§5.5) | | `max_concurrent_uploads` | 8 | node-wide upload leases. A separate pool from downloads, because the two cost different things and one queue for both makes each cap meaningless | | `max_upload_gb` | 8 | the largest single upload, §6.4. Fractions are allowed, and it is read per chunk so a change reaches an upload already running | -| `connection_processes` | 1 | how many processes carry the peer connections, §5.5. One is the node itself; above it, that many cores serve transfers. Bounded by the cores of the machine, which the Node page shows beside the field | +| `connection_processes` | one per core, up to 4 | how many processes carry the peer connections, §5.5. One is the node itself; above it, that many cores serve transfers. Bounded by the cores of the machine, which the Node page shows beside the field. `node.toml` as `init` writes it leaves the line commented, so the default reaches every node that has not chosen | | `transcode_incompatible_video` | true | whether browser-incompatible video is transcoded during streaming. Unlike remuxing this costs real CPU per viewer | `hardware_video_encode` (default true) governs whether that transcode may use the diff --git a/docs/USERGUIDE.md b/docs/USERGUIDE.md index 1808598..ab5ee8c 100644 --- a/docs/USERGUIDE.md +++ b/docs/USERGUIDE.md @@ -852,15 +852,16 @@ and running `meshbay-node reload`: | `max_concurrent_downloads` | 8 | node-wide download slots | | `max_concurrent_uploads` | 8 | node-wide upload slots | | `max_upload_gb` | 8 | the largest single file a member may send you | -| `connection_processes` | 1 | how many CPU cores carry members' connections | +| `connection_processes` | one per core, up to 4 | how many CPU cores carry members' connections | | `transcode_incompatible_video` | true | re-encode video no browser can play | | `hardware_video_encode` | true | use the GPU for that, if one actually works | -`connection_processes` is the one for speed. With 1, everything a node sends -goes through one core, about 35 MB/s shared between everyone downloading at -that moment. Raise it when several people download at once and the machine has -cores to spare: each one adds a core's worth, and about 85 MB of memory. It -applies to connections opened after the change. +`connection_processes` is the one for speed. Each process is a core's worth +of sending, shared between everyone downloading at that moment, and about 85 MB +of memory. With 1, everything a node sends goes through one core: about 35 MB/s +on Linux, much less on Windows. Lower it on a machine short of memory; raise it +above 4 only when many people download at once. It applies to connections +opened after the change. `max_concurrent_streams` is the one to think about: a stream costs a video process for as long as the film lasts, so this counts **viewers**, not diff --git a/packages/meshbay-cli/tests/test_e2e_local.py b/packages/meshbay-cli/tests/test_e2e_local.py index 85e5e85..385693d 100644 --- a/packages/meshbay-cli/tests/test_e2e_local.py +++ b/packages/meshbay-cli/tests/test_e2e_local.py @@ -196,7 +196,14 @@ def _build(root: Path, procs: list) -> dict: _wait_http(f"http://127.0.0.1:{ui_port}/", timeout=90) # Attaching reloads the node by itself; the key needs the group loaded. - run_node("group", "add", group_id, "--dir", str(share)) + # The control API answers before the hub socket is up, later still now + # that the connection processes start first. + deadline = time.monotonic() + 90 + while "Hub not connected" in (out := _try(run_node, "group", "add", group_id, + "--dir", str(share))): + assert time.monotonic() < deadline, f"group add never succeeded: {out}" + time.sleep(1) + assert "meshbay-node (" not in out, out deadline = time.monotonic() + 90 while "GEK ready" not in (out := _try(run_node, "gek", "init", "--group", group_id)): assert time.monotonic() < deadline, f"gek init never succeeded: {out}" @@ -1379,17 +1386,18 @@ def test_tab_completes_remote_paths_from_the_real_node(world): assert complete("ls", "No such group/") == [] # nothing, quietly # bash itself: the function the script installs, on a word typed with its - # spaces escaped, as bash hands it over. - script = cli(world, "completion", "bash").stdout - escaped = group[:10].replace(" ", "\\ ") - bash = subprocess.run( - ["bash", "-c", script + "\nCOMP_WORDS=(meshbay-cli ls " + escaped + ")\n" - "COMP_CWORD=2\n_meshbay_cli\nprintf '%s\\n' \"${COMPREPLY[@]}\"\n"], - env={**os.environ, "MESHBAY_CLI_HOME": str(world["root"] / "cli-chat"), - "PYTHON_KEYRING_BACKEND": "keyring.backends.fail.Keyring", - "PATH": f"{BIN}{os.pathsep}{os.environ['PATH']}"}, - capture_output=True, text=True, encoding="utf-8", timeout=60) - assert bash.stdout.splitlines() == [group + "/"], (bash.stdout, bash.stderr) + # spaces escaped, as bash hands it over. On Windows, where Git's is on PATH. + if shutil.which("bash"): + script = cli(world, "completion", "bash").stdout + escaped = group[:10].replace(" ", "\\ ") + bash = subprocess.run( + ["bash", "-c", script + "\nCOMP_WORDS=(meshbay-cli ls " + escaped + ")\n" + "COMP_CWORD=2\n_meshbay_cli\nprintf '%s\\n' \"${COMPREPLY[@]}\"\n"], + env={**os.environ, "MESHBAY_CLI_HOME": str(world["root"] / "cli-chat"), + "PYTHON_KEYRING_BACKEND": "keyring.backends.fail.Keyring", + "PATH": f"{BIN}{os.pathsep}{os.environ['PATH']}"}, + capture_output=True, text=True, encoding="utf-8", timeout=60) + assert bash.stdout.splitlines() == [group + "/"], (bash.stdout, bash.stderr) # PowerShell 7, or Windows PowerShell 5.1, which decodes a program's output # with the console's code page unless told otherwise: an accented name is the diff --git a/packages/meshbay-client/build/appx-extensions.xml b/packages/meshbay-client/build/appx-extensions.xml index db6e345..3326dd8 100644 --- a/packages/meshbay-client/build/appx-extensions.xml +++ b/packages/meshbay-client/build/appx-extensions.xml @@ -13,10 +13,3 @@ <uap5:ExecutionAlias Alias="meshbay-node.exe" /> </uap5:AppExecutionAlias> </uap5:Extension> - <!-- `meshbay-cli` in a terminal, the command-line client: the same alias - mechanism, for the same reason. --> - <uap5:Extension Category="windows.appExecutionAlias" Executable="app\resources\node-runtime\meshbay-cli.exe" EntryPoint="Windows.FullTrustApplication"> - <uap5:AppExecutionAlias> - <uap5:ExecutionAlias Alias="meshbay-cli.exe" /> - </uap5:AppExecutionAlias> - </uap5:Extension> diff --git a/packages/meshbay-client/build/appx-manifest.xml b/packages/meshbay-client/build/appx-manifest.xml index c097fdd..0542bd5 100644 --- a/packages/meshbay-client/build/appx-manifest.xml +++ b/packages/meshbay-client/build/appx-manifest.xml @@ -43,6 +43,27 @@ </uap:VisualElements> ${extensions} </Application> + <!-- `meshbay-cli` in a terminal, by the same alias mechanism as the node. + An application declares one execution alias, and meshbay-node.exe has + the app's (appx-extensions.xml): a second one there made makeappx + refuse the package. So the command-line client is an application of + its own, listed nowhere. --> + <Application Id="MeshBayCli" Executable="app\resources\node-runtime\meshbay-cli.exe" EntryPoint="Windows.FullTrustApplication"> + <uap:VisualElements + AppListEntry="none" + BackgroundColor="${backgroundColor}" + DisplayName="MeshBay command-line client" + Square150x150Logo="${square150x150Logo}" + Square44x44Logo="${square44x44Logo}" + Description="MeshBay command-line client" /> + <Extensions> + <uap5:Extension Category="windows.appExecutionAlias" Executable="app\resources\node-runtime\meshbay-cli.exe" EntryPoint="Windows.FullTrustApplication"> + <uap5:AppExecutionAlias> + <uap5:ExecutionAlias Alias="meshbay-cli.exe" /> + </uap5:AppExecutionAlias> + </uap5:Extension> + </Extensions> + </Application> </Applications> <!-- The node accepts connections from peers (WebRTC, QUIC, casting), and the command-line client does during a transfer (WebRTC). Rules declared diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/core.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/core.py index d197044..07820b2 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/core.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/core.py @@ -15,6 +15,7 @@ from meshbay_common.protocol import MNP from meshbay_node import ops from meshbay_node import transfers as transfers_mod from meshbay_node.transport.webrtc import ( + proactor_udp, # noqa: F401 (patches asyncio's Windows loop on import) sctp_fix, # noqa: F401 (patches aiortc on import) sctp_stats, # noqa: F401 (patches aiortc on import) udp_buffer, # noqa: F401 (patches aioice on import) diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/proactor_udp.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/proactor_udp.py new file mode 100644 index 0000000..38cb97c --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/proactor_udp.py @@ -0,0 +1,139 @@ +"""Datagrams moved straight through the socket on Windows' event loop, applied +on import. + +The node runs on the ProactorEventLoop, the one that can start ffmpeg there +(`platform.configure_event_loop`), and so do its connection processes. Its +datagram transport keeps one overlapped operation in each direction and starts +the next only from the completion callback of the one before, which runs on a +later turn of the loop. So a socket sends **one datagram per turn of the loop, +and receives one**, however many are waiting, and a turn lasts as long as +everything else the loop has to do: the other connections, reading the file, +the hub. The busier the node, the slower every connection, with cores to spare. + +Measured on loopback, 1200-byte datagrams, with the loop otherwise busy 0.5 ms +a turn: 1,928 sent a second (2.3 MB/s), and 1,954 received, the kernel dropping +87 % of what arrived; straight through the socket, 88,000 sent, and every +datagram received at the 20,000 a second it was sent at. In the node holding the +connections itself, a browser's download over eight connections ran at 9.1 MB/s +with the node at 45 % of a core and an SCTP round trip of 745 ms on loopback: +that is the queue in front of each socket. In a connection process serving the +command-line client, 42 % of the time was `IocpProactor.sendto`. Linux never +paid for the sending half: the selector loop's transport tries the socket first +and queues only what it refuses. + +Sending: a datagram goes out with a non-blocking `sendto` when nothing is queued +and no overlapped send is pending, so order is kept; anything else, the socket +refusing included, goes through the proactor's own queue as before. + +Receiving: once an overlapped read completes, its datagram is delivered, then +whatever else the socket holds is read with non-blocking `recvfrom`, and only +then is the next overlapped read posted. Never while one is pending, which +could hand a later datagram to the completion port ahead of the ones read here. +The overlapped path turns the ICMP "port unreachable" a send to a silent +address brings back into an empty read; read directly it is a +ConnectionResetError, and it is skipped the same way. + +No effect off Windows, where the class is never used. +""" + +import sys +from asyncio import exceptions, proactor_events + +_Transport = proactor_events._ProactorDatagramTransport +_proactor_sendto = _Transport.sendto + +# Datagrams read straight from the socket after one completion, at most: a +# socket under a flood still lets the loop turn. +DRAIN_MAX = 256 + + +def sendto(self, data, addr=None) -> None: + if (self._write_fut is None and not self._buffer and not self._conn_lost + and isinstance(data, (bytes, bytearray, memoryview)) and data + and (self._address is None or addr in (None, self._address))): + try: + if self._address is not None: + self._sock.send(data) + else: + self._sock.sendto(data, addr) + return + except (BlockingIOError, InterruptedError): + pass + except OSError as exc: + # What the overlapped path does with a failed send. + self._protocol.error_received(exc) + return + _proactor_sendto(self, data, addr) + + +def _deliver(self, data, addr) -> None: + """To the protocol, and whatever it raises reported as a callback's would + be: in CPython's order the next read was posted first, and an exception + here must not end the reading either.""" + try: + self._protocol.datagram_received(data, addr) + except (SystemExit, KeyboardInterrupt): + raise + except BaseException as exc: + self._loop.call_exception_handler({ + "message": "Exception in datagram_received", "exception": exc, + "transport": self, "protocol": self._protocol}) + + +def _drain(self) -> None: + for _ in range(DRAIN_MAX): + if self._conn_lost or self._closing: + return + try: + if self._address is not None: + data, addr = self._sock.recv(self.max_size), self._address + else: + data, addr = self._sock.recvfrom(self.max_size) + except (BlockingIOError, InterruptedError): + return + except ConnectionResetError: + continue + except OSError as exc: + # The overlapped read posted next meets it again, and reports it + # the proactor's way. + self._protocol.error_received(exc) + return + if data: + _deliver(self, data, addr) + + +def _loop_reading(self, fut=None) -> None: + """CPython's own (3.12), but a completed read is delivered and the socket + drained before the next overlapped read is posted, not after.""" + try: + if self._conn_lost: + return + assert self._read_fut is fut or (self._read_fut is None and self._closing) + self._read_fut = None + if fut is not None: + res = fut.result() + if self._closing: + return + data, addr = (res, self._address) if self._address is not None else res + if data: + _deliver(self, data, addr) + _drain(self) + if self._conn_lost: + return + if self._address is not None: + self._read_fut = self._loop._proactor.recv(self._sock, self.max_size) + else: + self._read_fut = self._loop._proactor.recvfrom(self._sock, self.max_size) + except OSError as exc: + self._protocol.error_received(exc) + except exceptions.CancelledError: + if not self._closing: + raise + else: + if self._read_fut is not None: + self._read_fut.add_done_callback(self._loop_reading) + + +if sys.platform == "win32": + _Transport.sendto = sendto + _Transport._loop_reading = _loop_reading diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/worker.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/worker.py index 832428a..c933aba 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/worker.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/worker.py @@ -224,9 +224,10 @@ async def _main() -> None: root.handlers[:] = [_Forward(writer.write)] root.setLevel(hello.get("level", logging.INFO)) - # Patches aiortc on import, as the node does for the connections it holds - # itself. + # Patches aiortc, and asyncio's Windows loop, on import, as the node does + # for the connections it holds itself. from meshbay_node.transport.webrtc import ( # noqa: F401 + proactor_udp, sctp_fix, sctp_stats, udp_buffer, diff --git a/packages/meshbay-node/tests/test_packaging_win.py b/packages/meshbay-node/tests/test_packaging_win.py index 6ee4891..0110b6c 100644 --- a/packages/meshbay-node/tests/test_packaging_win.py +++ b/packages/meshbay-node/tests/test_packaging_win.py @@ -1497,6 +1497,27 @@ def test_msix_gives_the_cli_an_execution_alias_and_the_windows_it_needs(): assert 'xmlns:uap5="http://schemas.microsoft.com/appx/manifest/uap/windows10/5"' in xml +def test_msix_gives_the_command_line_client_an_alias_of_its_own_application(): + """An Application may declare one windows.appExecutionAlias extension. + meshbay-node.exe has the app's; a second one beside it for meshbay-cli.exe + made makeappx refuse the whole package ("must only be declared once"), at + the very end of a Store build. The client is an Application of its own, + kept out of the Start menu, with its alias and its firewall rule.""" + xml = MSIX_MANIFEST_XML.read_text(encoding="utf-8") + ext = MSIX_EXTENSIONS_XML.read_text(encoding="utf-8") + assert ext.count('Category="windows.appExecutionAlias"') == 1 + apps = re.findall(r"<Application\b.*?</Application>", xml, re.S) + for app in apps: + assert app.count('Category="windows.appExecutionAlias"') <= 1, app[:80] + [cli] = [a for a in apps if 'Executable="app\\resources\\node-runtime\\meshbay-cli.exe"' + in a.split(">", 1)[0]] + assert 'AppListEntry="none"' in cli + assert 'Alias="meshbay-cli.exe"' in cli + package_level = xml.split("</Applications>", 1)[1] + rule = package_level.split('Executable="app\\resources\\node-runtime\\meshbay-cli.exe"', 1) + assert len(rule) == 2 and 'IPProtocol="UDP"' in rule[1].split("</desktop2:", 1)[0] + + def test_the_node_runtime_has_a_windowless_daemon_beside_the_cli(): spec = (WIN / "meshbay-node.spec").read_text(encoding="utf-8") windowless = spec.split('name="meshbay-nodew"', 1) diff --git a/packages/meshbay-node/tests/test_proactor_udp.py b/packages/meshbay-node/tests/test_proactor_udp.py new file mode 100644 index 0000000..0b22b4e --- /dev/null +++ b/packages/meshbay-node/tests/test_proactor_udp.py @@ -0,0 +1,150 @@ +""" +On Windows the node's datagrams go straight through the socket, not one +overlapped operation per datagram and per turn of the loop (`proactor_udp.py`). + +Run on a real ProactorEventLoop, which only Windows has. +""" + +import asyncio +import socket +import sys +import time + +import pytest +from meshbay_node.transport.webrtc import proactor_udp + +pytestmark = pytest.mark.skipif(sys.platform != "win32", reason="the proactor is Windows'") + +N = 2000 + + +class _Collect(asyncio.DatagramProtocol): + def __init__(self, raise_on: int = -1): + self.got: list[int] = [] + self.raise_on = raise_on + self.all = asyncio.get_running_loop().create_future() + + def datagram_received(self, data, addr): + self.got.append(int.from_bytes(data[:4], "big")) + if len(self.got) == N and not self.all.done(): + self.all.set_result(None) + if len(self.got) == self.raise_on: + raise RuntimeError("a protocol's own bug") + + +class _Refusing(socket.socket): + """A socket that refuses a send every so often, as a full buffer does, and + can answer one read with the reset an ICMP port unreachable leaves. The + overlapped path works by handle and never calls these.""" + + def __init__(self, every: int = 0, reset_once: bool = False): + super().__init__(socket.AF_INET, socket.SOCK_DGRAM) + self.every, self.calls, self.reset_once = every, 0, reset_once + + def sendto(self, data, addr): + self.calls += 1 + if self.every and self.calls % self.every == 0: + raise BlockingIOError + return super().sendto(data, addr) + + def recvfrom(self, n): + if self.reset_once: + self.reset_once = False + raise ConnectionResetError(10054, "port unreachable") + return super().recvfrom(n) + + +def _busy_turns(ms: float): + """A loop with other work to do: every turn takes `ms` longer.""" + async def busy(): + while True: + end = time.perf_counter() + ms / 1000 + while time.perf_counter() < end: + pass + await asyncio.sleep(0) + return asyncio.ensure_future(busy()) + + +async def _exchange(*, refuse_every=0, reset_once=False, raise_on=-1, busy_ms=0.0): + """N datagrams from one proactor transport to another on loopback: what + arrived in what order, and how many overlapped sends and reads it took.""" + loop = asyncio.get_running_loop() + rsock = _Refusing(reset_once=reset_once) + rsock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 8 << 20) + rsock.bind(("127.0.0.1", 0)) + rtr, rx = await loop.create_datagram_endpoint(lambda: _Collect(raise_on), sock=rsock) + tsock = _Refusing(refuse_every) + tsock.bind(("127.0.0.1", 0)) + tx, _ = await loop.create_datagram_endpoint(asyncio.DatagramProtocol, sock=tsock) + counts = {"sendto": 0, "recvfrom": 0} + originals = {name: getattr(loop._proactor, name) for name in counts} + + def counting(name): + def call(*a, **kw): + counts[name] += 1 + return originals[name](*a, **kw) + return call + for name in counts: + setattr(loop._proactor, name, counting(name)) + errors = [] + loop.set_exception_handler(lambda _loop, ctx: errors.append(ctx)) + busy = _busy_turns(busy_ms) if busy_ms else None + try: + dest = rsock.getsockname() + for i in range(N): + tx.sendto(i.to_bytes(4, "big") * 50, dest) + if i % 50 == 49: + await asyncio.sleep(0) + await asyncio.wait_for(rx.all, 20) + return rx.got, counts, errors + finally: + if busy: + busy.cancel() + for name, fn in originals.items(): + setattr(loop._proactor, name, fn) + tx.close() + rtr.close() + + +def _run(**kw): + return asyncio.run(_exchange(**kw), loop_factory=asyncio.ProactorEventLoop) + + +def test_the_transport_is_patched(): + cls = asyncio.proactor_events._ProactorDatagramTransport + assert cls.sendto is proactor_udp.sendto + assert cls._loop_reading is proactor_udp._loop_reading + + +def test_datagrams_go_straight_out_in_order(): + got, counts, _ = _run() + assert got == list(range(N)) + assert counts["sendto"] == 0 + + +def test_what_the_socket_refuses_is_queued_and_order_is_kept(): + """A refusal sends that datagram the overlapped way, and the ones after it + wait behind it rather than overtake it.""" + got, counts, _ = _run(refuse_every=97) + assert got == list(range(N)) + assert counts["sendto"] >= N // 97 + + +def test_a_busy_loop_reads_everything_waiting_in_order(): + """Unpatched, this loop read one datagram a turn: about 2,000 a second + whatever was waiting, and the kernel dropped the rest.""" + got, counts, _ = _run(busy_ms=2) + assert got == list(range(N)) + assert counts["recvfrom"] < N // 10 + + +def test_a_port_unreachable_on_a_direct_read_is_skipped(): + got, _, errors = _run(reset_once=True, busy_ms=2) + assert got == list(range(N)) + assert not errors + + +def test_a_protocol_that_raises_does_not_end_the_reading(): + got, _, errors = _run(raise_on=10) + assert got == list(range(N)) + assert [str(e["exception"]) for e in errors] == ["a protocol's own bug"] diff --git a/packaging/win/README.md b/packaging/win/README.md index 89185fe..959e651 100644 --- a/packaging/win/README.md +++ b/packaging/win/README.md @@ -37,7 +37,7 @@ way — not LocalSystem/NetworkService, which would have none of this. `build/installer.nsh` also adds `…\resources\node-runtime` to the **per-user** `Path` (`HKCU\Environment`) so `meshbay-node` and `meshbay-cli` work in a terminal, and takes it back out on uninstall. The Store package declares an -execution alias for each instead (`build/appx-extensions.xml`). New shells only — a `WM_SETTINGCHANGE` broadcast nudges +execution alias for each instead (the node's in `build/appx-extensions.xml`; the client's on an application of its own in `build/appx-manifest.xml`, since an application declares one alias). New shells only — a `WM_SETTINGCHANGE` broadcast nudges open ones. It uses stock `WordFunc.nsh` (the `EnVar` plugin is not in electron-builder's NSIS bundle). |