Initially reported in https://github.com/django/daphne/issues/163

I suspect this is due to the lack of calling close_old_connections, as seen in Channels 1 here: https://github.com/django/channels/blob/1.x/channels/signals.py#L10

As we discussed earlier in #494 we need live server test case for example to run selenium against it. This is work in progress PR, but I want some feedback already.

Few questions we need to answer before merging it.

1. Channels documentation mention test_channel_aliases as ability to test complex layers setup. I don't know what to do in this case for live server test. It is possible to spawn multiple live server instances. But its not clear to me what user API we should provide.

2. Regular live server thread contains connection override logic. I don't understand right now what problem does it solve. It will be awesome if any one can describe the reason it present there or provide test case for it.

3. How we can validate that twisted bind port successfully? For now I select first port from the possible range without any validation.

4. apply_routes setup test instance with overriding setUp and tearDown methods. Live server spawns its threads in the test class setup (as original one do). So apply routes hook works after we start live server threads and overrides not available in those threads. For now I change apply_routes decorator to work on class basis. But we also can consider running live server for each individual test method however this will slow things down and inconsistent with django behavior.

5. We need to write docs.

cc @bittner

I just upgraded Django channels to 1.1.3 as well as the following related libraries:

channels==1.1.3 daphne==1.2.0 asgi-redis==1.3.0 asgiref==1.1.1

Everything ran fine in my dev environment. But once I pushed to production, installed the new libraries, and completely restarted all services, I immediately started seeing the following error:

ERROR: HTTP/WS send decode error: Cannot dispatch message on channel u'daphne.response.dUTtOgoynk!bNiNpvqevU'

The client was showing the following error:

WebSocket connection to 'wss://<url>.com/' failed: Error during WebSocket handshake: Unexpected response code: 403

When I downgraded to the previous library versions, everything worked fine again.

It appears the issue is during the websocket connect / initial handshake?

I'm using the enforce_ordering decorator on all my websocket consumers as well as the asgi redis back-end.

Any thoughts on what's going on?

Yes, I updated all channels related stuff this morning :-)

Traceback (most recent call last):
  File "/home/brian/.pyenv/versions/3.6.4/envs/myrocc/lib/python3.6/site-packages/nose/suite.py", line 210, in run
    self.setUp()
  File "/home/brian/.pyenv/versions/3.6.4/envs/myrocc/lib/python3.6/site-packages/nose/suite.py", line 293, in setUp
    self.setupContext(ancestor)
  File "/home/brian/.pyenv/versions/3.6.4/envs/myrocc/lib/python3.6/site-packages/nose/suite.py", line 316, in setupContext
    try_run(context, names)
  File "/home/brian/.pyenv/versions/3.6.4/envs/myrocc/lib/python3.6/site-packages/nose/util.py", line 471, in try_run
    return func()
  File "/home/brian/.pyenv/versions/3.6.4/envs/myrocc/lib/python3.6/site-packages/django/test/testcases.py", line 1352, in setUpClass
    raise cls.server_thread.error
  File "/home/brian/.pyenv/versions/3.6.4/envs/myrocc/lib/python3.6/site-packages/channels/testing/live.py", line 28, in run
    self.daphne.run()
  File "/home/brian/.pyenv/versions/3.6.4/envs/myrocc/lib/python3.6/site-packages/daphne/server.py", line 132, in run
    reactor.run(installSignalHandlers=self.signal_handlers)
  File "/home/brian/.pyenv/versions/3.6.4/envs/myrocc/lib/python3.6/site-packages/twisted/internet/asyncioreactor.py", line 266, in run
    self.startRunning(installSignalHandlers=installSignalHandlers)
  File "/home/brian/.pyenv/versions/3.6.4/envs/myrocc/lib/python3.6/site-packages/twisted/internet/base.py", line 1222, in startRunning
    ReactorBase.startRunning(self)
  File "/home/brian/.pyenv/versions/3.6.4/envs/myrocc/lib/python3.6/site-packages/twisted/internet/base.py", line 730, in startRunning
    raise error.ReactorNotRestartable()
twisted.internet.error.ReactorNotRestartable: 

Currently channels has daphne as dependency. So it is not possible to install channels on one machine without installing daphne with its dependencies.

The only place in code of channels, which needs daphne, is the runserver command, which kind of a development command that should not be used in production. So there is no need to require daphne as a requirment for a production setup.

Could it be possible to create a package, that you can install without installing daphne as well?

Our background is as follows: We have a django-channels app, that has be used also on windows from low-tech-users. But it is quite hard to install a current version of twisted on windows. So we build geiss as an replacement for daphne, which does not need twisted. But with the current package of channels, it is not possible to get rid of twisted, even when we don't want to use daphne.

But I think, that there are other use cases where someone wants to install channels without daphne. For example if you want to separate the worker on different containers/hardware then the protocol server.

Pipfile:

[packages]
Django = "==3.1.3"
asgiref = "==3.2.10"
channels = "==3.0.2"
daphne = "==3.0.1"

(Note: asgiref 3.3.10 was also tried)

Code / Minimal example

I started a completely new Django + Django channels project to verify I can still see the same behaviour.

views.py:

def view_1(request):
    for i in range(10):
        time.sleep(1)
        print(request, i)

    return HttpResponse(b'Hello World #1!')


def view_2(request):
    return HttpResponse(b'Hello World #2!')

asgi.py:

import os

from channels.routing import ProtocolTypeRouter, URLRouter
from django.core.asgi import get_asgi_application
from django.urls import re_path

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'channels_test.settings')

application = ProtocolTypeRouter({
    'http': URLRouter([
        re_path('', get_asgi_application())
    ])
})

urls.py:

urlpatterns = [
    path('admin/', admin.site.urls),
    path('view1/', views.view_1),
    path('view2/', views.view_2),
]

I have two Django views. View1 and View2. Both are synchronous. As you can see, View1 sleeps for 10 seconds before returning a response. Calls to View2 respond instantaneously.

How you're running Channels (runserver? daphne/runworker? Nginx/Apache in front?)

Issue is seen in both runserver and also when running a single daphne worker. Nginx is serving the requests.

What Happened

If I open up a browser and go to localhost/view1, I can see in the console the printouts of the sleep for 10 seconds. If, while view1 is still processing, I go to localhost/view2, the page does not respond until view1 has finished processing.

What I expected

I should not expect View2 to have to wait for View1 to finish before returning a response, the requests should run concurrently. If I revert back to channels 2.4.0 with the following pip setup:

Pipfile:

[packages]
Django = "==3.1.3"
asgiref = "==3.2.10"
channels = "==2.4.0"
daphne = "==2.5.0"

and the following asgi.py:

import os

from channels.routing import ProtocolTypeRouter, URLRouter
from channels.routing import AsgiHandler
from django.urls import re_path

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'channels_test.settings')

import django
django.setup()
application = ProtocolTypeRouter({
    'http': URLRouter([
        re_path('', AsgiHandler)
    ])
})

I no longer see the behaviour of the ticket. As I did not expect a regression in behaviour for synchronous views after upgrading django / channels, I expected to see this same older behaviour after upgrading channels.

Workarounds

1. Convert the synchronous views to asynchronous views. This is not feasible for me as my entire project uses django rest framework which does not support asynchronous views (as Django does not provide async generic views).

2. Reverting back to Channels 2.X / Daphne 2.X.

Investigation / Notes

• While I know Django introduced async views, I did not expect a regression in behaviour for existing uses of synchronous views with ASGI. I did not see documentation indicating this behaviour should be expected.

• For my project, all APIs sent are now processed sequentially resulting in a very unresponsive front-end.

• I am certain this is down to the use of Django's ASGIHandler and its use of thread_sensitive=True in sync_to_async. Synchronous views are all placed in the same thread with thread_sensitive=True, so I'm assuming this is the cause of the issue - and wasn't an issue with channel's own version of ASGIHandler. If I access django.core.handlers.base and modify thread_sensitive of _get_response_async to False, the view responds instantly, like normal:

Line 228 of django.core.handlers.base

            if not asyncio.iscoroutinefunction(wrapped_callback):
                wrapped_callback = sync_to_async(wrapped_callback, thread_sensitive=False)

• I wasn't sure whether to report as a Channels or a Django issue. As using Django's get_asgi_application is part of the recommended way of upgrading channels from 2.X to 3.X, and this is all part of the Django ecosystem, it felt more appropriate here.

I am using a SyncConsumer which adds itself to a group on websocket.connect with group_add method of channel_layer.

class TaskConsumer(JsonWebsocketConsumer):
    def connect(self):
        self.accept()
        AsyncToSync(self.channel_layer.group_add)('task-i-1', self.channel_name)

    def task_message(self, event):
        self.send_json(event["text"])

If I try to send message to this group using group_send method that is wrapped with AsyncToSync, first message is succeed but further messages throw this exception:

>>> c = get_channel_layer() >>> AsyncToSync(c.group_send)('task-i-1', {'type': 'task.message', 'text':{}}) >>> AsyncToSync(c.group_send)('task-i-1', {'type': 'task.message', 'text':{}}) Connection <RedisConnection [db:0]> has pending commands, closing it. Traceback (most recent call last): File "/home/ahmet/websocket_channel/env/lib64/python3.6/site-packages/channel s_redis/core.py", line 316, in group_send await connection.zremrangebyscore(key, min=0, max=int(time.time()) - self.group_expiry) RuntimeError: Task <Task pending coro=<AsyncToSync.main_wrap() running at /home/ahmet/webso cket_channel/env/lib64/python3.6/site-packages/asgiref/sync.py:57> cb=[_run_until_complete_cb() at /usr/lib64/python3.6/asyncio/base_events.py:176]> got Future attached to a different loop

During handling of the above exception, another exception occurred:

Traceback (most recent call last): File "", line 1, in File "/home/ahmet/websocket_channel/env/lib64/python3.6/site-packages/asgiref /sync.py", line 49, in call call_result.result() File "/usr/lib64/python3.6/concurrent/futures/_base.py", line 425, in result return self.__get_result() File "/usr/lib64/python3.6/concurrent/futures/_base.py", line 384, in __get_result raise self._exception File "/home/ahmet/websocket_channel/env/lib64/python3.6/site-packages/asgiref /sync.py", line 57, in main_wrap result = await self.awaitable(*args, **kwargs) File "/home/ahmet/websocket_channel/env/lib64/python3.6/site-packages/channel s_redis/core.py", line 320, in group_send await connection.zrange(key, 0, -1) File "/home/ahmet/websocket_channel/env/lib64/python3.6/site-packages/aioredi s/commands/init.py", line 152, in exit self._release_callback(conn) File "/home/ahmet/websocket_channel/env/lib64/python3.6/site-packages/aioredi s/pool.py", line 361, in release conn.close() File "/home/ahmet/websocket_channel/env/lib64/python3.6/site-packages/aioredi s/connection.py", line 352, in close self._do_close(ConnectionForcedCloseError()) File "/home/ahmet/websocket_channel/env/lib64/python3.6/site-packages/aioredi s/connection.py", line 359, in _do_close self._writer.transport.close() File "/usr/lib64/python3.6/asyncio/selector_events.py", line 621, in close self._loop.call_soon(self._call_connection_lost, None) File "/usr/lib64/python3.6/asyncio/base_events.py", line 574, in call_soon self._check_closed() File "/usr/lib64/python3.6/asyncio/base_events.py", line 357, in _check_closed raise RuntimeError('Event loop is closed') RuntimeError: Event loop is closed

• OpenSuse Leap(42.2)
• python[3.6.4], django[2.0.2], channels[2.0.0], daphne[2.0.2], channels-redis[2.0.2]. asgiref[2.1.3]
• Django running with runserver

Hi, this is a duplicate of this question.

I'm trying to test new channels 2.0 with pytest-asyncio (0.8.0). If I place different assertions in the same function like:

import json
import pytest
from concurrent.futures._base import TimeoutError
from channels.testing import WebsocketCommunicator
from someapp.consumers import MyConsumer


@pytest.mark.django_db
@pytest.mark.asyncio
async def setup_database_and_websocket():
    path = 'foo'
    communicator = WebsocketCommunicator(MyConsumer, path)
    connected, subprotocol = await communicator.connect()
    assert connected
    return communicator


@pytest.mark.django_db
@pytest.mark.asyncio
async def test_1_and_2():
    communicator = await setup_database_and_websocket()
    sent = {"message": 'abc'}
    await communicator.send_json_to(sent)
    with pytest.raises(TimeoutError):
        await communicator.receive_from()
    await communicator.send_input({
        "type": "websocket.disconnect",
        "code": 1000,
    })

    communicator = await setup_database_and_websocket()
    sent = {"message": 1}
    await communicator.send_json_to(sent)
    with pytest.raises(TimeoutError):
        await communicator.receive_from()
    await communicator.send_input({
        "type": "websocket.disconnect",
        "code": 1000,
    })

then I'm not getting an error. But if I separate test cases like:

@pytest.mark.django_db
@pytest.mark.asyncio
async def test_1():
    communicator = await setup_database_and_websocket()
    sent = {"message": 'abc'}
    await communicator.send_json_to(sent)
    with pytest.raises(TimeoutError):
        await communicator.receive_from()
    await communicator.send_input({
        "type": "websocket.disconnect",
        "code": 1000,
    })


@pytest.mark.django_db
@pytest.mark.asyncio
async def test_2():
    communicator = await setup_database_and_websocket()
    sent = {"message": 1}
    await communicator.send_json_to(sent)
    with pytest.raises(TimeoutError):
        await communicator.receive_from()
    await communicator.send_input({
        "type": "websocket.disconnect",
        "code": 1000,
    })

then I'm getting following error upon the second receive_form call:

with pytest.raises(TimeoutError):
>           await communicator.receive_from()

someapp/tests/test_consumers_async.py:106: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
../../../.virtualenvs/some_env/lib/python3.6/site-packages/channels/testing/websocket.py:71: in receive_from
response = await self.receive_output(timeout)
../../../.virtualenvs/some_env/lib/python3.6/site-packages/asgiref/testing.py:66: in receive_output
self.future.result()
../../../.virtualenvs/some_env/lib/python3.6/site-packages/channels/consumer.py:54: in __call__
await await_many_dispatch([receive, self.channel_receive], self.dispatch)
../../../.virtualenvs/some_env/lib/python3.6/site-packages/channels/utils.py:48: in await_many_dispatch
await dispatch(result)
../../../.virtualenvs/some_env/lib/python3.6/site-packages/asgiref/sync.py:95: in __call__
return await asyncio.wait_for(future, timeout=None)
/usr/lib/python3.6/asyncio/tasks.py:339: in wait_for
return (yield from fut)
/usr/lib/python3.6/concurrent/futures/thread.py:56: in run
result = self.fn(*self.args, **self.kwargs)
../../../.virtualenvs/some_env/lib/python3.6/site-packages/channels/db.py:13: in thread_handler
return super().thread_handler(loop, *args, **kwargs)
../../../.virtualenvs/some_env/lib/python3.6/site-packages/asgiref/sync.py:110: in thread_handler
return self.func(*args, **kwargs)
../../../.virtualenvs/some_env/lib/python3.6/site-packages/channels/consumer.py:99: in dispatch
handler(message)
../../../.virtualenvs/some_env/lib/python3.6/site-packages/channels/generic/websocket.py:19: in websocket_connect
self.connect()
someapp/consumers.py:22: in connect
self.group_name, self.channel_name)
../../../.virtualenvs/some_env/lib/python3.6/site-packages/asgiref/sync.py:49: in __call__
return call_result.result()
/usr/lib/python3.6/concurrent/futures/_base.py:432: in result
return self.__get_result()
/usr/lib/python3.6/concurrent/futures/_base.py:384: in __get_result
raise self._exception
../../../.virtualenvs/some_env/lib/python3.6/site-packages/asgiref/sync.py:63: in main_wrap
result = await self.awaitable(*args, **kwargs)
../../../.virtualenvs/some_env/lib/python3.6/site-packages/channels_redis/core.py:290: in group_add
await connection.expire(group_key, self.group_expiry)
../../../.virtualenvs/some_env/lib/python3.6/site-packages/aioredis/commands/__init__.py:152: in __exit__
self._release_callback(conn)
../../../.virtualenvs/some_env/lib/python3.6/site-packages/aioredis/pool.py:361: in release
conn.close()
../../../.virtualenvs/some_env/lib/python3.6/site-packages/aioredis/connection.py:352: in close
self._do_close(ConnectionForcedCloseError())
../../../.virtualenvs/some_env/lib/python3.6/site-packages/aioredis/connection.py:359: in _do_close
self._writer.transport.close()
/usr/lib/python3.6/asyncio/selector_events.py:621: in close
self._loop.call_soon(self._call_connection_lost, None)
/usr/lib/python3.6/asyncio/base_events.py:574: in call_soon
self._check_closed()
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <_UnixSelectorEventLoop running=False closed=True debug=False>

def _check_closed(self):
if self._closed:
>           raise RuntimeError('Event loop is closed')
E           RuntimeError: Event loop is closed

/usr/lib/python3.6/asyncio/base_events.py:357: RuntimeError

Also if I do (as in https://channels.readthedocs.io/en/latest/topics/testing.html):

await communicator.disconnect()

instead of:

await communicator.send_input({
    "type": "websocket.disconnect",
    "code": 1000,
})

then the following error is raised:

>       await communicator.disconnect()

someapp/tests/test_consumers_async.py:96: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
../../../.virtualenvs/some_env/lib/python3.6/site-packages/channels/testing/websocket.py:100: in disconnect
    await self.future
../../../.virtualenvs/some_env/lib/python3.6/site-packages/channels/consumer.py:54: in __call__
    await await_many_dispatch([receive, self.channel_receive], self.dispatch)
../../../.virtualenvs/some_env/lib/python3.6/site-packages/channels/utils.py:48: in await_many_dispatch
    await dispatch(result)
../../../.virtualenvs/some_env/lib/python3.6/site-packages/asgiref/sync.py:95: in __call__
    return await asyncio.wait_for(future, timeout=None)
/usr/lib/python3.6/asyncio/tasks.py:339: in wait_for
    return (yield from fut)
/usr/lib/python3.6/concurrent/futures/thread.py:56: in run
    result = self.fn(*self.args, **self.kwargs)
../../../.virtualenvs/some_env/lib/python3.6/site-packages/channels/db.py:13: in thread_handler
    return super().thread_handler(loop, *args, **kwargs)
../../../.virtualenvs/some_env/lib/python3.6/site-packages/asgiref/sync.py:110: in thread_handler
    return self.func(*args, **kwargs)
../../../.virtualenvs/some_env/lib/python3.6/site-packages/channels/consumer.py:99: in dispatch
    handler(message)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <someapp.consumers.ChatConsumer object at 0x7f38fcc55240>
message = {'code': 1000, 'type': 'websocket.disconnect'}

    def websocket_disconnect(self, message):
        """
            Called when a WebSocket connection is closed. Base level so you don't
            need to call super() all the time.
            """
        # TODO: group leaving
>       self.disconnect(message["code"])
E       TypeError: disconnect() takes 1 positional argument but 2 were given

What should I do to separate those test cases in the respective individual test functions?

So this was the very first scenario in which the Single thread executor error was found and that lead to me opening django/asgiref#275

While trying to get a simple repro-case for it, we figured out a way to trigger an error related to it in a very simple way and this was fixed with https://github.com/django/asgiref/releases/tag/3.4.1

But testing the new 3.4.1 version against our code-base still yielded the same 500 errors while serving static files (at least) in the dev environment.

I've updated https://github.com/rdmrocha/asgiref-thread-bug with this new repro-case, by loading a crapload of JS files (1500) but that can be changed in the views.py file.

It doesn't ALWAYS happen (so you might need a hard-refresh or two) but when it does, you'll be greeted with something like this:

I believe this is still related https://github.com/django/asgiref/commit/13d0b82a505a753ef116e11b62a6dfcae6a80987 as reverting to v3.3.4 via requirements.txt makes the error go away.

Looking at the offending code inside channels/http.py it looks like this might be a thread exhaustion issue but this is pure speculation.

since the handle is decorated as sync_to_async: This is forcing the send to become sync and we're waiting on it like this: await self.handle(scope, async_to_sync(send), body_stream). If there's no more threads available, I speculate that they might end up in a deadlock waiting for the unwrap of this await async_to_sync(async_to_sync) call, eventually triggering the protection introduced in https://github.com/django/asgiref/commit/13d0b82a505a753ef116e11b62a6dfcae6a80987

But take this last part with a grain of salt as this is pure speculation without diving into the code and debugging it. Hope it helps

Lifespan is a new addition to the ASGI specification that provides events pertaining to the main event loop. You can find its documentation here: https://asgi.readthedocs.io/en/latest/specs/lifespan.html

Is there any interest in implementing this in Django Channels?

Hello,

I currently have a problem deploying my Django application with django-channels. I'm trying to setup daphne with nginx and for that, I made this ASGI file:

import os
from channels.asgi import get_channel_layer

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "myapp.settings_production")

channel_layer = get_channel_layer()

But somehow, it keeps using myapp.settings (and thus, fails at the database connection)

I use this command to launch daphne: daphne myapp.asgi:channel_layer -v 2

Python version: 2.7.11+ Django version: 1.9.7 django-channels version: 0.15.0

Am I doing something wrong?

Why

Currently the official way is to use channels_redis for layer functionality but this requires an extra service which adds complexity to setups.

Python has already some wonderful infrastructure to do this without extra services: the multiprocessing module.

It would be really nice to have an full fledged implementation of layers on base of multiprocessing.

Initial work

I experimented a bit and this is some initial work (basing on InMemoryChannelLayer and tests for it). I am currently short in time to develop it myself

https://github.com/devkral/channels-multiprocessing

State: Non-functional, hangs up. Combining multiprocessing with async is hard.

An unexpected behaviour occurs when trying to send a message to the group via the Redis channel when called from the celery shared task in the except block or any subsequent code if the exception occurs.

OS: macOS: 13.0.1 (22A400) Browser: Google Chrome (Version 108.0.5359.98 (Official Build) (arm64)) Server: Django Daphne ASGI Server

pip freeze

amqp==5.1.1
asgiref==3.5.2
async-timeout==4.0.2
attrs==22.1.0
autobahn==22.7.1
Automat==22.10.0
autopep8==2.0.0
azure-core==1.26.1
azure-identity==1.12.0
azure-storage-blob==12.14.1
billiard==3.6.4.0
celery==5.2.7
certifi==2022.9.14
cffi==1.15.1
channels==4.0.0
channels-redis==4.0.0
charset-normalizer==2.1.1
click==8.1.3
click-didyoumean==0.3.0
click-plugins==1.1.1
click-repl==0.2.0
colorama==0.4.5
constantly==15.1.0
cryptography==38.0.1
daphne==4.0.0
Deprecated==1.2.13
Django==4.0.6
django-cors-headers==3.13.0
django-filter==22.1
django-grip==3.2.0
django-rest-passwordreset==1.3.0
django-storages==1.13.1
django-storages-azure==1.6.8
djangorestframework==3.13.1
djangorestframework-simplejwt==5.2.0
et-xmlfile==1.1.0
fillpdf==0.7.2
gripcontrol==4.1.0
gunicorn==20.1.0
hyperlink==21.0.0
idna==3.4
incremental==22.10.0
install==1.3.5
isodate==0.6.1
kombu==5.2.4
lxml==4.9.1
Markdown==3.4.1
MarkupSafe==2.1.1
msal==1.18.0
msal-extensions==1.0.0
msgpack==1.0.4
msrest==0.7.1
ntlm-auth==1.5.0
numpy==1.23.5
oauthlib==3.2.2
Office365-REST-Python-Client==2.3.14
openpyxl==3.0.10
packaging==21.3
pandas==1.5.2
pdf2image==1.16.0
pdfrw2==0.5.0
Pillow==9.3.0
portalocker==2.6.0
prompt-toolkit==3.0.31
psycopg2-binary==2.9.5
pubcontrol==3.3.0
pyasn1==0.4.8
pyasn1-modules==0.2.8
pycodestyle==2.10.0
pycparser==2.21
PyJWT==2.4.0
PyMuPDF==1.21.0
pyOpenSSL==22.1.0
pyparsing==3.0.9
PyPDF2==2.11.2
python-dateutil==2.8.2
pytz==2022.1
redis==4.3.4
requests==2.28.1
requests-ntlm==1.1.0
requests-oauthlib==1.3.1
requests-toolbelt==0.9.1
service-identity==21.1.0
SharePlum==0.5.1
six==1.16.0
sqlparse==0.4.2
tomli==2.0.1
tqdm==4.64.1
Twisted==22.10.0
txaio==22.2.1
typing_extensions==4.4.0
tzdata==2022.1
urllib3==1.26.12
vine==5.0.0
wcwidth==0.2.5
Werkzeug==2.2.2
whitenoise==6.2.0
wrapt==1.14.1
zope.interface==5.5.2

I have the following celery shared task:

@shared_task(bind=True)
def upload_to_sharepoint_task(self, user_id: str, document_type: str, upload_file: str, filename: str) -> None:
    """Uploads a file to SharePoint."""
    task_id = self.request.id

    try:
        upload_to_sharepoint(user_id, document_type, upload_file, filename)

        print(f"Task {task_id} completed successfully.")

        async_to_sync(get_channel_layer().group_send)(task_id, {
            "type": "send_task_status",
            "message": {'task_id': task_id, 'status': status, "error": error}
        })
   except Exception as e:
       print(f"Task {task_id} failed to execute. Posting message to the group.")
       async_to_sync(get_channel_layer().group_send)(task_id, {
            "type": "send_task_status",
            "message": {'task_id': task_id, 'status': status, "error": error}
       })
       print("Message send to the group.")

Expected Behaviour When the exception is thrown, an error message is sent via channels to the subscribed group, and the client is notified.

Actual Behaviour The problem is that when an exception is thrown, then both print messages are called; however, no message is sent to the group, and therefore client gets no notification. When no exception is thrown, then everything works as expected.

Logs No logs are provided as the system acts like everything is alright.

Please also try and include, if you can:

Channels Configuration Daphne and build the server of the React client. An issue occurred in the development environment.

settings.py :

CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "channels_redis.core.RedisChannelLayer",
        "CONFIG": {
            "hosts": [("127.0.0.1", 6379)],
        },
    },
}

asgi.py

import os

from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
from django.core.asgi import get_asgi_application
from channels.security.websocket import AllowedHostsOriginValidator


os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'octo_manager.settings')

# Initialize Django ASGI application early to ensure the AppRegistry
# is populated before importing code that may import ORM models.
django_asgi_app = get_asgi_application()


import api.routing

application = ProtocolTypeRouter({
    "http": django_asgi_app,
    "websocket": AllowedHostsOriginValidator(
            AuthMiddlewareStack(URLRouter(api.routing.websocket_urlpatterns))
        ),
})

Problem when we use channels v3, after some time in production, connection to establish websockets start to work more and more slowly, sometimes it take 1-3 seconds to accept websockets connection and with more time on prod time increases to 170 sec and more, and at the end nginx kill connections (in our setup it`s 300 sec) and you see 502 timeouts.

when we use --limit-max-requests=1000 and --proxy-headers we see errors [Exception in ASGI application], but after we remove --limit-max-requests=1000 and --proxy-headers all works as expected (only in channels v3 in v4 nothing is working).

Django==4.0.6
daphne==3.0.2
asgiref==3.5.2
channels==3.0.5
channels-redis==3.3.0
websockets==10.4
uvicorn==0.19.0
uvloop==0.17.0
=========================
| ERROR:    Exception in ASGI application
| Traceback (most recent call last):
|   File "/usr/local/lib/python3.10/site-packages/uvicorn/protocols/websockets/websockets_impl.py", line 230, in run_asgi
|     result = await self.app(self.scope, self.asgi_receive, self.asgi_send)
|   File "/usr/local/lib/python3.10/site-packages/uvicorn/middleware/proxy_headers.py", line 78, in __call__
|     return await self.app(scope, receive, send)
|   File "/usr/local/lib/python3.10/site-packages/channels/routing.py", line 71, in __call__
|     return await application(scope, receive, send)
|   File "/usr/local/lib/python3.10/site-packages/channels/routing.py", line 150, in __call__
|     return await application(
|   File "/usr/local/lib/python3.10/site-packages/channels/consumer.py", line 94, in app
|     return await consumer(scope, receive, send)
|   File "/usr/local/lib/python3.10/site-packages/channels/consumer.py", line 58, in __call__
|     await await_many_dispatch(
|   File "/usr/local/lib/python3.10/site-packages/channels/utils.py", line 58, in await_many_dispatch
|     await task
| TypeError: An asyncio.Future, a coroutine or an awaitable is required

=========================
## uvicorn
uvicorn --host=0.0.0.0 --port=8002 --workers=4 --loop=uvloop --ws=websockets --limit-max-requests=1000 --proxy-headers --log-level=debug core.asgi:application

# simple test class that produce this error

```python
class TestWebSocket(AsyncJsonWebsocketConsumer):
    async def connect(self):
        try:
            await self.create_channel()
            await self.accept()
       except Exception as exc:
            logger.exception(exc)
            await self.close()

    @database_sync_to_async
    def create_channel(self):
         # some logic with database.
        return

Django==4.1.3
daphne==4.0.0
asgiref==3.5.2
channels==4.0.0
channels-redis==4.0.0
websockets==10.4
uvicorn==0.20.0
uvloop==0.17.0
=========================
| ERROR:    Exception in ASGI application
| Traceback (most recent call last):
|   File "/usr/local/lib/python3.10/site-packages/uvicorn/protocols/websockets/websockets_impl.py", line 230, in run_asgi
|     result = await self.app(self.scope, self.asgi_receive, self.asgi_send)
|   File "/usr/local/lib/python3.10/site-packages/uvicorn/middleware/proxy_headers.py", line 78, in __call__
|     return await self.app(scope, receive, send)
|   File "/usr/local/lib/python3.10/site-packages/channels/routing.py", line 71, in __call__
|     return await application(scope, receive, send)
|   File "/usr/local/lib/python3.10/site-packages/channels/routing.py", line 150, in __call__
|     return await application(
|   File "/usr/local/lib/python3.10/site-packages/channels/consumer.py", line 94, in app
|     return await consumer(scope, receive, send)
|   File "/usr/local/lib/python3.10/site-packages/channels/consumer.py", line 58, in __call__
|     await await_many_dispatch(
|   File "/usr/local/lib/python3.10/site-packages/channels/utils.py", line 58, in await_many_dispatch
|     await task
| AttributeError: 'list' object has no attribute 'decode'
=========================
uvicorn --host=0.0.0.0 --port=8002 --workers=4 --loop=uvloop --ws=websockets --limit-max-requests=1000 --proxy-headers --log-level=debug core.asgi:application

OS

• MacOs Monterey
• Amazon Linux AMI 2018.03

PIP Freeze

Problem

While testing, it seems that when we have some numbers of requests at the same time, the exception SynchronousOnlyOperation is thrown, after testing for a bit, I found that this part of the code here:

def notify_websocket_of_update(self, id, updated):
    channel_layer = get_channel_layer()
    payload = self.build_websocket_payload(id, 'update', updated)
    async_to_sync(channel_layer.group_send)("name", payload)

and when I comment the async_to_sync(channel_layer.group_send)("name", payload) line, the problem stops from happening. Which makes me almost certain that the problems happens because of the async context inside .group_send

To me, it seems that when we hit some async event_loop other requests coming in that are reusing the thread stumbles to the exception. But then I thought that the async_to_sync should solve this problem, or am I wrong?

At this point, I'm mainly checking if this is a common behavior, or if anyone had face this problem before and could share a fix.

StackTrace

I know this StackTrace is not directly related to the code snippet I put above, but it's an example of the problem that is created when I have the channel group_send method call (when I remove the .group_send call, this exception never happens).

Traceback (most recent call last):
  File "/Users/ms/team/project/venv/lib/python3.9/site-packages/django/core/handlers/exception.py", line 47, in inner
    response = get_response(request)
  File "/Users/ms/team/project/venv/lib/python3.9/site-packages/django/utils/deprecation.py", line 116, in __call__
    response = self.process_request(request)
  File "/Users/ms/team/project/helpers/tenant.py", line 33, in process_request
    tenant = self.get_tenant(domain_model, hostname)
  File "/Users/ms/team/project/venv/lib/python3.9/site-packages/django_tenants/middleware/main.py", line 27, in get_tenant
    domain = domain_model.objects.select_related('tenant').get(domain=hostname)
  File "/Users/ms/team/project/venv/lib/python3.9/site-packages/django/db/models/query.py", line 431, in get
    num = len(clone)
  File "/Users/ms/team/project/venv/lib/python3.9/site-packages/django/db/models/query.py", line 262, in __len__
    self._fetch_all()
  File "/Users/ms/team/project/venv/lib/python3.9/site-packages/django/db/models/query.py", line 1324, in _fetch_all
    self._result_cache = list(self._iterable_class(self))
  File "/Users/ms/team/project/venv/lib/python3.9/site-packages/django/db/models/query.py", line 51, in __iter__
    results = compiler.execute_sql(chunked_fetch=self.chunked_fetch, chunk_size=self.chunk_size)
  File "/Users/ms/team/project/venv/lib/python3.9/site-packages/django/db/models/sql/compiler.py", line 1173, in execute_sql
    cursor = self.connection.cursor()
  File "/Users/ms/team/project/venv/lib/python3.9/site-packages/django/utils/asyncio.py", line 24, in inner
    raise SynchronousOnlyOperation(message)
django.core.exceptions.SynchronousOnlyOperation: You cannot call this from an async context - use a thread or sync_to_async.

Service

The service is started using gunicorn, on a wsgi server.

• Ubuntu 22.04, Firefox 106.0.3, python 3.6(yes, we know it's old, we are migrating starting next year, but as far as I read the docs it's still supported)
• A pip freeze output showing your package versions

aiocontextvars==0.2.2
aioredis==1.3.1
alabaster==0.7.12
amqp==5.1.1
arrow==1.2.3
asgiref==3.4.1
async-timeout==4.0.2
attrs==22.1.0
autobahn==21.2.1
Automat==20.2.0
Babel==2.10.3
billiard==3.6.4.0
cached-property==1.5.2
celery==5.1.2
certifi==2022.6.15.1
cffi==1.15.1
channels==3.0.4
channels-redis==2.4.2
charset-normalizer==2.0.12
click==7.1.2
click-didyoumean==0.3.0
click-plugins==1.1.1
click-repl==0.2.0
colorama==0.4.5
ConfigArgParse==1.5.3
constantly==15.1.0
construct==2.10.56
contextvars==2.4
cryptography==38.0.1
cycler==0.11.0
daphne==3.0.2
dataclasses==0.8
decorator==5.1.1
Deprecated==1.2.13
Django==3.2.15
django-grappelli==3.0.3
django-mathfilters==1.0.0
django-wkhtmltopdf==3.4.0
docutils==0.18.1
Dozer==0.8
em==0.4.0
ephem==4.1.3
Flask==2.0.3
Flask-Executor==1.0.0
FormEncode==2.0.1
geographiclib==1.52
geopy==2.2.0
gphoto2==2.3.4
gps3==0.33.3
hiredis==2.0.0
hyperlink==21.0.0
idna==3.3
imagesize==1.4.1
immutables==0.18
importlib-metadata==4.8.3
importlib-resources==5.4.0
imutils==0.5.4
incremental==21.3.0
iniconfig==1.1.1
itsdangerous==2.0.1
Jinja2==3.0.3
kiwisolver==1.3.1
kombu==5.1.0
lark==1.1.2
loguru==0.6.0
Mako==1.1.6
MarkupSafe==2.0.1
massedit==0.69.1
matplotlib==3.3.4
memory-profiler==0.60.0
mido==1.2.10
mpmath==1.2.1
msgpack==0.6.2
msgpack-python==0.5.6
mysqlclient==2.1.1
natsort==8.2.0
netifaces==0.11.0
numpy==1.19.5
opencv-python==4.5.4.60
osa==0.2
packaging==21.3
paho-mqtt==1.6.1
pandas==1.1.5
pendulum==2.1.2
piexif==1.1.3
Pillow==8.4.0
pkg_resources==0.0.0
pluggy==1.0.0
prompt-toolkit==3.0.31
protobuf==3.19.4
psutil==5.9.2
py==1.11.0
pyasn1==0.4.8
pyasn1-modules==0.2.8
pycparser==2.21
PyDispatcher==2.0.6
Pygments==2.13.0
Pympler==1.0.1
pynmea2==1.18.0
pyOpenSSL==22.0.0
pyparsing==3.0.7
pyserial==3.5
pytest==7.0.1
python-dateutil==2.8.2
python-gammu==2.6
python-rtmidi==1.4.9
pytz==2022.2.1
pytzdata==2020.1
pyudev==0.23.2
PyYAML==6.0
pyzmq==23.2.1
redis==4.3.4
redis-dump-load==1.1
requests==2.27.1
schedule==1.1.0
scipy==1.5.4
seaborn==0.11.2
service-identity==21.1.0
six==1.16.0
snowballstemmer==2.2.0
Sphinx==5.1.1
sphinxcontrib-applehelp==1.0.2
sphinxcontrib-devhelp==1.0.2
sphinxcontrib-htmlhelp==2.0.0
sphinxcontrib-jsmath==1.0.1
sphinxcontrib-qthelp==1.0.3
sphinxcontrib-serializinghtml==1.1.5
SQLObject==3.9.1
sqlparse==0.4.2
sympy==1.9
tomli==1.2.3
tqdm==4.63.2
transitions==0.9.0
Twisted==22.4.0
txaio==22.2.1
typing_extensions==4.1.1
uptime==3.0.1
urllib3==1.26.12
validators==0.20.0
vine==5.0.0
Wand==0.6.10
wcwidth==0.2.5
WebOb==1.8.7
Werkzeug==2.0.3
wrapt==1.14.1
zipp==3.6.0
zmq==0.0.0
zope.interface==5.4.0

• Channels http.py handle method errors with following stacktrace

2022-11-09 16:38:14.502 ERROR daphne.server:server.py:293 Exception inside application: 'NoneType' object has no attribute 'items'
Traceback (most recent call last):
  File "/home/username/app/lib/python3.6/site-packages/channels/staticfiles.py", line 41, in __call__
    dict(scope, static_base_url=self.base_url), receive, send
  File "/home/username/app/lib/python3.6/site-packages/channels/staticfiles.py", line 56, in __call__
    return await super().__call__(scope, receive, send)
  File "/home/username/app/lib/python3.6/site-packages/channels/http.py", line 198, in __call__
    await self.handle(scope, async_to_sync(send), body_stream)
  File "/home/username/app/lib/python3.6/site-packages/asgiref/sync.py", line 444, in __call__
    ret = await asyncio.wait_for(future, timeout=None)
  File "/usr/lib/python3.6/asyncio/tasks.py", line 339, in wait_for
    return (yield from fut)
  File "/usr/lib/python3.6/concurrent/futures/thread.py", line 56, in run
    result = self.fn(*self.args, **self.kwargs)
  File "/home/username/app/lib/python3.6/site-packages/asgiref/sync.py", line 486, in thread_handler
    return func(*args, **kwargs)
  File "/home/username/app/lib/python3.6/site-packages/channels/http.py", line 255, in handle
    for response_message in self.encode_response(response):
  File "/home/username/app/lib/python3.6/site-packages/channels/http.py", line 311, in encode_response
    for header, value in response.items():
AttributeError: 'NoneType' object has no attribute 'items'

Djangos get_response method seems to return a None. Not sure if this is a channels bug and it should be able to handle None as a response or this is a Django bug and self.get_response(request) should never return None

• How you're running Channels (runserver? daphne/runworker? Nginx/Apache in front?) sudo /home/username/app/bin/python manage.py runserver 0.0.0.0:80 --insecure no nginx or apache

The application definition is:

from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
from channels.security.websocket import AllowedHostsOriginValidator
from django.urls import re_path

application = ProtocolTypeRouter({
    # (http->django views is added by default)
    'websocket': AllowedHostsOriginValidator(
        AuthMiddlewareStack(
            URLRouter([
                re_path(r"^app_name/ws", AppConsumer.as_asgi()),
            ])
        )
    )
})

and the consumer is defined as:

from django.db import close_old_connections
from channels.generic.websocket import AsyncJsonWebsocketConsumer

class AppConsumer(AsyncJsonWebsocketConsumer):
    async def connect(self):
        await self.accept()
        await self.channel_layer.group_add("app_ws", self.channel_name)

    async def disconnect(self, close_code):
        await self.channel_layer.group_discard("app_ws", self.channel_name)

    async def app_message(self, event):
        await self.send_json(event)

    async def receive_json(self, content, **kwargs):
        if 'request' in content:
            close_old_connections()
            await self.send_json({"text": get_text(), "channel": content['request']})