витой реактор не работает должным образом внутри сельдерея - PullRequest
Новая
       39

витой реактор не работает должным образом внутри сельдерея

2 голосов
/ 09 апреля 2020

Сведения о системе / зависимостях: 

CPU --> 4
requirements --> celery==4.3.0, twisted==19.7.0 , python3.7

Ниже приведены настройки сельдерея, которые у меня есть 

from threading import Thread
from celery import Celery
from twisted.internet import threads, reactor, defer
from twisted.web.error import Error
from celery import signals

app =   Celery('tasks', broker='pyamqp://guest@localhost//')

@signals.worker_process_init.connect
def configure_infrastructure(**kwargs):
    Thread(target=reactor.run, name="reactor.run", args=(False,)).start()
    print('started new thread')

@signals.worker_process_shutdown.connect()
def shutdown_reactor(**kwargs):
    """
    This is invoked when the individual workers shut down. It just stops the twisted reactor
    @param kwargs:
    @return:
    """
    reactor.callFromThread(reactor.stop)
    print('REACTOR SHUTDOWN')

def getPage(inp):
    print(inp)
    return inp

def inThread():
    print('inside inthread method')
    try:
        result = threads.blockingCallFromThread(
            reactor, getPage, "http://twistedmatrix.com/")
    except Exception as exc:
        print(exc)
    else:
        print(result)


@app.task
def add(x, y):
    print('inside add method')
    inThread()
    return x + y

Работающий работник сельдерея, как показано ниже: 

celery -A run worker --loglevel=info

Журналы, когда начинается сельдерей: 

(2_env) ubuntu@gpy:~/app/env/src$ celery -A run worker --loglevel=info

[tasks]
  . run.add

[2020-04-09 07:25:29,357: WARNING/Worker-1] started new thread
[2020-04-09 07:25:29,362: WARNING/Worker-4] started new thread
[2020-04-09 07:25:29,362: WARNING/Worker-3] started new thread
[2020-04-09 07:25:29,364: WARNING/Worker-2] started new thread
[2020-04-09 07:25:29,367: INFO/MainProcess] Connected to amqp://guest:**@127.0.0.1:5672//

метод вызова, как показано ниже: 

>>> run.add.delay(1,2)
<AsyncResult: d41680fd-7cc1-4e75-81be-6496bad0cc16>
>>>

иногда я вижу, что он работает нормально. 

[2020-04-09 07:27:17,998: INFO/MainProcess] Received task: run.add[00934769-48c4-48b8-852c-8b746bdd5e03]
[2020-04-09 07:27:17,999: WARNING/Worker-4] inside add method
[2020-04-09 07:27:17,999: WARNING/Worker-4] inside inthread method
[2020-04-09 07:27:18,000: WARNING/Worker-4] http://twistedmatrix.com/
[2020-04-09 07:27:18,000: WARNING/Worker-4] http://twistedmatrix.com/
[2020-04-09 07:27:18,000: INFO/MainProcess] Task run.add[00934769-48c4-48b8-852c-8b746bdd5e03] succeeded in 0.00144551398989s: 3

Иногда я вижу, что он не может вызвать getPage метод и зависает, как показано ниже 

[2020-04-09 07:27:22,318: INFO/MainProcess] Received task: run.add[d41680fd-7cc1-4e75-81be-6496bad0cc16]
[2020-04-09 07:27:22,319: WARNING/Worker-2] inside add method
[2020-04-09 07:27:22,319: WARNING/Worker-2] inside inthread method

есть ли проблемы с использованием reactor.run внутри Thread?

ОБНОВЛЕНИЕ

Я положил печать в *twisted.internet.threads.blockingCallFromThread*. 

def blockingCallFromThread(reactor, f, *a, **kw):
    queue = Queue.Queue()
    def _callFromThread():
        print('inside _callFromThread')
        result = defer.maybeDeferred(f, *a, **kw)
        result.addBoth(queue.put)
    print('before calling _callFromThread')
    reactor.callFromThread(_callFromThread)
    print('after calling _callFromThread')
    result = queue.get()
    if isinstance(result, failure.Failure):
        result.raiseException()
    return result

Я вижу, что работник из сельдерея повесился только тогда, когда _callFromThread метод не вызывается в reactor.callFromThread(_callFromThread), но когда я вручную останавливаю работника с помощью CTRL + c, тогда я могу вызвать его.

Каждый раз, когда я останавливаю работника там, где работа зависла, он начинает обрабатывать задание.

Обновление: 27 апреля 2020 г.

Решено, если я использую вязание крючком для запуска скрученного реактора. Я обновляю нижеприведенную функцию. 

@signals.worker_process_init.connect
def configure_infrastructure(**kwargs):
    from crochet import setup
    setup()
    print('started new thread')

1 Ответ

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

С некоторой осторожностью, которую вы, похоже, проявили, вы можете запустить витой реактор в потоке one . Однако вы не сможете запустить его более чем в одном потоке, что, как я полагаю, происходит, когда вы используете его с Celery. У него есть как экземпляр, так и глобальное состояние, которое будет вытеснено, если его запустить в более чем одном потоке.

Вместо этого попробуйте использовать вязание крючком для координации вызовов на реактор, работающий в одном не -основная нить из любого количества тем, которые вам нравятся.

...