refactor(jni): take results from json_callback instead of piping stdout

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 <noreply@anthropic.com>
This commit is contained in:
2026-08-26 19:35:10 +09:00
parent d448d5f885
commit 598a2d99c9
+67 -153
View File
@@ -5,16 +5,25 @@
* the Flutter app to run iperf3 tests natively inside the app process * the Flutter app to run iperf3 tests natively inside the app process
* (no Process.start needed, no SELinux restrictions). * (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: * THREADING NOTE:
* runIperfLive() BLOCKS for the whole duration of the test. Kotlin must * runIperfLive() BLOCKS for the whole duration of the test. Kotlin must
* call it from a worker thread, never from the Flutter platform thread. * call it from a worker thread, never from the Flutter platform thread.
* * json_callback fires on whichever thread iperf3 reports from, so it resolves
* PIPE / SHUTDOWN NOTE: * a JNIEnv for the calling thread rather than reusing a stored one.
* 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.
* *
* SIGNAL HANDLING NOTE: * SIGNAL HANDLING NOTE:
* iperf3's run() function (called from iperf_run_client()) uses * 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 volatile int g_stop = 0;
static char g_tmp_template[1024] = "/data/local/tmp/iperf3.XXXXXX"; static char g_tmp_template[1024] = "/data/local/tmp/iperf3.XXXXXX";
/* Per-call state shared between the JNI caller and the reader thread. /* Per-call state, reachable from the json_callback.
* We store the callback object (global ref) and method IDs.
* Callbacks are invoked via CallVoidMethod, NOT as function pointers. */ * Callbacks are invoked via CallVoidMethod, NOT as function pointers. */
struct iperf_run_state { struct iperf_run_state {
jobject callback_ref; /* Global ref to the callback object */ jobject callback_ref; /* Global ref to the callback object */
jmethodID mid_output; /* Method ID for onOutput(String) */ jmethodID mid_output; /* Method ID for onOutput(String) */
jmethodID mid_error; /* Method ID for onError(String) */ jmethodID mid_error; /* Method ID for onError(String) */
jmethodID mid_complete; /* Method ID for onComplete() */ jmethodID mid_complete; /* Method ID for onComplete() */
FILE *outfile; /* Pipe write end; NULL once closed */
int pipe_fd[2];
JavaVM *jvm; 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 */ /* Helper: invoke the output callback via CallVoidMethod */
static void signal_output(JNIEnv *env, struct iperf_run_state *state, static void signal_output(JNIEnv *env, struct iperf_run_state *state,
const char *msg) { 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 */ /* json_callback — receives one JSON event string per report from */
/* each line to the Java callback via JNI. */ /* 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 */ /* Runs on iperf3's reporting thread. In the client that is the thread */
/* `state`: the JNI caller frees it after joining this 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) { static void json_event_callback(struct iperf_test *test, char *json) {
struct iperf_run_state *state = (struct iperf_run_state *) arg; struct iperf_run_state *state = g_current_state;
JNIEnv *env; if (!state || !json) return;
int ret;
/* Attach this native thread to the Java VM */ JNIEnv *env = NULL;
ret = (*state->jvm)->AttachCurrentThread(state->jvm, &env, NULL); int attached = 0;
if (ret != JNI_OK) {
LOGE("AttachCurrentThread failed: %d", ret); if ((*state->jvm)->GetEnv(state->jvm, (void **) &env, JNI_VERSION_1_6)
/* Still drain the pipe so writers never block on a full pipe. */ != JNI_OK) {
char drain[1024]; if ((*state->jvm)->AttachCurrentThread(state->jvm, &env, NULL)
while (read(state->pipe_fd[0], drain, sizeof(drain)) > 0) { } != JNI_OK) {
close(state->pipe_fd[0]); LOGE("json_event_callback: could not reach a JNIEnv");
return NULL; return;
}
attached = 1;
} }
char buf[1024]; signal_output(env, state, json);
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);
}
}
fclose(fp); /* also closes pipe_fd[0] */
} else {
close(state->pipe_fd[0]);
}
/* Detach this native thread */ if (attached) {
(*state->jvm)->DetachCurrentThread(state->jvm); (*state->jvm)->DetachCurrentThread(state->jvm);
return NULL; }
} }
/* ------------------------------------------------------------------ */ /* ------------------------------------------------------------------ */
@@ -198,30 +194,12 @@ static void free_argv(char **argv, int argc) {
free(argv); 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. */ /* Retire the global test pointer and free it, under the mutex. */
static void retire_test(void) { static void retire_test(void) {
pthread_mutex_lock(&g_test_mutex); pthread_mutex_lock(&g_test_mutex);
struct iperf_test *test = g_test; struct iperf_test *test = g_test;
g_test = NULL; g_test = NULL;
if (test) { 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); iperf_free_test(test);
} }
pthread_mutex_unlock(&g_test_mutex); pthread_mutex_unlock(&g_test_mutex);
@@ -236,14 +214,8 @@ static void iperf_cleanup_and_signal(JNIEnv *env,
struct iperf_run_state *state, struct iperf_run_state *state,
char **argv, int argc, char **argv, int argc,
int ret, int stopped) { int ret, int stopped) {
/* Close the write end FIRST: the reader thread is blocked in fgets() /* No more events can be delivered once the run is over. */
* and only unblocks on EOF. Joining before this would deadlock. */ g_current_state = NULL;
close_outfile(state);
if (state->reader_started) {
pthread_join(state->reader_thread, NULL);
state->reader_started = 0;
}
/* Restore default signal handlers before g_sigend_jmp_buf goes stale. */ /* Restore default signal handlers before g_sigend_jmp_buf goes stale. */
iperf_catch_sigend(SIG_DFL); iperf_catch_sigend(SIG_DFL);
@@ -259,15 +231,11 @@ static void iperf_cleanup_and_signal(JNIEnv *env,
/* Signal result to Java */ /* Signal result to Java */
if (user_stopped) { 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) { } else if (ret < 0) {
char errbuf[256]; char errbuf[256];
snprintf(errbuf, sizeof(errbuf), snprintf(errbuf, sizeof(errbuf), "%s", iperf_strerror(i_errno));
"[iPerf JNI] Test failed: %s\n",
iperf_strerror(i_errno));
signal_error(env, state, errbuf); signal_error(env, state, errbuf);
} else {
signal_output(env, state, "[iPerf JNI] Test completed successfully.\n");
} }
/* Signal completion to Java */ /* Signal completion to Java */
@@ -325,20 +293,20 @@ Java_com_novoyuuparosk_iperf3gui_IperfRunner_runIperfLive(
if (mid_complete) (*env)->CallVoidMethod(env, callback, mid_complete); \ if (mid_complete) (*env)->CallVoidMethod(env, callback, mid_complete); \
} while (0) } 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. */ * exactly one test's worth of state. */
pthread_mutex_lock(&g_test_mutex); pthread_mutex_lock(&g_test_mutex);
int busy = (g_test != NULL); int busy = (g_test != NULL);
pthread_mutex_unlock(&g_test_mutex); pthread_mutex_unlock(&g_test_mutex);
if (busy) { if (busy) {
FAIL_EARLY("[iPerf JNI] A test is already running."); FAIL_EARLY("A test is already running.");
return; return;
} }
/* Create the iperf test object */ /* Create the iperf test object */
struct iperf_test *test = iperf_new_test(); struct iperf_test *test = iperf_new_test();
if (!test) { if (!test) {
FAIL_EARLY("[iPerf JNI] Failed to create iperf test"); FAIL_EARLY("Failed to create iperf test");
return; return;
} }
@@ -352,7 +320,7 @@ Java_com_novoyuuparosk_iperf3gui_IperfRunner_runIperfLive(
char **argv = build_argv(env, arguments, &total_argc); char **argv = build_argv(env, arguments, &total_argc);
if (!argv) { if (!argv) {
retire_test(); retire_test();
FAIL_EARLY("[iPerf JNI] Failed to allocate argv"); FAIL_EARLY("Failed to allocate argv");
return; return;
} }
@@ -365,19 +333,7 @@ Java_com_novoyuuparosk_iperf3gui_IperfRunner_runIperfLive(
if (iperf_parse_arguments(test, total_argc, argv) < 0) { if (iperf_parse_arguments(test, total_argc, argv) < 0) {
char errbuf[256]; char errbuf[256];
snprintf(errbuf, sizeof(errbuf), snprintf(errbuf, sizeof(errbuf),
"[iPerf JNI] Argument parse error: %s", iperf_strerror(i_errno)); "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));
free_argv(argv, total_argc); free_argv(argv, total_argc);
retire_test(); retire_test();
FAIL_EARLY(errbuf); FAIL_EARLY(errbuf);
@@ -389,25 +345,11 @@ Java_com_novoyuuparosk_iperf3gui_IperfRunner_runIperfLive(
calloc(1, sizeof(struct iperf_run_state)); calloc(1, sizeof(struct iperf_run_state));
if (!state) { if (!state) {
free_argv(argv, total_argc); free_argv(argv, total_argc);
close(pipefd[0]);
close(pipefd[1]);
retire_test(); retire_test();
FAIL_EARLY("[iPerf JNI] Failed to allocate run state"); FAIL_EARLY("Failed to allocate run state");
return; 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; JavaVM *jvm;
(*env)->GetJavaVM(env, &jvm); (*env)->GetJavaVM(env, &jvm);
@@ -415,14 +357,16 @@ Java_com_novoyuuparosk_iperf3gui_IperfRunner_runIperfLive(
state->mid_error = mid_error; state->mid_error = mid_error;
state->mid_complete = mid_complete; state->mid_complete = mid_complete;
state->callback_ref = (*env)->NewGlobalRef(env, callback); 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; state->jvm = jvm;
/* Redirect iperf's output to our pipe */ /* Deliver results as a stream of JSON events straight to
test->outfile = state->outfile; * json_event_callback, rather than as text through test->outfile.
setvbuf(test->outfile, NULL, _IOLBF, 0); /* Line-buffered */ * --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). /* Set temp directory for stream buffers (mkstemp in iperf_new_stream).
* On Android, /data/local/tmp is NOT writable from the app sandbox — * 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. */ * the strdup internally. */
iperf_set_test_template(test, g_tmp_template); 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 #undef FAIL_EARLY
/* /*
@@ -462,8 +390,6 @@ Java_com_novoyuuparosk_iperf3gui_IperfRunner_runIperfLive(
signal(SIGPIPE, SIG_IGN); /* Ignore SIGPIPE for socket errors */ signal(SIGPIPE, SIG_IGN); /* Ignore SIGPIPE for socket errors */
iperf_catch_sigend(sigend_handler); iperf_catch_sigend(sigend_handler);
signal_output(env, state, "[iPerf JNI] Starting iPerf3 client...\n");
g_stop = 0; g_stop = 0;
int ret = iperf_run_client(test); int ret = iperf_run_client(test);
LOGI("iperf_run_client returned: %d i_errno=%d strerror=%s", LOGI("iperf_run_client returned: %d i_errno=%d strerror=%s",
@@ -478,8 +404,8 @@ Java_com_novoyuuparosk_iperf3gui_IperfRunner_runIperfLive(
} else { } else {
/* /*
* We got here via longjmp() from sigend_handler. * We got here via longjmp() from sigend_handler.
* iperf_cleanup_and_signal() closes the pipe, joins the reader and * iperf_cleanup_and_signal() frees everything, so there is nothing
* frees everything, so there is nothing to unwind by hand here. * to unwind by hand here.
*/ */
LOGI("Test interrupted by signal %d", g_signed_sig); LOGI("Test interrupted by signal %d", g_signed_sig);
g_stop = 1; g_stop = 1;
@@ -515,19 +441,7 @@ Java_com_novoyuuparosk_iperf3gui_IperfRunner_forceStopIperfTest(
pthread_mutex_unlock(&g_test_mutex); pthread_mutex_unlock(&g_test_mutex);
if (!requested) return; if (!requested) return;
LOGI("Graceful stop requested");
/* 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);
}
}
} }
/* ------------------------------------------------------------------ */ /* ------------------------------------------------------------------ */