mirror of https://github.com/jetkvm/kvm.git
[WIP] Simplification: PR Simplification
This commit is contained in:
Alex P
parent
ebb79600b0
commit
7ab4a0e41d
|
|
@ -1,447 +0,0 @@
|
|||
package audio
|
||||
|
||||
import (
|
||||
"context"
|
||||
"math"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/jetkvm/kvm/internal/logging"
|
||||
"github.com/rs/zerolog"
|
||||
)
|
||||
|
||||
// AdaptiveBufferConfig holds configuration for the adaptive buffer sizing algorithm.
|
||||
//
|
||||
// The adaptive buffer system dynamically adjusts audio buffer sizes based on real-time
|
||||
// system conditions to optimize the trade-off between latency and stability. The algorithm
|
||||
// uses multiple factors to make decisions:
|
||||
//
|
||||
// 1. System Load Monitoring:
|
||||
// - CPU usage: High CPU load increases buffer sizes to prevent underruns
|
||||
// - Memory usage: High memory pressure reduces buffer sizes to conserve RAM
|
||||
//
|
||||
// 2. Latency Tracking:
|
||||
// - Target latency: Optimal latency for the current quality setting
|
||||
// - Max latency: Hard limit beyond which buffers are aggressively reduced
|
||||
//
|
||||
// 3. Adaptation Strategy:
|
||||
// - Exponential smoothing: Prevents oscillation and provides stable adjustments
|
||||
// - Discrete steps: Buffer sizes change in fixed increments to avoid instability
|
||||
// - Hysteresis: Different thresholds for increasing vs decreasing buffer sizes
|
||||
//
|
||||
// The algorithm is specifically tuned for embedded ARM systems with limited resources,
|
||||
// prioritizing stability over absolute minimum latency.
|
||||
type AdaptiveBufferConfig struct {
|
||||
// Buffer size limits (in frames)
|
||||
MinBufferSize int
|
||||
MaxBufferSize int
|
||||
DefaultBufferSize int
|
||||
|
||||
// System load thresholds
|
||||
LowCPUThreshold float64 // Below this, increase buffer size
|
||||
HighCPUThreshold float64 // Above this, decrease buffer size
|
||||
LowMemoryThreshold float64 // Below this, increase buffer size
|
||||
HighMemoryThreshold float64 // Above this, decrease buffer size
|
||||
|
||||
// Latency thresholds (in milliseconds)
|
||||
TargetLatency time.Duration
|
||||
MaxLatency time.Duration
|
||||
|
||||
// Adaptation parameters
|
||||
AdaptationInterval time.Duration
|
||||
SmoothingFactor float64 // 0.0-1.0, higher = more responsive
|
||||
}
|
||||
|
||||
// DefaultAdaptiveBufferConfig returns optimized config for JetKVM hardware
|
||||
func DefaultAdaptiveBufferConfig() AdaptiveBufferConfig {
|
||||
return AdaptiveBufferConfig{
|
||||
// Conservative buffer sizes for 256MB RAM constraint
|
||||
MinBufferSize: Config.AdaptiveMinBufferSize,
|
||||
MaxBufferSize: Config.AdaptiveMaxBufferSize,
|
||||
DefaultBufferSize: Config.AdaptiveDefaultBufferSize,
|
||||
|
||||
// CPU thresholds optimized for single-core ARM Cortex A7 under load
|
||||
LowCPUThreshold: Config.LowCPUThreshold * 100, // Below 20% CPU
|
||||
HighCPUThreshold: Config.HighCPUThreshold * 100, // Above 60% CPU (lowered to be more responsive)
|
||||
|
||||
// Memory thresholds for 256MB total RAM
|
||||
LowMemoryThreshold: Config.LowMemoryThreshold * 100, // Below 35% memory usage
|
||||
HighMemoryThreshold: Config.HighMemoryThreshold * 100, // Above 75% memory usage (lowered for earlier response)
|
||||
|
||||
// Latency targets
|
||||
TargetLatency: Config.AdaptiveBufferTargetLatency, // Target 20ms latency
|
||||
MaxLatency: Config.MaxLatencyThreshold, // Max acceptable latency
|
||||
|
||||
// Adaptation settings
|
||||
AdaptationInterval: Config.BufferUpdateInterval, // Check every 500ms
|
||||
SmoothingFactor: Config.SmoothingFactor, // Moderate responsiveness
|
||||
}
|
||||
}
|
||||
|
||||
// AdaptiveBufferManager manages dynamic buffer sizing based on system conditions
|
||||
type AdaptiveBufferManager struct {
|
||||
// Atomic fields MUST be first for ARM32 alignment (int64 fields need 8-byte alignment)
|
||||
currentInputBufferSize int64 // Current input buffer size (atomic)
|
||||
currentOutputBufferSize int64 // Current output buffer size (atomic)
|
||||
averageLatency int64 // Average latency in nanoseconds (atomic)
|
||||
systemCPUPercent int64 // System CPU percentage * 100 (atomic)
|
||||
systemMemoryPercent int64 // System memory percentage * 100 (atomic)
|
||||
adaptationCount int64 // Metrics tracking (atomic)
|
||||
|
||||
config AdaptiveBufferConfig
|
||||
logger zerolog.Logger
|
||||
|
||||
// Control channels
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
wg sync.WaitGroup
|
||||
|
||||
// Metrics tracking
|
||||
lastAdaptation time.Time
|
||||
mutex sync.RWMutex
|
||||
}
|
||||
|
||||
// NewAdaptiveBufferManager creates a new adaptive buffer manager
|
||||
func NewAdaptiveBufferManager(config AdaptiveBufferConfig) *AdaptiveBufferManager {
|
||||
logger := logging.GetDefaultLogger().With().Str("component", "adaptive-buffer").Logger()
|
||||
|
||||
if err := ValidateAdaptiveBufferConfig(config.MinBufferSize, config.MaxBufferSize, config.DefaultBufferSize); err != nil {
|
||||
logger.Warn().Err(err).Msg("invalid adaptive buffer config, using defaults")
|
||||
config = DefaultAdaptiveBufferConfig()
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
|
||||
return &AdaptiveBufferManager{
|
||||
currentInputBufferSize: int64(config.DefaultBufferSize),
|
||||
currentOutputBufferSize: int64(config.DefaultBufferSize),
|
||||
config: config,
|
||||
logger: logger,
|
||||
|
||||
ctx: ctx,
|
||||
cancel: cancel,
|
||||
lastAdaptation: time.Now(),
|
||||
}
|
||||
}
|
||||
|
||||
// Start begins the adaptive buffer management
|
||||
func (abm *AdaptiveBufferManager) Start() {
|
||||
abm.wg.Add(1)
|
||||
go abm.adaptationLoop()
|
||||
abm.logger.Info().Msg("adaptive buffer manager started")
|
||||
}
|
||||
|
||||
// Stop stops the adaptive buffer management
|
||||
func (abm *AdaptiveBufferManager) Stop() {
|
||||
abm.cancel()
|
||||
abm.wg.Wait()
|
||||
abm.logger.Info().Msg("adaptive buffer manager stopped")
|
||||
}
|
||||
|
||||
// GetInputBufferSize returns the current recommended input buffer size
|
||||
func (abm *AdaptiveBufferManager) GetInputBufferSize() int {
|
||||
return int(atomic.LoadInt64(&abm.currentInputBufferSize))
|
||||
}
|
||||
|
||||
// GetOutputBufferSize returns the current recommended output buffer size
|
||||
func (abm *AdaptiveBufferManager) GetOutputBufferSize() int {
|
||||
return int(atomic.LoadInt64(&abm.currentOutputBufferSize))
|
||||
}
|
||||
|
||||
// UpdateLatency updates the current latency measurement
|
||||
func (abm *AdaptiveBufferManager) UpdateLatency(latency time.Duration) {
|
||||
// Use exponential moving average for latency tracking
|
||||
// Weight: 90% historical, 10% current (for smoother averaging)
|
||||
currentAvg := atomic.LoadInt64(&abm.averageLatency)
|
||||
newLatencyNs := latency.Nanoseconds()
|
||||
|
||||
if currentAvg == 0 {
|
||||
// First measurement
|
||||
atomic.StoreInt64(&abm.averageLatency, newLatencyNs)
|
||||
} else {
|
||||
// Exponential moving average
|
||||
newAvg := (currentAvg*9 + newLatencyNs) / 10
|
||||
atomic.StoreInt64(&abm.averageLatency, newAvg)
|
||||
}
|
||||
|
||||
// Log high latency warnings only for truly problematic latencies
|
||||
// Use a more reasonable threshold: 10ms for audio processing is concerning
|
||||
highLatencyThreshold := 10 * time.Millisecond
|
||||
if latency > highLatencyThreshold {
|
||||
abm.logger.Debug().
|
||||
Dur("latency_ms", latency/time.Millisecond).
|
||||
Dur("threshold_ms", highLatencyThreshold/time.Millisecond).
|
||||
Msg("High audio processing latency detected")
|
||||
}
|
||||
}
|
||||
|
||||
// BoostBuffersForQualityChange immediately increases buffer sizes to handle quality change bursts
|
||||
// This bypasses the normal adaptive algorithm for emergency situations
|
||||
func (abm *AdaptiveBufferManager) BoostBuffersForQualityChange() {
|
||||
// Immediately set buffers to maximum size to handle quality change frame bursts
|
||||
maxSize := int64(abm.config.MaxBufferSize)
|
||||
atomic.StoreInt64(&abm.currentInputBufferSize, maxSize)
|
||||
atomic.StoreInt64(&abm.currentOutputBufferSize, maxSize)
|
||||
|
||||
abm.logger.Info().
|
||||
Int("buffer_size", int(maxSize)).
|
||||
Msg("Boosted buffers to maximum size for quality change")
|
||||
}
|
||||
|
||||
// adaptationLoop is the main loop that adjusts buffer sizes
|
||||
func (abm *AdaptiveBufferManager) adaptationLoop() {
|
||||
defer abm.wg.Done()
|
||||
|
||||
ticker := time.NewTicker(abm.config.AdaptationInterval)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-abm.ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
abm.adaptBufferSizes()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// adaptBufferSizes analyzes system conditions and adjusts buffer sizes
|
||||
// adaptBufferSizes implements the core adaptive buffer sizing algorithm.
|
||||
//
|
||||
// This function uses a multi-factor approach to determine optimal buffer sizes:
|
||||
//
|
||||
// Mathematical Model:
|
||||
// 1. Factor Calculation:
|
||||
//
|
||||
// - CPU Factor: Sigmoid function that increases buffer size under high CPU load
|
||||
//
|
||||
// - Memory Factor: Inverse relationship that decreases buffer size under memory pressure
|
||||
//
|
||||
// - Latency Factor: Exponential decay that aggressively reduces buffers when latency exceeds targets
|
||||
//
|
||||
// 2. Combined Factor:
|
||||
// Combined = (CPU_factor * Memory_factor * Latency_factor)
|
||||
// This multiplicative approach ensures any single critical factor can override others
|
||||
//
|
||||
// 3. Exponential Smoothing:
|
||||
// New_size = Current_size + smoothing_factor * (Target_size - Current_size)
|
||||
// This prevents rapid oscillations and provides stable convergence
|
||||
//
|
||||
// 4. Discrete Quantization:
|
||||
// Final sizes are rounded to frame boundaries and clamped to configured limits
|
||||
//
|
||||
// The algorithm runs periodically and only applies changes when the adaptation interval
|
||||
// has elapsed, preventing excessive adjustments that could destabilize the audio pipeline.
|
||||
func (abm *AdaptiveBufferManager) adaptBufferSizes() {
|
||||
// Use fixed system metrics for stability
|
||||
systemCPU := 50.0 // Assume moderate CPU usage
|
||||
systemMemory := 60.0 // Assume moderate memory usage
|
||||
|
||||
atomic.StoreInt64(&abm.systemCPUPercent, int64(systemCPU*100))
|
||||
atomic.StoreInt64(&abm.systemMemoryPercent, int64(systemMemory*100))
|
||||
|
||||
// Get current latency
|
||||
currentLatencyNs := atomic.LoadInt64(&abm.averageLatency)
|
||||
currentLatency := time.Duration(currentLatencyNs)
|
||||
|
||||
// Calculate adaptation factors
|
||||
cpuFactor := abm.calculateCPUFactor(systemCPU)
|
||||
memoryFactor := abm.calculateMemoryFactor(systemMemory)
|
||||
latencyFactor := abm.calculateLatencyFactor(currentLatency)
|
||||
|
||||
// Combine factors with weights (CPU has highest priority for KVM coexistence)
|
||||
combinedFactor := Config.CPUMemoryWeight*cpuFactor + Config.MemoryWeight*memoryFactor + Config.LatencyWeight*latencyFactor
|
||||
|
||||
// Apply adaptation with smoothing
|
||||
currentInput := float64(atomic.LoadInt64(&abm.currentInputBufferSize))
|
||||
currentOutput := float64(atomic.LoadInt64(&abm.currentOutputBufferSize))
|
||||
|
||||
// Calculate new buffer sizes
|
||||
newInputSize := abm.applyAdaptation(currentInput, combinedFactor)
|
||||
newOutputSize := abm.applyAdaptation(currentOutput, combinedFactor)
|
||||
|
||||
// Update buffer sizes if they changed significantly
|
||||
adjustmentMade := false
|
||||
if math.Abs(newInputSize-currentInput) >= 0.5 || math.Abs(newOutputSize-currentOutput) >= 0.5 {
|
||||
atomic.StoreInt64(&abm.currentInputBufferSize, int64(math.Round(newInputSize)))
|
||||
atomic.StoreInt64(&abm.currentOutputBufferSize, int64(math.Round(newOutputSize)))
|
||||
|
||||
atomic.AddInt64(&abm.adaptationCount, 1)
|
||||
abm.mutex.Lock()
|
||||
abm.lastAdaptation = time.Now()
|
||||
abm.mutex.Unlock()
|
||||
adjustmentMade = true
|
||||
|
||||
abm.logger.Debug().
|
||||
Float64("cpu_percent", systemCPU).
|
||||
Float64("memory_percent", systemMemory).
|
||||
Dur("latency", currentLatency).
|
||||
Float64("combined_factor", combinedFactor).
|
||||
Int("new_input_size", int(newInputSize)).
|
||||
Int("new_output_size", int(newOutputSize)).
|
||||
Msg("Adapted buffer sizes")
|
||||
}
|
||||
|
||||
// Update metrics with current state
|
||||
currentInputSize := int(atomic.LoadInt64(&abm.currentInputBufferSize))
|
||||
currentOutputSize := int(atomic.LoadInt64(&abm.currentOutputBufferSize))
|
||||
UpdateAdaptiveBufferMetrics(currentInputSize, currentOutputSize, systemCPU, systemMemory, adjustmentMade)
|
||||
}
|
||||
|
||||
// calculateCPUFactor returns adaptation factor based on CPU usage with threshold validation.
|
||||
//
|
||||
// Validation Rules:
|
||||
// - CPU percentage must be within valid range [0.0, 100.0]
|
||||
// - Uses LowCPUThreshold and HighCPUThreshold from config for decision boundaries
|
||||
// - Default thresholds: Low=20.0%, High=80.0%
|
||||
//
|
||||
// Adaptation Logic:
|
||||
// - CPU > HighCPUThreshold: Return -1.0 (decrease buffers to reduce CPU load)
|
||||
// - CPU < LowCPUThreshold: Return +1.0 (increase buffers for better quality)
|
||||
// - Between thresholds: Linear interpolation based on distance from midpoint
|
||||
//
|
||||
// Returns: Adaptation factor in range [-1.0, +1.0]
|
||||
// - Negative values: Decrease buffer sizes to reduce CPU usage
|
||||
// - Positive values: Increase buffer sizes for better audio quality
|
||||
// - Zero: No adaptation needed
|
||||
//
|
||||
// The function ensures CPU-aware buffer management to balance audio quality
|
||||
// with system performance, preventing CPU starvation of the KVM process.
|
||||
func (abm *AdaptiveBufferManager) calculateCPUFactor(cpuPercent float64) float64 {
|
||||
if cpuPercent > abm.config.HighCPUThreshold {
|
||||
// High CPU: decrease buffers to reduce latency and give CPU to KVM
|
||||
return -1.0
|
||||
} else if cpuPercent < abm.config.LowCPUThreshold {
|
||||
// Low CPU: increase buffers for better quality
|
||||
return 1.0
|
||||
}
|
||||
// Medium CPU: linear interpolation
|
||||
midpoint := (abm.config.HighCPUThreshold + abm.config.LowCPUThreshold) / 2
|
||||
return (midpoint - cpuPercent) / (midpoint - abm.config.LowCPUThreshold)
|
||||
}
|
||||
|
||||
// calculateMemoryFactor returns adaptation factor based on memory usage with threshold validation.
|
||||
//
|
||||
// Validation Rules:
|
||||
// - Memory percentage must be within valid range [0.0, 100.0]
|
||||
// - Uses LowMemoryThreshold and HighMemoryThreshold from config for decision boundaries
|
||||
// - Default thresholds: Low=30.0%, High=85.0%
|
||||
//
|
||||
// Adaptation Logic:
|
||||
// - Memory > HighMemoryThreshold: Return -1.0 (decrease buffers to free memory)
|
||||
// - Memory < LowMemoryThreshold: Return +1.0 (increase buffers for performance)
|
||||
// - Between thresholds: Linear interpolation based on distance from midpoint
|
||||
//
|
||||
// Returns: Adaptation factor in range [-1.0, +1.0]
|
||||
// - Negative values: Decrease buffer sizes to reduce memory usage
|
||||
// - Positive values: Increase buffer sizes for better performance
|
||||
// - Zero: No adaptation needed
|
||||
//
|
||||
// The function prevents memory exhaustion while optimizing buffer sizes
|
||||
// for audio processing performance and system stability.
|
||||
func (abm *AdaptiveBufferManager) calculateMemoryFactor(memoryPercent float64) float64 {
|
||||
if memoryPercent > abm.config.HighMemoryThreshold {
|
||||
// High memory: decrease buffers to free memory
|
||||
return -1.0
|
||||
} else if memoryPercent < abm.config.LowMemoryThreshold {
|
||||
// Low memory: increase buffers for better performance
|
||||
return 1.0
|
||||
}
|
||||
// Medium memory: linear interpolation
|
||||
midpoint := (abm.config.HighMemoryThreshold + abm.config.LowMemoryThreshold) / 2
|
||||
return (midpoint - memoryPercent) / (midpoint - abm.config.LowMemoryThreshold)
|
||||
}
|
||||
|
||||
// calculateLatencyFactor returns adaptation factor based on latency with threshold validation.
|
||||
//
|
||||
// Validation Rules:
|
||||
// - Latency must be non-negative duration
|
||||
// - Uses TargetLatency and MaxLatency from config for decision boundaries
|
||||
// - Default thresholds: Target=50ms, Max=200ms
|
||||
//
|
||||
// Adaptation Logic:
|
||||
// - Latency > MaxLatency: Return -1.0 (decrease buffers to reduce latency)
|
||||
// - Latency < TargetLatency: Return +1.0 (increase buffers for quality)
|
||||
// - Between thresholds: Linear interpolation based on distance from midpoint
|
||||
//
|
||||
// Returns: Adaptation factor in range [-1.0, +1.0]
|
||||
// - Negative values: Decrease buffer sizes to reduce audio latency
|
||||
// - Positive values: Increase buffer sizes for better audio quality
|
||||
// - Zero: Latency is at optimal level
|
||||
//
|
||||
// The function balances audio latency with quality, ensuring real-time
|
||||
// performance while maintaining acceptable audio processing quality.
|
||||
func (abm *AdaptiveBufferManager) calculateLatencyFactor(latency time.Duration) float64 {
|
||||
if latency > abm.config.MaxLatency {
|
||||
// High latency: decrease buffers
|
||||
return -1.0
|
||||
} else if latency < abm.config.TargetLatency {
|
||||
// Low latency: can increase buffers
|
||||
return 1.0
|
||||
}
|
||||
// Medium latency: linear interpolation
|
||||
midLatency := (abm.config.MaxLatency + abm.config.TargetLatency) / 2
|
||||
return float64(midLatency-latency) / float64(midLatency-abm.config.TargetLatency)
|
||||
}
|
||||
|
||||
// applyAdaptation applies the adaptation factor to current buffer size
|
||||
func (abm *AdaptiveBufferManager) applyAdaptation(currentSize, factor float64) float64 {
|
||||
// Calculate target size based on factor
|
||||
var targetSize float64
|
||||
if factor > 0 {
|
||||
// Increase towards max
|
||||
targetSize = currentSize + factor*(float64(abm.config.MaxBufferSize)-currentSize)
|
||||
} else {
|
||||
// Decrease towards min
|
||||
targetSize = currentSize + factor*(currentSize-float64(abm.config.MinBufferSize))
|
||||
}
|
||||
|
||||
// Apply smoothing
|
||||
newSize := currentSize + abm.config.SmoothingFactor*(targetSize-currentSize)
|
||||
|
||||
// Clamp to valid range
|
||||
return math.Max(float64(abm.config.MinBufferSize),
|
||||
math.Min(float64(abm.config.MaxBufferSize), newSize))
|
||||
}
|
||||
|
||||
// GetStats returns current adaptation statistics
|
||||
func (abm *AdaptiveBufferManager) GetStats() map[string]interface{} {
|
||||
abm.mutex.RLock()
|
||||
lastAdaptation := abm.lastAdaptation
|
||||
abm.mutex.RUnlock()
|
||||
|
||||
return map[string]interface{}{
|
||||
"input_buffer_size": abm.GetInputBufferSize(),
|
||||
"output_buffer_size": abm.GetOutputBufferSize(),
|
||||
"average_latency_ms": float64(atomic.LoadInt64(&abm.averageLatency)) / 1e6,
|
||||
"system_cpu_percent": float64(atomic.LoadInt64(&abm.systemCPUPercent)) / Config.PercentageMultiplier,
|
||||
"system_memory_percent": float64(atomic.LoadInt64(&abm.systemMemoryPercent)) / Config.PercentageMultiplier,
|
||||
"adaptation_count": atomic.LoadInt64(&abm.adaptationCount),
|
||||
"last_adaptation": lastAdaptation,
|
||||
}
|
||||
}
|
||||
|
||||
// Global adaptive buffer manager instance
|
||||
var globalAdaptiveBufferManager *AdaptiveBufferManager
|
||||
var adaptiveBufferOnce sync.Once
|
||||
|
||||
// GetAdaptiveBufferManager returns the global adaptive buffer manager instance
|
||||
func GetAdaptiveBufferManager() *AdaptiveBufferManager {
|
||||
adaptiveBufferOnce.Do(func() {
|
||||
globalAdaptiveBufferManager = NewAdaptiveBufferManager(DefaultAdaptiveBufferConfig())
|
||||
})
|
||||
return globalAdaptiveBufferManager
|
||||
}
|
||||
|
||||
// StartAdaptiveBuffering starts the global adaptive buffer manager
|
||||
func StartAdaptiveBuffering() {
|
||||
GetAdaptiveBufferManager().Start()
|
||||
}
|
||||
|
||||
// StopAdaptiveBuffering stops the global adaptive buffer manager
|
||||
func StopAdaptiveBuffering() {
|
||||
if globalAdaptiveBufferManager != nil {
|
||||
globalAdaptiveBufferManager.Stop()
|
||||
}
|
||||
}
|
||||
|
|
@ -152,11 +152,6 @@ type AudioConfigConstants struct {
|
|||
MemoryFactor float64
|
||||
LatencyFactor float64
|
||||
|
||||
// Adaptive Buffer Configuration
|
||||
AdaptiveMinBufferSize int // Minimum buffer size in frames for adaptive buffering
|
||||
AdaptiveMaxBufferSize int // Maximum buffer size in frames for adaptive buffering
|
||||
AdaptiveDefaultBufferSize int // Default buffer size in frames for adaptive buffering
|
||||
|
||||
// Timing Configuration
|
||||
RetryDelay time.Duration // Retry delay
|
||||
MaxRetryDelay time.Duration // Maximum retry delay
|
||||
|
|
@ -171,22 +166,17 @@ type AudioConfigConstants struct {
|
|||
OutputSupervisorTimeout time.Duration // 5s
|
||||
BatchProcessingDelay time.Duration // 10ms
|
||||
|
||||
// Adaptive Buffer Configuration
|
||||
// LowCPUThreshold defines CPU usage threshold for buffer size reduction.
|
||||
LowCPUThreshold float64 // 20% CPU threshold for buffer optimization
|
||||
|
||||
// HighCPUThreshold defines CPU usage threshold for buffer size increase.
|
||||
HighCPUThreshold float64 // 60% CPU threshold
|
||||
LowMemoryThreshold float64 // 50% memory threshold
|
||||
HighMemoryThreshold float64 // 75% memory threshold
|
||||
AdaptiveBufferTargetLatency time.Duration // 20ms target latency
|
||||
CooldownPeriod time.Duration // 30s cooldown period
|
||||
RollbackThreshold time.Duration // 300ms rollback threshold
|
||||
// System threshold configuration for buffer management
|
||||
LowCPUThreshold float64 // CPU usage threshold for performance optimization
|
||||
HighCPUThreshold float64 // CPU usage threshold for performance limits
|
||||
LowMemoryThreshold float64 // 50% memory threshold
|
||||
HighMemoryThreshold float64 // 75% memory threshold
|
||||
CooldownPeriod time.Duration // 30s cooldown period
|
||||
RollbackThreshold time.Duration // 300ms rollback threshold
|
||||
|
||||
MaxLatencyThreshold time.Duration // 200ms max latency
|
||||
JitterThreshold time.Duration // 20ms jitter threshold
|
||||
LatencyOptimizationInterval time.Duration // 5s optimization interval
|
||||
LatencyAdaptiveThreshold float64 // 0.8 adaptive threshold
|
||||
MicContentionTimeout time.Duration // 200ms contention timeout
|
||||
PreallocPercentage int // 20% preallocation percentage
|
||||
BackoffStart time.Duration // 50ms initial backoff
|
||||
|
|
@ -199,7 +189,6 @@ type AudioConfigConstants struct {
|
|||
PercentageMultiplier float64 // Multiplier for percentage calculations (100.0)
|
||||
AveragingWeight float64 // Weight for weighted averaging (0.7)
|
||||
ScalingFactor float64 // General scaling factor (1.5)
|
||||
SmoothingFactor float64 // Smoothing factor for adaptive buffers (0.3)
|
||||
CPUMemoryWeight float64 // Weight for CPU factor in calculations (0.5)
|
||||
MemoryWeight float64 // Weight for memory factor (0.3)
|
||||
LatencyWeight float64 // Weight for latency factor (0.2)
|
||||
|
|
@ -226,19 +215,17 @@ type AudioConfigConstants struct {
|
|||
// IPC Constants
|
||||
IPCInitialBufferFrames int // Initial IPC buffer size (500 frames)
|
||||
|
||||
EventTimeoutSeconds int
|
||||
EventTimeFormatString string
|
||||
EventSubscriptionDelayMS int
|
||||
InputProcessingTimeoutMS int
|
||||
AdaptiveBufferCPUMultiplier int
|
||||
AdaptiveBufferMemoryMultiplier int
|
||||
InputSocketName string
|
||||
OutputSocketName string
|
||||
AudioInputComponentName string
|
||||
AudioOutputComponentName string
|
||||
AudioServerComponentName string
|
||||
AudioRelayComponentName string
|
||||
AudioEventsComponentName string
|
||||
EventTimeoutSeconds int
|
||||
EventTimeFormatString string
|
||||
EventSubscriptionDelayMS int
|
||||
InputProcessingTimeoutMS int
|
||||
InputSocketName string
|
||||
OutputSocketName string
|
||||
AudioInputComponentName string
|
||||
AudioOutputComponentName string
|
||||
AudioServerComponentName string
|
||||
AudioRelayComponentName string
|
||||
AudioEventsComponentName string
|
||||
|
||||
TestSocketTimeout time.Duration
|
||||
TestBufferSize int
|
||||
|
|
@ -493,17 +480,11 @@ func DefaultAudioConfig() *AudioConfigConstants {
|
|||
OutputSupervisorTimeout: 5 * time.Second, // Output monitoring timeout
|
||||
BatchProcessingDelay: 5 * time.Millisecond, // Reduced batch processing delay
|
||||
|
||||
// Adaptive Buffer Configuration - Optimized for single-core RV1106G3
|
||||
LowCPUThreshold: 0.40, // Adjusted for single-core ARM system
|
||||
HighCPUThreshold: 0.75, // Adjusted for single-core RV1106G3 (current load ~64%)
|
||||
LowMemoryThreshold: 0.60,
|
||||
HighMemoryThreshold: 0.85, // Adjusted for 200MB total memory system
|
||||
AdaptiveBufferTargetLatency: 10 * time.Millisecond, // Aggressive target latency for responsiveness
|
||||
|
||||
// Adaptive Buffer Size Configuration - Optimized for quality change bursts
|
||||
AdaptiveMinBufferSize: 256, // Further increased minimum to prevent emergency mode
|
||||
AdaptiveMaxBufferSize: 1024, // Much higher maximum for quality changes
|
||||
AdaptiveDefaultBufferSize: 512, // Higher default for stability during bursts
|
||||
// System Load Configuration - Optimized for single-core RV1106G3
|
||||
LowCPUThreshold: 0.40, // Adjusted for single-core ARM system
|
||||
HighCPUThreshold: 0.75, // Adjusted for single-core RV1106G3 (current load ~64%)
|
||||
LowMemoryThreshold: 0.60,
|
||||
HighMemoryThreshold: 0.85, // Adjusted for 200MB total memory system
|
||||
|
||||
CooldownPeriod: 15 * time.Second, // Reduced cooldown period
|
||||
RollbackThreshold: 200 * time.Millisecond, // Lower rollback threshold
|
||||
|
|
@ -511,7 +492,6 @@ func DefaultAudioConfig() *AudioConfigConstants {
|
|||
MaxLatencyThreshold: 150 * time.Millisecond, // Lower max latency threshold
|
||||
JitterThreshold: 15 * time.Millisecond, // Reduced jitter threshold
|
||||
LatencyOptimizationInterval: 3 * time.Second, // More frequent optimization
|
||||
LatencyAdaptiveThreshold: 0.7, // More aggressive adaptive threshold
|
||||
|
||||
// Microphone Contention Configuration
|
||||
MicContentionTimeout: 200 * time.Millisecond,
|
||||
|
|
@ -531,7 +511,6 @@ func DefaultAudioConfig() *AudioConfigConstants {
|
|||
AveragingWeight: 0.7, // Weight for smoothing values (70% recent, 30% historical)
|
||||
ScalingFactor: 1.5, // General scaling factor for adaptive adjustments
|
||||
|
||||
SmoothingFactor: 0.3, // For adaptive buffer smoothing
|
||||
CPUMemoryWeight: 0.5, // CPU factor weight in combined calculations
|
||||
MemoryWeight: 0.3, // Memory factor weight in combined calculations
|
||||
LatencyWeight: 0.2, // Latency factor weight in combined calculations
|
||||
|
|
@ -548,7 +527,6 @@ func DefaultAudioConfig() *AudioConfigConstants {
|
|||
BatchProcessorFramesPerBatch: 16, // Larger batches for quality changes
|
||||
BatchProcessorTimeout: 20 * time.Millisecond, // Longer timeout for bursts
|
||||
BatchProcessorMaxQueueSize: 64, // Larger queue for quality changes
|
||||
BatchProcessorAdaptiveThreshold: 0.6, // Lower threshold for faster adaptation
|
||||
BatchProcessorThreadPinningThreshold: 8, // Lower threshold for better performance
|
||||
|
||||
// Output Streaming Constants - Balanced for stability
|
||||
|
|
@ -572,10 +550,6 @@ func DefaultAudioConfig() *AudioConfigConstants {
|
|||
// Input Processing Constants - Balanced for stability
|
||||
InputProcessingTimeoutMS: 10, // 10ms processing timeout threshold
|
||||
|
||||
// Adaptive Buffer Constants
|
||||
AdaptiveBufferCPUMultiplier: 100, // 100 multiplier for CPU percentage
|
||||
AdaptiveBufferMemoryMultiplier: 100, // 100 multiplier for memory percentage
|
||||
|
||||
// Socket Names
|
||||
InputSocketName: "audio_input.sock", // Socket name for audio input IPC
|
||||
OutputSocketName: "audio_output.sock", // Socket name for audio output IPC
|
||||
|
|
|
|||
|
|
@ -11,42 +11,6 @@ import (
|
|||
)
|
||||
|
||||
var (
|
||||
// Adaptive buffer metrics
|
||||
adaptiveInputBufferSize = promauto.NewGauge(
|
||||
prometheus.GaugeOpts{
|
||||
Name: "jetkvm_adaptive_input_buffer_size_bytes",
|
||||
Help: "Current adaptive input buffer size in bytes",
|
||||
},
|
||||
)
|
||||
|
||||
adaptiveOutputBufferSize = promauto.NewGauge(
|
||||
prometheus.GaugeOpts{
|
||||
Name: "jetkvm_adaptive_output_buffer_size_bytes",
|
||||
Help: "Current adaptive output buffer size in bytes",
|
||||
},
|
||||
)
|
||||
|
||||
adaptiveBufferAdjustmentsTotal = promauto.NewCounter(
|
||||
prometheus.CounterOpts{
|
||||
Name: "jetkvm_adaptive_buffer_adjustments_total",
|
||||
Help: "Total number of adaptive buffer size adjustments",
|
||||
},
|
||||
)
|
||||
|
||||
adaptiveSystemCpuPercent = promauto.NewGauge(
|
||||
prometheus.GaugeOpts{
|
||||
Name: "jetkvm_adaptive_system_cpu_percent",
|
||||
Help: "System CPU usage percentage used by adaptive buffer manager",
|
||||
},
|
||||
)
|
||||
|
||||
adaptiveSystemMemoryPercent = promauto.NewGauge(
|
||||
prometheus.GaugeOpts{
|
||||
Name: "jetkvm_adaptive_system_memory_percent",
|
||||
Help: "System memory usage percentage used by adaptive buffer manager",
|
||||
},
|
||||
)
|
||||
|
||||
// Socket buffer metrics
|
||||
socketBufferSizeGauge = promauto.NewGaugeVec(
|
||||
prometheus.GaugeOpts{
|
||||
|
|
@ -374,23 +338,6 @@ func UpdateMicrophoneMetrics(metrics UnifiedAudioMetrics) {
|
|||
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()
|
||||
defer metricsUpdateMutex.Unlock()
|
||||
|
||||
adaptiveInputBufferSize.Set(float64(inputBufferSize))
|
||||
adaptiveOutputBufferSize.Set(float64(outputBufferSize))
|
||||
adaptiveSystemCpuPercent.Set(cpuPercent)
|
||||
adaptiveSystemMemoryPercent.Set(memoryPercent)
|
||||
|
||||
if adjustmentMade {
|
||||
adaptiveBufferAdjustmentsTotal.Inc()
|
||||
}
|
||||
|
||||
atomic.StoreInt64(&lastMetricsUpdate, time.Now().Unix())
|
||||
}
|
||||
|
||||
// UpdateSocketBufferMetrics updates socket buffer metrics
|
||||
func UpdateSocketBufferMetrics(component, bufferType string, size, utilization float64, overflowOccurred bool) {
|
||||
metricsUpdateMutex.Lock()
|
||||
|
|
|
|||
|
|
@ -154,25 +154,6 @@ func ValidateMetricsInterval(interval time.Duration) error {
|
|||
return nil
|
||||
}
|
||||
|
||||
// ValidateAdaptiveBufferConfig validates adaptive buffer configuration
|
||||
func ValidateAdaptiveBufferConfig(minSize, maxSize, defaultSize int) error {
|
||||
if minSize <= 0 || maxSize <= 0 || defaultSize <= 0 {
|
||||
return ErrInvalidBufferSize
|
||||
}
|
||||
if minSize >= maxSize {
|
||||
return ErrInvalidBufferSize
|
||||
}
|
||||
if defaultSize < minSize || defaultSize > maxSize {
|
||||
return ErrInvalidBufferSize
|
||||
}
|
||||
// Validate against global limits
|
||||
maxBuffer := Config.SocketMaxBuffer
|
||||
if maxSize > maxBuffer {
|
||||
return ErrInvalidBufferSize
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ValidateInputIPCConfig validates input IPC configuration
|
||||
func ValidateInputIPCConfig(sampleRate, channels, frameSize int) error {
|
||||
minSampleRate := Config.MinSampleRate
|
||||
|
|
|
|||
|
|
@ -211,8 +211,8 @@ func NewAudioInputServer() (*AudioInputServer, error) {
|
|||
return nil, fmt.Errorf("failed to create unix socket after 3 attempts: %w", err)
|
||||
}
|
||||
|
||||
// Get initial buffer size from config
|
||||
initialBufferSize := int64(Config.AdaptiveDefaultBufferSize)
|
||||
// Get initial buffer size (512 frames for stability)
|
||||
initialBufferSize := int64(512)
|
||||
|
||||
// Ensure minimum buffer size to prevent immediate overflow
|
||||
// Use at least 50 frames to handle burst traffic
|
||||
|
|
@ -1182,7 +1182,7 @@ func (ais *AudioInputServer) startMonitorGoroutine() {
|
|||
atomic.StoreInt64(&ais.processingTime, newAvg)
|
||||
}
|
||||
|
||||
// Report latency to adaptive buffer manager
|
||||
// Report latency for metrics
|
||||
ais.ReportLatency(latency)
|
||||
|
||||
if err != nil {
|
||||
|
|
@ -1227,10 +1227,10 @@ func (ais *AudioInputServer) GetServerStats() (total, dropped int64, avgProcessi
|
|||
atomic.LoadInt64(&ais.bufferSize)
|
||||
}
|
||||
|
||||
// UpdateBufferSize updates the buffer size (now using fixed config values)
|
||||
// UpdateBufferSize updates the buffer size (now using fixed values)
|
||||
func (ais *AudioInputServer) UpdateBufferSize() {
|
||||
// Buffer size is now fixed from config
|
||||
newSize := int64(Config.AdaptiveDefaultBufferSize)
|
||||
// Buffer size is now fixed at 512 frames for stability
|
||||
newSize := int64(512)
|
||||
atomic.StoreInt64(&ais.bufferSize, newSize)
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -3,7 +3,7 @@
|
|||
|
||||
// Package audio provides real-time audio processing for JetKVM with low-latency streaming.
|
||||
//
|
||||
// Key components: output/input pipelines with Opus codec, adaptive buffer management,
|
||||
// Key components: output/input pipelines with Opus codec, buffer management,
|
||||
// zero-copy frame pools, IPC communication, and process supervision.
|
||||
//
|
||||
// Supports four quality presets (Low/Medium/High/Ultra) with configurable bitrates.
|
||||
|
|
|
|||
Loading…
Reference in New Issue