hassanselim0 commented 3 months ago

Checklist

[x] I have included information about relevant versions
[x] I have verified that the issue persists when using the master branch of Mode.

Steps to reproduce

Create a Django backend that runs with gunicorn in gthread mode (eg: --workers=1 --threads=2)
In one of your views.py files import a module that tries to create a Faust App: consumer = faust.App(…)

Expected behavior

No exceptions happen

Actual behavior

A Runtime Error is raised, no current event loop in thread. The problem basically lies here: https://github.com/faust-streaming/mode/blob/0.4.0/mode/services.py#L599 (and the three other similar methods). Each Event should take self._loop and not self.loop as it used to do in older versions (like 0.2.x), because creating a Faust App doesn't necessarily mean I'll be running a consumer in this process (in my case I'm using it to create topics for the django app to use), so there might not be an event loop on the current thread, and in that case self._loop will be None (because I'm not passing a loop for the faust app constructor, and that arg gets passed all the way down to that Event) and that's "fine" because I'm not going to run the mode service or await on it or anything (and if I do, then the Event has a loop property that will rightfully try and get_event_loop, so it will still work).

Full traceback

Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/gunicorn/workers/gthread.py", line 271, in handle
    keepalive = self.handle_request(req, conn)
  …
  File "/usr/local/lib/python3.9/site-packages/django/urls/resolvers.py", line 595, in urlconf_module
    return import_module(self.urlconf_name)
  …
  File "/code/…/users/api/urls.py", line 4, in <module>
    from .views import UserView
  File "/code/…/users/api/views.py", line 103, in <module>
    from .topics import users_topic
  File "/code/…/users/api/topics.py", line 23, in <module>
    from .faust import users_consumer
  File "/code/…/faust.py", line 14, in <module>
    users_consumer = faust.App(
  File "/usr/local/lib/python3.9/site-packages/faust/app/base.py", line 466, in __init__
    self.agents = AgentManager(self, loop=loop)
  File "/usr/local/lib/python3.9/site-packages/faust/agents/manager.py", line 48, in __init__
    Service.__init__(self, **kwargs)
  File "/usr/local/lib/python3.9/site-packages/mode/services.py", line 583, in __init__
    self._started = self._new_started_event()
  File "/usr/local/lib/python3.9/site-packages/mode/services.py", line 599, in _new_started_event
    return Event(loop=self.loop)
  File "/usr/local/lib/python3.9/site-packages/mode/services.py", line 146, in loop
    self._loop = asyncio.get_event_loop_policy().get_event_loop()
  File "/usr/local/lib/python3.9/asyncio/events.py", line 642, in get_event_loop
    raise RuntimeError('There is no current event loop in thread %r.'
RuntimeError: There is no current event loop in thread 'ThreadPoolExecutor-0_0'.

Versions

Python version: 3.9.17
Mode version: 0.4.0
Operating system: Debian 10 (running in Docker)

wbarnha commented 3 months ago

Each Event should take self._loop and not self.loop as it used to do in older versions (like 0.2.x)

Looks like the changes were made in https://github.com/faust-streaming/mode/commit/96d4f72dab3ea3035d91c19db8f8fe9ee5d9d8e4 and originally used asyncio.Event objects until they got their own dedicated class. The change you're talking about was made by me in https://github.com/faust-streaming/mode/commit/cd40e6f23523be71fffee8ebf9ae9082e6ed5a4e to address https://github.com/faust-streaming/mode/issues/38 so that thread loops would shut down properly. It's been a while so I'll need to test your changes further.

loop.run_in_executor exists for similar reasons since there should only be one event loop per process, otherwise things get messy. I haven't used Faust with Django before and it's a bit in deprecated territory in my opinion. I've been wanting to move the project more towards FastAPI since it properly integrates with asynchronous Python patterns where Django creates more potentially blocking code that could mess up event loop timing.

By the way, do you only need these threads for creating Kafka topics? Because if so, I recommend using aiokafka to create the topics manually. Even though it's not officially supported yet, it has worked fine for me so far.

hassanselim0 commented 3 months ago

Sorry for the late reply, I slept and took a long holiday after writing this issue and related PR. I'm happy that the fix has been merged.

I haven't used Faust with Django before and it's a bit in deprecated territory in my opinion. I've been wanting to move the project more towards FastAPI since it properly integrates with asynchronous Python patterns where Django creates more potentially blocking code that could mess up event loop timing.

Yeah I've been stumbling onto hard-to-reproduce issues with consumers getting stuck and I believe it has to do with django blocking code (despite all db access being correctly wrapped in sync_to_async which I believe places blocking code onto a thread pool, which should be "fine"). So yeah, leaving away django is something I hope to do in future projects, but it's such a big headache to rebuild all the niceties I got used to into the FastAPI world.

faust-streaming / mode