Args: get_df ( -> DataFlow): a callable which returns a DataFlow. Each thread will call this function to get the DataFlow to use. Therefore do not return the same DataFlow object for each call, unless your dataflow is stateless.
(self, get_df, num_prefetch, num_thread)
| 430 | self.stop() |
| 431 | |
| 432 | def __init__(self, get_df, num_prefetch, num_thread): |
| 433 | """ |
| 434 | Args: |
| 435 | get_df ( -> DataFlow): a callable which returns a DataFlow. |
| 436 | Each thread will call this function to get the DataFlow to use. |
| 437 | Therefore do not return the same DataFlow object for each call, |
| 438 | unless your dataflow is stateless. |
| 439 | num_prefetch (int): size of the queue |
| 440 | num_thread (int): number of threads |
| 441 | """ |
| 442 | assert num_thread > 0, num_thread |
| 443 | assert num_prefetch > 0, num_prefetch |
| 444 | self.num_thread = num_thread |
| 445 | self.queue = queue.Queue(maxsize=num_prefetch) |
| 446 | self.threads = [ |
| 447 | MultiThreadRunner._Worker(get_df, self.queue) |
| 448 | for _ in range(num_thread)] |
| 449 | |
| 450 | try: |
| 451 | self._size = self.__len__() |
| 452 | except NotImplementedError: |
| 453 | self._size = -1 |
| 454 | |
| 455 | def reset_state(self): |
| 456 | for th in self.threads: |