我使用 python 多处理库来实现一种算法,其中有许多工作人员处理某些数据并将结果返回给父进程。我使用 multiprocessing.Queue 将作业传递给工作人员,然后收集结果。
一切都运行得很好,直到工作人员无法处理某些数据块。在下面的简化示例中,每个工作人员都有两个阶段:
- 初始化 - 可能会失败,在这种情况下,worker 应该被销毁
- 数据处理 - 处理一块数据可能会失败,在这种情况下,工作人员应该跳过该块并继续处理下一个数据。
当这两个阶段中的任何一个失败时,我都会在脚本完成后陷入僵局。这段代码模拟了我的问题:
import multiprocessing as mp
import random
workers_count = 5
# Probability of failure, change to simulate failures
fail_init_p = 0.2
fail_job_p = 0.3
#========= Worker =========
def do_work(job_state, arg):
if random.random() < fail_job_p:
raise Exception("Job failed")
return "job %d processed %d" % (job_state, arg)
def init(args):
if random.random() < fail_init_p:
raise Exception("Worker init failed")
return args
def worker_function(args, jobs_queue, result_queue):
# INIT
# What to do when init() fails?
try:
state = init(args)
except:
print "!Worker %d init fail" % args
return
# DO WORK
# Process data in the jobs queue
for job in iter(jobs_queue.get, None):
try:
# Can throw an exception!
result = do_work(state, job)
result_queue.put(result)
except:
print "!Job %d failed, skip..." % job
finally:
jobs_queue.task_done()
# Telling that we are done with processing stop token
jobs_queue.task_done()
#========= Parent =========
jobs = mp.JoinableQueue()
results = mp.Queue()
for i in range(workers_count):
mp.Process(target=worker_function, args=(i, jobs, results)).start()
# Populate jobs queue
results_to_expect = 0
for j in range(30):
jobs.put(j)
results_to_expect += 1
# Collecting the results
# What if some workers failed to process the job and we have
# less results than expected
for r in range(results_to_expect):
result = results.get()
print result
#Signal all workers to finish
for i in range(workers_count):
jobs.put(None)
#Wait for them to finish
jobs.join()
我对这段代码有两个问题:
- When
init()
失败,如何检测该工作人员无效而不是等待它完成?
- When
do_work()
失败,如何通知父进程结果队列中应该有更少的结果?
谢谢你的帮助!
我稍微更改了您的代码以使其正常工作(请参阅下面的说明)。
import multiprocessing as mp
import random
workers_count = 5
# Probability of failure, change to simulate failures
fail_init_p = 0.5
fail_job_p = 0.4
#========= Worker =========
def do_work(job_state, arg):
if random.random() < fail_job_p:
raise Exception("Job failed")
return "job %d processed %d" % (job_state, arg)
def init(args):
if random.random() < fail_init_p:
raise Exception("Worker init failed")
return args
def worker_function(args, jobs_queue, result_queue):
# INIT
# What to do when init() fails?
try:
state = init(args)
except:
print "!Worker %d init fail" % args
result_queue.put('init failed')
return
# DO WORK
# Process data in the jobs queue
for job in iter(jobs_queue.get, None):
try:
# Can throw an exception!
result = do_work(state, job)
result_queue.put(result)
except:
print "!Job %d failed, skip..." % job
result_queue.put('job failed')
#========= Parent =========
jobs = mp.Queue()
results = mp.Queue()
for i in range(workers_count):
mp.Process(target=worker_function, args=(i, jobs, results)).start()
# Populate jobs queue
results_to_expect = 0
for j in range(30):
jobs.put(j)
results_to_expect += 1
init_failures = 0
job_failures = 0
successes = 0
while job_failures + successes < 30 and init_failures < workers_count:
result = results.get()
init_failures += int(result == 'init failed')
job_failures += int(result == 'job failed')
successes += int(result != 'init failed' and result != 'job failed')
#print init_failures, job_failures, successes
for ii in range(workers_count):
jobs.put(None)
我的改变:
- Changed
jobs
成为一个正常人Queue
(代替JoinableQueue
).
- 现在,工作人员会返回特殊结果字符串“init failed”和“job failed”。
- 只要特定条件有效,主进程就会监视所述特殊结果。
- 最后,提出“停止”请求(即
None
工作),无论你有多少工人,无论如何。请注意,并非所有这些都可以从队列中提取(以防工作线程初始化失败)。
顺便说一句,您的原始代码很好并且易于使用。随机概率位非常酷。
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)