Compare commits

...

2 Commits

Author SHA1 Message Date
Dmitry Sergeev a473c84b0e dropped-frame tracke 2026-09-01 23:45:10 +03:00
Dmitry Sergeev 179768ca4a playlist retry-exhaustion handling 2026-09-01 23:35:41 +03:00
18 changed files with 545 additions and 56 deletions
+10 -8
View File
@@ -467,7 +467,7 @@ func main() {
displayedVideoStride uint32 = placeholderStride
fps float64
lastIndex uint64
dropTracker videoDropTracker
dropped uint64
droppedTotal uint64
frameCount uint64
@@ -530,6 +530,8 @@ func main() {
resized = false
}
var shownIndex uint64
var shownGeneration uint64
var shownSource playback.FeedConfig
hasFrame := false
frameCtx, frameCancel := context.WithTimeout(ctx, 100*time.Millisecond)
@@ -555,6 +557,8 @@ func main() {
}
shownIndex = pendingFrame.Frame.Index
shownGeneration = pendingFrame.Generation
shownSource = pendingFrame.Source
displayedVideoWidth = pendingFrame.Frame.Width
displayedVideoHeight = pendingFrame.Frame.Height
displayedVideoStride = pendingFrame.Frame.Stride
@@ -565,15 +569,14 @@ func main() {
panic(frameErr)
}
snapshot, hasSnapshot := player.Controller.Snapshot()
// stats
if hasFrame {
if lastIndex != 0 && shownIndex > lastIndex {
if g := shownIndex - lastIndex - 1; g > 0 {
dropped += g
droppedTotal += g
if gap := dropTracker.Observe(shownGeneration, shownSource, shownIndex); gap > 0 {
dropped += gap
droppedTotal += gap
}
}
lastIndex = shownIndex
frameCount++
if now := time.Now(); now.Sub(lastReport) >= time.Second {
dt := now.Sub(lastReport).Seconds()
@@ -588,7 +591,6 @@ func main() {
// end of stats
if r != nil {
gui.BeginFrame(time.Since(lastFrame), int32(r.Extent().Width), int32(r.Extent().Height))
snapshot, hasSnapshot := player.Controller.Snapshot()
if showStats {
cimgui.SetNextWindowPos(cimgui.Vec2{X: 0, Y: 0})
cimgui.SetNextWindowSize(cimgui.Vec2{X: 460, Y: 510})
+11
View File
@@ -13,6 +13,7 @@ import (
type playlistFile struct {
Entries []playlistFileEntry `json:"entries"`
Loop bool `json:"loop"`
OnFailure string `json:"on_failure"`
}
type playlistFileEntry struct {
@@ -62,6 +63,16 @@ func decodePlaylistFile(reader io.Reader) (playback.Playlist, error) {
Entries: make([]playback.PlaylistEntry, len(file.Entries)),
Loop: file.Loop,
}
switch file.OnFailure {
case "", "wait":
playlist.OnFailure = playback.PlaylistFailureWait
case "next":
playlist.OnFailure = playback.PlaylistFailureNext
default:
return playback.Playlist{}, fmt.Errorf(
"on_failure %q: %w", file.OnFailure, playback.ErrPlaylistFailurePolicy,
)
}
for index, entry := range file.Entries {
duration := time.Duration(0)
if entry.Duration != "" {
+4 -1
View File
@@ -14,6 +14,7 @@ import (
func TestDecodePlaylistFile(t *testing.T) {
input := `{
"loop": true,
"on_failure": "next",
"entries": [
{
"name": "sync",
@@ -45,6 +46,7 @@ func TestDecodePlaylistFile(t *testing.T) {
}
want := playback.Playlist{
Loop: true,
OnFailure: playback.PlaylistFailureNext,
Entries: []playback.PlaylistEntry{
{
Name: "sync",
@@ -69,7 +71,7 @@ func TestDecodePlaylistFile(t *testing.T) {
},
},
}
if len(got.Entries) != len(want.Entries) || got.Loop != want.Loop {
if len(got.Entries) != len(want.Entries) || got.Loop != want.Loop || got.OnFailure != want.OnFailure {
t.Fatalf("decodePlaylistFile() = %#v, want %#v", got, want)
}
for index := range want.Entries {
@@ -100,6 +102,7 @@ func TestDecodePlaylistFileRejectsInvalidInput(t *testing.T) {
{name: "malformed JSON", input: `{"entries": [`, wantText: "decode JSON"},
{name: "unknown field", input: `{"unknown": true}`, wantText: "unknown field"},
{name: "multiple roots", input: `{"entries": []} {"entries": []}`, wantText: "multiple root values"},
{name: "invalid failure policy", input: `{"on_failure":"skip","entries":[]}`, wantErr: playback.ErrPlaylistFailurePolicy},
{
name: "invalid duration",
input: `{"entries":[{"video":{"domain":"/video","uuid":"video"},"duration":"later"}]}`,
+37
View File
@@ -0,0 +1,37 @@
package main
import "mxl-player/internal/playback"
// videoDropTracker counts gaps only within one playback generation. Frame
// indices belong to their source and cannot be compared across feed changes.
type videoDropTracker struct {
generation uint64
source playback.FeedConfig
lastIndex uint64
hasIndex bool
}
func (t *videoDropTracker) Observe(
generation uint64,
source playback.FeedConfig,
index uint64,
) uint64 {
if !t.hasIndex ||
generation != t.generation ||
!sameVideoSource(source, t.source) ||
index <= t.lastIndex {
t.generation = generation
t.source = source
t.lastIndex = index
t.hasIndex = true
return 0
}
dropped := index - t.lastIndex - 1
t.lastIndex = index
return dropped
}
func sameVideoSource(a, b playback.FeedConfig) bool {
return a.Domain == b.Domain && a.UUID == b.UUID
}
+71
View File
@@ -0,0 +1,71 @@
package main
import (
"testing"
"mxl-player/internal/playback"
)
func TestVideoDropTracker(t *testing.T) {
tests := []struct {
name string
observations [][2]uint64
want []uint64
}{
{
name: "counts gaps within generation",
observations: [][2]uint64{{1, 10}, {1, 11}, {1, 15}},
want: []uint64{0, 0, 3},
},
{
name: "higher index from new generation resets baseline",
observations: [][2]uint64{{1, 10}, {2, 1000000}, {2, 1000001}},
want: []uint64{0, 0, 0},
},
{
name: "lower index from new generation resets baseline",
observations: [][2]uint64{{1, 100}, {2, 5}, {2, 7}},
want: []uint64{0, 0, 1},
},
{
name: "index restart within generation resets baseline",
observations: [][2]uint64{{1, 100}, {1, 0}, {1, 1}},
want: []uint64{0, 0, 0},
},
{
name: "zero is a valid first index",
observations: [][2]uint64{{1, 0}, {1, 2}},
want: []uint64{0, 1},
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
var tracker videoDropTracker
source := playback.FeedConfig{Domain: "/mxl", UUID: "video"}
for index, observation := range test.observations {
got := tracker.Observe(observation[0], source, observation[1])
if got != test.want[index] {
t.Fatalf("Observe(%d, %d) = %d, want %d",
observation[0], observation[1], got, test.want[index])
}
}
})
}
}
func TestVideoDropTrackerResetsWhenSourceChanges(t *testing.T) {
var tracker videoDropTracker
first := playback.FeedConfig{Domain: "/mxl", UUID: "first"}
second := playback.FeedConfig{Domain: "/mxl", UUID: "second"}
if got := tracker.Observe(1, first, 10); got != 0 {
t.Fatalf("first Observe() = %d, want 0", got)
}
if got := tracker.Observe(1, second, 1000000); got != 0 {
t.Fatalf("source-changing Observe() = %d, want 0", got)
}
if got := tracker.Observe(1, second, 1000002); got != 1 {
t.Fatalf("same-source Observe() = %d, want 1", got)
}
}
+41
View File
@@ -0,0 +1,41 @@
{
"description": "sample for mxl reader go player",
"id": "5fbec3b1-1b0f-417d-9059-8b94a47197ed",
"tags": {
"urn:x-nmos:tag:grouphint/v1.0": [
"mxl-gst-testsrc pattern"
]
},
"format": "urn:x-nmos:format:video",
"label": "SMPTE bars test video",
"parents": [],
"media_type": "video/v210",
"grain_rate": {
"numerator": 25,
"denominator": 1
},
"frame_width": 1920,
"frame_height": 1080,
"interlace_mode": "progressive",
"colorspace": "BT709",
"components": [
{
"name": "Y",
"width": 1920,
"height": 1080,
"bit_depth": 10
},
{
"name": "Cb",
"width": 960,
"height": 1080,
"bit_depth": 10
},
{
"name": "Cr",
"width": 960,
"height": 1080,
"bit_depth": 10
}
]
}
+3 -28
View File
@@ -1,30 +1,6 @@
[Window][Debug##Default]
Pos=519,181
Size=400,398
Collapsed=0
[Window][Test]
Pos=60,60
Size=251,92
Collapsed=0
[Window][Stats]
Size=460,510
Collapsed=0
[Window][Connection]
Pos=42,264
Size=605,416
Collapsed=0
[Window][Test slider]
Pos=1370,0
Size=550,1080
Collapsed=0
[Window][Settings]
Pos=730,0
Size=550,720
Size=400,400
Collapsed=0
[Window][Settings & Info]
@@ -32,8 +8,7 @@ Pos=580,0
Size=700,720
Collapsed=0
[Window][Seetings & Info]
Pos=1220,0
Size=700,1080
[Window][Stats]
Size=460,510
Collapsed=0
+30
View File
@@ -11,8 +11,34 @@ var (
ErrPlaylistEntryEmpty = errors.New("playlist entry must contain at least one feed")
ErrPlaylistSyncFeedsRequired = errors.New("synchronized playlist entry requires both video and audio feeds")
ErrPlaylistDurationNegative = errors.New("playlist entry duration cannot be negative")
ErrPlaylistFailurePolicy = errors.New("invalid playlist failure policy")
)
type PlaylistFailurePolicy uint8
const (
PlaylistFailureWait PlaylistFailurePolicy = iota
PlaylistFailureNext
)
func (p PlaylistFailurePolicy) String() string {
switch p {
case PlaylistFailureWait:
return "wait"
case PlaylistFailureNext:
return "next"
default:
return fmt.Sprintf("PlaylistFailurePolicy(%d)", uint8(p))
}
}
func (p PlaylistFailurePolicy) Validate() error {
if p != PlaylistFailureWait && p != PlaylistFailureNext {
return ErrPlaylistFailurePolicy
}
return nil
}
type PlaylistFeed struct {
Domain string
UUID string
@@ -29,6 +55,7 @@ type PlaylistEntry struct {
type Playlist struct {
Entries []PlaylistEntry
Loop bool
OnFailure PlaylistFailurePolicy
}
func (f PlaylistFeed) IsConfigured() bool {
@@ -65,6 +92,9 @@ func (e PlaylistEntry) Validate() error {
}
func (p Playlist) Validate() error {
if err := p.OnFailure.Validate(); err != nil {
return err
}
for index, entry := range p.Entries {
if err := entry.Validate(); err != nil {
return fmt.Errorf("playlist entry %d: %w", index, err)
+47 -1
View File
@@ -7,10 +7,22 @@ import (
"time"
)
type PlaylistReadiness struct {
type PlaylistEventKind uint8
const (
PlaylistEventReady PlaylistEventKind = iota
PlaylistEventFailed
)
type PlaylistEvent struct {
Revision uint64
Kind PlaylistEventKind
}
// PlaylistReadiness is retained as an alias for callers that only publish
// ready events. Its zero Kind is PlaylistEventReady.
type PlaylistReadiness = PlaylistEvent
type playlistTimer interface {
C() <-chan time.Time
Stop() bool
@@ -162,6 +174,40 @@ func (c *PlaylistController) Run(
readiness = nil
continue
}
if ready.Kind == PlaylistEventFailed {
if ready.Revision != revision {
continue
}
stopTimer()
// A failed entry must not retain a live or apparently active
// duration clock, even when the policy is to wait.
timing = NewPlaylistTiming(revision, timing.Duration)
if c.playlist.OnFailure != PlaylistFailureNext {
c.publish(state, revision, timing)
continue
}
next, sessionCommand, apply, err := ApplyPlaylistSelection(
c.playlist,
state,
PlaylistCommand{Kind: PlaylistNext},
c.retry,
)
if err != nil || !apply {
c.publish(state, revision, timing)
continue
}
select {
case <-ctx.Done():
return ctx.Err()
case c.sessions <- sessionCommand:
}
revision++
state = next
entry, _ := next.Entry(c.playlist)
timing = NewPlaylistTiming(revision, entry.Duration)
c.publish(state, revision, timing)
continue
}
if timing.Paused &&
ready.Revision == timing.Revision &&
timing.Duration > 0 &&
+181
View File
@@ -0,0 +1,181 @@
package playback
import (
"context"
"testing"
"time"
)
func TestIsSessionFailed(t *testing.T) {
const generation = 7
video := FeedConfig{Domain: "/video", UUID: "video", Active: true}
audio := FeedConfig{Domain: "/audio", UUID: "audio", Active: true}
failedFeed := func(unit Unit, feed FeedConfig) Status {
return Status{
Unit: unit, State: StateFailed, Generation: generation, Feed: feed,
}
}
independent := SessionSnapshot{
Generation: generation,
Plan: SessionPlan{
Topology: TopologyIndependent,
Video: video,
Audio: audio,
},
}
synchronized := SessionSnapshot{
Generation: generation,
Plan: SessionPlan{
Topology: TopologySynchronized,
Sync: SyncPairConfig{Video: video, Audio: audio},
},
}
tests := []struct {
name string
session SessionSnapshot
statuses PlaybackStatusSnapshot
want bool
}{
{
name: "video failure in independent pair",
session: independent,
statuses: PlaybackStatusSnapshot{
Generation: generation,
Video: failedFeed(UnitVideo, video), HasVideo: true,
},
want: true,
},
{
name: "audio failure in independent pair",
session: independent,
statuses: PlaybackStatusSnapshot{
Generation: generation,
Audio: failedFeed(UnitAudio, audio), HasAudio: true,
},
want: true,
},
{
name: "sync failure",
session: synchronized,
statuses: PlaybackStatusSnapshot{
Generation: generation,
Sync: Status{
Unit: UnitSync, State: StateFailed, Generation: generation,
Pair: SyncPairConfig{Video: video, Audio: audio},
},
HasSync: true,
},
want: true,
},
{
name: "stale snapshot generation",
session: independent,
statuses: PlaybackStatusSnapshot{
Generation: generation - 1,
Video: failedFeed(UnitVideo, video), HasVideo: true,
},
},
{
name: "wrong source",
session: independent,
statuses: PlaybackStatusSnapshot{
Generation: generation,
Video: failedFeed(UnitVideo, FeedConfig{
Domain: "/video", UUID: "other", Active: true,
}),
HasVideo: true,
},
},
{
name: "inactive failed unit is ignored",
session: SessionSnapshot{
Generation: generation,
Plan: SessionPlan{
Topology: TopologyIndependent,
Video: video,
Audio: FeedConfig{Domain: audio.Domain, UUID: audio.UUID},
},
},
statuses: PlaybackStatusSnapshot{
Generation: generation,
Audio: failedFeed(UnitAudio, audio), HasAudio: true,
},
},
{
name: "reconnecting has not exhausted retries",
session: independent,
statuses: PlaybackStatusSnapshot{
Generation: generation,
Video: Status{
Unit: UnitVideo, State: StateReconnecting,
Generation: generation, Feed: video,
},
HasVideo: true,
},
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
if got := IsSessionFailed(test.session, test.statuses); got != test.want {
t.Fatalf("IsSessionFailed() = %v, want %v", got, test.want)
}
})
}
}
func TestPlaylistControllerFailurePolicy(t *testing.T) {
for _, test := range []struct {
name string
policy PlaylistFailurePolicy
wantAdvance bool
}{
{name: "wait", policy: PlaylistFailureWait},
{name: "next", policy: PlaylistFailureNext, wantAdvance: true},
} {
t.Run(test.name, func(t *testing.T) {
playlist := navigationPlaylist(false)
playlist.OnFailure = test.policy
sessions := make(chan SessionCommand, 4)
controller, err := NewPlaylistController(
playlist, validPlaylistRetryPolicy(), sessions,
)
if err != nil {
t.Fatalf("NewPlaylistController() error = %v", err)
}
commands := make(chan PlaylistCommand, 2)
events := make(chan PlaylistReadiness, 2)
ctx, cancel := context.WithCancel(context.Background())
result := make(chan error, 1)
go func() { result <- controller.Run(ctx, commands, events) }()
commands <- PlaylistCommand{Kind: PlaylistSelect, Index: 0}
<-sessions
events <- PlaylistEvent{Revision: 1, Kind: PlaylistEventFailed}
if test.wantAdvance {
select {
case <-sessions:
case <-time.After(time.Second):
t.Fatal("failure did not advance playlist")
}
snapshot, _ := controller.Snapshot()
if snapshot.State.CurrentIndex != 1 || snapshot.Revision != 2 {
t.Fatalf("snapshot = %#v, want index 1 revision 2", snapshot)
}
} else {
select {
case command := <-sessions:
t.Fatalf("unexpected session command: %#v", command)
case <-time.After(20 * time.Millisecond):
}
}
cancel()
if err := <-result; err != context.Canceled {
t.Fatalf("Run() error = %v, want context canceled", err)
}
})
}
}
+60 -8
View File
@@ -89,7 +89,8 @@ func (c *PlaylistReadinessCoordinator) Run(ctx context.Context) error {
ticker := c.newTicker(c.interval)
defer ticker.Stop()
var emittedRevision uint64
var emittedReadyRevision uint64
var emittedFailedRevision uint64
for {
select {
case <-ctx.Done():
@@ -99,11 +100,7 @@ func (c *PlaylistReadinessCoordinator) Run(ctx context.Context) error {
playlistSnapshot, ok := c.playlist.Snapshot()
if !ok ||
!playlistSnapshot.State.HasSelection ||
playlistSnapshot.Revision == 0 ||
playlistSnapshot.Entry.Duration <= 0 ||
playlistSnapshot.Timing.Started ||
playlistSnapshot.Timing.Paused ||
playlistSnapshot.Revision == emittedRevision {
playlistSnapshot.Revision == 0 {
continue
}
@@ -114,7 +111,30 @@ func (c *PlaylistReadinessCoordinator) Run(ctx context.Context) error {
) {
continue
}
if !IsSessionPlaying(sessionSnapshot, c.statuses.SnapshotAll()) {
statuses := c.statuses.SnapshotAll()
if IsSessionFailed(sessionSnapshot, statuses) {
if playlistSnapshot.Revision == emittedFailedRevision {
continue
}
failed := PlaylistEvent{
Revision: playlistSnapshot.Revision,
Kind: PlaylistEventFailed,
}
select {
case <-ctx.Done():
return ctx.Err()
case c.output <- failed:
emittedFailedRevision = playlistSnapshot.Revision
}
continue
}
if playlistSnapshot.Entry.Duration <= 0 ||
playlistSnapshot.Timing.Started ||
playlistSnapshot.Timing.Paused ||
playlistSnapshot.Revision == emittedReadyRevision {
continue
}
if !IsSessionPlaying(sessionSnapshot, statuses) {
continue
}
@@ -123,7 +143,7 @@ func (c *PlaylistReadinessCoordinator) Run(ctx context.Context) error {
case <-ctx.Done():
return ctx.Err()
case c.output <- ready:
emittedRevision = playlistSnapshot.Revision
emittedReadyRevision = playlistSnapshot.Revision
}
}
}
@@ -193,6 +213,38 @@ func IsSessionPlaying(
}
}
func IsSessionFailed(
session SessionSnapshot,
statuses PlaybackStatusSnapshot,
) bool {
if statuses.Generation != session.Generation {
return false
}
switch session.Plan.Topology {
case TopologyIndependent:
return (session.Plan.Video.Active && statusIsFailed(
statuses.Video, statuses.HasVideo, session.Generation, session.Plan.Video,
)) || (session.Plan.Audio.Active && statusIsFailed(
statuses.Audio, statuses.HasAudio, session.Generation, session.Plan.Audio,
))
case TopologySynchronized:
return statuses.HasSync &&
statuses.Sync.Generation == session.Generation &&
statuses.Sync.State == StateFailed &&
sameSyncSource(statuses.Sync.Pair, session.Plan.Sync)
default:
return false
}
}
func statusIsFailed(status Status, present bool, generation uint64, feed FeedConfig) bool {
return present &&
status.Generation == generation &&
status.State == StateFailed &&
sameFeedSource(status.Feed, feed)
}
func statusIsPlaying(
status Status,
present bool,
+2 -1
View File
@@ -14,6 +14,7 @@ func runSyncAttempt(
videoConfig FeedConfig,
audioConfig FeedConfig,
) (resultErr error) {
videoCtx := withVideoSource(ctx, videoConfig)
reader, err := factory.OpenSync(
ctx,
videoConfig,
@@ -39,7 +40,7 @@ func runSyncAttempt(
return fmt.Errorf("read sync group: %w", err)
}
if err := videoSink.ConsumeVideo(ctx, frame.Video); err != nil {
if err := videoSink.ConsumeVideo(videoCtx, frame.Video); err != nil {
if ctx.Err() != nil {
return ctx.Err()
}
+11
View File
@@ -2,6 +2,17 @@ package playback
import "context"
type videoSourceContextKey struct{}
func withVideoSource(ctx context.Context, source FeedConfig) context.Context {
return context.WithValue(ctx, videoSourceContextKey{}, source)
}
func videoSourceFromContext(ctx context.Context) FeedConfig {
source, _ := ctx.Value(videoSourceContextKey{}).(FeedConfig)
return source
}
// VideoFrame contains metadata and borrowed source payload.
//
// Payload is valid only until the next VideoReader.ReadVideo call or until the
+1
View File
@@ -24,6 +24,7 @@ func runVideoAttempt(
sink VideoSink,
config FeedConfig,
) (resultErr error) {
ctx = withVideoSource(ctx, config)
reader, err := factory.OpenVideo(ctx, config)
if err != nil {
return fmt.Errorf("open video: %w", err)
+4
View File
@@ -7,6 +7,8 @@ import (
type PendingVideoFrame struct {
Frame VideoFrame
Generation uint64
Source FeedConfig
completeOnce sync.Once
result chan error
@@ -28,6 +30,8 @@ func (b *VideoBridge) ConsumeVideo(
) error {
pending := &PendingVideoFrame{
Frame: frame,
Generation: generationFromContext(ctx),
Source: videoSourceFromContext(ctx),
result: make(chan error, 1),
}
+9 -1
View File
@@ -21,8 +21,10 @@ func TestVideoBridgeDeliversFrameAndCompletionResult(t *testing.T) {
}
consumeResult := make(chan error, 1)
wantSource := FeedConfig{Domain: "/video", UUID: "video", Active: true}
go func() {
consumeResult <- bridge.ConsumeVideo(context.Background(), wantFrame)
ctx := withGeneration(context.Background(), 17)
consumeResult <- bridge.ConsumeVideo(withVideoSource(ctx, wantSource), wantFrame)
}()
ctx, cancel := context.WithTimeout(context.Background(), videoBridgeTestTimeout)
@@ -34,6 +36,12 @@ func TestVideoBridgeDeliversFrameAndCompletionResult(t *testing.T) {
if pending.Frame.Index != wantFrame.Index {
t.Fatalf("Next() frame index = %d, want %d", pending.Frame.Index, wantFrame.Index)
}
if pending.Generation != 17 {
t.Fatalf("Next() generation = %d, want 17", pending.Generation)
}
if pending.Source != wantSource {
t.Fatalf("Next() source = %#v, want %#v", pending.Source, wantSource)
}
if &pending.Frame.Payload[0] != &wantFrame.Payload[0] {
t.Fatal("Next() copied the borrowed payload")
}
+1
View File
@@ -1,5 +1,6 @@
{
"loop": true,
"on_failure": "next",
"entries": [
{
"name": "timelapse",
+14
View File
@@ -1,5 +1,6 @@
{
"loop": true,
"on_failure": "next",
"entries": [
{
"name": "timelapse",
@@ -14,6 +15,19 @@
"sync": true,
"duration": "10s"
},
{
"name": "fail",
"video": {
"domain": "/dev/shm/mxl",
"uuid": "6fbec3b1-1b0f-417d-9059-8b94a47197ed"
},
"audio": {
"domain": "/dev/shm/mxl",
"uuid": "6fbec3b1-1b0f-417d-9059-8b94a47197ec"
},
"sync": true,
"duration": "10s"
},
{
"name": "F1 Highlights",
"video": {