Repository: gradio-app/gradio
Pull request: fix : queue could be blocked
# Description

Surround the code in gradio.queue.Queue.process_event with try...finally... to ensure that the processed events in the queue are cleared to avoid the queue being blocked.

Closes: #2261

# Checklist:

- [X] I have performed a self-review of my own code
- [X] My code follows the style guidelines of this project
- [ ] I have commented my code in hard-to-understand areas
- [ ] I have made corresponding changes to the documentation
- [ ] I have added tests that prove my fix is effective or that my feature works
- [ ] New and existing unit tests pass locally with my changes


Problem statement:
Queue will be blocked if `gradio.event_queue.Queue.process_event` has an exception
### Describe the bug

If there is an exception in gradio.event_queue.Queue.process_event when predicting, the queue will be blocked.

### Is there an existing issue for this?

- [X] I have searched the existing issues

### Reproduction

First add an exception manually in `gradio.event_queue.Queue.process_event`

```python
    async def process_event(self, event: Event) -> None:
        client_awake = await self.gather_event_data(event)
        if not client_awake:
            return
        client_awake = await self.send_message(event, {"msg": "process_starts"})
        if not client_awake:
            return
        begin_time = time.time()
        response = await self.call_prediction(event)

        raise ValueError  # Here is the exception we add

        if response.json.get("is_generating", False):
            while response.json.get("is_generating", False):
                old_response = response
                await self.send_message(
                    event,
                    {
                        "msg": "process_generating",
                        "output": old_response.json,
                        "success": old_response.status == 200,
                    },
                )
                response = await self.call_prediction(event)
            await self.send_message(
                event,
                {
                    "msg": "process_completed",
                    "output": old_response.json,
                    "success": old_response.status == 200,
                },
            )
        else:
            await self.send_message(
                event,
                {
                    "msg": "process_completed",
                    "output": response.json,
                    "success": response.status == 200,
                },
            )
        end_time = time.time()
        if response.status == 200:
            self.update_estimation(end_time - begin_time)

        await event.disconnect()
        await self.clean_event(event)
```
Then create an example app

```python
import gradio as gr

if __name__ == '__main__':
    app = gr.Blocks()
    with app:
        text = gr.TextArea()
        btn = gr.Button()
        btn.click(lambda x: x + "test", [text], [text])
    app.launch(enable_queue=True)
```

Refresh the page multiple times, click the button repeatedly, you will see the queue is blocked.

### Screenshot

_No response_

### Logs

```shell
Task exception was never retrieved
future: <Task finished coro=<Queue.process_event() done, defined at C:\Users\iiiiiii\AppData\Local\Programs\Python\Python37\lib\site-packages\gradio\event_queue.py:221> exception=ValueError()>
Traceback (most recent call last):
  File "C:\Users\iiiiiii\AppData\Local\Programs\Python\Python37\lib\site-packages\gradio\event_queue.py", line 231, in process_event
    raise ValueError  # Here is the exception we add
ValueError
```


### System Info

```shell
Gradio3.3
```


### How to fix this

Use try...finally...

```python
    async def process_event(self, event: Event) -> None:
        try:
            client_awake = await self.gather_event_data(event)
            if not client_awake:
                return
            client_awake = await self.send_message(event, {"msg": "process_starts"})
            if not client_awake:
                return
            begin_time = time.time()
            response = await self.call_prediction(event)
            if response.json.get("is_generating", False):
                while response.json.get("is_generating", False):
                    old_response = response
                    await self.send_message(
                        event,
                        {
                            "msg": "process_generating",
                            "output": old_response.json,
                            "success": old_response.status == 200,
                        },
                    )
                    response = await self.call_prediction(event)
                await self.send_message(
                    event,
                    {
                        "msg": "process_completed",
                        "output": old_response.json,
                        "success": old_response.status == 200,
                    },
                )
            else:
                await self.send_message(
                    event,
                    {
                        "msg": "process_completed",
                        "output": response.json,
                        "success": response.status == 200,
                    },
                )
            end_time = time.time()
            if response.status == 200:
                self.update_estimation(end_time - begin_time)
        
        finally:
            await event.disconnect()
            await self.clean_event(event)
```


### Severity

serious, but I can work around it

Hints from issue discussion:
Hi @SkyTNT can you please provide more context for when you run into this issue? 

Specifically, why are you adding an Exception manually?

> First add an exception manually in gradio.event_queue.Queue.process_event
I sometimes find my space app's queue is blocked and get error log like this
```shell
ERROR:asyncio:Task exception was never retrieved
future: <Task finished name='Task-6912' coro=<Queue.process_event() done, defined at /home/user/.local/lib/python3.8/site-packages/gradio/event_queue.py:221> exception=AttributeError("'Request' object has no attribute '_json_response_data'")>
Traceback (most recent call last):
  File "/home/user/.local/lib/python3.8/site-packages/gradio/event_queue.py", line 230, in process_event
    if response.json.get("is_generating", False):
  File "/home/user/.local/lib/python3.8/site-packages/gradio/utils.py", line 592, in json
    return self._json_response_data
AttributeError: 'Request' object has no attribute '_json_response_data'
```
I don't know how to reproduce this. But i think any exception or return before `await self.clean_event(event)` will result in same problem.
@abidlabs

Commit message: fix: queue could be blocked

Patch to review:
diff --git a/gradio/queue.py b/gradio/queue.py
index a829317027..caabaeab26 100644
--- a/gradio/queue.py
+++ b/gradio/queue.py
@@ -220,49 +220,52 @@ async def call_prediction(self, event: Event):
         return response
 
     async def process_event(self, event: Event) -> None:
-        client_awake = await self.gather_event_data(event)
-        if not client_awake:
-            return
-        client_awake = await self.send_message(event, {"msg": "process_starts"})
-        if not client_awake:
-            return
-        begin_time = time.time()
-        response = await self.call_prediction(event)
-        if response.json.get("is_generating", False):
-            while response.json.get("is_generating", False):
-                old_response = response
+        try:
+            client_awake = await self.gather_event_data(event)
+            if not client_awake:
+                return
+            client_awake = await self.send_message(event, {"msg": "process_starts"})
+            if not client_awake:
+                return
+            begin_time = time.time()
+            response = await self.call_prediction(event)
+            if response.json.get("is_generating", False):
+                while response.json.get("is_generating", False):
+                    old_response = response
+                    await self.send_message(
+                        event,
+                        {
+                            "msg": "process_generating",
+                            "output": old_response.json,
+                            "success": old_response.status == 200,
+                        },
+                    )
+                    response = await self.call_prediction(event)
                 await self.send_message(
                     event,
                     {
-                        "msg": "process_generating",
+                        "msg": "process_completed",
                         "output": old_response.json,
                         "success": old_response.status == 200,
                     },
                 )
-                response = await self.call_prediction(event)
-            await self.send_message(
-                event,
-                {
-                    "msg": "process_completed",
-                    "output": old_response.json,
-                    "success": old_response.status == 200,
-                },
-            )
-        else:
-            await self.send_message(
-                event,
-                {
-                    "msg": "process_completed",
-                    "output": response.json,
-                    "success": response.status == 200,
-                },
-            )
-        end_time = time.time()
-        if response.status == 200:
-            self.update_estimation(end_time - begin_time)
-
-        await event.disconnect()
-        await self.clean_event(event)
+            else:
+                await self.send_message(
+                    event,
+                    {
+                        "msg": "process_completed",
+                        "output": response.json,
+                        "success": response.status == 200,
+                    },
+                )
+            end_time = time.time()
+            if response.status == 200:
+                self.update_estimation(end_time - begin_time)
+        finally:
+            try:
+                await event.disconnect()
+            finally:
+                await self.clean_event(event)
 
     async def send_message(self, event, data: Dict) -> bool:
         try:


Code review comments:
diff_hunk: @@ -220,49 +220,52 @@ async def call_prediction(self, event: Event):
         return response
 
     async def process_event(self, event: Event) -> None:
-        client_awake = await self.gather_event_data(event)
-        if not client_awake:
-            return
-        client_awake = await self.send_message(event, {"msg": "process_starts"})
-        if not client_awake:
-            return
-        begin_time = time.time()
-        response = await self.call_prediction(event)
-        if response.json.get("is_generating", False):
-            while response.json.get("is_generating", False):
-                old_response = response
+        try:
+            client_awake = await self.gather_event_data(event)
+            if not client_awake:
+                return
+            client_awake = await self.send_message(event, {"msg": "process_starts"})
+            if not client_awake:
+                return
+            begin_time = time.time()
+            response = await self.call_prediction(event)
+            if response.json.get("is_generating", False):
path: gradio/queue.py
text: @user2:
It's hard to know what the best solution is since we don't have a reproducer. However, looking at the [code](https://github.com/gradio-app/gradio/blob/main/gradio/utils.py#L499) for the Response class, the `json_response_data` is not set only if there is an exception raised while sending the request. Given that, I wonder if the better solution is to check  whether `response.is_valid()` or `response.has_exception` prior to accessing the JSON and if it is not valid, send the following message

```python
{"msg": "process_completed",
  "output": {"error": str(response.exception)}
 "success": False
 }
```

What do you think @user1 @author 

@author:
Sounds good. I think we also need to `raise response.exception`