Link test: UDP echo sweep, soak and TCP each way, driven by the app

The tablet's app sockets are the only ones bound to the USB link, so the
measurement has to run in the app. tool/linktest_server.py serves the
Orin end and logs every result with its own TCP_INFO. Startable without
touching the screen: am start ... --es linktest <label>.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01DDwfd6Ki8UfG4hK6W5YpJz

android/app/src/main/kotlin/dev/yukart/display/MainActivity.kt

@@ -1,6 +1,7 @@
 package dev.yukart.display
 
 import android.content.Context
+import android.content.Intent
 import android.net.ConnectivityManager
 import android.net.Network
 import android.net.NetworkCapabilities
@@ -20,12 +21,16 @@ import io.flutter.plugin.common.MethodChannel
  *   default network while WiFi or cellular is up, and ordinary sockets go elsewhere.
  *   When asked, the whole process is bound to the Ethernet (NCM) network while it
  *   exists, and unbound when it goes.
+ *
+ * Also a development hook: `am start -n dev.yukart.display/.MainActivity --es linktest <label>`
+ * runs the link test (lib/linktest.dart) without touching the screen.
  */
 class MainActivity : FlutterActivity() {
     private lateinit var channel: MethodChannel
     private lateinit var cm: ConnectivityManager
     private var ethernet: Network? = null
     private var wantBinding = false
+    private var pendingLinkTest: String? = null
 
     private val callback = object : ConnectivityManager.NetworkCallback() {
         override fun onAvailable(network: Network) {
@@ -44,6 +49,12 @@ class MainActivity : FlutterActivity() {
     override fun onCreate(savedInstanceState: Bundle?) {
         super.onCreate(savedInstanceState)
         window.addFlags(WindowManager.LayoutParams.FLAG_KEEP_SCREEN_ON)
+        pendingLinkTest = intent?.getStringExtra("linktest")
+    }
+
+    override fun onNewIntent(intent: Intent) {
+        super.onNewIntent(intent)
+        intent.getStringExtra("linktest")?.let { channel.invokeMethod("linktest", it) }
     }
 
     override fun configureFlutterEngine(flutterEngine: FlutterEngine) {
@@ -57,6 +68,10 @@ class MainActivity : FlutterActivity() {
                     apply()
                     result.success(ethernet != null)
                 }
+                "takeLinkTest" -> {
+                    result.success(pendingLinkTest)
+                    pendingLinkTest = null
+                }
                 else -> result.notImplemented()
             }
         }

lib/linktest.dart

@@ -0,0 +1,231 @@
+// Link test: measures the tablet ⇄ Orin link from the app, whose sockets are the
+// only ones on the tablet bound to the USB network (`adb shell` traffic goes over
+// WiFi). Pairs with tool/linktest_server.py on the Orin, which also logs every
+// result, so a run is readable over SSH whatever happens to the screen.
+
+import 'dart:async';
+import 'dart:convert';
+import 'dart:io';
+import 'dart:math';
+import 'dart:typed_data';
+
+import 'package:flutter/material.dart';
+
+import 'net.dart';
+
+const _down = 7791, _up = 7792, _echo = 7793, _results = 7794;
+
+class LinkTest extends ChangeNotifier {
+  LinkTest(this.usb);
+  final UsbLink usb;
+
+  bool running = false;
+  String status = 'idle';
+  final List<String> lines = [];
+
+  /// UDP echo at each (size, rate) for [udpSecs], an app-like soak, then TCP each way.
+  Future<void> run(String host, String label, {int udpSecs = 3, int soakSecs = 30, int tcpSecs = 5}) async {
+    if (running) return;
+    running = true;
+    lines.clear();
+    final id = '$label-${DateTime.now().toUtc().toIso8601String().substring(0, 19)}';
+    _line('run $id → $host');
+    try {
+      await usb.bindFor(Uri(scheme: 'ws', host: host));
+      await _report(host, {'id': id, 'kind': 'start', 'label': label});
+      for (final size in [64, 1400]) {
+        for (final rate in [10, 50, 100, 200, 500]) {
+          _status('udp $size B × $rate/s');
+          final r = await udpEcho(host, size: size, rate: rate, secs: udpSecs);
+          _line('udp ${size}B $rate/s: ${r['loss_pct']}% loss, rtt p50 ${r['rtt_p50_ms']} p99 ${r['rtt_p99_ms']} max ${r['rtt_max_ms']} ms');
+          await _report(host, {'id': id, 'kind': 'udp', ...r});
+        }
+      }
+      _status('soak: 64 B × 10/s for $soakSecs s');
+      final soak = await udpEcho(host, size: 64, rate: 10, secs: soakSecs);
+      _line('soak ${soakSecs}s: ${soak['loss_pct']}% loss, max gap ${soak['max_gap_ms']} ms, rtt max ${soak['rtt_max_ms']} ms');
+      await _report(host, {'id': id, 'kind': 'soak', ...soak});
+      _status('tcp down $tcpSecs s');
+      final d = await tcpDown(host, id, tcpSecs);
+      _line('tcp down: ${d['mb_s']} MB/s');
+      await _report(host, {'id': id, 'kind': 'tcp_down', ...d});
+      _status('tcp up $tcpSecs s');
+      final u = await tcpUp(host, id, tcpSecs);
+      _line('tcp up: ${u['mb_s']} MB/s');
+      await _report(host, {'id': id, 'kind': 'tcp_up', ...u});
+      await _report(host, {'id': id, 'kind': 'end'});
+      _status('done');
+    } catch (e) {
+      _line('error: $e');
+      _status('failed');
+      try {
+        await _report(host, {'id': id, 'kind': 'error', 'error': '$e'});
+      } catch (_) {}
+    } finally {
+      running = false;
+      notifyListeners();
+    }
+  }
+
+  /// Sends [rate] packets/s of [size] bytes for [secs], each carrying its sequence
+  /// number and send time; loss, RTT and the longest silence come from the echoes.
+  static Future<Map<String, dynamic>> udpEcho(String host, {required int size, required int rate, required int secs}) async {
+    final addr = InternetAddress(host);
+    final sock = await RawDatagramSocket.bind(InternetAddress.anyIPv4, 0);
+    final clock = Stopwatch()..start();
+    final rtts = <int>[];
+    final seen = <int>{};
+    var dups = 0, sendFails = 0, sent = 0;
+    int? lastRx;
+    var maxGap = 0;
+    sock.listen((ev) {
+      if (ev != RawSocketEvent.read) return;
+      Datagram? d;
+      while ((d = sock.receive()) != null) {
+        if (d!.data.length < 12) continue;
+        final b = ByteData.sublistView(d.data);
+        final seq = b.getUint32(0);
+        final now = clock.elapsedMicroseconds;
+        if (!seen.add(seq)) {
+          dups++;
+          continue;
+        }
+        rtts.add(now - b.getInt64(4));
+        if (lastRx != null) maxGap = max(maxGap, now - lastRx!);
+        lastRx = now;
+      }
+    });
+    final total = rate * secs;
+    final buf = Uint8List(size);
+    final done = Completer<void>();
+    Timer.periodic(const Duration(milliseconds: 2), (t) {
+      final due = min(total, clock.elapsedMicroseconds * rate ~/ 1000000);
+      while (sent < due) {
+        final b = ByteData.sublistView(buf);
+        b.setUint32(0, sent);
+        b.setInt64(4, clock.elapsedMicroseconds);
+        if (sock.send(buf, addr, _echo) == 0) sendFails++;
+        sent++;
+      }
+      if (sent >= total) {
+        t.cancel();
+        done.complete();
+      }
+    });
+    await done.future;
+    await Future<void>.delayed(const Duration(seconds: 1)); // stragglers
+    sock.close();
+    rtts.sort();
+    double ms(int us) => (us / 100).round() / 10;
+    int pct(double p) => rtts.isEmpty ? 0 : rtts[min(rtts.length - 1, (rtts.length * p).floor())];
+    return {
+      'size': size,
+      'rate': rate,
+      'secs': secs,
+      'sent': sent,
+      'received': seen.length,
+      'loss_pct': sent == 0 ? 0 : ((sent - seen.length) * 1000 / sent).round() / 10,
+      'dups': dups,
+      'send_fails': sendFails,
+      'rtt_p50_ms': ms(pct(0.5)),
+      'rtt_p99_ms': ms(pct(0.99)),
+      'rtt_max_ms': rtts.isEmpty ? null : ms(rtts.last),
+      'max_gap_ms': ms(maxGap),
+    };
+  }
+
+  static Future<Map<String, dynamic>> tcpDown(String host, String id, int secs) async {
+    final s = await Socket.connect(host, _down, timeout: const Duration(seconds: 5));
+    s.write('$id $secs\n');
+    final clock = Stopwatch()..start();
+    var bytes = 0;
+    await s.timeout(Duration(seconds: secs + 30)).forEach((b) => bytes += b.length);
+    s.destroy();
+    final t = clock.elapsedMicroseconds / 1e6;
+    return {'bytes': bytes, 'secs': (t * 1000).round() / 1000, 'mb_s': (bytes / t / 1e4).round() / 100};
+  }
+
+  static Future<Map<String, dynamic>> tcpUp(String host, String id, int secs) async {
+    final s = await Socket.connect(host, _up, timeout: const Duration(seconds: 5));
+    final reply = utf8.decoder.bind(s).join();
+    s.write('$id\n');
+    final chunk = Uint8List(65536);
+    final clock = Stopwatch()..start();
+    var bytes = 0;
+    while (clock.elapsed < Duration(seconds: secs)) {
+      s.add(chunk);
+      await s.flush();
+      bytes += chunk.length;
+    }
+    await s.close();
+    final got = int.tryParse((await reply.timeout(const Duration(seconds: 30))).trim()) ?? -1;
+    final t = clock.elapsedMicroseconds / 1e6;
+    s.destroy();
+    return {'bytes': bytes, 'orin_bytes': got, 'secs': (t * 1000).round() / 1000, 'mb_s': (got / t / 1e4).round() / 100};
+  }
+
+  static Future<void> _report(String host, Map<String, dynamic> rec) async {
+    final s = await Socket.connect(host, _results, timeout: const Duration(seconds: 5));
+    s.write('${jsonEncode({...rec, 'tablet_ts': DateTime.now().millisecondsSinceEpoch / 1000})}\n');
+    await s.close();
+    s.destroy();
+  }
+
+  void _status(String s) {
+    status = s;
+    notifyListeners();
+  }
+
+  void _line(String s) {
+    lines.add(s);
+    notifyListeners();
+  }
+}
+
+class LinkTestPage extends StatefulWidget {
+  const LinkTestPage({super.key, required this.test, required this.host, this.restoreBinding});
+  final LinkTest test;
+  final String host;
+
+  /// Called after a run, to return the binding to what the gateway link needs.
+  final Future<void> Function()? restoreBinding;
+
+  @override
+  State<LinkTestPage> createState() => _LinkTestPageState();
+}
+
+class _LinkTestPageState extends State<LinkTestPage> {
+  late final _host = TextEditingController(text: widget.host);
+  final _label = TextEditingController(text: 'setup');
+
+  Future<void> _run() async {
+    await widget.test.run(_host.text.trim(), _label.text.trim().replaceAll(' ', '_'));
+    await widget.restoreBinding?.call();
+  }
+
+  @override
+  Widget build(BuildContext context) {
+    return Scaffold(
+      appBar: AppBar(title: const Text('Link test')),
+      body: ListenableBuilder(
+        listenable: widget.test,
+        builder: (context, _) {
+          final t = widget.test;
+          return ListView(padding: const EdgeInsets.all(16), children: [
+            Row(children: [
+              Expanded(child: TextField(controller: _host, decoration: const InputDecoration(labelText: 'Orin address'))),
+              const SizedBox(width: 16),
+              Expanded(child: TextField(controller: _label, decoration: const InputDecoration(labelText: 'Setup label'))),
+              const SizedBox(width: 16),
+              FilledButton(onPressed: t.running ? null : _run, child: Text(t.running ? 'Running…' : 'Run')),
+            ]),
+            const SizedBox(height: 12),
+            Text('Status: ${t.status}', style: Theme.of(context).textTheme.titleMedium),
+            const SizedBox(height: 8),
+            for (final l in t.lines) Text(l, style: const TextStyle(fontFamily: 'monospace')),
+          ]);
+        },
+      ),
+    );
+  }
+}

lib/main.dart

@@ -6,6 +6,7 @@ import 'dart:math';
 import 'package:flutter/material.dart';
 
 import 'boundary.dart';
+import 'linktest.dart';
 import 'net.dart';
 
 /// Override for development, e.g. `--dart-define=YUKART_GW=ws://halogen.local:7700/`.
@@ -16,32 +17,56 @@ Future<void> main() async {
   final usb = UsbLink();
   await usb.bindFor(Uri.parse(defaultUrl));
   final boundary = Boundary(defaultUrl)..start();
-  runApp(YukartApp(boundary: boundary, usb: usb));
+  final linkTest = LinkTest(usb);
+  // Development hook: `am start ... --es linktest <label>` runs the test on the gateway's host.
+  Future<void> runLinkTest(String label) async {
+    navigatorKey.currentState?.push(MaterialPageRoute<void>(builder: (_) => linkTestPage(boundary, linkTest, usb)));
+    await linkTest.run(Uri.parse(boundary.url).host, label);
+    await usb.bindFor(Uri.parse(boundary.url));
+  }
+
+  usb.onLinkTest = runLinkTest;
+  runApp(YukartApp(boundary: boundary, usb: usb, linkTest: linkTest));
+  final pending = await usb.takeLinkTest();
+  if (pending != null) {
+    WidgetsBinding.instance.addPostFrameCallback((_) => runLinkTest(pending));
+  }
 }
 
+final navigatorKey = GlobalKey<NavigatorState>();
+
+Widget linkTestPage(Boundary b, LinkTest t, UsbLink usb) => LinkTestPage(
+      test: t,
+      host: Uri.parse(b.url).host,
+      restoreBinding: () => usb.bindFor(Uri.parse(b.url)),
+    );
+
 class YukartApp extends StatelessWidget {
-  const YukartApp({super.key, required this.boundary, required this.usb});
+  const YukartApp({super.key, required this.boundary, required this.usb, required this.linkTest});
   final Boundary boundary;
   final UsbLink usb;
+  final LinkTest linkTest;
 
   @override
   Widget build(BuildContext context) {
     return MaterialApp(
       title: 'Yukart',
+      navigatorKey: navigatorKey,
       debugShowCheckedModeBanner: false,
       theme: ThemeData(colorSchemeSeed: Colors.teal, brightness: Brightness.dark, useMaterial3: true),
       home: ListenableBuilder(
         listenable: Listenable.merge([boundary, usb]),
-        builder: (context, _) => Readout(boundary: boundary, usb: usb),
+        builder: (context, _) => Readout(boundary: boundary, usb: usb, linkTest: linkTest),
       ),
     );
   }
 }
 
 class Readout extends StatelessWidget {
-  const Readout({super.key, required this.boundary, required this.usb});
+  const Readout({super.key, required this.boundary, required this.usb, required this.linkTest});
   final Boundary boundary;
   final UsbLink usb;
+  final LinkTest linkTest;
 
   @override
   Widget build(BuildContext context) {
@@ -53,7 +78,7 @@ class Readout extends StatelessWidget {
           children: [
             // Mock data must look like mock data: on the surface, not in a menu.
             if (b.mock) const MockBanner(),
-            LinkBar(boundary: b, usb: usb),
+            LinkBar(boundary: b, usb: usb, linkTest: linkTest),
             Expanded(
               child: LayoutBuilder(
                 builder: (context, c) {
@@ -100,9 +125,10 @@ class MockBanner extends StatelessWidget {
 }
 
 class LinkBar extends StatelessWidget {
-  const LinkBar({super.key, required this.boundary, required this.usb});
+  const LinkBar({super.key, required this.boundary, required this.usb, required this.linkTest});
   final Boundary boundary;
   final UsbLink usb;
+  final LinkTest linkTest;
 
   @override
   Widget build(BuildContext context) {
@@ -127,6 +153,13 @@ class LinkBar extends StatelessWidget {
             Text('last msg $age'),
             Text('Orin−tablet clock $offset'),
             if (usb.supported) Text('USB link ${usb.up ? 'up' : 'absent'}${usb.bound ? ', bound' : ''}'),
+            TextButton.icon(
+              onPressed: () => Navigator.of(context).push(
+                MaterialPageRoute<void>(builder: (_) => linkTestPage(b, linkTest, usb)),
+              ),
+              icon: const Icon(Icons.speed),
+              label: const Text('Link test'),
+            ),
           ]),
         ),
       ),

lib/net.dart

@@ -13,6 +13,9 @@ class UsbLink extends ChangeNotifier {
   bool up = false;
   bool bound = false;
 
+  /// Development hook: a link test requested by launch intent (see MainActivity.kt).
+  void Function(String label)? onLinkTest;
+
   UsbLink() {
     if (!supported) return;
     _ch.setMethodCallHandler((call) async {
@@ -21,10 +24,15 @@ class UsbLink extends ChangeNotifier {
         up = a['up'] == true;
         bound = a['bound'] == true;
         notifyListeners();
+      } else if (call.method == 'linktest') {
+        onLinkTest?.call(call.arguments as String);
       }
     });
   }
 
+  /// A link test requested by the intent that launched the app, once.
+  Future<String?> takeLinkTest() async => supported ? _ch.invokeMethod<String>('takeLinkTest') : null;
+
   /// Bind only when the target is on the USB link; a workstation over WiFi must stay reachable.
   Future<void> bindFor(Uri target) async {
     if (!supported) return;

tool/linktest_server.py

@@ -0,0 +1,123 @@
+#!/usr/bin/env python3
+"""Orin side of the app's link test (lib/linktest.dart). Development tool, not a service.
+
+The tablet's own sockets are the only ones bound to the USB link (`adb shell`
+traffic goes over WiFi), so the app drives the test and this end only serves:
+
+  7791/tcp  down: client sends "<id> <secs>\\n", gets zeros for <secs>
+  7792/tcp  up:   client sends "<id>\\n" then data until EOF; gets "<bytes>\\n" back
+  7793/udp  echo
+  7794/tcp  results: one JSON line per connection, appended to the results file
+
+Every TCP test also logs this end's TCP_INFO (retransmits, RTT), which the app
+cannot read. Run: python3 linktest_server.py [results.jsonl]
+"""
+
+import json
+import os
+import socket
+import struct
+import sys
+import threading
+import time
+
+RESULTS = os.path.expanduser(sys.argv[1] if len(sys.argv) > 1 else "~/linktest/results.jsonl")
+CHUNK = b"\0" * 65536
+lock = threading.Lock()
+
+
+def log(rec):
+    rec = {"orin_ts": round(time.time(), 3), **rec}
+    with lock, open(RESULTS, "a") as f:
+        f.write(json.dumps(rec) + "\n")
+    print(json.dumps(rec), flush=True)
+
+
+def tcp_info(sock):
+    # struct tcp_info: 8 bytes of u8 fields, then u32s; rtt is the 16th, total_retrans the 24th.
+    raw = sock.getsockopt(socket.IPPROTO_TCP, socket.TCP_INFO, 104)
+    u = struct.unpack_from("8B24I", raw)
+    return {"retrans": u[8 + 23], "rtt_ms": u[8 + 15] / 1000, "rto_ms": u[8] / 1000}
+
+
+def readline(conn):
+    buf = b""
+    while not buf.endswith(b"\n") and len(buf) < 256:
+        b = conn.recv(1)
+        if not b:
+            break
+        buf += b
+    return buf.decode().strip()
+
+
+def down(conn, peer):
+    test_id, secs = readline(conn).split()
+    end = time.monotonic() + float(secs)
+    sent = 0
+    try:
+        while time.monotonic() < end:
+            sent += conn.send(CHUNK)
+    except OSError as e:
+        log({"id": test_id, "kind": "tcp_down_orin", "error": str(e)})
+    log({"id": test_id, "kind": "tcp_down_orin", "peer": peer, "bytes": sent, **tcp_info(conn)})
+
+
+def up(conn, peer):
+    test_id = readline(conn)
+    got = 0
+    t0 = time.monotonic()
+    while True:
+        b = conn.recv(65536)
+        if not b:
+            break
+        got += len(b)
+    secs = time.monotonic() - t0
+    conn.sendall(f"{got}\n".encode())
+    log({"id": test_id, "kind": "tcp_up_orin", "peer": peer, "bytes": got, "secs": round(secs, 3),
+         **tcp_info(conn)})
+
+
+def results(conn, peer):
+    data = b""
+    while True:
+        b = conn.recv(65536)
+        if not b:
+            break
+        data += b
+    for line in data.decode().splitlines():
+        if line.strip():
+            log({"peer": peer, **json.loads(line)})
+
+
+def serve_tcp(port, handler):
+    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    s.bind(("0.0.0.0", port))
+    s.listen(4)
+    while True:
+        conn, addr = s.accept()
+
+        def run(conn=conn, peer=addr[0]):
+            with conn:
+                try:
+                    handler(conn, peer)
+                except Exception as e:  # a broken test must not stop the server
+                    print(f"{handler.__name__} {peer}: {e}", flush=True)
+
+        threading.Thread(target=run, daemon=True).start()
+
+
+def serve_echo(port):
+    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    s.bind(("0.0.0.0", port))
+    while True:
+        data, addr = s.recvfrom(2048)
+        s.sendto(data, addr)
+
+
+if __name__ == "__main__":
+    os.makedirs(os.path.dirname(RESULTS), exist_ok=True)
+    for port, h in ((7791, down), (7792, up), (7794, results)):
+        threading.Thread(target=serve_tcp, args=(port, h), daemon=True).start()
+    print(f"linktest server: tcp 7791 down, 7792 up, 7794 results; udp 7793 echo; -> {RESULTS}", flush=True)
+    serve_echo(7793)