Link test: download series (bulk sampled, warm bursts, paced) and burst-vs-idle; car results
Orin -> tablet over NCM: paced streams up to 4 MB/s are clean; unpaced bursts and bulk hit multi-second holds that inflate the RTO to 10-16 s. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01DDwfd6Ki8UfG4hK6W5YpJz
This commit is contained in:
@@ -145,6 +145,219 @@ class LinkTest extends ChangeNotifier {
|
||||
return {'bytes': bytes, 'secs': (t * 1000).round() / 1000, 'mb_s': (bytes / t / 1e4).round() / 100};
|
||||
}
|
||||
|
||||
/// Orin → tablet only: sustained bulk sampled every 250 ms, TTS-like bursts on a
|
||||
/// warm connection, and camera-like paced streams.
|
||||
Future<void> runDownload(String host, String label, {int repeats = 5, int bulkSecs = 10, int pacedSecs = 10}) async {
|
||||
if (running) return;
|
||||
running = true;
|
||||
lines.clear();
|
||||
final id = '$label-${DateTime.now().toUtc().toIso8601String().substring(0, 19)}';
|
||||
_line('download series $id → $host');
|
||||
try {
|
||||
await usb.bindFor(Uri(scheme: 'ws', host: host));
|
||||
await _report(host, {'id': id, 'kind': 'start', 'label': label, 'series': 'download'});
|
||||
for (var i = 0; i < repeats; i++) {
|
||||
_status('bulk ${i + 1}/$repeats, $bulkSecs s');
|
||||
final r = await tcpBulkSampled(host, id, bulkSecs);
|
||||
_line('bulk ${i + 1}: ${r['mb_s']} MB/s, 250 ms windows min ${r['win_min_mb_s']} p10 ${r['win_p10_mb_s']} max ${r['win_max_mb_s']}, longest stall ${r['max_gap_ms']} ms');
|
||||
await _report(host, {'id': id, 'kind': 'dl_bulk', 'n': i, ...r});
|
||||
}
|
||||
_status('bursts');
|
||||
final b = await tcpBursts(host, id, sizes: const [65536, 262144, 1048576, 4194304], repeats: repeats);
|
||||
for (final e in (b['sizes'] as List)) {
|
||||
_line('burst ${e['bytes'] ~/ 1024} KB: p50 ${e['p50_ms']} ms, max ${e['max_ms']} ms (${e['mb_s_p50']} MB/s at p50)');
|
||||
}
|
||||
await _report(host, {'id': id, 'kind': 'dl_bursts', ...b});
|
||||
for (final rate in const [500000, 1000000, 2000000, 4000000]) {
|
||||
_status('paced ${rate / 1e6} MB/s, $pacedSecs s');
|
||||
final r = await tcpPaced(host, id, rate, pacedSecs);
|
||||
_line('paced ${rate / 1e6} MB/s: got ${r['mb_s']} MB/s, ${r['frames']}/${r['expected']} frames, queueing p50 ${r['q_p50_ms']} p99 ${r['q_p99_ms']} max ${r['q_max_ms']} ms, >100 ms ${r['late_100']}, >500 ms ${r['late_500']}');
|
||||
await _report(host, {'id': id, 'kind': 'dl_paced', ...r});
|
||||
}
|
||||
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();
|
||||
}
|
||||
}
|
||||
|
||||
/// Burst latency against the idle gap before each burst, with and without keepalive traffic.
|
||||
Future<void> runBurstIdle(String host, String label) async {
|
||||
if (running) return;
|
||||
running = true;
|
||||
lines.clear();
|
||||
final id = '$label-${DateTime.now().toUtc().toIso8601String().substring(0, 19)}';
|
||||
_line('burst/idle series $id → $host');
|
||||
try {
|
||||
await usb.bindFor(Uri(scheme: 'ws', host: host));
|
||||
await _report(host, {'id': id, 'kind': 'start', 'label': label, 'series': 'burst-idle'});
|
||||
for (final (idleMs, ka) in const [(0, 0), (50, 0), (200, 0), (1000, 0), (1000, 50), (0, 0)]) {
|
||||
_status('256 KB bursts, idle $idleMs ms, keepalive $ka/s');
|
||||
final r = await tcpBursts(host, id, sizes: const [262144], repeats: 8, idle: Duration(milliseconds: idleMs), keepalivePps: ka);
|
||||
final e = (r['sizes'] as List).first;
|
||||
_line('idle $idleMs ms ka $ka/s: p50 ${e['p50_ms']} max ${e['max_ms']} ms ${e['ms']}');
|
||||
await _report(host, {'id': id, 'kind': 'dl_bursts', ...r});
|
||||
}
|
||||
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();
|
||||
}
|
||||
}
|
||||
|
||||
static Future<Map<String, dynamic>> tcpBulkSampled(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();
|
||||
final windows = <int, int>{};
|
||||
var bytes = 0, maxGap = 0;
|
||||
int? first, last;
|
||||
await s.timeout(Duration(seconds: secs + 60)).forEach((b) {
|
||||
final now = clock.elapsedMicroseconds;
|
||||
first ??= now;
|
||||
if (last != null) maxGap = max(maxGap, now - last!);
|
||||
last = now;
|
||||
bytes += b.length;
|
||||
windows.update((now - first!) ~/ 250000, (v) => v + b.length, ifAbsent: () => b.length);
|
||||
});
|
||||
s.destroy();
|
||||
final t = ((last ?? 0) - (first ?? 0)) / 1e6;
|
||||
final n = windows.isEmpty ? 0 : windows.keys.reduce(max); // drop the last, partial window
|
||||
final rates = [for (var i = 0; i < n; i++) (windows[i] ?? 0) * 4 / 1e6]..sort();
|
||||
double r2(num v) => (v * 100).round() / 100;
|
||||
return {
|
||||
'bytes': bytes,
|
||||
'secs': r2(t),
|
||||
'mb_s': t > 0 ? r2(bytes / t / 1e6) : 0,
|
||||
'win_min_mb_s': rates.isEmpty ? null : r2(rates.first),
|
||||
'win_p10_mb_s': rates.isEmpty ? null : r2(rates[rates.length ~/ 10]),
|
||||
'win_p50_mb_s': rates.isEmpty ? null : r2(rates[rates.length ~/ 2]),
|
||||
'win_max_mb_s': rates.isEmpty ? null : r2(rates.last),
|
||||
'zero_windows': rates.where((r) => r == 0).length,
|
||||
'max_gap_ms': (maxGap / 100).round() / 10,
|
||||
};
|
||||
}
|
||||
|
||||
/// Each burst is requested on the same connection after 1 s idle, as utterances would be.
|
||||
static Future<Map<String, dynamic>> tcpBursts(String host, String id,
|
||||
{required List<int> sizes, required int repeats, Duration idle = const Duration(seconds: 1), int keepalivePps = 0}) async {
|
||||
// Optional light UDP traffic during the idle gaps, to tell link power saving from TCP behaviour.
|
||||
RawDatagramSocket? ka;
|
||||
Timer? kaTimer;
|
||||
if (keepalivePps > 0) {
|
||||
ka = await RawDatagramSocket.bind(InternetAddress.anyIPv4, 0);
|
||||
final addr = InternetAddress(host);
|
||||
kaTimer = Timer.periodic(Duration(microseconds: 1000000 ~/ keepalivePps), (_) => ka!.send(Uint8List(16), addr, _echo));
|
||||
}
|
||||
final s = await Socket.connect(host, _down, timeout: const Duration(seconds: 5));
|
||||
s.write('$id burst\n');
|
||||
var got = 0;
|
||||
Completer<void>? want;
|
||||
var target = 0;
|
||||
final sub = s.listen((b) {
|
||||
got += b.length;
|
||||
final w = want;
|
||||
if (w != null && !w.isCompleted && got >= target) w.complete();
|
||||
});
|
||||
final out = <Map<String, dynamic>>[];
|
||||
for (final size in sizes) {
|
||||
final ms = <double>[];
|
||||
for (var i = 0; i < repeats; i++) {
|
||||
await Future<void>.delayed(idle);
|
||||
got = 0;
|
||||
target = size;
|
||||
final w = want = Completer<void>();
|
||||
final clock = Stopwatch()..start();
|
||||
s.write('$size\n');
|
||||
await w.future.timeout(const Duration(seconds: 60));
|
||||
want = null;
|
||||
ms.add(clock.elapsedMicroseconds / 1000);
|
||||
}
|
||||
ms.sort();
|
||||
final p50 = ms[ms.length ~/ 2];
|
||||
out.add({
|
||||
'bytes': size,
|
||||
'ms': [for (final m in ms) (m * 10).round() / 10],
|
||||
'p50_ms': (p50 * 10).round() / 10,
|
||||
'max_ms': (ms.last * 10).round() / 10,
|
||||
'mb_s_p50': (size / p50 / 10).round() / 100,
|
||||
});
|
||||
}
|
||||
await sub.cancel();
|
||||
s.destroy();
|
||||
kaTimer?.cancel();
|
||||
ka?.close();
|
||||
return {'idle_ms': idle.inMilliseconds, 'keepalive_pps': keepalivePps, 'sizes': out};
|
||||
}
|
||||
|
||||
/// Frames carry the Orin's send time. The clocks are not compared absolutely: queueing
|
||||
/// delay is each frame's one-way delay minus the run's smallest.
|
||||
static Future<Map<String, dynamic>> tcpPaced(String host, String id, int rate, int secs) async {
|
||||
final s = await Socket.connect(host, _down, timeout: const Duration(seconds: 5));
|
||||
s.write('$id paced $rate $secs\n');
|
||||
final header = BytesBuilder();
|
||||
var skip = 0, bytes = 0;
|
||||
final owd = <double>[];
|
||||
double? firstRx, lastRx;
|
||||
await s.timeout(Duration(seconds: secs + 60)).forEach((chunk) {
|
||||
bytes += chunk.length;
|
||||
var i = 0;
|
||||
while (i < chunk.length) {
|
||||
if (skip > 0) {
|
||||
final n = min(skip, chunk.length - i);
|
||||
skip -= n;
|
||||
i += n;
|
||||
continue;
|
||||
}
|
||||
final n = min(12 - header.length, chunk.length - i);
|
||||
header.add(Uint8List.sublistView(chunk, i, i + n));
|
||||
i += n;
|
||||
if (header.length == 12) {
|
||||
final h = ByteData.sublistView(header.takeBytes());
|
||||
final size = h.getUint32(0);
|
||||
final sentAt = h.getFloat64(4);
|
||||
skip = size - 12;
|
||||
// Arrival of the header stands for the frame; with skip pending, the rest follows.
|
||||
final now = DateTime.now().microsecondsSinceEpoch / 1e6;
|
||||
firstRx ??= now;
|
||||
lastRx = now;
|
||||
owd.add(now - sentAt);
|
||||
}
|
||||
}
|
||||
});
|
||||
s.destroy();
|
||||
final base = owd.isEmpty ? 0.0 : owd.reduce(min);
|
||||
final q = [for (final d in owd) (d - base) * 1000]..sort();
|
||||
double r1(num v) => (v * 10).round() / 10;
|
||||
final t = (lastRx ?? 0) - (firstRx ?? 0);
|
||||
return {
|
||||
'rate': rate,
|
||||
'secs': secs,
|
||||
'expected': secs * 30,
|
||||
'frames': owd.length,
|
||||
'mb_s': t > 0 ? (bytes / t / 1e4).round() / 100 : 0,
|
||||
'q_p50_ms': q.isEmpty ? null : r1(q[q.length ~/ 2]),
|
||||
'q_p99_ms': q.isEmpty ? null : r1(q[min(q.length - 1, (q.length * 0.99).floor())]),
|
||||
'q_max_ms': q.isEmpty ? null : r1(q.last),
|
||||
'late_100': q.where((v) => v > 100).length,
|
||||
'late_500': q.where((v) => v > 500).length,
|
||||
};
|
||||
}
|
||||
|
||||
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();
|
||||
|
||||
+9
-3
@@ -18,14 +18,20 @@ Future<void> main() async {
|
||||
await usb.bindFor(Uri.parse(defaultUrl));
|
||||
final boundary = Boundary(defaultUrl)..start();
|
||||
final linkTest = LinkTest(usb);
|
||||
// Development hook: `am start ... --es linktest <label>[@<host>]` runs the test,
|
||||
// on the gateway's host unless one is given.
|
||||
// Development hook: `am start ... --es linktest [dl:|bi:]<label>[@<host>]` runs the test
|
||||
// (`dl:` the download series, `bi:` burst latency against idle gaps), on the gateway's host unless one is given.
|
||||
Future<void> runLinkTest(String request) async {
|
||||
final mode = request.startsWith('dl:') || request.startsWith('bi:') ? request.substring(0, 2) : '';
|
||||
if (mode.isNotEmpty) request = request.substring(3);
|
||||
final at = request.indexOf('@');
|
||||
final label = at < 0 ? request : request.substring(0, at);
|
||||
final host = at < 0 ? Uri.parse(boundary.url).host : request.substring(at + 1);
|
||||
navigatorKey.currentState?.push(MaterialPageRoute<void>(builder: (_) => linkTestPage(boundary, linkTest, usb)));
|
||||
await linkTest.run(host, label);
|
||||
await switch (mode) {
|
||||
'dl' => linkTest.runDownload(host, label),
|
||||
'bi' => linkTest.runBurstIdle(host, label),
|
||||
_ => linkTest.run(host, label),
|
||||
};
|
||||
await usb.bindFor(Uri.parse(boundary.url));
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user