Перезапустите Async Coroutine, не дожидаясь завершения других сопрограмм

Question

Я практикую асинхронное программирование на Python со следующей проблемой:

Имитация нескольких людей, едящих из одной миски с заданным количеством порций еды.Каждый человек может принять x порций еды за один раз, а затем жует еду в течение y секунд (смоделировано с помощью блокирующего вызова).Человек может принимать и пережевывать пищу независимо от других людей, пока в миске есть еда.

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

Например, если у меня есть миска с 25 порциями еды и три человека, A, B и C:

• A принимает 2 порции пищи одновременно и жует в течение 3 секунд
• B принимает 3 порции пищи одновременно и жует 4 секунды
• C принимает 5 порций еды за раз и жует в течение 2 секунд

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

(t=0) Person A takes 2 servings of food, leaving 23 servings in the bowl.
(t=0) Person B takes 3 servings of food, leaving 20 servings in the bowl.
(t=0) Person C takes 5 servings of food, leaving 15 servings in the bowl.
(t=2) Person C takes 5 servings of food, leaving 10 servings in the bowl.
(t=3) Person A takes 2 servings of food, leaving 8 servings in the bowl.
(t=4) Person B takes 3 servings of food, leaving 5 servings in the bowl.
(t=4) Person C takes 5 servings of food, leaving 0 servings in the bowl.
(t=4) The bowl is empty!

(Время от временикак t=4, когда два человека готовы принять другую порцию, порядок не имеет значения) Код - моя попытка:

import asyncio
import time


class Person():
    def __init__(self, name, serving_size, time_to_eat):
        self.name = name
        self.serving_size = serving_size
        self.time_to_eat = time_to_eat

    async def eat_from(self, foodbowl):
        servings_taken = self.serving_size if foodbowl.qty >= self.serving_size else foodbowl.qty
        foodbowl.qty -= servings_taken
        t = round(time.time() - foodbowl.start_time)
        print("(t={}) Person {} picks up {} servings of food, leaving {} servings in the bowl.".format(t, self.name, servings_taken, foodbowl.qty))
        await asyncio.sleep(self.time_to_eat)
        return servings_taken


class FoodBowl():
    def __init__(self, qty):
        self.qty = qty

    async def assign_eaters(self, eaters):
        self.start_time = time.time()
        while self.qty > 0:
            await asyncio.gather(*[eater.eat_from(self) for eater in eaters])
        t = round(time.time() - self.start_time)
        print("The bowl is empty!")


bowl = FoodBowl(25)
person_1 = Person("A", 2, 3)
person_2 = Person("B", 3, 4)
person_3 = Person("C", 5, 2)
asyncio.run(bowl.assign_eaters([person_1, person_2, person_3]))

Однако моя попытка приводит к следующему поведению:

(t=0) Person A picks up 2 servings of food, leaving 23 servings in the bowl.
(t=0) Person B picks up 3 servings of food, leaving 20 servings in the bowl.
(t=0) Person C picks up 5 servings of food, leaving 15 servings in the bowl.
(t=4) Person A picks up 2 servings of food, leaving 13 servings in the bowl.
(t=4) Person B picks up 3 servings of food, leaving 10 servings in the bowl.
(t=4) Person C picks up 5 servings of food, leaving 5 servings in the bowl.
(t=8) Person A picks up 2 servings of food, leaving 3 servings in the bowl.
(t=8) Person B picks up 3 servings of food, leaving 0 servings in the bowl.
(t=8) Person C picks up 0 servings of food, leaving 0 servings in the bowl.
The bowl is empty!

Видно, что каждый человек ждет, пока все закончат есть, прежде чем снова дотянуться до чаши.Глядя на мой код, я знаю, что это потому, что я ждал asyncio.gather() функций приема пищи, и поэтому он будет ждать, пока все три человека закончат есть, прежде чем кто-нибудь сможет начать есть снова.

Я знаю, что это неправильно, но я не знаю, что я могу использовать в библиотеке asyncio, чтобы решить эту проблему.Я думаю о том, что сопрограмма eat_from автоматически перезапускается, пока в миске все еще есть еда.Как мне это сделать, или есть лучший подход к этой проблеме?

Answer 1

Я знаю, что [ждать, пока все три человека закончат есть, прежде чем кто-то сможет снова начать есть], неправильно, но я не знаю, что я могу использовать в библиотеке asyncio, чтобы решить эту проблему.

Вы можете использовать wait(return_when=asyncio.FIRST_COMPLETED), чтобы дождаться окончания действия любого едоков, вместо того, чтобы ждать всех из них, как это делает текущий код. Всякий раз, когда едок заканчивает есть, порождает новую сопрограмму для того же едока, эффективно «перезапуская» ее. Для этого требуется ссылка из задачи, возвращаемой едоком в wait; такая ссылка может быть легко прикреплена к объекту Task. Код может выглядеть так:

async def assign_eaters(self, eaters):
    self.start_time = time.time()
    # create the initial tasks...
    pending = [asyncio.create_task(eater.eat_from(self))
               for eater in eaters]
    # ...and store references to their respective eaters
    for t, eater in zip(pending, eaters):
        t.eater = eater

    while True:
        done, pending = await asyncio.wait(
            pending, return_when=asyncio.FIRST_COMPLETED)
        if self.qty == 0:
            break
        for t in done:
            # re-create the coroutines that have finished
            new = asyncio.create_task(t.eater.eat_from(self))
            new.eater = t.eater
            pending.add(new)
    t = round(time.time() - self.start_time)
    print("The bowl is empty!")

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

async def eat_from(self, foodbowl):
    while foodbowl.qty:
        servings_taken = self.serving_size \
            if foodbowl.qty >= self.serving_size else foodbowl.qty
        foodbowl.qty -= servings_taken
        t = round(time.time() - foodbowl.start_time)
        print("(t={}) Person {} picks up {} servings of food, "
              "leaving {} servings in the bowl."
              .format(t, self.name, servings_taken, foodbowl.qty))
        await asyncio.sleep(self.time_to_eat)

assign_eaters больше не нуждается в цикле и возвращается к использованию простого gather:

async def assign_eaters(self, eaters):
    self.start_time = time.time()
    await asyncio.gather(*[eater.eat_from(self) for eater in eaters])
    t = round(time.time() - self.start_time)
    print("The bowl is empty!")

Этот более простой код снова приводит к ожидаемому результату. Единственный «недостаток» заключается в том, что изменение потребовало инвертирования управления: чаша больше не управляет процессом приема пищи, теперь это делается автономно каждым едоком, причем чаша пассивно ожидает их завершения. Однако, если взглянуть на постановку проблемы, это кажется не только приемлемым, но, возможно, даже искомым решением. Утверждается, что функция миски для еды должна заставить людей «начать есть из миски, пока миска не опустеет». «Начинай есть» подразумевает, что миска просто инициирует процесс, и что каждый человек ест самостоятельно - так работает вторая версия.