多处理 BrokenPipeError：[WinError 232] 使用 .send() 方法时管道正在关闭答案

【问题标题】：Multiprocessing BrokenPipeError: [WinError 232] The pipe is being closed when using .send() method多处理 BrokenPipeError：[WinError 232] 使用 .send() 方法时管道正在关闭
【发布时间】：2022-10-13 09:03:37
【问题描述】：

我目前正在开发一个使用在线 OCR API 的程序。这个 API 需要 2-5 秒来向我发送处理后的图像，因此用户可以开始处理第一张图像，而其余的则使用多处理在不同的 python 实例上处理，而不是让用户等待所有图像被处理.我一直在使用multiprocessing.Pipe() 来回发送值。代码在这里：

import multiprocessing as mp
# importing cv2, PIL, os, json, other stuff

def image_processor():
    # processes the first image in the list, then moves the remaining images to a different python instance:
    p_conn, c_conn = mp.Pipe()
    p = mp.Process(target=Processing.worker, args=([c_conn, images, path], 5))
    p.start()
    
    while True:
        out = p_conn.recv()
        if not out:
            break
        else:
            im_data.append(out)
            p_conn.send(True)


class Processing:
    def worker(data, mode, headers=0):
        # (some if statements go here)
        elif mode == 5:
        print(data[0])
        for im_name in data[1]:
            if data[1].index(im_name) != 0:
                im_path = f'{data[2]}\{im_name}'  # find image path
                im = pil_img.open(im_path).convert('L')  # open and grayscale image with PIL
                os.rename(im_path, f'{data[2]}\Archive\{im_name}')  # move original to archive
                im_grayscale = f'{data[2]}\g_{im_name}'  # create grayscale image path
                im.save(im_grayscale)  # save grayscale image
                    
                ocr_data = json.loads(bl.Visual.OCR.ocr_space_file(im_grayscale)).get('ParsedResults')[0].get('ParsedText').splitlines()
                print(ocr_data)
                data[0].send([{im_name}, f'{data[2]}\Archive\{im_name}', ocr_data])
                data[0].recv()
            
            data[0].send(False)

这给我留下了以下回溯：

Process Process-1:
Traceback (most recent call last):
  File "C:\Users\BruhK\AppData\Local\Programs\Python\Python310\lib\multiprocessing\process.py", line 315, in _bootstrap
    self.run()
  File "C:\Users\BruhK\AppData\Local\Programs\Python\Python310\lib\multiprocessing\process.py", line 108, in run
    self._target(*self._args, **self._kwargs)
  File "c:\Users\BruhK\PycharmProjects\pythonProject\FleetFeet-OCR-Final.py", line 275, in worker
    data[0].send([{im_name}, f'{data[2]}\Archive\{im_name}', ocr_data])
  File "C:\Users\BruhK\AppData\Local\Programs\Python\Python310\lib\multiprocessing\connection.py", line 211, in send
    self._send_bytes(_ForkingPickler.dumps(obj))
  File "C:\Users\BruhK\AppData\Local\Programs\Python\Python310\lib\multiprocessing\connection.py", line 285, in _send_bytes
    ov, err = _winapi.WriteFile(self._handle, buf, overlapped=True)
BrokenPipeError: [WinError 232] The pipe is being closed

请注意，从子函数发送到父函数的数据是 2d 或 3d 数组。在测试中，我已经能够在子函数和父函数之间来回发送 2d 和 3d 数组。

我用于测试的代码示例如下：

import multiprocessing as mp
import random
import time


def hang(p):
  hang_time = random.randint(1, 5)
  time.sleep(hang_time)
  print(p)
  p.send(hang_time)
  time.sleep(1)


class Child:
  def process():
    start = time.time()
    p_conn, c_conn = mp.Pipe()
    p = mp.Process(target=hang, args=(c_conn,))
    p.start()
    out = p_conn.recv()
    print(f'Waited for {time.time() - start}')
    p.join()
    print(f'New time: {time.time() - start}')
    return out


class Parent:
  def run():
    # do some stuff
    
    print(f'Hang time: {Child.process()}')
    
    # do some stuff


if __name__ == '__main__':
  Parent.run()

我该如何解决这个问题？是否需要任何其他信息？

【问题讨论】：

老实说，我没有通过您的完整代码。但是根据您描述的问题，我建议使用两个 queues 而不是管道。一个队列是为 ocr 进程提供“作业”，另一个是将结果发送回用户进程。我认为这将是一个更清洁的解决方案，这有帮助吗？
..我也想知道你为什么使用类？
Aaaand 看起来你有错误的意图：就像for 不在elif 内，更重要的是data[0].send(False) 在for 循环内，所以它在处理第一张图像和你的之后发送False主进程退出while(True)
@tturbo 我计算机上的文件中的缩进是正确的，只是将其放入堆栈交换错误。至于data[0].send(False) 在for 循环内，你是对的。我已经将它移出 for 循环，虽然它改变了结果，但我仍然需要做一些测试，因为执行程序现在卡在某个地方，没有新的 OCR 请求正在处理。这个我应该可以弄清楚，如果不是这种情况，我将打开一个新线程。我使用类来组织事物，以便更好地理解它们，Processing 中的主程序中有更多功能
@tturbo（续）类。我一直在努力理解队列系统，以及队列和管道之间的区别。我对多处理还是比较陌生，不太了解它。我会研究使用类。不过应该注意的是，将data[0].send(False) 移出for 循环已经完全停止了初始错误。谢谢你。

标签： python python-multiprocessing

【解决方案1】：

正如@tturbo 指出的那样，代码data[0].send(False) 位于它应该位于外部的for 循环内，这阻止了损坏的管道错误。我不确定为什么要修复它，如果其他人愿意对此有所了解，请成为我的客人。对我来说，重要的是它奏效了。谢谢你。

【讨论】：