Как реализовать беспрерывный тайм-сон в Торнадо?

Question

Я пишу процесс (называемый процессом запроса), который периодически отправляет HTTP-запросы, но может быть прерван другим потоком / процессом (называемым основным процессом) в любое время. Первоначально я использовал пул потоков для запросов с multiprocessing.Event Объект для ожидания соответствующей задержки перед отправкой следующего запроса:

# in the main process
poison_pill_event= multiprocessing.Event()

# pass "poison_pill_event" to the requesting process

# in the requesting process which sends requests every wait_interval seconds
poison_pill_event.wait(timeout=time_interval)

Это позволило бы мне иметь прерываемую версию time.sleep(), на случай, если мне нужно будет завершить процесс. Иначе, это позволило бы коду продолжить работу после тайм-аута.

Недавно я обнаружил, что, учитывая мою необходимую пропускную способность и ресурсы, асинхронный дизайн является более подходящей альтернативой, и я попытался использовать Tornado. Начал с использования gen.sleepкак неблокирующее ожидание, но оно не может быть прервано. Затем переключился на использование функции ожидания Toro.Event, чтобы разрешить прерывания. Тем не менее, Toro.Event's delay отличается от многопроцессорности. Событие timeout в том, что это вызывает исключение Timeout и таким образом прекращает выполнение. Кроме того, я не верю, что могу разделить его между процессами, поэтому сейчас я пытаюсь объединить мой основной процесс с процессом запроса, но это не должно быть слишком сложным.

Итак, мой вопрос, как я могу переформулировать поведение, которое я имел с multiprocessing.Event.wait в Торнадо?

Answer 1

Вам больше не нужен Торо. Tornado 4.2 и более поздние версии включают все функции Toro.

Попробуйте что-то вроде этого, с условием вместо события:

import datetime
import logging

from tornado import gen, options
from tornado.ioloop import IOLoop
from tornado.locks import Condition

condition = Condition()


@gen.coroutine
def waiter():
    for _ in range(10):
        yield condition.wait(timeout=datetime.timedelta(seconds=1))
        logging.info("Tick")


@gen.coroutine
def notifier():
    yield gen.sleep(4.5)
    logging.info("Notify")
    condition.notify()


@gen.coroutine
def runner():
    # Yield two Futures; wait for waiter() and notifier() to finish.
    yield [waiter(), notifier()]


options.parse_command_line()  # Configures logging.
IOLoop.current().run_sync(runner)

Вы увидите вывод журнала как:

[I 160712 12:00:28 foo:15] Tick
[I 160712 12:00:29 foo:15] Tick
[I 160712 12:00:30 foo:15] Tick
[I 160712 12:00:31 foo:15] Tick
[I 160712 12:00:31 foo:21] Notify
[I 160712 12:00:31 foo:15] Tick
[I 160712 12:00:32 foo:15] Tick
[I 160712 12:00:33 foo:15] Tick
[I 160712 12:00:34 foo:15] Tick
[I 160712 12:00:35 foo:15] Tick
[I 160712 12:00:36 foo:15] Tick

Обратите внимание, что во втором, когда "Уведомить" было зарегистрировано, есть два "Тика", но в противном случае Тики появляются только один раз в секунду.