Задачи python asyncio.create_task завершаются раньше, чем ожидалось?

Проблема в том, что цикл в main не ожидает завершения задач, из-за чего задачи не завершаются. Используйте asyncio.gather() для запуска и дождитесь выполнения всех сопрограмм.

import asyncio


async def myfunc(i):
    print("hello", i)
    await asyncio.sleep(i)
    print("world", i)


async def main():
    await asyncio.gather(myfunc(2), myfunc(1))


asyncio.run(main())

Логика, которую вы описываете в комментариях, более сложная и нет функции, которая ее реализует, поэтому вам нужно спроектировать логику, в этом случае есть 2 условия, которые должны выполниться для завершения приложения:

• Все задачи должны быть запущены.
• Не должно быть активной задачи.

Учитывая это, я использую Future для создания флага, который указывает на это, также используйте add_done_callback, чтобы уведомить меня о завершении задания.

import asyncio
from collections import deque
import random


async def f(identifier, seconds):
    print(f"Starting task: {identifier}, seconds: {seconds}s")
    await asyncio.sleep(seconds)
    print(f"Finished task: {identifier}")
    return identifier


T = 3


class Manager:
    def __init__(self, q):
        self._q = q
        self._future = asyncio.get_event_loop().create_future()
        self._active_tasks = set()
        self._status = False

    @property
    def q(self):
        return self._q

    def launch_task(self):
        try:
            identifier = self.q.pop()
        except IndexError:
            return False
        else:
            seconds = random.uniform(T - 1, T + 1)
            task = asyncio.create_task(f(identifier, seconds))
            self._active_tasks.add(identifier)
            task.add_done_callback(self._finished_callback)
            return True

    async def work(self):
        self._status = True
        while self.launch_task():
            await asyncio.sleep(T)
        self._status = False
        await self._future

    def _finished_callback(self, c):
        self._active_tasks.remove(c.result())
        if not self._active_tasks and not self._status:
            self._future.set_result(None)


async def main():
    identifiers = deque(range(10))
    manager = Manager(identifiers)
    await manager.work()


if __name__ == "__main__":
    asyncio.run(main())

Нашел гораздо более простое решение, чем то, которое предоставил @eyllanesc здесь. Оказывается, есть функция, которая это реализует.