Параллельный запуск функции по расписанию с разными параметрами

Как на python >3.8 реализовать параллельный запуск функций по расписанию, но с разными параметрами и чтобы работа скрипта не была привязана к версии операционой системы?

Ниже приведен пример конфиг файла. Скрип на python будет считывать конфиг, а далее в зависимости от заданной в конфиге периодичности должен запускать функцию fnc с заданным параметром в параллельном режиме. Для примера если пользователь задал в конфиге задал параметры 1m, 1h, 1d то в 00:00:00 каждого дня должно параллельно выполниться функции с параметрами fnc(timeframe = 1m), fnc(timeframe = 1h), fnc(timeframe = 1d). Как реализовать заданный функционал и максимально оптимально и правильно?

ini file

1m:fnc(timeframe = 1m)
5m:fnc(timeframe = 5m)
1h:fnc(timeframe = 1h)
4h:fnc(timeframe = 4h)
1d:fnc(timeframe = 1d)
1w:fnc(timeframe = 1w)

Ответы (2 шт):

Автор решения: eri

Можно, к примеру, опереться на монотонный таймер asyncio и concurrent.futures.Executor для распаралеливания.

import asyncio
import datetime
import time
import re
import concurrent.futures

tab = """
1m:fnc(timeframe = 1m)
5m:fnc(timeframe = 5m)
1h:fnc(timeframe = 1h)
4h:fnc(timeframe = 4h)
1d:fnc(timeframe = 1d)
1w:fnc(timeframe = 1w)
"""

TIMEZONE = 3*60
ms = {
  'm':1,
  'h':60,
  'd':24*60,
  'w':7*24*60
}


def fnc(**args):
    print(datetime.datetime.now(), args)


def dispatcher(loop, pool, crontab):
    time_seconds = time.time()
    delay_seconds = (time_seconds//60+1) * 60 - time_seconds
    loop.call_later(delay_seconds, dispatcher, loop, pool, crontab)
    minute = time_seconds // 60 + TIMEZONE
    for interval, callback, name, value in crontab:
        if minute % interval == 0:
            pool.submit(callback, **{name: value})


def get_crontab():
    reg = re.compile("([0-9][mhdw]*):([a-z]*)\(([a-z][a-z0-9]*).*([0-9][mhdw]*)\)", re.M)
    crontab = [list(line) for line in reg.findall(tab)]
        
    for line in crontab:
        v,m = re.match(r'([0-9]*)([mhdw])', line[0]).groups()
        v = int(v)
        m = ms[m]
        line[0] = v * m
        line[1] = globals()[line[1]]
    return crontab


pool = concurrent.futures.ThreadPoolExecutor(max_workers=5)
loop = asyncio.get_event_loop()
dispatcher(loop, pool, get_crontab())
loop.run_forever()

На моем компютере задержка от часов получилась в пределах 0,003-0,060 секунды

get_crontab - переводит формат из примера текста в список: интервал, функция, аргумент, значение.

pool - это 5 потоков, которые почти одновременно запустят функции. Замените на ProcessPoolExecutor если нужно разгрузиться по всем ядрам.

loop - это асинхронный цикл, который в этом случае получает сигналы монотонного таймера от ядра системы.

dispatcher - в первую очередь запланирует себя к выполнению в 00 секунд следующей минуты. А потом выполнит функции из списка, которые должны выполнится в текущей минуте.

pool.submit - паралельно запускает функцию в одном из потоков пула.

Конкурентный вариант (не паралельно, но асинхронно)

import asyncio
import datetime
import time
import re
import concurrent.futures
import inspect
import functools

tab = """
1m:fnc(timeframe = 1m)
5m:fnc(timeframe = 5m)
1h:fnc(timeframe = 1h)
4h:fnc(timeframe = 4h)
1d:fnc(timeframe = 1d)
1w:fnc(timeframe = 1w)
"""

TIMEZONE = 3*60
ms = {
  'm':1,
  'h':60,
  'd':24*60,
  'w':7*24*60
}


async def fnc(**args):
    print(datetime.datetime.now(), args)


def dispatcher(loop, crontab):
    time_seconds = time.time()
    delay_seconds = (time_seconds//60+1) * 60 - time_seconds
    loop.call_later(delay_seconds, dispatcher, loop, crontab)
    minute = time_seconds // 60 + TIMEZONE
    if time_seconds % 60 > 5:
        return 
    for interval, callback, name, value in crontab:
        if minute % interval == 0:
            if inspect.iscoroutinefunction(callback):
                loop.create_task(callback(**{name: value}))
            elif inspect.isfunction(callback):
                loop.call_soon(functools.partial(callback, **{name: value}))


def get_crontab():
    reg = re.compile("([0-9][mhdw]*):([a-z]*)\(([a-z][a-z0-9]*).*([0-9][mhdw]*)\)", re.M)
    crontab = [list(m) for m in reg.findall(tab)]
        
    for line in crontab:
        v,m = re.match(r'([0-9]*)([mhdw])', line[0]).groups()
        v = int(v)
        m = ms[m]
        line[0] = v * m
        line[1] = globals()[line[1]]
    return crontab


async def main():
    dispatcher(loop, get_crontab())

loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.run_forever()
→ Ссылка
Автор решения: eri

Вариант через time.sleep и multiprocessing

import multiprocessing
import datetime
import time
import re


tab = """
1m:fnc(timeframe = 1m)
5m:fnc(timeframe = 5m)
1h:fnc(timeframe = 1h)
4h:fnc(timeframe = 4h)
1d:fnc(timeframe = 1d)
1w:fnc(timeframe = 1w)
"""

TIMEZONE = 3*60
ms = {
  'm':1,
  'h':60,
  'd':24*60,
  'w':7*24*60
}


def fnc(**args):
    print(datetime.datetime.now(), args)


def dispatcher(pool, crontab, time_seconds):
    minute = time_seconds // 60 + TIMEZONE
    for interval, callback, name, value in crontab:
        if minute % interval == 0:
            pool.apply_async(callback, kwds={name: value})


def get_crontab():
    reg = re.compile("([0-9][mhdw]*):([a-z]*)\(([a-z][a-z0-9]*).*([0-9][mhdw]*)\)", re.M)
    crontab = [list(m) for m in reg.findall(tab)]
        
    for line in crontab:
        v,m = re.match(r'([0-9]*)([mhdw])', line[0]).groups()
        v = int(v)
        m = ms[m]
        line[0] = v * m
        line[1] = globals()[line[1]]
    return crontab

pool = multiprocessing.Pool()
crontab =  get_crontab()


while True:
    time_seconds = time.time()
    dispatcher(pool, crontab, time_seconds)
    delay_seconds = (time_seconds//60+1) * 60 - time.time()
    time.sleep(delay_seconds)


pool.terminate()
pool.join()
→ Ссылка