20

multiprocessing.Poolおよびを使用して、Python でプロデューサー/コンシューマー パターンを実装していmultiprocessing.Queueます。geventコンシューマーは、複数のタスクを生成するために使用する事前にフォークされたプロセスです。

以下は、コードの縮小版です。

import gevent
from Queue import Empty as QueueEmpty
from multiprocessing import Process, Queue, Pool
import signal
import time

# Task queue
queue = Queue()

def init_worker ():
    # Ignore signals in worker
    signal.signal( signal.SIGTERM, signal.SIG_IGN )
    signal.signal( signal.SIGINT, signal.SIG_IGN )
    signal.signal( signal.SIGQUIT, signal.SIG_IGN )

# One of the worker task
def worker_task1( ):
    while True:
        try:
            m = queue.get( timeout = 2 )

            # Break out if producer says quit
            if m == 'QUIT':
                print 'TIME TO QUIT'
                break

        except QueueEmpty:
            pass

# Worker
def work( ):
    gevent.joinall([
        gevent.spawn( worker_task1 ),
    ])

pool = Pool( 2, init_worker )
for i in xrange( 2 ):
    pool.apply_async( work )

try:
    while True:
        queue.put( 'Some Task' )
        time.sleep( 2 )

except KeyboardInterrupt as e:
    print 'STOPPING'

    # Signal all workers to quit
    for i in xrange( 2 ):
        queue.put( 'QUIT' )

    pool.join()

終了しようとすると、次の状態になります。

  1. 親プロセスは、子の 1 つが参加するのを待っています。
  2. 子供の 1 人は死亡した状態です。終了しましたが、親は他の子が終了するのを待っています。
  3. 他の子が表示されています: futex(0x7f99d9188000, FUTEX_WAIT, 0, NULL ....

では、そのようなプロセスをきれいに終了させる正しい方法は何ですか?

4

1 に答える 1

14

私は問題を理解しました。のドキュメントにmultiprocessing.Pool.join()よると、になる前に にpoolなる必要があります。問題を解決する前に追加しました。close()edjoin()edpool.close()pool.join()

于 2013-05-23T08:12:48.590 に答える