updated version that uses ray queues so it doesn't block.

debug_app.py

@@ -22,50 +22,68 @@ logger = logging.getLogger(__name__)
 import subprocess
 import os
 
-class FFMpegConverter:
-    def __init__(self, input_format='mp3', output_format='s16le'):
-        self.input_format = input_format
+import ray
+from ray.util.queue import Queue
+if not ray.is_initialized():
+    # Try to connect to a running Ray cluster
+    ray_address = os.getenv('RAY_ADDRESS')
+    if ray_address:
+        ray.init(ray_address, namespace="project_charles")
+    else:
+        ray.init(namespace="project_charles")
+
+@ray.remote
+class FFMpegConverterActor:
+    def __init__(self, queue: Queue, buffer_size: int = 1920, output_format: str='s16le'):
+        self.queue = queue
+        self.buffer_size = 1920
         self.output_format = output_format
 
         self.input_pipe = None
         self.output_pipe = None
 
         self.process = None
-        self.start_process()
 
-    def start_process(self):
+    async def run(self):
+        while True:
+            chunk = await self.output_pipe.read(self.buffer_size)
+            await self.queue.put_async(chunk)       
+
+    async def start_process(self):
         cmd = [
             'ffmpeg',
-            # '-y',
-            # '-f', self.input_format,
             '-i', 'pipe:0',  # read from stdin
             '-f', self.output_format,
             '-ar', '48000',
             '-ac', '1',
-            # '-acodec', 'pcm_s16le',  # output format
             'pipe:1'  # write to stdout
         ]
 
-        self.process = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE)
+        self.process = await asyncio.create_subprocess_exec(
+            *cmd,
+            stdin=asyncio.subprocess.PIPE,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE
+        )
+
         self.input_pipe = self.process.stdin
         self.output_pipe = self.process.stdout
 
-    def push_chunk(self, chunk):
+        assert self.input_pipe is not None, "input_pipe was not initialized"
+        print (f"input_pipe: {self.input_pipe}")
+
+    async def push_chunk(self, chunk):
         try:
             self.input_pipe.write(chunk)
-            self.input_pipe.flush()
+            # await self.input_pipe.drain()
         except BrokenPipeError:
             # If the pipe is broken, restart the process.
-            self.start_process()
+            await self.start_process()
             self.input_pipe.write(chunk)
-            self.input_pipe.flush()
-
-    def read_output(self, num_bytes=1024):
-        frame = self.output_pipe.read(num_bytes)
-        return frame
+            # await self.input_pipe.drain()
 
-    def has_processed_all_data(self):
-        return self.process.poll() is not None
+    # def has_processed_all_data(self):
+    #     return self.process.poll() is not None
 
     def close(self):
         self.input_pipe.close()
@@ -83,9 +101,12 @@ with open("chunks.pkl", "rb") as f:
     import pickle
     debug_chunks = pickle.load(f)
 
-    converter = FFMpegConverter()
+    converter_queue = Queue(maxsize=100)
+    converter_actor = FFMpegConverterActor.remote(converter_queue)
+    ray.get(converter_actor.start_process.remote())
+    converter_actor.run.remote()
     for chunk in debug_chunks:
-        converter.push_chunk(chunk)
+        ray.get(converter_actor.push_chunk.remote(chunk))
 
 
 # emptry array of type int16
@@ -98,8 +119,12 @@ def process_frame(old_frame):
         output_sample_rate = 44100
         required_samples = old_frame.samples
 
-        frame_as_bytes = converter.read_output(required_samples*2*1) # 2 bytes per sample, mono
-        samples = np.frombuffer(frame_as_bytes, dtype=np.int16)
+        if not converter_queue.empty():
+            frame_as_bytes = converter_queue.get()
+            samples = np.frombuffer(frame_as_bytes, dtype=np.int16)
+        else: 
+            # create a byte array of zeros
+            samples = np.zeros((required_samples * 2 * 1), dtype=np.int16)
 
         # Duplicate mono channel for stereo
         if output_channels == 2: