| 536 | out_queue.put(end) |
| 537 | |
| 538 | def xreader(): |
| 539 | in_queue = Queue(buffer_size) |
| 540 | out_queue = Queue(buffer_size) |
| 541 | out_order = [0] |
| 542 | # start a read worker in a thread |
| 543 | target = order_read_worker if order else read_worker |
| 544 | t = Thread(target=target, args=(reader, in_queue)) |
| 545 | t.daemon = True |
| 546 | t.start() |
| 547 | # start several handle_workers |
| 548 | target = order_handle_worker if order else handle_worker |
| 549 | args = ( |
| 550 | (in_queue, out_queue, mapper, out_order) |
| 551 | if order |
| 552 | else (in_queue, out_queue, mapper) |
| 553 | ) |
| 554 | workers = [] |
| 555 | for i in range(process_num): |
| 556 | worker = Thread(target=target, args=args) |
| 557 | worker.daemon = True |
| 558 | workers.append(worker) |
| 559 | for w in workers: |
| 560 | w.start() |
| 561 | |
| 562 | sample = out_queue.get() |
| 563 | while not isinstance(sample, XmapEndSignal): |
| 564 | yield sample |
| 565 | sample = out_queue.get() |
| 566 | finish = 1 |
| 567 | while finish < process_num: |
| 568 | sample = out_queue.get() |
| 569 | if isinstance(sample, XmapEndSignal): |
| 570 | finish += 1 |
| 571 | else: |
| 572 | yield sample |
| 573 | |
| 574 | return xreader |
| 575 | |