zero-alloc audio pipeline: pointer-based IO from ring buffer to stdout

Replace Data/AudioPacket allocations with raw pointer callbacks through
the entire audio pipeline. Ring buffer hands out direct pointers (or
linearizes into a pre-allocated scratch buffer on wrap-around), converter
accepts/emits pointers via its cached buffers, and output handler writes
to stdout via write(2) with EINTR handling.

Remove AudioPacket (dead code), --flush flag (no-op with raw write(2)).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Nick Payne
2026-03-07 07:14:16 +00:00
parent 65c2d58c82
commit a1eb465142
9 changed files with 122 additions and 174 deletions
+1 -6
View File
@@ -9,7 +9,6 @@ struct AudioTee {
var stereo: Bool = false var stereo: Bool = false
var sampleRate: Double? var sampleRate: Double?
var chunkDuration: Double = 0.2 var chunkDuration: Double = 0.2
var flush: Bool = false
init() {} init() {}
@@ -33,7 +32,6 @@ struct AudioTee {
audiotee --include-processes 1234 5678 9012 # Tap only these processes audiotee --include-processes 1234 5678 9012 # Tap only these processes
audiotee --exclude-processes 1234 5678 # Tap everything except these audiotee --exclude-processes 1234 5678 # Tap everything except these
audiotee --mute # Mute processes being tapped audiotee --mute # Mute processes being tapped
audiotee --flush # Flush stdout after each chunk
""" """
) )
@@ -45,8 +43,6 @@ struct AudioTee {
name: "exclude-processes", help: "Process IDs to exclude (space-separated)") name: "exclude-processes", help: "Process IDs to exclude (space-separated)")
parser.addFlag(name: "mute", help: "Mute processes being tapped") parser.addFlag(name: "mute", help: "Mute processes being tapped")
parser.addFlag(name: "stereo", help: "Records in stereo") parser.addFlag(name: "stereo", help: "Records in stereo")
parser.addFlag(
name: "flush", help: "Flush stdout after each audio chunk (reduces latency when piping)")
parser.addOption( parser.addOption(
name: "sample-rate", name: "sample-rate",
help: "Target sample rate (8000, 16000, 22050, 24000, 32000, 44100, 48000)") help: "Target sample rate (8000, 16000, 22050, 24000, 32000, 44100, 48000)")
@@ -64,7 +60,6 @@ struct AudioTee {
audioTee.excludeProcesses = try parser.getArrayValue("exclude-processes", as: Int32.self) audioTee.excludeProcesses = try parser.getArrayValue("exclude-processes", as: Int32.self)
audioTee.mute = parser.getFlag("mute") audioTee.mute = parser.getFlag("mute")
audioTee.stereo = parser.getFlag("stereo") audioTee.stereo = parser.getFlag("stereo")
audioTee.flush = parser.getFlag("flush")
audioTee.sampleRate = try parser.getOptionalValue("sample-rate", as: Double.self) audioTee.sampleRate = try parser.getOptionalValue("sample-rate", as: Double.self)
audioTee.chunkDuration = try parser.getValue("chunk-duration", as: Double.self) audioTee.chunkDuration = try parser.getValue("chunk-duration", as: Double.self)
@@ -142,7 +137,7 @@ struct AudioTee {
throw ExitCode.failure throw ExitCode.failure
} }
let outputHandler = BinaryAudioOutputHandler(flushAfterWrite: flush) let outputHandler = BinaryAudioOutputHandler()
let recorder = try AudioRecorder( let recorder = try AudioRecorder(
deviceID: deviceID, outputHandler: outputHandler, convertToSampleRate: sampleRate, deviceID: deviceID, outputHandler: outputHandler, convertToSampleRate: sampleRate,
chunkDuration: chunkDuration) chunkDuration: chunkDuration)
+11 -9
View File
@@ -4,17 +4,19 @@ import Foundation
/// CLI-specific output handler that writes raw PCM audio to stdout /// CLI-specific output handler that writes raw PCM audio to stdout
/// and lifecycle messages to stderr via the logger. /// and lifecycle messages to stderr via the logger.
class BinaryAudioOutputHandler: AudioOutputHandler { class BinaryAudioOutputHandler: AudioOutputHandler {
private let flushAfterWrite: Bool private let fd = STDOUT_FILENO
init(flushAfterWrite: Bool = false) { func handleAudioData(_ pointer: UnsafeRawPointer, count: Int) {
self.flushAfterWrite = flushAfterWrite var written = 0
while written < count {
let result = write(fd, pointer.advanced(by: written), count - written)
if result >= 0 {
written += result
} else if errno == EINTR {
continue
} else {
break // EPIPE, EIO, etc consumer gone or real error
} }
func handleAudioPacket(_ packet: AudioPacket) {
// Write raw binary audio data directly to stdout
FileHandle.standardOutput.write(packet.data)
if flushAfterWrite {
fflush(stdout)
} }
} }
+27 -38
View File
@@ -10,20 +10,21 @@ import Foundation
public class AudioBuffer { public class AudioBuffer {
/// Raw heap-allocated ring buffer backing store. /// Raw heap-allocated ring buffer backing store.
private let buffer: UnsafeMutableRawPointer private let buffer: UnsafeMutableRawPointer
/// Pre-allocated buffer for linearizing chunks that straddle the ring
/// buffer boundary. Avoids a heap allocation on the wrap-around path.
private let linearizationBuffer: UnsafeMutableRawPointer
private var writeIndex: Int = 0 private var writeIndex: Int = 0
private var readIndex: Int = 0 private var readIndex: Int = 0
private var availableBytes: Int = 0 private var availableBytes: Int = 0
private let maxBufferSize: Int private let maxBufferSize: Int
private let bytesPerChunk: Int public let bytesPerChunk: Int
private let chunkDuration: Double
public init(format: AudioStreamBasicDescription, chunkDuration: Double = 0.2) { public init(format: AudioStreamBasicDescription, chunkDuration: Double = 0.2) {
// Pre-calculate chunk parameters // Pre-calculate chunk parameters
let bytesPerFrame = Int(format.mBytesPerFrame) let bytesPerFrame = Int(format.mBytesPerFrame)
let samplesPerChunk = Int(format.mSampleRate * chunkDuration) let samplesPerChunk = Int(format.mSampleRate * chunkDuration)
self.bytesPerChunk = samplesPerChunk * bytesPerFrame self.bytesPerChunk = samplesPerChunk * bytesPerFrame
self.chunkDuration = Double(samplesPerChunk) / format.mSampleRate
// Calculate max buffer size to hold ~10 seconds of audio (safety limit) // Calculate max buffer size to hold ~10 seconds of audio (safety limit)
let bytesPerSecond = Int(format.mSampleRate) * bytesPerFrame let bytesPerSecond = Int(format.mSampleRate) * bytesPerFrame
@@ -36,10 +37,16 @@ public class AudioBuffer {
alignment: MemoryLayout<UInt8>.alignment alignment: MemoryLayout<UInt8>.alignment
) )
buffer.initializeMemory(as: UInt8.self, repeating: 0, count: maxBufferSize) buffer.initializeMemory(as: UInt8.self, repeating: 0, count: maxBufferSize)
self.linearizationBuffer = UnsafeMutableRawPointer.allocate(
byteCount: bytesPerChunk,
alignment: MemoryLayout<UInt8>.alignment
)
} }
deinit { deinit {
buffer.deallocate() buffer.deallocate()
linearizationBuffer.deallocate()
} }
/// Appends audio data directly from a raw pointer into the ring buffer. /// Appends audio data directly from a raw pointer into the ring buffer.
@@ -82,51 +89,33 @@ public class AudioBuffer {
availableBytes += count availableBytes += count
} }
/// Extracts all complete chunks currently available in the buffer. /// Calls `handler` once for each complete chunk available in the buffer.
public func processChunks() -> [AudioPacket] { /// The pointer passed to the handler is valid only for the duration of
var packets: [AudioPacket] = [] /// that call. In the common (contiguous) case this points directly into
/// the ring buffer zero copies. In the wrap-around case the chunk is
while let packet = nextChunk() { /// linearized into a pre-allocated scratch buffer one memcpy, zero
packets.append(packet) /// heap allocations.
} public func processChunks(_ handler: (UnsafeRawPointer, Int) -> Void) {
while availableBytes >= bytesPerChunk {
return packets
}
private func nextChunk() -> AudioPacket? {
// Check if we have enough data for a complete chunk
guard availableBytes >= bytesPerChunk else { return nil }
let chunkData: Data
// Check if we can copy in one block (no wrap-around)
if readIndex + bytesPerChunk <= maxBufferSize { if readIndex + bytesPerChunk <= maxBufferSize {
// one copy needed // Contiguous: point directly into the ring buffer
chunkData = Data(bytes: buffer.advanced(by: readIndex), count: bytesPerChunk) handler(buffer.advanced(by: readIndex), bytesPerChunk)
readIndex = (readIndex + bytesPerChunk) % maxBufferSize readIndex = (readIndex + bytesPerChunk) % maxBufferSize
} else { } else {
// two copies needed due to wrap-around // Wrap-around: linearize into the pre-allocated scratch buffer
let firstChunkSize = maxBufferSize - readIndex let firstChunkSize = maxBufferSize - readIndex
let secondChunkSize = bytesPerChunk - firstChunkSize let secondChunkSize = bytesPerChunk - firstChunkSize
var assembled = Data(capacity: bytesPerChunk) linearizationBuffer.copyMemory(
assembled.append( from: buffer.advanced(by: readIndex), byteCount: firstChunkSize)
buffer.advanced(by: readIndex).assumingMemoryBound(to: UInt8.self), linearizationBuffer.advanced(by: firstChunkSize).copyMemory(
count: firstChunkSize) from: buffer, byteCount: secondChunkSize)
assembled.append(
buffer.assumingMemoryBound(to: UInt8.self),
count: secondChunkSize)
chunkData = assembled
handler(linearizationBuffer, bytesPerChunk)
readIndex = secondChunkSize readIndex = secondChunkSize
} }
availableBytes -= bytesPerChunk availableBytes -= bytesPerChunk
}
return AudioPacket(
timestamp: Date(),
duration: chunkDuration,
data: chunkData
)
} }
} }
@@ -126,27 +126,28 @@ public class AudioFormatConverter {
return (inputBuf, outputBuf) return (inputBuf, outputBuf)
} }
public func transform(_ packet: AudioPacket) -> AudioPacket { /// Converts audio data in-place through the pre-allocated converter buffers.
let inputData = packet.data /// Calls `handler` with a pointer to the converted output, valid only for
/// the duration of that call. Returns false on failure (caller should
// Calculate frame count from the input data size /// pass through the original data or drop it).
@discardableResult
public func transform(
from source: UnsafeRawPointer, count: Int,
handler: (UnsafeRawPointer, Int) -> Void
) -> Bool {
let bytesPerFrame = Int(sourceFormat.streamDescription.pointee.mBytesPerFrame) let bytesPerFrame = Int(sourceFormat.streamDescription.pointee.mBytesPerFrame)
let inputFrameCount = AVAudioFrameCount(inputData.count / bytesPerFrame) let inputFrameCount = AVAudioFrameCount(count / bytesPerFrame)
// Get or create pre-allocated buffers
guard let (inputBuffer, outputBuffer) = getBuffers(inputFrameCount: inputFrameCount) else { guard let (inputBuffer, outputBuffer) = getBuffers(inputFrameCount: inputFrameCount) else {
return packet return false
} }
// Copy input data into the reusable input buffer // Copy source data into the reusable input buffer
inputData.withUnsafeBytes { bytes in
let dest = inputBuffer.audioBufferList.pointee.mBuffers.mData! let dest = inputBuffer.audioBufferList.pointee.mBuffers.mData!
dest.copyMemory(from: bytes.baseAddress!, byteCount: inputData.count) dest.copyMemory(from: source, byteCount: count)
}
inputBuffer.frameLength = inputFrameCount inputBuffer.frameLength = inputFrameCount
// Perform conversion the block-based API lets AVAudioConverter pull // Perform conversion we do NOT call avConverter.reset() between
// input data as needed. We do NOT call avConverter.reset() between
// calls because the resampler maintains internal state for continuity // calls because the resampler maintains internal state for continuity
// across chunks (avoiding discontinuity artifacts). // across chunks (avoiding discontinuity artifacts).
var error: NSError? var error: NSError?
@@ -157,7 +158,6 @@ public class AudioFormatConverter {
return inputBuffer return inputBuffer
} }
// Check if conversion produced output (regardless of status code)
guard outputBuffer.frameLength > 0 else { guard outputBuffer.frameLength > 0 else {
AudioTeeLogging.logger.error( AudioTeeLogging.logger.error(
"Audio conversion produced no output", "Audio conversion produced no output",
@@ -167,19 +167,13 @@ public class AudioFormatConverter {
"input_frames": String(inputBuffer.frameLength), "input_frames": String(inputBuffer.frameLength),
"output_capacity": String(outputBuffer.frameCapacity), "output_capacity": String(outputBuffer.frameCapacity),
]) ])
return packet return false
} }
// Extract converted data from the reusable output buffer let outputCount = Int(
let outputData = Data( outputBuffer.frameLength * targetFormat.streamDescription.pointee.mBytesPerFrame)
bytes: outputBuffer.audioBufferList.pointee.mBuffers.mData!, handler(outputBuffer.audioBufferList.pointee.mBuffers.mData!, outputCount)
count: Int(outputBuffer.frameLength * targetFormat.streamDescription.pointee.mBytesPerFrame)) return true
return AudioPacket(
timestamp: packet.timestamp,
duration: packet.duration,
data: outputData
)
} }
public static func toSampleRate( public static func toSampleRate(
@@ -1,17 +0,0 @@
import Foundation
public struct AudioPacket {
public let timestamp: Date
public let duration: Double
public let data: Data
public init(
timestamp: Date,
duration: Double,
data: Data
) {
self.timestamp = timestamp
self.duration = duration
self.data = data
}
}
+11 -4
View File
@@ -132,10 +132,17 @@ public class AudioRecorder {
} }
private func processAudioBuffer() { private func processAudioBuffer() {
// Process and send complete chunks, applying conversion if needed audioBuffer?.processChunks { pointer, count in
audioBuffer?.processChunks().forEach { packet in if let converter = self.converter {
let processedPacket = converter?.transform(packet) ?? packet if !converter.transform(from: pointer, count: count, handler: { outPtr, outCount in
outputHandler.handleAudioPacket(processedPacket) self.outputHandler.handleAudioData(outPtr, count: outCount)
}) {
// Conversion failed pass through unconverted audio
self.outputHandler.handleAudioData(pointer, count: count)
}
} else {
self.outputHandler.handleAudioData(pointer, count: count)
}
} }
} }
@@ -2,7 +2,9 @@ import Foundation
/// Protocol for handling audio output in different formats /// Protocol for handling audio output in different formats
public protocol AudioOutputHandler { public protocol AudioOutputHandler {
func handleAudioPacket(_ packet: AudioPacket) /// Called with a pointer to raw PCM audio data. The pointer is only
/// valid for the duration of this call.
func handleAudioData(_ pointer: UnsafeRawPointer, count: Int)
func handleMetadata(_ metadata: AudioStreamMetadata) func handleMetadata(_ metadata: AudioStreamMetadata)
func handleStreamStart() func handleStreamStart()
func handleStreamStop() func handleStreamStop()
+39 -32
View File
@@ -44,6 +44,15 @@ final class AudioBufferTests: XCTestCase {
} }
} }
/// Collects chunks from the buffer as Data objects for test verification.
private func collectChunks(from buffer: AudioBuffer) -> [Data] {
var chunks: [Data] = []
buffer.processChunks { pointer, count in
chunks.append(Data(bytes: pointer, count: count))
}
return chunks
}
// MARK: - Basic append + processChunks // MARK: - Basic append + processChunks
func testSingleChunkExtraction() { func testSingleChunkExtraction() {
@@ -55,10 +64,10 @@ final class AudioBufferTests: XCTestCase {
let data = makeData(byte: 0xAB, count: chunkSize) let data = makeData(byte: 0xAB, count: chunkSize)
appendData(data, to: buffer) appendData(data, to: buffer)
let packets = buffer.processChunks() let chunks = collectChunks(from: buffer)
XCTAssertEqual(packets.count, 1) XCTAssertEqual(chunks.count, 1)
XCTAssertEqual(packets[0].data.count, chunkSize) XCTAssertEqual(chunks[0].count, chunkSize)
XCTAssertEqual(packets[0].data, data) XCTAssertEqual(chunks[0], data)
} }
func testMultipleChunksExtracted() { func testMultipleChunksExtracted() {
@@ -69,11 +78,11 @@ final class AudioBufferTests: XCTestCase {
// Append 2.5 chunks worth // Append 2.5 chunks worth
appendData(makeData(byte: 0x01, count: chunkSize * 2 + chunkSize / 2), to: buffer) appendData(makeData(byte: 0x01, count: chunkSize * 2 + chunkSize / 2), to: buffer)
let packets = buffer.processChunks() let chunks = collectChunks(from: buffer)
// Should get 2 complete chunks, remainder stays in buffer // Should get 2 complete chunks, remainder stays in buffer
XCTAssertEqual(packets.count, 2) XCTAssertEqual(chunks.count, 2)
XCTAssertEqual(packets[0].data.count, chunkSize) XCTAssertEqual(chunks[0].count, chunkSize)
XCTAssertEqual(packets[1].data.count, chunkSize) XCTAssertEqual(chunks[1].count, chunkSize)
} }
func testInsufficientDataReturnsNoChunks() { func testInsufficientDataReturnsNoChunks() {
@@ -84,8 +93,8 @@ final class AudioBufferTests: XCTestCase {
// Append less than one chunk // Append less than one chunk
appendData(makeData(byte: 0xFF, count: chunkSize - 1), to: buffer) appendData(makeData(byte: 0xFF, count: chunkSize - 1), to: buffer)
let packets = buffer.processChunks() let chunks = collectChunks(from: buffer)
XCTAssertEqual(packets.count, 0) XCTAssertEqual(chunks.count, 0)
} }
// MARK: - Wrap-around // MARK: - Wrap-around
@@ -103,7 +112,7 @@ final class AudioBufferTests: XCTestCase {
for _ in 0..<33 { for _ in 0..<33 {
appendData(makeData(byte: 0x00, count: chunkSize), to: buffer) appendData(makeData(byte: 0x00, count: chunkSize), to: buffer)
} }
let drained = buffer.processChunks() let drained = collectChunks(from: buffer)
XCTAssertEqual(drained.count, 33) XCTAssertEqual(drained.count, 33)
// Next write of 4800 bytes starts at 158400. 158400 + 4800 = 163200 > 160000. // Next write of 4800 bytes starts at 158400. 158400 + 4800 = 163200 > 160000.
@@ -117,9 +126,9 @@ final class AudioBufferTests: XCTestCase {
XCTAssertEqual(wrappingData.count, chunkSize) XCTAssertEqual(wrappingData.count, chunkSize)
appendData(wrappingData, to: buffer) appendData(wrappingData, to: buffer)
let packets = buffer.processChunks() let chunks = collectChunks(from: buffer)
XCTAssertEqual(packets.count, 1) XCTAssertEqual(chunks.count, 1)
XCTAssertEqual(packets[0].data, wrappingData) XCTAssertEqual(chunks[0], wrappingData)
} }
func testWrapAroundRead() { func testWrapAroundRead() {
@@ -133,7 +142,7 @@ final class AudioBufferTests: XCTestCase {
for _ in 0..<33 { for _ in 0..<33 {
appendData(makeData(byte: 0x00, count: chunkSize), to: buffer) appendData(makeData(byte: 0x00, count: chunkSize), to: buffer)
} }
_ = buffer.processChunks() _ = collectChunks(from: buffer)
// Write one chunk starting at 158400. The write itself wraps (tested above), // Write one chunk starting at 158400. The write itself wraps (tested above),
// but crucially the READ will also wrap: readIndex = 158400, // but crucially the READ will also wrap: readIndex = 158400,
@@ -145,9 +154,9 @@ final class AudioBufferTests: XCTestCase {
crossBoundaryData.append(makeData(byte: 0xDD, count: 3200)) crossBoundaryData.append(makeData(byte: 0xDD, count: 3200))
appendData(crossBoundaryData, to: buffer) appendData(crossBoundaryData, to: buffer)
let packets = buffer.processChunks() let chunks = collectChunks(from: buffer)
XCTAssertEqual(packets.count, 1) XCTAssertEqual(chunks.count, 1)
XCTAssertEqual(packets[0].data, crossBoundaryData) XCTAssertEqual(chunks[0], crossBoundaryData)
} }
// MARK: - Overflow guard // MARK: - Overflow guard
@@ -164,13 +173,13 @@ final class AudioBufferTests: XCTestCase {
appendData(makeData(byte: 0x02, count: 100), to: buffer) appendData(makeData(byte: 0x02, count: 100), to: buffer)
// Drain and verify we only got the original data // Drain and verify we only got the original data
let packets = buffer.processChunks() let chunks = collectChunks(from: buffer)
let totalBytes = packets.reduce(0) { $0 + $1.data.count } let totalBytes = chunks.reduce(0) { $0 + $1.count }
XCTAssertEqual(totalBytes, maxBuffer) XCTAssertEqual(totalBytes, maxBuffer)
// Every byte should be 0x01, not 0x02 // Every byte should be 0x01, not 0x02
for packet in packets { for chunk in chunks {
XCTAssertTrue(packet.data.allSatisfy { $0 == 0x01 }) XCTAssertTrue(chunk.allSatisfy { $0 == 0x01 })
} }
} }
@@ -187,26 +196,24 @@ final class AudioBufferTests: XCTestCase {
appendData(makeData(byte: UInt8(i), count: callbackSize), to: buffer) appendData(makeData(byte: UInt8(i), count: callbackSize), to: buffer)
} }
let packets = buffer.processChunks() let chunks = collectChunks(from: buffer)
XCTAssertEqual(packets.count, 1) XCTAssertEqual(chunks.count, 1)
XCTAssertEqual(packets[0].data.count, chunkSize) XCTAssertEqual(chunks[0].count, chunkSize)
// Verify the data is in the correct order // Verify the data is in the correct order
for i in 0..<10 { for i in 0..<10 {
let slice = packets[0].data.subdata(in: (i * callbackSize)..<((i + 1) * callbackSize)) let slice = chunks[0].subdata(in: (i * callbackSize)..<((i + 1) * callbackSize))
XCTAssertTrue(slice.allSatisfy { $0 == UInt8(i) }) XCTAssertTrue(slice.allSatisfy { $0 == UInt8(i) })
} }
} }
// MARK: - Packet metadata // MARK: - Chunk size
func testChunkDurationIsCorrect() { func testBytesPerChunkIsCorrect() {
let format = makeFormat() let format = makeFormat()
let buffer = AudioBuffer(format: format, chunkDuration: 0.1) let buffer = AudioBuffer(format: format, chunkDuration: 0.1)
appendData(makeData(byte: 0x00, count: 3200), to: buffer) // 16kHz * 0.1s * 2 bytes/frame = 3200
let packets = buffer.processChunks() XCTAssertEqual(buffer.bytesPerChunk, 3200)
XCTAssertEqual(packets[0].duration, 0.1, accuracy: 0.001)
} }
} }
@@ -1,31 +0,0 @@
import XCTest
@testable import AudioTeeCore
final class AudioPacketTests: XCTestCase {
func testPacketCreation() {
let timestamp = Date()
let duration = 1.0
let data = Data([0x01, 0x02, 0x03, 0x04])
let packet = AudioPacket(
timestamp: timestamp,
duration: duration,
data: data
)
XCTAssertEqual(packet.timestamp, timestamp)
XCTAssertEqual(packet.duration, duration)
XCTAssertEqual(packet.data, data)
}
func testPacketDataSize() {
let packet = AudioPacket(
timestamp: Date(),
duration: 0.5,
data: Data(repeating: 0xFF, count: 1024)
)
XCTAssertEqual(packet.data.count, 1024)
}
}