处理池结果而无需等待所有任务完成

如何解决处理池结果而无需等待所有任务完成

from multiprocessing import Pool
from functools import partial
from time import sleep
import random
import string
import uuid
import os
import glob



def task_a(param1,param2,mydata):

    thread_id = str(uuid.uuid4().hex)   # this may not be robust enough to guarantee no collisions,address
    output_filename = ''.join([str(thread_id),'.txt'])
    # part 1 - create output file for task_b to use
    with open(output_filename,'w') as outfile:
        for line in mydata:
            outfile.write(line)
    # part 2 - do some extra stuff (whilst task_b is running)
    sleep(5)
    print('Task A finished')
    return output_filename # not interested in return val


def task_b(expected_num_files):
    processed_files = 0
    while processed_files<expected_num_files:
        print('I am task_b,waiting for {} files ({} so far)'.format(expected_num_files,processed_files))
        path_to_search = ''
        for filename in glob.iglob(path_to_search + '*.txt',recursive=True):
            print('Got file : {}'.format(filename))
            # would do something complicated here
            os.rename(filename,filename+'.done')
            processed_files+=1
        sleep(10)



if __name__ == '__main__':

    param1 = ''     # dummy variable,need to support in solution
    param2 = ''     # dummy variable,need to support in solution

    num_workers = 2
    full_data = [[random.choice(string.ascii_lowercase) for _ in range(5)] for _ in range(100)]
    print(full_data)
    for i in range(0,len(full_data),num_workers):
        print('Going to process {}'.format(full_data[i:i+num_workers]))
        p = Pool(num_workers)
        task_a_func = partial(task_a,param1,param2)
        results = p.map(task_a_func,full_data[i:i+num_workers])
        p.close()
        p.join()
        task_b(expected_num_files=num_workers) # want this running sooner
        print('Iteration {} complete'.format(i))
        #want to wait for task_a's and task_b to finish

我无法安排这些任务并发运行。

task_a是一个多处理池，在执行过程中会生成输出文件。

task_b必须按任意顺序顺序处理输出文件（可以尽快处理），WHILST task_a继续运行（它将不再更改输出文件）

仅当所有task_a均已完成且task_b均已完成时才开始下一次迭代。

我发布的玩具代码显然在task_b启动之前等待task_a完全完成（这不是我想要的）

我已经看过多处理/子进程等，但是找不到同时启动池和单个task_b进程并等待两者都完成的方法。

task_b的编写就好像可以将其更改为外部脚本一样，但是我仍然对如何管理执行保持执着。

我是否应该有效地将task_b中的代码合并到task_a中，并以某种方式传递一个标志以确保每个池中的一个工作人员通过if / else'运行task_b代码'-至少那么我只是在池中等待完成？ / p>

解决方法

您可以使用进程间队列在任务a和任务b之间传递文件名。

此外，在循环内部重复初始化池是有害的，并且不必要地缓慢。 最好一开始就初始化池。

from multiprocessing import Pool,Manager,Event
from functools import partial
from time import sleep
import random
import string
import uuid
import os
import glob



def task_a(param1,param2,queue,mydata):
    thread_id = str(uuid.uuid4().hex)
    output_filename = ''.join([str(thread_id),'.txt'])
    output_filename = 'data/' + output_filename
    with open(output_filename,'w') as outfile:
        for line in mydata:
            outfile.write(line)
    print(f'{thread_id}: Task A file write complete for data {mydata}')
    queue.put(output_filename)
    print('Task A finished')


def task_b(queue,num_workers,data_size,event_task_b_done):
    print('Task b started!')
    processed_files = 0
    while True:
        filename = queue.get()
        if filename == 'QUIT':
            # Whenever you want task_b to quit,just push 'quit' to the queue
            print('Task b quitting')
            break
        print('Got file : {}'.format(filename))
        os.rename(filename,filename+'.done')
        processed_files+=1
        print(f'Have processed {processed_files} so far!')
        if (processed_files % num_workers == 0) or (processed_files ==  data_size):
            event_task_b_done.set()



if __name__ == '__main__':

    param1 = ''     # dummy variable,need to support in solution
    param2 = ''     # dummy variable,need to support in solution


    num_workers = 2
    data_size = 100
    full_data = [[random.choice(string.ascii_lowercase) for _ in range(5)] for _ in range(data_size)]
    mgr = Manager()
    queue = mgr.Queue()
    event_task_b_done = mgr.Event()
    # One extra worker for task b
    p = Pool(num_workers + 1)
    p.apply_async(task_b,args=(queue,event_task_b_done))
    task_a_func = partial(task_a,param1,queue)
    for i in range(0,len(full_data),num_workers):
        data = full_data[i:i+num_workers]
        print('Going to process {}'.format(data))
        p.map_async(task_a_func,full_data[i:i+num_workers])
        print(f'Waiting for task b to process all {num_workers} files...')
        event_task_b_done.wait()
        event_task_b_done.clear()
        print('Iteration {} complete'.format(i))
    queue.put('QUIT')
    p.close()
    p.join()
    exit(0)