0

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

import asyncio

async def f1():
    print("First function...")
    while True:
        pass

async def f2():
    print("Second function...")
    while True:
        pass


ioloop = asyncio.get_event_loop()
tasks = [
    ioloop.create_task(f1()),
    ioloop.create_task(f2())
]
ioloop.run_until_complete(asyncio.wait(tasks))
ioloop.close()

Программа выводит только:

First function

А нужно чтобы:

First function...
Second function...

Как такое реализовать?

Дополнение: Скорее всего мой вопрос был некорректным вместо pass я имел в виду определенное действие а не ожидание. Думаю так будет понятнее:

import asyncio

async def f1():
    print("Starting first function")
    while True: #Этот код работает бесконечно
        print('First code is working...') 

async def f2():
    print("Starting second function")
    while True: #Этот код тоже работает бесконечно, но не мешает первому
        print('Second code is working too...') 


ioloop = asyncio.get_event_loop()
tasks = [
    ioloop.create_task(f1()),
    ioloop.create_task(f2())
]
ioloop.run_until_complete(asyncio.wait(tasks))
ioloop.close()

Будет выводить:

First code is working...
First code is working...
First code is working...
(До бесконечности)

А надо что-то на подобии:

First code is working...
Second code is working too...
First code is working...
Second code is working too...
(До бесконечности)

Возможно я не очень понял суть библиотеки asyncio (Возможно лучше использовать multiprocessing или что-то ещё)

11
  • Какие реальные задачи будут выполнять функции?
    – user207200
    Commented 6 июл. 2021 в 10:22
  • Вообще это для telegram-бота. Одна функция будет ждать сообщения и отправлять ответ, а вторая - отправлять сообщение в конкретное время.
    – ilapro 53
    Commented 6 июл. 2021 в 10:25
  • Документацию aiogram просматривали?
    – user207200
    Commented 6 июл. 2021 в 10:30
  • 1
    поскольку в примере внутри циклов нет await, то и переключиться на другую задачу и "дать другим поработать" он не может.
    – vitidev
    Commented 6 июл. 2021 в 18:40
  • 1
    Зависит от задачи. Из примеров не видно. А вообще await означает "я запустил что-то неблокирующее и сам блокировать не буду, подожду тут результата и дам другим поработать". Если у вас есть такой неблокирующий код (это ввод-вывод- запись в файл, работа с сетью..), то на них и можно переключить задачи. Если же неблокирующего ничего нет, то и переключения не будет - тогда придется использовать костыль await asyncio.sleep или же городить кооперативную многозадачность через генераторы.
    – vitidev
    Commented 6 июл. 2021 в 18:59

4 ответа 4

1

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

Ядро асинхронного приложения - это событийный цикл. В вашем примере, не вдаваясь в детали, ioloop = asyncio.get_event_loop() - вы его получаете, tasks = [ ioloop.create_task(f1()), ioloop.create_task(f2()) ] - создаете задачи для него, ioloop.run_until_complete(asyncio.wait(tasks)) запускаете эти задачи внутри него. Цикл событий выполняет только одну задачу, до тех пор пока не встречает неблокирующую операцию, когда задача Task выполняется с оператором await, то выполняющаяся задача приостанавливается, а цикл обработки событий выполняет следующую задачу. Для вашего примера достаточно добавить asyncio.sleep(0.1) что бы цикл событий мог передавать контекст выполнения между задачами.

import asyncio

async def f1():
    print("First function...")
    while True:  # Этот код работает бесконечно
        await asyncio.sleep(0.1) #-> на этом месте цикл событий переключит выполнение на другую задачу
        print('First code is working...')

async def f2():
    print("Second function...")
    while True:  # Этот код тоже работает бесконечно, но не мешает первому
        await asyncio.sleep(0.1)#-> на этом месте цикл событий переключит выполнение на другую задачу
        print('Second code is working too...')


ioloop = asyncio.get_event_loop()
tasks = [
    ioloop.create_task(f1()),
    ioloop.create_task(f2())
]
ioloop.run_until_complete(asyncio.wait(tasks))
ioloop.close()

Если Вы не используете асинхронные библиотеки в задачах, или другие i/o wait операции ( к примеру открытий файла и чтение из него, тут можно было использовать потоки), а распаралеливание затрагивает только cpu операции, то необходимо использовать процессы multiprocessing. Как пример:

import multiprocessing as mp
def f1():
    print("First function...")
    while True:  # Этот код работает бесконечно
        print('First code is working...')
        time.sleep(0.1)

def f2():
    print("Second function...")
    while True:  # Этот код тоже работает бесконечно, но не мешает первому
        print('Second code is working too...')
        time.sleep(0.1) # это не обязательно

def main(func: callable):
    func()

tasks = [f1, f2]
procs = [mp.Process(target=main, args=(i,)) for i in tasks] # создаем столько процессов, сколько имеем функций
for proc in procs:
    proc.start()

for proc in procs:
    proc.join()

На практике чаще имеется одна функция и много аргументов, и необходимость распаралелить выполнение одной функции. Тогда кол-во аргументов бьется на пачки и передается пачка в каждый процесс, если кол-во процессов < "пачек" то лучше использовать пул процессов (multiprocessing.Pool), так как создание процесса считается тяжелой операцией, а так у Вас заранее будет выделено N процессов в вашем пуле.

2
  • await asyncio.sleep(0) можно - тогда будет работать без задержки. и не забывайте про экзекуторы - они удобнее поднятия треда вручную
    – eri
    Commented 7 июл. 2021 в 8:23
  • А, понял. Спасибо вам огромное.
    – ilapro 53
    Commented 7 июл. 2021 в 8:32
0
import asyncio


async def f1():
    print("First function...")
    while True:
        await asyncio.sleep(4)


async def f2():
    print("Second function...")
    while True:
        await asyncio.sleep(4)


ioloop = asyncio.get_event_loop()
tasks = [
    ioloop.create_task(f1()),
    ioloop.create_task(f2())
]
ioloop.run_until_complete(asyncio.wait(tasks))
ioloop.close()

в новых версиях питон Python (≥ 3.7) лучше запускать асинхронные функции так:

async def f1():
    print("First function...")
    await asyncio.sleep(4) # Имитирование работы программы
    print(f'{f1.__name__} Done')


async def f2():
    print("Second function...")
    await asyncio.sleep(4) # Имитирование работы программы
    print(f'{f2.__name__} Done')


async def main():
    await asyncio.gather(f1(), f2())


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

pass блокирующая команда, вместо нее лучше использовать asyncio.sleep как имитирование работы функции. Для одновременного выполнения асинхронных программ нужно добавить их в asyncio.gather(func(), func()...) а затем запустить с помощью asyncio.run

2
  • Наверное я не правильно объяснил суть вопроса. Я дополнил его
    – ilapro 53
    Commented 6 июл. 2021 в 18:36
  • лучше запускать асинхронные функции так - не лучше. для задачи в 2 таска наоборот хуже)
    – eri
    Commented 7 июл. 2021 в 8:08
0

Если функция асинхронная

async def f1():
    print("Async function...")
    while True:
        await asyncio.sleep(5)

Если функция синхронная

async def f2():
    print("Blocking function...")
    while True:
        await ioloop.run_in_executor(None, time.sleep(5))

Если на функции не блокирующий опрос состояния можно сделать так - await asyncio.sleep(0) выполняется на каждом шаге лупа пока другой таск висит на слове await в конкурентном таске

async def f2():
    print("Check function...")
    while True:
        await asyncio.sleep(0)
        if driver.state == 1:
            ioloop.call_soon(cb)

await asyncio.sleep(0) можно использовать как переключатель между тасками если не беда что они будут блокироваться - например при какой-нибудь математике, когда ещё не выгодно уходить на мультипроцессинг.

Тут колбэк может снова заблокировать луп - так что смотрим выше

0

Как я понял из слов @vitdev (спасибо ему огромное) с помощью asincio параллельно можно запустить "только неблокирующий код (это ввод-вывод- запись в файл, работа с сетью..)". Поэтому asincio подойдёт не для любых задач.

P.S @asanisimov объяснил как работает Этот модуль. Всетаки asincio может помочь, просто работает по другому принципу. Спасибо большое @asanisimov

Я нашел решение данного вопроса с помощью модуля threading. Например так:

from threading import Thread

def f1():
    print("Starting first function")
    while True: #Этот код работает бесконечно
        print('First code is working...')

def f2():
    print("Starting second function")
    while True: #Этот код тоже работает бесконечно
        print('Second code is working too...')



class MyThread(Thread):
    """A threading example"""
    
    def __init__(self, name, function): # name - некое имя потока (Можно было и без этого обойтись)
                                        # function - функция, которая будет запущена параллельно основному потоку
        """Инициализация потока"""
        Thread.__init__(self)
        self.name = name
        self.function = function
    
    def run(self):
        """Запуск потока"""
        self.function()

    
def create_threads():
    """Создаем группу потоков"""
    
    name = "Thread 1"
    function = f1
    my_thread1 = MyThread(name, function)
    my_thread1.start()

    name = "Thread 2"
    function = f2
    my_thread2 = MyThread(name, function)
    my_thread2.start()
 
 
if __name__ == "__main__":
    create_threads()

Вывод у этой программы довольно страшный (Не буду вставлять), но это значит что все работает.

Можно заменить print('First code is working...') и print('Second code is working too...') на pass и убедиться, что обе функции запустились:

from threading import Thread

def f1():
    print("Starting first function")
    while True: #Этот код работает бесконечно
        pass

def f2():
    print("Starting second function")
    while True: #Этот код тоже работает бесконечно
        pass



class MyThread(Thread):
    """A threading example"""
    
    def __init__(self, name, function): # name - некое имя потока
                                        # function - функция, которая будет запущена параллельно основному потоку
        """Инициализация потока"""
        Thread.__init__(self)
        self.name = name
        self.function = function
    
    def run(self):
        """Запуск потока"""
        self.function()

    
def create_threads():
    """Создаем группу потоков"""
    
    name = "Thread 1"
    function = f1
    my_thread1 = MyThread(name, function)
    my_thread1.start()

    name = "Thread 2"
    function = f2
    my_thread2 = MyThread(name, function)
    my_thread2.start()
 
 
if __name__ == "__main__":
    create_threads()

Вывод:

Starting first function
Starting second function

Ваш ответ

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

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.