คุณสามารถใช้ Queue
เพื่อป้อนกลับความล้มเหลวใน Pool
ผ่านการวนซ้ำในการเริ่มต้น Process
:
import multiprocessing as mp
import random
def f(x):
if random.getrandbits(1):
# on failure / exception catch
f.q.put(x)
return None
return x*x
def f_init(q):
f.q = q
def main(pending):
total_items = len(pending)
successful = []
failure_tracker = []
q = mp.Queue()
p = mp.Pool(None, f_init, [q])
results = p.imap(f, pending)
retry_results = []
while len(successful) < total_items:
successful.extend([r for r in results if not r is None])
successful.extend([r for r in retry_results if not r is None])
failed_items = []
while not q.empty():
failed_items.append(q.get())
if failed_items:
failure_tracker.append(failed_items)
retry_results = p.imap(f, failed_items);
p.close()
p.join()
print "Results: %s" % successful
print "Failures: %s" % failure_tracker
if __name__ == '__main__':
main(range(1, 10))
ผลลัพธ์จะเป็นดังนี้:
Results: [1, 4, 36, 49, 25, 81, 16, 64, 9]
Failures: [[3, 4, 5, 8, 9], [3, 8, 4], [8, 3], []]
A Pool
ไม่สามารถแชร์ระหว่างหลายกระบวนการได้ ดังนั้นแนวทางที่ใช้ Queue
นี้ หากคุณพยายามส่งพูลเป็นพารามิเตอร์ไปยังกระบวนการของพูล คุณจะได้รับข้อผิดพลาดนี้:
NotImplementedError: pool objects cannot be passed between processes or pickled
คุณสามารถลองอีกครั้งทันทีภายในฟังก์ชัน f
ของคุณเพื่อหลีกเลี่ยงค่าใช้จ่ายในการซิงโครไนซ์ จริงๆ แล้วมันเป็นเรื่องของเวลาที่ฟังก์ชันของคุณควรรอเพื่อลองอีกครั้ง และโอกาสที่จะประสบความสำเร็จหากลองใหม่ทันที
คำตอบเก่า: เพื่อความสมบูรณ์ นี่คือคำตอบเก่าของฉัน ซึ่งไม่เหมาะสมเท่ากับการส่งเข้าสู่กลุ่มโดยตรงอีกครั้ง แต่อาจยังคงมีความเกี่ยวข้อง ขึ้นอยู่กับกรณีการใช้งาน เนื่องจากเป็นวิธีธรรมชาติในการจัดการ/จำกัดการลองใหม่ระดับ n
:
คุณสามารถใช้ Queue
เพื่อรวมความล้มเหลวและส่งใหม่อีกครั้งเมื่อสิ้นสุดการทดสอบแต่ละครั้ง โดยดำเนินการหลายครั้ง:
import multiprocessing as mp
import random
def f(x):
if random.getrandbits(1):
# on failure / exception catch
f.q.put(x)
return None
return x*x
def f_init(q):
f.q = q
def main(pending):
run_number = 1
while pending:
jobs = pending
pending = []
q = mp.Queue()
p = mp.Pool(None, f_init, [q])
results = p.imap(f, jobs)
p.close()
p.join()
failed_items = []
while not q.empty():
failed_items.append(q.get())
successful = [r for r in results if not r is None]
print "(%d) Succeeded: %s" % (run_number, successful)
print "(%d) Failed: %s" % (run_number, failed_items)
print
pending = failed_items
run_number += 1
if __name__ == '__main__':
main(range(1, 10))
ด้วยผลลัพธ์ดังนี้:
(1) Succeeded: [9, 16, 36, 81]
(1) Failed: [2, 1, 5, 7, 8]
(2) Succeeded: [64]
(2) Failed: [2, 1, 5, 7]
(3) Succeeded: [1, 25]
(3) Failed: [2, 7]
(4) Succeeded: [49]
(4) Failed: [2]
(5) Succeeded: [4]
(5) Failed: []
person
Preet Kukreti
schedule
24.07.2012
return f(x)
แทนที่จะเพิ่มValueError
? แค่คาดเดา... - person Paulo Freitas   schedule 24.07.2012