diff --git a/cmd/mxl-player/main.go b/cmd/mxl-player/main.go index 4cd42b4..eca8a0b 100644 --- a/cmd/mxl-player/main.go +++ b/cmd/mxl-player/main.go @@ -25,19 +25,22 @@ const ( ) func main() { - videoFlowId := "5fbec3b1-1b0f-417d-9059-8b94a47197ed" - audioFlowId := "5fbec3b1-1b0f-417d-9059-8b94a47197ec" + // video: 5fbec3b1-1b0f-417d-9059-8b94a47197ed + // audio: 5fbec3b1-1b0f-417d-9059-8b94a47197ec + var videoFlowId, audioFlowId string mxlDomain := flag.String("d", "/dev/shm/mxl", "MXL domain") mxlVideoFlowID := flag.String("v", videoFlowId, "MXL video flow UUID") mxlAudioFlowID := flag.String("a", audioFlowId, "MXL audio flow UUID") - fmt.Println(*mxlAudioFlowID) flag.Parse() + if *mxlVideoFlowID == "" && *mxlAudioFlowID == "" { + log.Fatal("need at least -v or -a ") + } runtime.LockOSThread() if err := sdl.Load(); err != nil { panic(err) } - if !sdl.Init(sdl.InitVideo) { + if !sdl.Init(sdl.InitVideo | sdl.InitAudio) { log.Fatalf("SDL_Init: %s", sdl.GetError()) return } @@ -108,29 +111,123 @@ func main() { } defer vkDevice.Destroy() - mxlSrc, err := source.Open(*mxlDomain, *mxlVideoFlowID) - if err != nil { - log.Fatalf("source: %v", err) - } - defer func() { _ = mxlSrc.Close() }() - fmt.Printf("source: %dx%d stride=%d\n", mxlSrc.Width(), mxlSrc.Height(), mxlSrc.Stride()) + var ( + syncSrc *source.SyncSource + videoSrc *source.Source + audioSrc *source.AudioSource + audioStream uintptr + audioBatch uint64 + aChans uint64 + ) - r, err := renderer.New(renderer.Config{ - PhysDevice: vkPhysDevice, - Device: vkDevice, - Queue: vkQueue, - Surface: vkSurf, - Window: windowHandler, - GraphicsFamily: gfx, - VideoWidth: mxlSrc.Width(), - VideoHeight: mxlSrc.Height(), - VideoStride: mxlSrc.Stride(), - }) - if err != nil { - panic(err) + 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 *mxlVideoFlowID != "" && *mxlAudioFlowID != "": + syncSrc, err = source.OpenSync(*mxlDomain, *mxlVideoFlowID, *mxlAudioFlowID) + 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(sdl.AudioDeviceDefaultPlayback, 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) + + case *mxlVideoFlowID != "": + videoSrc, err = source.Open(*mxlDomain, *mxlVideoFlowID) + if err != nil { + log.Fatalf("source: %v", err) + } + fmt.Printf("video: %dx%d stride=%d\n", videoSrc.Width(), videoSrc.Height(), videoSrc.Stride()) + + default: + audioSrc, err = source.OpenAudio(*mxlDomain, *mxlAudioFlowID) + if err != nil { + log.Fatalf("audio source: %v", err) + } + aChans = audioSrc.Channels() + audioBatch = uint64(audioSrc.Rate().Num) / (100 * uint64(audioSrc.Rate().Den)) + if audioBatch == 0 { + audioBatch = 1 + } + audioStream = sdl.OpenAudioDeviceStream(sdl.AudioDeviceDefaultPlayback, sdl.AudioSpec{ + Format: sdl.AudioF32, + Channels: int32(aChans), + Freq: int32(audioSrc.Rate().Num / audioSrc.Rate().Den), + }) + if audioStream == 0 { + log.Fatalf("audio: %s", sdl.GetError()) + } + sdl.ResumeAudioStreamDevice(audioStream) + fmt.Printf("audio: %dch %d/%d Hz\n", aChans, audioSrc.Rate().Num, audioSrc.Rate().Den) + } + + defer func() { + if syncSrc != nil { + _ = syncSrc.Close() + } + if videoSrc != nil { + _ = videoSrc.Close() + } + if audioSrc != nil { + _ = audioSrc.Close() + } + }() + if audioStream != 0 { + defer sdl.DestroyAudioStream(audioStream) + } + + var r *renderer.Renderer + if *mxlVideoFlowID != "" { + var w, h, stride uint32 + if syncSrc != nil { + w, h, stride = syncSrc.Width(), syncSrc.Height(), syncSrc.Stride() + } else { + w, h, stride = videoSrc.Width(), videoSrc.Height(), videoSrc.Stride() + } + r, err = renderer.New(renderer.Config{ + PhysDevice: vkPhysDevice, + Device: vkDevice, + Queue: vkQueue, + Surface: vkSurf, + Window: windowHandler, + GraphicsFamily: gfx, + VideoWidth: w, + VideoHeight: h, + VideoStride: stride, + }) + if err != nil { + panic(err) + } + defer r.Destroy() + defer vkDevice.WaitIdle() } - defer r.Destroy() - defer vkDevice.WaitIdle() ctx, cancel := context.WithCancel(context.Background()) defer cancel() @@ -139,8 +236,14 @@ func main() { failed := make(chan struct{}) reopen := func() error { - if err := mxlSrc.Close(); err != nil { - return fmt.Errorf("close old source: %w", err) + if syncSrc != nil { + _ = syncSrc.Close() + } + if videoSrc != nil { + _ = videoSrc.Close() + } + if audioSrc != nil { + _ = audioSrc.Close() } for { select { @@ -148,56 +251,166 @@ func main() { return ctx.Err() default: } - s, err := source.Open(*mxlDomain, *mxlVideoFlowID) - if err == nil { - newSize := vk.DeviceSize(s.Stride()) * vk.DeviceSize(s.Height()) - if newSize != r.FrameSize() { - if err := r.RecreateBuffers(newSize); err != nil { - return err + if syncSrc != nil { + s, e := source.OpenSync(*mxlDomain, *mxlVideoFlowID, *mxlAudioFlowID) + if e == nil { + if r != nil { + newSize := vk.DeviceSize(s.Stride()) * vk.DeviceSize(s.Height()) + if newSize != r.FrameSize() { + if e = r.RecreateBuffers(newSize); e != nil { + return e + } + } } - fmt.Printf("source: resolution changed, frameSize=%d\n", newSize) + syncSrc = s + return nil } - mxlSrc = s - return nil + log.Printf("source: reopen retry: %v", e) + } else if videoSrc != nil { + s, e := source.Open(*mxlDomain, *mxlVideoFlowID) + if e == nil { + if r != nil { + newSize := vk.DeviceSize(s.Stride()) * vk.DeviceSize(s.Height()) + if newSize != r.FrameSize() { + if e = r.RecreateBuffers(newSize); e != nil { + return e + } + } + } + videoSrc = s + return nil + } + log.Printf("source: reopen retry: %v", e) + } else if audioSrc != nil { + s, e := source.OpenAudio(*mxlDomain, *mxlAudioFlowID) + if e == nil { + audioSrc = s + return nil + } + log.Printf("source: reopen retry: %v", e) } - log.Printf("source: reopen retry: %v", err) time.Sleep(500 * time.Millisecond) } } go func() { + // Audio-only mode: independent loop, no grant/staged handshake. + if audioSrc != nil && syncSrc == nil && videoSrc == nil { + for { + select { + case <-ctx.Done(): + return + 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 + } + if errors.Is(err, mxl.ErrFlowInvalid) { + log.Printf("source: flow invalid, reopening") + if rerr := reopen(); rerr != nil { + log.Printf("source: reopen failed: %v", rerr) + cancel() + return + } + continue + } + log.Printf("source: %v", err) + cancel() + return + } + if f.Samples != nil && audioStream != 0 { + sdl.PutAudioStreamData(audioStream, interleaveAudio(f.Samples)) + } + } + } + + // Video (with or without sync) mode: grant/staged handshake. for { select { case <-grant: case <-ctx.Done(): return } - f, err := mxlSrc.NextCtx(ctx, 200*time.Millisecond) - if err != nil { - if errors.Is(err, context.Canceled) { + + var payload []byte + var grainIdx uint64 + + if syncSrc != nil { + vFrame, aFrame, err := syncSrc.NextSync(ctx, audioBatch, 200*time.Millisecond) + if err != nil { + if errors.Is(err, context.Canceled) { + return + } + if errors.Is(err, mxl.ErrFlowInvalid) { + log.Printf("source: flow invalid, reopening") + if rerr := reopen(); rerr != nil { + log.Printf("source: reopen failed: %v", rerr) + cancel() + return + } + select { + case failed <- struct{}{}: + case <-ctx.Done(): + return + } + continue + } + log.Printf("source: %v", err) + cancel() return } - if errors.Is(err, mxl.ErrFlowInvalid) { - log.Printf("source: flow invalid, reopening") - if rerr := reopen(); rerr != nil { - log.Printf("source: reopen failed: %v", rerr) - cancel() - return - } - select { - case failed <- struct{}{}: - case <-ctx.Done(): - return - } - continue + payload = vFrame.Payload + grainIdx = vFrame.Index + if aFrame.Samples != nil && audioStream != 0 { + sdl.PutAudioStreamData(audioStream, interleaveAudio(aFrame.Samples)) } - log.Printf("source: %v", err) - cancel() - return + + } else if videoSrc != nil { + f, err := videoSrc.NextCtx(ctx, 200*time.Millisecond) + if err != nil { + if errors.Is(err, context.Canceled) { + return + } + if errors.Is(err, mxl.ErrFlowInvalid) { + log.Printf("source: flow invalid, reopening") + if rerr := reopen(); rerr != nil { + log.Printf("source: reopen failed: %v", rerr) + cancel() + return + } + select { + case failed <- struct{}{}: + case <-ctx.Done(): + return + } + continue + } + log.Printf("source: %v", err) + cancel() + return + } + payload = f.Payload + grainIdx = f.Index + + } + + if r != nil { + vk.CopyToMapped(r.StagingMapped(), payload) } - vk.CopyToMapped(r.StagingMapped(), f.Payload) select { - case staged <- f.Index: + case staged <- grainIdx: case <-ctx.Done(): return } @@ -239,7 +452,7 @@ func main() { if !running { break } - if resized { + if resized && r != nil { if err := r.RecreateSwapchain(); err != nil { if errors.Is(err, renderer.ErrMinimized) { resized = true @@ -272,36 +485,46 @@ func main() { continue } - err := r.DrawFrame(mxlSrc.Width(), mxlSrc.Height(), mxlSrc.Stride()) - if errors.Is(err, renderer.ErrOutOfDate) { - if rerr := r.RecreateSwapchain(); rerr != nil { - if errors.Is(rerr, renderer.ErrMinimized) { - resized = true - continue + if r != nil { + var w, h, stride uint32 + if syncSrc != nil { + w, h, stride = syncSrc.Width(), syncSrc.Height(), syncSrc.Stride() + } else if videoSrc != nil { + w, h, stride = videoSrc.Width(), videoSrc.Height(), videoSrc.Stride() + } + err := r.DrawFrame(w, h, stride) + if errors.Is(err, renderer.ErrOutOfDate) { + if rerr := r.RecreateSwapchain(); rerr != nil { + if errors.Is(rerr, renderer.ErrMinimized) { + resized = true + continue + } + panic(rerr) } - panic(rerr) + continue + } + if err != nil { + panic(err) } - continue - } - if err != nil { - panic(err) - } - if lastIndex != 0 && shownIndex > lastIndex { - if g := shownIndex - lastIndex - 1; g > 0 { - dropped += g + if lastIndex != 0 && shownIndex > lastIndex { + if g := shownIndex - lastIndex - 1; g > 0 { + dropped += g + } } - } - lastIndex = shownIndex - frameCount++ - if now := time.Now(); now.Sub(lastReport) >= time.Second { - dt := now.Sub(lastReport).Seconds() - fps := float64(frameCount) / dt - fmt.Printf("fps=%.1f dropped=%d idx=%d frameTime=%.2fms\n", - fps, dropped, shownIndex, float64(now.Sub(frameStart).Microseconds())/1000.0) - frameCount = 0 - dropped = 0 - lastReport = now + lastIndex = shownIndex + frameCount++ + if now := time.Now(); now.Sub(lastReport) >= time.Second { + dt := now.Sub(lastReport).Seconds() + fps := float64(frameCount) / dt + fmt.Printf("fps=%.1f dropped=%d idx=%d frameTime=%.2fms\n", + fps, dropped, shownIndex, float64(now.Sub(frameStart).Microseconds())/1000.0) + frameCount = 0 + dropped = 0 + lastReport = now + } + } else { + time.Sleep(10 * time.Millisecond) } } } diff --git a/internal/source/source.go b/internal/source/source.go index ed26a6f..beb0f1e 100644 --- a/internal/source/source.go +++ b/internal/source/source.go @@ -71,6 +71,11 @@ func Open(domain, flowID string) (*Source, error) { inst.Close() return nil, fmt.Errorf("parse flow def: %w", err) } + if fd.FrameWidth == 0 || fd.FrameHeight == 0 { + r.Close() + inst.Close() + return nil, fmt.Errorf("flow has no video dimensions (not a video flow?)") + } rate := info.Config.Common.GrainRate idx := mxl.CurrentIndex(rate) if idx == mxl.UndefinedIndex { @@ -270,3 +275,199 @@ func (s *AudioSource) Close() error { func (s *AudioSource) Rate() mxl.Rational { return s.rate } func (s *AudioSource) Channels() uint64 { return s.chans } + +type SyncSource struct { + inst *mxl.Instance + vr *mxl.Reader + ar *mxl.Reader + group *mxl.SyncGroup + rate mxl.Rational // video rate + aRate mxl.Rational + chans uint64 + idx uint64 + width, height, stride uint32 +} + +func OpenSync(domain, videoFlow, audioFlow string) (*SyncSource, error) { + inst, err := mxl.NewInstance(domain, "") + if err != nil { + return nil, fmt.Errorf("NewInstance: %w", err) + } + + vr, err := inst.NewReader(videoFlow) + if err != nil { + inst.Close() + return nil, fmt.Errorf("NewReader(video): %w", err) + } + ar, err := inst.NewReader(audioFlow) + if err != nil { + vr.Close() + inst.Close() + return nil, fmt.Errorf("NewReader(audio): %w", err) + } + + vInfo, err := vr.Info() + if err != nil { + ar.Close() + vr.Close() + inst.Close() + return nil, fmt.Errorf("Info(video): %w", err) + } + if !vInfo.Config.Common.Format.IsDiscrete() { + ar.Close() + vr.Close() + inst.Close() + return nil, fmt.Errorf("video flow is not discrete") + } + aInfo, err := ar.Info() + if err != nil { + ar.Close() + vr.Close() + inst.Close() + return nil, fmt.Errorf("Info(audio): %w", err) + } + if aInfo.Config.Common.Format.IsDiscrete() { + ar.Close() + vr.Close() + inst.Close() + return nil, fmt.Errorf("audio flow is not continuous") + } + + def, err := inst.FlowDef(videoFlow) + if err != nil { + ar.Close() + vr.Close() + inst.Close() + return nil, fmt.Errorf("FlowDef: %w", err) + } + var fd flowDef + if err := json.Unmarshal([]byte(def), &fd); err != nil { + ar.Close() + vr.Close() + inst.Close() + return nil, fmt.Errorf("parse flow def: %w", err) + } + + vRate := vInfo.Config.Common.GrainRate + idx := mxl.CurrentIndex(vRate) + if idx == mxl.UndefinedIndex { + ar.Close() + vr.Close() + inst.Close() + return nil, fmt.Errorf("invalid video rate %d/%d", vRate.Num, vRate.Den) + } + + group, err := inst.NewSyncGroup() + if err != nil { + ar.Close() + vr.Close() + inst.Close() + return nil, fmt.Errorf("NewSyncGroup: %w", err) + } + if err := group.AddReader(vr); err != nil { + group.Close() + ar.Close() + vr.Close() + inst.Close() + return nil, fmt.Errorf("AddReader(video): %w", err) + } + if err := group.AddReader(ar); err != nil { + group.Close() + ar.Close() + vr.Close() + inst.Close() + return nil, fmt.Errorf("AddReader(audio): %w", err) + } + + return &SyncSource{ + inst: inst, + vr: vr, + ar: ar, + group: group, + rate: vRate, + aRate: aInfo.Config.Common.GrainRate, + chans: uint64(aInfo.Config.Continuous.ChannelCount), + idx: idx, + width: uint32(fd.FrameWidth), + height: uint32(fd.FrameHeight), + stride: vInfo.Config.Discrete.SliceSizes[0], + }, nil +} + +func (s *SyncSource) Close() error { + _ = s.group.Close() + _ = s.ar.Close() + _ = s.vr.Close() + 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) { + for { + select { + case <-ctx.Done(): + return Frame{}, AudioFrame{}, ctx.Err() + default: + } + + ts := mxl.IndexToTimestamp(s.rate, s.idx) + err := s.group.WaitForDataAt(ts, timeout) + switch { + case err == nil: + // read video + g, gerr := s.vr.GetGrain(s.idx, 50*time.Millisecond) + if gerr != nil { + if errors.Is(gerr, mxl.ErrFlowInvalid) { + s.idx = mxl.CurrentIndex(s.rate) + continue + } + return Frame{}, AudioFrame{}, fmt.Errorf("GetGraing: %w", gerr) + } + // read audio at the same timestamp + aIdx := mxl.TimestampToIndex(s.aRate, ts) + av, aerr := s.ar.GetSamples(aIdx, int(audioBatch), 50*time.Millisecond) + vFrame := Frame{ + Index: g.Index, Width: s.width, Height: s.height, + Stride: s.stride, Size: g.GrainSize, + Invalid: g.Invalid(), Payload: g.Payload, + } + 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, + } + } + // even if audio failed, video returns + return vFrame, aFrame, nil + case errors.Is(err, mxl.ErrTimeout), errors.Is(err, mxl.ErrOutOfRangeEarly): + select { + case <-time.After(5 * time.Millisecond): + case <-ctx.Done(): + return Frame{}, AudioFrame{}, ctx.Err() + } + case errors.Is(err, mxl.ErrOutOfRangeLate): + s.idx = mxl.CurrentIndex(s.rate) + default: + return Frame{}, AudioFrame{}, fmt.Errorf("WaitForDataAt: %w", err) + } + } +} + +func (s *SyncSource) Width() uint32 { return s.width } +func (s *SyncSource) Height() uint32 { return s.height } +func (s *SyncSource) Stride() uint32 { return s.stride } +func (s *SyncSource) Rate() mxl.Rational { return s.rate } +func (s *SyncSource) AudioRate() mxl.Rational { return s.aRate } +func (s *SyncSource) Channels() uint64 { return s.chans }