fix(jni): detach native threads and preserve callbacks

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)
 }
 

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]

jni/build.sh

@@ -46,4 +46,4 @@ then
 fi
 
 cmake "$RELATIVE_PROJECT_PATH" "${cmake_options[@]}"
-make -j"${JOBS:-1}"
+make -j"${JOBS:-1}"

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;
-}
+}

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;
-}
+}

src/test/java/tel/schich/libdatachannel/NativeThreadLifecycleTest.java

@@ -0,0 +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.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 {
+    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
+    @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;
+    }
+
+    private static void churnPeer() {
+        PeerConnectionConfiguration config = PeerConnectionConfiguration.DEFAULT.withDisableAutoNegotiation(true);
+        try (PeerConnection peer = PeerConnection.createPeer(config)) {
+            peer.createDataChannel("thread-lifecycle");
+            peer.setLocalDescription("offer");
+        }
+    }
+}

src/test/java/tel/schich/libdatachannel/PeerCallbackLifecycleTest.java

@@ -0,0 +1,113 @@
+package tel.schich.libdatachannel;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+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 static org.junit.jupiter.api.Assertions.assertEquals;
+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.
+ *
+ * <p>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 {
+    @Test
+    @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();
+
+        assertEquals(0, process.exitValue(), "the churn process died, its output was:\n" + output);
+    }
+
+    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));
+        }
+        return text.toString();
+    }
+
+    /**
+     * 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<PeerConnection> peers = new ArrayList<>();
+            for (int i = 0; i < ITERATIONS; i++) {
+                peers.add(deleteWhileCallbackIsRunning());
+            }
+            System.out.println("survived " + ITERATIONS + " deletions, kept " + peers.size() + " peers");
+        }
+
+        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");
+            }
+
+            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 static void await(CountDownLatch latch) {
+            try {
+                latch.await();
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+            }
+        }
+    }
+}