Многопроцессорная обработка показывает несколько индикаторов выполнения

Question

Для моей программы у меня есть файл, который записывает случайные целые числа в файл.CSV.

from __future__ import absolute_import, division, print_function
from numpy.random import randint as randrange
import os, argparse, time
from tqdm import tqdm

def write_to_csv(filename, *args, newline = True):
    write_string = ''
    for arg in args:
        if type(arg) == list:
            for i in arg:
                write_string += str(i) + ','
        else:
            write_string += str(arg) + ','
    if newline:
        write_string = write_string.rstrip(',') + '\n'
    else:
        write_string = write_string.rstrip(',')
    with open(filename+'.csv', 'a') as file:
        file.write(write_string)

def move_dir(dirname, parent = False):
    if not parent:
        dirname = str(dirname)
        exists = os.path.isfile(dirname)
        try:
            os.mkdir(dirname)
            os.chdir(dirname)
        except FileExistsError:
            os.chdir(dirname)
    else:
        os.chdir("..")

def calculate_probability(odds, exitmode = False, low_cpu = 0):
    try:
        file_count = 0
        move_dir('Probability')
        move_dir(str(odds))
        d = {}
        writelist = []
        percentlist = []
        for i in tqdm(range(odds)):
            d[str(i)] = 0
            writelist.append(f'Times {i}')
            percentlist.append(f'Percent {i}')
        while True:
            if os.path.isfile(str(file_count)+'.csv'):
                file_count += 1
            else:
                break
        filename = str(file_count)
        write_to_csv(filename, 'Number', 'Value')
        rep = 500 * odds
        if rep > 10000:
            rep = 10000
        for i in tqdm(range(rep)):
            ran = randrange(odds)
            ran = int(ran)
            d[str(ran)] += 1
            if i == 999:
                write_to_csv(filename, i, ran+1, newline = False)
            else:
                write_to_csv(filename, i, ran+1)
            if low_cpu:
                time.sleep(0.01*float(low_cpu))
        writelist2 = []
        percentlist2 = []
        for i in tqdm(range(odds)):
            val = d[str(i)]
            writelist2.append(val)
            percentlist2.append(round(((val/rep)*100), 2))
        if os.path.isfile('runs.csv'):
            write_to_csv('runs', file_count, writelist2, percentlist2)
        else:
            write_to_csv('runs', 'Run #', writelist, percentlist)
            write_to_csv('runs', file_count, writelist2, percentlist2)
        if exitmode:
            exit()
    except(KeyboardInterrupt, SystemExit):
        if exitmode:
            os.remove(str(file_count)+'.csv')
            exit()
        else:
            try:
                os.system('cls')
                print('User/program interrupted, lauching shutdown mode...')
                os.remove(str(file_count)+'.csv')
                print('Finilizaing current trial...')
                os.chdir("..")
                os.chdir("..")
            except FileNotFoundError:
                exit()
            calculate_probability(odds, exitmode = True)

У меня также есть система повторения, чтобы сделать это несколько раз.

def run_tests(times, odds, low_cpu = 0, shutdown = False):
    for i in tqdm(range(times)):
        calculate_probability(odds, low_cpu = low_cpu)
        os.chdir("..")
        os.chdir("..")
    if shutdown:
        os.system('shutdown /S /F /T 0 /hybrid')

Однако, если бы я пробежал 30 трасс, это заняло бы вечность. Поэтому я решил использовать многопроцессорный модуль для ускорения процесса. Поскольку каждый прогон должен в конце записывать один и тот же файл, мне пришлось собирать данные и записывать их после завершения процессов.

def calculate_probability(odds, low_cpu = 0):
    try:
        file_count = 0
        move_dir('Probability')
        move_dir(str(odds))
        d = {}
        writelist = []
        percentlist = []
        for i in tqdm(range(odds)):
            d[str(i)] = 0
            writelist.append(f'Times {i}')
            percentlist.append(f'Percent {i}')
        while True:
            if os.path.isfile(str(file_count)+'.csv'):
                file_count += 1
            else:
                break
        filename = str(file_count)
        write_to_csv(filename, 'Number', 'Value')
        rep = 500 * odds
        if rep > 10000:
            rep = 10000
        for i in range(rep):
            ran = randrange(odds)
            ran = int(ran)
            d[str(ran)] += 1
            if i == 999:
                write_to_csv(filename, i, ran+1, newline = False)
            else:
                write_to_csv(filename, i, ran+1)
            if low_cpu:
                time.sleep(0.01*float(low_cpu))
        writelist2 = []
        percentlist2 = []
        for i in range(odds):
            val = d[str(i)]
            writelist2.append(val)
            percentlist2.append(round(((val/rep)*100), 2))
        return (writelist, percentlist, writelist2, percentlist2)
    except(KeyboardInterrupt, SystemExit):
        try:
            os.remove(str(file_count)+'.csv')
        finally:
            exit()

def worker(odds, returndict, num, low_cpu = 0):
    returndict[f'write{num}'] = calculate_probability(odds, low_cpu = low_cpu)
    os.chdir("..")
    os.chdir("..")
    os.system('cls')

def run_tests(times, odds, low_cpu = 0, shutdown = False):
    print('Starting...')
    manager = Manager()
    return_dict = manager.dict()
    job_list = []
    for i in range(times):
        p = Process(target=worker, args=(odds,return_dict,i), kwargs = {'low_cpu' : low_cpu})
        job_list.append(p)
        p.start()

    try:
        for proc in job_list:
            proc.join()
    except KeyboardInterrupt:
        print('User quit program...')
        time.sleep(5)
        for proc in job_list:
            proc.join()
        exit()
    else:
        move_dir('Probability')
        move_dir(str(odds))
        if not os.path.isfile('runs.csv'):
            write_to_csv('runs', return_dict.values()[0][0], return_dict.values()[0][1])
        for value in return_dict.values():
            write_to_csv('runs', value[2], value[3])
        print('Done!')
    finally:
        if shutdown:
            os.system('shutdown /S /F /T 0 /hybrid')

Однако, когда я запускаю этот новый код, появляется один индикатор выполнения, и каждый процесс перезаписывает панель, поэтому панель мигает случайными числами, что делает панель полезной. Я хочу иметь стек баров, по одному для каждого процесса, чтобы каждое обновление не прерывало другие. Бары не нужно заказывать; Мне просто нужно понять, как быстро каждый процесс выполняет свои задачи.

Answer 1

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

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