From 947b4f95280956f9eb554e405a8c75ea3d4bb00b Mon Sep 17 00:00:00 2001 From: Alex P Date: Fri, 5 Sep 2025 21:47:21 +0000 Subject: [PATCH] [WIP] Updates: Reduce PR complexity --- internal/audio/core_metrics.go | 83 ------ internal/audio/core_metrics_registry.go | 57 +--- internal/audio/latency_monitor.go | 333 ------------------------ internal/audio/quality_presets.go | 5 + 4 files changed, 10 insertions(+), 468 deletions(-) delete mode 100644 internal/audio/latency_monitor.go diff --git a/internal/audio/core_metrics.go b/internal/audio/core_metrics.go index 4247b5cd..03fafae6 100644 --- a/internal/audio/core_metrics.go +++ b/internal/audio/core_metrics.go @@ -230,63 +230,6 @@ var ( }, ) - // Audio configuration metrics - audioConfigQuality = promauto.NewGauge( - prometheus.GaugeOpts{ - Name: "jetkvm_audio_config_quality", - Help: "Current audio quality setting (0=Low, 1=Medium, 2=High, 3=Ultra)", - }, - ) - - audioConfigBitrate = promauto.NewGauge( - prometheus.GaugeOpts{ - Name: "jetkvm_audio_config_bitrate_kbps", - Help: "Current audio bitrate in kbps", - }, - ) - - audioConfigSampleRate = promauto.NewGauge( - prometheus.GaugeOpts{ - Name: "jetkvm_audio_config_sample_rate_hz", - Help: "Current audio sample rate in Hz", - }, - ) - - audioConfigChannels = promauto.NewGauge( - prometheus.GaugeOpts{ - Name: "jetkvm_audio_config_channels", - Help: "Current audio channel count", - }, - ) - - microphoneConfigQuality = promauto.NewGauge( - prometheus.GaugeOpts{ - Name: "jetkvm_microphone_config_quality", - Help: "Current microphone quality setting (0=Low, 1=Medium, 2=High, 3=Ultra)", - }, - ) - - microphoneConfigBitrate = promauto.NewGauge( - prometheus.GaugeOpts{ - Name: "jetkvm_microphone_config_bitrate_kbps", - Help: "Current microphone bitrate in kbps", - }, - ) - - microphoneConfigSampleRate = promauto.NewGauge( - prometheus.GaugeOpts{ - Name: "jetkvm_microphone_config_sample_rate_hz", - Help: "Current microphone sample rate in Hz", - }, - ) - - microphoneConfigChannels = promauto.NewGauge( - prometheus.GaugeOpts{ - Name: "jetkvm_microphone_config_channels", - Help: "Current microphone channel count", - }, - ) - // Device health metrics // Removed device health metrics - functionality not used @@ -539,32 +482,6 @@ func UpdateMicrophoneProcessMetrics(metrics ProcessMetrics, isRunning bool) { atomic.StoreInt64(&lastMetricsUpdate, time.Now().Unix()) } -// UpdateAudioConfigMetrics updates Prometheus metrics with audio configuration -func UpdateAudioConfigMetrics(config AudioConfig) { - metricsUpdateMutex.Lock() - defer metricsUpdateMutex.Unlock() - - audioConfigQuality.Set(float64(config.Quality)) - audioConfigBitrate.Set(float64(config.Bitrate)) - audioConfigSampleRate.Set(float64(config.SampleRate)) - audioConfigChannels.Set(float64(config.Channels)) - - atomic.StoreInt64(&lastMetricsUpdate, time.Now().Unix()) -} - -// UpdateMicrophoneConfigMetrics updates Prometheus metrics with microphone configuration -func UpdateMicrophoneConfigMetrics(config AudioConfig) { - metricsUpdateMutex.Lock() - defer metricsUpdateMutex.Unlock() - - microphoneConfigQuality.Set(float64(config.Quality)) - microphoneConfigBitrate.Set(float64(config.Bitrate)) - microphoneConfigSampleRate.Set(float64(config.SampleRate)) - microphoneConfigChannels.Set(float64(config.Channels)) - - atomic.StoreInt64(&lastMetricsUpdate, time.Now().Unix()) -} - // UpdateAdaptiveBufferMetrics updates Prometheus metrics with adaptive buffer information func UpdateAdaptiveBufferMetrics(inputBufferSize, outputBufferSize int, cpuPercent, memoryPercent float64, adjustmentMade bool) { metricsUpdateMutex.Lock() diff --git a/internal/audio/core_metrics_registry.go b/internal/audio/core_metrics_registry.go index 7ecdffd3..b842af08 100644 --- a/internal/audio/core_metrics_registry.go +++ b/internal/audio/core_metrics_registry.go @@ -14,8 +14,6 @@ type MetricsRegistry struct { mu sync.RWMutex audioMetrics AudioMetrics audioInputMetrics AudioInputMetrics - audioConfig AudioConfig - microphoneConfig AudioConfig lastUpdate int64 // Unix timestamp } @@ -56,28 +54,6 @@ func (mr *MetricsRegistry) UpdateAudioInputMetrics(metrics AudioInputMetrics) { UpdateMicrophoneMetrics(convertAudioInputMetricsToUnified(metrics)) } -// UpdateAudioConfig updates the centralized audio configuration -func (mr *MetricsRegistry) UpdateAudioConfig(config AudioConfig) { - mr.mu.Lock() - mr.audioConfig = config - mr.lastUpdate = time.Now().Unix() - mr.mu.Unlock() - - // Update Prometheus metrics directly - UpdateAudioConfigMetrics(config) -} - -// UpdateMicrophoneConfig updates the centralized microphone configuration -func (mr *MetricsRegistry) UpdateMicrophoneConfig(config AudioConfig) { - mr.mu.Lock() - mr.microphoneConfig = config - mr.lastUpdate = time.Now().Unix() - mr.mu.Unlock() - - // Update Prometheus metrics directly - UpdateMicrophoneConfigMetrics(config) -} - // GetAudioMetrics returns the current audio output metrics func (mr *MetricsRegistry) GetAudioMetrics() AudioMetrics { mr.mu.RLock() @@ -92,20 +68,6 @@ func (mr *MetricsRegistry) GetAudioInputMetrics() AudioInputMetrics { return mr.audioInputMetrics } -// GetAudioConfig returns the current audio configuration -func (mr *MetricsRegistry) GetAudioConfig() AudioConfig { - mr.mu.RLock() - defer mr.mu.RUnlock() - return mr.audioConfig -} - -// GetMicrophoneConfig returns the current microphone configuration -func (mr *MetricsRegistry) GetMicrophoneConfig() AudioConfig { - mr.mu.RLock() - defer mr.mu.RUnlock() - return mr.microphoneConfig -} - // GetLastUpdate returns the timestamp of the last metrics update func (mr *MetricsRegistry) GetLastUpdate() time.Time { timestamp := atomic.LoadInt64(&mr.lastUpdate) @@ -132,20 +94,11 @@ func (mr *MetricsRegistry) StartMetricsCollector() { mr.UpdateAudioInputMetrics(metrics) } - // Collect audio output metrics directly from global metrics variable to avoid circular dependency - audioMetrics := AudioMetrics{ - FramesReceived: atomic.LoadInt64(&metrics.FramesReceived), - FramesDropped: atomic.LoadInt64(&metrics.FramesDropped), - BytesProcessed: atomic.LoadInt64(&metrics.BytesProcessed), - ConnectionDrops: atomic.LoadInt64(&metrics.ConnectionDrops), - LastFrameTime: metrics.LastFrameTime, - AverageLatency: metrics.AverageLatency, - } - mr.UpdateAudioMetrics(audioMetrics) - - // Collect configuration directly from global variables to avoid circular dependency - mr.UpdateAudioConfig(currentConfig) - mr.UpdateMicrophoneConfig(currentMicrophoneConfig) + // Collect audio output metrics from global audio output manager + // Note: We need to get metrics from the actual audio output system + // For now, we'll use the global metrics variable from quality_presets.go + globalAudioMetrics := GetGlobalAudioMetrics() + mr.UpdateAudioMetrics(globalAudioMetrics) } }() } diff --git a/internal/audio/latency_monitor.go b/internal/audio/latency_monitor.go deleted file mode 100644 index 40b2381d..00000000 --- a/internal/audio/latency_monitor.go +++ /dev/null @@ -1,333 +0,0 @@ -package audio - -import ( - "context" - "sync" - "sync/atomic" - "time" - - "github.com/rs/zerolog" -) - -// LatencyMonitor tracks and optimizes audio latency in real-time -type LatencyMonitor struct { - // Atomic fields MUST be first for ARM32 alignment (int64 fields need 8-byte alignment) - currentLatency int64 // Current latency in nanoseconds (atomic) - averageLatency int64 // Rolling average latency in nanoseconds (atomic) - minLatency int64 // Minimum observed latency in nanoseconds (atomic) - maxLatency int64 // Maximum observed latency in nanoseconds (atomic) - latencySamples int64 // Number of latency samples collected (atomic) - jitterAccumulator int64 // Accumulated jitter for variance calculation (atomic) - lastOptimization int64 // Timestamp of last optimization in nanoseconds (atomic) - - config LatencyConfig - logger zerolog.Logger - - // Control channels - ctx context.Context - cancel context.CancelFunc - wg sync.WaitGroup - - // Optimization callbacks - optimizationCallbacks []OptimizationCallback - mutex sync.RWMutex - - // Performance tracking - latencyHistory []LatencyMeasurement - historyMutex sync.RWMutex -} - -// LatencyConfig holds configuration for latency monitoring -type LatencyConfig struct { - TargetLatency time.Duration // Target latency to maintain - MaxLatency time.Duration // Maximum acceptable latency - OptimizationInterval time.Duration // How often to run optimization - HistorySize int // Number of latency measurements to keep - JitterThreshold time.Duration // Jitter threshold for optimization - AdaptiveThreshold float64 // Threshold for adaptive adjustments (0.0-1.0) -} - -// LatencyMeasurement represents a single latency measurement -type LatencyMeasurement struct { - Timestamp time.Time - Latency time.Duration - Jitter time.Duration - Source string // Source of the measurement (e.g., "input", "output", "processing") -} - -// OptimizationCallback is called when latency optimization is triggered -type OptimizationCallback func(metrics LatencyMetrics) error - -// LatencyMetrics provides comprehensive latency statistics -type LatencyMetrics struct { - Current time.Duration - Average time.Duration - Min time.Duration - Max time.Duration - Jitter time.Duration - SampleCount int64 - Trend LatencyTrend -} - -// LatencyTrend indicates the direction of latency changes -type LatencyTrend int - -const ( - LatencyTrendStable LatencyTrend = iota - LatencyTrendIncreasing - LatencyTrendDecreasing - LatencyTrendVolatile -) - -// DefaultLatencyConfig returns a sensible default configuration -func DefaultLatencyConfig() LatencyConfig { - config := GetConfig() - return LatencyConfig{ - TargetLatency: config.LatencyMonitorTarget, - MaxLatency: config.MaxLatencyThreshold, - OptimizationInterval: config.LatencyOptimizationInterval, - HistorySize: config.LatencyHistorySize, - JitterThreshold: config.JitterThreshold, - AdaptiveThreshold: config.LatencyAdaptiveThreshold, - } -} - -// NewLatencyMonitor creates a new latency monitoring system -func NewLatencyMonitor(config LatencyConfig, logger zerolog.Logger) *LatencyMonitor { - // Validate latency configuration - if err := ValidateLatencyConfig(config); err != nil { - // Log validation error and use default configuration - logger.Error().Err(err).Msg("Invalid latency configuration provided, using defaults") - config = DefaultLatencyConfig() - } - - ctx, cancel := context.WithCancel(context.Background()) - - return &LatencyMonitor{ - config: config, - logger: logger.With().Str("component", "latency-monitor").Logger(), - ctx: ctx, - cancel: cancel, - latencyHistory: make([]LatencyMeasurement, 0, config.HistorySize), - minLatency: int64(time.Hour), // Initialize to high value - } -} - -// Start begins latency monitoring and optimization -func (lm *LatencyMonitor) Start() { - lm.wg.Add(1) - go lm.monitoringLoop() -} - -// Stop stops the latency monitor -func (lm *LatencyMonitor) Stop() { - lm.cancel() - lm.wg.Wait() -} - -// RecordLatency records a new latency measurement -func (lm *LatencyMonitor) RecordLatency(latency time.Duration, source string) { - now := time.Now() - latencyNanos := latency.Nanoseconds() - - // Update atomic counters - atomic.StoreInt64(&lm.currentLatency, latencyNanos) - atomic.AddInt64(&lm.latencySamples, 1) - - // Update min/max - for { - oldMin := atomic.LoadInt64(&lm.minLatency) - if latencyNanos >= oldMin || atomic.CompareAndSwapInt64(&lm.minLatency, oldMin, latencyNanos) { - break - } - } - - for { - oldMax := atomic.LoadInt64(&lm.maxLatency) - if latencyNanos <= oldMax || atomic.CompareAndSwapInt64(&lm.maxLatency, oldMax, latencyNanos) { - break - } - } - - // Update rolling average using exponential moving average - oldAvg := atomic.LoadInt64(&lm.averageLatency) - newAvg := oldAvg + (latencyNanos-oldAvg)/10 // Alpha = 0.1 - atomic.StoreInt64(&lm.averageLatency, newAvg) - - // Calculate jitter (difference from average) - jitter := latencyNanos - newAvg - if jitter < 0 { - jitter = -jitter - } - atomic.AddInt64(&lm.jitterAccumulator, jitter) - - // Store in history - lm.historyMutex.Lock() - measurement := LatencyMeasurement{ - Timestamp: now, - Latency: latency, - Jitter: time.Duration(jitter), - Source: source, - } - - if len(lm.latencyHistory) >= lm.config.HistorySize { - // Remove oldest measurement - copy(lm.latencyHistory, lm.latencyHistory[1:]) - lm.latencyHistory[len(lm.latencyHistory)-1] = measurement - } else { - lm.latencyHistory = append(lm.latencyHistory, measurement) - } - lm.historyMutex.Unlock() -} - -// GetMetrics returns current latency metrics -func (lm *LatencyMonitor) GetMetrics() LatencyMetrics { - current := atomic.LoadInt64(&lm.currentLatency) - average := atomic.LoadInt64(&lm.averageLatency) - min := atomic.LoadInt64(&lm.minLatency) - max := atomic.LoadInt64(&lm.maxLatency) - samples := atomic.LoadInt64(&lm.latencySamples) - jitterSum := atomic.LoadInt64(&lm.jitterAccumulator) - - var jitter time.Duration - if samples > 0 { - jitter = time.Duration(jitterSum / samples) - } - - return LatencyMetrics{ - Current: time.Duration(current), - Average: time.Duration(average), - Min: time.Duration(min), - Max: time.Duration(max), - Jitter: jitter, - SampleCount: samples, - Trend: lm.calculateTrend(), - } -} - -// AddOptimizationCallback adds a callback for latency optimization -func (lm *LatencyMonitor) AddOptimizationCallback(callback OptimizationCallback) { - lm.mutex.Lock() - lm.optimizationCallbacks = append(lm.optimizationCallbacks, callback) - lm.mutex.Unlock() -} - -// monitoringLoop runs the main monitoring and optimization loop -func (lm *LatencyMonitor) monitoringLoop() { - defer lm.wg.Done() - - ticker := time.NewTicker(lm.config.OptimizationInterval) - defer ticker.Stop() - - for { - select { - case <-lm.ctx.Done(): - return - case <-ticker.C: - lm.runOptimization() - } - } -} - -// runOptimization checks if optimization is needed and triggers callbacks with threshold validation. -// -// Validation Rules: -// - Current latency must not exceed MaxLatency (default: 200ms) -// - Average latency checked against adaptive threshold: TargetLatency * (1 + AdaptiveThreshold) -// - Jitter must not exceed JitterThreshold (default: 20ms) -// - All latency values must be non-negative durations -// -// Optimization Triggers: -// - Current latency > MaxLatency: Immediate optimization needed -// - Average latency > adaptive threshold: Gradual optimization needed -// - Jitter > JitterThreshold: Stability optimization needed -// -// Threshold Calculations: -// - Adaptive threshold = TargetLatency * (1.0 + AdaptiveThreshold) -// - Default: 50ms * (1.0 + 0.8) = 90ms adaptive threshold -// - Provides buffer above target before triggering optimization -// -// The function ensures real-time audio performance by monitoring multiple -// latency metrics and triggering optimization callbacks when thresholds are exceeded. -func (lm *LatencyMonitor) runOptimization() { - metrics := lm.GetMetrics() - - // Check if optimization is needed - needsOptimization := false - - // Check if current latency exceeds threshold - if metrics.Current > lm.config.MaxLatency { - needsOptimization = true - lm.logger.Warn().Dur("current_latency", metrics.Current).Dur("max_latency", lm.config.MaxLatency).Msg("latency exceeds maximum threshold") - } - - // Check if average latency is above adaptive threshold - adaptiveThreshold := time.Duration(float64(lm.config.TargetLatency.Nanoseconds()) * (1.0 + lm.config.AdaptiveThreshold)) - if metrics.Average > adaptiveThreshold { - needsOptimization = true - } - - // Check if jitter is too high - if metrics.Jitter > lm.config.JitterThreshold { - needsOptimization = true - } - - if needsOptimization { - atomic.StoreInt64(&lm.lastOptimization, time.Now().UnixNano()) - - // Run optimization callbacks - lm.mutex.RLock() - callbacks := make([]OptimizationCallback, len(lm.optimizationCallbacks)) - copy(callbacks, lm.optimizationCallbacks) - lm.mutex.RUnlock() - - for _, callback := range callbacks { - if err := callback(metrics); err != nil { - lm.logger.Error().Err(err).Msg("optimization callback failed") - } - } - } -} - -// calculateTrend analyzes recent latency measurements to determine trend -func (lm *LatencyMonitor) calculateTrend() LatencyTrend { - lm.historyMutex.RLock() - defer lm.historyMutex.RUnlock() - - if len(lm.latencyHistory) < 10 { - return LatencyTrendStable - } - - // Analyze last 10 measurements - recentMeasurements := lm.latencyHistory[len(lm.latencyHistory)-10:] - - var increasing, decreasing int - for i := 1; i < len(recentMeasurements); i++ { - if recentMeasurements[i].Latency > recentMeasurements[i-1].Latency { - increasing++ - } else if recentMeasurements[i].Latency < recentMeasurements[i-1].Latency { - decreasing++ - } - } - - // Determine trend based on direction changes - if increasing > 6 { - return LatencyTrendIncreasing - } else if decreasing > 6 { - return LatencyTrendDecreasing - } else if increasing+decreasing > 7 { - return LatencyTrendVolatile - } - - return LatencyTrendStable -} - -// GetLatencyHistory returns a copy of recent latency measurements -func (lm *LatencyMonitor) GetLatencyHistory() []LatencyMeasurement { - lm.historyMutex.RLock() - defer lm.historyMutex.RUnlock() - - history := make([]LatencyMeasurement, len(lm.latencyHistory)) - copy(history, lm.latencyHistory) - return history -} diff --git a/internal/audio/quality_presets.go b/internal/audio/quality_presets.go index 1a381feb..0c0b259e 100644 --- a/internal/audio/quality_presets.go +++ b/internal/audio/quality_presets.go @@ -330,6 +330,11 @@ func GetMicrophoneConfig() AudioConfig { return currentMicrophoneConfig } +// GetGlobalAudioMetrics returns the current global audio metrics +func GetGlobalAudioMetrics() AudioMetrics { + return metrics +} + // Batched metrics to reduce atomic operations frequency var ( batchedFramesReceived int64