时间戳

首页 > 解决方案 > Python多处理，从子进程读取输入

python - Python多处理，从子进程读取输入

问题描述

我有一个多处理系统，有一个主进程和两个子进程。

其中一个子进程（例如，C1）通过队列向另一个子进程（C2）发送消息。C2 分析来自 C1 的消息，当某些情况发生时，它需要用户的一些输入。

所以，情况是这样的：

主文件

from child_process_01 import Child_Process_01
from child_process_02 import Child_Process_02

set_start_method("spawn")

c1_c2_q = Queue(maxsize=1)

c1 = mp.Process(target=Child_Process_01, args=(c1_c2_q,))
c1.daemon = True
c1.start()

c2 = mp.Process(target=Child_Process_02, args=(c1_c2_q,))
c2.daemon = True
c2.start()

Child_Process_01

message = produced somehow
c1_c2_q.put((message))

Child_Process_02

## infinite loop, keep checking messages and ask for user input under certain conditions

while True:
    try:
        message = c1_c2_q.get_nowait()
    except:
        message = None

    ## when message arrives
    if message:
        if message = something:
            usr_input = input('In this situation, I need user input')

代码不能按原样工作，因为多处理模块关闭了它创建的所有进程的标准输入，正如我在这里的许多答案中发现的那样。一个很好的建议似乎是在主流程中重新定义标准并将其发送给孩子，就像解释here或here一样，所以我尝试了：

主文件

newstdin = os.fdopen(os.dup(sys.stdin.fileno()))

c2 = mp.Process(target=Child_Process_02, args=(c1_c2_q, newstdin))
c2.daemon = True
c2.start()

Child_Process_02

def Child_Process_02(c1_c2_q, newstdin):
    sys.stdin = newstdin
    while True:
        try:
            message = c1_c2_q.get_nowait()
        except:
            message = None

        ## when message arrives
        if message:
            if message = something:
                usr_input = input('In this situation, I need user input')

但这也不起作用，因为我无法通过队列传递在主进程中创建的 newstdin 对象，我收到错误消息：

TypeError: cannot pickle '_io.TextIOWrapper' object

此外，对类似问题的一些评论通常不鼓励从子进程读取输入的做法，但我无法想象如何以不同的方式做到这一点。有什么建议么？

标签： pythonmultiprocessingpython-multiprocessing

解决方案

首先，您应该修改Child_Process_02以便c1_c2_q.get_nowait()通过使用阻塞调用不会消耗太多 CPU 周期来进行非生产性调用，阻塞调用c1_c2_q.get()在收到消息之前不会返回（并且您可以摆脱try/except围绕此调用的脚手架）。

其次，这是可选的，因为您从Child_Process_01传递到Child_Process_02的消息有一个生产者和一个消费者，您可以通过multiprocessing.Queue调用来提高效率multiprocessing.Pipe(duplex=False)，这将返回两个单向multiprocessing.connection.Connection实例，其中第二个适用于发送任意对象例如字符串，其中第一个用于接收对象（顺便说一下multiprocessing.Queue是在 之上实现的Pipe）。但是一个全双工Pipe，其中每个连接都可以用于发送和接收，这绝对是您想要的以下程序，其中由主进程启动的线程正在“侦听”input通过调用进行调用的请求multiprocessing.connection.Connection.recv()在一个循环中。此调用返回的消息是用于调用的提示字符串input。然后函数返回的值input将sent返回到连接上：

from multiprocessing import Process, Pipe
from threading import Thread

def inputter(input_conn):
    """ get requests to do input calls """
    while True:
        input_msg = input_conn.recv()
        value = input(input_msg)
        input_conn.send(value) # send inputted value:


def worker(msg_conn, input_conn):
    while True:
        message = msg_conn.recv()
        if message is None:
            break
        if message == 'do input':
            # send inputter our prompt message:
            input_conn.send('Enter x: ')
            # get back the result of the input:
            x = (int)(input_conn.recv())
            print('The value entered was', x)
        else:
            print('Got message:', message)


if __name__ == '__main__':
    import time

    # create the connections for sending messages from one process to another:
    recv_conn, send_conn = Pipe(duplex=False)

    # create the connections for doing the input requests:
    input_conn1, input_conn2 = Pipe(duplex=True) # each connection is bi-drectional

    # start the inputter thread with one of the inputter duplex connections:
    t = Thread(target=inputter, args=(input_conn1,), daemon=True)
    t.start()

    # start a child process with the message connection in lieu of a Queue
    # and the other inputter connection:
    p = Process(target=worker, args=(recv_conn, input_conn2))
    p.start()

    # send messages to worker process:
    send_conn.send('a')
    send_conn.send('do input')
    send_conn.send('b')
    # signal the child process to terminate:
    send_conn.send(None)
    p.join()

印刷：

Got message: a
Enter x: 8
The value entered was 8
Got message: b

笔记

应该注意的是，amultiprocessing.Queue为底层启动了一个 feeder 线程，multiprocessing.Pipe以防止“推杆”在调用maxsizeput之前过早阻塞，其中maxsize是用于实例化队列的值。但这也是为什么multiprocessing.Queue性能不如multiprocessing.Pipe. 但这也意味着send在一个连接上重复recv调用而另一端没有相应的调用最终会阻塞。但鉴于您已在队列中指定maxsize=1，这几乎不是问题，因为您会在与队列相同的情况下阻塞。

推荐阅读