diff --git a/app/build.gradle.kts b/app/build.gradle.kts index ceb80d9..e96ef74 100644 --- a/app/build.gradle.kts +++ b/app/build.gradle.kts @@ -37,8 +37,8 @@ android { minSdk = 33 targetSdk = 35 // Remain newer than the 0.6.3 rendezvous candidate (code 26). - versionCode = 32 - versionName = "0.6.9" + versionCode = 33 + versionName = "0.6.10" testInstrumentationRunner = "androidx.test.runner.AndroidJUnitRunner" diff --git a/app/src/main/kotlin/dev/forgesworn/kithmoot/media/LinkTable.kt b/app/src/main/kotlin/dev/forgesworn/kithmoot/media/LinkTable.kt new file mode 100644 index 0000000..cb18c9c --- /dev/null +++ b/app/src/main/kotlin/dev/forgesworn/kithmoot/media/LinkTable.kt @@ -0,0 +1,58 @@ +package dev.forgesworn.kithmoot.media + +import java.util.concurrent.ConcurrentHashMap + +/** + * The engine's table of one link per remote device, readable from any thread. + * + * The engine mutates it under its own lock, so two reconciles cannot open one + * device twice. What this class exists for is the other direction: a libwebrtc + * callback, which arrives on the signalling thread, may ask "is this link still + * the device's current one" WITHOUT that lock. + * + * Closing a peer connection waits for the signalling thread. If the signalling + * thread is waiting for the engine lock, and the engine is holding that lock + * while it closes, neither can move, and five seconds later Android calls the + * app not responding. That was every freeze on 0.6.9: three traces, each with + * the main thread inside `PeerConnection.close` under the lock and the + * signalling thread "waiting to lock, held by thread 1". So identity reads here + * take no lock, and [reconcile] hands back what it removed so the engine can + * close it after the lock has gone. + */ +internal class LinkTable { + private val links = ConcurrentHashMap() + + operator fun get(device: String): L? = links[device] + + /** Read from any thread, never under the engine lock. */ + fun isCurrent(device: String, link: L): Boolean = links[device] === link + + val devices: Set get() = links.keys.toSet() + + fun values(): List = links.values.toList() + + fun snapshot(): List> = links.entries.map { it.key to it.value } + + fun put(device: String, link: L) { + links[device] = link + } + + fun remove(device: String): L? = links.remove(device) + + /** Empties the table and hands back what was in it, for closing outside the lock. */ + fun clear(): List { + val all = values() + links.clear() + return all + } + + /** + * Applies a roster: opens, under [lock], every device not yet linked, and + * removes every link whose device has gone. The removed links are handed + * back, not closed: the caller closes them once it has released [lock]. + */ + fun reconcile(lock: Any, devices: Set, open: (String) -> L): List> = synchronized(lock) { + for (device in devices - links.keys) links[device] = open(device) + (links.keys - devices).mapNotNull { device -> links.remove(device)?.let { device to it } } + } +} diff --git a/app/src/main/kotlin/dev/forgesworn/kithmoot/media/WebRtcEngine.kt b/app/src/main/kotlin/dev/forgesworn/kithmoot/media/WebRtcEngine.kt index 6a2f285..d6afd96 100644 --- a/app/src/main/kotlin/dev/forgesworn/kithmoot/media/WebRtcEngine.kt +++ b/app/src/main/kotlin/dev/forgesworn/kithmoot/media/WebRtcEngine.kt @@ -105,9 +105,14 @@ class WebRtcEngine( * caller cancels this scope moments after asking for it. Anything launched * here would be cancelled before it ran, leaving every peer connection * alive behind a factory that had already gone. + * + * Two rules keep it from deadlocking against libwebrtc, whose callbacks + * arrive on its signalling thread and whose `close` waits for that thread: + * nothing that runs on a WebRTC callback takes this lock (see + * [LinkTable]), and a link is closed only after this lock is released. */ private val lock = Any() - private val links = mutableMapOf() + private val links = LinkTable() /** * Which remote devices this device's media may be sent to. @@ -156,15 +161,21 @@ class WebRtcEngine( fun setCallActive(active: Boolean) { synchronized(lock) { callActive = active } - if (active) reconcile(session.remoteDevices.value) else stop() + if (active) scope.launch(MediaDispatcher) { reconcile(session.remoteDevices.value) } else stop() } + /** + * Everything here runs on [MediaDispatcher], never on the caller's thread. + * Opening and closing a peer connection blocks until libwebrtc's signalling + * thread has done it, and a roster change used to do that on the main + * thread, which is where the input timeout is measured. + */ fun start() { // The set of devices to connect to is derived from the roster, so a // device that joins, leaves or lapses is reconciled here rather than // being handled as an event. - scope.launch { session.remoteDevices.collect { reconcile(it) } } - scope.launch { + scope.launch(MediaDispatcher) { session.remoteDevices.collect { reconcile(it) } } + scope.launch(MediaDispatcher) { session.signals.collect { signal -> val target = managedLinkFor(signal.from) ?: return@collect try { @@ -173,22 +184,20 @@ class WebRtcEngine( catch (failure: Exception) { // A signal can finish after its link was deliberately // replaced. That old link's exception must not relabel the - // replacement as failed. Check identity and publish the - // state while holding the same lock used for replacement. - synchronized(lock) { - if (callActive && links[signal.from] === target) { - Log.w("KithMootMedia", "signal failed peer=${signal.from.take(8)}", failure) - _connections.update { it + (signal.from to "failed") } - } + // replacement as failed: publish only while this is still + // the device's current link. + if (callActive && links.isCurrent(signal.from, target)) { + Log.w("KithMootMedia", "signal failed peer=${signal.from.take(8)}", failure) + _connections.update { it + (signal.from to "failed") } } } } } - scope.launch { localMedia.tracks.collect { onLocalTracksChanged(it) } } - scope.launch { + scope.launch(MediaDispatcher) { localMedia.tracks.collect { onLocalTracksChanged(it) } } + scope.launch(MediaDispatcher) { while (isActive) { delay(10_000) - synchronized(lock) { links.values.forEach { it.reportMediaProgress() } } + synchronized(lock) { links.values().forEach { it.reportMediaProgress() } } } } // One statistics sample per profile-2 pair every two seconds, which is @@ -196,10 +205,10 @@ class WebRtcEngine( // the only honest evidence that a direction is alive: a connection can // be `connected`, the signalling quiet, every object healthy, and one // direction carrying nothing at all. - scope.launch { + scope.launch(MediaDispatcher) { while (isActive) { delay(HEALTH_SAMPLE_MS) - val sampling = synchronized(lock) { links.values.filter { it.profileTwo } } + val sampling = synchronized(lock) { links.values().filter { it.profileTwo } } for (link in sampling) runCatching { link.sampleHealth() } } } @@ -230,27 +239,30 @@ class WebRtcEngine( } private fun closeLinks() { - val closing = synchronized(lock) { - val all = links.values.toList() - links.clear() - all - } + val closing = synchronized(lock) { links.clear() } for (link in closing) runCatching { link.close() } _remoteTracks.value = emptyList() _connections.value = emptyMap() } - private fun reconcile(devices: Set) = synchronized(lock) { - if (!callActive) return@synchronized - for (device in devices - links.keys) links[device] = openLink(device) - for (device in links.keys - devices) { - links.remove(device)?.close() + /** + * Opened under the lock, so two reconciles cannot open one device twice; + * closed after it, because a close waits for the signalling thread, and + * the signalling thread must never find this lock held while it waits. + */ + private fun reconcile(devices: Set) { + val closing = synchronized(lock) { + if (!callActive) return + links.reconcile(lock, devices) { openLink(it) } + } + for ((device, link) in closing) { + runCatching { link.close() } _connections.update { it - device } _remoteTracks.update { current -> current.filterNot { it.device == device } } } } - private fun managedLinkFor(device: String): ManagedLink? = synchronized(lock) { links[device] } + private fun managedLinkFor(device: String): ManagedLink? = links[device] /** * Throw a pair's connection away and open another. @@ -262,11 +274,16 @@ class WebRtcEngine( * rather than opening a generation of its own and glaring with it. */ private fun rebuildLink(device: String, gen: Long, open: Boolean) { - synchronized(lock) { - if (!callActive || device !in links) return - links.remove(device)?.let { runCatching { it.close() } } - links[device] = openLink(device, gen, open) + val old = synchronized(lock) { + val current = links[device] + if (!callActive || current == null) return + links.remove(device) + links.put(device, openLink(device, gen, open)) + current } + // Outside the lock: see `reconcile`. The old link's callbacks are + // already ignored, because it is no longer the device's current link. + runCatching { old.close() } _remoteTracks.update { current -> current.filterNot { it.device == device } } } @@ -279,11 +296,14 @@ class WebRtcEngine( * it was addressed to a session the far end no longer has. */ private fun reopenAsProfileOne(device: String) { - synchronized(lock) { - if (!callActive || device !in links) return - links.remove(device)?.let { runCatching { it.close() } } - links[device] = openLink(device, profileTwo = false) + val old = synchronized(lock) { + val current = links[device] + if (!callActive || current == null) return + links.remove(device) + links.put(device, openLink(device, profileTwo = false)) + current } + runCatching { old.close() } _remoteTracks.update { current -> current.filterNot { it.device == device } } } @@ -359,7 +379,7 @@ class WebRtcEngine( audience = rule val tracks = localMedia.tracks.value synchronized(lock) { - for ((device, link) in links) link.syncLocalTracks(tracksFor(device, tracks)) + for ((device, link) in links.snapshot()) link.syncLocalTracks(tracksFor(device, tracks)) } } @@ -408,7 +428,7 @@ class WebRtcEngine( // byte-identical for everyone who never mutes. session.setTracks(tracks.map { TrackRef(it.trackId, it.role, if (it.muted) true else null) }) synchronized(lock) { - for ((device, link) in links) link.syncLocalTracks(tracksFor(device, tracks)) + for ((device, link) in links.snapshot()) link.syncLocalTracks(tracksFor(device, tracks)) } } @@ -565,15 +585,19 @@ class WebRtcEngine( } } - /** Publish state only while this is still the device's current link. + /** + * Publish state only while this is still the device's current link. * Native WebRTC callbacks can arrive after close/rebuild; without the * identity check an old callback can overwrite the new link's state. + * + * Never under the engine lock. This runs on libwebrtc's signalling + * thread, and closing a connection waits for that thread: taking the + * lock here while the engine held it through a close was the deadlock + * behind every freeze on 0.6.9. */ private fun updateConnectionState(state: String) { - synchronized(lock) { - if (!closed && callActive && links[device] === this) { - _connections.update { it + (device to state) } - } + if (!closed && callActive && links.isCurrent(device, this)) { + _connections.update { it + (device to state) } } } diff --git a/app/src/test/kotlin/dev/forgesworn/kithmoot/media/LinkTableTest.kt b/app/src/test/kotlin/dev/forgesworn/kithmoot/media/LinkTableTest.kt new file mode 100644 index 0000000..b207626 --- /dev/null +++ b/app/src/test/kotlin/dev/forgesworn/kithmoot/media/LinkTableTest.kt @@ -0,0 +1,80 @@ +package dev.forgesworn.kithmoot.media + +import java.util.concurrent.CountDownLatch +import java.util.concurrent.TimeUnit +import kotlin.concurrent.thread +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFalse +import kotlin.test.assertNotNull +import kotlin.test.assertNull +import kotlin.test.assertTrue + +/** + * The lock discipline that keeps the engine and libwebrtc's signalling thread + * from waiting on each other. Three freezes on 0.6.9 had the same shape: the + * main thread inside a peer-connection close under the engine lock, and the + * signalling thread waiting for that lock to publish a state. + */ +class LinkTableTest { + private class Link(val device: String) + + @Test fun `opens under the lock and hands back what is gone for closing outside it`() { + val lock = Any() + val table = LinkTable() + var openedUnderLock = true + table.reconcile(lock, setOf("a", "b")) { device -> + if (!Thread.holdsLock(lock)) openedUnderLock = false + Link(device) + } + assertTrue(openedUnderLock, "a link is opened under the lock, so a device is never opened twice") + assertEquals(setOf("a", "b"), table.devices) + val b = assertNotNull(table["b"]) + + val gone = table.reconcile(lock, setOf("b")) { Link(it) } + assertFalse(Thread.holdsLock(lock), "the lock is released before the caller closes anything") + assertEquals(listOf("a"), gone.map { it.first }) + assertNull(table["a"]) + assertTrue(table.isCurrent("b", b)) + assertFalse(table.isCurrent("a", gone.single().second)) + } + + @Test fun `a state read never waits for the engine lock`() { + val lock = Any() + val table = LinkTable() + val a = Link("a") + table.put("a", a) + val read = CountDownLatch(1) + var current = false + var completed = false + val engine = thread { + synchronized(lock) { + // The worst case, and the shape of every 0.6.9 freeze: the + // engine holds its lock and waits for the signalling thread. + val signalling = thread { + current = table.isCurrent("a", a) + read.countDown() + } + completed = read.await(5, TimeUnit.SECONDS) + signalling.join() + } + } + engine.join() + assertTrue(completed, "the callback waited for the engine lock") + assertTrue(current) + } + + @Test fun `clear hands everything back and leaves nothing current`() { + val lock = Any() + val table = LinkTable() + val a = Link("a") + val b = Link("b") + table.put("a", a) + table.put("b", b) + val all = synchronized(lock) { table.clear() } + assertEquals(setOf(a, b), all.toSet()) + assertTrue(table.devices.isEmpty()) + assertFalse(table.isCurrent("a", a)) + assertEquals(emptyList(), table.snapshot()) + } +} diff --git a/docs/android-release.md b/docs/android-release.md index 35b58d0..9d932f7 100644 --- a/docs/android-release.md +++ b/docs/android-release.md @@ -2,6 +2,25 @@ The public website currently offers production-signed 0.6.7 (30), Android 13 or later. Signing, publication and physical acceptance are recorded separately. +## 0.6.10 the freeze on calls + +Version code 33 fixes the "KithMoot isn't responding" that 0.6.9 produced three +times in two days on the owner's Pixel 10 Pro XL, always during or around a +call. All three system traces show the same deadlock: a roster change ran the +engine's reconcile on the main thread, which closed the departed device's peer +connection while holding the engine lock; that close waits for libwebrtc's +signalling thread, and the signalling thread was delivering an ICE state +callback that took the same lock. Now nothing that runs on a WebRTC callback +takes the engine lock, a link is closed only after the lock is released, and +the engine's collectors run off the main thread. `LinkTableTest` pins the two +rules. It also carries the epoch request admission proof (PR #70), which the +web keeper deployed on 22 September now requires: without this build an +Android member who misses a removal in a keeper room cannot catch up. + +Not changed here, and worth knowing while judging a hot phone: the camera is +captured at 1280 by 720 at 30 frames a second and every remote peer gets its own +encoder with no bitrate, resolution or framerate cap. + ## 0.6.8 sharing and listening candidate Version code 31 adds app playback capture during screen sharing, preserves an