Recover radio worker failures on the owning I/O executor
This commit is contained in:
evgeny committed
1 parent
a04cfe4441
commit
ed8d208454
1 file changed
+65
-27
@@ -724,17 +724,20 @@ class RadioAudioEngine {
|
||||
var statsFrames = 0
|
||||
var fedFrames = 0
|
||||
var skippedFrames = 0
|
||||
var failedFrames = 0
|
||||
var minRmsDb = 0
|
||||
var maxRmsDb = -120
|
||||
var maxReadNs = 0L
|
||||
var maxFeedNs = 0L
|
||||
var previousFeedAllowed: Boolean? = null
|
||||
while (captureRunning) {
|
||||
var operation = "poll"
|
||||
try {
|
||||
while (captureRunning && active && groupId == group) {
|
||||
operation = "poll"
|
||||
val readStartedNs = SystemClock.elapsedRealtimeNanos()
|
||||
val event = subscription.poll() ?: continue
|
||||
if (!captureRunning) break
|
||||
if (!captureRunning || !active || groupId != group) break
|
||||
if (event is MicrophoneCapture.Ptt) {
|
||||
operation = if (event.pressed) "control begin" else "control end"
|
||||
if (event.pressed) NativeLib.radioTalkBegin(group) else NativeLib.radioTalkEnd(group)
|
||||
talking = event.pressed
|
||||
continue
|
||||
@@ -746,7 +749,12 @@ class RadioAudioEngine {
|
||||
discontinuity = true
|
||||
continue
|
||||
}
|
||||
operation = "frame"
|
||||
check(captured.pcm.size == FRAME_SAMPLES * ch) {
|
||||
"invalid capture frame samples=${captured.pcm.size} expected=${FRAME_SAMPLES * ch} sequence=${captured.sequence}"
|
||||
}
|
||||
if (discontinuity || (expected >= 0 && (captured.sequence != expected || captured.generation != captureEpoch))) {
|
||||
operation = "discontinuity"
|
||||
LogManager.addLog("INFO", "radio", "shared capture boundary group=$group epoch=$captureEpoch->${captured.generation}")
|
||||
NativeLib.radioCaptureDiscontinuity(group)
|
||||
}
|
||||
@@ -769,20 +777,20 @@ class RadioAudioEngine {
|
||||
previousFeedAllowed = feedAllowed
|
||||
}
|
||||
if (feedAllowed) {
|
||||
operation = "feed"
|
||||
val requestNs = talkRequestNs
|
||||
val feedStartedNs = SystemClock.elapsedRealtimeNanos()
|
||||
val fed = NativeLib.radioAudioFeed(group, buf)
|
||||
check(NativeLib.radioAudioFeed(group, buf)) { "native PCM feed rejected" }
|
||||
maxFeedNs = maxOf(maxFeedNs, SystemClock.elapsedRealtimeNanos() - feedStartedNs)
|
||||
if (fed) fedFrames++ else {
|
||||
failedFrames++
|
||||
if (failedFrames == 1) LogManager.addLog("WARN", "radio", "Auto/PTT PCM rejected group=$group frame=$frameN")
|
||||
}
|
||||
if (fed && requestedTalk && requestNs != reportedRequestNs) {
|
||||
fedFrames++
|
||||
if (requestedTalk && requestNs != reportedRequestNs) {
|
||||
reportedRequestNs = requestNs
|
||||
LogManager.addLog("INFO", "RadioAudio", "PTT first frame queued elapsedMs=${(SystemClock.elapsedRealtimeNanos() - requestNs) / 1_000_000}")
|
||||
}
|
||||
operation = "TX status"
|
||||
if (captureRunning) _talkingFlow.value = requestedTalk || NativeLib.radioTransmitting()
|
||||
} else skippedFrames++
|
||||
operation = "TX status"
|
||||
val transmitting = if (captureRunning && active && groupId == group && NativeLib.radioTransmitting()) group else null
|
||||
if (_transmittingGroup.value != transmitting) {
|
||||
_transmittingGroup.value = transmitting
|
||||
@@ -791,19 +799,28 @@ class RadioAudioEngine {
|
||||
val nowMs = SystemClock.elapsedRealtime()
|
||||
if (nowMs - statsMs >= 10_000L) {
|
||||
LogManager.addLog("DEBUG", "radio", "Auto capture summary group=$group spanMs=${nowMs - statsMs} " +
|
||||
"frames=$statsFrames audioMs=${statsFrames * FRAME_MS} fed=$fedFrames skipped=$skippedFrames rejected=$failedFrames " +
|
||||
"frames=$statsFrames audioMs=${statsFrames * FRAME_MS} fed=$fedFrames skipped=$skippedFrames " +
|
||||
"rmsDb=$minRmsDb..$maxRmsDb readMaxMs=${maxReadNs / 1_000_000.0} feedMaxMs=${maxFeedNs / 1_000_000.0} " +
|
||||
"channels=$ch input=${MediaAudioController.state.value.inputName} " +
|
||||
"route=${route()} mode=$autoMode screen=$screenInteractive manual=$requestedTalk tx=${_talkingFlow.value}")
|
||||
statsMs = nowMs
|
||||
statsFrames = 0; fedFrames = 0; skippedFrames = 0; failedFrames = 0
|
||||
statsFrames = 0; fedFrames = 0; skippedFrames = 0
|
||||
minRmsDb = 0; maxRmsDb = -120; maxReadNs = 0L; maxFeedNs = 0L
|
||||
}
|
||||
}
|
||||
LogManager.addLog("DEBUG", "RadioAudio", "capture loop done")
|
||||
} catch (e: Exception) {
|
||||
if (e is InterruptedException && !captureRunning) {
|
||||
LogManager.addLog("DEBUG", "RadioAudio", "capture cancelled operation=$operation group=$group")
|
||||
} else {
|
||||
LogManager.addLog("ERROR", "RadioAudio", "capture failed operation=$operation group=$group channels=$ch " +
|
||||
"frame=$frameN running=$captureRunning currentGroup=$groupId: $e cause=${e.cause}")
|
||||
}
|
||||
} finally {
|
||||
LogManager.addLog("DEBUG", "RadioAudio", "capture loop done group=$group frames=$frameN")
|
||||
}
|
||||
}
|
||||
|
||||
/** RMS захваченного кадра в dBFS (0 = полная шкала; тишина ~ -90). */
|
||||
/** RMS непустого захваченного кадра в dBFS (0 = полная шкала; тишина = -120). */
|
||||
private fun rmsDb(buf: ShortArray): Int {
|
||||
var sum = 0L
|
||||
for (s in buf) sum += s.toLong() * s.toLong()
|
||||
@@ -811,32 +828,34 @@ class RadioAudioEngine {
|
||||
return if (rms < 1.0) -120 else (20.0 * Math.log10(rms / 32768.0)).toInt()
|
||||
}
|
||||
|
||||
private fun playLoop() {
|
||||
/** Использовать только переданный track; ошибку своей сессии очистить на io до уведомления main. */
|
||||
private fun playLoop(track: AudioTrack, group: Long, version: Long) {
|
||||
val frame = FRAME_SAMPLES * PLAY_CHANNELS
|
||||
val buf = ShortArray(frame)
|
||||
val frameNs = FRAME_MS * 1_000_000L /* 20мс в наносекундах */
|
||||
var nextNs = System.nanoTime()
|
||||
var previousHead = 0L
|
||||
var playedFrames = 0L
|
||||
val group = audioGroup
|
||||
val track = audioTrack ?: return
|
||||
var operation = "head"
|
||||
var written = 0
|
||||
LogManager.addLog("DEBUG", "radio", "playback progress tracking group=$group guard=200ms")
|
||||
try {
|
||||
while (playbackRunning) {
|
||||
operation = "head"
|
||||
/* AudioTrack возвращает unsigned 32-bit позицию; разворачиваем wrap (~25 часов @48k). */
|
||||
val head = track.playbackHeadPosition.toLong() and 0xffffffffL
|
||||
playedFrames += (head - previousHead) and 0xffffffffL
|
||||
previousHead = head
|
||||
operation = "pull"
|
||||
val n = NativeLib.radioAudioPull(group, buf, playedFrames)
|
||||
if (n < frame) buf.fill(0, n.coerceAtLeast(0), frame)
|
||||
var written = 0
|
||||
check(n in 0..frame && n % PLAY_CHANNELS == 0) { "invalid native playback count=$n capacity=$frame" }
|
||||
if (n < frame) buf.fill(0, n, frame)
|
||||
written = 0
|
||||
operation = "write"
|
||||
while (written < frame && playbackRunning) {
|
||||
val count = track.write(buf, written, frame - written)
|
||||
if (count <= 0) {
|
||||
if (playbackRunning) {
|
||||
LogManager.addLog("ERROR", "radio", "playback write failed group=$group rc=$count written=$written/$frame")
|
||||
mainHandler.post { if (audioTrack === track) stop() }
|
||||
}
|
||||
return
|
||||
check(count > 0 && count <= frame - written && count % PLAY_CHANNELS == 0) {
|
||||
"invalid playback write rc=$count written=$written/$frame channels=$PLAY_CHANNELS"
|
||||
}
|
||||
written += count
|
||||
}
|
||||
@@ -845,13 +864,32 @@ class RadioAudioEngine {
|
||||
nextNs += frameNs
|
||||
val waitNs = nextNs - System.nanoTime()
|
||||
if (waitNs > 0) {
|
||||
try { Thread.sleep(waitNs / 1_000_000L, (waitNs % 1_000_000L).toInt()) }
|
||||
catch (_: InterruptedException) { return }
|
||||
operation = "sleep"
|
||||
Thread.sleep(waitNs / 1_000_000L, (waitNs % 1_000_000L).toInt())
|
||||
} else {
|
||||
nextNs = System.nanoTime() /* отстали — сбрасываем, не копим долг */
|
||||
}
|
||||
}
|
||||
LogManager.addLog("DEBUG", "RadioAudio", "play loop done")
|
||||
} catch (e: Exception) {
|
||||
if (e is InterruptedException && !playbackRunning) {
|
||||
LogManager.addLog("DEBUG", "RadioAudio", "playback cancelled operation=$operation group=$group generation=$version")
|
||||
} else {
|
||||
LogManager.addLog("ERROR", "RadioAudio", "playback failed operation=$operation group=$group generation=$version " +
|
||||
"written=$written/$frame running=$playbackRunning: $e cause=${e.cause}")
|
||||
io.execute {
|
||||
if (audioTrack !== track) {
|
||||
LogManager.addLog("DEBUG", "RadioAudio", "obsolete playback failure ignored group=$group generation=$version")
|
||||
return@execute
|
||||
}
|
||||
try { releaseLocked() } catch (cleanup: Exception) {
|
||||
LogManager.addLog("ERROR", "RadioAudio", "playback failure cleanup failed group=$group; retained for retry: $cleanup")
|
||||
}
|
||||
postReady(group, version, false, null)
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
LogManager.addLog("DEBUG", "RadioAudio", "play loop done group=$group generation=$version")
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
Reference in new issue
Block a user