Convert a DataFlow to a :class:`multiprocessing.Queue`. The DataFlow will only be reset in the spawned process. Args: df (DataFlow): the DataFlow to dump. size (int): size of the queue nr_consumer (int): number of consumer of the queue. The producer
(df, size, nr_consumer)
| 161 | |
| 162 | # for internal use only |
| 163 | def dump_dataflow_to_process_queue(df, size, nr_consumer): |
| 164 | """ |
| 165 | Convert a DataFlow to a :class:`multiprocessing.Queue`. |
| 166 | The DataFlow will only be reset in the spawned process. |
| 167 | |
| 168 | Args: |
| 169 | df (DataFlow): the DataFlow to dump. |
| 170 | size (int): size of the queue |
| 171 | nr_consumer (int): number of consumer of the queue. |
| 172 | The producer will add this many of ``DIE`` sentinel to the end of the queue. |
| 173 | |
| 174 | Returns: |
| 175 | tuple(queue, process): |
| 176 | The process will take data from ``df`` and fill |
| 177 | the queue, once you start it. Each element in the queue is (idx, |
| 178 | dp). idx can be the ``DIE`` sentinel when ``df`` is exhausted. |
| 179 | """ |
| 180 | q = mp.Queue(size) |
| 181 | |
| 182 | class EnqueProc(mp.Process): |
| 183 | |
| 184 | def __init__(self, df, q, nr_consumer): |
| 185 | super(EnqueProc, self).__init__() |
| 186 | self.df = df |
| 187 | self.q = q |
| 188 | |
| 189 | def run(self): |
| 190 | self.df.reset_state() |
| 191 | try: |
| 192 | for idx, dp in enumerate(self.df): |
| 193 | self.q.put((idx, dp)) |
| 194 | finally: |
| 195 | for _ in range(nr_consumer): |
| 196 | self.q.put((DIE, None)) |
| 197 | |
| 198 | proc = EnqueProc(df, q, nr_consumer) |
| 199 | return q, proc |
| 200 | |
| 201 | |
| 202 | if __name__ == '__main__': |