Написание EventLoop без использования asyncio - PullRequest
Новая
       55

Написание EventLoop без использования asyncio

0 голосов
/ 13 декабря 2018

Я очень хорошо знаком с Python asyncio , асинхронным программированием на Python, сопрограммами и т. Д. Я хочу иметь возможность выполнять несколько сопрограмм с моим собственным eventloop.

Мне любопытно, могу ли я написать свой собственный eventloop без импорта asyncio вообще

Ответы [ 2 ]

0 голосов
/ 29 декабря 2018

Хорошо, я нашел где-то пример (извините, не помню, где, без ссылки) и немного изменил.

eventloop и co-routins без импорта asyncio: 

import datetime
import heapq
import types
import time

class Task:
    def __init__(self, wait_until, coro):
        self.coro = coro
        self.waiting_until = wait_until

    def __eq__(self, other):
        return self.waiting_until == other.waiting_until

    def __lt__(self, other):
        return self.waiting_until < other.waiting_until

class SleepingLoop:
    def __init__(self, *coros):
        self._new = coros
        self._waiting = []

    def run_until_complete(self):
        # Start all the coroutines.
        for coro in self._new:
            wait_for = coro.send(None)
            heapq.heappush(self._waiting, Task(wait_for, coro))

        # Keep running until there is no more work to do.
        while self._waiting:
            now = datetime.datetime.now()
            # Get the coroutine with the soonest resumption time.
            task = heapq.heappop(self._waiting)
            if now < task.waiting_until:
                # We're ahead of schedule; wait until it's time to resume.
                delta = task.waiting_until - now
                time.sleep(delta.total_seconds())
                now = datetime.datetime.now()
            try:
                # It's time to resume the coroutine.
                wait_until = task.coro.send(now)
                heapq.heappush(self._waiting, Task(wait_until, task.coro))
            except StopIteration:
                # The coroutine is done.
                pass


@types.coroutine
def async_sleep(seconds):
    now = datetime.datetime.now()
    wait_until = now + datetime.timedelta(seconds=seconds)
    actual = yield wait_until

    return actual - now


async def countdown(label, total_seconds_wait, *, delay=0):
    print(label, 'waiting', delay, 'seconds before starting countdown')
    delta = await async_sleep(delay)
    print(label, 'starting after waiting', delta)
    while total_seconds_wait:
        print(label, 'T-minus', total_seconds_wait)
        waited = await async_sleep(1)
        total_seconds_wait -= 1
    print(label, 'lift-off!')


def main():
    loop = SleepingLoop(countdown('A', 5, delay=0),
                        countdown('B', 3, delay=2),
                        countdown('C', 4, delay=1))
    start = datetime.datetime.now()
    loop.run_until_complete()

    print('Total elapsed time is', datetime.datetime.now() - start)



if __name__ == '__main__':
    main()
0 голосов
/ 13 декабря 2018

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

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

Мне интересно, могу ли я написать свой собственный Eventloop без импорта asyncio вообще

Это определенно возможно - ведь asyncio - это просто библиотека, но для того, чтобы цикл обработки событий был полезным, потребуется некоторая работа.Смотрите этот превосходный доклад Дэвида Бизли, где он демонстрирует написание цикла событий перед живой аудиторией.(Не стоит откладывать на то, что Дэвид использует старый синтаксис yield from - await работает точно так же.)

...