Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
2.9 kB · 118 lines
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119package 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) })}