Как создать индикатор выполнения в командной строке для процессов пула?

Question

У меня есть несколько сценариев, которые я запускаю с помощью многопроцессорного пула. Я пытаюсь создать индикатор выполнения на основе выполненных сценариев.

Я проверил { ссылка } =

, но я не могу понять, как я могу объединить сценарии, выполненные в счетчике

import os                                                                       
from multiprocessing import Pool

def run_process(process):                                                             
    os.system('python {}'.format(process))

processes = ('script1.py', 'script2.py','script3.py','script4.py')

if __name__ == "__main__":

    pool = Pool(processes=2)
    pool.map(run_process, processes)

Answer 1

Вот немного другой подход, который использует concurrent.futures.ThreadPoolExecutor вместо multiprocessing.Pool, что делает его проще и эффективнее, чем в моем другом ответе .

Примечание он использует тот же модуль print_progress_bar.py, что и в моем другом ответе.

import concurrent.futures
import os
import subprocess
import sys

from print_progress_bar import print_progress_bar
progress_bar_kwargs = dict(prefix='Progress:', suffix='Complete', length=40)

# To simplify testing just using one script multiple times.
processes = ('./mp_scripts/script1.py', './mp_scripts/script1.py',
             './mp_scripts/script1.py', './mp_scripts/script1.py')
process_count = 0


def run_process(process):
    global process_count

    subprocess.run([sys.executable,  process])
    # Update process count and progress bar when it's done.
    process_count += 1
    print_progress_bar(process_count, len(processes), **progress_bar_kwargs)


print_progress_bar(0, len(processes), **progress_bar_kwargs) # Print 0% progress.
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:

    future_to_process = {executor.submit(run_process, process): process
                            for process in processes}
    for future in concurrent.futures.as_completed(future_to_process):
        process = future_to_process[future]
        try:
            _ = future.result()
        except Exception as exc:
            print()
            print(f'{process} generated an exception: {exc}')

    print('Done')

Answer 2

Вы можете сделать это, используя pool.apply_async(), потому что он поддерживает функцию обратного вызова, которую можно использовать, чтобы узнать, когда была возвращена целевая функция.

Я использовал @ Greenstick's ответьте , чтобы отобразить индикатор выполнения, но я в основном изменил его, чтобы он соответствовал рекомендациям по кодированию PEP-8, и поместил его в отдельный модуль с именем print_progress_bar - см. ниже.

Примечание по производительности: Хотя один может использовать для этого multiprocessing.Pool - я сильно подозреваю, что код в вашем вопросе является дословной копией того, что есть в статье Как запускать параллельные процессы - выполнение это крайне неэффективно, потому что каждый процесс будет инициализировать свой собственный Python интерпретатор double количество раз, которое действительно необходимо. Сначала выполнить саму функцию run_process(), а затем снова запустить процесс сценария.

Процессы нереста требуют значительных накладных расходов. Эти издержки можно уменьшить, вместо этого запустив run_process() как отдельный поток в процессе current , который является более легким.

Переключиться на ThreadPool очень просто, просто изменить строку: from multiprocessing import Pool в from multiprocessing.pool import ThreadPool as Pool

В качестве альтернативы вы можете использовать concurrent.futures.ThreadPoolExecutor, как показано в моем другом ответе .

import os
from multiprocessing import Pool
import subprocess
import sys

from print_progress_bar import print_progress_bar
progress_bar_kwargs = dict(prefix='Progress:', suffix='Complete', length=40)

def run_process(process):
     os.system('{} {}'.format(sys.executable,  process))

def callback(_):
    """Update process count and progress bar."""
    global process_count
    process_count += 1
    print_progress_bar(process_count, len(processes), **progress_bar_kwargs)


# To simplify testing just using one script multiple times.
processes = ('./mp_scripts/script1.py', './mp_scripts/script1.py',
             './mp_scripts/script1.py', './mp_scripts/script1.py')
process_count = 0

if __name__ == '__main__':

    print_progress_bar(0, len(processes), **progress_bar_kwargs) # Print 0% progress.

    with Pool(processes=2) as pool:
        results = []
        for process in processes:
            r = pool.apply_async(run_process, (process,), {}, callback)
            results.append(r)

        while results:  # Processes still running?
            results = [r for r in results if not r.ready()]

    print('Done')

---

print_progress_bar.py:

# from https://stackoverflow.com/a/34325723/355230
def print_progress_bar(iteration, total, prefix='', suffix='', decimals=1, length=100,
                       fill='█', print_end="\r"):
    """ Print iterations progress.

        Call in a loop to create terminal progress bar
        @params:
            iteration   - Required  : current iteration (Int)
            total       - Required  : total iterations (Int)
            prefix      - Optional  : prefix string (Str)
            suffix      - Optional  : suffix string (Str)
            decimals    - Optional  : positive number of decimals in percent complete (Int)
            length      - Optional  : character length of bar (Int)
            fill        - Optional  : bar fill character (Str)
            print_end   - Optional  : end character (e.g. "\r", "\r\n") (Str)
    """
    percent = ("{0:." + str(decimals) + "f}").format(100 * (iteration / float(total)))
    filledLength = int(length * iteration // total)
    bar = fill * filledLength + '-' * (length - filledLength)
    print('\r%s |%s| %s%% %s' % (prefix, bar, percent, suffix), end=print_end, flush=True)

    if iteration == total:  # Print newline on completion.
        print(flush=True)