From 598a2d99c95877c3e6875ea15b250c6f9a905bb8 Mon Sep 17 00:00:00 2001 From: Mikkeli Date: Wed, 26 Aug 2026 19:35:10 +0900 Subject: [PATCH 1/4] refactor(jni): take results from json_callback instead of piping stdout MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit iperf3 exposes a hook for exactly this use — iperf.h describes json_callback as "allow user apps to receive the JSON strings, instead of writing them to the output file". Setting it alongside --json-stream's two flags makes iperf3 hand each report ("start", "interval", "end", "error") straight to the bridge as a JSON string, and test->outfile is never touched. That removes the pipe, the reader thread, the fdopen/fgets loop, and AttachCurrentThread from the output path — roughly 90 lines. It also removes the two constraints that came with them and were both gotten wrong: that the write end must be closed before joining a reader blocked in fgets(), and that exactly one side may own the shared state struct. Neither can be violated now because neither exists. json_callback carries no user-data pointer, so the run state is reached through a global published before iperf_run_client() and cleared after it. Only one test runs at a time, which runIperfLive already enforced. The callback fires on iperf3's reporting thread, so it resolves a JNIEnv for the calling thread rather than trusting a stored one. Co-Authored-By: Claude Opus 5 --- android/app/src/main/cpp/iperf_jni.c | 220 ++++++++------------------- 1 file changed, 67 insertions(+), 153 deletions(-) diff --git a/android/app/src/main/cpp/iperf_jni.c b/android/app/src/main/cpp/iperf_jni.c index 5b4cfac..ec6dca4 100644 --- a/android/app/src/main/cpp/iperf_jni.c +++ b/android/app/src/main/cpp/iperf_jni.c @@ -5,16 +5,25 @@ * the Flutter app to run iperf3 tests natively inside the app process * (no Process.start needed, no SELinux restrictions). * + * OUTPUT NOTE: + * The test runs in --json-stream mode with a json_callback installed, so + * iperf3 hands each event ("start", "interval", "end", "error") straight to + * us as a JSON string and never touches test->outfile. That is what the hook + * exists for -- iperf.h calls it "allow user apps to receive the JSON + * strings, instead of writing them to the output file". + * + * An earlier version piped test->outfile to a reader thread instead. That + * required closing the pipe's write end before joining the reader (which + * blocks in fgets() until EOF) and single ownership of the shared state + * struct. Both were gotten wrong and deadlocked every test. Reading events + * from the callback removes the pipe, the reader thread, and that whole class + * of ordering bug. + * * THREADING NOTE: * runIperfLive() BLOCKS for the whole duration of the test. Kotlin must * call it from a worker thread, never from the Flutter platform thread. - * - * PIPE / SHUTDOWN NOTE: - * iperf's output goes to a pipe drained by a reader thread that blocks in - * fgets(). The reader only sees EOF once the write end is closed, so the - * write end MUST be closed before pthread_join() — joining first deadlocks. - * The state struct is owned solely by the JNI caller, which frees it after - * joining the reader; the reader must never free it. + * json_callback fires on whichever thread iperf3 reports from, so it resolves + * a JNIEnv for the calling thread rather than reusing a stored one. * * SIGNAL HANDLING NOTE: * iperf3's run() function (called from iperf_run_client()) uses @@ -61,21 +70,22 @@ static pthread_mutex_t g_test_mutex = PTHREAD_MUTEX_INITIALIZER; static volatile int g_stop = 0; static char g_tmp_template[1024] = "/data/local/tmp/iperf3.XXXXXX"; -/* Per-call state shared between the JNI caller and the reader thread. - * We store the callback object (global ref) and method IDs. +/* Per-call state, reachable from the json_callback. * Callbacks are invoked via CallVoidMethod, NOT as function pointers. */ struct iperf_run_state { jobject callback_ref; /* Global ref to the callback object */ jmethodID mid_output; /* Method ID for onOutput(String) */ jmethodID mid_error; /* Method ID for onError(String) */ jmethodID mid_complete; /* Method ID for onComplete() */ - FILE *outfile; /* Pipe write end; NULL once closed */ - int pipe_fd[2]; JavaVM *jvm; - pthread_t reader_thread; - int reader_started; /* Only join if pthread_create succeeded */ }; +/* iperf3's json_callback signature carries no user-data pointer, so the + * current run's state has to be reachable from a global. Only one test runs + * at a time (enforced in runIperfLive), and this is published before + * iperf_run_client() and cleared after it returns. */ +static struct iperf_run_state *g_current_state = NULL; + /* Helper: invoke the output callback via CallVoidMethod */ static void signal_output(JNIEnv *env, struct iperf_run_state *state, const char *msg) { @@ -108,50 +118,36 @@ static void signal_complete(JNIEnv *env, struct iperf_run_state *state) { } /* ------------------------------------------------------------------ */ -/* Output reader thread — reads from iperf's output pipe and forwards */ -/* each line to the Java callback via JNI. */ +/* json_callback — receives one JSON event string per report from */ +/* iperf3 and forwards it to Java verbatim. Parsing happens in Dart. */ /* */ -/* Exits when the write end of the pipe is closed (EOF). Does NOT own */ -/* `state`: the JNI caller frees it after joining this thread. */ +/* Runs on iperf3's reporting thread. In the client that is the thread */ +/* that called iperf_run_client() (already attached, since Kotlin's */ +/* executor thread is a Java thread), but GetEnv is checked rather */ +/* than assumed, with an attach as fallback. */ /* ------------------------------------------------------------------ */ -static void *reader_thread_func(void *arg) { - struct iperf_run_state *state = (struct iperf_run_state *) arg; - JNIEnv *env; - int ret; +static void json_event_callback(struct iperf_test *test, char *json) { + struct iperf_run_state *state = g_current_state; + if (!state || !json) return; - /* Attach this native thread to the Java VM */ - ret = (*state->jvm)->AttachCurrentThread(state->jvm, &env, NULL); - if (ret != JNI_OK) { - LOGE("AttachCurrentThread failed: %d", ret); - /* Still drain the pipe so writers never block on a full pipe. */ - char drain[1024]; - while (read(state->pipe_fd[0], drain, sizeof(drain)) > 0) { } - close(state->pipe_fd[0]); - return NULL; - } + JNIEnv *env = NULL; + int attached = 0; - char buf[1024]; - FILE *fp = fdopen(state->pipe_fd[0], "r"); - if (fp) { - while (fgets(buf, sizeof(buf), fp) != NULL) { - /* Skip empty lines */ - if (buf[0] == '\n' || buf[0] == '\0') - continue; - jstring line = (*env)->NewStringUTF(env, buf); - if (line != NULL) { - (*env)->CallVoidMethod(env, state->callback_ref, - state->mid_output, line); - (*env)->DeleteLocalRef(env, line); - } + if ((*state->jvm)->GetEnv(state->jvm, (void **) &env, JNI_VERSION_1_6) + != JNI_OK) { + if ((*state->jvm)->AttachCurrentThread(state->jvm, &env, NULL) + != JNI_OK) { + LOGE("json_event_callback: could not reach a JNIEnv"); + return; } - fclose(fp); /* also closes pipe_fd[0] */ - } else { - close(state->pipe_fd[0]); + attached = 1; } - /* Detach this native thread */ - (*state->jvm)->DetachCurrentThread(state->jvm); - return NULL; + signal_output(env, state, json); + + if (attached) { + (*state->jvm)->DetachCurrentThread(state->jvm); + } } /* ------------------------------------------------------------------ */ @@ -198,30 +194,12 @@ static void free_argv(char **argv, int argc) { free(argv); } -/* ------------------------------------------------------------------ */ -/* Close the pipe write end. Idempotent — both the normal and the */ -/* longjmp path call it, and the second call must be a no-op. */ -/* */ -/* Closing this is what makes the reader thread see EOF and exit, so */ -/* it must happen before any pthread_join(). */ -/* ------------------------------------------------------------------ */ -static void close_outfile(struct iperf_run_state *state) { - if (state && state->outfile) { - fflush(state->outfile); - fclose(state->outfile); - state->outfile = NULL; - } -} - /* Retire the global test pointer and free it, under the mutex. */ static void retire_test(void) { pthread_mutex_lock(&g_test_mutex); struct iperf_test *test = g_test; g_test = NULL; if (test) { - /* outfile is a closed FILE* by now — point it somewhere harmless so - * nothing in the teardown path can write to a dead stream. */ - test->outfile = stdout; iperf_free_test(test); } pthread_mutex_unlock(&g_test_mutex); @@ -236,14 +214,8 @@ static void iperf_cleanup_and_signal(JNIEnv *env, struct iperf_run_state *state, char **argv, int argc, int ret, int stopped) { - /* Close the write end FIRST: the reader thread is blocked in fgets() - * and only unblocks on EOF. Joining before this would deadlock. */ - close_outfile(state); - - if (state->reader_started) { - pthread_join(state->reader_thread, NULL); - state->reader_started = 0; - } + /* No more events can be delivered once the run is over. */ + g_current_state = NULL; /* Restore default signal handlers before g_sigend_jmp_buf goes stale. */ iperf_catch_sigend(SIG_DFL); @@ -259,15 +231,11 @@ static void iperf_cleanup_and_signal(JNIEnv *env, /* Signal result to Java */ if (user_stopped) { - signal_output(env, state, "[iPerf JNI] Test was stopped by user.\n"); + signal_output(env, state, "{\"event\":\"jni_stopped\",\"data\":{}}"); } else if (ret < 0) { char errbuf[256]; - snprintf(errbuf, sizeof(errbuf), - "[iPerf JNI] Test failed: %s\n", - iperf_strerror(i_errno)); + snprintf(errbuf, sizeof(errbuf), "%s", iperf_strerror(i_errno)); signal_error(env, state, errbuf); - } else { - signal_output(env, state, "[iPerf JNI] Test completed successfully.\n"); } /* Signal completion to Java */ @@ -325,20 +293,20 @@ Java_com_novoyuuparosk_iperf3gui_IperfRunner_runIperfLive( if (mid_complete) (*env)->CallVoidMethod(env, callback, mid_complete); \ } while (0) - /* Refuse to start a second concurrent test — the globals below hold + /* Refuse to start a second concurrent test — the globals above hold * exactly one test's worth of state. */ pthread_mutex_lock(&g_test_mutex); int busy = (g_test != NULL); pthread_mutex_unlock(&g_test_mutex); if (busy) { - FAIL_EARLY("[iPerf JNI] A test is already running."); + FAIL_EARLY("A test is already running."); return; } /* Create the iperf test object */ struct iperf_test *test = iperf_new_test(); if (!test) { - FAIL_EARLY("[iPerf JNI] Failed to create iperf test"); + FAIL_EARLY("Failed to create iperf test"); return; } @@ -352,7 +320,7 @@ Java_com_novoyuuparosk_iperf3gui_IperfRunner_runIperfLive( char **argv = build_argv(env, arguments, &total_argc); if (!argv) { retire_test(); - FAIL_EARLY("[iPerf JNI] Failed to allocate argv"); + FAIL_EARLY("Failed to allocate argv"); return; } @@ -365,19 +333,7 @@ Java_com_novoyuuparosk_iperf3gui_IperfRunner_runIperfLive( if (iperf_parse_arguments(test, total_argc, argv) < 0) { char errbuf[256]; snprintf(errbuf, sizeof(errbuf), - "[iPerf JNI] Argument parse error: %s", iperf_strerror(i_errno)); - free_argv(argv, total_argc); - retire_test(); - FAIL_EARLY(errbuf); - return; - } - - /* Create a pipe to capture iperf's output */ - int pipefd[2]; - if (pipe(pipefd) != 0) { - char errbuf[256]; - snprintf(errbuf, sizeof(errbuf), - "[iPerf JNI] Failed to create output pipe: %s", strerror(errno)); + "Argument parse error: %s", iperf_strerror(i_errno)); free_argv(argv, total_argc); retire_test(); FAIL_EARLY(errbuf); @@ -389,25 +345,11 @@ Java_com_novoyuuparosk_iperf3gui_IperfRunner_runIperfLive( calloc(1, sizeof(struct iperf_run_state)); if (!state) { free_argv(argv, total_argc); - close(pipefd[0]); - close(pipefd[1]); retire_test(); - FAIL_EARLY("[iPerf JNI] Failed to allocate run state"); + FAIL_EARLY("Failed to allocate run state"); return; } - FILE *outfile = fdopen(pipefd[1], "w"); - if (!outfile) { - free(state); - free_argv(argv, total_argc); - close(pipefd[0]); - close(pipefd[1]); - retire_test(); - FAIL_EARLY("[iPerf JNI] Failed to open output stream"); - return; - } - - /* Get JavaVM for reader-thread attachment */ JavaVM *jvm; (*env)->GetJavaVM(env, &jvm); @@ -415,14 +357,16 @@ Java_com_novoyuuparosk_iperf3gui_IperfRunner_runIperfLive( state->mid_error = mid_error; state->mid_complete = mid_complete; state->callback_ref = (*env)->NewGlobalRef(env, callback); - state->outfile = outfile; - state->pipe_fd[0] = pipefd[0]; - state->pipe_fd[1] = pipefd[1]; state->jvm = jvm; - /* Redirect iperf's output to our pipe */ - test->outfile = state->outfile; - setvbuf(test->outfile, NULL, _IOLBF, 0); /* Line-buffered */ + /* Deliver results as a stream of JSON events straight to + * json_event_callback, rather than as text through test->outfile. + * --json-stream sets both of these flags, so setting json_stream alone + * would leave iperf3 printing human-readable text as well. */ + iperf_set_test_json_output(test, 1); + iperf_set_test_json_stream(test, 1); + iperf_set_test_json_callback(test, json_event_callback); + g_current_state = state; /* Set temp directory for stream buffers (mkstemp in iperf_new_stream). * On Android, /data/local/tmp is NOT writable from the app sandbox — @@ -430,22 +374,6 @@ Java_com_novoyuuparosk_iperf3gui_IperfRunner_runIperfLive( * the strdup internally. */ iperf_set_test_template(test, g_tmp_template); - /* Spawn reader thread to forward iperf output to Java. Started before - * the test so early output is not lost, and before setjmp so the - * longjmp path can rely on reader_started being accurate. */ - if (pthread_create(&state->reader_thread, NULL, - reader_thread_func, state) != 0) { - close_outfile(state); - close(pipefd[0]); - free_argv(argv, total_argc); - retire_test(); - (*env)->DeleteGlobalRef(env, state->callback_ref); - free(state); - FAIL_EARLY("[iPerf JNI] Failed to start output reader thread"); - return; - } - state->reader_started = 1; - #undef FAIL_EARLY /* @@ -462,8 +390,6 @@ Java_com_novoyuuparosk_iperf3gui_IperfRunner_runIperfLive( signal(SIGPIPE, SIG_IGN); /* Ignore SIGPIPE for socket errors */ iperf_catch_sigend(sigend_handler); - signal_output(env, state, "[iPerf JNI] Starting iPerf3 client...\n"); - g_stop = 0; int ret = iperf_run_client(test); LOGI("iperf_run_client returned: %d i_errno=%d strerror=%s", @@ -478,8 +404,8 @@ Java_com_novoyuuparosk_iperf3gui_IperfRunner_runIperfLive( } else { /* * We got here via longjmp() from sigend_handler. - * iperf_cleanup_and_signal() closes the pipe, joins the reader and - * frees everything, so there is nothing to unwind by hand here. + * iperf_cleanup_and_signal() frees everything, so there is nothing + * to unwind by hand here. */ LOGI("Test interrupted by signal %d", g_signed_sig); g_stop = 1; @@ -515,19 +441,7 @@ Java_com_novoyuuparosk_iperf3gui_IperfRunner_forceStopIperfTest( pthread_mutex_unlock(&g_test_mutex); if (!requested) return; - - /* Notify Java */ - jclass callback_cls = (*env)->GetObjectClass(env, callback); - jmethodID mid_output = (*env)->GetMethodID( - env, callback_cls, "onOutput", "(Ljava/lang/String;)V"); - if (mid_output) { - jstring msg = (*env)->NewStringUTF(env, - "[iPerf JNI] Requested graceful stop.\n"); - if (msg) { - (*env)->CallVoidMethod(env, callback, mid_output, msg); - (*env)->DeleteLocalRef(env, msg); - } - } + LOGI("Graceful stop requested"); } /* ------------------------------------------------------------------ */ From 7d8d06e2e5ba0207c2a01558288302b822f21845 Mon Sep 17 00:00:00 2001 From: Mikkeli Date: Wed, 26 Aug 2026 19:35:10 +0900 Subject: [PATCH 2/4] feat(app): parse json-stream reports into structured events MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The native layer now emits one JSON object per report, so the service parses each into a sealed Iperf3Event — start, interval, end, error — and publishes those instead of opaque lines. Interval samples carry bytes, bits per second, retransmits and per-stream socket/cwnd/rtt detail; the end event carries sent and received totals and CPU utilisation. None of that was available while the output was scraped text. Anything unrecognised — non-JSON, malformed JSON, an event name a future iperf3 adds — becomes an Iperf3LogLine rather than being dropped, so unexpected output still reaches the user. Failures arrive as an `error` event rather than a non-zero return from iperf_run_client, so that event is what now marks a run as failed. The result's rawOutput keeps the JSON lines verbatim, since that is what the native layer actually produced. Co-Authored-By: Claude Opus 5 --- lib/data/services/iperf3_service.dart | 56 ++--- lib/domain/models/iperf3_event.dart | 238 ++++++++++++++++++++++ test/domain/models/iperf3_event_test.dart | 107 ++++++++++ 3 files changed, 375 insertions(+), 26 deletions(-) create mode 100644 lib/domain/models/iperf3_event.dart create mode 100644 test/domain/models/iperf3_event_test.dart diff --git a/lib/data/services/iperf3_service.dart b/lib/data/services/iperf3_service.dart index a3e6296..a9b3036 100644 --- a/lib/data/services/iperf3_service.dart +++ b/lib/data/services/iperf3_service.dart @@ -2,6 +2,7 @@ import 'dart:async'; import 'package:flutter/foundation.dart'; import 'package:flutter/services.dart'; +import 'package:iperf3gui/domain/models/iperf3_event.dart'; import 'package:iperf3gui/domain/models/iperf3_result.dart'; import 'package:iperf3gui/domain/models/test_configuration.dart'; @@ -10,9 +11,11 @@ import 'package:iperf3gui/domain/models/test_configuration.dart'; /// On Android, iperf3 is compiled into the app as a shared library (.so). /// This avoids SELinux restrictions on executing ELF binaries from assets. /// -/// The service talks to the native layer over a MethodChannel. Native output -/// arrives line by line on [outputLines] while the test runs; [run] resolves -/// with the accumulated transcript once the native layer reports completion. +/// The service talks to the native layer over a MethodChannel. The test runs +/// in `--json-stream` mode, so each native report arrives as one JSON object; +/// they are parsed and published on [events] while the test runs. [run] +/// resolves once the native layer reports completion, carrying the raw JSON +/// lines as the result's output. class Iperf3Service { Iperf3Service() { _channel.setMethodCallHandler(_handleNativeCall); @@ -33,16 +36,15 @@ class Iperf3Service { /// Exit code reported when the native layer never signalled completion. static const int exitCodeTimeout = -3; - /// Lines emitted by the native layer during the current run. - final StreamController _outputController = - StreamController.broadcast(); + /// Events parsed from the native layer during the current run. + final StreamController _eventController = + StreamController.broadcast(); - /// Live output, one line per event. Use this to render progress; [run]'s - /// result carries the same text joined together. - Stream get outputLines => _outputController.stream; + /// Live reports, one event per native report. Use this to render progress. + Stream get events => _eventController.stream; - /// Transcript of the current (or most recent) run. - final List _outputBuffer = []; + /// Raw JSON lines received during the current (or most recent) run. + final List _rawLines = []; /// Completer that resolves when the native iperf3 test finishes. Completer? _runCompleter; @@ -71,7 +73,7 @@ class Iperf3Service { /// Release the stream controller. Call from the owner's dispose(). Future dispose() async { _channel.setMethodCallHandler(null); - await _outputController.close(); + await _eventController.close(); } /// Run iperf3 with the given configuration. @@ -88,7 +90,7 @@ class Iperf3Service { _isRunning = true; _sawError = false; _stopRequested = false; - _outputBuffer.clear(); + _rawLines.clear(); final args = config.toCliArgs(); debugPrint('Iperf3Service: running: ${args.join(" ")}'); @@ -121,9 +123,9 @@ class Iperf3Service { debugPrint('Iperf3Service: error stopping after timeout: $e'); return null; }); - _emit('[ERROR] Timed out waiting for iperf3 to finish.'); + _publish(const Iperf3Error('Timed out waiting for iperf3 to finish.')); return Iperf3Result( - rawOutput: _transcript(), + rawOutput: _rawOutput(), exitCode: exitCodeTimeout, ); }, @@ -145,7 +147,12 @@ class Iperf3Service { case 'onOutput': final output = call.arguments['output'] as String?; if (output != null) { - _emit(output); + _rawLines.add(output); + final event = Iperf3Event.parse(output); + // iperf3 reports a failed test as an `error` event rather than a + // non-zero return, so this is where most failures surface. + if (event is Iperf3Error) _sawError = true; + _publish(event); } case 'onError': final error = call.arguments['error'] as String?; @@ -153,7 +160,7 @@ class Iperf3Service { // Not terminal on its own — iperf3 reports the failure reason and // then completes, so wait for onComplete before resolving. _sawError = true; - _emit('[ERROR] $error'); + _publish(Iperf3Error(error)); } case 'onComplete': debugPrint('Iperf3Service: test completed'); @@ -161,7 +168,7 @@ class Iperf3Service { if (completer != null && !completer.isCompleted) { completer.complete( Iperf3Result( - rawOutput: _transcript(), + rawOutput: _rawOutput(), // A user-requested stop is a normal outcome, not a failure. exitCode: (_sawError && !_stopRequested) ? exitCodeNativeError : 0, @@ -174,17 +181,14 @@ class Iperf3Service { } } - void _emit(String line) { - _outputBuffer.add(line); - if (!_outputController.isClosed) { - _outputController.add(line); + void _publish(Iperf3Event event) { + if (!_eventController.isClosed) { + _eventController.add(event); } - debugPrint('Iperf3Service output: ${line.trimRight()}'); } - /// Native output already carries its own newlines; keep it verbatim except - /// for lines we synthesised ourselves. - String _transcript() => _outputBuffer + /// The raw JSON stream as received, one event per line. + String _rawOutput() => _rawLines .map((line) => line.endsWith('\n') ? line : '$line\n') .join(); diff --git a/lib/domain/models/iperf3_event.dart b/lib/domain/models/iperf3_event.dart new file mode 100644 index 0000000..36147cc --- /dev/null +++ b/lib/domain/models/iperf3_event.dart @@ -0,0 +1,238 @@ +import 'dart:convert'; + +/// One report from iperf3's `--json-stream` output. +/// +/// iperf3 emits `{"event": "", "data": {...}}` per report; the events a +/// client run produces are `start`, `interval`, `end` and `error`. The bridge +/// adds `jni_stopped` when a run ended because the user stopped it. +sealed class Iperf3Event { + const Iperf3Event(); + + /// Parse one line of `--json-stream` output. + /// + /// Returns an [Iperf3LogLine] for anything that is not a recognised JSON + /// event, so unexpected output is surfaced rather than dropped. + static Iperf3Event parse(String line) { + final trimmed = line.trim(); + if (trimmed.isEmpty) return const Iperf3LogLine(''); + if (!trimmed.startsWith('{')) return Iperf3LogLine(trimmed); + + final Object? decoded; + try { + decoded = jsonDecode(trimmed); + } on FormatException { + return Iperf3LogLine(trimmed); + } + if (decoded is! Map) return Iperf3LogLine(trimmed); + + final event = decoded['event']; + final data = decoded['data']; + if (event is! String) return Iperf3LogLine(trimmed); + final map = data is Map ? data : const {}; + + return switch (event) { + 'start' => Iperf3Start.fromJson(map), + 'interval' => Iperf3Interval.fromJson(map), + 'end' => Iperf3End.fromJson(map), + 'error' => Iperf3Error(_stringOf(data) ?? 'unknown error'), + 'jni_stopped' => const Iperf3Stopped(), + _ => Iperf3LogLine(trimmed), + }; + } + + static String? _stringOf(Object? value) { + if (value is String) return value; + if (value is Map) { + final error = value['error']; + if (error is String) return error; + } + return null; + } +} + +/// Connection details, emitted once at the start of a test. +class Iperf3Start extends Iperf3Event { + final String? targetHost; + final int? targetPort; + final int? streamCount; + final int? durationSeconds; + final String? protocol; + + const Iperf3Start({ + this.targetHost, + this.targetPort, + this.streamCount, + this.durationSeconds, + this.protocol, + }); + + factory Iperf3Start.fromJson(Map data) { + final connecting = data['connecting_to']; + final testStart = data['test_start']; + return Iperf3Start( + targetHost: connecting is Map ? connecting['host'] as String? : null, + targetPort: connecting is Map ? _asInt(connecting['port']) : null, + streamCount: testStart is Map ? _asInt(testStart['num_streams']) : null, + durationSeconds: testStart is Map ? _asInt(testStart['duration']) : null, + protocol: testStart is Map ? testStart['protocol'] as String? : null, + ); + } +} + +/// One reporting interval, summed across all streams. +class Iperf3Interval extends Iperf3Event { + final double start; + final double end; + final int bytes; + final double bitsPerSecond; + final int? retransmits; + final bool omitted; + + /// Per-stream figures, when more than one stream is running. + final List streams; + + const Iperf3Interval({ + required this.start, + required this.end, + required this.bytes, + required this.bitsPerSecond, + this.retransmits, + this.omitted = false, + this.streams = const [], + }); + + factory Iperf3Interval.fromJson(Map data) { + // Non-bidirectional runs use "sum"; bidir adds "sum_bidir_reverse". + final sum = data['sum'] ?? data['sum_bidir_reverse']; + final sumMap = sum is Map ? sum : const {}; + final rawStreams = data['streams']; + + return Iperf3Interval( + start: _asDouble(sumMap['start']) ?? 0, + end: _asDouble(sumMap['end']) ?? 0, + bytes: _asInt(sumMap['bytes']) ?? 0, + bitsPerSecond: _asDouble(sumMap['bits_per_second']) ?? 0, + retransmits: _asInt(sumMap['retransmits']), + omitted: sumMap['omitted'] == true, + streams: rawStreams is List + ? rawStreams + .whereType>() + .map(Iperf3StreamInterval.fromJson) + .toList(growable: false) + : const [], + ); + } +} + +/// One stream's contribution to an interval. +class Iperf3StreamInterval { + final int? socket; + final int bytes; + final double bitsPerSecond; + final int? retransmits; + final int? sendCwnd; + final int? rtt; + + const Iperf3StreamInterval({ + this.socket, + required this.bytes, + required this.bitsPerSecond, + this.retransmits, + this.sendCwnd, + this.rtt, + }); + + factory Iperf3StreamInterval.fromJson(Map data) { + return Iperf3StreamInterval( + socket: _asInt(data['socket']), + bytes: _asInt(data['bytes']) ?? 0, + bitsPerSecond: _asDouble(data['bits_per_second']) ?? 0, + retransmits: _asInt(data['retransmits']), + sendCwnd: _asInt(data['snd_cwnd']), + rtt: _asInt(data['rtt']), + ); + } +} + +/// Totals for the whole test, emitted once at the end. +class Iperf3End extends Iperf3Event { + final Iperf3Total? sent; + final Iperf3Total? received; + final double? hostCpuPercent; + final double? remoteCpuPercent; + + const Iperf3End({ + this.sent, + this.received, + this.hostCpuPercent, + this.remoteCpuPercent, + }); + + factory Iperf3End.fromJson(Map data) { + final cpu = data['cpu_utilization_percent']; + return Iperf3End( + sent: Iperf3Total.fromJson( + data['sum_sent'] ?? data['sum_sent_bidir_reverse']), + received: Iperf3Total.fromJson( + data['sum_received'] ?? data['sum_received_bidir_reverse']), + hostCpuPercent: cpu is Map ? _asDouble(cpu['host_total']) : null, + remoteCpuPercent: cpu is Map ? _asDouble(cpu['remote_total']) : null, + ); + } +} + +/// A sent or received total from the `end` event. +class Iperf3Total { + final double seconds; + final int bytes; + final double bitsPerSecond; + final int? retransmits; + + const Iperf3Total({ + required this.seconds, + required this.bytes, + required this.bitsPerSecond, + this.retransmits, + }); + + static Iperf3Total? fromJson(Object? data) { + if (data is! Map) return null; + return Iperf3Total( + seconds: _asDouble(data['seconds']) ?? 0, + bytes: _asInt(data['bytes']) ?? 0, + bitsPerSecond: _asDouble(data['bits_per_second']) ?? 0, + retransmits: _asInt(data['retransmits']), + ); + } +} + +/// iperf3 reported an error for this test. +class Iperf3Error extends Iperf3Event { + final String message; + const Iperf3Error(this.message); +} + +/// The run ended because the user stopped it. +class Iperf3Stopped extends Iperf3Event { + const Iperf3Stopped(); +} + +/// Output that was not a recognised JSON event. +class Iperf3LogLine extends Iperf3Event { + final String text; + const Iperf3LogLine(this.text); +} + +int? _asInt(Object? value) => switch (value) { + int v => v, + double v => v.round(), + String v => int.tryParse(v), + _ => null, + }; + +double? _asDouble(Object? value) => switch (value) { + double v => v, + int v => v.toDouble(), + String v => double.tryParse(v), + _ => null, + }; diff --git a/test/domain/models/iperf3_event_test.dart b/test/domain/models/iperf3_event_test.dart new file mode 100644 index 0000000..a5de174 --- /dev/null +++ b/test/domain/models/iperf3_event_test.dart @@ -0,0 +1,107 @@ +import 'package:flutter_test/flutter_test.dart'; +import 'package:iperf3gui/domain/models/iperf3_event.dart'; + +void main() { + group('Iperf3Event.parse', () { + test('parses a start event', () { + const line = '{"event":"start","data":{' + '"connecting_to":{"host":"192.168.2.182","port":5201},' + '"test_start":{"protocol":"TCP","num_streams":4,"duration":10}}}'; + final event = Iperf3Event.parse(line); + expect(event, isA()); + final start = event as Iperf3Start; + expect(start.targetHost, equals('192.168.2.182')); + expect(start.targetPort, equals(5201)); + expect(start.streamCount, equals(4)); + expect(start.durationSeconds, equals(10)); + expect(start.protocol, equals('TCP')); + }); + + test('parses an interval event with per-stream detail', () { + const line = '{"event":"interval","data":{' + '"streams":[{"socket":143,"bytes":49283072,' + '"bits_per_second":394264576.0,"retransmits":0,"snd_cwnd":2023424,' + '"rtt":3521}],' + '"sum":{"start":0.0,"end":1.000151,"seconds":1.000151,' + '"bytes":98566144,"bits_per_second":788410352.0,"retransmits":0,' + '"omitted":false,"sender":true}}}'; + final event = Iperf3Event.parse(line); + expect(event, isA()); + final interval = event as Iperf3Interval; + expect(interval.bytes, equals(98566144)); + expect(interval.bitsPerSecond, closeTo(788410352.0, 0.1)); + expect(interval.retransmits, equals(0)); + expect(interval.omitted, isFalse); + expect(interval.streams, hasLength(1)); + expect(interval.streams.single.socket, equals(143)); + expect(interval.streams.single.rtt, equals(3521)); + }); + + test('marks omitted intervals', () { + const line = '{"event":"interval","data":{"streams":[],' + '"sum":{"start":0.0,"end":1.0,"bytes":1,"bits_per_second":8.0,' + '"omitted":true}}}'; + final interval = Iperf3Event.parse(line) as Iperf3Interval; + expect(interval.omitted, isTrue); + }); + + test('parses an end event with sent and received totals', () { + const line = '{"event":"end","data":{' + '"sum_sent":{"seconds":10.0,"bytes":987654321,' + '"bits_per_second":790123456.0,"retransmits":7},' + '"sum_received":{"seconds":10.01,"bytes":987000000,' + '"bits_per_second":789000000.0},' + '"cpu_utilization_percent":{"host_total":12.5,"remote_total":3.25}}}'; + final end = Iperf3Event.parse(line) as Iperf3End; + expect(end.sent!.bytes, equals(987654321)); + expect(end.sent!.retransmits, equals(7)); + expect(end.received!.bitsPerSecond, closeTo(789000000.0, 0.1)); + expect(end.hostCpuPercent, closeTo(12.5, 0.01)); + expect(end.remoteCpuPercent, closeTo(3.25, 0.01)); + }); + + test('falls back to bidir-reverse keys when present', () { + const line = '{"event":"interval","data":{"streams":[],' + '"sum_bidir_reverse":{"start":0.0,"end":1.0,"bytes":42,' + '"bits_per_second":336.0}}}'; + final interval = Iperf3Event.parse(line) as Iperf3Interval; + expect(interval.bytes, equals(42)); + }); + + test('parses an error event', () { + const line = '{"event":"error","data":"unable to connect to server"}'; + final event = Iperf3Event.parse(line); + expect(event, isA()); + expect((event as Iperf3Error).message, + equals('unable to connect to server')); + }); + + test('parses the bridge stop marker', () { + final event = Iperf3Event.parse('{"event":"jni_stopped","data":{}}'); + expect(event, isA()); + }); + + test('surfaces non-JSON output as a log line rather than dropping it', () { + final event = Iperf3Event.parse('iperf3: something unexpected'); + expect(event, isA()); + expect((event as Iperf3LogLine).text, + equals('iperf3: something unexpected')); + }); + + test('surfaces malformed JSON as a log line', () { + final event = Iperf3Event.parse('{"event":"interval",'); + expect(event, isA()); + }); + + test('treats an unknown event name as a log line', () { + final event = Iperf3Event.parse('{"event":"future_thing","data":{}}'); + expect(event, isA()); + }); + + test('tolerates a missing data object', () { + final event = Iperf3Event.parse('{"event":"interval"}'); + expect(event, isA()); + expect((event as Iperf3Interval).bytes, equals(0)); + }); + }); +} From cbb2aba8b00f5079188bc3fa1c66f27b096c4738 Mon Sep 17 00:00:00 2001 From: Mikkeli Date: Wed, 26 Aug 2026 19:35:10 +0900 Subject: [PATCH 3/4] feat(ui): render the event stream and show live throughput figures MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Consuming JSON means the familiar iperf3 text layout has to be rebuilt, so a pure formatter turns each event back into a line — binary units for transfer totals, decimal units for bitrates, matching how iperf3 quotes each. The view model keeps the interval series and the end summary as data alongside the transcript, and a stat row shows current/total, mean, peak and sample count off that series. A chart can be driven from the same list without re-parsing text. Repaints are coalesced on a 250ms timer rather than one per event. Reports can now arrive several times a second across many streams, and each rebuild re-renders the whole transcript, so notifying per event made the cost of drawing scale with the report rate — which competes with the test itself for CPU and skews the measurement. Warm-up intervals (-O) are marked omitted by iperf3 and are kept out of the series so they cannot drag the mean down. Co-Authored-By: Claude Opus 5 --- .../test_run/iperf3_event_formatter.dart | 95 +++++++++++++++ lib/ui/features/test_run/test_run_screen.dart | 59 +++++++++ .../test_run/test_run_view_model.dart | 113 ++++++++++++++---- .../test_run/iperf3_event_formatter_test.dart | 87 ++++++++++++++ .../test_run/test_run_view_model_test.dart | 95 +++++++++++++-- 5 files changed, 415 insertions(+), 34 deletions(-) create mode 100644 lib/ui/features/test_run/iperf3_event_formatter.dart create mode 100644 test/ui/features/test_run/iperf3_event_formatter_test.dart diff --git a/lib/ui/features/test_run/iperf3_event_formatter.dart b/lib/ui/features/test_run/iperf3_event_formatter.dart new file mode 100644 index 0000000..898533f --- /dev/null +++ b/lib/ui/features/test_run/iperf3_event_formatter.dart @@ -0,0 +1,95 @@ +import 'package:iperf3gui/domain/models/iperf3_event.dart'; + +/// Renders parsed iperf3 events as the kind of line iperf3 itself would print. +/// +/// The app consumes `--json-stream`, so the familiar text layout has to be +/// rebuilt here. Keeping it in one pure function makes it testable and leaves +/// the structured events untouched for anything else that wants them. +String? formatIperf3Event(Iperf3Event event) { + return switch (event) { + Iperf3Start e => _formatStart(e), + Iperf3Interval e => _formatInterval(e), + Iperf3End e => _formatEnd(e), + Iperf3Error e => 'iperf3 error: ${e.message}', + Iperf3Stopped _ => 'Stopped by user.', + Iperf3LogLine e => e.text.isEmpty ? null : e.text, + }; +} + +String _formatStart(Iperf3Start e) { + final parts = []; + if (e.targetHost != null) { + final port = e.targetPort != null ? ', port ${e.targetPort}' : ''; + parts.add('Connecting to host ${e.targetHost}$port'); + } + final detail = [ + if (e.protocol != null) e.protocol!, + if (e.streamCount != null) + '${e.streamCount} stream${e.streamCount == 1 ? '' : 's'}', + if (e.durationSeconds != null) '${e.durationSeconds}s', + ]; + if (detail.isNotEmpty) parts.add(detail.join(' · ')); + return parts.isEmpty ? '' : parts.join('\n'); +} + +String _formatInterval(Iperf3Interval e) { + final window = '${_seconds(e.start)}-${_seconds(e.end)} sec'; + final line = StringBuffer('[SUM] $window ' + '${formatBytes(e.bytes)} ${formatBitrate(e.bitsPerSecond)}'); + if (e.retransmits != null) { + line.write(' ${e.retransmits} retr'); + } + if (e.omitted) line.write(' (omitted)'); + return line.toString(); +} + +String _formatEnd(Iperf3End e) { + final lines = ['- ' * 20]; + final sent = e.sent; + final received = e.received; + if (sent != null) { + lines.add('[SUM] 0.00-${_seconds(sent.seconds)} sec ' + '${formatBytes(sent.bytes)} ${formatBitrate(sent.bitsPerSecond)}' + '${sent.retransmits != null ? ' ${sent.retransmits} retr' : ''}' + ' sender'); + } + if (received != null) { + lines.add('[SUM] 0.00-${_seconds(received.seconds)} sec ' + '${formatBytes(received.bytes)} ' + '${formatBitrate(received.bitsPerSecond)} receiver'); + } + if (e.hostCpuPercent != null) { + lines.add('CPU: local ${e.hostCpuPercent!.toStringAsFixed(1)}%' + '${e.remoteCpuPercent != null ? ', remote ${e.remoteCpuPercent!.toStringAsFixed(1)}%' : ''}'); + } + return lines.join('\n'); +} + +String _seconds(double value) => value.toStringAsFixed(2); + +/// Byte totals the way iperf3 prints them — `93.8 MBytes`. +String formatBytes(int bytes) { + const units = ['Bytes', 'KBytes', 'MBytes', 'GBytes', 'TBytes']; + var value = bytes.toDouble(); + var unit = 0; + while (value >= 1024 && unit < units.length - 1) { + value /= 1024; + unit++; + } + final precision = value >= 100 ? 0 : (value >= 10 ? 1 : 2); + return '${value.toStringAsFixed(precision)} ${units[unit]}'; +} + +/// Bitrates the way iperf3 prints them — `786 Mbits/sec`. Decimal units, since +/// network rates are powers of ten while transfer totals are powers of two. +String formatBitrate(double bitsPerSecond) { + const units = ['bits/sec', 'Kbits/sec', 'Mbits/sec', 'Gbits/sec']; + var value = bitsPerSecond; + var unit = 0; + while (value >= 1000 && unit < units.length - 1) { + value /= 1000; + unit++; + } + final precision = value >= 100 ? 0 : (value >= 10 ? 1 : 2); + return '${value.toStringAsFixed(precision)} ${units[unit]}'; +} diff --git a/lib/ui/features/test_run/test_run_screen.dart b/lib/ui/features/test_run/test_run_screen.dart index fb2a85d..16aa122 100644 --- a/lib/ui/features/test_run/test_run_screen.dart +++ b/lib/ui/features/test_run/test_run_screen.dart @@ -3,6 +3,7 @@ import 'package:flutter/services.dart'; import 'package:provider/provider.dart'; import 'package:iperf3gui/data/services/iperf3_service.dart'; import 'package:iperf3gui/domain/models/test_configuration.dart'; +import 'package:iperf3gui/ui/features/test_run/iperf3_event_formatter.dart'; import 'package:iperf3gui/ui/features/test_run/test_run_view_model.dart'; /// Main test run screen. @@ -193,6 +194,35 @@ class _TestRunContentState extends State<_TestRunContent> { ), const SizedBox(height: 12), + // Live throughput figures, derived from the parsed interval series + Consumer( + builder: (context, vm, _) { + final intervals = vm.intervals; + if (intervals.isEmpty) return const SizedBox.shrink(); + final rates = + intervals.map((i) => i.bitsPerSecond).toList(growable: false); + final latest = rates.last; + final peak = rates.reduce((a, b) => a > b ? a : b); + final mean = rates.reduce((a, b) => a + b) / rates.length; + final summary = vm.summary?.sent; + return Padding( + padding: const EdgeInsets.only(bottom: 12), + child: Row( + mainAxisAlignment: MainAxisAlignment.spaceBetween, + children: [ + _Stat( + label: summary != null ? 'total' : 'now', + value: formatBitrate(summary?.bitsPerSecond ?? latest), + ), + _Stat(label: 'avg', value: formatBitrate(mean)), + _Stat(label: 'peak', value: formatBitrate(peak)), + _Stat(label: 'samples', value: '${rates.length}'), + ], + ), + ); + }, + ), + // Live output Expanded( child: Container( @@ -229,3 +259,32 @@ class _TestRunContentState extends State<_TestRunContent> { ); } } + + +/// One figure in the throughput row. +class _Stat extends StatelessWidget { + const _Stat({required this.label, required this.value}); + + final String label; + final String value; + + @override + Widget build(BuildContext context) { + final theme = Theme.of(context); + return Column( + crossAxisAlignment: CrossAxisAlignment.start, + children: [ + Text( + label, + style: theme.textTheme.labelSmall + ?.copyWith(color: theme.colorScheme.onSurfaceVariant), + ), + Text( + value, + style: theme.textTheme.titleSmall + ?.copyWith(fontFeatures: const [FontFeature.tabularFigures()]), + ), + ], + ); + } +} diff --git a/lib/ui/features/test_run/test_run_view_model.dart b/lib/ui/features/test_run/test_run_view_model.dart index ec267ec..965cb14 100644 --- a/lib/ui/features/test_run/test_run_view_model.dart +++ b/lib/ui/features/test_run/test_run_view_model.dart @@ -2,24 +2,33 @@ import 'dart:async'; import 'package:flutter/foundation.dart'; import 'package:iperf3gui/data/services/iperf3_service.dart'; +import 'package:iperf3gui/domain/models/iperf3_event.dart'; import 'package:iperf3gui/domain/models/iperf3_result.dart'; import 'package:iperf3gui/domain/models/test_configuration.dart'; +import 'package:iperf3gui/ui/features/test_run/iperf3_event_formatter.dart'; /// State of the test run. enum TestRunState { idle, running, completed, error } /// ViewModel for managing test run state. /// -/// Subscribes to [Iperf3Service.outputLines] so the transcript grows while the -/// test runs rather than appearing all at once at the end. +/// Subscribes to [Iperf3Service.events] so the transcript and the interval +/// series both grow while the test runs, rather than appearing at the end. class TestRunViewModel extends ChangeNotifier { final Iperf3Service iperf3Service; + /// How often to notify listeners while output is arriving. iperf3 can report + /// several times a second across many streams, and each rebuild re-renders + /// the whole transcript — so repaints are coalesced instead of one per event. + static const Duration displayInterval = Duration(milliseconds: 250); + TestRunViewModel({required this.iperf3Service}) { - _outputSubscription = iperf3Service.outputLines.listen(_onOutputLine); + _eventSubscription = iperf3Service.events.listen(_onEvent); } - StreamSubscription? _outputSubscription; + StreamSubscription? _eventSubscription; + Timer? _repaintTimer; + bool _dirty = false; TestRunState _state = TestRunState.idle; TestRunState get state => _state; @@ -30,11 +39,26 @@ class TestRunViewModel extends ChangeNotifier { TestConfiguration? _currentConfig; TestConfiguration? get currentConfig => _currentConfig; - final StringBuffer _liveOutput = StringBuffer(); + final StringBuffer _transcript = StringBuffer(); - /// Output accumulated so far. Grows line by line during a run, and holds the - /// finished transcript once the test completes. - String get output => _liveOutput.toString(); + /// Human-readable output accumulated so far. + String get output => _transcript.toString(); + + final List _intervals = []; + + /// Per-interval throughput samples, in arrival order. Structured, so a chart + /// can be driven from this without re-parsing text. + List get intervals => List.unmodifiable(_intervals); + + Iperf3Start? _start; + + /// Connection details reported at the start of the run. + Iperf3Start? get start => _start; + + Iperf3End? _summary; + + /// Totals reported at the end of the run. + Iperf3End? get summary => _summary; /// Whether a test is currently in flight. bool get isRunning => _state == TestRunState.running; @@ -47,35 +71,32 @@ class TestRunViewModel extends ChangeNotifier { if (_state == TestRunState.running) return; if (!config.isValid) { - _liveOutput - ..clear() - ..writeln('Invalid configuration: check the server address and port.'); + _resetRunData(); + _transcript + .writeln('Invalid configuration: check the server address and port.'); _state = TestRunState.error; - _result = null; notifyListeners(); return; } + _resetRunData(); _state = TestRunState.running; _currentConfig = config; - _result = null; - _liveOutput.clear(); notifyListeners(); try { final result = await iperf3Service.run(config); _result = result; - // Prefer the streamed transcript — it is what the user watched build up. - // Fall back to the service's copy if nothing streamed through. - if (_liveOutput.isEmpty && result.rawOutput.isNotEmpty) { - _liveOutput.write(result.rawOutput); + if (_transcript.isEmpty && result.rawOutput.isNotEmpty) { + _transcript.write(result.rawOutput); } _state = result.exitCode == 0 ? TestRunState.completed : TestRunState.error; } catch (e) { - _liveOutput.writeln('Test failed: $e'); + _transcript.writeln('Test failed: $e'); _state = TestRunState.error; } finally { + _stopRepaintTimer(); notifyListeners(); } } @@ -88,21 +109,63 @@ class TestRunViewModel extends ChangeNotifier { /// Reset to idle state. void reset() { + _resetRunData(); _state = TestRunState.idle; - _result = null; _currentConfig = null; - _liveOutput.clear(); notifyListeners(); } - void _onOutputLine(String line) { - _liveOutput.write(line.endsWith('\n') ? line : '$line\n'); - notifyListeners(); + void _resetRunData() { + _result = null; + _transcript.clear(); + _intervals.clear(); + _start = null; + _summary = null; + _dirty = false; + } + + void _onEvent(Iperf3Event event) { + switch (event) { + case Iperf3Start e: + _start = e; + case Iperf3Interval e: + // Omitted intervals are warm-up (-O) and are not part of the series. + if (!e.omitted) _intervals.add(e); + case Iperf3End e: + _summary = e; + case Iperf3Error _: + case Iperf3Stopped _: + case Iperf3LogLine _: + break; + } + + final line = formatIperf3Event(event); + if (line != null && line.isNotEmpty) { + _transcript.writeln(line); + } + _scheduleRepaint(); + } + + /// Coalesce repaints: mark dirty and let a timer flush at [displayInterval]. + void _scheduleRepaint() { + _dirty = true; + _repaintTimer ??= Timer.periodic(displayInterval, (_) { + if (!_dirty) return; + _dirty = false; + notifyListeners(); + }); + } + + void _stopRepaintTimer() { + _repaintTimer?.cancel(); + _repaintTimer = null; + _dirty = false; } @override void dispose() { - _outputSubscription?.cancel(); + _stopRepaintTimer(); + _eventSubscription?.cancel(); iperf3Service.dispose(); super.dispose(); } diff --git a/test/ui/features/test_run/iperf3_event_formatter_test.dart b/test/ui/features/test_run/iperf3_event_formatter_test.dart new file mode 100644 index 0000000..f5abc90 --- /dev/null +++ b/test/ui/features/test_run/iperf3_event_formatter_test.dart @@ -0,0 +1,87 @@ +import 'package:flutter_test/flutter_test.dart'; +import 'package:iperf3gui/domain/models/iperf3_event.dart'; +import 'package:iperf3gui/ui/features/test_run/iperf3_event_formatter.dart'; + +void main() { + group('formatBytes', () { + test('scales in binary units the way iperf3 prints totals', () { + expect(formatBytes(0), equals('0.00 Bytes')); + expect(formatBytes(1024), equals('1.00 KBytes')); + expect(formatBytes(98566144), equals('94.0 MBytes')); + expect(formatBytes(1073741824), equals('1.00 GBytes')); + }); + }); + + group('formatBitrate', () { + test('scales in decimal units, as network rates are quoted', () { + expect(formatBitrate(0), equals('0.00 bits/sec')); + expect(formatBitrate(8000), equals('8.00 Kbits/sec')); + expect(formatBitrate(790000000), equals('790 Mbits/sec')); + expect(formatBitrate(2500000000), equals('2.50 Gbits/sec')); + }); + }); + + group('formatIperf3Event', () { + test('renders a start event', () { + const event = Iperf3Start( + targetHost: '192.168.2.182', + targetPort: 5201, + protocol: 'TCP', + streamCount: 1, + durationSeconds: 10, + ); + final text = formatIperf3Event(event)!; + expect(text, contains('Connecting to host 192.168.2.182, port 5201')); + expect(text, contains('TCP')); + expect(text, contains('1 stream')); + }); + + test('renders an interval with retransmits', () { + const event = Iperf3Interval( + start: 0, + end: 1, + bytes: 98566144, + bitsPerSecond: 788410352, + retransmits: 3, + ); + final text = formatIperf3Event(event)!; + expect(text, contains('0.00-1.00 sec')); + expect(text, contains('94.0 MBytes')); + expect(text, contains('788 Mbits/sec')); + expect(text, contains('3 retr')); + }); + + test('marks an omitted interval', () { + const event = Iperf3Interval( + start: 0, end: 1, bytes: 1, bitsPerSecond: 8, omitted: true); + expect(formatIperf3Event(event), contains('omitted')); + }); + + test('renders sender and receiver lines from an end event', () { + const event = Iperf3End( + sent: Iperf3Total( + seconds: 10, bytes: 987654321, bitsPerSecond: 790000000, + retransmits: 7), + received: Iperf3Total( + seconds: 10.01, bytes: 987000000, bitsPerSecond: 789000000), + hostCpuPercent: 12.5, + ); + final text = formatIperf3Event(event)!; + expect(text, contains('sender')); + expect(text, contains('receiver')); + expect(text, contains('7 retr')); + expect(text, contains('CPU: local 12.5%')); + }); + + test('renders errors and stops', () { + expect(formatIperf3Event(const Iperf3Error('boom')), + equals('iperf3 error: boom')); + expect(formatIperf3Event(const Iperf3Stopped()), + equals('Stopped by user.')); + }); + + test('drops an empty log line', () { + expect(formatIperf3Event(const Iperf3LogLine('')), isNull); + }); + }); +} diff --git a/test/ui/features/test_run/test_run_view_model_test.dart b/test/ui/features/test_run/test_run_view_model_test.dart index 6f83d36..025019c 100644 --- a/test/ui/features/test_run/test_run_view_model_test.dart +++ b/test/ui/features/test_run/test_run_view_model_test.dart @@ -2,6 +2,7 @@ import 'dart:async'; import 'package:flutter_test/flutter_test.dart'; import 'package:iperf3gui/data/services/iperf3_service.dart'; +import 'package:iperf3gui/domain/models/iperf3_event.dart'; import 'package:iperf3gui/domain/models/iperf3_result.dart'; import 'package:iperf3gui/domain/models/test_configuration.dart'; import 'package:iperf3gui/ui/features/test_run/test_run_view_model.dart'; @@ -56,11 +57,16 @@ void main() { expect(vm.result!.exitCode, equals(1)); }); - test('streamed lines accumulate into output during the run', () async { + test('streamed events accumulate into output during the run', () async { final mock = _MockIperf3Service(); mock.onRun = (_) async { - mock.emit('Connecting to host 192.168.1.1\n'); - mock.emit('[ 5] 0.00-1.00 sec 11.2 MBytes 94.1 Mbits/sec\n'); + mock.emit(const Iperf3Start(targetHost: '192.168.1.1', targetPort: 5201)); + mock.emit(const Iperf3Interval( + start: 0, + end: 1, + bytes: 11744051, + bitsPerSecond: 94100000, + )); // Let the broadcast stream deliver before the run resolves. await Future.delayed(Duration.zero); return const Iperf3Result(rawOutput: 'ignored', exitCode: 0); @@ -69,12 +75,83 @@ void main() { final vm = TestRunViewModel(iperf3Service: mock); await vm.runTest(const TestConfiguration(server: '192.168.1.1')); - expect(vm.output, contains('Connecting to host')); + expect(vm.output, contains('Connecting to host 192.168.1.1')); expect(vm.output, contains('94.1 Mbits/sec')); // The streamed transcript wins over the service's own copy. expect(vm.output, isNot(contains('ignored'))); }); + test('intervals are collected as structured samples', () async { + final mock = _MockIperf3Service(); + mock.onRun = (_) async { + mock.emit(const Iperf3Interval( + start: 0, end: 1, bytes: 1000, bitsPerSecond: 8000)); + mock.emit(const Iperf3Interval( + start: 1, end: 2, bytes: 2000, bitsPerSecond: 16000)); + await Future.delayed(Duration.zero); + return const Iperf3Result(rawOutput: '', exitCode: 0); + }; + + final vm = TestRunViewModel(iperf3Service: mock); + await vm.runTest(const TestConfiguration(server: '192.168.1.1')); + + expect(vm.intervals, hasLength(2)); + expect(vm.intervals[1].bitsPerSecond, equals(16000)); + }); + + test('omitted warm-up intervals are excluded from the series', () async { + final mock = _MockIperf3Service(); + mock.onRun = (_) async { + mock.emit(const Iperf3Interval( + start: 0, end: 1, bytes: 1, bitsPerSecond: 1, omitted: true)); + mock.emit(const Iperf3Interval( + start: 1, end: 2, bytes: 2000, bitsPerSecond: 16000)); + await Future.delayed(Duration.zero); + return const Iperf3Result(rawOutput: '', exitCode: 0); + }; + + final vm = TestRunViewModel(iperf3Service: mock); + await vm.runTest(const TestConfiguration(server: '192.168.1.1')); + + expect(vm.intervals, hasLength(1)); + expect(vm.intervals.single.bitsPerSecond, equals(16000)); + }); + + test('end event is retained as the summary', () async { + final mock = _MockIperf3Service(); + mock.onRun = (_) async { + mock.emit(const Iperf3End( + sent: Iperf3Total( + seconds: 10, bytes: 987654321, bitsPerSecond: 790000000), + )); + await Future.delayed(Duration.zero); + return const Iperf3Result(rawOutput: '', exitCode: 0); + }; + + final vm = TestRunViewModel(iperf3Service: mock); + await vm.runTest(const TestConfiguration(server: '192.168.1.1')); + + expect(vm.summary?.sent?.bitsPerSecond, equals(790000000)); + expect(vm.output, contains('790 Mbits/sec')); + }); + + test('a new run clears the previous interval series', () async { + final mock = _MockIperf3Service(); + mock.onRun = (_) async { + mock.emit(const Iperf3Interval( + start: 0, end: 1, bytes: 1000, bitsPerSecond: 8000)); + await Future.delayed(Duration.zero); + return const Iperf3Result(rawOutput: '', exitCode: 0); + }; + + final vm = TestRunViewModel(iperf3Service: mock); + await vm.runTest(const TestConfiguration(server: '192.168.1.1')); + expect(vm.intervals, hasLength(1)); + + await vm.runTest(const TestConfiguration(server: '192.168.1.1')); + expect(vm.intervals, hasLength(1), reason: 'not appended to the old run'); + }); + test('falls back to result output when nothing streamed', () async { final mock = _MockIperf3Service( onRun: (_) async => @@ -166,19 +243,19 @@ class _MockIperf3Service implements Iperf3Service { Future Function(TestConfiguration)? onRun; - final StreamController _controller = - StreamController.broadcast(); + final StreamController _controller = + StreamController.broadcast(); int runCallCount = 0; int stopCallCount = 0; bool disposed = false; bool _isRunning = false; - /// Push a line as if the native layer had emitted it. - void emit(String line) => _controller.add(line); + /// Push an event as if the native layer had reported it. + void emit(Iperf3Event event) => _controller.add(event); @override - Stream get outputLines => _controller.stream; + Stream get events => _controller.stream; @override bool get isRunning => _isRunning; From 857b79fbaf56d61cffb7734b0f0de8bf40995071 Mon Sep 17 00:00:00 2001 From: Mikkeli Date: Wed, 26 Aug 2026 19:35:41 +0900 Subject: [PATCH 4/4] docs: replace the pipe design with json_callback and note the embedding costs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The output-streaming section described the pipe and reader thread that no longer exist, including the join-ordering rule that was the reason they were dangerous. Replaced with how the json_callback path works and why, and an explicit note not to reintroduce the pipe. Adds a section on what embedding a CLI in a long-lived app process actually costs, since that is the recurring trap here: iperf3's main() can leave process-global state behind because it is about to exit, and this app cannot. Records what iperf3 does clean up — joined stream threads, closed fds, an immediately unlinked temp file — with the measurements taken on device, so the next person does not have to re-derive whether repeated runs leak. Co-Authored-By: Claude Opus 5 --- docs/README.md | 58 ++++++++++++++++++++++++++++++++++++++++---------- 1 file changed, 47 insertions(+), 11 deletions(-) diff --git a/docs/README.md b/docs/README.md index 321cfbc..9728d37 100644 --- a/docs/README.md +++ b/docs/README.md @@ -42,8 +42,9 @@ User taps GO → Kotlin IperfRunner: submits to the iperf3-run worker thread, returns at once → JNI C: build_argv() (prepends argv[0]) → iperf_parse_arguments() → iperf_run_client(test) [blocks on the worker thread] - → Reader thread pipes output → CallVoidMethod → main-thread hop → Dart onOutput - → Dart streams each line to the UI via Iperf3Service.outputLines + → iperf3 json_callback → one JSON event per report → CallVoidMethod + → main-thread hop → Dart onOutput → Iperf3Event.parse() + → Iperf3Service.events → view model: transcript + interval series + summary → On completion: Dart Completer resolves with Iperf3Result ``` @@ -61,16 +62,51 @@ to the main looper — stalled every callback until the test was already over. ### Callback Mechanism The JNI bridge uses `CallVoidMethod` on stored `jmethodID`s to invoke the Kotlin `IperfCallback` interface, NOT function pointer casts. Direct function pointer casts caused SIGBUS crashes on ARM due to JNINativeInterface method ID layout differences. -### Output Streaming -iperf3 output is redirected via pipe to a background reader thread that attaches -to the Java VM via `AttachCurrentThread` and invokes callbacks via -`CallVoidMethod`. Dart re-publishes each line on `Iperf3Service.outputLines`, so -the transcript grows while the test runs. +### Output via json_callback, not a pipe +The test runs with `json_output` and `json_stream` set and a `json_callback` +installed, so iperf3 hands each report to the bridge as a JSON string and never +writes to `test->outfile`. `iperf.h` describes the hook as "allow user apps to +receive the JSON strings, instead of writing them to the output file" — this is +its intended use. -The reader blocks in `fgets()` and only unblocks on EOF, so the pipe's write end -must be closed **before** `pthread_join()` — joining first deadlocks the worker -thread. The JNI caller solely owns the run-state struct and frees it after the -join; the reader must never free it. +An earlier design piped `test->outfile` to a reader thread. That imposed two +constraints, both of which were gotten wrong and deadlocked every test: the +pipe's write end had to be closed *before* joining a reader blocked in +`fgets()`, and exactly one side could own the shared state struct. Reading from +the callback removes the pipe, the reader thread and both constraints. Don't +reintroduce it. + +`json_callback` carries no user-data pointer, so the run state is reached +through a global published before `iperf_run_client()` and cleared afterwards; +only one test may run at a time, which `runIperfLive` enforces. The callback +fires on iperf3's reporting thread, so it resolves a `JNIEnv` for the calling +thread rather than reusing a stored one. + +Dart parses each event into a sealed `Iperf3Event`; unrecognised output becomes +an `Iperf3LogLine` rather than being discarded. The UI rebuilds iperf3's text +layout from the events and keeps the interval series as data. + +### Embedding a CLI in a long-lived process +iperf3's `main()` runs one test and exits, so anything it leaves behind is +reclaimed by the OS. Linked into an app process that runs tests repeatedly for +hours, every such leftover becomes this app's problem. What that costs in +practice: + +- **Signal handlers are process-global.** `iperf_catch_sigend()` and + `signal(SIGPIPE, SIG_IGN)` change behaviour for the whole process, Flutter and + ART included. Handlers are restored to `SIG_DFL` when a test ends. +- **`g_sigend_jmp_buf` outlives its frame.** Harmless in a process about to + exit; a jump into a dead stack frame otherwise. Same fix as above. +- **Globals mean one test at a time.** `i_errno` and the bridge's own statics + hold a single test's worth of state, so concurrent runs are refused. + +What iperf3 itself handles correctly, verified rather than assumed: per-stream +threads are joined (`iperf_client_api.c`), stream buffer fds are closed, and the +`mkstemp` temp file is `unlink`ed immediately after creation +(`iperf_api.c:4652`), so nothing accumulates in the cache dir. Measured across +four consecutive 4-stream runs on device, thread count returned to baseline +(33 → 40 mid-run → 34) and fd count returned exactly to baseline (175 → 183 → +175). ### Stop Handling A stop shuts down the control socket, so `iperf_run_client()` returns -1 with