Improvements, Fixes: reduce mouse lag when audio is on

Author: pennycoders

.golangci.yml

@@ -1,4 +1,7 @@
 version: "2"
+run:
+  build-tags:
+    - nolint
 linters:
   enable:
     - forbidigo

cloud.go

@@ -454,7 +454,7 @@ func handleSessionRequest(
 	// Check if we have an existing session and handle renegotiation
 	if currentSession != nil {
 		scopedLogger.Info().Msg("handling renegotiation for existing session")
-		
+
 		// Handle renegotiation with existing session
 		sd, err = currentSession.ExchangeOffer(req.Sd)
 		if err != nil {

internal/audio/cgo_audio.go

@@ -1,3 +1,5 @@
+//go:build !nolint
+
 package audio
 
 import (
@@ -54,7 +56,7 @@ int jetkvm_audio_read_encode(void *opus_buf) {
 	short pcm_buffer[1920]; // max 2ch*960
 	unsigned char *out = (unsigned char*)opus_buf;
 	int pcm_rc = snd_pcm_readi(pcm_handle, pcm_buffer, frame_size);
-	
+
 	// Handle ALSA errors with recovery
 	if (pcm_rc < 0) {
 		if (pcm_rc == -EPIPE) {
@@ -70,12 +72,12 @@ int jetkvm_audio_read_encode(void *opus_buf) {
 			return -1;
 		}
 	}
-	
+
 	// If we got fewer frames than expected, pad with silence
 	if (pcm_rc < frame_size) {
 		memset(&pcm_buffer[pcm_rc * channels], 0, (frame_size - pcm_rc) * channels * sizeof(short));
 	}
-	
+
 	int nb_bytes = opus_encode(encoder, pcm_buffer, frame_size, out, max_packet_size);
 	return nb_bytes;
 }
@@ -85,15 +87,15 @@ int jetkvm_audio_playback_init() {
 	int err;
 	snd_pcm_hw_params_t *params;
 	if (pcm_playback_handle) return 0;
-	
+
 	// Try to open the USB gadget audio device for playback
 	// This should correspond to the capture endpoint of the USB gadget
 	if (snd_pcm_open(&pcm_playback_handle, "hw:1,0", SND_PCM_STREAM_PLAYBACK, 0) < 0) {
 		// Fallback to default device if hw:1,0 doesn't work for playback
 		if (snd_pcm_open(&pcm_playback_handle, "default", SND_PCM_STREAM_PLAYBACK, 0) < 0)
 			return -1;
 	}
-	
+
 	snd_pcm_hw_params_malloc(&params);
 	snd_pcm_hw_params_any(pcm_playback_handle, params);
 	snd_pcm_hw_params_set_access(pcm_playback_handle, params, SND_PCM_ACCESS_RW_INTERLEAVED);
@@ -104,23 +106,23 @@ int jetkvm_audio_playback_init() {
 	snd_pcm_hw_params(pcm_playback_handle, params);
 	snd_pcm_hw_params_free(params);
 	snd_pcm_prepare(pcm_playback_handle);
-	
+
 	// Initialize Opus decoder
 	decoder = opus_decoder_create(sample_rate, channels, &err);
 	if (!decoder) return -2;
-	
+
 	return 0;
 }
 
 // Decode Opus and write PCM to playback device
 int jetkvm_audio_decode_write(void *opus_buf, int opus_size) {
 	short pcm_buffer[1920]; // max 2ch*960
 	unsigned char *in = (unsigned char*)opus_buf;
-	
+
 	// Decode Opus to PCM
 	int pcm_frames = opus_decode(decoder, in, opus_size, pcm_buffer, frame_size, 0);
 	if (pcm_frames < 0) return -1;
-	
+
 	// Write PCM to playback device
 	int pcm_rc = snd_pcm_writei(pcm_playback_handle, pcm_buffer, pcm_frames);
 	if (pcm_rc < 0) {
@@ -131,7 +133,7 @@ int jetkvm_audio_decode_write(void *opus_buf, int opus_size) {
 		}
 		if (pcm_rc < 0) return -2;
 	}
-	
+
 	return pcm_frames;
 }
 
@@ -148,8 +150,6 @@ void jetkvm_audio_close() {
 */
 import "C"
 
-
-
 // Go wrappers for initializing, starting, stopping, and controlling audio
 func cgoAudioInit() error {
 	ret := C.jetkvm_audio_init()
@@ -179,8 +179,6 @@ func cgoAudioReadEncode(buf []byte) (int, error) {
 	return int(n), nil
 }
 
-
-
 // Go wrappers for audio playback (microphone input)
 func cgoAudioPlaybackInit() error {
 	ret := C.jetkvm_audio_playback_init()
@@ -206,8 +204,6 @@ func cgoAudioDecodeWrite(buf []byte) (int, error) {
 	return int(n), nil
 }
 
-
-
 // Wrapper functions for non-blocking audio manager
 func CGOAudioInit() error {
 	return cgoAudioInit()

internal/audio/cgo_audio_stub.go

@@ -28,4 +28,30 @@ func cgoAudioPlaybackClose() {
 
 func cgoAudioDecodeWrite(buf []byte) (int, error) {
 	return 0, errors.New("audio not available in lint mode")
-}
+}
+
+// Uppercase wrapper functions (called by nonblocking_audio.go)
+
+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 CGOAudioDecodeWrite(buf []byte) (int, error) {
+	return cgoAudioDecodeWrite(buf)
+}

internal/audio/input.go

@@ -11,7 +11,7 @@ import (
 // AudioInputMetrics holds metrics for microphone input
 // Note: int64 fields must be 64-bit aligned for atomic operations on ARM
 type AudioInputMetrics struct {
-	FramesSent      int64         // Must be first for alignment
+	FramesSent      int64 // Must be first for alignment
 	FramesDropped   int64
 	BytesProcessed  int64
 	ConnectionDrops int64
@@ -22,8 +22,8 @@ type AudioInputMetrics struct {
 // AudioInputManager manages microphone input stream from WebRTC to USB gadget
 type AudioInputManager struct {
 	// metrics MUST be first for ARM32 alignment (contains int64 fields)
-	metrics     AudioInputMetrics
-	
+	metrics AudioInputMetrics
+
 	inputBuffer chan []byte
 	logger      zerolog.Logger
 	running     int32
@@ -44,7 +44,7 @@ func (aim *AudioInputManager) Start() error {
 	}
 
 	aim.logger.Info().Msg("Starting audio input manager")
-	
+
 	// Start the non-blocking audio input stream
 	err := StartNonBlockingAudioInput(aim.inputBuffer)
 	if err != nil {
@@ -62,11 +62,11 @@ func (aim *AudioInputManager) Stop() {
 	}
 
 	aim.logger.Info().Msg("Stopping audio input manager")
-	
+
 	// Stop the non-blocking audio input stream
 	// Note: This is handled by the global non-blocking audio manager
 	// Individual input streams are managed centrally
-	
+
 	// Drain the input buffer
 	go func() {
 		for {
@@ -115,4 +115,4 @@ func (aim *AudioInputManager) GetMetrics() AudioInputMetrics {
 // IsRunning returns whether the audio input manager is running
 func (aim *AudioInputManager) IsRunning() bool {
 	return atomic.LoadInt32(&aim.running) == 1
-}
+}

internal/audio/nonblocking_api.go

@@ -14,11 +14,14 @@ func StartNonBlockingAudioStreaming(send func([]byte)) error {
 	managerMutex.Lock()
 	defer managerMutex.Unlock()
 
-	if globalNonBlockingManager != nil && globalNonBlockingManager.IsRunning() {
-		return ErrAudioAlreadyRunning
+	if globalNonBlockingManager != nil && globalNonBlockingManager.IsOutputRunning() {
+		return nil // Already running, this is not an error
+	}
+
+	if globalNonBlockingManager == nil {
+		globalNonBlockingManager = NewNonBlockingAudioManager()
 	}
 
-	globalNonBlockingManager = NewNonBlockingAudioManager()
 	return globalNonBlockingManager.StartAudioOutput(send)
 }
 
@@ -31,6 +34,11 @@ func StartNonBlockingAudioInput(receiveChan <-chan []byte) error {
 		globalNonBlockingManager = NewNonBlockingAudioManager()
 	}
 
+	// Check if input is already running to avoid unnecessary operations
+	if globalNonBlockingManager.IsInputRunning() {
+		return nil // Already running, this is not an error
+	}
+
 	return globalNonBlockingManager.StartAudioInput(receiveChan)
 }
 
@@ -45,6 +53,16 @@ func StopNonBlockingAudioStreaming() {
 	}
 }
 
+// StopNonBlockingAudioInput stops only the audio input without affecting output
+func StopNonBlockingAudioInput() {
+	managerMutex.Lock()
+	defer managerMutex.Unlock()
+
+	if globalNonBlockingManager != nil && globalNonBlockingManager.IsInputRunning() {
+		globalNonBlockingManager.StopAudioInput()
+	}
+}
+
 // GetNonBlockingAudioStats returns statistics from the non-blocking audio system
 func GetNonBlockingAudioStats() NonBlockingAudioStats {
 	managerMutex.Lock()
@@ -62,4 +80,12 @@ func IsNonBlockingAudioRunning() bool {
 	defer managerMutex.Unlock()
 
 	return globalNonBlockingManager != nil && globalNonBlockingManager.IsRunning()
-}
+}
+
+// IsNonBlockingAudioInputRunning returns true if the non-blocking audio input is running
+func IsNonBlockingAudioInputRunning() bool {
+	managerMutex.Lock()
+	defer managerMutex.Unlock()
+
+	return globalNonBlockingManager != nil && globalNonBlockingManager.IsInputRunning()
+}

internal/audio/nonblocking_audio.go

@@ -23,14 +23,14 @@ type NonBlockingAudioManager struct {
 	logger *zerolog.Logger
 
 	// Audio output (capture from device, send to WebRTC)
-	outputSendFunc  func([]byte)
-	outputWorkChan  chan audioWorkItem
+	outputSendFunc   func([]byte)
+	outputWorkChan   chan audioWorkItem
 	outputResultChan chan audioResult
 
-	// Audio input (receive from WebRTC, playback to device)  
+	// Audio input (receive from WebRTC, playback to device)
 	inputReceiveChan <-chan []byte
-	inputWorkChan   chan audioWorkItem
-	inputResultChan chan audioResult
+	inputWorkChan    chan audioWorkItem
+	inputResultChan  chan audioResult
 
 	// Worker threads and flags - int32 fields grouped together
 	outputRunning       int32
@@ -69,7 +69,7 @@ type NonBlockingAudioStats struct {
 	InputFramesDropped    int64
 	WorkerErrors          int64
 	// time.Time is int64 internally, so it's also aligned
-	LastProcessTime       time.Time
+	LastProcessTime time.Time
 }
 
 // NewNonBlockingAudioManager creates a new non-blocking audio manager
@@ -81,8 +81,8 @@ func NewNonBlockingAudioManager() *NonBlockingAudioManager {
 		ctx:              ctx,
 		cancel:           cancel,
 		logger:           &logger,
-		outputWorkChan:   make(chan audioWorkItem, 10),   // Buffer for work items
-		outputResultChan: make(chan audioResult, 10),     // Buffer for results
+		outputWorkChan:   make(chan audioWorkItem, 10), // Buffer for work items
+		outputResultChan: make(chan audioResult, 10),   // Buffer for results
 		inputWorkChan:    make(chan audioWorkItem, 10),
 		inputResultChan:  make(chan audioResult, 10),
 	}
@@ -327,7 +327,7 @@ func (nam *NonBlockingAudioManager) inputCoordinatorThread() {
 			return
 
 		case frame := <-nam.inputReceiveChan:
-			if frame == nil || len(frame) == 0 {
+			if len(frame) == 0 {
 				continue
 			}
 
@@ -397,6 +397,16 @@ func (nam *NonBlockingAudioManager) Stop() {
 	nam.logger.Info().Msg("non-blocking audio manager stopped")
 }
 
+// StopAudioInput stops only the audio input operations
+func (nam *NonBlockingAudioManager) StopAudioInput() {
+	nam.logger.Info().Msg("stopping audio input")
+
+	// Stop only the input coordinator
+	atomic.StoreInt32(&nam.inputRunning, 0)
+
+	nam.logger.Info().Msg("audio input stopped")
+}
+
 // GetStats returns current statistics
 func (nam *NonBlockingAudioManager) GetStats() NonBlockingAudioStats {
 	return NonBlockingAudioStats{
@@ -412,4 +422,14 @@ func (nam *NonBlockingAudioManager) GetStats() NonBlockingAudioStats {
 // IsRunning returns true if any audio operations are running
 func (nam *NonBlockingAudioManager) IsRunning() bool {
 	return atomic.LoadInt32(&nam.outputRunning) == 1 || atomic.LoadInt32(&nam.inputRunning) == 1
-}
+}
+
+// IsInputRunning returns true if audio input is running
+func (nam *NonBlockingAudioManager) IsInputRunning() bool {
+	return atomic.LoadInt32(&nam.inputRunning) == 1
+}
+
+// IsOutputRunning returns true if audio output is running
+func (nam *NonBlockingAudioManager) IsOutputRunning() bool {
+	return atomic.LoadInt32(&nam.outputRunning) == 1
+}

jsonrpc.go

@@ -21,8 +21,8 @@ import (
 
 // Mouse event processing with single worker
 var (
-	mouseEventChan   = make(chan mouseEventData, 100) // Buffered channel for mouse events
-	mouseWorkerOnce  sync.Once
+	mouseEventChan  = make(chan mouseEventData, 100) // Buffered channel for mouse events
+	mouseWorkerOnce sync.Once
 )
 
 type mouseEventData struct {
@@ -35,15 +35,15 @@ func startMouseWorker() {
 	go func() {
 		ticker := time.NewTicker(16 * time.Millisecond) // ~60 FPS
 		defer ticker.Stop()
-		
+
 		var latestMouseEvent *mouseEventData
-		
+
 		for {
 			select {
 			case event := <-mouseEventChan:
 				// Always keep the latest mouse event
 				latestMouseEvent = &event
-				
+
 			case <-ticker.C:
 				// Process the latest mouse event at regular intervals
 				if latestMouseEvent != nil {
@@ -68,7 +68,7 @@ func onRPCMessageThrottled(message webrtc.DataChannelMessage, session *Session)
 	if isMouseEvent(request.Method) {
 		// Start the mouse worker if not already started
 		mouseWorkerOnce.Do(startMouseWorker)
-		
+
 		// Send to mouse worker (non-blocking)
 		select {
 		case mouseEventChan <- mouseEventData{message: message, session: session}: