kvm/internal/audio/cgo_audio.go

//go:build cgo

package audio

import (
	"errors"
	"fmt"
	"sync"
	"sync/atomic"
	"time"
	"unsafe"
)

/*
#cgo CFLAGS: -I$HOME/.jetkvm/audio-libs/alsa-lib-$ALSA_VERSION/include -I$HOME/.jetkvm/audio-libs/opus-$OPUS_VERSION/include -I$HOME/.jetkvm/audio-libs/opus-$OPUS_VERSION/celt
#cgo LDFLAGS: -L$HOME/.jetkvm/audio-libs/alsa-lib-$ALSA_VERSION/src/.libs -lasound -L$HOME/.jetkvm/audio-libs/opus-$OPUS_VERSION/.libs -lopus -lm -ldl -static

#include "c/audio.c"
*/
import "C"

// Optimized Go wrappers with reduced overhead
var (
	// Base error types for wrapping with context
	errAudioInitFailed   = errors.New("failed to init ALSA/Opus")
	errAudioReadEncode   = errors.New("audio read/encode error")
	errAudioDecodeWrite  = errors.New("audio decode/write error")
	errAudioPlaybackInit = errors.New("failed to init ALSA playback/Opus decoder")
	errEmptyBuffer       = errors.New("empty buffer")
	errNilBuffer         = errors.New("nil buffer")
	errInvalidBufferPtr  = errors.New("invalid buffer pointer")
)

// Error creation functions with enhanced context
func newBufferTooSmallError(actual, required int) error {
	baseErr := fmt.Errorf("buffer too small: got %d bytes, need at least %d bytes", actual, required)
	return WrapWithMetadata(baseErr, "cgo_audio", "buffer_validation", map[string]interface{}{
		"actual_size":   actual,
		"required_size": required,
		"error_type":    "buffer_undersize",
	})
}

func newBufferTooLargeError(actual, max int) error {
	baseErr := fmt.Errorf("buffer too large: got %d bytes, maximum allowed %d bytes", actual, max)
	return WrapWithMetadata(baseErr, "cgo_audio", "buffer_validation", map[string]interface{}{
		"actual_size": actual,
		"max_size":    max,
		"error_type":  "buffer_oversize",
	})
}

func newAudioInitError(cErrorCode int) error {
	baseErr := fmt.Errorf("%w: C error code %d", errAudioInitFailed, cErrorCode)
	return WrapWithMetadata(baseErr, "cgo_audio", "initialization", map[string]interface{}{
		"c_error_code": cErrorCode,
		"error_type":   "init_failure",
		"severity":     "critical",
	})
}

func newAudioPlaybackInitError(cErrorCode int) error {
	baseErr := fmt.Errorf("%w: C error code %d", errAudioPlaybackInit, cErrorCode)
	return WrapWithMetadata(baseErr, "cgo_audio", "playback_init", map[string]interface{}{
		"c_error_code": cErrorCode,
		"error_type":   "playback_init_failure",
		"severity":     "high",
	})
}

func newAudioReadEncodeError(cErrorCode int) error {
	baseErr := fmt.Errorf("%w: C error code %d", errAudioReadEncode, cErrorCode)
	return WrapWithMetadata(baseErr, "cgo_audio", "read_encode", map[string]interface{}{
		"c_error_code": cErrorCode,
		"error_type":   "read_encode_failure",
		"severity":     "medium",
	})
}

func newAudioDecodeWriteError(cErrorCode int) error {
	baseErr := fmt.Errorf("%w: C error code %d", errAudioDecodeWrite, cErrorCode)
	return WrapWithMetadata(baseErr, "cgo_audio", "decode_write", map[string]interface{}{
		"c_error_code": cErrorCode,
		"error_type":   "decode_write_failure",
		"severity":     "medium",
	})
}

func cgoAudioInit() error {
	// Get cached config and ensure it's updated
	cache := GetCachedConfig()
	cache.Update()

	// Update C constants from cached config (atomic access, no locks)
	C.update_audio_constants(
		C.int(cache.opusBitrate.Load()),
		C.int(cache.opusComplexity.Load()),
		C.int(cache.opusVBR.Load()),
		C.int(cache.opusVBRConstraint.Load()),
		C.int(cache.opusSignalType.Load()),
		C.int(cache.opusBandwidth.Load()),
		C.int(cache.opusDTX.Load()),
		C.int(16), // LSB depth for improved bit allocation
		C.int(cache.sampleRate.Load()),
		C.int(cache.channels.Load()),
		C.int(cache.frameSize.Load()),
		C.int(cache.maxPacketSize.Load()),
		C.int(Config.CGOUsleepMicroseconds),
		C.int(Config.CGOMaxAttempts),
		C.int(Config.CGOMaxBackoffMicroseconds),
	)

	result := C.jetkvm_audio_init()
	if result != 0 {
		return newAudioInitError(int(result))
	}
	return nil
}

func cgoAudioClose() {
	C.jetkvm_audio_close()
}

// AudioConfigCache provides a comprehensive caching system for audio configuration
type AudioConfigCache struct {
	// Atomic int64 fields MUST be first for ARM32 alignment (8-byte alignment required)
	minFrameDuration         atomic.Int64 // Store as nanoseconds
	maxFrameDuration         atomic.Int64 // Store as nanoseconds
	maxLatency               atomic.Int64 // Store as nanoseconds
	minMetricsUpdateInterval atomic.Int64 // Store as nanoseconds
	maxMetricsUpdateInterval atomic.Int64 // Store as nanoseconds
	restartWindow            atomic.Int64 // Store as nanoseconds
	restartDelay             atomic.Int64 // Store as nanoseconds
	maxRestartDelay          atomic.Int64 // Store as nanoseconds

	// Atomic int32 fields for lock-free access to frequently used values
	minReadEncodeBuffer  atomic.Int32
	maxDecodeWriteBuffer atomic.Int32
	maxPacketSize        atomic.Int32
	maxPCMBufferSize     atomic.Int32
	opusBitrate          atomic.Int32
	opusComplexity       atomic.Int32
	opusVBR              atomic.Int32
	opusVBRConstraint    atomic.Int32
	opusSignalType       atomic.Int32
	opusBandwidth        atomic.Int32
	opusDTX              atomic.Int32
	sampleRate           atomic.Int32
	channels             atomic.Int32
	frameSize            atomic.Int32

	// Additional cached values for validation functions
	maxAudioFrameSize atomic.Int32
	maxChannels       atomic.Int32
	minOpusBitrate    atomic.Int32
	maxOpusBitrate    atomic.Int32

	// Socket and buffer configuration values
	socketMaxBuffer          atomic.Int32
	socketMinBuffer          atomic.Int32
	inputProcessingTimeoutMS atomic.Int32
	maxRestartAttempts       atomic.Int32

	// Batch processing related values
	BatchProcessingTimeout               time.Duration
	BatchProcessorFramesPerBatch         int
	BatchProcessorTimeout                time.Duration
	BatchProcessingDelay                 time.Duration
	MinBatchSizeForThreadPinning         int
	BatchProcessorMaxQueueSize           int
	BatchProcessorAdaptiveThreshold      float64
	BatchProcessorThreadPinningThreshold int

	// Mutex for updating the cache
	mutex       sync.RWMutex
	lastUpdate  time.Time
	cacheExpiry time.Duration
	initialized atomic.Bool

	// Pre-allocated errors to avoid allocations in hot path
	bufferTooSmallReadEncode  error
	bufferTooLargeDecodeWrite error
}

// Global audio config cache instance
var globalAudioConfigCache = &AudioConfigCache{
	cacheExpiry: 30 * time.Second, // Increased from 10s to 30s to further reduce cache updates
}

// GetCachedConfig returns the global audio config cache instance
func GetCachedConfig() *AudioConfigCache {
	return globalAudioConfigCache
}

// Update refreshes the cached config values if needed
func (c *AudioConfigCache) Update() {
	// Fast path: if cache is initialized and not expired, return immediately
	if c.initialized.Load() {
		c.mutex.RLock()
		cacheExpired := time.Since(c.lastUpdate) > c.cacheExpiry
		c.mutex.RUnlock()
		if !cacheExpired {
			return
		}
	}

	// Slow path: update cache
	c.mutex.Lock()
	defer c.mutex.Unlock()

	// Double-check after acquiring lock
	if !c.initialized.Load() || time.Since(c.lastUpdate) > c.cacheExpiry {
		// Update atomic values for lock-free access - CGO values
		c.minReadEncodeBuffer.Store(int32(Config.MinReadEncodeBuffer))
		c.maxDecodeWriteBuffer.Store(int32(Config.MaxDecodeWriteBuffer))
		c.maxPacketSize.Store(int32(Config.CGOMaxPacketSize))
		c.maxPCMBufferSize.Store(int32(Config.MaxPCMBufferSize))
		c.opusBitrate.Store(int32(Config.CGOOpusBitrate))
		c.opusComplexity.Store(int32(Config.CGOOpusComplexity))
		c.opusVBR.Store(int32(Config.CGOOpusVBR))
		c.opusVBRConstraint.Store(int32(Config.CGOOpusVBRConstraint))
		c.opusSignalType.Store(int32(Config.CGOOpusSignalType))
		c.opusBandwidth.Store(int32(Config.CGOOpusBandwidth))
		c.opusDTX.Store(int32(Config.CGOOpusDTX))
		c.sampleRate.Store(int32(Config.CGOSampleRate))
		c.channels.Store(int32(Config.CGOChannels))
		c.frameSize.Store(int32(Config.CGOFrameSize))

		// Update additional validation values
		c.maxAudioFrameSize.Store(int32(Config.MaxAudioFrameSize))
		c.maxChannels.Store(int32(Config.MaxChannels))
		c.minFrameDuration.Store(int64(Config.MinFrameDuration))
		c.maxFrameDuration.Store(int64(Config.MaxFrameDuration))
		c.minOpusBitrate.Store(int32(Config.MinOpusBitrate))
		c.maxOpusBitrate.Store(int32(Config.MaxOpusBitrate))

		// Update batch processing related values
		c.BatchProcessingTimeout = 100 * time.Millisecond // Fixed timeout for batch processing
		c.BatchProcessorFramesPerBatch = Config.BatchProcessorFramesPerBatch
		c.BatchProcessorTimeout = Config.BatchProcessorTimeout
		c.BatchProcessingDelay = Config.BatchProcessingDelay
		c.MinBatchSizeForThreadPinning = Config.MinBatchSizeForThreadPinning
		c.BatchProcessorMaxQueueSize = Config.BatchProcessorMaxQueueSize
		c.BatchProcessorAdaptiveThreshold = Config.BatchProcessorAdaptiveThreshold
		c.BatchProcessorThreadPinningThreshold = Config.BatchProcessorThreadPinningThreshold

		// Pre-allocate common errors
		c.bufferTooSmallReadEncode = newBufferTooSmallError(0, Config.MinReadEncodeBuffer)
		c.bufferTooLargeDecodeWrite = newBufferTooLargeError(Config.MaxDecodeWriteBuffer+1, Config.MaxDecodeWriteBuffer)

		c.lastUpdate = time.Now()
		c.initialized.Store(true)

		// Update the global validation cache as well
		if cachedMaxFrameSize != 0 {
			cachedMaxFrameSize = Config.MaxAudioFrameSize
		}
	}
}

// GetMinReadEncodeBuffer returns the cached MinReadEncodeBuffer value
func (c *AudioConfigCache) GetMinReadEncodeBuffer() int {
	return int(c.minReadEncodeBuffer.Load())
}

// GetMaxDecodeWriteBuffer returns the cached MaxDecodeWriteBuffer value
func (c *AudioConfigCache) GetMaxDecodeWriteBuffer() int {
	return int(c.maxDecodeWriteBuffer.Load())
}

// GetMaxPacketSize returns the cached MaxPacketSize value
func (c *AudioConfigCache) GetMaxPacketSize() int {
	return int(c.maxPacketSize.Load())
}

// GetMaxPCMBufferSize returns the cached MaxPCMBufferSize value
func (c *AudioConfigCache) GetMaxPCMBufferSize() int {
	return int(c.maxPCMBufferSize.Load())
}

// GetBufferTooSmallError returns the pre-allocated buffer too small error
func (c *AudioConfigCache) GetBufferTooSmallError() error {
	return c.bufferTooSmallReadEncode
}

// GetBufferTooLargeError returns the pre-allocated buffer too large error
func (c *AudioConfigCache) GetBufferTooLargeError() error {
	return c.bufferTooLargeDecodeWrite
}

// Removed duplicate config caching system - using AudioConfigCache instead

// updateCacheIfNeeded updates cache only if expired to avoid overhead
func updateCacheIfNeeded(cache *AudioConfigCache) {
	if cache.initialized.Load() {
		cache.mutex.RLock()
		cacheExpired := time.Since(cache.lastUpdate) > cache.cacheExpiry
		cache.mutex.RUnlock()
		if cacheExpired {
			cache.Update()
		}
	} else {
		cache.Update()
	}
}

func cgoAudioReadEncode(buf []byte) (int, error) {
	// Minimal buffer validation - assume caller provides correct size
	if len(buf) == 0 {
		return 0, errEmptyBuffer
	}

	// Direct CGO call - hotpath optimization
	n := C.jetkvm_audio_read_encode(unsafe.Pointer(&buf[0]))

	// Fast path for success
	if n > 0 {
		return int(n), nil
	}

	// Error handling with static errors
	if n < 0 {
		if n == -1 {
			return 0, errAudioInitFailed
		}
		return 0, errAudioReadEncode
	}

	return 0, nil
}

// Audio playback functions
func cgoAudioPlaybackInit() error {
	// Get cached config and ensure it's updated
	cache := GetCachedConfig()
	cache.Update()

	// No need to update C constants here as they're already set in cgoAudioInit

	ret := C.jetkvm_audio_playback_init()
	if ret != 0 {
		return newAudioPlaybackInitError(int(ret))
	}
	return nil
}

func cgoAudioPlaybackClose() {
	C.jetkvm_audio_playback_close()
}

func cgoAudioDecodeWrite(buf []byte) (int, error) {
	// Minimal validation - assume caller provides correct size
	if len(buf) == 0 {
		return 0, errEmptyBuffer
	}

	// Direct CGO call - hotpath optimization
	n := int(C.jetkvm_audio_decode_write(unsafe.Pointer(&buf[0]), C.int(len(buf))))

	// Fast path for success
	if n >= 0 {
		return n, nil
	}

	// Error handling with static errors
	if n == -1 {
		return 0, errAudioInitFailed
	}
	return 0, errAudioDecodeWrite
}

// updateOpusEncoderParams dynamically updates OPUS encoder parameters
func updateOpusEncoderParams(bitrate, complexity, vbr, vbrConstraint, signalType, bandwidth, dtx int) error {
	result := C.update_opus_encoder_params(
		C.int(bitrate),
		C.int(complexity),
		C.int(vbr),
		C.int(vbrConstraint),
		C.int(signalType),
		C.int(bandwidth),
		C.int(dtx),
	)
	if result != 0 {
		return fmt.Errorf("failed to update OPUS encoder parameters: C error code %d", result)
	}
	return nil
}

// Buffer pool for reusing buffers in CGO functions
var (
	// Using SizedBufferPool for better memory management
	// Track buffer pool usage
	cgoBufferPoolGets atomic.Int64
	cgoBufferPoolPuts atomic.Int64
	// Batch processing statistics - only enabled in debug builds
	batchProcessingCount atomic.Int64
	batchFrameCount      atomic.Int64
	batchProcessingTime  atomic.Int64
	// Batch time tracking removed
)

// GetBufferFromPool gets a buffer from the pool with at least the specified capacity
func GetBufferFromPool(minCapacity int) []byte {
	cgoBufferPoolGets.Add(1)
	return GetOptimalBuffer(minCapacity)
}

// ReturnBufferToPool returns a buffer to the pool
func ReturnBufferToPool(buf []byte) {
	cgoBufferPoolPuts.Add(1)
	ReturnOptimalBuffer(buf)
}

// ReadEncodeWithPooledBuffer reads audio data and encodes it using a buffer from the pool
func ReadEncodeWithPooledBuffer() ([]byte, int, error) {
	cache := GetCachedConfig()
	updateCacheIfNeeded(cache)

	bufferSize := cache.GetMinReadEncodeBuffer()
	if bufferSize == 0 {
		bufferSize = 1500
	}

	buf := GetBufferFromPool(bufferSize)
	n, err := cgoAudioReadEncode(buf)
	if err != nil {
		ReturnBufferToPool(buf)
		return nil, 0, err
	}

	return buf[:n], n, nil
}

// DecodeWriteWithPooledBuffer decodes and writes audio data using a pooled buffer
func DecodeWriteWithPooledBuffer(data []byte) (int, error) {
	if len(data) == 0 {
		return 0, errEmptyBuffer
	}

	cache := GetCachedConfig()
	updateCacheIfNeeded(cache)

	maxPacketSize := cache.GetMaxPacketSize()
	if len(data) > maxPacketSize {
		return 0, newBufferTooLargeError(len(data), maxPacketSize)
	}

	pcmBuffer := GetBufferFromPool(cache.GetMaxPCMBufferSize())
	defer ReturnBufferToPool(pcmBuffer)

	return CGOAudioDecodeWrite(data, pcmBuffer)
}

// BatchReadEncode reads and encodes multiple audio frames in a single batch
// with optimized zero-copy frame management and batch reference counting
func BatchReadEncode(batchSize int) ([][]byte, error) {
	// Simple batch processing without complex overhead
	frames := make([][]byte, 0, batchSize)
	frameSize := 4096 // Fixed frame size for performance

	for i := 0; i < batchSize; i++ {
		buf := make([]byte, frameSize)
		n, err := cgoAudioReadEncode(buf)
		if err != nil {
			if i > 0 {
				return frames, nil // Return partial batch
			}
			return nil, err
		}
		if n > 0 {
			frames = append(frames, buf[:n])
		}
	}

	return frames, nil
}

// BatchDecodeWrite decodes and writes multiple audio frames in a single batch
// This reduces CGO call overhead by processing multiple frames at once
// with optimized zero-copy frame management and batch reference counting
func BatchDecodeWrite(frames [][]byte) error {
	// Validate input
	if len(frames) == 0 {
		return nil
	}

	// Convert to zero-copy frames for optimized processing
	zeroCopyFrames := make([]*ZeroCopyAudioFrame, 0, len(frames))
	for _, frameData := range frames {
		if len(frameData) > 0 {
			frame := GetZeroCopyFrame()
			frame.SetDataDirect(frameData) // Direct assignment without copy
			zeroCopyFrames = append(zeroCopyFrames, frame)
		}
	}

	// Use batch reference counting for efficient management
	if len(zeroCopyFrames) > 0 {
		// Batch AddRef all frames at once
		err := BatchAddRefFrames(zeroCopyFrames)
		if err != nil {
			return err
		}
		// Ensure cleanup with batch release
		defer func() {
			if _, err := BatchReleaseFrames(zeroCopyFrames); err != nil {
				// Log release error but don't fail the operation
				_ = err
			}
		}()
	}

	// Get cached config
	cache := GetCachedConfig()
	// Only update cache if expired - avoid unnecessary overhead
	// Use proper locking to avoid race condition
	if cache.initialized.Load() {
		cache.mutex.RLock()
		cacheExpired := time.Since(cache.lastUpdate) > cache.cacheExpiry
		cache.mutex.RUnlock()
		if cacheExpired {
			cache.Update()
		}
	} else {
		cache.Update()
	}

	// Track batch processing statistics - only if enabled
	var startTime time.Time
	// Batch time tracking removed
	trackTime := false
	if trackTime {
		startTime = time.Now()
	}
	batchProcessingCount.Add(1)

	// Get a PCM buffer from the pool for optimized decode-write
	pcmBuffer := GetBufferFromPool(cache.GetMaxPCMBufferSize())
	defer ReturnBufferToPool(pcmBuffer)

	// Process each zero-copy frame with optimized batch processing
	frameCount := 0
	for _, zcFrame := range zeroCopyFrames {
		// Get frame data from zero-copy frame
		frameData := zcFrame.Data()[:zcFrame.Length()]
		if len(frameData) == 0 {
			continue
		}

		// Process this frame using optimized implementation
		_, err := CGOAudioDecodeWrite(frameData, pcmBuffer)
		if err != nil {
			// Update statistics before returning error
			batchFrameCount.Add(int64(frameCount))
			if trackTime {
				batchProcessingTime.Add(time.Since(startTime).Microseconds())
			}
			return err
		}

		frameCount++
	}

	// Update statistics
	batchFrameCount.Add(int64(frameCount))
	if trackTime {
		batchProcessingTime.Add(time.Since(startTime).Microseconds())
	}

	return nil
}

// GetBatchProcessingStats returns statistics about batch processing
func GetBatchProcessingStats() (count, frames, avgTimeUs int64) {
	count = batchProcessingCount.Load()
	frames = batchFrameCount.Load()
	totalTime := batchProcessingTime.Load()

	// Calculate average time per batch
	if count > 0 {
		avgTimeUs = totalTime / count
	}

	return count, frames, avgTimeUs
}

// cgoAudioDecodeWriteWithBuffers decodes opus data and writes to PCM buffer
// This implementation uses separate buffers for opus data and PCM output
func cgoAudioDecodeWriteWithBuffers(opusData []byte, pcmBuffer []byte) (int, error) {
	// Validate input
	if len(opusData) == 0 {
		return 0, errEmptyBuffer
	}
	if len(pcmBuffer) == 0 {
		return 0, errEmptyBuffer
	}

	// Get cached config
	cache := GetCachedConfig()
	// Only update cache if expired - avoid unnecessary overhead
	// Use proper locking to avoid race condition
	if cache.initialized.Load() {
		cache.mutex.RLock()
		cacheExpired := time.Since(cache.lastUpdate) > cache.cacheExpiry
		cache.mutex.RUnlock()
		if cacheExpired {
			cache.Update()
		}
	} else {
		cache.Update()
	}

	// Ensure data doesn't exceed max packet size
	maxPacketSize := cache.GetMaxPacketSize()
	if len(opusData) > maxPacketSize {
		return 0, newBufferTooLargeError(len(opusData), maxPacketSize)
	}

	// Direct CGO call with minimal overhead - unsafe.Pointer(&slice[0]) is never nil for non-empty slices
	n := int(C.jetkvm_audio_decode_write(unsafe.Pointer(&opusData[0]), C.int(len(opusData))))

	// Fast path for success case
	if n >= 0 {
		return n, nil
	}

	// Handle error cases with static error codes to reduce allocations
	switch n {
	case -1:
		return 0, errAudioInitFailed
	case -2:
		return 0, errAudioDecodeWrite
	default:
		return 0, newAudioDecodeWriteError(n)
	}
}

// Optimized CGO function aliases - use direct function calls to reduce overhead
// These are now direct function aliases instead of variable assignments
func CGOAudioInit() error                               { return cgoAudioInit() }
func CGOAudioClose()                                    { cgoAudioClose() }
func CGOAudioReadEncode(buf []byte) (int, error)        { return cgoAudioReadEncode(buf) }
func CGOAudioPlaybackInit() error                       { return cgoAudioPlaybackInit() }
func CGOAudioPlaybackClose()                            { cgoAudioPlaybackClose() }
func CGOAudioDecodeWriteLegacy(buf []byte) (int, error) { return cgoAudioDecodeWrite(buf) }
func CGOAudioDecodeWrite(opusData []byte, pcmBuffer []byte) (int, error) {
	return cgoAudioDecodeWriteWithBuffers(opusData, pcmBuffer)
}
func CGOUpdateOpusEncoderParams(bitrate, complexity, vbr, vbrConstraint, signalType, bandwidth, dtx int) error {
	return updateOpusEncoderParams(bitrate, complexity, vbr, vbrConstraint, signalType, bandwidth, dtx)
}