멀티 컨트롤러-파이프와 큐

멀티 컨트롤러-파이프와 큐

---

에서 큐와 파이프의 패키지 인 차이점은 무엇입니까?

어떤 시나리오에서 다른 시나리오를 선택해야합니까? 언제 사용하는 Pipe()것이 유리입니까? 언제 사용하는 Queue()것이 유리입니까?

---

• A 는 두 개의 끝 포인트 만 잇습니다.Pipe()

• 여러 생산자와 소비자가 있습니다 .Queue()

사용시기

의사 소통에 두 개 이상의 포인트가 필요한 경우를 사용하십시오 .Queue()

성능이 필요하기 때문에 절대적인하면 위에 구축되어 있기 때문에 가 훨씬 빠릅니다 .Pipe()Queue()Pipe()

성능 벤치마킹

두 개의 프로세스를 생성하고 가능한 빨리 그들 사이에 메시지를 보내려고 가정 해봅시다. 이것 Pipe()과 Queue()...를 사용하는 드래그 테스트 사이의 드래그 레이스의 타이밍 결과입니다 . 이 Ubuntu 11.10을 실행하는 ThinkpadT61과 Python 2.7.2에 있습니다.

참고로, 나는 개체로 결과를 던졌다 . 호출 될 때 작업을 설명합니다 (특정 작업에 알지 충분하지 않은 작업 만 계산 함 ). 따라서 작업이 완료 하실 수 있습니다.JoinableQueue()JoinableQueue()queue.task_done()queue.join()

이 답변의 맨 아래에있는 각 코드는 ...

mpenning@mpenning-T61:~$ python multi_pipe.py 
Sending 10000 numbers to Pipe() took 0.0369849205017 seconds
Sending 100000 numbers to Pipe() took 0.328398942947 seconds
Sending 1000000 numbers to Pipe() took 3.17266988754 seconds
mpenning@mpenning-T61:~$ python multi_queue.py 
Sending 10000 numbers to Queue() took 0.105256080627 seconds
Sending 100000 numbers to Queue() took 0.980564117432 seconds
Sending 1000000 numbers to Queue() took 10.1611330509 seconds
mpnening@mpenning-T61:~$ python multi_joinablequeue.py 
Sending 10000 numbers to JoinableQueue() took 0.172781944275 seconds
Sending 100000 numbers to JoinableQueue() took 1.5714070797 seconds
Sending 1000000 numbers to JoinableQueue() took 15.8527247906 seconds
mpenning@mpenning-T61:~$

요약하면 Pipe()a보다 약 3 배 빠 Queue(). JoinableQueue()당신이 가족으로부터 얻는 것을 가져옵니다 .

개체 자료 2

멀티 실행은 정보 흐름에 미묘한 변경을 도입하여 약간의 경우를 수행하기 어렵게 만듭니다. 예를 들어, 여러 조건에서 사전을 통해 색인을 생성 할 때 제대로 작동하지만 특정 입력에서 실패 할 가능성이 있습니다.

일반적으로 우리는 전체 프로세스가 충돌 할 때 실패 할 때 얻습니다. 그러나 다중 처리 기능이 충돌하는 경우 원치 않는 충돌 추적이 콘솔에 인쇄되지 않습니다. 알려지지 않은 멀티 프로세스 크래시를 추적하는 것은 프로세스 크래시의 단서가없이 어렵습니다.

멀티 프로세스 충돌 정보를 추적하는 가장 간단한 방법은 전체 멀티 가동 기능을 try/ 로서 except사용하는 것을 구석입니다 traceback.print_exc().

import traceback
def reader(args):
    try:
        # Insert stuff to be multiprocessed here
        return args[0]['that']
    except:
        print "FATAL: reader({0}) exited while multiprocessing".format(args) 
        traceback.print_exc()

이제 충돌을 발견하면 다음과 같은 내용이 표시됩니다.

FATAL: reader([{'crash', 'this'}]) exited while multiprocessing
Traceback (most recent call last):
  File "foo.py", line 19, in __init__
    self.run(task_q, result_q)
  File "foo.py", line 46, in run
    raise ValueError
ValueError

소스 코드 :

---

"""
multi_pipe.py
"""
from multiprocessing import Process, Pipe
import time

def reader_proc(pipe):
    ## Read from the pipe; this will be spawned as a separate Process
    p_output, p_input = pipe
    p_input.close()    # We are only reading
    while True:
        msg = p_output.recv()    # Read from the output pipe and do nothing
        if msg=='DONE':
            break

def writer(count, p_input):
    for ii in xrange(0, count):
        p_input.send(ii)             # Write 'count' numbers into the input pipe
    p_input.send('DONE')

if __name__=='__main__':
    for count in [10**4, 10**5, 10**6]:
        # Pipes are unidirectional with two endpoints:  p_input ------> p_output
        p_output, p_input = Pipe()  # writer() writes to p_input from _this_ process
        reader_p = Process(target=reader_proc, args=((p_output, p_input),))
        reader_p.daemon = True
        reader_p.start()     # Launch the reader process

        p_output.close()       # We no longer need this part of the Pipe()
        _start = time.time()
        writer(count, p_input) # Send a lot of stuff to reader_proc()
        p_input.close()
        reader_p.join()
        print("Sending {0} numbers to Pipe() took {1} seconds".format(count,
            (time.time() - _start)))

---

"""
multi_queue.py
"""

from multiprocessing import Process, Queue
import time
import sys

def reader_proc(queue):
    ## Read from the queue; this will be spawned as a separate Process
    while True:
        msg = queue.get()         # Read from the queue and do nothing
        if (msg == 'DONE'):
            break

def writer(count, queue):
    ## Write to the queue
    for ii in range(0, count):
        queue.put(ii)             # Write 'count' numbers into the queue
    queue.put('DONE')

if __name__=='__main__':
    pqueue = Queue() # writer() writes to pqueue from _this_ process
    for count in [10**4, 10**5, 10**6]:             
        ### reader_proc() reads from pqueue as a separate process
        reader_p = Process(target=reader_proc, args=((pqueue),))
        reader_p.daemon = True
        reader_p.start()        # Launch reader_proc() as a separate python process

        _start = time.time()
        writer(count, pqueue)    # Send a lot of stuff to reader()
        reader_p.join()         # Wait for the reader to finish
        print("Sending {0} numbers to Queue() took {1} seconds".format(count, 
            (time.time() - _start)))

---

"""
multi_joinablequeue.py
"""
from multiprocessing import Process, JoinableQueue
import time

def reader_proc(queue):
    ## Read from the queue; this will be spawned as a separate Process
    while True:
        msg = queue.get()         # Read from the queue and do nothing
        queue.task_done()

def writer(count, queue):
    for ii in xrange(0, count):
        queue.put(ii)             # Write 'count' numbers into the queue

if __name__=='__main__':
    for count in [10**4, 10**5, 10**6]:
        jqueue = JoinableQueue() # writer() writes to jqueue from _this_ process
        # reader_proc() reads from jqueue as a different process...
        reader_p = Process(target=reader_proc, args=((jqueue),))
        reader_p.daemon = True
        reader_p.start()     # Launch the reader process
        _start = time.time()
        writer(count, jqueue) # Send a lot of stuff to reader_proc() (in different process)
        jqueue.join()         # Wait for the reader to finish
        print("Sending {0} numbers to JoinableQueue() took {1} seconds".format(count, 
            (time.time() - _start)))

참고 URL : https://stackoverflow.com/questions/8463008/multiprocessing-pipe-vs-queue