【问题标题】:Python: multiprocessing Queue.put() in module won't send anything to parent processPython:模块中的多处理 Queue.put() 不会向父进程发送任何内容
【发布时间】:2019-10-27 18:45:09
【问题描述】:

我正在尝试使用 Python 中的多处理包,更准确地说是 Queue() 类,使 2 个进程相互通信。从父进程中,我想每 5 秒获取一次子进程的更新值。这个子进程是一个类函数。我做了一个玩具示例,一切正常。

但是,当我尝试在我的项目中实现这个解决方案时,子模块中子进程的 Queue.put() 方法似乎不会向父进程发送任何内容,因为父进程赢了'不打印所需的值,代码永远不会停止运行。实际上,父进程只打印发送给子进程的值,这里是True,但正如我所说,永远不会停止。

所以我的问题是:

  1. 我的玩具示例中是否有任何错误?

  2. 我应该如何修改我的项目才能让它像我的玩具示例一样工作?

玩具示例:作品

主模块

from multiprocessing import Process, Event, Lock, Queue, Pipe
import time 
import test_mod as test

def loop(output):
    stop_event = Event()
    q = Queue()
    child_process = Process(target=test.child.sub, args=(q,))
    child_process.start()
    i = 0
    print("started at {} ".format(time.time()))

    while not stop_event.is_set():
        i+=1
        time.sleep(5)
        q.put(True)
        print(q.get())
        if i == 5:
            child_process.terminate()
            stop_event.set()

    output.put("main process looped")

if __name__ == '__main__':
    stop_event, output = Event(), Queue()
    k = 0
    while k < 5:
        loop_process = Process(target=loop, args=(output,))
        loop_process.start()
        print(output.get())
        loop_process.join()
        k+=1

子模块

from multiprocessing import Process, Event, Lock, Queue, Pipe
import time


class child(object):
    def __init__(self):
        pass

    def sub(q):
        i = 0
        while i < 2000:
            latest_value = time.time()
            accord = q.get()
            if accord == True:
                q.put(latest_value)
            accord = False
            time.sleep(0.0000000005)
            i+=1

项目代码:不起作用

主模块

import neat #package in which the submodule is 
import *some other stuff*

def run(config_file):

    config = neat.Config(some configuration)

    p = neat.Population(config)

    **WHERE MY PROBLEM IS**

    stop_event = Event()
    q = Queue()
    pe = neat.ParallelEvaluator(**args)

    child_process = Process(target=p.run, args=(pe.evaluate, q, other args))
    child_process.start()

    i = 0
    while not stop_event.is_set():

        q.put(True)
        print(q.get())
        time.sleep(5)
        i += 1
        if i == 5:
            child_process.terminate()
            stop_event.set()

if __name__ == '__main__':
    run(config_file)

子模块

class Population(object):
    def __init__():
      *initialization*

    def run(self, q, other args):

        while n is None or k < n:
            *some stuff*
            accord = add_2.get()
            if accord == True:
                add_2.put(self.best_genome.fitness)
            accord = False

        return self.best_genome

注意:

  1. 我不习惯多处理

  2. 鉴于整个代码太长,我已尝试给出我项目中最相关的部分。

  3. 我也考虑过使用 Pipe(),但是这个选项也不起作用。

【问题讨论】:

    标签: python module multiprocessing queue communication


    【解决方案1】:

    如果我没看错,你想要的子模块是类Population。但是,您使用ParallelEvaluator 类型的参数开始您的过程。接下来,我看不到您将队列q 提供给子流程。这就是我从提供的代码中看到的:

    stop_event = Event()
    q = Queue()
    pe = neat.ParallelEvaluator(**args)
    
    child_process = Process(target=p.run, args=(pe.evaluate, **args)
    child_process.start()
    

    此外,以下几行创建了一个竞争条件:

    q.put(True)
    print(q.get())
    

    get 命令类似于pop。所以它需要一个元素并将其从队列中删除。如果您的子进程没有访问这两行之间的队列(因为它很忙),True 将永远不会进入子进程。因此,最好两个使用多个队列。每个方向一个。比如:

    stop_event = Event()
    q_in = Queue()
    q_out = Queue()
    pe = neat.ParallelEvaluator(**args)
    
    child_process = Process(target=p.run, args=(pe.evaluate, **args))
    child_process.start()
    
    i = 0
    while not stop_event.is_set():
    
         q_in.put(True)
         print(q_out.get())
         time.sleep(5)
         i += 1
         if i == 5:
             child_process.terminate()
             stop_event.set()
    

    这是你的子模块

    class Population(object):
        def __init__():
          *initialization*
    
        def run(self, **args):
    
            while n is None or k < n:
                *some stuff*
                accord = add_2.get()           # add_2 = q_in
                if accord == True:
                    add_3.put(self.best_genome.fitness)  #add_3 = q_out
                accord = False
    
            return self.best_genome
    

    【讨论】:

    • 非常感谢您的回复。我会在家里试试,我暂时做不到。但实际上,在我的项目中,我确实在子流程中传递了Queue 参数,它包含在run(self, **args) 中,我同意这不清楚,我已经更新了我的问题。但即便如此,它也不起作用。我会在尝试时通知您。再次,非常感谢
    • 实际上,我注意到问题出现在我的代码中。因为在这个模块中,在add_3 方法之前,我调用了一个包含pool.apply_async() 方法的函数。这个函数在pemodule里面,我没注意。深入研究程序永远运行的地方,我注意到我遇到了这个问题:bugs.python.org/issue25829 我想我会发布另一个问题,因为问题现在完全不同了。但是,非常感谢,我相信您的回答会进一步帮助我
    猜你喜欢
    • 1970-01-01
    • 1970-01-01
    • 2015-01-03
    • 2011-02-16
    • 1970-01-01
    • 2017-08-29
    • 2021-05-01
    • 1970-01-01
    • 2019-01-10
    相关资源
    最近更新 更多