storm/skald
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Start work on improving recordings.

Browse Source
This commit is contained in:
Storm Dragon
2026-05-23 23:33:30 -04:00
parent 08b484f106
commit 3e06de7501
8 changed files with 662 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files
diskwriter/diskwriter.go
+354 -75
View File
@@ -8,6 +8,7 @@ import (
"fmt"
"io"
"log"
"math"
"net"
"os"
"os/exec"
@@ -24,12 +25,16 @@ import (
)
const (
sampleRate = 48000
channels = 2
frameDuration = 20 * time.Millisecond
samplesPerFrame = sampleRate / 1000 * 20
bytesPerFrame = samplesPerFrame * channels * 2
sourceBufferSize = 16
sampleRate = 48000
channels = 2
frameDuration = 20 * time.Millisecond
samplesPerFrame = sampleRate / 1000 * 20
bytesPerFrame = samplesPerFrame * channels * 2
sourceBufferSize = 16
encoderBufferSize = 50
sourceStaleAfter = 2 * time.Second
limiterThreshold = 28000
limiterRatio = 4
)
var Directory string
@@ -197,17 +202,46 @@ func (client *Client) PushConn(g *hall.Hall, id string, up conn.Up, tracks []con
}
type recorder struct {
client *Client
path string
cmd *exec.Cmd
stdin io.WriteCloser
done chan error
client *Client
path string
cmd *exec.Cmd
stdin io.WriteCloser
done chan error
writeCh chan []byte
writerDone chan error
mu sync.Mutex
sources map[*diskTrack]chan []byte
sources map[*diskTrack]*recordingSource
metrics RecorderMetrics
closed bool
}
type recordingSource struct {
ch chan []byte
lastFrame time.Time
decoderStarted bool
decoderStopped bool
staleLogged bool
}
type RecorderMetrics struct {
Started time.Time
Duration time.Duration
ActiveSources int
DecoderStarts uint64
DecoderStops uint64
DroppedSourceFrames uint64
StaleSourceTicks uint64
DroppedMixedFrames uint64
EncoderWriteErrors uint64
MixerFrames uint64
}
type SyntheticRecordingResult struct {
Path string
Metrics RecorderMetrics
}
func newRecorder(client *Client, directory string) (*recorder, error) {
file, err := openDiskFile(directory, "ogg")
if err != nil {
@@ -245,35 +279,107 @@ func newRecorder(client *Client, directory string) (*recorder, error) {
go drainStderr("ffmpeg encoder", stderr)
r := &recorder{
client: client,
path: path,
cmd: cmd,
stdin: stdin,
done: make(chan error, 1),
sources: make(map[*diskTrack]chan []byte),
client: client,
path: path,
cmd: cmd,
stdin: stdin,
done: make(chan error, 1),
writeCh: make(chan []byte, encoderBufferSize),
writerDone: make(chan error, 1),
sources: make(map[*diskTrack]*recordingSource),
metrics: RecorderMetrics{Started: time.Now()},
}
activeRecordings.Store(path, struct{}{})
go r.writeEncoder()
go r.mix()
return r, nil
}
func (r *recorder) addSource(t *diskTrack) chan []byte {
func (r *recorder) addSource(t *diskTrack) *recordingSource {
r.mu.Lock()
defer r.mu.Unlock()
if r.closed {
return nil
}
ch := make(chan []byte, sourceBufferSize)
r.sources[t] = ch
return ch
source := &recordingSource{
ch: make(chan []byte, sourceBufferSize),
lastFrame: time.Now(),
}
r.sources[t] = source
return source
}
func (r *recorder) noteDecoderStarted(t *diskTrack) {
r.mu.Lock()
defer r.mu.Unlock()
source := r.sources[t]
if source == nil || source.decoderStarted {
return
}
source.decoderStarted = true
r.metrics.DecoderStarts++
log.Printf("Recording %s: decoder started, active sources %d",
filepath.Base(r.path), len(r.sources))
}
func (r *recorder) removeSource(t *diskTrack) {
r.mu.Lock()
defer r.mu.Unlock()
source := r.sources[t]
if source == nil {
return
}
if source.decoderStarted && !source.decoderStopped {
source.decoderStopped = true
r.metrics.DecoderStops++
}
delete(r.sources, t)
log.Printf("Recording %s: source removed, active sources %d",
filepath.Base(r.path), len(r.sources))
}
func (r *recorder) noteDecoderStopped(t *diskTrack, err error) {
r.mu.Lock()
defer r.mu.Unlock()
source := r.sources[t]
if source == nil || source.decoderStopped {
return
}
source.decoderStopped = true
r.metrics.DecoderStops++
delete(r.sources, t)
if err != nil && !errors.Is(err, io.EOF) && !errors.Is(err, io.ErrUnexpectedEOF) {
log.Printf("Recording %s: decoder stopped: %v",
filepath.Base(r.path), err)
} else {
log.Printf("Recording %s: decoder stopped, active sources %d",
filepath.Base(r.path), len(r.sources))
}
}
func (r *recorder) pushSourceFrame(t *diskTrack, frame []byte) bool {
r.mu.Lock()
defer r.mu.Unlock()
source := r.sources[t]
if source == nil {
return false
}
now := time.Now()
source.lastFrame = now
source.staleLogged = false
select {
case source.ch <- frame:
default:
<-source.ch
source.ch <- frame
r.metrics.DroppedSourceFrames++
}
return true
}
func (r *recorder) mix() {
@@ -281,74 +387,149 @@ func (r *recorder) mix() {
defer ticker.Stop()
silence := make([]byte, bytesPerFrame)
for range ticker.C {
r.mu.Lock()
if r.closed {
r.mu.Unlock()
r.done <- nil
return
}
sources := make([]chan []byte, 0, len(r.sources))
for _, ch := range r.sources {
sources = append(sources, ch)
}
r.mu.Unlock()
frame := mixFrame(sources)
if frame == nil {
frame = silence
}
if _, err := r.stdin.Write(frame); err != nil {
for {
select {
case err := <-r.writerDone:
if err != nil {
r.mu.Lock()
r.metrics.EncoderWriteErrors++
r.mu.Unlock()
}
r.done <- err
return
case <-ticker.C:
if r.isClosed() {
close(r.writeCh)
err := <-r.writerDone
if err != nil {
r.mu.Lock()
r.metrics.EncoderWriteErrors++
r.mu.Unlock()
}
r.done <- err
return
}
frame := r.nextMixedFrame(time.Now())
if frame == nil {
frame = silence
}
r.queueMixedFrame(frame)
}
}
}
func mixFrame(sources []chan []byte) []byte {
var mixed []int32
used := false
func (r *recorder) isClosed() bool {
r.mu.Lock()
defer r.mu.Unlock()
return r.closed
}
for _, ch := range sources {
var latest []byte
for {
select {
case b := <-ch:
latest = b
default:
goto drained
func (r *recorder) nextMixedFrame(now time.Time) []byte {
r.mu.Lock()
frames := make([][]byte, 0, len(r.sources))
for _, source := range r.sources {
select {
case frame := <-source.ch:
if len(frame) >= bytesPerFrame {
frames = append(frames, frame)
}
default:
if now.Sub(source.lastFrame) >= sourceStaleAfter && !source.staleLogged {
source.staleLogged = true
r.metrics.StaleSourceTicks++
log.Printf("Recording %s: source has no decoded frames for %v",
filepath.Base(r.path), sourceStaleAfter)
}
}
drained:
if len(latest) < bytesPerFrame {
continue
}
r.metrics.MixerFrames++
r.mu.Unlock()
return mixPCMFrames(frames)
}
func (r *recorder) queueMixedFrame(frame []byte) {
select {
case r.writeCh <- frame:
default:
r.mu.Lock()
r.metrics.DroppedMixedFrames++
r.mu.Unlock()
log.Printf("Recording %s: encoder queue full, dropping mixed frame",
filepath.Base(r.path))
}
}
func (r *recorder) writeEncoder() {
for frame := range r.writeCh {
if _, err := r.stdin.Write(frame); err != nil {
r.writerDone <- err
return
}
}
r.writerDone <- nil
}
func mixFrame(sources []chan []byte) []byte {
frames := make([][]byte, 0, len(sources))
for _, ch := range sources {
var frame []byte
select {
case frame = <-ch:
default:
}
if len(frame) >= bytesPerFrame {
frames = append(frames, frame)
}
}
return mixPCMFrames(frames)
}
func mixPCMFrames(frames [][]byte) []byte {
if len(frames) == 0 {
return nil
}
var mixed []int32
gain := 1.0
if len(frames) > 1 {
gain = 1 / math.Sqrt(float64(len(frames)))
}
for _, frame := range frames {
if mixed == nil {
mixed = make([]int32, samplesPerFrame*channels)
}
for i := range mixed {
sample := int16(binary.LittleEndian.Uint16(latest[i*2:]))
sample := int16(binary.LittleEndian.Uint16(frame[i*2:]))
mixed[i] += int32(sample)
}
used = true
}
if !used {
return nil
}
out := make([]byte, bytesPerFrame)
for i, sample := range mixed {
if sample > 32767 {
sample = 32767
} else if sample < -32768 {
sample = -32768
}
sample = int32(math.Round(float64(sample) * gain))
sample = limitSample(sample)
binary.LittleEndian.PutUint16(out[i*2:], uint16(int16(sample)))
}
return out
}
func limitSample(sample int32) int32 {
if sample > limiterThreshold {
sample = limiterThreshold + (sample-limiterThreshold)/limiterRatio
} else if sample < -limiterThreshold {
sample = -limiterThreshold + (sample+limiterThreshold)/limiterRatio
}
if sample > 32767 {
return 32767
}
if sample < -32768 {
return -32768
}
return sample
}
func (r *recorder) Close() error {
r.mu.Lock()
if r.closed {
@@ -362,6 +543,18 @@ func (r *recorder) Close() error {
stdinErr := r.stdin.Close()
waitErr := r.cmd.Wait()
activeRecordings.Delete(r.path)
metrics := r.snapshotMetrics()
log.Printf("Recording %s: closed after %v, sources=%d, decoders=%d/%d, source drops=%d, stale ticks=%d, mixed drops=%d, encoder errors=%d",
filepath.Base(r.path),
metrics.Duration,
metrics.ActiveSources,
metrics.DecoderStarts,
metrics.DecoderStops,
metrics.DroppedSourceFrames,
metrics.StaleSourceTicks,
metrics.DroppedMixedFrames,
metrics.EncoderWriteErrors,
)
if err != nil {
return err
@@ -379,6 +572,91 @@ func (r *recorder) Close() error {
return nil
}
func (r *recorder) snapshotMetrics() RecorderMetrics {
r.mu.Lock()
defer r.mu.Unlock()
metrics := r.metrics
metrics.Duration = time.Since(metrics.Started)
metrics.ActiveSources = len(r.sources)
return metrics
}
func RunSyntheticRecording(directory string, sources int, duration time.Duration) (SyntheticRecordingResult, error) {
if sources <= 0 {
return SyntheticRecordingResult{}, errors.New("sources must be greater than zero")
}
if duration <= 0 {
return SyntheticRecordingResult{}, errors.New("duration must be greater than zero")
}
if err := os.MkdirAll(directory, 0700); err != nil {
return SyntheticRecordingResult{}, err
}
r, err := newRecorder(nil, directory)
if err != nil {
return SyntheticRecordingResult{}, err
}
var wg sync.WaitGroup
for i := 0; i < sources; i++ {
track := &diskTrack{}
if r.addSource(track) == nil {
r.Close()
return SyntheticRecordingResult{}, errors.New("recorder is closed")
}
r.noteDecoderStarted(track)
wg.Add(1)
go func(index int, t *diskTrack) {
defer wg.Done()
defer r.removeSource(t)
ticker := time.NewTicker(frameDuration)
defer ticker.Stop()
timer := time.NewTimer(duration)
defer timer.Stop()
frequency := 220 + float64(index)*37
phase := 0.0
for {
select {
case <-timer.C:
return
case <-ticker.C:
r.pushSourceFrame(t, sinePCMFrame(frequency, &phase))
}
}
}(i, track)
}
wg.Wait()
if err := r.Close(); err != nil {
return SyntheticRecordingResult{}, err
}
return SyntheticRecordingResult{
Path: r.path,
Metrics: r.snapshotMetrics(),
}, nil
}
func sinePCMFrame(frequency float64, phase *float64) []byte {
frame := make([]byte, bytesPerFrame)
step := 2 * math.Pi * frequency / sampleRate
for i := 0; i < samplesPerFrame; i++ {
sample := int16(math.Sin(*phase) * 6000)
*phase += step
if *phase >= 2*math.Pi {
*phase -= 2 * math.Pi
}
offset := i * channels * 2
for ch := 0; ch < channels; ch++ {
binary.LittleEndian.PutUint16(
frame[offset+ch*2:], uint16(sample),
)
}
}
return frame
}
type diskConn struct {
client *Client
remote conn.Up
@@ -486,8 +764,8 @@ func (t *diskTrack) Write(buf []byte) (int, error) {
}
func (t *diskTrack) start(payload uint8) error {
ch := t.conn.client.recorder.addSource(t)
if ch == nil {
source := t.conn.client.recorder.addSource(t)
if source == nil {
return errors.New("recorder is closed")
}
@@ -560,25 +838,26 @@ func (t *diskTrack) start(payload uint8) error {
t.rtpConn = rtpConn
t.cmd = cmd
t.stdin = stdin
t.ch = ch
t.ch = source.ch
t.conn.client.recorder.noteDecoderStarted(t)
go drainStderr("ffmpeg decoder", stderr)
go t.readPCM(stdout, ch)
go t.readPCM(stdout)
return nil
}
func (t *diskTrack) readPCM(r io.Reader, ch chan []byte) {
func (t *diskTrack) readPCM(r io.Reader) {
for {
buf := make([]byte, bytesPerFrame)
_, err := io.ReadFull(r, buf)
if err != nil {
if t.conn.client.recorder != nil {
t.conn.client.recorder.noteDecoderStopped(t, err)
}
return
}
select {
case ch <- buf:
default:
<-ch
ch <- buf
if t.conn.client.recorder != nil {
t.conn.client.recorder.pushSourceFrame(t, buf)
}
}
}
diskwriter/diskwriter_test.go
+109 -3
View File
@@ -103,7 +103,7 @@ func TestCleanupOldRecordings(t *testing.T) {
}
}
func TestMixFrameClipsAndSumsSources(t *testing.T) {
func TestMixFrameAppliesHeadroomForMultipleSources(t *testing.T) {
a := make(chan []byte, 1)
b := make(chan []byte, 1)
a <- pcmFrame(20000)
@@ -116,12 +116,79 @@ func TestMixFrameClipsAndSumsSources(t *testing.T) {
for i := 0; i < len(out); i += 2 {
got := int16(binary.LittleEndian.Uint16(out[i:]))
if got != 32767 {
t.Fatalf("sample %d = %d, expected clipped 32767", i/2, got)
if got != 28071 {
t.Fatalf("sample %d = %d, expected headroom-adjusted 28071", i/2, got)
}
}
}
func TestMixFrameLimiterPreservesNormalSpeechLevel(t *testing.T) {
a := make(chan []byte, 1)
a <- pcmFrame(12000)
out := mixFrame([]chan []byte{a})
for i := 0; i < len(out); i += 2 {
got := int16(binary.LittleEndian.Uint16(out[i:]))
if got != 12000 {
t.Fatalf("sample %d = %d, expected unchanged 12000", i/2, got)
}
}
}
func TestMixFrameHandlesQuietAndLoudSources(t *testing.T) {
a := make(chan []byte, 1)
b := make(chan []byte, 1)
a <- pcmFrame(1000)
b <- pcmFrame(20000)
out := mixFrame([]chan []byte{a, b})
for i := 0; i < len(out); i += 2 {
got := int16(binary.LittleEndian.Uint16(out[i:]))
if got != 14849 {
t.Fatalf("sample %d = %d, expected mixed quiet/loud sample 14849", i/2, got)
}
}
}
func TestMixFrameLimiterBoundsOverloadedSamples(t *testing.T) {
a := make(chan []byte, 1)
b := make(chan []byte, 1)
a <- pcmFrame(32000)
b <- pcmFrame(32000)
out := mixFrame([]chan []byte{a, b})
for i := 0; i < len(out); i += 2 {
got := int16(binary.LittleEndian.Uint16(out[i:]))
if got <= 0 || got > 32767 {
t.Fatalf("sample %d = %d, expected bounded positive sample", i/2, got)
}
if got >= 32767 {
t.Fatalf("sample %d = %d, expected limiter below hard clip", i/2, got)
}
}
}
func TestMixFrameConsumesOldestQueuedFrame(t *testing.T) {
source := make(chan []byte, 2)
source <- pcmFrame(1000)
source <- pcmFrame(2000)
out := mixFrame([]chan []byte{source})
if len(out) != bytesPerFrame {
t.Fatalf("mixed frame has length %d, expected %d", len(out), bytesPerFrame)
}
for i := 0; i < len(out); i += 2 {
got := int16(binary.LittleEndian.Uint16(out[i:]))
if got != 1000 {
t.Fatalf("sample %d = %d, expected oldest queued sample 1000", i/2, got)
}
}
if queued := len(source); queued != 1 {
t.Fatalf("queued frames after mix = %d, expected 1", queued)
}
}
func TestMixFrameReturnsNilForSilence(t *testing.T) {
if out := mixFrame(nil); out != nil {
t.Fatalf("expected nil silent frame, got %d bytes", len(out))
@@ -162,6 +229,45 @@ func TestRecorderCreatesPlayableOgg(t *testing.T) {
}
}
func TestSyntheticRecordingCreatesPlayableOgg(t *testing.T) {
if _, err := exec.LookPath("ffmpeg"); err != nil {
t.Skip("ffmpeg not installed")
}
if _, err := exec.LookPath("ffprobe"); err != nil {
t.Skip("ffprobe not installed")
}
result, err := RunSyntheticRecording(t.TempDir(), 3, 100*time.Millisecond)
if err != nil {
t.Fatal(err)
}
if result.Metrics.DecoderStarts != 3 {
t.Fatalf("decoder starts = %d, expected 3", result.Metrics.DecoderStarts)
}
if result.Metrics.DecoderStops != 3 {
t.Fatalf("decoder stops = %d, expected 3", result.Metrics.DecoderStops)
}
if result.Metrics.DroppedMixedFrames != 0 {
t.Fatalf("dropped mixed frames = %d, expected 0", result.Metrics.DroppedMixedFrames)
}
cmd := exec.Command(
"ffprobe",
"-v", "error",
"-select_streams", "a:0",
"-show_entries", "stream=codec_name",
"-of", "default=noprint_wrappers=1:nokey=1",
result.Path,
)
out, err := cmd.Output()
if err != nil {
t.Fatal(err)
}
if strings.TrimSpace(string(out)) != "opus" {
t.Fatalf("expected opus stream, got %q", strings.TrimSpace(string(out)))
}
}
func pcmFrame(sample int16) []byte {
out := make([]byte, bytesPerFrame)
for i := 0; i < len(out); i += 2 {