diff --git a/cmd/mxl-player/main.go b/cmd/mxl-player/main.go index 6984573..f7b7cf0 100644 --- a/cmd/mxl-player/main.go +++ b/cmd/mxl-player/main.go @@ -6,16 +6,12 @@ import ( "fmt" "io" "log" - mxladapter "mxl-player/internal/adapter/mxl" "mxl-player/internal/imgui" - "mxl-player/internal/output" "mxl-player/internal/playback" "mxl-player/internal/renderer" "mxl-player/internal/sdl" - "mxl-player/internal/source" "os" "runtime" - "sync" "time" "unsafe" @@ -170,22 +166,6 @@ func main() { if !args.ListAudio && !args.ListGPU { checkMXLargs(args) } - if args.SyncRequested && - args.VideoFlowId != "" && - args.AudioFlowId != "" && - args.VideoDomain != args.AudioDomain { - fmt.Fprintln( - os.Stderr, - "--sync currently requires video and audio to use the same MXL domain", - ) - os.Exit(2) - } - // path selection - useLegacySync := args.SyncRequested && - args.VideoFlowId != "" && - args.AudioFlowId != "" - useIndependentSlots := !useLegacySync - runtime.LockOSThread() if err := sdl.Load(); err != nil { panic(err) @@ -289,77 +269,6 @@ func main() { } defer vkDevice.Destroy() - var ( - syncSrc *source.SyncSource - videoSrc *source.Source - audioSrc *source.AudioSource - audioStream uintptr - audioBatch uint64 - aChans uint64 - ) - - interleaveAudio := func(samples [][]byte) []byte { - frameBytes := int(audioBatch) * int(aChans) * 4 - out := make([]byte, frameBytes) - for ch := uint64(0); ch < aChans; ch++ { - srcBytes := samples[ch] - for i := uint64(0); i < audioBatch; i++ { - srcOff := i * 4 - dstOff := (i*aChans + ch) * 4 - if srcOff+4 <= uint64(len(srcBytes)) { - copy(out[dstOff:dstOff+4], srcBytes[srcOff:srcOff+4]) - } - } - } - return out - } - - switch { - case useLegacySync: - syncSrc, err = source.OpenSameDomainSync( - args.VideoDomain, - args.VideoFlowId, - args.AudioFlowId, - ) - if err != nil { - log.Fatalf("sync source: %v", err) - } - aChans = syncSrc.Channels() - audioBatch = uint64(syncSrc.AudioRate().Num) / uint64(syncSrc.Rate().Num) - if audioBatch == 0 { - audioBatch = 1 - } - audioStream = sdl.OpenAudioDeviceStream(sdlAudioDevice, sdl.AudioSpec{ - Format: sdl.AudioF32, - Channels: int32(aChans), - Freq: int32(syncSrc.AudioRate().Num / syncSrc.AudioRate().Den), - }) - if audioStream == 0 { - log.Fatalf("audio: %s", sdl.GetError()) - } - sdl.ResumeAudioStreamDevice(audioStream) - fmt.Printf("sync: video %dx%d audio %dch batch=%d\n", - syncSrc.Width(), syncSrc.Height(), aChans, audioBatch) - default: - // Independent slots own their readers. - // Empty startup opens nothing. - } - - defer func() { - if syncSrc != nil { - _ = syncSrc.Close() - } - if videoSrc != nil { - _ = videoSrc.Close() - } - if audioSrc != nil { - _ = audioSrc.Close() - } - }() - if audioStream != 0 { - defer sdl.DestroyAudioStream(audioStream) - } - // Create renderer r, err := renderer.New(renderer.Config{ PhysDevice: vkPhysDevice, @@ -414,419 +323,58 @@ func main() { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - type reconnectParams struct { - domain string - video string - audio string - } - // Reconnect requests from GUI or automatic retry - control := make(chan reconnectParams, 1) - videoBridge := playback.NewVideoBridge() - statusStore := playback.NewStatusStore() - - videoWorker, err := playback.NewVideoWorker( - mxladapter.VideoFactory{}, - videoBridge, - retryPolicy, - mxladapter.ShouldRetry, - func(status playback.Status) { - statusStore.Observe(status) - if status.Err != nil { - log.Printf( - "video: state=%v attempt=%d failed=%d: %v", - status.State, - status.Attempt, - status.FailedAttempts, - status.Err, - ) - return - } - log.Printf( - "video: state=%v attempt=%d failed=%d", - status.State, - status.Attempt, - status.FailedAttempts, - ) - }, - ) + player, err := newPlayerPlayback(sdlAudioDevice, retryPolicy) if err != nil { panic(err) } - videoSlot, err := playback.NewVideoSlot(videoWorker) - if err != nil { - panic(err) - } - videoCommands := make(chan playback.FeedConfig, 1) + videoBridge := player.Video + statusStore := player.Status + syncRequested := args.SyncRequested - audioOutput := output.NewSDLAudioSink(sdlAudioDevice) - defer audioOutput.Close() - audioWorker, err := playback.NewAudioWorker( - mxladapter.AudioFactory{}, - audioOutput, - retryPolicy, - mxladapter.ShouldRetry, - func(status playback.Status) { - statusStore.Observe(status) - - if status.Err != nil { - log.Printf( - "audio: state=%v attempt=%d failed=%d: %v", - status.State, - status.Attempt, - status.FailedAttempts, - status.Err, - ) - return - } - - log.Printf( - "audio: state=%v attempt=%d failed=%d", - status.State, - status.Attempt, - status.FailedAttempts, - ) - }, - ) - if err != nil { - panic(err) - } - audioSlot, err := playback.NewAudioSlot(audioWorker) - if err != nil { - panic(err) - } - audioCommands := make(chan playback.FeedConfig, 1) - - reopen := func(params reconnectParams) error { - // Close current sources - if syncSrc != nil { - _ = syncSrc.Close() - syncSrc = nil - } - if videoSrc != nil { - _ = videoSrc.Close() - videoSrc = nil - } - if audioSrc != nil { - _ = audioSrc.Close() - audioSrc = nil - } - // Try once. Return error if fails — caller loops back to select - // and can pick up newer reconnect request. - if params.video != "" && params.audio != "" { - s, e := source.OpenSameDomainSync(params.domain, params.video, params.audio) - if e == nil { - syncSrc = s - aChans = s.Channels() - audioBatch = uint64(s.AudioRate().Num) / uint64(s.Rate().Num) - if audioBatch == 0 { - audioBatch = 1 - } - return nil - } - return e - } else if params.video != "" { - s, e := source.Open(params.domain, params.video) - if e == nil { - videoSrc = s - return nil - } - return e - } else if params.audio != "" { - s, e := source.OpenAudio(params.domain, params.audio) - if e == nil { - audioSrc = s - aChans = s.Channels() - audioBatch = uint64(s.Rate().Num) / (100 * uint64(s.Rate().Den)) - if audioBatch == 0 { - audioBatch = 1 - } - return nil - } - return e - } - return fmt.Errorf("reopen: no flow specified") - } - - enqueueVideoConfig := func(config playback.FeedConfig) { + enqueueCommand := func(command playback.SessionCommand) { select { - case <-videoCommands: - default: - } - - select { - case videoCommands <- config: + case player.Commands <- command: default: + log.Printf("playback command queue is full; ignoring command %d", command.Kind) } } - enqueueAudioConfig := func(config playback.FeedConfig) { - select { - case <-audioCommands: - default: - } - - select { - case audioCommands <- config: - default: - } - } - doReconnect := func() { - if useIndependentSlots { - videoActive = videoStr != "" - audioActive = audioStr != "" + videoActive = videoStr != "" + audioActive = audioStr != "" - videoConfig := playback.FeedConfig{} - if videoStr != "" { - videoConfig = playback.FeedConfig{ + if videoStr == "" { + enqueueCommand(playback.SessionCommand{Kind: playback.CommandRemoveVideo}) + } else { + enqueueCommand(playback.SessionCommand{ + Kind: playback.CommandSetVideo, + Config: playback.FeedConfig{ Domain: videoDomainStr, UUID: videoStr, Active: true, - } - } - - audioConfig := playback.FeedConfig{} - if audioStr != "" { - audioConfig = playback.FeedConfig{ + }, + }) + } + if audioStr == "" { + enqueueCommand(playback.SessionCommand{Kind: playback.CommandRemoveAudio}) + } else { + enqueueCommand(playback.SessionCommand{ + Kind: playback.CommandSetAudio, + Config: playback.FeedConfig{ Domain: audioDomainStr, UUID: audioStr, Active: true, - } - } - - enqueueVideoConfig(videoConfig) - enqueueAudioConfig(audioConfig) - return + }, + }) } - // legacy - if videoDomainStr != audioDomainStr { - log.Printf( - "sync reconnect rejected: video domain %q differs from audio domain %q", - videoDomainStr, - audioDomainStr, - ) - return - } - select { - case <-control: - default: - } - control <- reconnectParams{domain: videoDomainStr, video: videoStr, audio: audioStr} } - playbackDone := make(chan struct{}) + playbackDone := make(chan error, 1) go func() { - defer close(playbackDone) - - if useIndependentSlots { - var slots sync.WaitGroup - slots.Add(2) - - go func() { - defer slots.Done() - - err := videoSlot.Run( - ctx, - playback.FeedConfig{ - Domain: args.VideoDomain, - UUID: args.VideoFlowId, - Active: args.VideoFlowId != "", - }, - videoCommands, - ) - if err != nil && !errors.Is(err, context.Canceled) { - log.Printf("video slot: %v", err) - } - }() - - go func() { - defer slots.Done() - - err := audioSlot.Run( - ctx, - playback.FeedConfig{ - Domain: args.AudioDomain, - UUID: args.AudioFlowId, - Active: args.AudioFlowId != "", - }, - audioCommands, - ) - if err != nil && !errors.Is(err, context.Canceled) { - log.Printf("audio slot: %v", err) - } - }() - - <-ctx.Done() - slots.Wait() - return - } - - // Audio-only mode: independent loop. - if audioSrc != nil && syncSrc == nil && videoSrc == nil { - for { - select { - case <-ctx.Done(): - return - case params := <-control: - if params.video != "" || params.audio != "" { - if rerr := reopen(params); rerr != nil { - if errors.Is(rerr, context.Canceled) { - return - } - log.Printf("source: reopen failed: %v, retrying", rerr) - select { - case <-time.After(500 * time.Millisecond): - case <-ctx.Done(): - return - } - select { - case control <- params: - default: - } - } - } - continue - default: - } - queued := sdl.GetAudioStreamQueued(audioStream) - maxQueued := int32(audioBatch) * int32(aChans) * 4 * 20 - if queued > maxQueued { - select { - case <-time.After(10 * time.Millisecond): - case <-ctx.Done(): - return - } - continue - } - f, err := audioSrc.NextAudio(ctx, audioBatch, 20*time.Millisecond) - if err != nil { - if errors.Is(err, context.Canceled) { - return - } - log.Printf("source: %v", err) - params := reconnectParams{domain: videoDomainStr, video: videoStr, audio: audioStr} - select { - case <-control: - default: - } - select { - case <-time.After(500 * time.Millisecond): - case <-ctx.Done(): - return - } - select { - case control <- params: - default: - } - continue - } - if f.Samples != nil && audioStream != 0 { - sdl.PutAudioStreamData(audioStream, interleaveAudio(f.Samples)) - } - } - } - - // Video bridge provides backpressure: only one borrowed frame is in flight. - for { - select { - case <-ctx.Done(): - return - - case params := <-control: - if rerr := reopen(params); rerr != nil { - if errors.Is(rerr, context.Canceled) { - return - } - log.Printf("source: reopen failed: %v, retrying", rerr) - - select { - case <-time.After(500 * time.Millisecond): - case <-ctx.Done(): - return - } - - select { - case control <- params: - default: - // Preserve an already queued, potentially newer request. - } - } - continue - - default: - } - - var videoFrame playback.VideoFrame - if syncSrc != nil { - vFrame, aFrame, err := syncSrc.NextSync(ctx, audioBatch, 200*time.Millisecond) - if err != nil { - if errors.Is(err, context.Canceled) { - return - } - log.Printf("source: %v", err) - // Request a reconnect after the read failure. - params := reconnectParams{ - domain: videoDomainStr, - video: videoStr, - audio: audioStr, - } - select { - case control <- params: - default: - // Preserve an already queued, potentially newer request. - } - continue - } - videoFrame = playback.VideoFrame{ - Index: vFrame.Index, - Width: vFrame.Width, - Height: vFrame.Height, - Stride: vFrame.Stride, - Size: vFrame.Size, - Invalid: vFrame.Invalid, - Payload: vFrame.Payload, - } - if aFrame.Samples != nil && audioStream != 0 { - sdl.PutAudioStreamData(audioStream, interleaveAudio(aFrame.Samples)) - } - - } else if videoSrc != nil { - f, err := videoSrc.NextCtx(ctx, 200*time.Millisecond) - if err != nil { - if errors.Is(err, context.Canceled) { - return - } - log.Printf("source: %v", err) - params := reconnectParams{ - domain: videoDomainStr, - video: videoStr, - audio: audioStr, - } - select { - case control <- params: - default: - // Preserve an already queued, potentially newer request. - } - continue - } - videoFrame = playback.VideoFrame{ - Index: f.Index, - Width: f.Width, - Height: f.Height, - Stride: f.Stride, - Size: f.Size, - Invalid: f.Invalid, - Payload: f.Payload, - } - } - - if err := videoBridge.ConsumeVideo(ctx, videoFrame); err != nil { - if errors.Is(err, context.Canceled) { - return - } - log.Printf("video output: %v", err) - return - } - } + playbackDone <- player.Controller.Run( + ctx, + args.playbackConfig(), + player.Commands, + ) }() running := true @@ -979,136 +527,109 @@ func main() { cimgui.InputTextWithHint("Audio domain", "/dev/shm/mxl", &audioDomainStr, 0, nil) cimgui.InputTextWithHint("Video UUID", "", &videoStr, 0, nil) cimgui.InputTextWithHint("Audio UUID", "", &audioStr, 0, nil) + if snapshot, ok := player.Controller.Snapshot(); ok { + videoActive = snapshot.Desired.Video.Active + audioActive = snapshot.Desired.Audio.Active + syncRequested = snapshot.Desired.SyncRequested + } if cimgui.Button("Connect") { doReconnect() } cimgui.SameLine() cimgui.Checkbox("Show stats", &showStats) - if useIndependentSlots && videoActive { + if cimgui.Checkbox("Synchronize", &syncRequested) { + kind := playback.CommandDisableSync + if syncRequested { + kind = playback.CommandEnableSync + } + enqueueCommand(playback.SessionCommand{Kind: kind}) + } + if videoActive { if cimgui.Button("Stop video") { videoActive = false - enqueueVideoConfig( - playback.FeedConfig{ - Domain: videoDomainStr, - UUID: videoStr, - Active: false, - }) + enqueueCommand(playback.SessionCommand{Kind: playback.CommandStopVideo}) } } - if useIndependentSlots && !videoActive && videoStr != "" { + if !videoActive && videoStr != "" { cimgui.SameLine() if cimgui.Button("Resume video") { videoActive = true - enqueueVideoConfig(playback.FeedConfig{ - Domain: videoDomainStr, - UUID: videoStr, - Active: true, - }) + enqueueCommand(playback.SessionCommand{Kind: playback.CommandResumeVideo}) } } - if useIndependentSlots && videoStr != "" { + if videoStr != "" { if cimgui.Button("Remove video") { videoActive = false videoStr = "" - enqueueVideoConfig(playback.FeedConfig{}) + enqueueCommand(playback.SessionCommand{Kind: playback.CommandRemoveVideo}) } } - if useIndependentSlots { - if videoActive { - cimgui.Text("Video desired: active") - } else if videoStr != "" { - cimgui.Text("Video desired: stopped") - } else { - cimgui.Text("Video desired: not configured") - } - - if status, ok := statusStore.Snapshot(playback.UnitVideo); ok { - cimgui.Text(fmt.Sprintf( - "Video actual: %s", - status.State, - )) - cimgui.Text(fmt.Sprintf( - "Attempt: %d, failed: %d", - status.Attempt, - status.FailedAttempts, - )) - - if status.RetryIn > 0 { - cimgui.Text(fmt.Sprintf( - "Retry in: %s", - status.RetryIn.Round(time.Millisecond), - )) - } - - if status.Err != nil { - cimgui.TextWrapped(status.Err.Error()) - } - } else { - cimgui.Text("Video actual: not started") - } + if videoActive { + cimgui.Text("Video desired: active") + } else if videoStr != "" { + cimgui.Text("Video desired: stopped") + } else { + cimgui.Text("Video desired: not configured") } - if useIndependentSlots && audioActive { + + if status, ok := statusStore.Snapshot(playback.UnitVideo); ok { + cimgui.Text(fmt.Sprintf("Video actual: %s", status.State)) + cimgui.Text(fmt.Sprintf("Attempt: %d, failed: %d", status.Attempt, status.FailedAttempts)) + if status.RetryIn > 0 { + cimgui.Text(fmt.Sprintf("Retry in: %s", status.RetryIn.Round(time.Millisecond))) + } + if status.Err != nil { + cimgui.TextWrapped(status.Err.Error()) + } + } else { + cimgui.Text("Video actual: not started") + } + if audioActive { if cimgui.Button("Stop audio") { audioActive = false - enqueueAudioConfig(playback.FeedConfig{ - Domain: audioDomainStr, - UUID: audioStr, - Active: false, - }) + enqueueCommand(playback.SessionCommand{Kind: playback.CommandStopAudio}) } } - if useIndependentSlots && !audioActive && audioStr != "" { + if !audioActive && audioStr != "" { if cimgui.Button("Resume audio") { audioActive = true - enqueueAudioConfig(playback.FeedConfig{ - Domain: audioDomainStr, - UUID: audioStr, - Active: true, - }) + enqueueCommand(playback.SessionCommand{Kind: playback.CommandResumeAudio}) } } - if useIndependentSlots && audioStr != "" { + if audioStr != "" { if cimgui.Button("Remove audio") { audioActive = false audioStr = "" - enqueueAudioConfig(playback.FeedConfig{}) + enqueueCommand(playback.SessionCommand{Kind: playback.CommandRemoveAudio}) } } - if useIndependentSlots { - if audioActive { - cimgui.Text("Audio desired: active") - } else if audioStr != "" { - cimgui.Text("Audio desired: stopped") - } else { - cimgui.Text("Audio desired: not configured") + if audioActive { + cimgui.Text("Audio desired: active") + } else if audioStr != "" { + cimgui.Text("Audio desired: stopped") + } else { + cimgui.Text("Audio desired: not configured") + } + + if status, ok := statusStore.Snapshot(playback.UnitAudio); ok { + cimgui.Text(fmt.Sprintf("Audio actual: %s", status.State)) + cimgui.Text(fmt.Sprintf("Attempt: %d, failed: %d", status.Attempt, status.FailedAttempts)) + if status.RetryIn > 0 { + cimgui.Text(fmt.Sprintf("Retry in: %s", status.RetryIn.Round(time.Millisecond))) } - - if status, ok := statusStore.Snapshot(playback.UnitAudio); ok { - cimgui.Text(fmt.Sprintf( - "Audio actual: %s", - status.State, - )) - cimgui.Text(fmt.Sprintf( - "Attempt: %d, failed: %d", - status.Attempt, - status.FailedAttempts, - )) - - if status.RetryIn > 0 { - cimgui.Text(fmt.Sprintf( - "Retry in: %s", - status.RetryIn.Round(time.Millisecond), - )) - } - - if status.Err != nil { - cimgui.TextWrapped(status.Err.Error()) - } - } else { - cimgui.Text("Audio actual: not started") + if status.Err != nil { + cimgui.TextWrapped(status.Err.Error()) + } + } else { + cimgui.Text("Audio actual: not started") + } + if status, ok := statusStore.Snapshot(playback.UnitSync); ok { + cimgui.Text(fmt.Sprintf("Sync actual: %s", status.State)) + if status.Err != nil { + cimgui.TextWrapped(status.Err.Error()) } } @@ -1141,5 +662,10 @@ func main() { } cancel() - <-playbackDone + if err := <-playbackDone; err != nil && !errors.Is(err, context.Canceled) { + log.Printf("playback controller: %v", err) + } + if err := player.Close(); err != nil { + log.Printf("close playback: %v", err) + } } diff --git a/imgui.ini b/imgui.ini index e3ffc14..564fc96 100644 --- a/imgui.ini +++ b/imgui.ini @@ -14,7 +14,7 @@ Size=200,200 Collapsed=0 [Window][Connection] -Pos=1322,937 -Size=613,164 +Pos=322,387 +Size=618,275 Collapsed=0 diff --git a/internal/adapter/mxl/sync.go b/internal/adapter/mxl/sync.go index d45dfc4..c8194b2 100644 --- a/internal/adapter/mxl/sync.go +++ b/internal/adapter/mxl/sync.go @@ -13,8 +13,7 @@ import ( ) const ( - DefaultSyncReadTimeout = 200 * time.Millisecond - DefaultSyncBatchDuration = 10 * time.Millisecond + DefaultSyncReadTimeout = 200 * time.Millisecond ) var ErrNativeSyncDifferentDomains = errors.New( @@ -22,15 +21,13 @@ var ErrNativeSyncDifferentDomains = errors.New( ) type SyncFactory struct { - ReadTimeout time.Duration - BatchDuration time.Duration - open func(string, string, string) (localSyncSource, error) + ReadTimeout time.Duration + open func(string, string, string) (localSyncSource, error) } type localSyncSource interface { NextSync( context.Context, - uint64, time.Duration, ) (source.Frame, source.AudioFrame, error) @@ -41,7 +38,6 @@ type localSyncSource interface { type syncReader struct { source localSyncSource readTimeout time.Duration - audioBatch uint64 rateNumerator int64 rateDenominator int64 } @@ -118,25 +114,11 @@ func (f SyncFactory) OpenSync( if readTimeout <= 0 { readTimeout = DefaultSyncReadTimeout } - batchDuration := f.BatchDuration - if batchDuration <= 0 { - batchDuration = DefaultSyncBatchDuration - } audioRate := src.AudioRate() - batch, err := audioBatchSize(audioRate.Num, audioRate.Den, batchDuration) - if err != nil { - _ = src.Close() - return nil, &source.SourceError{ - Op: "calculate sync audio batch", - Kind: source.ErrorKindInvalidConfig, - Err: err, - } - } return &syncReader{ source: src, readTimeout: readTimeout, - audioBatch: batch, rateNumerator: audioRate.Num, rateDenominator: audioRate.Den, }, nil @@ -145,7 +127,7 @@ func (f SyncFactory) OpenSync( func (r *syncReader) ReadSync( ctx context.Context, ) (playback.SyncFrame, error) { - video, audio, err := r.source.NextSync(ctx, r.audioBatch, r.readTimeout) + video, audio, err := r.source.NextSync(ctx, r.readTimeout) if err != nil { return playback.SyncFrame{}, err } diff --git a/internal/adapter/mxl/sync_test.go b/internal/adapter/mxl/sync_test.go index d91f341..f859207 100644 --- a/internal/adapter/mxl/sync_test.go +++ b/internal/adapter/mxl/sync_test.go @@ -17,7 +17,6 @@ type fakeLocalSyncSource struct { audio source.AudioFrame readErr error rate mxl.Rational - batch uint64 timeout time.Duration closed bool closeError error @@ -25,10 +24,8 @@ type fakeLocalSyncSource struct { func (s *fakeLocalSyncSource) NextSync( _ context.Context, - batch uint64, timeout time.Duration, ) (source.Frame, source.AudioFrame, error) { - s.batch = batch s.timeout = timeout return s.video, s.audio, s.readErr } @@ -79,23 +76,8 @@ func TestSyncFactoryUsesDefaultsAndForwardsFeeds(t *testing.T) { t.Fatalf("open args = %q %q %q", domain, videoUUID, audioUUID) } got := reader.(*syncReader) - if got.readTimeout != DefaultSyncReadTimeout || got.audioBatch != 480 { - t.Fatalf("reader timeout=%s batch=%d, want %s and 480", got.readTimeout, got.audioBatch, DefaultSyncReadTimeout) - } -} - -func TestSyncFactoryClosesSourceForInvalidAudioRate(t *testing.T) { - fake := &fakeLocalSyncSource{rate: mxl.Rational{}} - factory := SyncFactory{open: func(string, string, string) (localSyncSource, error) { - return fake, nil - }} - video, audio := syncFeedConfigs() - reader, err := factory.OpenSync(context.Background(), video, audio) - if reader != nil { - t.Fatal("OpenSync() reader is not nil") - } - if !errors.Is(err, ErrInvalidAudioBatch) || !fake.closed { - t.Fatalf("OpenSync() error=%v closed=%t", err, fake.closed) + if got.readTimeout != DefaultSyncReadTimeout { + t.Fatalf("reader timeout=%s, want %s", got.readTimeout, DefaultSyncReadTimeout) } } @@ -108,7 +90,7 @@ func TestSyncReaderConvertsPairWithoutCopying(t *testing.T) { rate: mxl.Rational{Num: 48_000, Den: 1}, } reader := &syncReader{ - source: fake, readTimeout: 7 * time.Millisecond, audioBatch: 12, + source: fake, readTimeout: 7 * time.Millisecond, rateNumerator: 48_000, rateDenominator: 1, } @@ -116,8 +98,8 @@ func TestSyncReaderConvertsPairWithoutCopying(t *testing.T) { if err != nil { t.Fatal(err) } - if fake.batch != 12 || fake.timeout != 7*time.Millisecond { - t.Fatalf("NextSync() batch=%d timeout=%s", fake.batch, fake.timeout) + if fake.timeout != 7*time.Millisecond { + t.Fatalf("NextSync() timeout=%s", fake.timeout) } if frame.Video.Index != 10 || frame.Audio.Index != 40 || frame.Audio.SampleRateNumerator != 48_000 { t.Fatalf("frame = %+v", frame) diff --git a/internal/source/source.go b/internal/source/source.go index 1be29f2..c1215c5 100644 --- a/internal/source/source.go +++ b/internal/source/source.go @@ -598,8 +598,10 @@ func (s *SyncSource) Close() error { return s.inst.Close() } -// NextSync reads both at a synced timestamp. Returns video Frame + audio AudioFrame -func (s *SyncSource) NextSync(ctx context.Context, audioBatch uint64, timeout time.Duration) (Frame, AudioFrame, error) { +// NextSync reads one video frame and the audio interval between this video +// timestamp and the next. Deriving the interval for every frame preserves +// exact long-term timing for fractional video rates. +func (s *SyncSource) NextSync(ctx context.Context, timeout time.Duration) (Frame, AudioFrame, error) { var timeouts int for { select { @@ -635,7 +637,30 @@ func (s *SyncSource) NextSync(ctx context.Context, audioBatch uint64, timeout ti } // read audio at the same timestamp aIdx := mxl.TimestampToIndex(s.aRate, ts) + nextTimestamp := mxl.IndexToTimestamp(s.rate, s.idx+1) + nextAudioIndex := mxl.TimestampToIndex(s.aRate, nextTimestamp) + if nextAudioIndex <= aIdx { + return Frame{}, AudioFrame{}, wrapError( + "calculate synchronized audio interval", + ErrorKindInvalidConfig, + fmt.Errorf("invalid audio interval: %d..%d", aIdx, nextAudioIndex), + ) + } + audioBatch := nextAudioIndex - aIdx av, aerr := s.ar.GetSamples(aIdx, int(audioBatch), 50*time.Millisecond) + if aerr != nil { + kind := ErrorKindUnavailable + if errors.Is(aerr, mxl.ErrTimeout) || + errors.Is(aerr, mxl.ErrOutOfRangeEarly) || + errors.Is(aerr, mxl.ErrOutOfRangeLate) { + kind = ErrorKindTemporary + } + return Frame{}, AudioFrame{}, wrapError( + "read synchronized audio", + kind, + aerr, + ) + } vFrame := Frame{ Index: g.Index, Width: s.width, Height: s.height, Stride: s.stride, Size: g.GrainSize, @@ -643,23 +668,19 @@ func (s *SyncSource) NextSync(ctx context.Context, audioBatch uint64, timeout ti } s.idx++ - var aFrame AudioFrame - if aerr == nil { - samples := make([][]byte, s.chans) - for ch := uint64(0); ch < s.chans; ch++ { - f1, f2, _ := av.ChannelFragments(ch) - if len(f2) > 0 { - samples[ch] = append(f1, f2...) - } else { - samples[ch] = f1 - } - } - aFrame = AudioFrame{ - Index: aIdx, SampleCount: audioBatch, - Channels: s.chans, Samples: samples, + samples := make([][]byte, s.chans) + for ch := uint64(0); ch < s.chans; ch++ { + f1, f2, _ := av.ChannelFragments(ch) + if len(f2) > 0 { + samples[ch] = append(f1, f2...) + } else { + samples[ch] = f1 } } - // even if audio failed, video returns + aFrame := AudioFrame{ + Index: aIdx, SampleCount: audioBatch, + Channels: s.chans, Samples: samples, + } return vFrame, aFrame, nil case errors.Is(err, mxl.ErrTimeout), errors.Is(err, mxl.ErrOutOfRangeEarly), errors.Is(err, mxl.ErrOutOfRangeLate): timeouts++