Returns an iterator equivalent to map(fn, iter). Args: fn: A callable that will take as many arguments as there are passed iterables. timeout: The maximum number of seconds to wait. If None, then there is no limit on the wait time.
(self, fn, *iterables, timeout=None, chunksize=1, buffersize=None)
| 602 | raise NotImplementedError() |
| 603 | |
| 604 | def map(self, fn, *iterables, timeout=None, chunksize=1, buffersize=None): |
| 605 | """Returns an iterator equivalent to map(fn, iter). |
| 606 | |
| 607 | Args: |
| 608 | fn: A callable that will take as many arguments as there are |
| 609 | passed iterables. |
| 610 | timeout: The maximum number of seconds to wait. If None, then there |
| 611 | is no limit on the wait time. |
| 612 | chunksize: The size of the chunks the iterable will be broken into |
| 613 | before being passed to a child process. This argument is only |
| 614 | used by ProcessPoolExecutor; it is ignored by |
| 615 | ThreadPoolExecutor. |
| 616 | buffersize: The number of submitted tasks whose results have not |
| 617 | yet been yielded. If the buffer is full, iteration over the |
| 618 | iterables pauses until a result is yielded from the buffer. |
| 619 | If None, all input elements are eagerly collected, and a task is |
| 620 | submitted for each. |
| 621 | |
| 622 | Returns: |
| 623 | An iterator equivalent to: map(func, *iterables) but the calls may |
| 624 | be evaluated out-of-order. |
| 625 | |
| 626 | Raises: |
| 627 | TimeoutError: If the entire result iterator could not be generated |
| 628 | before the given timeout. |
| 629 | Exception: If fn(*args) raises for any values. |
| 630 | """ |
| 631 | if buffersize is not None and not isinstance(buffersize, int): |
| 632 | raise TypeError("buffersize must be an integer or None") |
| 633 | if buffersize is not None and buffersize < 1: |
| 634 | raise ValueError("buffersize must be None or > 0") |
| 635 | |
| 636 | if timeout is not None: |
| 637 | end_time = timeout + time.monotonic() |
| 638 | |
| 639 | zipped_iterables = zip(*iterables) |
| 640 | if buffersize: |
| 641 | fs = collections.deque( |
| 642 | self.submit(fn, *args) for args in islice(zipped_iterables, buffersize) |
| 643 | ) |
| 644 | else: |
| 645 | fs = [self.submit(fn, *args) for args in zipped_iterables] |
| 646 | |
| 647 | # Use a weak reference to ensure that the executor can be garbage |
| 648 | # collected independently of the result_iterator closure. |
| 649 | executor_weakref = weakref.ref(self) |
| 650 | |
| 651 | # Yield must be hidden in closure so that the futures are submitted |
| 652 | # before the first iterator value is required. |
| 653 | def result_iterator(): |
| 654 | try: |
| 655 | # reverse to keep finishing order |
| 656 | fs.reverse() |
| 657 | while fs: |
| 658 | if ( |
| 659 | buffersize |
| 660 | and (executor := executor_weakref()) |
| 661 | and (args := next(zipped_iterables, None)) |