Задача Python asyncio получила плохую доходность

Я смущен тем, как играть с модулем asyncio в Python 3.4. Я использую API searching для поисковой системы и хочу, чтобы каждый запрос на поиск выполнялся либо параллельно, либо асинхронно, так что мне не нужно ждать завершения одного поиска.

Вот мой высокоуровневый API поиска для создания некоторых объектов с необработанными результатами поиска. В самой поисковой системе используется какой-то механизм асинхронности, поэтому я не буду с этим документировать.

# No asyncio module used here now
class search(object):
  ...
  self.s = some_search_engine()
  ...
  def searching(self, *args, **kwargs):
    ret = {}
    # do some raw searching according to args and kwargs and build the wrapped results
    ...
    return ret

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

# Here is my testing script
@asyncio.coroutine
def handle(f, *args, **kwargs):
  r = yield from f(*args, **kwargs)
  return r

s = search()
loop = asyncio.get_event_loop()
loop.run_until_complete(handle(s.searching, arg1, arg2, ...))
loop.close()

Запустив pytest, он вернет RuntimeError: Task got bad yield : {results from searching...}, когда он попадет на строку r = yield from ....

Я также пробовал по-другому.

# same handle as above
def handle(..):
  ....
s = search()
loop = asyncio.get_event_loop()
tasks = [
        asyncio.async(handle(s.searching, arg11, arg12, ...)),
        asyncio.async(handle(s.searching, arg21, arg22, ...)),
        ...
        ]
loop.run_until_complete(asyncio.wait(tasks))
loop.close()

Запустив этот тестовый файл с помощью pytest, он пройдет, но какое-то странное исключение из поисковой системы поднимет. И он говорит Future/Task exception was never retrieved.

Вещи, которые я хочу задать:

  • Для моей первой попытки, правильно ли использовать yield from, вернув фактический результат из вызова функции?
  • Я думаю, мне нужно добавить немного сна к моему 2-му тестовому примеру, чтобы дождаться завершения задачи, но как мне это сделать? И как я могу получить свои вызовы функций для возврата во втором тестовом случае?
  • Это хороший способ реализовать asyncio с существующим модулем, создав обработчик async для обработки запросов?
  • Если ответ на вопрос 2 НЕТ, каждый клиент вызывает класс search, должен включать loop = get_event_loop() этот тип материалов для асинхронизации запросов?

Ответы

Ответ 1

Проблема заключается в том, что вы не можете просто вызвать существующий синхронный код, как если бы он был asyncio.coroutine и получить асинхронное поведение. Когда вы вызываете yield from searching(...), вы получите асинхронное поведение, если searching сам по себе является asyncio.coroutine или, по крайней мере, возвращает asyncio.Future. Прямо сейчас searching является просто регулярной синхронной функцией, поэтому вызов yield from searching(...) просто собирается выдать ошибку, потому что она не возвращает Future или сопрограмму.

Чтобы получить нужное поведение, вам нужно иметь асинхронную версию searching в дополнение к версии synchronous (или просто полностью отказаться от синхронной версии, если она вам не понадобится). У вас есть несколько вариантов поддержки обоих:

  • Перепишите searching как asyncio.coroutine, что он использует asyncio -компонентные вызовы для выполнения ввода-вывода, а не для блокировки ввода-вывода. Это заставит его работать в контексте asyncio, но это означает, что вы больше не сможете вызывать его прямо в синхронном контексте. Вместо этого вам также необходимо предоставить альтернативный синхронный метод searching, который запускает цикл событий asyncio и вызывает return loop.run_until_complete(self.searching(...)). Подробнее см. этот вопрос.

  • Сохраняйте синхронную реализацию searching и предоставляйте альтернативный асинхронный API, который использует BaseEventLoop.run_in_executor для запуска searching в фоновом потоке:

    class search(object):
  ...
  self.s = some_search_engine()
  ...
  def searching(self, *args, **kwargs):
    ret = {}
    ...
    return ret

   @asyncio.coroutine
   def searching_async(self, *args, **kwargs):
      loop = kwargs.get('loop', asyncio.get_event_loop())
      try:
          del kwargs['loop']  # assuming searching doesn't take loop as an arg
      except KeyError:
          pass
      r = yield from loop.run_in_executor(None, self.searching, *args)  # Passing None tells asyncio to use the default ThreadPoolExecutor
      return r

    Тестирование script:

    s = search()
loop = asyncio.get_event_loop()
loop.run_until_complete(s.searching_async(arg1, arg2, ...))
loop.close()

    Таким образом, вы можете сохранить свой синхронный код как есть и по крайней мере предоставить методы, которые могут использоваться в asyncio, без блокировки цикла событий. Это не так чистое решение, как если бы вы использовали асинхронный ввод-вывод в вашем коде, но лучше, чем ничего.

  • Предоставьте две полностью отдельные версии searching, которые используют блокирующий ввод-вывод, и один, который asyncio -совместим. Это дает идеальные реализации для обоих контекстов, но требует удвоенной работы.