Python多处理过程如何重用流程?

2024-09-30 22:27:34 发布

您现在位置:Python中文网/ 问答频道 /正文

我使用python多处理模块并行运行一些长时间运行的任务。我使用start()方法运行作业,但是一旦作业返回,我想再次运行它们。在

是否可以重用我创建的流程?还是每次运行作业时都必须创建一个新的进程对象?在

Pyton文档中有一节建议我不能使用start()方法超过一个,但也许有人知道另一种重用实例的方法:

start()

启动流程活动。

每个进程对象最多只能调用一次。 它安排在单独的进程中调用对象的run()方法。

这是我的Process类版本:

class Process(multiprocessing.Process):
    def __init__(self, result_queue, MCMCinstance):
        assert isinstance(MCMCinstance, MCMC)
        multiprocessing.Process.__init__(self)
        self.result_queue = result_queue
        self.mcmc = MCMCinstance
        self.interface = C_interface(self.mcmc)
        self.burn_in = False

    def run(self):
        if self.burn_in: interface.burn_in()
        self.interface.sample(self.mcmc.options.runs)
        self.interface.update(self.mcmc)
        self.result_queue.put(self.mcmc)

然后,我实例化这些进程并使用start()方法运行它们:

^{pr2}$

Tags: 对象方法inselfqueue进程作业流程
3条回答

要重用流程,您应该使用池。像这样的东西应该行得通,不过我还没有测试过。在

SENTINEL = "SENTINEL"

class Worker(object):
    def __init__(self, result_queue, MCMCinstance):
        assert isinstance(MCMCinstance, MCMC)
        self.result_queue = result_queue
        self.mcmc = MCMCinstance
        self.interface = C_interface(self.mcmc)
        self.burn_in = False

    def run(self):
        if self.burn_in: interface.burn_in()
        self.interface.sample(self.mcmc.options.runs)
        self.interface.update(self.mcmc)
        #Signal exit by putting SENTINEL in the queue 
        if True:       
            self.result_queue.put(SENTINEL)
        else:
            self.result_queue.put(self.mcmc)

def run(result_queue):
    while True:
        instance = result_queue.get(True)
        if instance == SENTINEL:
            break
        worker = Worker(result_queue, instance)
        worker.run()

if __name__ == "__main__":
    result_queue = multiprocessing.Queue()
    pool = multiprocessing.pool.Pool(3, run, (result_queue,)) # Use a pool with 5 process

    mcmc1 = MCMC(options, donors, clusters)
    mcmc2 = MCMC(options, donors, clusters)
    mcmc3 = MCMC(options, donors, clusters)
    mcmc4 = MCMC(options, donors, clusters)

    result_queue.put(mcmc1)  
    result_queue.put(mcmc2)  
    result_queue.put(mcmc3)  
    result_queue.put(mcmc4)  

    pool.close()
    pool.join()

不,不可能。在^{}中有一个特定的防范措施。在

我只能推测为什么它不可重用,但我认为这是一个设计选择。它可能会给类添加太多的逻辑来回收对象,这是值得的。但我觉得更有趣的是,问为什么会这样。在

不过,在阅读了过去20分钟的源代码之后,我可以肯定地说,仅仅创建整个python进程的一个分支,比创建一个对象的新实例要花更多的时间,所以从性能上讲,这一点都不重要。在

至于您的代码,您可以稍微压缩一下,您不必使用命名的Process实例,并利用列表理解。在

# setup the jobs and run
result_queue = multiprocessing.Queue()

mcmc_list = [MCMC(options, donors, clusters)]*4
jobs = [Process(result_queue, mcmc) for mcmc in mcmc_list ]
for job in jobs:
    #job.debug_level = 1
    job.start()

results = [result_queue.get() for job in jobs]

#for res in results: res.traceplot(show=False)
jobs2 = [Process(result_queue, result) for result in results]

for j in jobs2:
    j.start()

results2 = [result_queue.get() for job in jobs2]

编辑: 我也认为你有点误用Queue这是为了在进程之间进行通信,我想你不需要它。要创建线程池,应该使用^{}Pool.map。但是我不能给出确切的代码示例,没有看到原始的目标函数。我认为这需要调整。在

正如文档所说,您只能调用一次.start()方法,我相信每次都必须创建新的进程:

# setup the jobs and run
result_queue = multiprocessing.Queue()

mcmc1 = MCMC(options, donors, clusters)
mcmc2 = MCMC(options, donors, clusters)
mcmc3 = MCMC(options, donors, clusters)
mcmc4 = MCMC(options, donors, clusters)

p1 = Process(result_queue, mcmc1)
p2 = Process(result_queue, mcmc2)
p3 = Process(result_queue, mcmc3)
p4 = Process(result_queue, mcmc4)

jobs = [p1, p2, p3, p4]

for job in jobs:
    #job.debug_level = 1
    job.start()

results = [result_queue.get() for job in jobs]

#for res in results: res.traceplot(show=False)
p5 = Process(result_queue, results[0])
p6 = Process(result_queue, results[1])
p7 = Process(result_queue, results[2])
p8 = Process(result_queue, results[3])

jobs2 = [p5, p6, p7, p8]

for j in jobs2:
    j.start()


results2 = [result_queue.get() for job in jobs2]

相关问题 更多 >