diff --git a/pkg/bus/segchanman.go b/pkg/bus/segchanman.go index 570a6d4de..b7c7333eb 100644 --- a/pkg/bus/segchanman.go +++ b/pkg/bus/segchanman.go @@ -66,17 +66,21 @@ func (b *Bus) SubscribeSegmentBuf(ctx context.Context, user string, rendition st chs = []*SegChan{} b.segChans[key] = chs } - ch := make(chan *Seg) + // Buffered, and fed in publication order by PublishSegment itself (no + // goroutine per segment, which delivered bursts in random order); a + // subscriber that falls chanSize segments behind loses segments rather + // than holding the publisher. The most recent segments are replayed + // into it first so a late subscriber can start at once. + ch := make(chan *Seg, chanSize) b.segBufMutex.RLock() defer b.segBufMutex.RUnlock() curBuf, ok := b.segBuf[key] - myCh := make(chan *Seg, chanSize) if ok { if bufSize > len(curBuf) { bufSize = len(curBuf) } for i := 0; i < bufSize; i += 1 { - myCh <- curBuf[len(curBuf)-bufSize+i] + ch <- curBuf[len(curBuf)-bufSize+i] } } segChan := &SegChan{C: ch, Context: ctx} @@ -127,17 +131,13 @@ func (b *Bus) PublishSegment(ctx context.Context, user string, rendition string, if !ok { return } - for _, ch := range chs { - go func(segChan *SegChan) { - select { - case segChan.C <- seg: - case <-segChan.Context.Done(): - return - case <-time.After(1 * time.Minute): - log.Warn(ctx, "failed to send segment to channel, timing out", "user", user, "rendition", rendition) - } - - }(ch) + for _, segChan := range chs { + select { + case segChan.C <- seg: + case <-segChan.Context.Done(): + default: + log.Warn(ctx, "segment subscriber is not keeping up, dropping a segment", "user", user, "rendition", rendition, "behind", len(segChan.C)) + } } } diff --git a/pkg/bus/segchanman_test.go b/pkg/bus/segchanman_test.go new file mode 100644 index 000000000..24693a528 --- /dev/null +++ b/pkg/bus/segchanman_test.go @@ -0,0 +1,47 @@ +package bus + +import ( + "context" + "fmt" + "testing" + + "github.com/stretchr/testify/require" +) + +// Segments reach a subscriber in the order they were published, and the +// recent ones are replayed to a subscriber that arrives late. +func TestPublishSegmentKeepsOrder(t *testing.T) { + b := NewBus() + ctx := context.Background() + for i := 0; i < 3; i++ { + b.PublishSegment(ctx, "u", "source", &Seg{Filepath: fmt.Sprint("early", i)}) + } + sub := b.SubscribeSegmentBuf(ctx, "u", "source", 2) + defer b.UnsubscribeSegment(ctx, "u", "source", sub) + const n = 500 + for i := 0; i < n; i++ { + b.PublishSegment(ctx, "u", "source", &Seg{Filepath: fmt.Sprint(i)}) + } + require.Equal(t, "early1", (<-sub.C).Filepath, "the last two are replayed") + require.Equal(t, "early2", (<-sub.C).Filepath) + for i := 0; i < n; i++ { + require.Equal(t, fmt.Sprint(i), (<-sub.C).Filepath) + } +} + +// A subscriber that never reads does not hold the publisher. +func TestPublishSegmentDropsForStalledSubscriber(t *testing.T) { + b := NewBus() + ctx := context.Background() + sub := b.SubscribeSegment(ctx, "u", "source") + defer b.UnsubscribeSegment(ctx, "u", "source", sub) + done := make(chan struct{}) + go func() { + for i := 0; i < chanSize+50; i++ { + b.PublishSegment(ctx, "u", "source", &Seg{}) + } + close(done) + }() + <-done + require.Len(t, sub.C, chanSize) +} diff --git a/pkg/director/director.go b/pkg/director/director.go index 82ce16fe6..fa3bc14e2 100644 --- a/pkg/director/director.go +++ b/pkg/director/director.go @@ -75,6 +75,8 @@ func (d *Director) Start(ctx context.Context) error { cli: d.cli, bus: d.bus, segmentChan: make(chan struct{}), + sourceLane: newLane(), + renditions: newLane(), op: d.op, packets: make([]bus.PacketizedSegment, 0), started: make(chan struct{}), diff --git a/pkg/director/lane.go b/pkg/director/lane.go new file mode 100644 index 000000000..baea26059 --- /dev/null +++ b/pkg/director/lane.go @@ -0,0 +1,118 @@ +package director + +import ( + "context" + "sync" + "time" +) + +// publishPatience bounds how long a segment waits for the segments before +// it to publish. Work behind a segment (the transcoder round trip, the +// packetizer) normally finishes within a second or two; a segment stuck +// far longer than that is abandoned by the ones behind it, which then +// publish, and it publishes late if it ever finishes. +const publishPatience = 15 * time.Second + +// A lane hands out tickets in the order segments arrive and lets each +// segment's publication take its turn in that order, whatever order the +// work behind it finishes in. Segments are worked on concurrently (two +// transcodes in flight, a packetizer per segment), and without this the +// rendition playlists, the WebRTC feeds and the peers pulling the stream +// all received segments in completion order: out of order in every burst. +type lane struct { + mu sync.Mutex + next uint64 + head uint64 // the ticket whose turn it is + done map[uint64]bool + waiters map[uint64]chan struct{} +} + +func newLane() *lane { + return &lane{done: map[uint64]bool{}, waiters: map[uint64]chan struct{}{}} +} + +// turn claims the next ticket. Every claimed turn must be released. +func (l *lane) turn() *turn { + l.mu.Lock() + defer l.mu.Unlock() + t := l.next + l.next++ + return &turn{lane: l, ticket: t} +} + +func (l *lane) wait(ctx context.Context, ticket uint64, patience time.Duration) { + l.mu.Lock() + if l.head >= ticket { + l.mu.Unlock() + return + } + ch, ok := l.waiters[ticket] + if !ok { + ch = make(chan struct{}) + l.waiters[ticket] = ch + } + l.mu.Unlock() + timer := time.NewTimer(patience) + defer timer.Stop() + select { + case <-ch: + case <-ctx.Done(): + case <-timer.C: + l.mu.Lock() + if l.head < ticket { + // Give up on the segments before it: they publish when they + // finish, late; everything from here on stays in order. + for t := l.head; t < ticket; t++ { + delete(l.done, t) + if w, ok := l.waiters[t]; ok { + close(w) + delete(l.waiters, t) + } + } + l.head = ticket + delete(l.waiters, ticket) + } + l.mu.Unlock() + } +} + +func (l *lane) release(ticket uint64) { + l.mu.Lock() + defer l.mu.Unlock() + if ticket < l.head { + return // abandoned by the segments behind it + } + l.done[ticket] = true + for l.done[l.head] { + delete(l.done, l.head) + l.head++ + } + if w, ok := l.waiters[l.head]; ok { + close(w) + delete(l.waiters, l.head) + } +} + +// A turn is one segment's place in a lane. wait blocks until the segments +// before it have published (or given up on, see publishPatience); release +// lets the next one go. Both are safe on a nil turn, so callers without a +// lane need no branches. +type turn struct { + lane *lane + ticket uint64 + once sync.Once +} + +func (t *turn) wait(ctx context.Context) { + if t == nil { + return + } + t.lane.wait(ctx, t.ticket, publishPatience) +} + +func (t *turn) release() { + if t == nil { + return + } + t.once.Do(func() { t.lane.release(t.ticket) }) +} diff --git a/pkg/director/lane_test.go b/pkg/director/lane_test.go new file mode 100644 index 000000000..026a1877c --- /dev/null +++ b/pkg/director/lane_test.go @@ -0,0 +1,77 @@ +package director + +import ( + "context" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// Turns claimed in order publish in order even when their work finishes +// backwards. +func TestLaneOrdersCompletions(t *testing.T) { + l := newLane() + const n = 20 + turns := make([]*turn, n) + for i := range turns { + turns[i] = l.turn() + } + var mu sync.Mutex + var order []int + var wg sync.WaitGroup + for i := n - 1; i >= 0; i-- { // finish backwards, staggered + wg.Add(1) + go func(i int) { + defer wg.Done() + time.Sleep(time.Duration(n-i) * 2 * time.Millisecond) + turns[i].wait(context.Background()) + mu.Lock() + order = append(order, i) + mu.Unlock() + turns[i].release() + }(i) + } + wg.Wait() + for i, v := range order { + require.Equal(t, i, v, "publication order") + } +} + +// A turn released without waiting (a failed transcode) does not hold the +// ones behind it. +func TestLaneReleaseWithoutWait(t *testing.T) { + l := newLane() + a, b := l.turn(), l.turn() + a.release() + done := make(chan struct{}) + go func() { b.wait(context.Background()); close(done) }() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("b never got its turn") + } + b.release() +} + +// A segment stuck longer than the patience is abandoned: the ones behind +// it go, and its late release is harmless. +func TestLanePatience(t *testing.T) { + l := newLane() + stuck, next := l.turn(), l.turn() + start := time.Now() + l.wait(context.Background(), next.ticket, 50*time.Millisecond) + require.Less(t, time.Since(start), time.Second) + next.release() + stuck.release() // late, ignored + third := l.turn() + l.wait(context.Background(), third.ticket, time.Second) + third.release() +} + +func TestNilTurnIsNoop(t *testing.T) { + var tn *turn + tn.wait(context.Background()) + tn.release() +} diff --git a/pkg/director/stream_session.go b/pkg/director/stream_session.go index cd1be81d4..a691884ac 100644 --- a/pkg/director/stream_session.go +++ b/pkg/director/stream_session.go @@ -40,6 +40,9 @@ import ( ) type StreamSession struct { + // Publication order for the source segments (WebRTC, peers) and for + // the transcoded renditions: see lane. + sourceLane, renditions *lane mm *media.MediaManager mod model.Model cli *config.CLI @@ -281,13 +284,16 @@ func (ss *StreamSession) NewSegment(ctx context.Context, notif *media.NewSegment } ss.bus.Publish(spseg.Creator, spseg) + // Claimed here, in arrival order, so the source segments publish in it + // however long each one's packetizing takes. + sourceTurn := ss.sourceLane.turn() ss.Go(ctx, func() error { return ss.AddPlaybackSegment(ctx, spseg, "source", &bus.Seg{ Filepath: notif.Segment.ID, Data: notif.Data, Muxl: notif.Muxl, Published: notif.Metadata.Published, - }) + }, sourceTurn) }) if notif.Local { @@ -317,9 +323,12 @@ func (ss *StreamSession) NewSegment(ctx context.Context, notif *media.NewSegment // Transcoding is the ingest node's job, like recording: a node that // merely syndicates the stream receives the renditions with it. if ss.cli.LivepeerGatewayURL != "" && notif.Local { + // Claimed in arrival order: the renditions publish in it however + // the gateway's round trips finish. + renditionTurn := ss.renditions.turn() ss.Go(ctx, func() error { start := time.Now() - err := ss.Transcode(ctx, spseg, notif) + err := ss.Transcode(ctx, spseg, notif, renditionTurn) took := time.Since(start) spmetrics.QueuedTranscodeDuration.WithLabelValues(spseg.Creator).Set(float64(took.Milliseconds())) return err @@ -843,7 +852,10 @@ func (ss *StreamSession) doUpdateViewCount(ctx context.Context, repoDID string) return nil } -func (ss *StreamSession) Transcode(ctx context.Context, spseg *placestream.Segment, notif *media.NewSegmentNotification) error { +func (ss *StreamSession) Transcode(ctx context.Context, spseg *placestream.Segment, notif *media.NewSegmentNotification, tn *turn) error { + // Released on every exit: a segment that is skipped or fails must not + // hold the ones behind it. + defer tn.release() data := notif.Data if len(notif.Muxl) > 0 { // Hand the transcoder the Opus presentation: it muxes that audio @@ -893,65 +905,93 @@ func (ss *StreamSession) Transcode(ctx context.Context, spseg *placestream.Segme if err != nil { return err } - // The renditions as signed canonical tracks: into the live window (HLS - // variants) now; the WebRTC path below keeps using the plain MP4s. + // Everything the renditions publish as is prepared first, concurrently + // where it can be; then the segment takes its turn and publishes all + // of it at once, so every consumer sees renditions in source order. + var addendum []byte if len(notif.Muxl) > 0 { inputs := make([]media.RenditionInput, len(segs)) for i, seg := range segs { inputs[i] = media.RenditionInput{Name: rs[i].Name, MP4: seg} } - addendum, err := ss.mm.MintVideoRenditions(ctx, notif.Muxl, inputs) + addendum, err = ss.mm.MintVideoRenditions(ctx, notif.Muxl, inputs) if err != nil { log.Warn(ctx, "could not mint rendition tracks", "error", err) - } else if addendum != nil { - ss.mm.FeedLiveRenditions(context.WithoutCancel(ctx), spseg.Creator, addendum, notif.Metadata.Published) - // And to the peers pulling this stream (see subscribeSegments). - ss.bus.PublishSegment(ctx, spseg.Creator, media.RenditionsChannel, &bus.Seg{ - Muxl: addendum, - Published: notif.Metadata.Published, - }) + addendum = nil } } + playback := make([]*bus.Seg, len(segs)) + var pg errgroup.Group for i, seg := range segs { ctx := log.WithLogValues(ctx, "rendition", rs[i].Name) - log.Debug(ctx, "publishing segment", "rendition", rs[i]) fd, err := ss.cli.SegmentFileCreate(spseg.Creator, aqt, fmt.Sprintf("%s.mp4", rs[i].Name)) if err != nil { return fmt.Errorf("failed to create transcoded segment file: %w", err) } - defer fd.Close() _, err = fd.Write(seg) + fd.Close() if err != nil { return fmt.Errorf("failed to write transcoded segment file: %w", err) } - ss.Go(ctx, func() error { - return ss.AddPlaybackSegment(ctx, spseg, rs[i].Name, &bus.Seg{ - Filepath: fd.Name(), - Data: seg, - // Carries the source's state: WebRTC playback hands an - // unpublished segment only to the streamer, and a rendition - // with the flag unset was refused to every viewer. - Published: notif.Metadata.Published, - }) + playback[i] = &bus.Seg{ + Filepath: fd.Name(), + Data: seg, + // Carries the source's state: WebRTC playback hands an + // unpublished segment only to the streamer, and a rendition + // with the flag unset was refused to every viewer. + Published: notif.Metadata.Published, + } + pg.Go(func() error { + packet, err := media.Packetize(ctx, ss.cli, playback[i]) + if err != nil { + log.Error(ctx, "failed to packetize rendition segment", "error", err) + playback[i] = nil + return nil + } + playback[i].PacketizedData = packet + return nil }) + } + _ = pg.Wait() + tn.wait(ctx) + if addendum != nil { + // Into the live window (HLS variants), and to the peers pulling + // this stream (see subscribeSegments). + ss.mm.FeedLiveRenditions(context.WithoutCancel(ctx), spseg.Creator, addendum, notif.Metadata.Published) + ss.bus.PublishSegment(ctx, spseg.Creator, media.RenditionsChannel, &bus.Seg{ + Muxl: addendum, + Published: notif.Metadata.Published, + }) + } + for i, seg := range playback { + if seg == nil { + continue + } + log.Debug(ctx, "publishing segment", "rendition", rs[i]) + ss.bus.PublishSegment(ctx, spseg.Creator, rs[i].Name, seg) } return nil } -func (ss *StreamSession) AddPlaybackSegment(ctx context.Context, spseg *placestream.Segment, rendition string, seg *bus.Seg) error { +// AddPlaybackSegment packetizes a segment for WebRTC and publishes it on +// the bus, taking its turn (tn) in the stream's publication order once +// the packetizing is done. +func (ss *StreamSession) AddPlaybackSegment(ctx context.Context, spseg *placestream.Segment, rendition string, seg *bus.Seg, tn *turn) error { ss.Go(ctx, func() error { - return ss.AddToWebRTC(ctx, spseg, rendition, seg) + return ss.AddToWebRTC(ctx, spseg, rendition, seg, tn) }) return nil } -func (ss *StreamSession) AddToWebRTC(ctx context.Context, spseg *placestream.Segment, rendition string, seg *bus.Seg) error { +func (ss *StreamSession) AddToWebRTC(ctx context.Context, spseg *placestream.Segment, rendition string, seg *bus.Seg, tn *turn) error { + defer tn.release() packet, err := media.Packetize(ctx, ss.cli, seg) if err != nil { return fmt.Errorf("failed to packetize segment: %w", err) } seg.PacketizedData = packet + tn.wait(ctx) ss.bus.PublishSegment(ctx, spseg.Creator, rendition, seg) return nil } diff --git a/pkg/media/fanout_test.go b/pkg/media/fanout_test.go new file mode 100644 index 000000000..8978d7181 --- /dev/null +++ b/pkg/media/fanout_test.go @@ -0,0 +1,25 @@ +package media + +import ( + "context" + "fmt" + "testing" + + "github.com/stretchr/testify/require" + "stream.place/streamplace/pkg/localdb" +) + +// Validated segments reach a subscriber in validation order. +func TestNewSegmentFanoutKeepsOrder(t *testing.T) { + mm := &MediaManager{} + ch := mm.NewSegment() + const n = 200 + go func() { + for i := 0; i < n; i++ { + mm.notifySubscribers(context.Background(), &NewSegmentNotification{Segment: &localdb.Segment{ID: fmt.Sprint(i)}}) + } + }() + for i := 0; i < n; i++ { + require.Equal(t, fmt.Sprint(i), (<-ch).Segment.ID) + } +} diff --git a/pkg/media/media.go b/pkg/media/media.go index 0b5c04db3..7ec2cd3f7 100644 --- a/pkg/media/media.go +++ b/pkg/media/media.go @@ -10,6 +10,7 @@ import ( "io" "sync" "sync/atomic" + "time" "github.com/google/uuid" "github.com/pion/webrtc/v4" @@ -54,7 +55,7 @@ type MediaManager struct { modBuffersMut sync.Mutex httpPipes map[string]io.Writer httpPipesMutex sync.Mutex - newSegmentSubs []chan *NewSegmentNotification + newSegmentSubs []*segmentSubscriber newSegmentSubsMutex sync.RWMutex model model.Model bus *bus.Bus @@ -184,13 +185,54 @@ func (mm *MediaManager) GetHTTPPipeWriter(uu string) io.Writer { return mm.httpPipes[uu] } +// segmentQueueSize is how many validated segments may wait for one +// subscriber before further ones are dropped for it. +const segmentQueueSize = 256 + +// A segmentSubscriber receives every validated segment, in validation +// order: notifications queue per subscriber and one goroutine forwards +// them. (A goroutine per notification, as before, handed a burst of tiny +// segments to the director in whatever order the scheduler ran them.) +type segmentSubscriber struct { + ch chan *NewSegmentNotification + queue chan *NewSegmentNotification +} + +func (s *segmentSubscriber) forward() { + for not := range s.queue { + select { + case s.ch <- not: + case <-time.After(time.Minute): + log.Warn(context.Background(), "segment subscriber did not take a segment within a minute, dropping it", "streamer", not.Segment.RepoDID, "segmentID", not.Segment.ID) + } + } +} + // register a handler for all new segments that come in func (mm *MediaManager) NewSegment() <-chan *NewSegmentNotification { - ch := make(chan *NewSegmentNotification) + sub := &segmentSubscriber{ + ch: make(chan *NewSegmentNotification), + queue: make(chan *NewSegmentNotification, segmentQueueSize), + } + go sub.forward() mm.newSegmentSubsMutex.Lock() defer mm.newSegmentSubsMutex.Unlock() - mm.newSegmentSubs = append(mm.newSegmentSubs, ch) - return ch + mm.newSegmentSubs = append(mm.newSegmentSubs, sub) + return sub.ch +} + +// notifySubscribers hands a validated segment to every subscriber, in +// order, without waiting on any of them. +func (mm *MediaManager) notifySubscribers(ctx context.Context, not *NewSegmentNotification) { + mm.newSegmentSubsMutex.RLock() + defer mm.newSegmentSubsMutex.RUnlock() + for _, sub := range mm.newSegmentSubs { + select { + case sub.queue <- not: + default: + log.Error(ctx, "segment subscriber is not keeping up, dropping a segment", "streamer", not.Segment.RepoDID, "segmentID", not.Segment.ID, "behind", len(sub.queue)) + } + } } type obj map[string]any diff --git a/pkg/media/validate.go b/pkg/media/validate.go index ba4aa39bf..7e4c8bd64 100644 --- a/pkg/media/validate.go +++ b/pkg/media/validate.go @@ -270,27 +270,13 @@ func (mm *MediaManager) distributeSegment(ctx context.Context, vs *validatedSegm return fmt.Errorf("wrap segment for distribution: %w", err) } - mm.newSegmentSubsMutex.RLock() - defer mm.newSegmentSubsMutex.RUnlock() - not := &NewSegmentNotification{ + mm.notifySubscribers(ctx, &NewSegmentNotification{ Segment: dbSeg, Data: playable.Bytes(), Muxl: seg, Metadata: meta, Local: vs.local, - } - for _, ch := range mm.newSegmentSubs { - go func() { - select { - case ch <- not: - case <-ctx.Done(): - return - case <-time.After(1 * time.Minute): - log.Warn(ctx, "failed to send segment to channel, timing out", "streamer", vs.repoDID, "signingKey", vs.signingKeyDID, "segmentID", vs.label) - return - } - }() - } + }) aqt := aqtime.FromTime(meta.StartTime.Time()) log.Log(ctx, "successfully ingested segment", "user", vs.repoDID, "signingKey", vs.signingKeyDID, "timestamp", aqt.FileSafeString(), "segmentID", vs.label) return nil