perf(cw): pool detection samples in a ring buffer

The detection pool shifted its whole array down one slot per incoming sample
once full. Detection is throttled to 2 s but the pool fills in 400 ms, so for
the remaining 1.6 s of every cycle each chunk arrived at a full buffer: 320
copies of 1280 floats per chunk, measured at 24320 whole-array moves per 10 s
of audio, all on the capture thread.

Writing to a ring index is O(1) per sample. Draining walks the ring from the
oldest slot so the analyser still receives the most recent audio in
chronological order - a test feeds a ramp past capacity and asserts the exact
contents, since getting the wrap wrong would splice the waveform and corrupt
every estimate silently.
This commit is contained in:
mckero committed 2026-08-22 01:37:16 +00:00
1 parent 1b8f8c46f6
commit f6db55b35c
2 files changed
+95 -11

No files matched your search

@@ -145,6 +145,7 @@ class CwDeepDecoder(
*/ */
private val detectBuffer = FloatArray(DETECT_MIN_SAMPLES) private val detectBuffer = FloatArray(DETECT_MIN_SAMPLES)
private var detectFill = 0 private var detectFill = 0
private var detectWriteIndex = 0
/** Carries Hilbert filter history and mixer phase across capture chunks. */ /** Carries Hilbert filter history and mixer phase across capture chunks. */
private val streamingShifter = CwToneShifter.Streaming() private val streamingShifter = CwToneShifter.Streaming()
@@ -328,6 +329,7 @@ class CwDeepDecoder(
detectedToneHz = null detectedToneHz = null
lastDetectAtMs = 0L lastDetectAtMs = 0L
detectFill = 0 detectFill = 0
detectWriteIndex = 0
streamingShifter.reset() streamingShifter.reset()
} }
@@ -339,9 +341,7 @@ class CwDeepDecoder(
val elapsed = now - lastDetectAtMs val elapsed = now - lastDetectAtMs
if (detectFill >= DETECT_MIN_SAMPLES && elapsed >= DETECT_INTERVAL_MS) { if (detectFill >= DETECT_MIN_SAMPLES && elapsed >= DETECT_INTERVAL_MS) {
lastDetectAtMs = now lastDetectAtMs = now
val sample = detectBuffer.copyOf(detectFill) runDetection(drainDetectionBuffer())
detectFill = 0
runDetection(sample)
} }
// Streaming keeps the Hilbert filter history and mixer phase across chunks; // Streaming keeps the Hilbert filter history and mixer phase across chunks;
@@ -349,21 +349,39 @@ class CwDeepDecoder(
return streamingShifter.process(resampled, activeShiftHz, CwDeepSpectrogram.SAMPLE_RATE) return streamingShifter.process(resampled, activeShiftHz, CwDeepSpectrogram.SAMPLE_RATE)
} }
/** Collect resampled chunks until [DETECT_MIN_SAMPLES] is available. */ /**
* Collect resampled chunks until [DETECT_MIN_SAMPLES] is available.
*
* A ring buffer rather than a sliding array: detection is throttled to
* [DETECT_INTERVAL_MS] but the buffer fills in 400 ms, so for the remaining 1.6 s
* every chunk arrives at a full buffer. Shifting the array down one slot per sample
* cost 320 copies of 1280 floats per chunk - measured at 24320 whole-array moves per
* 10 s of audio, on the capture thread. Overwriting the oldest slot is O(1).
*/
private fun accumulateForDetection(chunk: FloatArray) { private fun accumulateForDetection(chunk: FloatArray) {
if (chunk.isEmpty()) return if (chunk.isEmpty()) return
// A chunk larger than the detection buffer only needs to contribute its tail. // A chunk longer than the buffer can only contribute its tail.
val start = maxOf(0, chunk.size - detectBuffer.size) val start = maxOf(0, chunk.size - detectBuffer.size)
for (i in start until chunk.size) { for (i in start until chunk.size) {
if (detectFill == detectBuffer.size) { detectBuffer[detectWriteIndex] = chunk[i]
// Slide the window so detection always sees the most recent audio. detectWriteIndex = (detectWriteIndex + 1) % detectBuffer.size
detectBuffer.copyInto(detectBuffer, 0, 1, detectFill) if (detectFill < detectBuffer.size) detectFill++
detectFill--
}
detectBuffer[detectFill++] = chunk[i]
} }
} }
/** Copy the buffered audio out in chronological order, oldest sample first. */
private fun drainDetectionBuffer(): FloatArray {
val out = FloatArray(detectFill)
// When full, the oldest sample sits at the write cursor; otherwise at index 0.
val oldest = if (detectFill == detectBuffer.size) detectWriteIndex else 0
for (i in 0 until detectFill) {
out[i] = detectBuffer[(oldest + i) % detectBuffer.size]
}
detectFill = 0
detectWriteIndex = 0
return out
}
/** /**
* Discard buffered audio that was shifted by a now-stale amount. * Discard buffered audio that was shifted by a now-stale amount.
* *
@@ -530,6 +548,7 @@ class CwDeepDecoder(
detectedToneHz = null detectedToneHz = null
lastDetectAtMs = 0L lastDetectAtMs = 0L
detectFill = 0 detectFill = 0
detectWriteIndex = 0
streamingShifter.reset() streamingShifter.reset()
// Leave toneShiftWasEnabled unset so the next chunk re-seeds it from the // Leave toneShiftWasEnabled unset so the next chunk re-seeds it from the
// current setting instead of reporting a spurious change. // current setting instead of reporting a spurious change.
@@ -161,4 +161,69 @@ class CwToneShiftGateTest {
abs(shiftAt1300 - shiftAt1400) >= hysteresisHz abs(shiftAt1300 - shiftAt1400) >= hysteresisHz
) )
} }
/**
* The detection pool must hand the analyser the most recent audio in chronological
* order. The decoder implements this as a ring buffer because detection is throttled
* to 2 s while the pool fills in 400 ms, so most chunks arrive at a full buffer.
*
* This pins the contract with the same ring semantics the decoder uses: getting the
* order wrong would feed the detector a spliced waveform and corrupt every estimate,
* silently, which no other assertion here would notice.
*/
@Test
fun `detection pool yields the most recent samples in order`() {
val capacity = 1280 // CwDeepDecoder.DETECT_MIN_SAMPLES
val pool = FloatArray(capacity)
var fill = 0
var writeIndex = 0
fun accumulate(chunk: FloatArray) {
val start = maxOf(0, chunk.size - pool.size)
for (i in start until chunk.size) {
pool[writeIndex] = chunk[i]
writeIndex = (writeIndex + 1) % pool.size
if (fill < pool.size) fill++
}
}
fun drain(): FloatArray {
val out = FloatArray(fill)
val oldest = if (fill == pool.size) writeIndex else 0
for (i in 0 until fill) out[i] = pool[(oldest + i) % pool.size]
fill = 0
writeIndex = 0
return out
}
// Feed a monotonically increasing ramp in 320-sample chunks, past capacity.
var next = 0f
repeat(10) {
accumulate(FloatArray(320) { next++ })
}
val drained = drain()
assertEquals("a full pool must hand over exactly its capacity", capacity, drained.size)
assertEquals(
"the pool must end on the newest sample fed",
next - 1f, drained.last(), 0f
)
assertEquals(
"the pool must start capacity-1 samples before the newest",
next - capacity, drained.first(), 0f
)
for (i in 1 until drained.size) {
assertEquals(
"sample $i is out of order, so the ring wrap is wrong",
drained[i - 1] + 1f, drained[i], 0f
)
}
// A short run must come back whole, without stale slots from the previous pass.
accumulate(FloatArray(320) { 9000f + it })
val partial = drain()
assertEquals("a partial pool must not report capacity", 320, partial.size)
assertEquals(9000f, partial.first(), 0f)
assertEquals(9319f, partial.last(), 0f)
}
} }