|
|
|
@@ -38,6 +38,19 @@ func (s EngineState) String() string { |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
type RuntimeState string |
|
|
|
|
|
|
|
const ( |
|
|
|
RuntimeStateIdle RuntimeState = "idle" |
|
|
|
RuntimeStateArming RuntimeState = "arming" |
|
|
|
RuntimeStatePrebuffering RuntimeState = "prebuffering" |
|
|
|
RuntimeStateRunning RuntimeState = "running" |
|
|
|
RuntimeStateDegraded RuntimeState = "degraded" |
|
|
|
RuntimeStateMuted RuntimeState = "muted" |
|
|
|
RuntimeStateFaulted RuntimeState = "faulted" |
|
|
|
RuntimeStateStopping RuntimeState = "stopping" |
|
|
|
) |
|
|
|
|
|
|
|
func updateMaxDuration(dst *atomic.Uint64, d time.Duration) { |
|
|
|
v := uint64(d) |
|
|
|
for { |
|
|
|
@@ -96,11 +109,12 @@ type Engine struct { |
|
|
|
deviceRate float64 |
|
|
|
frameQueue *output.FrameQueue |
|
|
|
|
|
|
|
mu sync.Mutex |
|
|
|
state EngineState |
|
|
|
cancel context.CancelFunc |
|
|
|
startedAt time.Time |
|
|
|
wg sync.WaitGroup |
|
|
|
mu sync.Mutex |
|
|
|
state EngineState |
|
|
|
cancel context.CancelFunc |
|
|
|
startedAt time.Time |
|
|
|
wg sync.WaitGroup |
|
|
|
runtimeState atomic.Value |
|
|
|
|
|
|
|
chunksProduced atomic.Uint64 |
|
|
|
totalSamples atomic.Uint64 |
|
|
|
@@ -177,7 +191,7 @@ func NewEngine(cfg cfgpkg.Config, driver platform.SoapyDriver) *Engine { |
|
|
|
log.Printf("engine: same-rate mode — DSP@%dHz", cfg.FM.CompositeRateHz) |
|
|
|
} |
|
|
|
|
|
|
|
return &Engine{ |
|
|
|
engine := &Engine{ |
|
|
|
cfg: cfg, |
|
|
|
driver: driver, |
|
|
|
generator: offpkg.NewGenerator(cfg), |
|
|
|
@@ -187,6 +201,8 @@ func NewEngine(cfg cfgpkg.Config, driver platform.SoapyDriver) *Engine { |
|
|
|
state: EngineIdle, |
|
|
|
frameQueue: output.NewFrameQueue(cfg.Runtime.FrameQueueCapacity), |
|
|
|
} |
|
|
|
engine.setRuntimeState(RuntimeStateIdle) |
|
|
|
return engine |
|
|
|
} |
|
|
|
|
|
|
|
func (e *Engine) SetChunkDuration(d time.Duration) { |
|
|
|
@@ -306,6 +322,7 @@ func (e *Engine) Start(ctx context.Context) error { |
|
|
|
runCtx, cancel := context.WithCancel(ctx) |
|
|
|
e.cancel = cancel |
|
|
|
e.state = EngineRunning |
|
|
|
e.setRuntimeState(RuntimeStateArming) |
|
|
|
e.startedAt = time.Now() |
|
|
|
e.wg.Add(1) |
|
|
|
e.mu.Unlock() |
|
|
|
@@ -321,6 +338,7 @@ func (e *Engine) Stop(ctx context.Context) error { |
|
|
|
return nil |
|
|
|
} |
|
|
|
e.state = EngineStopping |
|
|
|
e.setRuntimeState(RuntimeStateStopping) |
|
|
|
e.cancel() |
|
|
|
e.mu.Unlock() |
|
|
|
|
|
|
|
@@ -336,6 +354,7 @@ func (e *Engine) Stop(ctx context.Context) error { |
|
|
|
|
|
|
|
e.mu.Lock() |
|
|
|
e.state = EngineIdle |
|
|
|
e.setRuntimeState(RuntimeStateIdle) |
|
|
|
e.mu.Unlock() |
|
|
|
return nil |
|
|
|
} |
|
|
|
@@ -359,7 +378,7 @@ func (e *Engine) Stats() EngineStats { |
|
|
|
hasRecentLateBuffers := lateAlertAt > 0 && now.Sub(time.Unix(0, int64(lateAlertAt))) <= lateBufferIndicatorWindow |
|
|
|
ri := runtimeIndicator(queue.Health, hasRecentLateBuffers) |
|
|
|
return EngineStats{ |
|
|
|
State: state.String(), |
|
|
|
State: string(e.currentRuntimeState()), |
|
|
|
ChunksProduced: e.chunksProduced.Load(), |
|
|
|
TotalSamples: e.totalSamples.Load(), |
|
|
|
Underruns: e.underruns.Load(), |
|
|
|
@@ -401,6 +420,7 @@ func runtimeAlert(queueHealth output.QueueHealth, recentLateBuffers bool) string |
|
|
|
} |
|
|
|
|
|
|
|
func (e *Engine) run(ctx context.Context) { |
|
|
|
e.setRuntimeState(RuntimeStateRunning) |
|
|
|
e.wg.Add(1) |
|
|
|
go e.writerLoop(ctx) |
|
|
|
defer e.wg.Done() |
|
|
|
@@ -530,3 +550,16 @@ func cloneFrame(src *output.CompositeFrame) *output.CompositeFrame { |
|
|
|
Sequence: src.Sequence, |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
func (e *Engine) setRuntimeState(state RuntimeState) { |
|
|
|
e.runtimeState.Store(state) |
|
|
|
} |
|
|
|
|
|
|
|
func (e *Engine) currentRuntimeState() RuntimeState { |
|
|
|
if v := e.runtimeState.Load(); v != nil { |
|
|
|
if rs, ok := v.(RuntimeState); ok { |
|
|
|
return rs |
|
|
|
} |
|
|
|
} |
|
|
|
return RuntimeStateIdle |
|
|
|
} |