| 728 | raise e |
| 729 | |
| 730 | def pipe_reader(): |
| 731 | conns = [] |
| 732 | for reader in readers: |
| 733 | parent_conn, child_conn = fork_context.Pipe() |
| 734 | conns.append(parent_conn) |
| 735 | p = fork_context.Process( |
| 736 | target=_read_into_pipe, args=(reader, child_conn) |
| 737 | ) |
| 738 | p.start() |
| 739 | |
| 740 | reader_num = len(readers) |
| 741 | finish_num = 0 |
| 742 | conn_to_remove = [] |
| 743 | while finish_num < reader_num: |
| 744 | for conn in conn_to_remove: |
| 745 | conns.remove(conn) |
| 746 | conn_to_remove = [] |
| 747 | for conn in conns: |
| 748 | sample = json.loads(conn.recv()) |
| 749 | if sample is None: |
| 750 | finish_num += 1 |
| 751 | conn.close() |
| 752 | conn_to_remove.append(conn) |
| 753 | elif sample == "": |
| 754 | conn.close() |
| 755 | conn_to_remove.append(conn) |
| 756 | raise ValueError( |
| 757 | "multiprocess_reader failed to send data into the multiprocessing.Pipe." |
| 758 | ) |
| 759 | else: |
| 760 | yield sample |
| 761 | |
| 762 | if use_pipe: |
| 763 | return pipe_reader |