Repository: gradio-app/gradio

Issue:
load events don't work well on HF spaces if queue enabled
### Describe the bug

I've noticed that some demos don't work well on HF spaces. If you go to them, you'll see that the queue is really large and doesn't go down.

The one thing these demos all have in common is that they have load events and the queue enabled.

* [blocks_random_slider](https://huggingface.co/spaces/gradio/blocks_random_slider)
![image](https://user-images.githubusercontent.com/41651716/218547980-aaad9595-74f3-4fa6-8280-4b2ff2769680.png)

* https://huggingface.co/spaces/gradio/blocks_random_slider_main
![image](https://user-images.githubusercontent.com/41651716/218548089-20ef2b0b-def7-4a2e-bd34-872c618cb3fa.png)

* https://huggingface.co/spaces/gradio/xgboost-income-prediction-with-explainability
![image](https://user-images.githubusercontent.com/41651716/218548218-5e6bfb26-d9bd-413e-b664-3c93c0015fc0.png)

* https://huggingface.co/spaces/gradio/native_plots
![image](https://user-images.githubusercontent.com/41651716/218548305-41f7cac8-4a92-4d7b-8c98-8b475614bed3.png)

* https://huggingface.co/spaces/gradio/altair_plot
![image](https://user-images.githubusercontent.com/41651716/218548498-cba59060-3f3a-4a72-b5eb-23d19496030c.png)

Running these demos locally works as expected.

The reason I think this is related to the queue is that this demo has load events with the queue disabled and it works well:
https://huggingface.co/spaces/gradio/timeseries-forecasting-with-prophet

![image](https://user-images.githubusercontent.com/41651716/218548896-ba00aefe-031d-4b5d-8bfb-bdb3e075c7e2.png)


### Is there an existing issue for this?

- [X] I have searched the existing issues

### Reproduction

-

### Screenshot

-

### Logs

```shell
-
```


### System Info

```shell
-
```


### Severity

serious, but I can work around it

Discussion:
gradio-pr-bot: All the demos for this PR have been deployed at https://huggingface.co/spaces/gradio-pr-deploys/pr-3196-all-demos
pngwn: Isn't five seconds a really long time?
freddyaboulton: @pngwn Yea we can do 1 second? Sending the message should be fast but don't want to set something too low that would impact people on slow networks. 
abidlabs: Just getting a chance to look at this. Very cool hunting down the issue. What you're proposing is reasonable, although I worry that a 5 second timeout might actually be too *little* given the fact that these websocket messages could contain large files particularly if multiple images are involved, which are sent as base64 strings

Also, it is a very good sign that:
* https://huggingface.co/spaces/gradio/native_plots
* https://huggingface.co/spaces/gradio/blocks_random_slider

But currently the other Spaces from the issue such as:
* https://huggingface.co/spaces/gradio/blocks_random_slider_main
* https://huggingface.co/spaces/gradio/altair_plot

are also running fine. Have you seen these Spaces build up a large queue while the others haven't?
freddyaboulton: @abidlabs Thanks for the review! Good point about the latency for large images. Let me toy with setting variable timeouts for the different messages. I think the getting the data and sending the result should have long timeouts (a minute) but the other message payloads are quite small and I think a second will suffice.

Will ping when this is ready for review again!
freddyaboulton: Also, altair_plot just got stuck for me and that demo does not include the timeouts. So I think it's a good sign that instituting some timeouts is beneficial. Just a matter of making sure we don't set it too small and shoot ourselves in the foot!

![image](https://user-images.githubusercontent.com/41651716/219430272-bff1a696-fea6-4a2a-95d4-45b992298ff3.png)

abidlabs: Can confirm the `altair_plots` plot is stuck for me, while `native_plots` and `blocks_random_slider` are looking good!
freddyaboulton: @abidlabs I set the `get_message` timeout to 60 seconds (that's where the biggest data will be sent) and all other timeouts to 1 second.

That lets me upload a 44 mb image to this space: https://huggingface.co/spaces/gradio/image_mod and not get a timeout. I think 60 seconds is reasonable since I don't think total image payloads above 50mb will be common. And for other file types we will only send the filename when #3191 is ready. That being said, we can also set it higher.

Also, users will be notified if they reached the timeout uploading their data:

![timeout_message](https://user-images.githubusercontent.com/41651716/219752381-e849d7b5-9a3a-4df8-bf34-12c81fe79e8c.png)


abidlabs: Getting some interesting behavior when testing out the deployed PRs. The `image_mod` demo is not successfully running a prediction:

<img width="1005" alt="image" src="https://user-images.githubusercontent.com/1778297/219765523-b48e8b6e-be83-4f05-8f28-79a3df0221d8.png">

Gonna take a deeper look later today but thought I'd flag this for now @freddyaboulton 
freddyaboulton: @abidlabs I can't reproduce 🤔 

![image](https://user-images.githubusercontent.com/41651716/219767238-9d016f93-049c-427e-8826-0fed5b922497.png)

abidlabs: I think this looks good. Let me just resurrect the queue benchmarking script and test the performance on this branch vs. main
abidlabs: Ran the benchmarking script on this branch vs. main and here's what I found as the avg latency over 1000 runs (seconds):

|test | `main`  | `3187-load-events-queue` |
| ------------- | ------------- | --- | 
| `text` | ` 1.7082971923546104` | ` 1.7173802871112676`
| `image` | `1.761333772516626` | `1.7226356868743897`
| `audio` | `1.7082971923546104` | ` 1.7681459769481371`
| `video` | ` 2.231178657029026` | `2.14179096728798`

Given the results, I think we can go ahead and merge
freddyaboulton: Thank you for testing @abidlabs 🙏 Will merge once ci passes!

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 9fea5bb9e..e6e15fdc9 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -8,10 +8,10 @@
 ## Bug Fixes:
 - Ensure `mirror_webcam` is always respected by [@pngwn](https://github.com/pngwn) in [PR 3245](https://github.com/gradio-app/gradio/pull/3245)
 - Fix issue where updated markdown links were not being opened in a new tab by [@gante](https://github.com/gante) in [PR 3236](https://github.com/gradio-app/gradio/pull/3236)
+- Added a timeout to queue messages as some demos were experiencing infinitely growing queues from active jobs waiting forever for clients to respond by [@freddyaboulton](https://github.com/freddyaboulton) in [PR 3196](https://github.com/gradio-app/gradio/pull/3196)
 - Fixes the height of rendered LaTeX images so that they match the height of surrounding text by [@abidlabs](https://github.com/abidlabs) in [PR 3258](https://github.com/gradio-app/gradio/pull/3258)
 - Fix bug where matplotlib images where always too small on the front end by [@freddyaboulton](https://github.com/freddyaboulton) in [PR 3274](https://github.com/gradio-app/gradio/pull/3274) 
 
-
 ## Documentation Changes:
 No changes to highlight.
 
@@ -35,6 +35,7 @@ No changes to highlight.
 
 ## Bug Fixes:
 - UI fixes including footer and API docs by [@aliabid94](https://github.com/aliabid94) in [PR 3242](https://github.com/gradio-app/gradio/pull/3242)
+- Updated image upload component to accept all image formats, including lossless formats like .webp by [@fienestar](https://github.com/fienestar) in [PR 3225](https://github.com/gradio-app/gradio/pull/3225)
 
 ## Documentation Changes:
 No changes to highlight.
diff --git a/gradio/queueing.py b/gradio/queueing.py
index 269c1f98e..dbc8514ce 100644
--- a/gradio/queueing.py
+++ b/gradio/queueing.py
@@ -4,6 +4,7 @@ import asyncio
 import copy
 import sys
 import time
+from asyncio import TimeoutError as AsyncTimeOutError
 from collections import deque
 from typing import Any, Deque, Dict, List, Tuple
 
@@ -205,7 +206,7 @@ class Queue:
         if self.live_updates:
             await self.broadcast_estimations()
 
-    async def gather_event_data(self, event: Event) -> bool:
+    async def gather_event_data(self, event: Event, receive_timeout=60) -> bool:
         """
         Gather data for the event
 
@@ -216,7 +217,20 @@ class Queue:
             client_awake = await self.send_message(event, {"msg": "send_data"})
             if not client_awake:
                 return False
-            event.data = await self.get_message(event)
+            data, client_awake = await self.get_message(event, timeout=receive_timeout)
+            if not client_awake:
+                # In the event, we timeout due to large data size
+                # Let the client know, otherwise will hang
+                await self.send_message(
+                    event,
+                    {
+                        "msg": "process_completed",
+                        "output": {"error": "Time out uploading data to server"},
+                        "success": False,
+                    },
+                )
+                return False
+            event.data = data
         return True
 
     async def notify_clients(self) -> None:
@@ -424,21 +438,25 @@ class Queue:
                 # to start "from scratch"
                 await self.reset_iterators(event.session_hash, event.fn_index)
 
-    async def send_message(self, event, data: Dict) -> bool:
+    async def send_message(self, event, data: Dict, timeout: float | int = 1) -> bool:
         try:
-            await event.websocket.send_json(data=data)
+            await asyncio.wait_for(
+                event.websocket.send_json(data=data), timeout=timeout
+            )
             return True
         except:
             await self.clean_event(event)
             return False
 
-    async def get_message(self, event) -> PredictBody | None:
+    async def get_message(self, event, timeout=5) -> Tuple[PredictBody | None, bool]:
         try:
-            data = await event.websocket.receive_json()
-            return PredictBody(**data)
-        except:
+            data = await asyncio.wait_for(
+                event.websocket.receive_json(), timeout=timeout
+            )
+            return PredictBody(**data), True
+        except AsyncTimeOutError:
             await self.clean_event(event)
-            return None
+            return None, False
 
     async def reset_iterators(self, session_hash: str, fn_index: int):
         await AsyncRequest(
diff --git a/gradio/routes.py b/gradio/routes.py
index ccb1453d0..8ddaff56a 100644
--- a/gradio/routes.py
+++ b/gradio/routes.py
@@ -12,6 +12,7 @@ import posixpath
 import secrets
 import tempfile
 import traceback
+from asyncio import TimeoutError as AsyncTimeOutError
 from collections import defaultdict
 from copy import deepcopy
 from typing import Any, Dict, List, Optional, Type
@@ -479,8 +480,20 @@ class App(FastAPI):
             await websocket.accept()
             # In order to cancel jobs, we need the session_hash and fn_index
             # to create a unique id for each job
-            await websocket.send_json({"msg": "send_hash"})
-            session_info = await websocket.receive_json()
+            try:
+                await asyncio.wait_for(
+                    websocket.send_json({"msg": "send_hash"}), timeout=1
+                )
+            except AsyncTimeOutError:
+                return
+
+            try:
+                session_info = await asyncio.wait_for(
+                    websocket.receive_json(), timeout=1
+                )
+            except AsyncTimeOutError:
+                return
+
             event = Event(
                 websocket, session_info["session_hash"], session_info["fn_index"]
             )


Test patch:
diff --git a/test/test_queueing.py b/test/test_queueing.py
index ebb4b9c96..47a1ec4ed 100644
--- a/test/test_queueing.py
+++ b/test/test_queueing.py
@@ -1,3 +1,4 @@
+import asyncio
 import os
 import sys
 from collections import deque
@@ -31,7 +32,7 @@ def queue() -> Queue:
 
 @pytest.fixture()
 def mock_event() -> Event:
-    websocket = MagicMock()
+    websocket = AsyncMock()
     event = Event(websocket=websocket, session_hash="test", fn_index=0)
     yield event
 
@@ -53,9 +54,20 @@ class TestQueueMethods:
 
     @pytest.mark.asyncio
     async def test_receive(self, queue: Queue, mock_event: Event):
+        mock_event.websocket.receive_json.return_value = {"data": ["test"], "fn": 0}
         await queue.get_message(mock_event)
         assert mock_event.websocket.receive_json.called
 
+    @pytest.mark.asyncio
+    async def test_receive_timeout(self, queue: Queue, mock_event: Event):
+        async def take_too_long():
+            await asyncio.sleep(1)
+
+        mock_event.websocket.receive_json = take_too_long
+        data, is_awake = await queue.get_message(mock_event, timeout=0.5)
+        assert data is None
+        assert not is_awake
+
     @pytest.mark.asyncio
     async def test_send(self, queue: Queue, mock_event: Event):
         await queue.send_message(mock_event, {})
@@ -85,7 +97,7 @@ class TestQueueMethods:
         queue.send_message = AsyncMock()
         queue.get_message = AsyncMock()
         queue.send_message.return_value = True
-        queue.get_message.return_value = {"data": ["test"], "fn": 0}
+        queue.get_message.return_value = {"data": ["test"], "fn": 0}, True
 
         assert await queue.gather_event_data(mock_event)
         assert queue.send_message.called
@@ -95,6 +107,25 @@ class TestQueueMethods:
         assert await queue.gather_event_data(mock_event)
         assert not (queue.send_message.called)
 
+    @pytest.mark.asyncio
+    async def test_gather_event_data_timeout(self, queue: Queue, mock_event: Event):
+        async def take_too_long():
+            await asyncio.sleep(1)
+
+        queue.send_message = AsyncMock()
+        queue.send_message.return_value = True
+
+        mock_event.websocket.receive_json = take_too_long
+        is_awake = await queue.gather_event_data(mock_event, receive_timeout=0.5)
+        assert not is_awake
+
+        # Have to use awful [1][0][1] syntax cause of python 3.7
+        assert queue.send_message.call_args_list[1][0][1] == {
+            "msg": "process_completed",
+            "output": {"error": "Time out uploading data to server"},
+            "success": False,
+        }
+
 
 class TestQueueEstimation:
     def test_get_update_estimation(self, queue: Queue):
@@ -193,6 +224,8 @@ class TestQueueProcessEvents:
         self, queue: Queue, mock_event: Event
     ):
         mock_event.websocket.send_json = AsyncMock()
+        mock_event.websocket.receive_json.return_value = {"data": ["test"], "fn": 0}
+
         mock_event.websocket.send_json.side_effect = ["2", ValueError("Can't connect")]
         queue.call_prediction = AsyncMock()
         mock_event.disconnect = AsyncMock()
@@ -260,6 +293,7 @@ class TestQueueProcessEvents:
     async def test_process_event_handles_error_sending_process_completed_msg(
         self, queue: Queue, mock_event: Event
     ):
+        mock_event.websocket.receive_json.return_value = {"data": ["test"], "fn": 0}
         mock_event.websocket.send_json = AsyncMock()
         mock_event.websocket.send_json.side_effect = [
             "2",
@@ -289,6 +323,7 @@ class TestQueueProcessEvents:
     async def test_process_event_handles_exception_during_disconnect(
         self, mock_request, queue: Queue, mock_event: Event
     ):
+        mock_event.websocket.receive_json.return_value = {"data": ["test"], "fn": 0}
         mock_event.websocket.send_json = AsyncMock()
         queue.call_prediction = AsyncMock(
             return_value=MagicMock(has_exception=False, json=dict(is_generating=False))