Python多处理IOError:[Errno 232]管道正在关闭 [英] Python multiprocessing IOError: [Errno 232] The pipe is being closed

问题描述

我正在尝试在mutliprocessing上实现此教程在python中，但是当我尝试执行自己的任务时，出现以下错误:

I am trying to implement this tutorial on mutliprocessing in python, but when I tried to do my own task I get the following error:

Traceback (most recent call last):
>>>   File "C:\Python27\lib\multiprocessing\queues.py", line 262, in _feed
    send(obj)
IOError: [Errno 232] The pipe is being closed

这是我尝试做的可复制示例，它给出了相同的错误消息:

Here is a reproducible example of what I am trying to do which gives the same error message:

from multiprocessing import Lock, Process, Queue, current_process
import time

class Testclass(object):
    def __init__(self, x):
        self.x = x

def toyfunction(testclass):
    testclass.product = testclass.x * testclass.x
    return testclass


def worker(work_queue, done_queue):
    try:
        for testclass in iter(work_queue.get, 'STOP'):
            print(testclass.counter)
            newtestclass = toyfunction(testclass)
            done_queue.put(newtestclass)

    except:
        print('error')

    return True

def main():

    counter = 1

    database = []
    while counter <= 1000:
        database.append(Testclass(3))
        counter += 1
        print(counter)



    workers = 8
    work_queue = Queue()
    done_queue = Queue()
    processes = []

    start = time.clock()
    counter = 1

    for testclass in database:
        testclass.counter = counter
        work_queue.put(testclass)
        counter += 1
        print(counter)


    print('items loaded')
    for w in range(workers):
        p = Process(target=worker, args=(work_queue, done_queue))
        p.start()
        processes.append(p)
        work_queue.put('STOP')

    for p in processes:
        p.join()

    done_queue.put('STOP')

    print(time.clock()-start)
    print("Done")

if __name__ == '__main__':
    main()    

推荐答案

当我添加处理完成的队列的代码时，我不再收到错误.这是工作代码:

When I add code that processes the done queue I no longer get the error. Here is working code:

from multiprocessing import Lock, Process, Queue, current_process
import time

class Testclass(object):
    def __init__(self, x):
        self.x = x

def toyfunction(testclass):
    testclass.product = testclass.x * testclass.x
    return testclass


def worker(work_queue, done_queue):
    try:
        for testclass in iter(work_queue.get, 'STOP'):
            print(testclass.counter)
            newtestclass = toyfunction(testclass)
            done_queue.put(newtestclass)

    except:
        print('error')

    return True

def main():

    counter = 1

    database = []
    while counter <= 100:
        database.append(Testclass(10))
        counter += 1
        print(counter)



    workers = 8
    work_queue = Queue()
    done_queue = Queue()
    processes = []

    start = time.clock()
    counter = 1

    for testclass in database:
        testclass.counter = counter
        work_queue.put(testclass)
        counter += 1
        print(counter)


    print('items loaded')
    for w in range(workers):
        p = Process(target=worker, args=(work_queue, done_queue))
        p.start()
        processes.append(p)
        work_queue.put('STOP')

    for p in processes:
        p.join()

    done_queue.put('STOP')

    newdatabase = []
    for testclass in iter(done_queue.get, 'STOP'):
        newdatabase.append(testclass)

    print(time.clock()-start)
    print("Done")
    return(newdatabase)

if __name__ == '__main__':
    database = main()

这篇关于Python多处理IOError:[Errno 232]管道正在关闭的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！