transports(base_input): remove unnecessary task

CHANGELOG.md

@@ -28,6 +28,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Fixed an issue where `camera_out_enabled` would cause the highg CPU usage if
   no image was provided.
 
+### Performance
+
+- Removed unnecessary audio input tasks.
+
 ## [0.0.24] - 2024-05-29
 
 ### Added

src/pipecat/transports/base_input.py

@@ -59,10 +59,7 @@ class BaseInputTransport(FrameProcessor):
 
         if self._params.audio_in_enabled or self._params.vad_enabled:
             loop = self.get_event_loop()
-            self._audio_in_thread = loop.run_in_executor(
+            self._audio_thread = loop.run_in_executor(self._in_executor, self._audio_thread_handler)
-                self._in_executor, self._audio_in_thread_handler)
-            self._audio_out_thread = loop.run_in_executor(
-                self._in_executor, self._audio_out_thread_handler)
 
     async def stop(self):
         if not self._running:
@@ -73,8 +70,7 @@ class BaseInputTransport(FrameProcessor):
 
         # Wait for the threads to finish.
         if self._params.audio_in_enabled or self._params.vad_enabled:
-            await self._audio_in_thread
+            await self._audio_thread
-            await self._audio_out_thread
 
         self._push_frame_task.cancel()
 
@@ -167,7 +163,9 @@ class BaseInputTransport(FrameProcessor):
             vad_state = new_vad_state
         return vad_state
 
-    def _audio_in_thread_handler(self):
+    def _audio_thread_handler(self):
+        vad_state: VADState = VADState.QUIET
+
         sample_rate = self._params.audio_in_sample_rate
         num_channels = self._params.audio_in_channels
         num_frames = int(sample_rate / 100)  # 10ms of audio
@@ -179,32 +177,19 @@ class BaseInputTransport(FrameProcessor):
                         audio=audio_frames,
                         sample_rate=sample_rate,
                         num_channels=num_channels)
-                    self._audio_in_queue.put(frame)
+
+                    audio_passthrough = True
+
+                    # Check VAD and push event if necessary. We just care about
+                    # changes from QUIET to SPEAKING and vice versa.
+                    if self._params.vad_enabled:
+                        vad_state = self._handle_vad(frame.audio, vad_state)
+                        audio_passthrough = self._params.vad_audio_passthrough
+
+                    # Push audio downstream if passthrough.
+                    if audio_passthrough:
+                        future = asyncio.run_coroutine_threadsafe(
+                            self._internal_push_frame(frame), self.get_event_loop())
+                        future.result()
             except BaseException as e:
                 logger.error(f"Error reading audio frames: {e}")
-
-    def _audio_out_thread_handler(self):
-        vad_state: VADState = VADState.QUIET
-        while self._running:
-            try:
-                frame = self._audio_in_queue.get(timeout=1)
-
-                audio_passthrough = True
-
-                # Check VAD and push event if necessary. We just care about changes
-                # from QUIET to SPEAKING and vice versa.
-                if self._params.vad_enabled:
-                    vad_state = self._handle_vad(frame.audio, vad_state)
-                    audio_passthrough = self._params.vad_audio_passthrough
-
-                # Push audio downstream if passthrough.
-                if audio_passthrough:
-                    future = asyncio.run_coroutine_threadsafe(
-                        self._internal_push_frame(frame), self.get_event_loop())
-                    future.result()
-
-                self._audio_in_queue.task_done()
-            except queue.Empty:
-                pass
-            except BaseException as e:
-                logger.error(f"Error pushing audio frames: {e}")