| @@ -199,11 +199,12 @@ Generator/Upsampler und Hardwarewriter werden als getrennte Stufen mit kleinem, | |||||
| - **Akzeptanzpunkte:** | - **Akzeptanzpunkte:** | ||||
| - Keine unbounded Queue. | - Keine unbounded Queue. | ||||
| - Fill-Level (High/Low) ist aus `QueueStats` sichtbar. | - Fill-Level (High/Low) ist aus `QueueStats` sichtbar. | ||||
| - Queue-Health-Indikator (`queue.health`) liefert `critical`, `low` oder `normal` aus dem Fill-Level. EngineStats.`queue` zeigt den Status ebenfalls. | |||||
| - Drop/Repeat/Mute-Counter sind vorhanden und testbar. | - Drop/Repeat/Mute-Counter sind vorhanden und testbar. | ||||
| - **Nachweis:** | - **Nachweis:** | ||||
| - `FrameQueue`-Implementierung (`internal/output/frame_queue.go`) liefert kapazitätsgesteuerte Push/Pop-Logik und Counters. | - `FrameQueue`-Implementierung (`internal/output/frame_queue.go`) liefert kapazitätsgesteuerte Push/Pop-Logik und Counters. | ||||
| - Engine-Run nutzt Queue vor dem Writer und zeigt `QueueStats` in `EngineStats`. | - Engine-Run nutzt Queue vor dem Writer und zeigt `QueueStats` in `EngineStats`. | ||||
| - Tests (`internal/output/frame_queue_test.go` + `go test ./...`) decken Push/Pop, Timeout-Counters und Stats ab. | |||||
| - Tests (`internal/output/frame_queue_test.go` + `go test ./...`) decken Push/Pop, Timeout-Counters, Stats und den neuen Queue-Health-Indikator ab. | |||||
| - **Restrisiken:** | - **Restrisiken:** | ||||
| - Die Queue wird aktuell synchron getrieben; ein dedizierter Writer-Worker fehlt noch. | - Die Queue wird aktuell synchron getrieben; ein dedizierter Writer-Worker fehlt noch. | ||||
| - Queue-Close erwartet, dass Generator/Writer vor dem Schließen stoppen, sonst droht Panik beim Schreiben. | - Queue-Close erwartet, dass Generator/Writer vor dem Schließen stoppen, sonst droht Panik beim Schreiben. | ||||
| @@ -246,11 +247,13 @@ Generator/Upsampler und Hardwarewriter werden als getrennte Stufen mit kleinem, | |||||
| | Datum | Entscheidung | Notiz | | | Datum | Entscheidung | Notiz | | ||||
| |---|---|---| | |---|---|---| | ||||
| | 2026-04-05 | FrameQueue mit Engine-Integration | Queue lebt nach dem Upsampler auf DeviceFrame-Ebene, Kapazität via `runtime.frameQueueCapacity`, `EngineStats` zeigt `QueueStats`, Tests decken Timeouts und Counters ab. | | | 2026-04-05 | FrameQueue mit Engine-Integration | Queue lebt nach dem Upsampler auf DeviceFrame-Ebene, Kapazität via `runtime.frameQueueCapacity`, `EngineStats` zeigt `QueueStats`, Tests decken Timeouts und Counters ab. | | ||||
| | 2026-04-05 | Queue-Health-Indikator | `QueueStats.Health` gibt `critical`/`low`/`normal` zurück und ist über `EngineStats.Queue` im Runtime-Endpunkt sichtbar. | | |||||
| ## WS-01 Verifikation | ## WS-01 Verifikation | ||||
| | Datum | Fokus | Ergebnis | | | Datum | Fokus | Ergebnis | | ||||
| |---|---|---| | |---|---|---| | ||||
| | 2026-04-05 | FrameQueue + Engine integration | ✅ `go test ./...` (im `internal`-Modul incl. `frame_queue_test.go`) | | | 2026-04-05 | FrameQueue + Engine integration | ✅ `go test ./...` (im `internal`-Modul incl. `frame_queue_test.go`) | | ||||
| | 2026-04-05 | Queue-Health-Indikator | go test ./... deckt `TestFrameQueueHealthIndicator` und `queue.health` ab. | | |||||
| --- | --- | ||||
| @@ -12,17 +12,31 @@ var ErrFrameQueueClosed = errors.New("frame queue closed") | |||||
| // QueueStats exposes the runtime state of a frame queue. | // QueueStats exposes the runtime state of a frame queue. | ||||
| type QueueStats struct { | type QueueStats struct { | ||||
| Capacity int `json:"capacity"` | |||||
| Depth int `json:"depth"` | |||||
| FillLevel float64 `json:"fillLevel"` | |||||
| HighWaterMark int `json:"highWaterMark"` | |||||
| LowWaterMark int `json:"lowWaterMark"` | |||||
| PushTimeouts uint64 `json:"pushTimeouts"` | |||||
| PopTimeouts uint64 `json:"popTimeouts"` | |||||
| DroppedFrames uint64 `json:"droppedFrames"` | |||||
| RepeatedFrames uint64 `json:"repeatedFrames"` | |||||
| MutedFrames uint64 `json:"mutedFrames"` | |||||
| } | |||||
| Capacity int `json:"capacity"` | |||||
| Depth int `json:"depth"` | |||||
| FillLevel float64 `json:"fillLevel"` | |||||
| Health QueueHealth `json:"health"` | |||||
| HighWaterMark int `json:"highWaterMark"` | |||||
| LowWaterMark int `json:"lowWaterMark"` | |||||
| PushTimeouts uint64 `json:"pushTimeouts"` | |||||
| PopTimeouts uint64 `json:"popTimeouts"` | |||||
| DroppedFrames uint64 `json:"droppedFrames"` | |||||
| RepeatedFrames uint64 `json:"repeatedFrames"` | |||||
| MutedFrames uint64 `json:"mutedFrames"` | |||||
| } | |||||
| type QueueHealth string | |||||
| const ( | |||||
| QueueHealthCritical QueueHealth = "critical" | |||||
| QueueHealthLow QueueHealth = "low" | |||||
| QueueHealthNormal QueueHealth = "normal" | |||||
| ) | |||||
| const ( | |||||
| queueHealthCriticalThreshold = 0.2 | |||||
| queueHealthLowThreshold = 0.5 | |||||
| ) | |||||
| // FrameQueue is a bounded ring that holds CompositeFrame instances between the | // FrameQueue is a bounded ring that holds CompositeFrame instances between the | ||||
| // generator and the writer. Push blocks when the queue is full until space | // generator and the writer. Push blocks when the queue is full until space | ||||
| @@ -87,10 +101,12 @@ func (q *FrameQueue) Depth() int { | |||||
| // Stats returns a snapshot of the queue metrics. | // Stats returns a snapshot of the queue metrics. | ||||
| func (q *FrameQueue) Stats() QueueStats { | func (q *FrameQueue) Stats() QueueStats { | ||||
| q.mu.Lock() | q.mu.Lock() | ||||
| fill := q.fillLevelLocked() | |||||
| stats := QueueStats{ | stats := QueueStats{ | ||||
| Capacity: q.capacity, | Capacity: q.capacity, | ||||
| Depth: q.depth, | Depth: q.depth, | ||||
| FillLevel: q.fillLevelLocked(), | |||||
| FillLevel: fill, | |||||
| Health: queueHealthFromFill(fill), | |||||
| HighWaterMark: q.highWaterMark, | HighWaterMark: q.highWaterMark, | ||||
| LowWaterMark: q.lowWaterMark, | LowWaterMark: q.lowWaterMark, | ||||
| PushTimeouts: q.pushTimeouts, | PushTimeouts: q.pushTimeouts, | ||||
| @@ -209,3 +225,14 @@ func (q *FrameQueue) recordPopTimeout() { | |||||
| q.popTimeouts++ | q.popTimeouts++ | ||||
| q.mu.Unlock() | q.mu.Unlock() | ||||
| } | } | ||||
| func queueHealthFromFill(fill float64) QueueHealth { | |||||
| switch { | |||||
| case fill <= queueHealthCriticalThreshold: | |||||
| return QueueHealthCritical | |||||
| case fill <= queueHealthLowThreshold: | |||||
| return QueueHealthLow | |||||
| default: | |||||
| return QueueHealthNormal | |||||
| } | |||||
| } | |||||
| @@ -80,3 +80,44 @@ func TestFrameQueueCounters(t *testing.T) { | |||||
| t.Fatalf("expected 1 mute, got %d", stats.MutedFrames) | t.Fatalf("expected 1 mute, got %d", stats.MutedFrames) | ||||
| } | } | ||||
| } | } | ||||
| func TestFrameQueueHealthIndicator(t *testing.T) { | |||||
| q := NewFrameQueue(4) | |||||
| ctx := context.Background() | |||||
| stats := q.Stats() | |||||
| if stats.Health != QueueHealthCritical { | |||||
| t.Fatalf("expected initial health critical, got %s", stats.Health) | |||||
| } | |||||
| push := func(seq int) { | |||||
| frame := &CompositeFrame{Sequence: seq} | |||||
| if err := q.Push(ctx, frame); err != nil { | |||||
| t.Fatalf("push %d failed: %v", seq, err) | |||||
| } | |||||
| } | |||||
| push(1) | |||||
| stats = q.Stats() | |||||
| if stats.Health != QueueHealthLow { | |||||
| t.Fatalf("expected low after one frame, got %s", stats.Health) | |||||
| } | |||||
| push(2) | |||||
| stats = q.Stats() | |||||
| if stats.Health != QueueHealthLow { | |||||
| t.Fatalf("expected low at 50%% fill, got %s", stats.Health) | |||||
| } | |||||
| push(3) | |||||
| stats = q.Stats() | |||||
| if stats.Health != QueueHealthNormal { | |||||
| t.Fatalf("expected normal once queue has ~75%% fill, got %s", stats.Health) | |||||
| } | |||||
| for q.Depth() > 0 { | |||||
| if _, err := q.Pop(ctx); err != nil { | |||||
| t.Fatalf("cleanup pop failed: %v", err) | |||||
| } | |||||
| } | |||||
| } | |||||