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/build.sh b/jni/build.sh index 01e0761..58a93e1 100755 --- a/jni/build.sh +++ b/jni/build.sh @@ -46,4 +46,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/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 +} 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..d603e1e --- /dev/null +++ b/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"); + } + } +} 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..f678b76 --- /dev/null +++ b/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. + * + *

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