RuntimeError: одновременный вызов poll () с использованием сельдерея - PullRequest
Новая
       16

RuntimeError: одновременный вызов poll () с использованием сельдерея

5 голосов
/ 08 апреля 2019

При запуске Celery в Docker-контейнере, который получает restAPI из других контейнеров, я получаю вызов RuntimeError: concurrent poll ().

Кто-нибудь сталкивался с подобной ошибкой?

Я прикрепляю трассировку. 

Traceback (most recent call last):
  File "/usr/lib/python3.5/threading.py", line 914, in _bootstrap_inner
    self.run()
  File "/usr/lib/python3.5/threading.py", line 862, in run
    self._target(*self._args, **self._kwargs)
  File "/opt/www/api/api/training_call.py", line 187, in start_process
    result_state.get(on_message=self._on_raw_message, propagate=False)
  File "/usr/local/lib/python3.5/dist-packages/celery/result.py", line 226, in get
    on_message=on_message,
  File "/usr/local/lib/python3.5/dist-packages/celery/backends/asynchronous.py", line 188, in wait_for_pending
    for _ in self._wait_for_pending(result, **kwargs):
  File "/usr/local/lib/python3.5/dist-packages/celery/backends/asynchronous.py", line 255, in _wait_for_pending
    on_interval=on_interval):
  File "/usr/local/lib/python3.5/dist-packages/celery/backends/asynchronous.py", line 56, in drain_events_until
    yield self.wait_for(p, wait, timeout=1)
  File "/usr/local/lib/python3.5/dist-packages/celery/backends/asynchronous.py", line 65, in wait_for
    wait(timeout=timeout)
  File "/usr/local/lib/python3.5/dist-packages/celery/backends/redis.py", line 127, in drain_events
    message = self._pubsub.get_message(timeout=timeout)
  File "/usr/local/lib/python3.5/dist-packages/redis/client.py", line 3135, in get_message
    response = self.parse_response(block=False, timeout=timeout)
  File "/usr/local/lib/python3.5/dist-packages/redis/client.py", line 3034, in parse_response
    if not block and not connection.can_read(timeout=timeout):
  File "/usr/local/lib/python3.5/dist-packages/redis/connection.py", line 628, in can_read
    return self._parser.can_read() or self._selector.can_read(timeout)
  File "/usr/local/lib/python3.5/dist-packages/redis/selector.py", line 28, in can_read
    return self.check_can_read(timeout)
  File "/usr/local/lib/python3.5/dist-packages/redis/selector.py", line 156, in check_can_read
    events = self.read_poller.poll(timeout)
RuntimeError: concurrent poll() invocation

Ответы [ 2 ]

1 голос
/ 12 июня 2019

Соединение с брокером не является потокобезопасным, поэтому вам необходимо учитывать безопасность потоков в коде приложения. @ Laizer упомянул тикет , где эта ошибка была введена в базовой библиотеке python

Один из способов сделать это - обернуть все блокирующие вызовы до завершения задачи вобщий замок: 

import celery
import threading

@celery.shared_task
def debug_task(self):
    print('Hello, world')

def boom(nb_tasks):
    """ not thread safe - raises RuntimeError during concurrent executions """
    tasks = celery.group([debug_task.s() for _ in range(nb_tasks)])
    pool = tasks.apply_async()
    pool.join() # raised from here

CELERY_POLL_LOCK = threading.Lock()

def safe(nb_tasks):
    tasks = celery.group([debug_task.s() for _ in range(nb_tasks)])
    pool = tasks.apply_async()
    with CELERY_POLL_LOCK:  # prevents concurrent calls to poll()
        pool.join()

def main(nb_threads, nb_tasks_per_thread):
    for func in (safe, boom):
        threads = [threading.Thread(target=func, args=(nb_tasks_per_thread, )) for _ in range(nb_threads)]

        for a_thread in threads:
            a_thread.start()

        for a_thread in threads:
            a_thread.join()

main(10, 100)

Это наивный подход, который подходит мне, потому что я не ожидаю большого параллелизма и все задачи выполняются относительно быстро (~ 10 с).Если у вас другой «профиль», вам может понадобиться нечто более замысловатое (например, одна задача опроса, которая периодически опрашивает все ожидающие группы / задачи).

0 голосов
/ 10 апреля 2019

У меня была та же ошибка, что и приложение, которое использовало Redis pub / sub напрямую.Быстрое выполнение многих вызовов на redis.client.PubSub.getMessage привело к этому состоянию гонки.Моим решением было снизить скорость опроса новых сообщений.

...