From b4092d1bd632a88e068b8563645613b8ec6812f2 Mon Sep 17 00:00:00 2001 From: evgeny Date: Thu, 15 Oct 2026 02:47:33 +0200 Subject: [PATCH] Record bounded call audio traces and report event windows off playback thread --- .../com/utun/chat/data/CallAudioEngine.kt | 63 +++++++-- .../java/com/utun/chat/data/CallAudioTrace.kt | 131 ++++++++++++++++++ .../com/utun/chat/data/CallAudioTraceTest.kt | 64 +++++++++ 3 files changed, 246 insertions(+), 12 deletions(-) create mode 100644 tools/chatgui-android/app/src/main/java/com/utun/chat/data/CallAudioTrace.kt create mode 100644 tools/chatgui-android/app/src/test/java/com/utun/chat/data/CallAudioTraceTest.kt diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/CallAudioEngine.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/CallAudioEngine.kt index ebe3cae7..1c4f4d1c 100644 --- a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/CallAudioEngine.kt +++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/CallAudioEngine.kt @@ -426,6 +426,27 @@ class CallAudioEngine { } private fun playLoop() { + val trace = CallAudioTrace(callId) { level, message -> LogManager.addLog(level, "call", message) } + val reporter = Thread({ + try { + while (!Thread.currentThread().isInterrupted) { + trace.flush() + Thread.sleep(200) + } + } catch (_: InterruptedException) { + // Playback уже закончен; ниже сохраняем неполное окно события. + } catch (e: Exception) { + LogManager.addLog("ERROR", "CallAudio", "diagnostic reporter failed: ${e.message}") + } finally { + try { trace.flush(true) } + catch (e: Exception) { LogManager.addLog("ERROR", "CallAudio", "diagnostic flush failed: ${e.message}") } + } + }, "utun-call-diagnostics").apply { start() } + try { playLoop(trace) } + finally { reporter.interrupt(); reporter.join() } + } + + private fun playLoop(trace: CallAudioTrace) { val buf = ShortArray(FRAME_SAMPLES) val silence = ShortArray(FRAME_SAMPLES) val frameNs = FRAME_MS * 1_000_000L @@ -439,11 +460,17 @@ class CallAudioEngine { val epoch = snapshot[0] val head = CallPlaybackHead() var clockFailed = false - var logAtUs = 0L - var lastUnderruns = 0 + val diagnostic = LongArray(8) + val row = LongArray(CallAudioTrace.WIDTH) + val stamp = AudioTimestamp() + var stampAtUs = 0L + var stampOk = false while (active) { val track = audioTrack ?: return + val pullAtNs = System.nanoTime() val n = NativeLib.callAudioPull(callId, buf) + val pullUs = (System.nanoTime() - pullAtNs) / 1000 + trace.enabled = NativeLib.callAudioDiagnostics(callId, diagnostic) if (n !in 0..FRAME_SAMPLES) { reportFailure(callId, "callAudioPull=$n expected=0..$FRAME_SAMPLES") return @@ -452,6 +479,9 @@ class CallAudioEngine { if (n < FRAME_SAMPLES) silence.fill(0, n, FRAME_SAMPLES) var offset = 0 var blockedAt = 0L + var zeroWrites = 0L + var observedPlayed = -1L + val writeAtNs = System.nanoTime() while (active && offset < FRAME_SAMPLES) { val written = track.write(silence, offset, FRAME_SAMPLES - offset, AudioTrack.WRITE_NON_BLOCKING) if (written < 0) { @@ -459,6 +489,7 @@ class CallAudioEngine { return } if (written == 0) { + zeroWrites++ val now = SystemClock.elapsedRealtime() if (blockedAt == 0L) blockedAt = now if (now - blockedAt > 2000) { @@ -474,6 +505,7 @@ class CallAudioEngine { val nowUs = System.nanoTime() / 1000 val played = head.update(track.playbackHeadPosition, playFramesWritten) if (played != null) { + observedPlayed = played NativeLib.callAudioProgress(callId, epoch, played, nowUs) } else if (!clockFailed) { clockFailed = true @@ -482,17 +514,24 @@ class CallAudioEngine { LogManager.addLog("ERROR", "CallAudio", "playback clock invalid id=$callId epoch=$epoch written=$playFramesWritten; audio continues") } - if (nowUs - logAtUs >= 1_000_000 && NativeLib.callAudioClock(callId, nowUs, snapshot)) { - logAtUs = nowUs - val underruns = track.underrunCount - val threshold = if (Build.VERSION.SDK_INT >= 31) track.startThresholdInFrames else track.bufferSizeInFrames - val queuedUs = (snapshot[3] - snapshot[2]).coerceAtLeast(0) * 1_000_000 / SAMPLE_RATE - LogManager.addLog(if (underruns > lastUnderruns) "WARN" else "DEBUG", "CallAudio", - "clock epoch=${snapshot[0]} source=${snapshot[1]} played=${snapshot[2]} written=${snapshot[3]} " + - "ptsUs=${snapshot[4]} valid=${snapshot[5]} measured=${snapshot[6]} heldUs=${snapshot[7]} " + - "queuedUs=$queuedUs bufferFrames=${track.bufferSizeInFrames} startFrames=$threshold underruns=$underruns") - lastUnderruns = underruns + } + val writeUs = (System.nanoTime() - writeAtNs) / 1000 + if (trace.enabled && active) { + val nowUs = System.nanoTime() / 1000 + if (nowUs - stampAtUs >= 200_000) { + stampAtUs = nowUs + stampOk = track.getTimestamp(stamp) } + val route = track.routedDevice + row[0] = nowUs; row[1] = route?.id?.toLong() ?: -1; row[2] = route?.type?.toLong() ?: -1 + row[3] = playFramesWritten; row[4] = observedPlayed + row[5] = pullUs; row[6] = writeUs; row[7] = ((pullAtNs - nextNs) / 1000).coerceAtLeast(0) + row[8] = zeroWrites; row[9] = track.underrunCount.toLong(); row[10] = track.bufferSizeInFrames.toLong() + row[11] = if (Build.VERSION.SDK_INT >= 31) track.startThresholdInFrames.toLong() else row[10] + row[12] = if (stampOk) stamp.framePosition else -1; row[13] = if (stampOk) stamp.nanoTime else -1 + row[14] = stampAtUs; row[15] = if (stampOk) 1 else 0 + diagnostic.copyInto(row, 16); row[24] = offset.toLong() + trace.record(row) } if (firstWrite && active) { diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/CallAudioTrace.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/CallAudioTrace.kt new file mode 100644 index 00000000..49270f8a --- /dev/null +++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/CallAudioTrace.kt @@ -0,0 +1,131 @@ +package com.utun.chat.data + +/** Только числа в playback-потоке; форматирование и вывод выполняет отдельный reporter. */ +internal class CallAudioTrace(private val id: Long, private val emit: (String, String) -> Unit) { + companion object { + const val WIDTH = 25 + const val CAPACITY = 512 + // monoUs, routeId, routeType, written, played, pullUs, writeUs, wakeLateUs, zeroWrites, + // deviceUnderruns, bufferFrames, startFrames, stampFrame, stampNs, stampReadUs, stampOk, + // encodedMs, processingMs, reserveMs, spreadMs, pcmSamples, tempo100, jbUnderruns, waiting, accepted + private const val PRE_US = 2_000_000L + private const val POST_US = 3_000_000L + } + @Volatile var enabled = false + private val ring = Array(CAPACITY) { LongArray(WIDTH) } + private var head = 0 + private var count = 0 + private var lost = 0L + private val history = ArrayDeque() // Только reporter. + private val summary = ArrayList(64) + private var previous: LongArray? = null + private var summaryAt = 0L + private var eventAt = -1L + private var eventBits = 0 + private var eventCount = 0 + private var lastDump = -10_000_000L + private var suppressed = 0 + private var highSince = -1L + private var stoppedSince = -1L + private var underrunAt = -10_000_000L + private var faults = 0 + + @Synchronized fun record(row: LongArray) { + if (!enabled) return + if (count == CAPACITY) { head = (head + 1) % CAPACITY; count--; lost++ } + row.copyInto(ring[(head + count) % CAPACITY]) + count++ + } + + // Копирование ограничено 512 короткими строками; никаких логов под lock. + @Synchronized private fun take(): Pair, Long> { + val rows = List(count) { ring[(head + it) % CAPACITY].copyOf() } + val overflow = lost + head = 0; count = 0; lost = 0 + return rows to overflow + } + + /** Вызывается только reporter, обычно раз в 200мс; finished сохраняет неполное окно при stop. */ + fun flush(finished: Boolean = false) { + val (rows, overflow) = take() + if (!enabled) { + history.clear(); summary.clear(); previous = null; eventAt = -1 + faults = 0; highSince = -1; stoppedSince = -1; underrunAt = -10_000_000 + return + } + if (overflow > 0) emit("WARN", "audio trace call=$id reporter overrun lost=$overflow") + for (row in rows) { + val now = row[0] + val old = previous + // Сначала завершаем старое окно, чтобы оно не поглотило новое событие после задержки reporter. + if (eventAt >= 0 && now > eventAt + POST_US) dump(false) + history.addLast(row) + while (history.size > 400 || (eventAt < 0 && history.first()[0] < now - PRE_US)) history.removeFirst() + var events = if (old == null) 1 else 0 + if (old != null && (row[1] != old[1] || row[2] != old[2])) events = events or 2 + val queueUs = queued(row) + if (queueUs > 100_000) { if (highSince < 0) highSince = now } else highSince = -1 + if (old != null && row[4] == old[4] && queueUs > 0) { + if (stoppedSince < 0) stoppedSince = old[0] + } else stoppedSince = -1 + if (old != null && row[9] > old[9]) underrunAt = now + val current = (if (highSince >= 0 && now - highSince >= 1_000_000) 4 else 0) or + (if (stoppedSince >= 0 && now - stoppedSince >= 100_000) 8 else 0) or + (if (now - underrunAt < 3_000_000) 16 else 0) + val added = current and faults.inv() + if (added != 0) { + emit("WARN", "audio trace call=$id fault=$added queueUs=$queueUs route=${row[1]}/${row[2]} monoUs=$now") + events = events or added + } + if (faults != 0 && current == 0) emit("INFO", "audio trace call=$id recovered monoUs=$now") + faults = current + if (events != 0) { + if (eventAt >= 0) { eventBits = eventBits or events; eventCount++ } + else if (now - lastDump >= 10_000_000) { eventAt = now; eventBits = events; eventCount = 1 } + else suppressed++ + } + if (summary.isEmpty()) summaryAt = now + summary.add(row) + if (now - summaryAt >= 1_000_000) summarize() + previous = row + } + if (finished) { + if (summary.isNotEmpty()) summarize() + if (eventAt >= 0) dump(true) + } + } + + private fun queued(r: LongArray): Long = if (r[4] < 0) -1 else (r[3] - r[4]) * 1_000_000 / 48000 + + private fun range(value: (LongArray) -> Long): String { + val values = summary.map(value) + return "${values.minOrNull()}/${values.sum() / values.size}/${values.maxOrNull()}" + } + + private fun summarize() { + val first = summary.first(); val last = summary.last() + emit("DEBUG", "audio summary call=$id monoUs=${last[0]} spanUs=${last[0] - first[0]} n=${summary.size} " + + "route=${last[1]}/${last[2]} min/avg/max queueUs=${range(::queued)} " + + "jbMs=${range { it[16] + it[17] }} reserveMs=${range { it[18] }} spreadMs=${range { it[19] }} " + + "pullUs=${range { it[5] }} writeUs=${range { it[6] }} wakeLateUs=${range { it[7] }} " + + "tempo=${last[21]} deviceUnderruns=${last[9]} jbUnderruns=${last[22]} " + + "zeroWrites=${summary.sumOf { it[8] }} missingPcm=${summary.sumOf { (960 - it[20]).coerceAtLeast(0) }} " + + "stamp=${last[12]}@${last[13]} readUs=${last[14]} ok=${last[15]} " + + "buffer=${last[10]} start=${last[11]} fault=$faults suppressed=$suppressed") + summary.clear() + } + + private fun dump(partial: Boolean) { + emit("DEBUG", "audio detail call=$id triggerUs=$eventAt reasons=$eventBits events=$eventCount partial=$partial " + + "bits=start:1,route:2,queue:4,stalled:8,underrun:16 " + + "columns=monoUs,routeId,routeType,written,played,pullUs,writeUs,wakeLateUs,zeroWrites,deviceUnderruns," + + "bufferFrames,startFrames,stampFrame,stampNs,stampReadUs,stampOk," + + "encodedMs,processingMs,reserveMs,spreadMs,pcmSamples,tempo100,jbUnderruns,waiting,accepted") + // До пяти секунд, по четыре снимка в строке, с явным временем каждого. + history.filter { it[0] in (eventAt - PRE_US)..(eventAt + POST_US) }.chunked(4).forEach { group -> + emit("DEBUG", "audio detail call=$id rows=" + group.joinToString(";") { it.joinToString(",") }) + } + lastDump = history.lastOrNull()?.get(0) ?: eventAt + eventAt = -1 + } +} diff --git a/tools/chatgui-android/app/src/test/java/com/utun/chat/data/CallAudioTraceTest.kt b/tools/chatgui-android/app/src/test/java/com/utun/chat/data/CallAudioTraceTest.kt new file mode 100644 index 00000000..d207817c --- /dev/null +++ b/tools/chatgui-android/app/src/test/java/com/utun/chat/data/CallAudioTraceTest.kt @@ -0,0 +1,64 @@ +package com.utun.chat.data + +import org.junit.Assert.* +import org.junit.Test + +class CallAudioTraceTest { + private fun sample(t: Long, queueMs: Long = 40, route: Long = 1): LongArray = LongArray(CallAudioTrace.WIDTH).also { + it[0] = t; it[1] = route; it[2] = 2; it[3] = t * 48 / 1000 + queueMs * 48; it[4] = t * 48 / 1000 + it[18] = 60; it[20] = 960; it[21] = 100 + } + + @Test fun fixedRingCopiesSamplesAndReportsOverflowWithoutLoggingFromProducer() { + val logs = mutableListOf() + val trace = CallAudioTrace(1) { _, text -> logs.add(text) }.apply { enabled = true } + val row = sample(0) + repeat(600) { row[0] = it * 20_000L; trace.record(row) } + assertTrue(logs.isEmpty()) + row[0] = -1 + trace.flush(true) + assertTrue(logs.any { it.contains("lost=88") }) + assertTrue(logs.any { it.contains("triggerUs=1760000") }) + } + + @Test fun sustainedHighQueueProducesOneEventAndRecoveryRearmsIt() { + val logs = mutableListOf() + val trace = CallAudioTrace(2) { _, text -> logs.add(text) }.apply { enabled = true } + repeat(1500) { + trace.record(sample(it * 20_000L, if (it in 700..899) 40 else 140)) + if (it % 10 == 0) trace.flush() + } + trace.flush(true) + assertEquals(2, logs.count { it.contains(" fault=4 queueUs=") }) + assertEquals(1, logs.count { it.contains(" recovered ") }) + assertEquals(2, logs.count { it.contains("triggerUs=") }) + assertTrue(logs.count { it.contains("audio summary") } in 28..31) + } + + @Test fun routeEventIncludesHistoryAndPostWindowAndDoesNotRetriggerEveryFrame() { + val logs = mutableListOf() + val trace = CallAudioTrace(3) { _, text -> logs.add(text) }.apply { enabled = true } + repeat(1000) { + trace.record(sample(it * 20_000L, route = if (it < 750) 1 else 2)) + if (it % 10 == 0) trace.flush() + } + trace.flush(true) + assertEquals(2, logs.count { it.contains("triggerUs=") }) + assertTrue(logs.any { it.contains("triggerUs=15000000 reasons=2") }) + assertTrue(logs.any { it.contains("rows=13000000,") }) + assertTrue(logs.any { it.contains("18000000,2,2,") }) + } + + @Test fun disabledLoggingAndShortCallDoNotLoseControlOfBuffer() { + val logs = mutableListOf() + val trace = CallAudioTrace(4) { _, text -> logs.add(text) } + trace.record(sample(0)); trace.flush(true) + assertTrue(logs.isEmpty()) + trace.enabled = true + trace.record(sample(20_000)); trace.flush(true) + assertTrue(logs.any { it.contains("partial=true") }) + logs.clear(); trace.enabled = false + trace.record(sample(40_000)); trace.flush(true) + assertTrue(logs.isEmpty()) + } +}