在工作进程死亡时恢复丢失的multiprocessing.Queue项

Question

我的情况是这样的：

• 我有一个工作程序，它会把任务放进一个叫做multiprocessing.Queue()的队列里，前提是这个队列是空的。这样做是为了确保任务按照一定的优先级执行，因为multiprocessing.Queue()本身不支持优先级。
• 有几个工作程序会从这个队列中取出任务并处理。有时候（不到0.1%的情况），这些工作程序会失败并退出，导致无法重新放回任务。
• 我的任务是通过一个中央数据库进行锁定的，每个任务只能执行一次（这是硬性要求）。为此，任务有一些状态，可以在这些状态之间转换。

我现在的解决方案是：让所有工作程序通过另一个队列反馈哪些任务已经完成，并设置一个截止时间，要求任务在这个时间之前完成。如果超过了截止时间，就重置任务并重新放回队列。但这个方案有个问题，就是它是“软”的，也就是说截止时间是随意的。

我在寻找最简单的解决方案。有没有更简单或者更严格的解决方案呢？

Answer 1

这个方案使用了三个队列来跟踪工作进度（用WORK_ID来模拟）：

• todo_q：待处理的工作（包括如果处理过程中出现问题需要重新做的工作）
• start_q：已经开始的工作
• finish_q：已经完成的工作

用这种方法，你不需要定时器。只要给每个进程分配一个标识符，并跟踪这些分配，检查一下Process.is_alive()。如果进程挂掉了，就把那项工作重新放回待处理队列。

在下面的代码中，我模拟了一个工作进程有25%的几率会挂掉……

from multiprocessing import Process, Queue
from Queue import Empty
from random import choice as rndchoice
import time

def worker(id, todo_q, start_q, finish_q):
    """multiprocessing worker"""
    msg = None
    while (msg!='DONE'):
        try:
            msg = todo_q.get_nowait()    # Poll non-blocking on todo_q
            if (msg!='DONE'):
                start_q.put((id, msg))   # Let the controller know work started
                time.sleep(0.05)
                if (rndchoice(range(3))==1):
                    # Die a fraction of the time before finishing
                    print "DEATH to worker %s who had task=%s" % (id, msg)
                    break
                finish_q.put((id, msg))  # Acknowledge work finished
        except Empty:
            pass
    return

if __name__ == '__main__':
    NUM_WORKERS = 5
    WORK_ID = set(['A','B','C','D','E']) # Work to be done, you will need to
                                    #    name work items so they are unique
    WORK_DONE = set([])             # Work that has been done
    ASSIGNMENTS = dict()            # Who was assigned a task
    workers = dict()
    todo_q = Queue()
    start_q = Queue()
    finish_q = Queue()

    print "Starting %s tasks" % len(WORK_ID)
    # Add work
    for work in WORK_ID:
        todo_q.put(work)

    # spawn workers
    for ii in xrange(NUM_WORKERS):
        p = Process(target=worker, args=(ii, todo_q, start_q, finish_q))
        workers[ii] = p
        p.start()

    finished = False
    while True:
        try:
            start_ack = start_q.get_nowait()  # Poll for work started
            ## Check for race condition between start_ack and finished_ack
            if not ASSIGNMENTS.get(start_ack[0], False):
                ASSIGNMENTS[start_ack[0]] = start_ack   # Track the assignment
                print "ASSIGNED worker=%s task=%s" % (start_ack[0], 
                    start_ack[1])
                WORK_ID.remove(start_ack[1])      # Account for started tasks
            else:
                # Race condition. Never overwrite existing assignments
                # Wait until the ASSIGNMENT is cleared
                start_q.put(start_ack)
        except Empty:
            pass

        try:
            finished_ack = finish_q.get_nowait()  # Poll for work finished
            # Check for race condition between start_ack and finished_ack
            if (ASSIGNMENTS[finished_ack[0]][1]==finished_ack[1]):
                # Clean up after the finished task
                print "REMOVED worker=%s task=%s" % (finished_ack[0], 
                    finished_ack[1])
                del ASSIGNMENTS[finished_ack[0]]
                WORK_DONE.add(finished_ack[1])
            else:
                # Race condition. Never overwrite existing assignments
                # It was received out of order... wait for the 'start_ack'
                finish_q.put(finished_ack)
            finished_ack = None
        except Empty:
            pass

        # Look for any dead workers, and put their work back on the todo_q
        if not finished:
            for id, p in workers.items():
                status = p.is_alive()
                if not status:
                    print "    WORKER %s FAILED!" % id
                    # Add to the work again...
                    todo_q.put(ASSIGNMENTS[id][1])
                    WORK_ID.add(ASSIGNMENTS[id][1])
                    del ASSIGNMENTS[id]      # Worker is dead now
                    del workers[id]
                    ii += 1
                    print "Spawning worker number", ii
                    # Respawn a worker to replace the one that died
                    p = Process(target=worker, args=(ii, todo_q, start_q, 
                        finish_q))
                    workers[ii] = p
                    p.start()
        else:
            for id, p in workers.items():
                p.join()
                del workers[id]
            break

        if (WORK_ID==set([])) and (ASSIGNMENTS.keys()==list()):
            finished = True
            [todo_q.put('DONE') for x in xrange(NUM_WORKERS)]
        else:
            pass
    print "We finished %s tasks" % len(WORK_DONE)

在我的笔记本上运行这个……

mpenning@mpenning-T61:~$ python queueack.py
Starting 5 tasks
ASSIGNED worker=2 task=C
ASSIGNED worker=0 task=A
ASSIGNED worker=4 task=B
ASSIGNED worker=3 task=E
ASSIGNED worker=1 task=D
DEATH to worker 4 who had task=B
DEATH to worker 3 who had task=E
    WORKER 3 FAILED!
Spawning worker number 5
    WORKER 4 FAILED!
Spawning worker number 6
REMOVED worker=2 task=C
REMOVED worker=0 task=A
REMOVED worker=1 task=D
ASSIGNED worker=0 task=B
ASSIGNED worker=2 task=E
REMOVED worker=2 task=E
DEATH to worker 0 who had task=B
    WORKER 0 FAILED!
Spawning worker number 7
ASSIGNED worker=5 task=B
REMOVED worker=5 task=B
We finished 5 tasks
mpenning@mpenning-T61:~$

我用超过10000个工作项进行了测试，挂掉率是25%。