From 374437023d237951d6792c61a1dadbad9cefc1c2 Mon Sep 17 00:00:00 2001 From: AlexProgrammerDE <40795980+AlexProgrammerDE@users.noreply.github.com> Date: Sun, 9 Aug 2026 22:20:10 +0200 Subject: [PATCH 1/4] fix(jni): detach terminated native threads --- build.gradle.kts | 1 + ...ibdatachannel.convention.common.gradle.kts | 1 + gradle/libs.versions.toml | 1 + jni/CMakeLists.txt | 4 ++ jni/build.sh | 3 +- jni/src/init.c | 6 +- jni/test/thread_lifecycle.c | 71 +++++++++++++++++++ .../NativeThreadLifecycleTest.java | 24 +++++++ 8 files changed, 107 insertions(+), 4 deletions(-) create mode 100644 jni/test/thread_lifecycle.c create mode 100644 src/test/java/tel/schich/libdatachannel/NativeThreadLifecycleTest.java diff --git a/build.gradle.kts b/build.gradle.kts index 0ee5e15..20c034a 100644 --- a/build.gradle.kts +++ b/build.gradle.kts @@ -184,6 +184,7 @@ val dockcrossOutputDir: Directory = project.layout.buildDirectory.get().dir("doc val nativeForHostOutputDir: Directory = dockcrossOutputDir.dir("host") val compileNativeForHost by tasks.registering(DockcrossRunTask::class) { baseConfigure(nativeForHostOutputDir, BuildTarget(image = null, family = "host", classifier = "host")) + extraEnv.put("BUILD_JNI_TESTS", "ON") unsafeWritableMountSource = true runner(NonContainerRunner) } diff --git a/conventions/src/main/kotlin/tel.schich.libdatachannel.convention.common.gradle.kts b/conventions/src/main/kotlin/tel.schich.libdatachannel.convention.common.gradle.kts index f3f8d1b..bb74dfa 100644 --- a/conventions/src/main/kotlin/tel.schich.libdatachannel.convention.common.gradle.kts +++ b/conventions/src/main/kotlin/tel.schich.libdatachannel.convention.common.gradle.kts @@ -44,6 +44,7 @@ dependencies { implementation(libs.slf4j) testImplementation(libs.junitJupiter) + testRuntimeOnly(libs.junitPlatformLauncher) testImplementation(libs.logbackClassic) } diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index 932513b..f5c0c09 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -12,6 +12,7 @@ jniAccessGenerator = { module = "tel.schich:jni-access-generator", version.ref = jdtAnnotations = { module = "org.eclipse.jdt:org.eclipse.jdt.annotation", version.ref = "jdtAnnotations" } slf4j = { module = "org.slf4j:slf4j-api", version.ref = "slf4j" } logbackClassic = { module = "ch.qos.logback:logback-classic", version.ref = "logback" } +junitPlatformLauncher = { module = "org.junit.platform:junit-platform-launcher", version.ref = "junitJupiter" } junitJupiter = { module = "org.junit.jupiter:junit-jupiter-engine", version.ref = "junitJupiter" } [plugins] diff --git a/jni/CMakeLists.txt b/jni/CMakeLists.txt index 92b23f8..a970714 100644 --- a/jni/CMakeLists.txt +++ b/jni/CMakeLists.txt @@ -5,6 +5,7 @@ set(CMAKE_C_STANDARD 11) set(CMAKE_CXX_STANDARD 11) option(PROJECT_VERSION "The version of the project" "unspecified") +option(BUILD_JNI_TESTS "Build JNI test support" OFF) set(NO_WEBSOCKET ON CACHE BOOL "configure libdatachannel build") set(NO_MEDIA ON CACHE BOOL "configure libdatachannel build") @@ -56,4 +57,7 @@ add_library(datachannel-java SHARED src/native_peer.c src/native_track.c src/callback.c) +if(BUILD_JNI_TESTS) + target_sources(datachannel-java PRIVATE test/thread_lifecycle.c) +endif() target_link_libraries(datachannel-java PRIVATE datachannel-static) diff --git a/jni/build.sh b/jni/build.sh index 01e0761..ff25206 100755 --- a/jni/build.sh +++ b/jni/build.sh @@ -27,6 +27,7 @@ cmake_options=( "-DCMAKE_PROJECT_TOP_LEVEL_INCLUDES=${MOUNT_SOURCE}/jni/cmake-conan/conan_provider.cmake" "-DPROJECT_VERSION=${PROJECT_VERSION}" "-DCMAKE_BUILD_TYPE=${PROJECT_BUILD_TYPE}" + "-DBUILD_JNI_TESTS=${BUILD_JNI_TESTS:-OFF}" ) if [ "$TARGET_FAMILY" = 'android' ] @@ -46,4 +47,4 @@ then fi cmake "$RELATIVE_PROJECT_PATH" "${cmake_options[@]}" -make -j"${JOBS:-1}" \ No newline at end of file +make -j"${JOBS:-1}" diff --git a/jni/src/init.c b/jni/src/init.c index ddf16ae..5289bd2 100644 --- a/jni/src/init.c +++ b/jni/src/init.c @@ -9,8 +9,8 @@ static JavaVM* global_JVM; static pthread_key_t thread_key; -void detach_thread() { - JavaVM* jvm = pthread_getspecific(thread_key); +static void detach_thread(void* value) { + JavaVM* jvm = value; if (jvm != NULL) { (*jvm)->DetachCurrentThread(jvm); } @@ -64,4 +64,4 @@ JNIEXPORT void JNICALL JNI_OnUnload(JavaVM* jvm, void* reserved) { JNIEnv* env = get_jni_env(); module_OnUnload(env); global_JVM = NULL; -} \ No newline at end of file +} diff --git a/jni/test/thread_lifecycle.c b/jni/test/thread_lifecycle.c new file mode 100644 index 0000000..8d7b618 --- /dev/null +++ b/jni/test/thread_lifecycle.c @@ -0,0 +1,71 @@ +#include "../src/global_jvm.h" +#include "../src/util.h" + +#include +#include + +struct thread_result { + jobject thread; + char* error; +}; + +static void* attach_and_terminate(void* data) { + struct thread_result* result = data; + JNIEnv* env = get_jni_env(); + if (env == NULL) { + result->error = "Failed to attach native test thread"; + return NULL; + } + + jclass thread_class = (*env)->FindClass(env, "java/lang/Thread"); + if (thread_class == NULL) { + (*env)->ExceptionClear(env); + result->error = "Failed to find java.lang.Thread"; + return NULL; + } + + jmethodID current_thread = (*env)->GetStaticMethodID(env, thread_class, "currentThread", "()Ljava/lang/Thread;"); + if (current_thread == NULL) { + (*env)->ExceptionClear(env); + result->error = "Failed to find Thread.currentThread"; + return NULL; + } + + jobject thread = (*env)->CallStaticObjectMethod(env, thread_class, current_thread); + if (thread == NULL || (*env)->ExceptionCheck(env)) { + (*env)->ExceptionClear(env); + result->error = "Failed to get current native thread"; + return NULL; + } + + result->thread = (*env)->NewGlobalRef(env, thread); + if (result->thread == NULL) { + if ((*env)->ExceptionCheck(env)) { + (*env)->ExceptionClear(env); + } + result->error = "Failed to retain current native thread"; + } + return NULL; +} + +JNIEXPORT jobject JNICALL +Java_tel_schich_libdatachannel_NativeThreadLifecycleTest_attachAndTerminateNativeThread(JNIEnv* env, jclass clazz) { + struct thread_result result = {0}; + pthread_t thread; + if (pthread_create(&thread, NULL, attach_and_terminate, &result) != 0) { + throw_native_exception(env, "Failed to create native test thread"); + return NULL; + } + if (pthread_join(thread, NULL) != 0) { + throw_native_exception(env, "Failed to join native test thread"); + return NULL; + } + if (result.error != NULL) { + throw_native_exception(env, result.error); + return NULL; + } + + jobject java_thread = (*env)->NewLocalRef(env, result.thread); + (*env)->DeleteGlobalRef(env, result.thread); + return java_thread; +} diff --git a/src/test/java/tel/schich/libdatachannel/NativeThreadLifecycleTest.java b/src/test/java/tel/schich/libdatachannel/NativeThreadLifecycleTest.java new file mode 100644 index 0000000..908f442 --- /dev/null +++ b/src/test/java/tel/schich/libdatachannel/NativeThreadLifecycleTest.java @@ -0,0 +1,24 @@ +package tel.schich.libdatachannel; + +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +class NativeThreadLifecycleTest { + static { + LibDataChannel.initialize(); + } + + private static native Thread attachAndTerminateNativeThread(); + + @Test + void detachesTerminatedNativeThread() { + Thread thread = attachAndTerminateNativeThread(); + + assertNotNull(thread); + assertFalse(thread.isAlive()); + assertEquals(Thread.State.TERMINATED, thread.getState()); + } +} From 932f58a6f0da551bbcb1ecf993c7a56c41e5faf0 Mon Sep 17 00:00:00 2001 From: AlexProgrammerDE <40795980+AlexProgrammerDE@users.noreply.github.com> Date: Sun, 9 Aug 2026 22:20:37 +0200 Subject: [PATCH 2/4] fix(jni): retain peer callbacks through deletion --- jni/src/native_peer.c | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/jni/src/native_peer.c b/jni/src/native_peer.c index 9feb7b3..6094a3f 100644 --- a/jni/src/native_peer.c +++ b/jni/src/native_peer.c @@ -139,11 +139,12 @@ JNIEXPORT jint JNICALL Java_tel_schich_libdatachannel_LibDataChannelNative_rtcDeletePeerConnection(JNIEnv* env, jclass clazz, jint peerHandle) { struct jvm_callback* callback = rtcGetUserPointer(peerHandle); - if (callback != NULL) { + jint result = rtcDeletePeerConnection(peerHandle); + if (result == RTC_ERR_SUCCESS && callback != NULL) { free_callback(env, callback); } - return rtcDeletePeerConnection(peerHandle); + return result; } @@ -270,4 +271,4 @@ JNIEXPORT jint JNICALL Java_tel_schich_libdatachannel_LibDataChannelNative_setup rtcSetUserPointer(peerHandle, jvm_callback); return RTC_ERR_SUCCESS; -} \ No newline at end of file +} From 180d4d3ce5e077344a19d93db7788f4fb6af33a6 Mon Sep 17 00:00:00 2001 From: AlexProgrammerDE <40795980+AlexProgrammerDE@users.noreply.github.com> Date: Thu, 20 Aug 2026 14:34:53 +0200 Subject: [PATCH 3/4] test(jni): cover peer callback deletion from Java Exercise a real scheduled peer callback while deletion runs and verify the Java listener remains reachable until native callback draining completes. --- jni/CMakeLists.txt | 5 +- jni/src/native_peer.c | 9 ++ jni/test/callback_lifecycle.c | 83 ++++++++++++ jni/test/callback_lifecycle.h | 8 ++ .../PeerCallbackLifecycleTest.java | 118 ++++++++++++++++++ 5 files changed, 222 insertions(+), 1 deletion(-) create mode 100644 jni/test/callback_lifecycle.c create mode 100644 jni/test/callback_lifecycle.h create mode 100644 src/test/java/tel/schich/libdatachannel/PeerCallbackLifecycleTest.java diff --git a/jni/CMakeLists.txt b/jni/CMakeLists.txt index a970714..e354569 100644 --- a/jni/CMakeLists.txt +++ b/jni/CMakeLists.txt @@ -58,6 +58,9 @@ add_library(datachannel-java SHARED src/native_track.c src/callback.c) if(BUILD_JNI_TESTS) - target_sources(datachannel-java PRIVATE test/thread_lifecycle.c) + target_compile_definitions(datachannel-java PRIVATE BUILD_JNI_TESTS) + target_sources(datachannel-java PRIVATE + test/callback_lifecycle.c + test/thread_lifecycle.c) endif() target_link_libraries(datachannel-java PRIVATE datachannel-static) diff --git a/jni/src/native_peer.c b/jni/src/native_peer.c index 6094a3f..12ecc8b 100644 --- a/jni/src/native_peer.c +++ b/jni/src/native_peer.c @@ -6,6 +6,10 @@ #include #include +#ifdef BUILD_JNI_TESTS +#include "../test/callback_lifecycle.h" +#endif + void RTC_API handle_local_description(int pc, const char* sdp, const char* type, void* ptr) { DISPATCH_JNI(call_tel_schich_libdatachannel_PeerConnectionListener_onLocalDescription_cstr, sdp, type); } @@ -32,6 +36,11 @@ void RTC_API handle_gathering_state_change(int pc, rtcGatheringState state, void SET_CALLBACK_INTERFACE_IMPL(rtcSetGatheringStateChangeCallback, handle_gathering_state_change) void RTC_API handle_signaling_state_change(int pc, rtcSignalingState state, void* ptr) { +#ifdef BUILD_JNI_TESTS + if (!wait_for_signaling_state_callback_test()) { + return; + } +#endif DISPATCH_JNI(call_tel_schich_libdatachannel_PeerConnectionListener_onSignalingStateChange, state); } SET_CALLBACK_INTERFACE_IMPL(rtcSetSignalingStateChangeCallback, handle_signaling_state_change) diff --git a/jni/test/callback_lifecycle.c b/jni/test/callback_lifecycle.c new file mode 100644 index 0000000..6282f1a --- /dev/null +++ b/jni/test/callback_lifecycle.c @@ -0,0 +1,83 @@ +#include "callback_lifecycle.h" + +#include +#include +#include +#include + +static pthread_mutex_t callback_mutex = PTHREAD_MUTEX_INITIALIZER; +static pthread_cond_t callback_condition = PTHREAD_COND_INITIALIZER; +static bool callback_armed; +static bool callback_entered; +static bool callback_released; +static bool callback_should_dispatch; + +static struct timespec deadline_after_millis(jlong timeout_millis) { + struct timespec deadline; + timespec_get(&deadline, TIME_UTC); + deadline.tv_sec += timeout_millis / 1000; + deadline.tv_nsec += (timeout_millis % 1000) * 1000000; + if (deadline.tv_nsec >= 1000000000) { + deadline.tv_sec++; + deadline.tv_nsec -= 1000000000; + } + return deadline; +} + +bool wait_for_signaling_state_callback_test(void) { + pthread_mutex_lock(&callback_mutex); + if (!callback_armed) { + pthread_mutex_unlock(&callback_mutex); + return true; + } + + callback_entered = true; + pthread_cond_broadcast(&callback_condition); + while (!callback_released) { + pthread_cond_wait(&callback_condition, &callback_mutex); + } + + bool should_dispatch = callback_should_dispatch; + callback_armed = false; + callback_entered = false; + callback_released = false; + callback_should_dispatch = false; + pthread_mutex_unlock(&callback_mutex); + return should_dispatch; +} + +JNIEXPORT void JNICALL +Java_tel_schich_libdatachannel_PeerCallbackLifecycleTest_armSignalingStateCallback(JNIEnv* env, jclass clazz) { + pthread_mutex_lock(&callback_mutex); + callback_armed = true; + callback_entered = false; + callback_released = false; + callback_should_dispatch = false; + pthread_mutex_unlock(&callback_mutex); +} + +JNIEXPORT jboolean JNICALL +Java_tel_schich_libdatachannel_PeerCallbackLifecycleTest_awaitSignalingStateCallback(JNIEnv* env, jclass clazz, + jlong timeout_millis) { + struct timespec deadline = deadline_after_millis(timeout_millis); + pthread_mutex_lock(&callback_mutex); + while (!callback_entered) { + int result = pthread_cond_timedwait(&callback_condition, &callback_mutex, &deadline); + if (result == ETIMEDOUT) { + pthread_mutex_unlock(&callback_mutex); + return JNI_FALSE; + } + } + pthread_mutex_unlock(&callback_mutex); + return JNI_TRUE; +} + +JNIEXPORT void JNICALL +Java_tel_schich_libdatachannel_PeerCallbackLifecycleTest_releaseSignalingStateCallback(JNIEnv* env, jclass clazz, + jboolean dispatch) { + pthread_mutex_lock(&callback_mutex); + callback_should_dispatch = dispatch == JNI_TRUE; + callback_released = true; + pthread_cond_broadcast(&callback_condition); + pthread_mutex_unlock(&callback_mutex); +} diff --git a/jni/test/callback_lifecycle.h b/jni/test/callback_lifecycle.h new file mode 100644 index 0000000..078e9eb --- /dev/null +++ b/jni/test/callback_lifecycle.h @@ -0,0 +1,8 @@ +#ifndef LIBDATACHANNEL_JNI_CALLBACK_LIFECYCLE_H +#define LIBDATACHANNEL_JNI_CALLBACK_LIFECYCLE_H + +#include + +bool wait_for_signaling_state_callback_test(void); + +#endif // LIBDATACHANNEL_JNI_CALLBACK_LIFECYCLE_H diff --git a/src/test/java/tel/schich/libdatachannel/PeerCallbackLifecycleTest.java b/src/test/java/tel/schich/libdatachannel/PeerCallbackLifecycleTest.java new file mode 100644 index 0000000..124f69e --- /dev/null +++ b/src/test/java/tel/schich/libdatachannel/PeerCallbackLifecycleTest.java @@ -0,0 +1,118 @@ +package tel.schich.libdatachannel; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +import java.lang.ref.WeakReference; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static tel.schich.libdatachannel.LibDataChannelNative.rtcDeletePeerConnection; + +class PeerCallbackLifecycleTest { + private static final long CALLBACK_TIMEOUT_MILLIS = TimeUnit.SECONDS.toMillis(5); + + static { + LibDataChannel.initialize(); + } + + private static native void armSignalingStateCallback(); + + private static native boolean awaitSignalingStateCallback(long timeoutMillis); + + private static native void releaseSignalingStateCallback(boolean dispatch); + + @Test + @Timeout(15) + void retainsCallbackUntilPeerDeletionDrainsScheduledCallbacks() throws InterruptedException { + BlockedPeer blockedPeer = createPeerWithBlockedCallback(); + AtomicReference deletionFailure = new AtomicReference<>(); + AtomicInteger deletionResult = new AtomicInteger(); + CountDownLatch deletionStarted = new CountDownLatch(1); + Thread deletionThread = new Thread(() -> { + deletionStarted.countDown(); + try { + deletionResult.set(rtcDeletePeerConnection(blockedPeer.peerHandle)); + } catch (Throwable failure) { + deletionFailure.set(failure); + } + }, "peer-deletion-test"); + + boolean dispatchCallback = false; + try { + deletionThread.start(); + assertTrue(deletionStarted.await(1, TimeUnit.SECONDS)); + awaitDeletionBlock(deletionThread); + + forceGarbageCollection(); + assertNotNull(blockedPeer.listener.get(), + "peer deletion released its callback before the blocked native callback returned"); + dispatchCallback = true; + } finally { + releaseSignalingStateCallback(dispatchCallback); + deletionThread.join(CALLBACK_TIMEOUT_MILLIS); + } + + assertFalse(deletionThread.isAlive(), "peer deletion did not finish after the callback returned"); + assertNull(deletionFailure.get()); + assertEquals(0, deletionResult.get()); + assertEquals(1, blockedPeer.callbackCount.get()); + } + + private static BlockedPeer createPeerWithBlockedCallback() { + PeerConnectionConfiguration configuration = PeerConnectionConfiguration.DEFAULT + .withDisableAutoNegotiation(true); + PeerConnection peer = PeerConnection.createPeer(configuration); + DataChannel channel = peer.createDataChannel("lifecycle-test"); + AtomicInteger callbackCount = new AtomicInteger(); + peer.onSignalingStateChange.register((ignoredPeer, ignoredState) -> callbackCount.incrementAndGet()); + + armSignalingStateCallback(); + peer.setLocalDescription("offer"); + boolean callbackEntered = awaitSignalingStateCallback(CALLBACK_TIMEOUT_MILLIS); + if (!callbackEntered) { + releaseSignalingStateCallback(false); + } + assertTrue(callbackEntered, "the native signaling callback did not start"); + + channel.close(); + return new BlockedPeer(peer.peerHandle, new WeakReference<>(peer.listener), callbackCount); + } + + private static void awaitDeletionBlock(Thread deletionThread) throws InterruptedException { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(1); + while (deletionThread.isAlive() && deletionThread.getState() == Thread.State.RUNNABLE + && System.nanoTime() < deadline) { + Thread.sleep(10); + } + assertTrue(deletionThread.isAlive(), "peer deletion returned before the native callback was released"); + } + + private static void forceGarbageCollection() throws InterruptedException { + for (int attempt = 0; attempt < 10; attempt++) { + System.gc(); + System.runFinalization(); + Thread.sleep(20); + } + } + + private static final class BlockedPeer { + private final int peerHandle; + private final WeakReference listener; + private final AtomicInteger callbackCount; + + private BlockedPeer(int peerHandle, WeakReference listener, + AtomicInteger callbackCount) { + this.peerHandle = peerHandle; + this.listener = listener; + this.callbackCount = callbackCount; + } + } +} From 86f1ae58ba6647d5a8ca199f2e6606fec99aecfe Mon Sep 17 00:00:00 2001 From: AlexProgrammerDE <40795980+AlexProgrammerDE@users.noreply.github.com> Date: Mon, 24 Aug 2026 22:18:46 +0200 Subject: [PATCH 4/4] test: reproduce the JNI lifecycle bugs from plain Java The previous tests reached both race conditions through extra JNI sources compiled into the shared library, which put a test hook in the middle of a production callback. Neither bug needs that. The thread test churns peer connections and watches the JVM thread list. Every peer attaches a native thread that terminates with the peer, so a missing detach leaves one permanent thread record behind per peer. The callback test parks a peer inside a Java callback, deletes the peer from another thread and then lets the callback return, so the callbacks queued behind it are drained while the deletion waits. Freeing the callback state before the deletion turns that into a use after free, which takes the whole JVM down, so the scenario runs in a child process and the test looks at how that process ended. --- build.gradle.kts | 1 - jni/CMakeLists.txt | 7 - jni/build.sh | 1 - jni/src/native_peer.c | 9 - jni/test/callback_lifecycle.c | 83 -------- jni/test/callback_lifecycle.h | 8 - jni/test/thread_lifecycle.c | 71 ------- .../NativeThreadLifecycleTest.java | 62 ++++-- .../PeerCallbackLifecycleTest.java | 177 +++++++++--------- 9 files changed, 135 insertions(+), 284 deletions(-) delete mode 100644 jni/test/callback_lifecycle.c delete mode 100644 jni/test/callback_lifecycle.h delete mode 100644 jni/test/thread_lifecycle.c diff --git a/build.gradle.kts b/build.gradle.kts index 20c034a..0ee5e15 100644 --- a/build.gradle.kts +++ b/build.gradle.kts @@ -184,7 +184,6 @@ val dockcrossOutputDir: Directory = project.layout.buildDirectory.get().dir("doc val nativeForHostOutputDir: Directory = dockcrossOutputDir.dir("host") val compileNativeForHost by tasks.registering(DockcrossRunTask::class) { baseConfigure(nativeForHostOutputDir, BuildTarget(image = null, family = "host", classifier = "host")) - extraEnv.put("BUILD_JNI_TESTS", "ON") unsafeWritableMountSource = true runner(NonContainerRunner) } diff --git a/jni/CMakeLists.txt b/jni/CMakeLists.txt index e354569..92b23f8 100644 --- a/jni/CMakeLists.txt +++ b/jni/CMakeLists.txt @@ -5,7 +5,6 @@ set(CMAKE_C_STANDARD 11) set(CMAKE_CXX_STANDARD 11) option(PROJECT_VERSION "The version of the project" "unspecified") -option(BUILD_JNI_TESTS "Build JNI test support" OFF) set(NO_WEBSOCKET ON CACHE BOOL "configure libdatachannel build") set(NO_MEDIA ON CACHE BOOL "configure libdatachannel build") @@ -57,10 +56,4 @@ add_library(datachannel-java SHARED src/native_peer.c src/native_track.c src/callback.c) -if(BUILD_JNI_TESTS) - target_compile_definitions(datachannel-java PRIVATE BUILD_JNI_TESTS) - target_sources(datachannel-java PRIVATE - test/callback_lifecycle.c - test/thread_lifecycle.c) -endif() target_link_libraries(datachannel-java PRIVATE datachannel-static) diff --git a/jni/build.sh b/jni/build.sh index ff25206..58a93e1 100755 --- a/jni/build.sh +++ b/jni/build.sh @@ -27,7 +27,6 @@ cmake_options=( "-DCMAKE_PROJECT_TOP_LEVEL_INCLUDES=${MOUNT_SOURCE}/jni/cmake-conan/conan_provider.cmake" "-DPROJECT_VERSION=${PROJECT_VERSION}" "-DCMAKE_BUILD_TYPE=${PROJECT_BUILD_TYPE}" - "-DBUILD_JNI_TESTS=${BUILD_JNI_TESTS:-OFF}" ) if [ "$TARGET_FAMILY" = 'android' ] diff --git a/jni/src/native_peer.c b/jni/src/native_peer.c index 12ecc8b..6094a3f 100644 --- a/jni/src/native_peer.c +++ b/jni/src/native_peer.c @@ -6,10 +6,6 @@ #include #include -#ifdef BUILD_JNI_TESTS -#include "../test/callback_lifecycle.h" -#endif - void RTC_API handle_local_description(int pc, const char* sdp, const char* type, void* ptr) { DISPATCH_JNI(call_tel_schich_libdatachannel_PeerConnectionListener_onLocalDescription_cstr, sdp, type); } @@ -36,11 +32,6 @@ void RTC_API handle_gathering_state_change(int pc, rtcGatheringState state, void SET_CALLBACK_INTERFACE_IMPL(rtcSetGatheringStateChangeCallback, handle_gathering_state_change) void RTC_API handle_signaling_state_change(int pc, rtcSignalingState state, void* ptr) { -#ifdef BUILD_JNI_TESTS - if (!wait_for_signaling_state_callback_test()) { - return; - } -#endif DISPATCH_JNI(call_tel_schich_libdatachannel_PeerConnectionListener_onSignalingStateChange, state); } SET_CALLBACK_INTERFACE_IMPL(rtcSetSignalingStateChangeCallback, handle_signaling_state_change) diff --git a/jni/test/callback_lifecycle.c b/jni/test/callback_lifecycle.c deleted file mode 100644 index 6282f1a..0000000 --- a/jni/test/callback_lifecycle.c +++ /dev/null @@ -1,83 +0,0 @@ -#include "callback_lifecycle.h" - -#include -#include -#include -#include - -static pthread_mutex_t callback_mutex = PTHREAD_MUTEX_INITIALIZER; -static pthread_cond_t callback_condition = PTHREAD_COND_INITIALIZER; -static bool callback_armed; -static bool callback_entered; -static bool callback_released; -static bool callback_should_dispatch; - -static struct timespec deadline_after_millis(jlong timeout_millis) { - struct timespec deadline; - timespec_get(&deadline, TIME_UTC); - deadline.tv_sec += timeout_millis / 1000; - deadline.tv_nsec += (timeout_millis % 1000) * 1000000; - if (deadline.tv_nsec >= 1000000000) { - deadline.tv_sec++; - deadline.tv_nsec -= 1000000000; - } - return deadline; -} - -bool wait_for_signaling_state_callback_test(void) { - pthread_mutex_lock(&callback_mutex); - if (!callback_armed) { - pthread_mutex_unlock(&callback_mutex); - return true; - } - - callback_entered = true; - pthread_cond_broadcast(&callback_condition); - while (!callback_released) { - pthread_cond_wait(&callback_condition, &callback_mutex); - } - - bool should_dispatch = callback_should_dispatch; - callback_armed = false; - callback_entered = false; - callback_released = false; - callback_should_dispatch = false; - pthread_mutex_unlock(&callback_mutex); - return should_dispatch; -} - -JNIEXPORT void JNICALL -Java_tel_schich_libdatachannel_PeerCallbackLifecycleTest_armSignalingStateCallback(JNIEnv* env, jclass clazz) { - pthread_mutex_lock(&callback_mutex); - callback_armed = true; - callback_entered = false; - callback_released = false; - callback_should_dispatch = false; - pthread_mutex_unlock(&callback_mutex); -} - -JNIEXPORT jboolean JNICALL -Java_tel_schich_libdatachannel_PeerCallbackLifecycleTest_awaitSignalingStateCallback(JNIEnv* env, jclass clazz, - jlong timeout_millis) { - struct timespec deadline = deadline_after_millis(timeout_millis); - pthread_mutex_lock(&callback_mutex); - while (!callback_entered) { - int result = pthread_cond_timedwait(&callback_condition, &callback_mutex, &deadline); - if (result == ETIMEDOUT) { - pthread_mutex_unlock(&callback_mutex); - return JNI_FALSE; - } - } - pthread_mutex_unlock(&callback_mutex); - return JNI_TRUE; -} - -JNIEXPORT void JNICALL -Java_tel_schich_libdatachannel_PeerCallbackLifecycleTest_releaseSignalingStateCallback(JNIEnv* env, jclass clazz, - jboolean dispatch) { - pthread_mutex_lock(&callback_mutex); - callback_should_dispatch = dispatch == JNI_TRUE; - callback_released = true; - pthread_cond_broadcast(&callback_condition); - pthread_mutex_unlock(&callback_mutex); -} diff --git a/jni/test/callback_lifecycle.h b/jni/test/callback_lifecycle.h deleted file mode 100644 index 078e9eb..0000000 --- a/jni/test/callback_lifecycle.h +++ /dev/null @@ -1,8 +0,0 @@ -#ifndef LIBDATACHANNEL_JNI_CALLBACK_LIFECYCLE_H -#define LIBDATACHANNEL_JNI_CALLBACK_LIFECYCLE_H - -#include - -bool wait_for_signaling_state_callback_test(void); - -#endif // LIBDATACHANNEL_JNI_CALLBACK_LIFECYCLE_H diff --git a/jni/test/thread_lifecycle.c b/jni/test/thread_lifecycle.c deleted file mode 100644 index 8d7b618..0000000 --- a/jni/test/thread_lifecycle.c +++ /dev/null @@ -1,71 +0,0 @@ -#include "../src/global_jvm.h" -#include "../src/util.h" - -#include -#include - -struct thread_result { - jobject thread; - char* error; -}; - -static void* attach_and_terminate(void* data) { - struct thread_result* result = data; - JNIEnv* env = get_jni_env(); - if (env == NULL) { - result->error = "Failed to attach native test thread"; - return NULL; - } - - jclass thread_class = (*env)->FindClass(env, "java/lang/Thread"); - if (thread_class == NULL) { - (*env)->ExceptionClear(env); - result->error = "Failed to find java.lang.Thread"; - return NULL; - } - - jmethodID current_thread = (*env)->GetStaticMethodID(env, thread_class, "currentThread", "()Ljava/lang/Thread;"); - if (current_thread == NULL) { - (*env)->ExceptionClear(env); - result->error = "Failed to find Thread.currentThread"; - return NULL; - } - - jobject thread = (*env)->CallStaticObjectMethod(env, thread_class, current_thread); - if (thread == NULL || (*env)->ExceptionCheck(env)) { - (*env)->ExceptionClear(env); - result->error = "Failed to get current native thread"; - return NULL; - } - - result->thread = (*env)->NewGlobalRef(env, thread); - if (result->thread == NULL) { - if ((*env)->ExceptionCheck(env)) { - (*env)->ExceptionClear(env); - } - result->error = "Failed to retain current native thread"; - } - return NULL; -} - -JNIEXPORT jobject JNICALL -Java_tel_schich_libdatachannel_NativeThreadLifecycleTest_attachAndTerminateNativeThread(JNIEnv* env, jclass clazz) { - struct thread_result result = {0}; - pthread_t thread; - if (pthread_create(&thread, NULL, attach_and_terminate, &result) != 0) { - throw_native_exception(env, "Failed to create native test thread"); - return NULL; - } - if (pthread_join(thread, NULL) != 0) { - throw_native_exception(env, "Failed to join native test thread"); - return NULL; - } - if (result.error != NULL) { - throw_native_exception(env, result.error); - return NULL; - } - - jobject java_thread = (*env)->NewLocalRef(env, result.thread); - (*env)->DeleteGlobalRef(env, result.thread); - return java_thread; -} diff --git a/src/test/java/tel/schich/libdatachannel/NativeThreadLifecycleTest.java b/src/test/java/tel/schich/libdatachannel/NativeThreadLifecycleTest.java index 908f442..d603e1e 100644 --- a/src/test/java/tel/schich/libdatachannel/NativeThreadLifecycleTest.java +++ b/src/test/java/tel/schich/libdatachannel/NativeThreadLifecycleTest.java @@ -1,24 +1,60 @@ package tel.schich.libdatachannel; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; -import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertFalse; -import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; +/** + * libdatachannel attaches its own native threads to the JVM whenever they reach a callback. Threads that only live for + * as long as a peer connection have to be detached again when they terminate, otherwise the JVM keeps a record for + * every one of them and its thread list grows for the rest of the process. + */ class NativeThreadLifecycleTest { - static { - LibDataChannel.initialize(); - } - - private static native Thread attachAndTerminateNativeThread(); + private static final int MEASURED_PEERS = 40; + /** + * The threads that stay around are already attached once the count settles, so a healthy run adds none. The slack + * only absorbs unrelated JVM and logging threads. + */ + private static final int TOLERATED_GROWTH = 10; + private static final int SETTLE_SAMPLES = 5; + private static final int SETTLE_LIMIT = 100; @Test - void detachesTerminatedNativeThread() { - Thread thread = attachAndTerminateNativeThread(); + @Timeout(120) + void terminatedNativeThreadsDoNotStayAttached() { + int threadsBefore = churnUntilThreadCountSettles(); + for (int i = 0; i < MEASURED_PEERS; i++) { + churnPeer(); + } + int growth = Thread.getAllStackTraces().size() - threadsBefore; + + assertTrue(growth <= TOLERATED_GROWTH, + MEASURED_PEERS + " peer connections left " + growth + " additional threads registered with the JVM"); + } + + /** + * libdatachannel's worker pool has a fixed size but attaches its threads lazily, so the thread count keeps climbing + * for the first few peers even when nothing leaks. Churn peers until it stops moving, or give up and let the + * assertion report what it sees. + */ + private static int churnUntilThreadCountSettles() { + int stableSamples = 0; + int previousCount = -1; + for (int i = 0; i < SETTLE_LIMIT && stableSamples < SETTLE_SAMPLES; i++) { + churnPeer(); + int count = Thread.getAllStackTraces().size(); + stableSamples = count == previousCount ? stableSamples + 1 : 0; + previousCount = count; + } + return previousCount; + } - assertNotNull(thread); - assertFalse(thread.isAlive()); - assertEquals(Thread.State.TERMINATED, thread.getState()); + private static void churnPeer() { + PeerConnectionConfiguration config = PeerConnectionConfiguration.DEFAULT.withDisableAutoNegotiation(true); + try (PeerConnection peer = PeerConnection.createPeer(config)) { + peer.createDataChannel("thread-lifecycle"); + peer.setLocalDescription("offer"); + } } } diff --git a/src/test/java/tel/schich/libdatachannel/PeerCallbackLifecycleTest.java b/src/test/java/tel/schich/libdatachannel/PeerCallbackLifecycleTest.java index 124f69e..f678b76 100644 --- a/src/test/java/tel/schich/libdatachannel/PeerCallbackLifecycleTest.java +++ b/src/test/java/tel/schich/libdatachannel/PeerCallbackLifecycleTest.java @@ -3,116 +3,111 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Timeout; -import java.lang.ref.WeakReference; +import java.io.IOException; +import java.io.InputStream; +import java.nio.charset.StandardCharsets; +import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.concurrent.atomic.AtomicReference; import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertFalse; -import static org.junit.jupiter.api.Assertions.assertNotNull; -import static org.junit.jupiter.api.Assertions.assertNull; -import static org.junit.jupiter.api.Assertions.assertTrue; import static tel.schich.libdatachannel.LibDataChannelNative.rtcDeletePeerConnection; +/** + * Deleting a peer connection blocks until libdatachannel has drained the callbacks it already scheduled. The callback + * state behind the peer's user pointer therefore has to outlive that call, otherwise a draining callback reads memory + * that has already been freed. + * + *

A use after free takes the whole JVM down, so the scenario runs in a child process and this test only looks at how + * that process ended. + */ class PeerCallbackLifecycleTest { - private static final long CALLBACK_TIMEOUT_MILLIS = TimeUnit.SECONDS.toMillis(5); - - static { - LibDataChannel.initialize(); - } - - private static native void armSignalingStateCallback(); - - private static native boolean awaitSignalingStateCallback(long timeoutMillis); - - private static native void releaseSignalingStateCallback(boolean dispatch); - @Test - @Timeout(15) - void retainsCallbackUntilPeerDeletionDrainsScheduledCallbacks() throws InterruptedException { - BlockedPeer blockedPeer = createPeerWithBlockedCallback(); - AtomicReference deletionFailure = new AtomicReference<>(); - AtomicInteger deletionResult = new AtomicInteger(); - CountDownLatch deletionStarted = new CountDownLatch(1); - Thread deletionThread = new Thread(() -> { - deletionStarted.countDown(); - try { - deletionResult.set(rtcDeletePeerConnection(blockedPeer.peerHandle)); - } catch (Throwable failure) { - deletionFailure.set(failure); - } - }, "peer-deletion-test"); - - boolean dispatchCallback = false; - try { - deletionThread.start(); - assertTrue(deletionStarted.await(1, TimeUnit.SECONDS)); - awaitDeletionBlock(deletionThread); - - forceGarbageCollection(); - assertNotNull(blockedPeer.listener.get(), - "peer deletion released its callback before the blocked native callback returned"); - dispatchCallback = true; - } finally { - releaseSignalingStateCallback(dispatchCallback); - deletionThread.join(CALLBACK_TIMEOUT_MILLIS); + @Timeout(300) + void peerDeletionKeepsCallbackStateUntilScheduledCallbacksReturn() throws Exception { + String java = Paths.get(System.getProperty("java.home"), "bin", "java").toString(); + Process process = new ProcessBuilder(java, "-cp", System.getProperty("java.class.path"), Churn.class.getName()) + .redirectErrorStream(true) + .start(); + + String output; + try (InputStream stdout = process.getInputStream()) { + output = readFully(stdout); } + process.waitFor(); - assertFalse(deletionThread.isAlive(), "peer deletion did not finish after the callback returned"); - assertNull(deletionFailure.get()); - assertEquals(0, deletionResult.get()); - assertEquals(1, blockedPeer.callbackCount.get()); + assertEquals(0, process.exitValue(), "the churn process died, its output was:\n" + output); } - private static BlockedPeer createPeerWithBlockedCallback() { - PeerConnectionConfiguration configuration = PeerConnectionConfiguration.DEFAULT - .withDisableAutoNegotiation(true); - PeerConnection peer = PeerConnection.createPeer(configuration); - DataChannel channel = peer.createDataChannel("lifecycle-test"); - AtomicInteger callbackCount = new AtomicInteger(); - peer.onSignalingStateChange.register((ignoredPeer, ignoredState) -> callbackCount.incrementAndGet()); - - armSignalingStateCallback(); - peer.setLocalDescription("offer"); - boolean callbackEntered = awaitSignalingStateCallback(CALLBACK_TIMEOUT_MILLIS); - if (!callbackEntered) { - releaseSignalingStateCallback(false); + private static String readFully(InputStream stream) throws IOException { + byte[] buffer = new byte[8192]; + StringBuilder text = new StringBuilder(); + int read; + while ((read = stream.read(buffer)) != -1) { + text.append(new String(buffer, 0, read, StandardCharsets.UTF_8)); } - assertTrue(callbackEntered, "the native signaling callback did not start"); - - channel.close(); - return new BlockedPeer(peer.peerHandle, new WeakReference<>(peer.listener), callbackCount); + return text.toString(); } - private static void awaitDeletionBlock(Thread deletionThread) throws InterruptedException { - long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(1); - while (deletionThread.isAlive() && deletionThread.getState() == Thread.State.RUNNABLE - && System.nanoTime() < deadline) { - Thread.sleep(10); + /** + * Repeatedly parks a peer connection inside one of its callbacks, deletes the peer from another thread and then + * lets the callback return, so that the callbacks queued behind it are delivered while the deletion is waiting. + */ + public static final class Churn { + private static final int ITERATIONS = 100; + private static final long TIMEOUT_SECONDS = 10; + + public static void main(String[] args) throws Exception { + // holding on to the peers keeps their cleaners from deleting handles that have since been reused + List peers = new ArrayList<>(); + for (int i = 0; i < ITERATIONS; i++) { + peers.add(deleteWhileCallbackIsRunning()); + } + System.out.println("survived " + ITERATIONS + " deletions, kept " + peers.size() + " peers"); } - assertTrue(deletionThread.isAlive(), "peer deletion returned before the native callback was released"); - } - private static void forceGarbageCollection() throws InterruptedException { - for (int attempt = 0; attempt < 10; attempt++) { - System.gc(); - System.runFinalization(); - Thread.sleep(20); - } - } + private static PeerConnection deleteWhileCallbackIsRunning() throws Exception { + CountDownLatch callbackEntered = new CountDownLatch(1); + CountDownLatch releaseCallback = new CountDownLatch(1); + + PeerConnectionConfiguration config = PeerConnectionConfiguration.DEFAULT.withDisableAutoNegotiation(true); + PeerConnection peer = PeerConnection.createPeer(config); + peer.onSignalingStateChange.register((ignoredPeer, ignoredState) -> { + callbackEntered.countDown(); + await(releaseCallback); + }); + // these queue up behind the parked callback and are drained while the deletion waits + peer.onLocalDescription.register((ignoredPeer, ignoredSdp, ignoredType) -> {}); + peer.onLocalCandidate.register((ignoredPeer, ignoredCandidate, ignoredMediaId) -> {}); + peer.onGatheringStateChange.register((ignoredPeer, ignoredState) -> {}); + peer.onStateChange.register((ignoredPeer, ignoredState) -> {}); + peer.createDataChannel("callback-lifecycle"); + peer.setLocalDescription("offer"); + + if (!callbackEntered.await(TIMEOUT_SECONDS, TimeUnit.SECONDS)) { + throw new IllegalStateException("the signaling state callback never ran"); + } - private static final class BlockedPeer { - private final int peerHandle; - private final WeakReference listener; - private final AtomicInteger callbackCount; + Thread deletion = new Thread(() -> rtcDeletePeerConnection(peer.peerHandle), "peer-deletion"); + deletion.start(); + // let the deletion reach the point where it waits for the scheduled callbacks + Thread.sleep(50); + releaseCallback.countDown(); + deletion.join(TimeUnit.SECONDS.toMillis(TIMEOUT_SECONDS)); + if (deletion.isAlive()) { + throw new IllegalStateException("the peer deletion never returned"); + } + return peer; + } - private BlockedPeer(int peerHandle, WeakReference listener, - AtomicInteger callbackCount) { - this.peerHandle = peerHandle; - this.listener = listener; - this.callbackCount = callbackCount; + private static void await(CountDownLatch latch) { + try { + latch.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } } } }