1package tuner23import (4 "context"5 "errors"6 "fmt"7 "io"8 "slices"9 "strings"10 "sync"11 "time"1213 "airwaves/internal/hdhr"14 "airwaves/internal/ts"15)1617// Priority orders who gets a tuner when there aren't enough: a recording18// takes one from a viewer, and a viewer one from a measurement.19type Priority int2021// Priorities, lowest first.22const (23 // Idle is a tuner left on its RF channel for a moment after its last24 // user, so coming back to it is instant. Anyone can have it.25 Idle Priority = iota26 Measuring27 Watching28 Recording29)3031func (p Priority) String() string {32 switch p {33 case Measuring:34 return "measuring"35 case Watching:36 return "watching"37 case Recording:38 return "recording"39 }40 return "idle"41}4243// ErrNoDevice is returned while no HDHomeRun has been found.44var ErrNoDevice = errors.New("no tuner: no HDHomeRun found on the network")4546// ErrNoProgram is returned for a program the RF channel doesn't carry47// (any more).48var ErrNoProgram = errors.New("the RF channel doesn't carry that program")4950// errIdle ends a tuner that stayed unused for Linger.51var errIdle = errors.New("no one is using it")5253// errTooSlow ends a user that fell too far behind the broadcast.54var errTooSlow = errors.New("fell too far behind the broadcast")5556// BusyError is a tuner asked for when every one is in use by others at57// the same priority or higher.58type BusyError struct {59 // Users says who has the tuners, "watching 7.1".60 Users []string61}6263func (e *BusyError) Error() string {64 if len(e.Users) == 0 {65 return "every tuner is in use"66 }67 return "every tuner is in use (" + strings.Join(e.Users, "; ") + ")"68}6970// PreemptedError ends a user whose tuner was given to someone at a higher71// priority.72type PreemptedError struct {73 By string // "recording The News"74}7576func (e *PreemptedError) Error() string { return e.By + " needed the tuner" }7778// Hardware is the tuners: an HDHomeRun.79type Hardware interface {80 OpenRF(ctx context.Context, tuner int, freqHz int64) (*hdhr.Stream, error)81 Status(ctx context.Context) ([]hdhr.TunerStatus, error)82}8384// Manager shares an HDHomeRun's tuners among everything that wants antenna85// TV: viewers in the app, HDHomeRun clients, recordings and Measure now.86// It is the device's only client. Each RF channel in use is read once, on87// one tuner, and handed to everyone on it, each getting their own program,88// so watching 9.2 while recording 9.1 takes one tuner.89type Manager struct {90 // Linger is how long a tuner stays on its RF channel after its last91 // user leaves.92 Linger time.Duration93 // StatusAge is how long a reading of the device's status is used:94 // it's asked at most this often.95 StatusAge time.Duration9697 mu sync.Mutex98 dev Hardware99 tuners int100 muxes map[int]*mux // by tuner101 // away marks tuners the device said someone else has, until then.102 away map[int]time.Time103104 statusMu sync.Mutex105 status []hdhr.TunerStatus106 statusAt time.Time107}108109// NewManager returns a manager with no device yet.110func NewManager() *Manager {111 return &Manager{Linger: 5 * time.Second, StatusAge: 500 * time.Millisecond, muxes: map[int]*mux{}, away: map[int]time.Time{}}112}113114// SetDevice makes dev, with its count of tuners, the one to use; nil for115// none. Tuners in use on a device that goes away end.116func (m *Manager) SetDevice(dev Hardware, tuners int) {117 m.mu.Lock()118 defer m.mu.Unlock()119 if dev == m.dev && tuners == m.tuners {120 return121 }122 for _, x := range m.muxes {123 x.stop(errors.New("the tuner went away"))124 }125 m.dev, m.tuners = dev, tuners126}127128// Count is how many tuners the device has; 0 without one.129func (m *Manager) Count() int {130 m.mu.Lock()131 defer m.mu.Unlock()132 if m.dev == nil {133 return 0134 }135 return m.tuners136}137138// Request asks for a tuner.139type Request struct {140 // FrequencyHz is the RF channel's center frequency.141 FrequencyHz int64142 // Program is the program to receive, 0 for none: only to keep the143 // tuner on the RF channel, as measuring does.144 Program int145 Priority Priority146 // Who says what it's for, "watching 7.1".147 Who string148 // Owner, when set, is who asks (a viewer's app): a tuner only they149 // use is theirs to retune, so changing channels never finds every150 // tuner busy with their own last one.151 Owner string152}153154// Open puts a tuner on the RF channel r asks for and subscribes to its155// program there. A tuner already on that RF channel is shared. With no156// tuner free, it takes the one whose users all have a lower priority157// (or are all r.Owner), ending them, or fails with a *BusyError. It158// returns once the tuner is asked for; Wait waits for the program.159func (m *Manager) Open(ctx context.Context, r Request) (*Sub, error) {160 for {161 m.mu.Lock()162 if m.dev == nil {163 m.mu.Unlock()164 return nil, ErrNoDevice165 }166 for _, x := range m.muxes {167 if x.freq == r.FrequencyHz && x.alive() {168 s := x.add(r)169 m.mu.Unlock()170 return s, nil171 }172 }173 tuner, victim, err := m.pick(r)174 if err != nil {175 m.mu.Unlock()176 return nil, err177 }178 if victim == nil {179 x := m.start(tuner, r.FrequencyHz)180 s := x.add(r)181 m.mu.Unlock()182 return s, nil183 }184 victim.stop(&PreemptedError{By: r.Priority.String() + " " + r.Who})185 m.mu.Unlock()186 // The device frees the tuner once its stream is closed.187 select {188 case <-victim.done:189 case <-time.After(3 * time.Second):190 return nil, fmt.Errorf("tuner %d didn't come free", victim.tuner)191 case <-ctx.Done():192 return nil, ctx.Err()193 }194 }195}196197// pick finds a tuner for r: a free one, or failing that one in use only198// by r's owner, or the one in use at the lowest priority below r's, to199// take over. Of those alike, the one with the fewest users goes, then the200// one tuned last: the least to lose.201func (m *Manager) pick(r Request) (tuner int, victim *mux, err error) {202 now := time.Now()203 for t := range m.tuners {204 if _, used := m.muxes[t]; !used && !now.Before(m.away[t]) {205 return t, nil, nil206 }207 }208 for _, x := range m.muxes {209 if x.ownedBy(r.Owner) {210 return x.tuner, x, nil211 }212 }213 for _, x := range m.muxes {214 if x.top() < r.Priority && (victim == nil || x.lessThan(victim)) {215 victim = x216 }217 }218 if victim != nil {219 return victim.tuner, victim, nil220 }221 busy := &BusyError{}222 for t := range m.tuners {223 if x := m.muxes[t]; x != nil {224 busy.Users = append(busy.Users, x.users()...)225 }226 }227 slices.Sort(busy.Users)228 return 0, nil, busy229}230231// start puts tuner on freqHz. Called with m.mu held.232func (m *Manager) start(tuner int, freqHz int64) *mux {233 ctx, cancel := context.WithCancelCause(context.Background())234 x := &mux{235 m: m, tuner: tuner, freq: freqHz, since: time.Now(), ctx: ctx, cancel: cancel,236 subs: map[*Sub]bool{}, demux: ts.NewDemux(), changed: make(chan struct{}), done: make(chan struct{}),237 }238 m.muxes[tuner] = x239 go x.run(m.dev)240 return x241}242243// ended forgets x once its stream has closed.244func (m *Manager) ended(x *mux, err error) {245 m.mu.Lock()246 defer m.mu.Unlock()247 if m.muxes[x.tuner] == x {248 delete(m.muxes, x.tuner)249 }250 var he *hdhr.Error251 if errors.As(err, &he) && he.Busy() {252 // Someone else (the device's own scan, another app) has it.253 m.away[x.tuner] = time.Now().Add(30 * time.Second)254 }255}256257// User is someone using a tuner.258type User struct {259 Who string `json:"who"`260 Priority Priority `json:"priority"`261}262263// State is what one tuner is doing now.264type State struct {265 Number int266 // FrequencyHz is the RF channel it's on; 0 while idle.267 FrequencyHz int64268 Since time.Time269 Users []User270 // Top is the highest priority of its users; Idle with none.271 Top Priority272 // Locked is set while data arrives: the tuner has a lock.273 Locked bool274 // Packets and Errors count what arrived since it was tuned, and the275 // transport and continuity errors in it.276 Packets, Errors int64277 // Device is what the HDHomeRun itself reports of the tuner: strength,278 // quality and symbol quality, while it's tuned.279 Device hdhr.TunerStatus280 // Elsewhere is set for a tuner the device says is tuned by something281 // other than this manager.282 Elsewhere bool283}284285// Tuners reports every tuner: what this manager has it doing, and what286// the device measures. The device is asked at most every StatusAge.287func (m *Manager) Tuners(ctx context.Context) ([]State, error) {288 m.mu.Lock()289 dev, n := m.dev, m.tuners290 muxes := make(map[int]*mux, len(m.muxes))291 for t, x := range m.muxes {292 muxes[t] = x293 }294 m.mu.Unlock()295 if dev == nil {296 return nil, ErrNoDevice297 }298 status, err := m.deviceStatus(ctx, dev)299 out := make([]State, n)300 for t := range out {301 out[t] = State{Number: t}302 if x := muxes[t]; x != nil {303 out[t] = x.state()304 }305 }306 for _, st := range status {307 if st.Tuner < 0 || st.Tuner >= n {308 continue309 }310 out[st.Tuner].Device = st311 if out[st.Tuner].FrequencyHz == 0 && st.Tuned() {312 out[st.Tuner].FrequencyHz, out[st.Tuner].Elsewhere = st.FrequencyHz, true313 }314 }315 return out, err316}317318func (m *Manager) deviceStatus(ctx context.Context, dev Hardware) ([]hdhr.TunerStatus, error) {319 m.statusMu.Lock()320 defer m.statusMu.Unlock()321 if !m.statusAt.IsZero() && time.Since(m.statusAt) < m.StatusAge {322 return m.status, nil323 }324 st, err := dev.Status(ctx)325 if err != nil {326 return nil, err327 }328 m.status, m.statusAt = st, time.Now()329 return st, nil330}331332// mux is a tuner on an RF channel, read once for all its users.333type mux struct {334 m *Manager335 tuner int336 freq int64337 since time.Time338 ctx context.Context339 cancel context.CancelCauseFunc340 done chan struct{}341342 mu sync.Mutex343 subs map[*Sub]bool344 idle *time.Timer345 demux *ts.Demux346 counter ts.Counter347 answered bool // the device answered: it locked348 lastData time.Time349 // changed is closed, and replaced, whenever there's more to see.350 changed chan struct{}351}352353func (x *mux) alive() bool { return x.ctx.Err() == nil }354355// stop ends the tuner and everyone on it with cause.356func (x *mux) stop(cause error) { x.cancel(cause) }357358func (x *mux) run(dev Hardware) {359 defer close(x.done)360 err := x.read(dev)361 x.cancel(err)362 cause := context.Cause(x.ctx)363 x.mu.Lock()364 for s := range x.subs {365 s.end(cause)366 }367 if x.idle != nil {368 x.idle.Stop()369 }370 x.notify()371 x.mu.Unlock()372 x.m.ended(x, cause)373}374375func (x *mux) read(dev Hardware) error {376 st, err := dev.OpenRF(x.ctx, x.tuner, x.freq)377 if err != nil {378 return err379 }380 defer st.Close()381 x.mu.Lock()382 x.answered = true383 x.notify()384 x.mu.Unlock()385 var split ts.Splitter386 buf := make([]byte, 64<<10)387 for {388 n, err := st.Read(buf)389 if n > 0 {390 x.mu.Lock()391 x.lastData = time.Now()392 split.Write(buf[:n], func(pkt []byte) {393 x.demux.Packet(pkt)394 x.counter.Packet(pkt)395 for s := range x.subs {396 s.packet(pkt)397 }398 })399 for s := range x.subs {400 s.flush()401 }402 x.notify()403 x.mu.Unlock()404 }405 if err != nil {406 if x.ctx.Err() != nil {407 return context.Cause(x.ctx)408 }409 return fmt.Errorf("the tuner stopped: %w", err)410 }411 }412}413414// notify wakes whoever waits on x. Called with x.mu held.415func (x *mux) notify() {416 close(x.changed)417 x.changed = make(chan struct{})418}419420// add subscribes r's asker to its program. Called with m.mu held.421func (x *mux) add(r Request) *Sub {422 ctx, cancel := context.WithCancelCause(x.ctx)423 s := &Sub{x: x, program: r.Program, prio: r.Priority, who: r.Who, owner: r.Owner, ctx: ctx, cancel: cancel, data: make(chan []byte, 256)}424 if r.Program > 0 {425 s.filter = ts.NewFilter(&s.pending, r.Program)426 }427 x.mu.Lock()428 x.subs[s] = true429 if x.idle != nil {430 x.idle.Stop()431 x.idle = nil432 }433 x.mu.Unlock()434 return s435}436437// remove unsubscribes s; with no one left, the tuner lingers, then ends.438func (x *mux) remove(s *Sub) {439 x.m.mu.Lock()440 defer x.m.mu.Unlock()441 x.mu.Lock()442 defer x.mu.Unlock()443 if !x.subs[s] {444 return445 }446 delete(x.subs, s)447 if len(x.subs) > 0 || !x.alive() {448 return449 }450 if x.m.Linger <= 0 {451 x.stop(errIdle)452 return453 }454 x.idle = time.AfterFunc(x.m.Linger, func() {455 x.mu.Lock()456 unused := len(x.subs) == 0457 x.mu.Unlock()458 if unused {459 x.stop(errIdle)460 }461 })462}463464// ownedBy reports whether x is used by owner alone (and not idle).465// Called with m.mu held.466func (x *mux) ownedBy(owner string) bool {467 if owner == "" {468 return false469 }470 x.mu.Lock()471 defer x.mu.Unlock()472 for s := range x.subs {473 if s.owner != owner {474 return false475 }476 }477 return len(x.subs) > 0478}479480// lessThan reports whether x has less to lose than y: lower priority481// users, fewer of them, or tuned later. Called with m.mu held.482func (x *mux) lessThan(y *mux) bool {483 if a, b := x.top(), y.top(); a != b {484 return a < b485 }486 if a, b := x.count(), y.count(); a != b {487 return a < b488 }489 return x.since.After(y.since)490}491492func (x *mux) count() int {493 x.mu.Lock()494 defer x.mu.Unlock()495 return len(x.subs)496}497498// top is the highest priority among x's users. Called with m.mu held.499func (x *mux) top() Priority {500 x.mu.Lock()501 defer x.mu.Unlock()502 top := Idle503 for s := range x.subs {504 top = max(top, s.prio)505 }506 return top507}508509// users describes who is on x, "watching 7.1". Called with m.mu held.510func (x *mux) users() []string {511 x.mu.Lock()512 defer x.mu.Unlock()513 var out []string514 for s := range x.subs {515 out = append(out, s.prio.String()+" "+s.who)516 }517 return out518}519520func (x *mux) state() State {521 x.mu.Lock()522 defer x.mu.Unlock()523 t := State{524 Number: x.tuner, FrequencyHz: x.freq, Since: x.since, Packets: x.counter.Packets, Errors: x.counter.Errors(),525 Locked: x.answered && time.Since(x.lastData) < 2*time.Second,526 }527 for s := range x.subs {528 t.Users = append(t.Users, User{Who: s.who, Priority: s.prio})529 t.Top = max(t.Top, s.prio)530 }531 slices.SortFunc(t.Users, func(a, b User) int { return strings.Compare(a.Who, b.Who) })532 return t533}534535// maxPending is how far a user may fall behind before it's dropped.536const maxPending = 16 << 20537538// Sub is one user of a tuner: a program it receives, or only the tuner539// held on an RF channel.540type Sub struct {541 x *mux542 program int543 prio Priority544 who string545 owner string546 ctx context.Context547 cancel context.CancelCauseFunc548 filter *ts.Filter549 // pending is what the filter wrote since the last flush; data hands550 // it to whoever copies the program. Both are kept under x.mu.551 pending buffer552 data chan []byte553 once sync.Once554}555556// buffer collects the filter's output.557type buffer []byte558559func (b *buffer) Write(p []byte) (int, error) {560 *b = append(*b, p...)561 return len(p), nil562}563564// packet passes one of the multiplex's packets to s. Called with x.mu held.565func (s *Sub) packet(pkt []byte) {566 if s.filter != nil && s.ctx.Err() == nil {567 _ = s.filter.Packet(pkt)568 }569}570571// flush hands on what s collected, without ever blocking the tuner: a572// user that has fallen too far behind is dropped. Called with x.mu held.573func (s *Sub) flush() {574 if s.filter != nil {575 _ = s.filter.Flush()576 }577 if len(s.pending) == 0 {578 return579 }580 select {581 case s.data <- s.pending:582 s.pending = nil583 default:584 if len(s.pending) > maxPending {585 s.pending = nil586 s.end(errTooSlow)587 }588 }589}590591// end ends s with cause.592func (s *Sub) end(cause error) { s.cancel(cause) }593594// Tuner is the tuner s is on.595func (s *Sub) Tuner() int { return s.x.tuner }596597// FrequencyHz is the RF channel s is on.598func (s *Sub) FrequencyHz() int64 { return s.x.freq }599600// Done is closed when s ends: closed, its tuner taken over, or the tuner601// gone. Err says why.602func (s *Sub) Done() <-chan struct{} { return s.ctx.Done() }603604// Err is why s ended; nil while it hasn't.605func (s *Sub) Err() error {606 if s.ctx.Err() == nil {607 return nil608 }609 return context.Cause(s.ctx)610}611612// Locked reports whether data is arriving: the tuner has a lock.613func (s *Sub) Locked() bool {614 s.x.mu.Lock()615 defer s.x.mu.Unlock()616 return s.x.answered && time.Since(s.x.lastData) < 2*time.Second617}618619// Wait waits until s's program can be read, and describes it: once its620// PMT has arrived, or for a subscription without a program, once the621// tuner locked. It fails when s ends first, when the RF channel turns out622// not to carry the program, or when ctx ends.623func (s *Sub) Wait(ctx context.Context) (ts.Program, error) {624 for {625 s.x.mu.Lock()626 changed := s.x.changed627 var p ts.Program628 ok := false629 switch {630 case s.program == 0:631 ok = s.x.answered && !s.x.lastData.IsZero()632 default:633 // Once every PMT the PAT lists is in, a program without one634 // isn't there.635 if p, ok = s.x.demux.Program(s.program); !ok && s.x.demux.Ready() {636 s.x.mu.Unlock()637 return ts.Program{}, fmt.Errorf("program %d: %w", s.program, ErrNoProgram)638 }639 }640 s.x.mu.Unlock()641 if ok {642 return p, nil643 }644 select {645 case <-changed:646 case <-s.ctx.Done():647 return ts.Program{}, context.Cause(s.ctx)648 case <-ctx.Done():649 return ts.Program{}, ctx.Err()650 }651 }652}653654// Copy writes s's program to w until s ends, w fails or ctx ends, and655// returns why it stopped.656func (s *Sub) Copy(ctx context.Context, w io.Writer) error {657 for {658 select {659 case b := <-s.data:660 if _, err := w.Write(b); err != nil {661 return err662 }663 case <-s.ctx.Done():664 return context.Cause(s.ctx)665 case <-ctx.Done():666 return ctx.Err()667 }668 }669}670671// Close lets go of the tuner. It's safe to call more than once.672func (s *Sub) Close() {673 s.once.Do(func() {674 s.cancel(context.Canceled)675 s.x.remove(s)676 })677}