diff --git a/app/common/audio/stream.go b/app/common/audio/stream.go index 2a3e72a..3003b38 100644 --- a/app/common/audio/stream.go +++ b/app/common/audio/stream.go @@ -7,13 +7,14 @@ import ( "sync" ) -// Producers never block: the queue is elastic so generation can run ahead of -// playback. The limit only guards against a consumer that never reads. +// The queue is elastic so generation can run ahead of playback, but it is +// bounded by streamQueueLimit as a backpressure high-water mark: once the +// queued bytes reach the limit, Write blocks until the consumer drains space +// instead of buffering unboundedly. Closing the stream unblocks the writer. var streamQueueLimit = 64 * 1024 * 1024 var ( - ErrStreamClosed = errors.New("audio stream is closed") - ErrStreamLimitReached = errors.New("audio stream queue limit exceeded") + ErrStreamClosed = errors.New("audio stream is closed") ) type StreamFormat struct { @@ -129,6 +130,12 @@ func (stream *Stream) Read(destination []byte) (int, error) { copied := copy(destination, stream.queue) stream.queue = stream.queue[copied:] + // Draining the queue may free space below the high-water mark; wake any + // writer blocked in Write applying backpressure. + if copied > 0 { + stream.condition.Broadcast() + } + if stream.teeSink != nil && copied > 0 { if _, err := stream.teeSink.Write(destination[:copied]); err != nil { stream.teeError = err @@ -189,11 +196,18 @@ func (writer *StreamWriter) Write(data []byte) (int, error) { stream.mutex.Lock() defer stream.mutex.Unlock() - if stream.consumerClosed || stream.producerClosed { - return 0, ErrStreamClosed - } - if len(stream.queue)+len(data) > streamQueueLimit { - return 0, ErrStreamLimitReached + // Apply backpressure: once the queue reaches the high-water mark, block + // until the consumer drains enough space instead of buffering unboundedly. + // A blocked writer wakes if the stream is closed or errored so a + // genuinely-gone consumer can never deadlock the producer forever. + for { + if stream.consumerClosed || stream.producerClosed { + return 0, ErrStreamClosed + } + if len(stream.queue) < streamQueueLimit { + break + } + stream.condition.Wait() } stream.queue = append(stream.queue, data...) diff --git a/app/common/audio/stream_test.go b/app/common/audio/stream_test.go index 3ca88b2..3388e27 100644 --- a/app/common/audio/stream_test.go +++ b/app/common/audio/stream_test.go @@ -161,18 +161,85 @@ func TestPipeStream(t *testing.T) { } }) - t.Run("QueueLimitFailsWrite", func(t *testing.T) { + t.Run("QueueLimitBlocksWriteUntilRead", func(t *testing.T) { originalLimit := streamQueueLimit streamQueueLimit = 8 t.Cleanup(func() { streamQueueLimit = originalLimit }) - _, writer := NewPipeStream(defaultStreamFormat()) + stream, writer := NewPipeStream(defaultStreamFormat()) + // Fill the queue up to the high-water mark. if _, err := writer.Write(patternBytes(8)); err != nil { t.Fatalf("Write within limit: %v", err) } - if _, err := writer.Write([]byte{0}); err != ErrStreamLimitReached { - t.Fatalf("got %v, want ErrStreamLimitReached", err) + + // The next write must block: the queue is at the limit. + writeDone := make(chan error, 1) + go func() { + _, err := writer.Write([]byte("more")) + writeDone <- err + }() + + select { + case err := <-writeDone: + t.Fatalf("Write returned %v while queue was full; expected it to block", err) + case <-time.After(50 * time.Millisecond): + // Still blocked, as intended. + } + + // Reading frees space below the limit, which must unblock the writer. + readExactly(t, stream, 8) + + select { + case err := <-writeDone: + if err != nil { + t.Fatalf("blocked Write returned %v after read freed space", err) + } + case <-time.After(5 * time.Second): + t.Fatal("Write never returned after a read freed queue space") + } + + data := readExactly(t, stream, 4) + if string(data) != "more" { + t.Fatalf("got %q, want %q", data, "more") + } + }) + + t.Run("ConsumerCloseUnblocksWrite", func(t *testing.T) { + originalLimit := streamQueueLimit + streamQueueLimit = 8 + t.Cleanup(func() { streamQueueLimit = originalLimit }) + + stream, writer := NewPipeStream(defaultStreamFormat()) + + if _, err := writer.Write(patternBytes(8)); err != nil { + t.Fatalf("Write within limit: %v", err) + } + + writeDone := make(chan error, 1) + go func() { + _, err := writer.Write([]byte("more")) + writeDone <- err + }() + + // Ensure the writer is parked on the condition before closing. + select { + case err := <-writeDone: + t.Fatalf("Write returned %v while queue was full; expected it to block", err) + case <-time.After(50 * time.Millisecond): + } + + // Closing the stream must wake the blocked writer with an error rather + // than deadlocking a producer whose consumer has gone away. + stream.Close() + + select { + case err := <-writeDone: + if err != ErrStreamClosed { + t.Fatalf("got %v, want ErrStreamClosed after consumer close", err) + } + case <-time.After(5 * time.Second): + t.Fatal("Write never returned after consumer Close unblocked it") } })