Правильная работа с asyncio (корутинами)

#python #python_3x #асинхронность #asyncio

Захотелось мне поразбираться с механизмом асинхронной работы через корунтины (сопрограммы)
используя стандартную библиотеку asyncio.

Написал тестовый пример и удивился тому, что асинхронный код не такой эффективный
как я думал.

Возможно, я неправильно работал с asyncio и есть более эффективный способ.

Прошу помочь с этим вопросом.



Код:

from timeit import default_timer as clock
from multiprocessing.dummy import Pool as ThreadPool
import asyncio
import hashlib


def my_algo(*args):
    next_value = b'HELLO WORLD!' * 10000000

    value = hashlib.sha1(next_value).hexdigest()
    next_value = bytes(value, 'utf-8')

    value = hashlib.sha256(next_value).hexdigest()
    next_value = bytes(value, 'utf-8')

    value = hashlib.sha512(next_value).hexdigest()
    next_value = bytes(value, 'utf-8')

    value = hashlib.md5(next_value).hexdigest()

    return value


@asyncio.coroutine
def foo(i):
    return my_algo(i)


def my_sync(items):
    new_items = []

    for i in items:
        x = my_algo(i)
        new_items.append(x)

    # print(min(new_items), max(new_items))


def my_threaded(items):
    pool = ThreadPool()
    new_items = pool.map(my_algo, items)

    # print(min(new_items), max(new_items))


def my_asyncio(items):
    ioloop = asyncio.get_event_loop()

    tasks = [foo(i) for i in items]

    wait_tasks = asyncio.wait(tasks)
    results = ioloop.run_until_complete(wait_tasks)[0]

    new_items = [x.result() for x in results]
    # print(min(new_items), max(new_items))

    ioloop.close()


if __name__ == '__main__':
    t = clock()
    print('my_algo')
    my_algo()
    print('  Elapsed:', clock() - t)
    print('\n')

    items = [1] * 100

    t = clock()
    print('my_sync')
    my_sync(items)
    print('  Elapsed:', clock() - t)
    print()

    t = clock()
    print('my_threaded')
    my_threaded(items)
    print('  Elapsed:', clock() - t)
    print()

    t = clock()
    print('my_asyncio')
    my_asyncio(items)
    print('  Elapsed:', clock() - t)


Результаты:

my_algo
  Elapsed: 0.22824488105526344


my_sync
  Elapsed: 23.38838779162658

my_threaded
  Elapsed: 6.768628799269269

my_asyncio
  Elapsed: 22.96235436593546

    

Ответы

Ответ 1

Хэширование - это CPU-bound операция, в то время как асинхронность позволяет получить
преимущество только в I/O-bound операциях.

Предположим, что на всех графиках ось time - это константа. Последовательное выполнение
my_algo трижды будет выглядеть так



Асинхронное выполнение трёх задач, выполняющих ту же функцию my_algo так



То есть это не настоящая параллельность, выполнение задач просто чередуется в том
же промежутке времени. Обычно, это получается ещё и немного медленнее из-за необходимости
переключения контекста.

Выигрыш появился бы в том случае, если бы внутри my_algo осуществлялся ввод\вывод,
блокирующий выполнение.



Тогда заблокированная задача могла бы на время ожидания отдать управление другой,
ещё не заблокированной.

Графики взяты из "Введения в Twisted"

Разница между asyncio.Event и asyncio.Future для реализации событий

#python #asyncio

В питоновой асинхронщине есть asyncio.Event, который позволяет уведомить соседний
таск о каком-нибудь событии:

import asyncio
event = asyncio.Event()

async def task1():
    event.set()

async def task2():
    while True:
        await event.wait()
        event.clear()
        # do_some_work()


Однако для тех же задач можно попробовать использовать и просто asyncio.Future:

import asyncio
event = asyncio.Future()

async def task1():
    if not event.done():
        event.set_result(True)

async def task2():
    global event
    while True:
        await event
        event = asyncio.Future()  # clear
        # do_some_work()


Эти примеры на первый взгляд эквивалентны и работают одинаково.

Но, заглянув в реализацию threading.Event, я увидел, что там используется несколько
футур в self._waiters = collections.deque(), и для каждого ожидающего создаётся своя
отдельная футура.

Однако ведь вроде можно await'ить одну и ту же футуру несколько раз. Значит, для
всех ожидающих достаточно одной футуры.

Или нельзя?

Короче, зачем asyncio.Event устроен так сложно? Зачем в нём deque и несколько футур?
На какие грабли я рискую напороться, если вместо asyncio.Event буду использовать одну-единственную
футуру, как во втором примере?
    

Ответы

Ответ 1

В своих экспериментах я забыл про отмену тасков. Если запустить task2 из второго
примера несколько раз, а потом один из этих task2 отменить:

t = asyncio.create_task(task2())
await asyncio.sleep(1)
t.cancel()


...то футура перейдёт в состояние cancelled, и все остальные экземпляры task2 тоже
отменятся, что скорее всего является нежелательным поведением.

Если делать для каждого ожидающего свою отдельную футуру, то эти отдельные футуры
могут без проблем отменяться, не мешая всем остальным ожидающим. Поэтому в asyncio.Event
и нужен self._waiters, хранящий по футуре для каждого ожидающего.

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

ASYNCIO.Отличие Tasks от Future

#python #asyncio

Проблема

Вот уже несколько дней не могу понять разницу между объектами Future и Task. Точнее,
не уверен что до конца понимаю что и где используется. 

Мое понимание

Из документации следует, что в asyncio существует три вида объектов курутины, задачи(Task)
и футуры(Future)(назовем их так).
Задачи - это курутины выполнение которых запланировано в eventloop.


  When a coroutine is wrapped into a Task with functions like
  asyncio.create_task() the coroutine is automatically scheduled to run
  soon


Футуры - это некий объект в котором хранится возможный результат асинхронной операции.


  A Future represents an eventual result of an asynchronous operation


т.е. для того, что бы запустить курутину на выполнение в асинхронном режиме необходимо
использовать asyncio.create_task, но зачем тогда нам ensure_future?, который фигурирует
во всех примерах кода по asyncio. На англоязычном stackoverflow нашел смежный вопрос,
но как мне показалось нормального ответа там и нет. Ниже представлен код в котором
я использую различные варианты запуска курутин - результат всегда одинаков:

import asyncio
import time

async def fetch(n):
    await asyncio.sleep(n)
    print(time.time())
    return n

loop = asyncio.get_event_loop()
# coros = [asyncio.ensure_future(fetch(i)) for i in range(2)]
# coros = [fetch(i) for i in range(3)]
coros = [loop.create_task(fetch(i)) for i in range(2)]
results = loop.run_until_complete(asyncio.gather(*coros))
print(results)


С вариантом 2 и 3 понятно, так как в соответствии с документацией gather преобразует
курутины в задачи, а вот зачем тогда вообще используют ensure_future. Так же в документации
сказано, что лучше использовать create_task.


  See also the create_task() function which is the preferred way for
  creating new Tasks.


Если подытожить, то хотелось бы услышать простое объяснение что такое футура, чем
она отличается от задачи и где используется.

p.s. Пока писал вопрос нашел вот такую статью, но все равно не понятно) и еще один
смежный вопрос
    

Ответы

Ответ 1

Не совсем понятно, что именно непонятно, поэтому я попробую упростить ответ максимально,
прибегнув при этом к грубым допущениям. Да простят меня коллеги.

Сопрограмма (coroutine) — результат вызова асинхронной функции, представляющий собой
выполнение этой функции, способное приостанавливаться. Так как в общем случае невозможно
определить сколько раз и на какое время выполнение будет приостановлено, невозможно
и сказать когда оно будет закончено. Ваш код может либо ждать завершения выполнения
сопрограммы с помощью оператора await, либо поручить ожидание циклу событий и продолжить
свой выполнение. 

В первом случае

async def callee():
    print('Hello')

async def caller():
    await callee()
    print('World')


выполнение caller приостановится до выполнения callee. В этот момент какие-то другие
операции в каких-то других сопрограммах могут продолжаться, но caller будет ждать там,
где выполнил await.

Во втором случае

async def callee():
    print('Hello')

async def caller():
    asyncio.create_task(callee())
    print('World')


caller сразу же продолжит свою работу. Строка "World" будет выведена раньше, чем
"Hello". Здесь мы видим, что caller поставил циклу событий задачу выполнить сопрограмму
callee.

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

Футура (Future) - будущий результат выполнения сопрограммы. Метод ensure_future поручает
циклу событий выполнить сопрограмму и сразу же, в момент вызова, возвращает футуру,
в которой будет значение, но неизвестно когда. Вызывающая сторона может подождать выполнения
футуры так же, как ожидало саму сопрограмму

async def callee():
    return 'Hello'

async def caller():
    loop = asyncio.get_event_loop()
    future = loop.ensure_future(callee())
    result = await future
    print(result + ' World')


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

async def caller():
    loop = asyncio.get_event_loop()
    future = loop.ensure_future(callee())
    while not future.done():
        # Какие-нибудь циклические дела
    print(future.result() + ' World')


Или установить на футуру колбэк

async def caller():
    loop = asyncio.get_event_loop()
    future = loop.ensure_future(callee())
    future.add_done_callback(lambda f: print(f.result() + ' World'))
    # какие-нибудь другие важные дела


Или может собрать их в список и ждать все. Или не все, а только ту, которая будет
выполнена первой, а остальные проигнорировать. Или передать футуру другой сопрограмме,
а самой заняться каким-нибудь другим делом. В общем, это "очень полезный горшок, куда
можно класть какие хочешь вещи".

Осталось только разобраться, чем отличаются футуры от задач. Ничем не отличаются,
по большому счёту. Класс Task - это наследник класса Future. А существенная разница
между asyncio.create_task() и loop.ensure_future() только в том, что первой не было
до Python 3.7.

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

asyncio: корректная остановка не своих сопрограмм

#python #websocket #python_3x #asyncio

Я опять с той же проблемой. В предыдущем вопросе все таски были под моим контролем,
и я мог их вручную закрывать. Но вот беда: мне тут потребовались вебсокеты. В примерах
в документации к одноимённому модулю никакого корректного закрытия не производится:

start_server = websockets.serve(hello, 'localhost', 8765)
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()


Я поигрался с примерами, и да, при Ctrl+C ругается, если есть подключенный клиент:

$ python websocket_server.py 
^C
KeyboardInterrupt
Exception ignored in: Task was destroyed but it is pending!
task:  
  wait_for=
  cb=[_wait.._on_completion() at asyncio/tasks.py:399]>
Task was destroyed but it is pending!
task:  
  wait_for=
  cb=[_wait.._on_completion() at asyncio/tasks.py:399]>
Task was destroyed but it is pending!
task:  
  wait_for=>


Так как корректного способа завершения в документации не представлено и быстрое тыкание
наличия всяких server.close() ни к какому продуктивному результату не привело, снова
появляется вопрос о корректном завершении приложения, если таски мной не контролируются.

Это просто эта библиотека кривая? Или это характерно для любых asyncio-библиотек
и я чего-то не понимаю в самой сути asyncio и кто-нибудь ткнёт меня а какую-нибудь
матчасть по этому поводу?

Как всё-таки корректно завершать всё это дело в общем случае?



С общим случаем принцип понятен, но websockets закрываться никак не хочет. Минимальный
пример для повторения (на основе ответа @jfs):

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import asyncio
import websockets


@asyncio.coroutine
def echo(ws, path):
    print('client started')
    while True:
        data = yield from ws.recv()
        if data is None:
            break
        yield from ws.send(data)
    print('client finished')


loop = asyncio.get_event_loop()
start_server = websockets.serve(echo, '127.0.0.1', 8888, loop=loop)
server = loop.run_until_complete(start_server)

print('Listen')
try:
    server = loop.run_forever()
except KeyboardInterrupt:
    pass

server.close()
loop.run_until_complete(server.wait_closed())
loop.close()
print('Finished')


Достаточно просто подключить к нему любой клиент (хоть из браузера — ws = new WebSocket('ws://127.0.0.1:8888/');)
и нажать Ctrl+C — отпечатается приведённая ранее ошибка (правда, уже без KeyboardInterrupt).
    

Ответы

Ответ 1

asyncio реализует кооперативную многозадачность. Это значит, что websockets библиотека
должна кооперировать (предоставлять возможность чисто завершить соединения). Был открыт
баг на websockets Review cancellation management.

Даже для обычных (preemptive) потоков не существует общего решения, которое бы корректно
остановило бы потоки вне зависимости от кода, который они исполняют (Java выучила этот
урок тяжёлым путём). Безопасным для остановки ресурсом является процесс операционной
системы, если нас не волнует потеря данных из-за неочищенного файлового буфера, уведомления
другой стороны о прекращении сетевого соединения, завершение транзакции базы данных
итд. Если волнует, то процесс также обязан кооперировать, если мы хотим его завершить
преждевременно.

Чтобы обработать Ctrl+C в Питоне, как обычно нужно поймать KeyboardInterrupt или
определить свой обработчик для SIGINT сигнала. Пример echo сервера в asyncio документации
показывает как это можно сделать (KeyboardInterrupt + wait_closed()):

import asyncio

@asyncio.coroutine
def handle_echo(reader, writer):
    data = yield from reader.read(100)
    message = data.decode()
    addr = writer.get_extra_info('peername')
    print("Received %r from %r" % (message, addr))

    print("Send: %r" % message)
    writer.write(data)
    yield from writer.drain()

    print("Close the client socket")
    writer.close()

loop = asyncio.get_event_loop()
coro = asyncio.start_server(handle_echo, '127.0.0.1', 8888, loop=loop)
server = loop.run_until_complete(coro)

# Serve requests until CTRL+c is pressed
print('Serving on {}'.format(server.sockets[0].getsockname()))
try:
    loop.run_forever()
except KeyboardInterrupt:
    pass

# Close the server
server.close()
loop.run_until_complete(server.wait_closed())
loop.close()


В gevent (кооперативная многозадачность) также необходимо KeyboardInterrtupt ловить.

Документация websockets.server.serve() явно упоминает:


  serve() yields a Server which provides a close() method and a wait_closed() coroutine
to stop serving requests.


то есть для websocket сервера можно точно такой же код использовать для корректной
остановки:

try:
    loop.run_forever()
except KeyboardInterrupt:
    pass

# Close the server
server.close()
loop.run_until_complete(server.wait_closed())
loop.close()

Python. Asyncio. Асинхронные чтение и запись файла

#python #python_3x #асинхронность #asyncio

Как с помощью asyncio реализовать чтение и запись файла ? 
    

Ответы

Ответ 1

Асинхронно никак, но можно в executor'е.

import asyncio

def read_file(file_name):
    return open(file_name).read()

async def main():
    loop = asyncio.get_event_loop()
    data = await loop.run_in_executor(None, read_file, 'data.txt')
    print(data)

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

Корутины выполняются в одном потоке?

#python_3x #asyncio

Короткий вопрос по asincio: Корутины выполняются в одном потоке?

Всегда разные треды, всегда один тред или потоки выделяются по необходимости? Зависит
ли это от реализации loop?

Этот вопрос связан с использованием threading.local.
    

Ответы

Ответ 1

Да.:


  An event loop runs in a thread and executes all callbacks and tasks in the same thread.


asyncio.get_event_loop() возвращает цикл событий соответствующий текущему контексту.
По умолчанию это текущий поток, то есть один цикл на поток. При вызове из coroutine,
возвращается цикл, который эту coroutine выполняет:


  When called from a coroutine or a callback (e.g. scheduled with call_soon
  or similar API), this function will always return the running event loop.


Большинство методов и классов в asyncio не являются потокобезопасными. Если хотите
из другого потока запустить задачу в цикле, то можно использовать asyncio.run_coroutine_threadsafe().

Даже в одном потоке могут быть проблемы с изменением нелокального состояния при возвращении
контроля циклу событий (await). См., PEP 550 -- Execution Context. В Питоне 3.7 доступен
contextvars модуль, реализующий PEP 567 -- Context Variables.

Как асинхронно работать с 2мя сокетами?

#python #python_3x #async_await #asyncio

В питоне хочу сделать не блокирующую работу с 2мя сокетами. Для упрощения примера
представим Proxy. 

Пытаюсь на asincio реализовать.

Клиент_1 устанавливает соединение с сервером - получаю reader и writer. Сервер устанавливает
новое соединение с внутренним процессом через файловый сокет (в данном проекте, но
может понадобится и другие типы соединения).

Процесс может создавать запросы клиенту, а клиент к процессу.

Очень давно на втором питоне делал несколько тредов которые общались через multiprocessing.Queue.
Эти схемы надо бы исключить потому как передача большого количества данных займет много
оперативной памяти.

С клиента читаю 4 байта заголовка, и по размеру тело сообщения. Тело обрабатывается
и отправляется к процессу.

С процесса читаю по мере поступления отдельные, их также надо перепаковать.

Запускаю читалки в циклах. Та что с клиента отрабатывает одно сообщение, а вторая
продолжает дергать процесс. Такое чувство что второй блокирует первый и вставляет awaitы
в LIFO, не давая первому читать сокет клиента.
    

Ответы

Ответ 1

Давайте попробуем реализовать асинхронность по классике =) Асинхронность в отличии
от синхронности заключается в том, что для каждого соединения мы должны создать контекст:
некую структуру данных, которая будет хранить состояние соединения, информацию о том,
что сейчас происходит на этом соединении. Действия же с соединением (чтение, запись
и т.д.) происходят только тогда, когда соединение к этому готово. Поэтому, ни одно
действие не вызывает ожидания. Ожидания в таком дизайне становятся недопустимы, так
как все обработчики будут ждать одного. 

Когда-то давно мы в универе на сетевом программировании постигли дзен и обрели знание
о методе select. Метод select предназначен для опроса блокирующих потоковых классов:
стандартный ввод (stdin), запуск приложения с ожидание результата (popen), получение
данных с сокета (socket). В Python она тоже есть и выглядит следующим образом:

readst,writest,errorst = select(rlist, wlist, xlist[ , timeout])


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

#!/usr/bin/python3

import select
import socket
import sys
import os
import fcntl
import logging

USAGE = "usage: python selectproxy.py proxyhost:proxyport desthost:destport"

class ProxyConnection(object):

    # определим максимальный размер буфера
    MAX_BUFFER_SIZE = 1024

    # ProxyConnection - класс, который пересылает данные между клиентским и целевым
сокетом

    def __init__(self,proxyserver,listensock,servaddr):
        self.proxyserver = proxyserver
        self.servaddr = servaddr    # адрес сервера

        # открываем сокеты
        self.clisock, self.cliaddr = listensock.accept() # ждем подключения клиента
        self.clisock.setblocking(0)             
        self.servsock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # сокет сервера
        self.servsock.setblocking(0)

        # буферы для данных, полученных из сокета
        self.buffers = { self.clisock:bytes(), self.servsock:bytes() }           

        self.connected = False      # подключен ли серверный сокет?

        # регистрируем сокеты и разрешать операции чтения
        self.proxyserver.registerSocket(self.clisock,self)
        self.proxyserver.registerSocket(self.servsock,self)
        self.proxyserver.activateRead(self.clisock)
        self.proxyserver.activateRead(self.servsock)

    # метод, возвращающий сокет на «другом конце» соединения
    def other(self,socket):
        if socket == self.clisock:
            return self.servsock
        else:
            return self.clisock

    # подключаемся к серверу
    def connect(self):
        # мы должны использовать метод connect_ex, потому что соединение является
асинхронным и не будет выполнено немедленно
        self.servsock.connect_ex(self.servaddr)

    # метод, читающий данные из сокета
    def readfrom(self,s):
        if s == self.servsock and not self.connected:
            self.proxyserver.connection_count += 1
            logging.getLogger("selectproxy") \
                .info("соединение с установлено %s, количество текущих соединений
%d"%(str(self.cliaddr),self.proxyserver.connection_count))
            self.connected = True
            return

        # читаем из сокета
        capacity = ProxyConnection.MAX_BUFFER_SIZE - len(self.buffers[s])

        try:
            data = s.recv(capacity)
        except Exception as ex:
            data = b""

        # если чтение не удалось, закрываем сокет (это происходит, когда клиент или
сервер закрывает соединение)
        if len(data) == 0:
            self.close()
            return

        self.buffers[s] += data
        self.proxyserver.activateWrite(self.other(s))

        # перестаем читать, если буфер полон
        capacity -= len(data)
        if capacity <= 0:
            self.proxyserver.deactivateRead(s)


    # метод, который пишет в сокет
    def writeto(self,s):
        # получаем буфер, содержащий данные для чтения
        buf = self.buffers[self.other(s)]

        try:

        # пишем его в сокет
            written = s.send(buf)
        except Exception as ex:
            self.close()
            return

        # удаляем записанные данные из буфера
        buf = buf[written:]
        self.buffers[self.other(s)] = buf

        if len(buf) == 0:
            self.proxyserver.deactivateWrite(s)  
        # разрешаем чтение, если все записали
        if written:
            self.proxyserver.activateRead(self.other(s))


    # метод, закрывающий соединение
    def close(self):
        for sock in [self.clisock,self.servsock]:
            if sock:
                self.proxyserver.deactivateRead(sock)
                self.proxyserver.deactivateWrite(sock)       
                sock.close()
                self.proxyserver.unregisterSocket(sock,self)

        self.proxyserver.connection_count -= 1
        logging.getLogger("selectproxy") \
                .info("соединение с %s разорвано, количество текущих соединений %d"%(self.cliaddr,self.proxyserver.connection_count))


class ProxyServer(object):

    def __init__(self,host,port,serverhost,serverport):
        self.address = (host,port)
        self.server = (serverhost,serverport)
        self.listensock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.listensock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.listensock.bind(self.address)
        self.listensock.listen(5)
        self.connections = {}               # словарь сокет <-> ProxyConnection
        self.readsockets = []               
        self.writesockets = []              
        self.allsockets = [self.listensock] 
        self.connection_count = 0           # количество активных соединений


    def run(self):
        loop = 0
        while True:
            # Блокировать до тех пор, пока в одном из сокетов не будет активности,
тайм-аут каждые 60 секунд по умолчанию
            r, w, e = select.select(
                            [self.listensock]+self.readsockets, 
                            self.writesockets, 
                            self.allsockets,                            
                            60)
            loop += 1
            # обрабатываем любые чтения           
            for s in r:
                if s is self.listensock:
                    self.open()
                else:
                    if s in self.connections:
                        self.connections[s].readfrom(s)
            # обрабатываем запись
            for s in w:
                if s in self.connections:
                    self.connections[s].writeto(s)
            # обрабатываем ошибки
            for s in e:
                if s in self.connections:
                    self.connections[s].close()

        self.sock.close()
        self.sock = None

    def activateRead(self,sock):
        if not sock in self.readsockets:
            self.readsockets.append(sock)

    def deactivateRead(self,sock):
        if sock in self.readsockets:
            self.readsockets.remove(sock)

    def activateWrite(self,sock):
        if not sock in self.writesockets:
            self.writesockets.append(sock)

    def deactivateWrite(self,sock):
        if sock in self.writesockets:
            self.writesockets.remove(sock)

    def registerSocket(self,sock,conn):
        self.connections[sock] = conn
        self.allsockets.append(sock)

    def unregisterSocket(self,sock,conn):
        del self.connections[sock]
        self.allsockets.remove(sock)

    def open(self):    
        conn = ProxyConnection(self,self.listensock,self.server)
        conn.connect()

if __name__ == '__main__':
    try:
        proxy = sys.argv[1].split(":")
        dest = sys.argv[2].split(":")
        proxyhost = proxy[0]
        proxyport = int(proxy[1])
        serverhost = dest[0]
        serverport = int(dest[1])
    except:    
        print(USAGE)
        sys.exit(-1)

    logger = logging.getLogger('selectproxy')
    logger.setLevel(logging.INFO)
    hdlr = logging.StreamHandler()
    hdlr.setLevel(logging.INFO)
    hdlr.setFormatter(logging.Formatter('%(asctime)s - %(name)s - %(levelname)s -
%(message)s'))
    logger.addHandler(hdlr)

    server = ProxyServer(proxyhost,proxyport,serverhost,serverport)
    server.run()

asyncio: корректная остановка не своих сопрограмм

Я опять с той же проблемой. В предыдущем вопросе все таски были под моим контролем, и я мог их вручную закрывать. Но вот беда: мне тут потребовались вебсокеты. В примерах в документации к одноимённому модулю никакого корректного закрытия не производится:
start_server = websockets.serve(hello, 'localhost', 8765) asyncio.get_event_loop().run_until_complete(start_server) asyncio.get_event_loop().run_forever()
Я поигрался с примерами, и да, при Ctrl+C ругается, если есть подключенный клиент:
$ python websocket_server.py ^C KeyboardInterrupt Exception ignored in: Task was destroyed but it is pending! task: wait_for= cb=[_wait.._on_completion() at asyncio/tasks.py:399]> Task was destroyed but it is pending! task: wait_for= cb=[_wait.._on_completion() at asyncio/tasks.py:399]> Task was destroyed but it is pending! task: wait_for=>
Так как корректного способа завершения в документации не представлено и быстрое тыкание наличия всяких server.close() ни к какому продуктивному результату не привело, снова появляется вопрос о корректном завершении приложения, если таски мной не контролируются.
Это просто эта библиотека кривая? Или это характерно для любых asyncio-библиотек и я чего-то не понимаю в самой сути asyncio и кто-нибудь ткнёт меня а какую-нибудь матчасть по этому поводу?
Как всё-таки корректно завершать всё это дело в общем случае?

С общим случаем принцип понятен, но websockets закрываться никак не хочет. Минимальный пример для повторения (на основе ответа @jfs):
#!/usr/bin/env python3 # -*- coding: utf-8 -*-
import asyncio import websockets
@asyncio.coroutine def echo(ws, path): print('client started') while True: data = yield from ws.recv() if data is None: break yield from ws.send(data) print('client finished')
loop = asyncio.get_event_loop() start_server = websockets.serve(echo, '127.0.0.1', 8888, loop=loop) server = loop.run_until_complete(start_server)
print('Listen') try: server = loop.run_forever() except KeyboardInterrupt: pass
server.close() loop.run_until_complete(server.wait_closed()) loop.close() print('Finished')
Достаточно просто подключить к нему любой клиент (хоть из браузера — ws = new WebSocket('ws://127.0.0.1:8888/');) и нажать Ctrl+C — отпечатается приведённая ранее ошибка (правда, уже без KeyboardInterrupt).

Ответ

asyncio реализует кооперативную многозадачность. Это значит, что websockets библиотека должна кооперировать (предоставлять возможность чисто завершить соединения). Был открыт баг на websockets Review cancellation management
Даже для обычных (preemptive) потоков не существует общего решения, которое бы корректно остановило бы потоки вне зависимости от кода, который они исполняют (Java выучила этот урок тяжёлым путём). Безопасным для остановки ресурсом является процесс операционной системы, если нас не волнует потеря данных из-за неочищенного файлового буфера, уведомления другой стороны о прекращении сетевого соединения, завершение транзакции базы данных итд. Если волнует, то процесс также обязан кооперировать, если мы хотим его завершить преждевременно.
Чтобы обработать Ctrl+C в Питоне, как обычно нужно поймать KeyboardInterrupt или определить свой обработчик для SIGINT сигнала. Пример echo сервера в asyncio документации показывает как это можно сделать (KeyboardInterrupt + wait_closed())
import asyncio
@asyncio.coroutine def handle_echo(reader, writer): data = yield from reader.read(100) message = data.decode() addr = writer.get_extra_info('peername') print("Received %r from %r" % (message, addr))
print("Send: %r" % message) writer.write(data) yield from writer.drain()
print("Close the client socket") writer.close()
loop = asyncio.get_event_loop() coro = asyncio.start_server(handle_echo, '127.0.0.1', 8888, loop=loop) server = loop.run_until_complete(coro)
# Serve requests until CTRL+c is pressed print('Serving on {}'.format(server.sockets[0].getsockname())) try: loop.run_forever() except KeyboardInterrupt: pass
# Close the server server.close() loop.run_until_complete(server.wait_closed()) loop.close()
В gevent (кооперативная многозадачность) также необходимо KeyboardInterrtupt ловить.
Документация websockets.server.serve() явно упоминает:
serve() yields a Server which provides a close() method and a wait_closed() coroutine to stop serving requests.
то есть для websocket сервера можно точно такой же код использовать для корректной остановки:
try: loop.run_forever() except KeyboardInterrupt: pass
# Close the server server.close() loop.run_until_complete(server.wait_closed()) loop.close()

Корутины выполняются в одном потоке?

Короткий вопрос по asincio: Корутины выполняются в одном потоке?
Всегда разные треды, всегда один тред или потоки выделяются по необходимости? Зависит ли это от реализации loop?
Этот вопрос связан с использованием threading.local

Ответ

Да.
An event loop runs in a thread and executes all callbacks and tasks in the same thread.
asyncio.get_event_loop() возвращает цикл событий соответствующий текущему контексту. По умолчанию это текущий поток, то есть один цикл на поток. При вызове из coroutine, возвращается цикл, который эту coroutine выполняет
When called from a coroutine or a callback (e.g. scheduled with call_soon or similar API), this function will always return the running event loop.
Большинство методов и классов в asyncio не являются потокобезопасными. Если хотите из другого потока запустить задачу в цикле, то можно использовать asyncio.run_coroutine_threadsafe()
Даже в одном потоке могут быть проблемы с изменением нелокального состояния при возвращении контроля циклу событий (await). См., PEP 550 -- Execution Context. В Питоне 3.7 доступен contextvars модуль, реализующий PEP 567 -- Context Variables