Queue will be blocked if `gradio.event_queue.Queue.process_event` has an exception

[SkyTNT]

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?

• I have searched the existing issues

Reproduction

First add an exception manually in gradio.event_queue.Queue.process_event

    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

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

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

Gradio3.3

How to fix this

Use try...finally...

    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

[abidlabs]

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

[SkyTNT]

I sometimes find my space app's queue is blocked and get error log like this

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

[aliabid94]

This seems critical @abidlabs, I can take a look

[freddyaboulton]

@aliabid94 Open PR for this in #2288