0

Задача заключается в том, что нужно указанное число(count) раз вызвать функцию(doubler), но одновременно может быть запущено лишь указанное число процессов(process)

Имеется следующий код:

def doubler(argument):
    pass

def start_processes_and_wait(process):
    procs = []
    for i in range(process):
        proc = Process(target=doubler, args=(i,))
        procs.append(proc)
        proc.start()

    for i in procs:
        i.join()

def run(count, process):

    count -= process
    while count > 0:
        start_processes_and_wait(process)    
        count -= process

    start_processes_and_wait(process + count)

Но у этого кода есть существенный минус. Он сначала запускает одну партию процессов, ждёт пока они все завершатся и лишь потом запускает следующую партию.

В функции doubler может получиться так, что один из процессов просто зависнет на очень долгий срок, и в итоге из-за этого остановится вся партия процессов.

3
  • Есть же библиотеки, которые управляют пулом потоков
    – danilshik
    29 сен 2018 в 12:12
  • Рассмотрите вариант, в котором один из процессом будет "ядром" и будет запускать и проверять состояние активных потоков. 29 сен 2018 в 12:29
  • Спасибо ребят, вы очень помогли) 29 сен 2018 в 17:03

2 ответа 2

1

Пример с потоками:

from concurrent.futures import ThreadPoolExecutor
count = 6000 #Количество раз
count_thread = 40 #Количество потоков

arguments = []

def doubler(argument):
    print (argument)

def start():
    with ThreadPoolExecutor(count_thread) as executor:
        for _ in executor.map(doubler, arguments):
            pass

if __name__ == "__main__":
    # Создаем массив параметров
    for i in range(count):
        arguments.append(i)
    start()

Тот же самый алгоритм, только с пулом процессов (работает медленнее, требует больше ресурсов для создания процессов нежели потоков. Оптимальное количество процессов равно количеству ядер вашего процессов, при указании больше, будут большие издержки на переключении между процессами)

from concurrent.futures import ProcessPoolExecutor
    count = 6000 #Количество раз
    count_thread = 40 #Количество процессов

    arguments = []

    def doubler(argument):
        print (argument)

    def start():
        with ProcessPoolExecutor(count_thread) as executor:
            for _ in executor.map(doubler, arguments):
                pass

    if __name__ == "__main__":
        # Создаем массив параметров
        for i in range(count):
            arguments.append(i)
        start()

Также вы можете посмотреть библиотеки Multiprocessing и Multiprocessing.Dumpy

0

Советую посмотреть в сторону пула процессов, в частности на ProcessPoolExecutor. Суть в том что у вас будет наперед заготовление набор процессов, которые будут по мере готовности принимать задачи на выполнение:

import time
from concurrent.futures import ProcessPoolExecutor, as_completed
from random import randint


def doubler(argument):
    sleep_time = randint(1, 10)
    time.sleep(sleep_time)
    return (argument, sleep_time)


def run(count, process):
    with ProcessPoolExecutor(process) as executor:
        futures = [executor.submit(doubler, x) for x in range(count)]
        for future in as_completed(futures):
            print(future.result())

Но в конце концов все зависит от содержимого функции doubler, если там будут очень простые и не частые вычисления на ЦПУ (выполнение Python кода) без выполнения I/O операция, то может лучше использовать ThreadPoolExecutor, кок более эффективный вариант.

Но если предполагается хоть сколько-то сложные вычисления, или I/O операция (запросы к удаленному серверу например) тогда стоит использовать именно ProcessPoolExecutor или асинхронность соответственно.

ThreadPoolExecutor при росте нагрузки начинает страдать от GIL, так как код по сути выполняется внутри одного процесса по очереди многими потоками. ProcessPoolExecutor от этого не страдает, так как GIL работает в пределах одного процесса, но на маленьких объемах работы больше ресурсов уйдет на создание самого пула процессов.

Раньше ThreadPoolExecutor часто использовался для выполнения I/O операций, так-как во время их выполнения поток отпускает GIL, и позволяет выполнятся другому потоку, но сейчас более эффективно для этих целей использовать асинхронность, так как там вообще все выполняется в одном потоке, а ограничение количества одновременно выполняемых запросов можно с помощью специальных менеджеров контекста.

Ваш ответ

By clicking “Отправить ответ”, you agree to our terms of service and acknowledge you have read our privacy policy.

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