From f7f0c44c3241fbeb1d99b5ba2d82510044c42f15 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= <aleix@daily.co>
Date: Thu, 24 Oct 2024 10:35:37 -0700
Subject: [PATCH 1/3] transports(daily): don't block event handlers

---
 src/pipecat/transports/services/daily.py | 11 +++++++++--
 1 file changed, 9 insertions(+), 2 deletions(-)

diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py
index da232c224..05cb96b0d 100644
--- a/src/pipecat/transports/services/daily.py
+++ b/src/pipecat/transports/services/daily.py
@@ -584,8 +584,15 @@ class DailyTransportClient(EventHandler):
         )
 
     def _call_async_callback(self, callback, *args):
-        future = asyncio.run_coroutine_threadsafe(callback(*args), self._loop)
-        future.result()
+        # Don't wait on the coroutine, otherwise if we call a `CallClient`
+        # function and wait for its completion this will currently result in a
+        # deadlock. This is because `_call_async_callback` is used inside
+        # `CallClient` event handlers which are holding the GIL in
+        # `daily-python`. So if the `callback` passed here makes a `CallClient`
+        # call and waits for it to finish using completions (and a future) we
+        # will deadlock because completions use event handlers (which are
+        # holding the GIL).
+        asyncio.run_coroutine_threadsafe(callback(*args), self._loop)
 
 
 class DailyInputTransport(BaseInputTransport):

From 88362db0345c6dff8e391d692beb8844f6426e1c Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= <aleix@daily.co>
Date: Thu, 24 Oct 2024 10:42:48 -0700
Subject: [PATCH 2/3] transports(daily): no more need for an output message
 queue

---
 src/pipecat/transports/services/daily.py | 29 +-----------------------
 1 file changed, 1 insertion(+), 28 deletions(-)

diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py
index 05cb96b0d..87fe497cf 100644
--- a/src/pipecat/transports/services/daily.py
+++ b/src/pipecat/transports/services/daily.py
@@ -741,37 +741,21 @@ class DailyOutputTransport(BaseOutputTransport):
 
         self._client = client
 
-        # Task to process outgoing messages.
-        self._messages_task = None
-        self._messages_queue = asyncio.Queue()
-
     async def start(self, frame: StartFrame):
         # Parent start.
         await super().start(frame)
         # Join the room.
         await self._client.join()
-        # Start messages task
-        self._messages_task = self.get_event_loop().create_task(self._messages_task_handler())
 
     async def stop(self, frame: EndFrame):
         # Parent stop.
         await super().stop(frame)
-        # Cancel messages task
-        if self._messages_task:
-            self._messages_task.cancel()
-            await self._messages_task
-            self._messages_task = None
         # Leave the room.
         await self._client.leave()
 
     async def cancel(self, frame: CancelFrame):
         # Parent stop.
         await super().cancel(frame)
-        # Cancel messages task
-        if self._messages_task:
-            self._messages_task.cancel()
-            await self._messages_task
-            self._messages_task = None
         # Leave the room.
         await self._client.leave()
 
@@ -780,7 +764,7 @@ class DailyOutputTransport(BaseOutputTransport):
         await self._client.cleanup()
 
     async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame):
-        await self._messages_queue.put(frame)
+        await self._client.send_message(frame)
 
     async def write_raw_audio_frames(self, frames: bytes):
         await self._client.write_raw_audio_frames(frames)
@@ -788,17 +772,6 @@ class DailyOutputTransport(BaseOutputTransport):
     async def write_frame_to_camera(self, frame: OutputImageRawFrame):
         await self._client.write_frame_to_camera(frame)
 
-    async def _messages_task_handler(self):
-        while True:
-            try:
-                message = await self._messages_queue.get()
-                await self._client.send_message(message)
-                self._messages_queue.task_done()
-            except asyncio.CancelledError:
-                break
-            except Exception as e:
-                logger.exception(f"{self} error processing message queue: {e}")
-
 
 class DailyTransport(BaseTransport):
     def __init__(

From 2e6b5d1843dc0a770a77be78124008675124e6d1 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= <aleix@daily.co>
Date: Thu, 24 Oct 2024 10:43:15 -0700
Subject: [PATCH 3/3] transports(daily): fix aiohttp timeout

---
 src/pipecat/transports/services/daily.py | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py
index 87fe497cf..3f13aa9de 100644
--- a/src/pipecat/transports/services/daily.py
+++ b/src/pipecat/transports/services/daily.py
@@ -935,7 +935,9 @@ class DailyTransport(BaseTransport):
             url = f"{self._params.api_url}/dialin/pinlessCallUpdate"
 
             try:
-                async with session.post(url, headers=headers, json=data, timeout=10) as r:
+                async with session.post(
+                    url, headers=headers, json=data, timeout=aiohttp.ClientTimeout(total=10)
+                ) as r:
                     if r.status != 200:
                         text = await r.text()
                         logger.error(