|
|
|
@@ -80,6 +80,8 @@ const ( |
|
|
|
RuntimeIndicatorQueueCritical RuntimeIndicator = "queueCritical" |
|
|
|
) |
|
|
|
|
|
|
|
const lateBufferIndicatorWindow = 5 * time.Second |
|
|
|
|
|
|
|
// Engine is the continuous TX loop. It generates composite IQ in chunks, |
|
|
|
// resamples to device rate, and pushes to hardware in a tight loop. |
|
|
|
// The hardware buffer_push call is blocking — it returns when the hardware |
|
|
|
@@ -100,15 +102,16 @@ type Engine struct { |
|
|
|
startedAt time.Time |
|
|
|
wg sync.WaitGroup |
|
|
|
|
|
|
|
chunksProduced atomic.Uint64 |
|
|
|
totalSamples atomic.Uint64 |
|
|
|
underruns atomic.Uint64 |
|
|
|
lateBuffers atomic.Uint64 |
|
|
|
maxCycleNs atomic.Uint64 |
|
|
|
maxGenerateNs atomic.Uint64 |
|
|
|
maxUpsampleNs atomic.Uint64 |
|
|
|
maxWriteNs atomic.Uint64 |
|
|
|
lastError atomic.Value // string |
|
|
|
chunksProduced atomic.Uint64 |
|
|
|
totalSamples atomic.Uint64 |
|
|
|
underruns atomic.Uint64 |
|
|
|
lateBuffers atomic.Uint64 |
|
|
|
lateBufferAlertAt atomic.Uint64 |
|
|
|
maxCycleNs atomic.Uint64 |
|
|
|
maxGenerateNs atomic.Uint64 |
|
|
|
maxUpsampleNs atomic.Uint64 |
|
|
|
maxWriteNs atomic.Uint64 |
|
|
|
lastError atomic.Value // string |
|
|
|
|
|
|
|
// Live config: pending frequency change, applied between chunks |
|
|
|
pendingFreq atomic.Pointer[float64] |
|
|
|
@@ -351,7 +354,10 @@ func (e *Engine) Stats() EngineStats { |
|
|
|
|
|
|
|
queue := e.frameQueue.Stats() |
|
|
|
lateBuffers := e.lateBuffers.Load() |
|
|
|
ri := runtimeIndicator(queue.Health, lateBuffers) |
|
|
|
now := time.Now() |
|
|
|
lateAlertAt := e.lateBufferAlertAt.Load() |
|
|
|
hasRecentLateBuffers := lateAlertAt > 0 && now.Sub(time.Unix(0, int64(lateAlertAt))) <= lateBufferIndicatorWindow |
|
|
|
ri := runtimeIndicator(queue.Health, hasRecentLateBuffers) |
|
|
|
return EngineStats{ |
|
|
|
State: state.String(), |
|
|
|
ChunksProduced: e.chunksProduced.Load(), |
|
|
|
@@ -366,26 +372,26 @@ func (e *Engine) Stats() EngineStats { |
|
|
|
MaxWriteMs: durationMs(e.maxWriteNs.Load()), |
|
|
|
Queue: queue, |
|
|
|
RuntimeIndicator: ri, |
|
|
|
RuntimeAlert: runtimeAlert(queue.Health, lateBuffers), |
|
|
|
RuntimeAlert: runtimeAlert(queue.Health, hasRecentLateBuffers), |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
func runtimeIndicator(queueHealth output.QueueHealth, lateBuffers uint64) RuntimeIndicator { |
|
|
|
func runtimeIndicator(queueHealth output.QueueHealth, recentLateBuffers bool) RuntimeIndicator { |
|
|
|
switch { |
|
|
|
case queueHealth == output.QueueHealthCritical: |
|
|
|
return RuntimeIndicatorQueueCritical |
|
|
|
case queueHealth == output.QueueHealthLow || lateBuffers > 0: |
|
|
|
case queueHealth == output.QueueHealthLow || recentLateBuffers: |
|
|
|
return RuntimeIndicatorDegraded |
|
|
|
default: |
|
|
|
return RuntimeIndicatorNormal |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
func runtimeAlert(queueHealth output.QueueHealth, lateBuffers uint64) string { |
|
|
|
func runtimeAlert(queueHealth output.QueueHealth, recentLateBuffers bool) string { |
|
|
|
switch { |
|
|
|
case queueHealth == output.QueueHealthCritical: |
|
|
|
return "queue health critical" |
|
|
|
case lateBuffers > 0: |
|
|
|
case recentLateBuffers: |
|
|
|
return "late buffers" |
|
|
|
case queueHealth == output.QueueHealthLow: |
|
|
|
return "queue health low" |
|
|
|
@@ -484,6 +490,7 @@ func (e *Engine) writerLoop(ctx context.Context) { |
|
|
|
|
|
|
|
if cycleDur > e.chunkDuration { |
|
|
|
late := e.lateBuffers.Add(1) |
|
|
|
e.lateBufferAlertAt.Store(uint64(time.Now().UnixNano())) |
|
|
|
if late <= 5 || late%20 == 0 { |
|
|
|
log.Printf("TX LATE: cycle=%s budget=%s write=%s over=%s", |
|
|
|
cycleDur, e.chunkDuration, writeDur, cycleDur-e.chunkDuration) |
|
|
|
|