Update to v4 results in "RuntimeError: Event loop is closed"

[stumpylog]

After upgrading to channels-redis==4.0.0, our celery tasks are all reporting the following traceback:

future: <Task finished name='Task-9' coro=<Connection.disconnect() done, defined at /usr/local/lib/python3.9/site-packages/red                                                                                                                                                                is/asyncio/connection.py:819> exception=RuntimeError('Event loop is closed')>
Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/redis/asyncio/connection.py", line 828, in disconnect
    self._writer.close()  # type: ignore[union-attr]
  File "/usr/local/lib/python3.9/asyncio/streams.py", line 353, in close
    return self._transport.close()
  File "/usr/local/lib/python3.9/asyncio/selector_events.py", line 698, in close
    self._loop.call_soon(self._call_connection_lost, None)
  File "/usr/local/lib/python3.9/asyncio/base_events.py", line 751, in call_soon
    self._check_closed()
  File "/usr/local/lib/python3.9/asyncio/base_events.py", line 515, in _check_closed
    raise RuntimeError('Event loop is closed')
RuntimeError: Event loop is closed
[2022-10-12 07:30:31,972] [ERROR] [asyncio] Task exception was never retrieved

Downgrading the image to channels-redis==3.4.1 resolves the issue, so I'm starting out here. This seems probably related to #312.

Image OS is Debian Bullseye, amd64. The django application is running with gunicorn.

Probably related packages:

celery==5.2.7
channels==3.0.5
channels-redis==4.0.0
hiredis==2.0.0
redis==4.3.4

Full Pipfile.lock: https://github.com/paperless-ngx/paperless-ngx/blob/dev/Pipfile.lock

[ljodal]

I’m seeing the exact same thing. Seem like the same issue as reported in #312

[carltongibson]

There was lots of discussion on the other issue. If you could investigate that would be handy.

A minimal reproduce would help too. It's hard to say anything without one. 🤔

[stumpylog]

The application is inherited, so I'm not really sure how the whole channels thing works yet or what connects where.

At best so far I can narrow it down to the only task which uses async_to_sync to send status updates. It will be called multiple times (roughly 10) to update a status from 0% to 100%

[aaronmader]

If I use ahaltindis's workaround detailed here, my code works. So it definitely seems to be a conflict/issue with async_to_sync.

loop = asyncio.get_event_loop()
coroutine = channel_layer.group_send(
    group_name,
    {
        'type': 'task.message',
        'text': context
    })
loop.run_until_complete(coroutine)

Perhaps channels, or channels_redis documentation could be improved, to provide an example of how users should trigger a group_send from a synchronous context (such that several calls can be made)? Perhaps a 'sync' version of group_send could be made available?

[ljodal]

I'm pretty sure the issue we have is very similar to what's reported above. We do this async_to_sync(channel_layer.group_send)(group_name, channel_layer_payload) from a sync gunicorn worker and I think that's where this stems from. The stack trace doesn't go back to our code though, so kinda hard to tell exactly.

[nblam1994]

I also get this issue after I upgrade to 4.0.0

[Elabbasy00]

Same here and sometimes

Task exception was never retrieved
future: <Task finished name='Task-378' coro=<Connection.disconnect() done, defined at /home/elabbasy/Desktop/BOT_TOOL/webserver/server/venv/lib/python3.10/site-packages/redis/asyncio/connection.py:819> exception=RuntimeError("Task <Task pending name='Task-378' coro=<Connection.disconnect() running at /home/elabbasy/Desktop/BOT_TOOL/webserver/server/venv/lib/python3.10/site-packages/redis/asyncio/connection.py:831>> got Future <Future pending> attached to a different loop")>
Traceback (most recent call last):
  File "/home/elabbasy/Desktop/BOT_TOOL/webserver/server/venv/lib/python3.10/site-packages/redis/asyncio/connection.py", line 831, in disconnect
    await self._writer.wait_closed()  # type: ignore[union-attr]
  File "/usr/lib/python3.10/asyncio/streams.py", line 344, in wait_closed
    await self._protocol._get_close_waiter(self)
RuntimeError: Task <Task pending name='Task-378' coro=<Connection.disconnect() running at /home/elabbasy/Desktop/BOT_TOOL/webserver/server/venv/lib/python3.10/site-packages/redis/asyncio/connection.py:831>> got Future <Future pending> attached to a different loop
Task exception was never retrieved
future: <Task finished name='Task-381' coro=<Connection.disconnect() done, defined at /home/elabbasy/Desktop/BOT_TOOL/webserver/server/venv/lib/python3.10/site-packages/redis/asyncio/connection.py:819> exception=RuntimeError("Task <Task pending name='Task-381' coro=<Connection.disconnect() running at /home/elabbasy/Desktop/BOT_TOOL/webserver/server/venv/lib/python3.10/site-packages/redis/asyncio/connection.py:831>> got Future <Future pending> attached to a different loop")>
Traceback (most recent call last):
  File "/home/elabbasy/Desktop/BOT_TOOL/webserver/server/venv/lib/python3.10/site-packages/redis/asyncio/connection.py", line 831, in disconnect
    await self._writer.wait_closed()  # type: ignore[union-attr]
  File "/usr/lib/python3.10/asyncio/streams.py", line 344, in wait_closed
    await self._protocol._get_close_waiter(self)
RuntimeError: Task <Task pending name='Task-381' coro=<Connection.disconnect() running at /home/elabbasy/Desktop/BOT_TOOL/webserver/server/venv/lib/python3.10/site-packages/redis/asyncio/connection.py:831>> got Future <Future pending> attached to a different loop
Task exception was never retrieved
future: <Task finished name='Task-382' coro=<Connection.disconnect() done, defined at /home/elabbasy/Desktop/BOT_TOOL/webserver/server/venv/lib/python3.10/site-packages/redis/asyncio/connection.py:819> exception=RuntimeError("Task <Task pending name='Task-382' coro=<Connection.disconnect() running at /home/elabbasy/Desktop/BOT_TOOL/webserver/server/venv/lib/python3.10/site-packages/redis/asyncio/connection.py:831>> got Future <Future pending> attached to a different loop")>
Traceback (most recent call last):
  File "/home/elabbasy/Desktop/BOT_TOOL/webserver/server/venv/lib/python3.10/site-packages/redis/asyncio/connection.py", line 831, in disconnect
    await self._writer.wait_closed()  # type: ignore[union-attr]
  File "/usr/lib/python3.10/asyncio/streams.py", line 344, in wait_closed
    await self._protocol._get_close_waiter(self)
RuntimeError: Task <Task pending name='Task-382' coro=<Connection.disconnect() running at /home/elabbasy/Desktop/BOT_TOOL/webserver/server/venv/lib/python3.10/site-packages/redis/asyncio/connection.py:831>> got Future <Future pending> attached to a different loop

[carltongibson]

Any chance of a minimal reproduce? It's hard to say anything with just a traceback

[Elabbasy00]

asgiref==3.5.2
channels==4.0.0
channels-redis==4.0.0
daphne==4.0.0

views.py

# passing request.data to serializer
if serializer.is_valid():
                serializer.save(interpreter=request.user)
                channel_layer = get_channel_layer()
                async_to_sync(channel_layer.group_send)(
                    "admins", {"type": "chat_message",  'message': {"command": "admin_new_ticket","ticket": serializer.data}})
                return Response(serializer.data, status=status.HTTP_200_OK)

and just normal consumer

import json
from channels.generic.websocket import AsyncWebsocketConsumer


class AdminsConsumer(AsyncWebsocketConsumer):

    async def connect(self):
        self.room_group_name = 'admins'

        # Join room group
        await self.channel_layer.group_add(
            self.room_group_name,
            self.channel_name
        )

        await self.accept()

    async def disconnect(self, close_code):
        # Leave room group
        await self.channel_layer.group_discard(self.room_group_name, self.channel_name)

    # Receive message from WebSocket
    async def receive(self, text_data):
        text_data_json = json.loads(text_data)
        message = text_data_json["message"]

        # Send message to room group
        await self.channel_layer.group_send(
            self.room_group_name, {"type": "chat_message", "message": message}
        )

    # Receive message from room group
    async def chat_message(self, event):
        message = event["message"]

        # Send message to WebSocket
        await self.send(text_data=json.dumps({"message": message}))

[abtinmo]

same issue here,

Task exception was never retrieved
future: <Task finished name='Task-90' coro=<Connection.disconnect() done, defined at /usr/local/lib/python3.8/site-packages/redis/asyncio/connection.py:819> exception=RuntimeError('Event loop is closed')>
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/redis/asyncio/connection.py", line 828, in disconnect
    self._writer.close()  # type: ignore[union-attr]
  File "/usr/local/lib/python3.8/asyncio/streams.py", line 353, in close
    return self._transport.close()
  File "/usr/local/lib/python3.8/asyncio/selector_events.py", line 692, in close
    self._loop.call_soon(self._call_connection_lost, None)
  File "/usr/local/lib/python3.8/asyncio/base_events.py", line 719, in call_soon
    self._check_closed()
  File "/usr/local/lib/python3.8/asyncio/base_events.py", line 508, in _check_closed
    raise RuntimeError('Event loop is closed')
RuntimeError: Event loop is closed

dependencies:

channels==4.0.0
channels-redis==4.0.0
daphne==4.0.0
Django==3.2.16
django-redis==5.0.0
redis==4.3.4

code:

async_to_sync(channel_layer.group_send)(
    group_name,
    {"type": "store.manager.handout", "data": {"type": request.POST["message_type"]}}
)

this comment did not worked either, I'm getting:

There is no current event loop in thread 'ThreadPoolExecutor-0_0'.

[bachloxo]

Any update this issue. I'm still face this issue