evgeny
4 days ago
3 changed files with 246 additions and 12 deletions
@ -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<LongArray>() // Только reporter. |
||||
private val summary = ArrayList<LongArray>(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<List<LongArray>, 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 |
||||
} |
||||
} |
||||
@ -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<String>() |
||||
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<String>() |
||||
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<String>() |
||||
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<String>() |
||||
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()) |
||||
} |
||||
} |
||||
Loading…
Reference in new issue