使用 gevent 执行池的 celery 任务的 SynchronousOnlyOperation
SynchronousOnlyOperation from celery task using gevent execution pool
给定芹菜 运行 这些选项:
celery -A openwisp2 worker -l info --pool=gevent --concurrency=15 -Ofair
鉴于此 celery task from openwisp-monitoring:
@shared_task
def perform_check(uuid):
"""
Retrieves check according to the passed UUID
and calls ``check.perform_check()``
"""
try:
check = get_check_model().objects.get(pk=uuid)
except ObjectDoesNotExist:
logger.warning(f'The check with uuid {uuid} has been deleted')
return
result = check.perform_check()
if settings.DEBUG: # pragma: nocover
print(json.dumps(result, indent=4, sort_keys=True))
大部分时间任务正常,但有时(通常是突发)会生成以下异常:
SynchronousOnlyOperation: You cannot call this from an async context - use a thread or sync_to_async.
完整堆栈跟踪:
SynchronousOnlyOperation: You cannot call this from an async context - use a thread or sync_to_async.
File "celery/app/trace.py", line 412, in trace_task
R = retval = fun(*args, **kwargs)
File "celery/app/trace.py", line 704, in __protected_call__
return self.run(*args, **kwargs)
File "openwisp_monitoring/check/tasks.py", line 44, in perform_check
check = get_check_model().objects.get(pk=uuid)
File "django/db/models/manager.py", line 85, in manager_method
return getattr(self.get_queryset(), name)(*args, **kwargs)
File "django/db/models/query.py", line 425, in get
num = len(clone)
File "django/db/models/query.py", line 269, in __len__
self._fetch_all()
File "django/db/models/query.py", line 1308, in _fetch_all
self._result_cache = list(self._iterable_class(self))
File "django/db/models/query.py", line 53, in __iter__
results = compiler.execute_sql(chunked_fetch=self.chunked_fetch, chunk_size=self.chunk_size)
File "django/db/models/sql/compiler.py", line 1154, in execute_sql
cursor = self.connection.cursor()
File "django/utils/asyncio.py", line 24, in inner
raise SynchronousOnlyOperation(message)
我不完全明白为什么会这样。
回顾一下,如有错误请指正:
- 通过这种配置,celery 可以并行执行任务,这种并行化是由 gevent 使用 asyncio 事件循环执行的
- gevent 然后调用每个任务,使用相同的线程
- 这些任务不是设计为异步的,它们使用纯同步代码,这些任务执行数据库查询和网络请求
- django 有一个
async_unsafe
装饰器用来装饰数据库操作,这个装饰器检查事件循环是否是 运行 并且在这种情况下它会引发一个 SynchronousOnlyOperation
但是为什么这个异常不是在 100% 的情况下出现,而是只在少数情况下出现?
这些任务确实有效,我可以看到它,因为它们产生了正常显示的图表数据集合,或者它们产生了设备模型中的状态变化(例如:从 ok 到 critical)。
是OpenWISP Monitoring的bug,配置错误还是Django的bug?
看起来 Django 中没有使用事件循环,但 Django 抛出了这个异常,尽管它并不关心它。这可能是一个错误,但希望在提交错误报告之前听取专家对该主题的意见。
我认为一个可能的快速解决方案可能是设置 env 变量 DJANGO_ALLOW_ASYNC_UNSAFE
但仅限于 celery 进程。
提前致谢。
事实证明我的假设是错误的,我使用的代码对 gevent 来说不是 greenlet 安全的。
由于暂时无法将代码重写为 greenlet 安全,因此我切换回了 prefork。
给定芹菜 运行 这些选项:
celery -A openwisp2 worker -l info --pool=gevent --concurrency=15 -Ofair
鉴于此 celery task from openwisp-monitoring:
@shared_task
def perform_check(uuid):
"""
Retrieves check according to the passed UUID
and calls ``check.perform_check()``
"""
try:
check = get_check_model().objects.get(pk=uuid)
except ObjectDoesNotExist:
logger.warning(f'The check with uuid {uuid} has been deleted')
return
result = check.perform_check()
if settings.DEBUG: # pragma: nocover
print(json.dumps(result, indent=4, sort_keys=True))
大部分时间任务正常,但有时(通常是突发)会生成以下异常:
SynchronousOnlyOperation: You cannot call this from an async context - use a thread or sync_to_async.
完整堆栈跟踪:
SynchronousOnlyOperation: You cannot call this from an async context - use a thread or sync_to_async.
File "celery/app/trace.py", line 412, in trace_task
R = retval = fun(*args, **kwargs)
File "celery/app/trace.py", line 704, in __protected_call__
return self.run(*args, **kwargs)
File "openwisp_monitoring/check/tasks.py", line 44, in perform_check
check = get_check_model().objects.get(pk=uuid)
File "django/db/models/manager.py", line 85, in manager_method
return getattr(self.get_queryset(), name)(*args, **kwargs)
File "django/db/models/query.py", line 425, in get
num = len(clone)
File "django/db/models/query.py", line 269, in __len__
self._fetch_all()
File "django/db/models/query.py", line 1308, in _fetch_all
self._result_cache = list(self._iterable_class(self))
File "django/db/models/query.py", line 53, in __iter__
results = compiler.execute_sql(chunked_fetch=self.chunked_fetch, chunk_size=self.chunk_size)
File "django/db/models/sql/compiler.py", line 1154, in execute_sql
cursor = self.connection.cursor()
File "django/utils/asyncio.py", line 24, in inner
raise SynchronousOnlyOperation(message)
我不完全明白为什么会这样。
回顾一下,如有错误请指正:
- 通过这种配置,celery 可以并行执行任务,这种并行化是由 gevent 使用 asyncio 事件循环执行的
- gevent 然后调用每个任务,使用相同的线程
- 这些任务不是设计为异步的,它们使用纯同步代码,这些任务执行数据库查询和网络请求
- django 有一个
async_unsafe
装饰器用来装饰数据库操作,这个装饰器检查事件循环是否是 运行 并且在这种情况下它会引发一个SynchronousOnlyOperation
但是为什么这个异常不是在 100% 的情况下出现,而是只在少数情况下出现?
这些任务确实有效,我可以看到它,因为它们产生了正常显示的图表数据集合,或者它们产生了设备模型中的状态变化(例如:从 ok 到 critical)。
是OpenWISP Monitoring的bug,配置错误还是Django的bug?
看起来 Django 中没有使用事件循环,但 Django 抛出了这个异常,尽管它并不关心它。这可能是一个错误,但希望在提交错误报告之前听取专家对该主题的意见。
我认为一个可能的快速解决方案可能是设置 env 变量 DJANGO_ALLOW_ASYNC_UNSAFE
但仅限于 celery 进程。
提前致谢。
事实证明我的假设是错误的,我使用的代码对 gevent 来说不是 greenlet 安全的。
由于暂时无法将代码重写为 greenlet 安全,因此我切换回了 prefork。