def map(self, fn, *iterables, timeout=None, prefetch=None):<EOL>
if timeout is not None:<EOL><INDENT>end_time = timeout + time.time()<EOL><DEDENT>if prefetch is None:<EOL><INDENT>prefetch = self._max_workers<EOL><DEDENT>if prefetch < <NUM_LIT:0>:<EOL><INDENT>raise ValueError('<STR_LIT>')<EOL><DEDENT>if len(iterables) > <NUM_LIT:1>:<EOL><INDENT>argsiter = zip(*iterables)<EOL><DEDENT>else:<EOL><INDENT>argsiter = iter(iterables[<NUM_LIT:0>])<EOL><DEDENT>fs = collections.deque(<EOL>self.submit(fn, *args)<EOL>for args in itertools.islice(argsiter, self._max_workers +<EOL>prefetch))<EOL>def _result_iterator():<EOL><INDENT>nonlocal argsiter<EOL>try:<EOL><INDENT>while fs:<EOL><INDENT>res = fs[<NUM_LIT:0>].result() if timeout is None else fs[<NUM_LIT:0>].result(<EOL>end_time - time.time())<EOL>del fs[<NUM_LIT:0>]<EOL>if argsiter:<EOL><INDENT>try:<EOL><INDENT>args = next(argsiter)<EOL><DEDENT>except StopIteration:<EOL><INDENT>argsiter = None<EOL><DEDENT>else:<EOL><INDENT>fs.append(self.submit(fn, *args))<EOL><DEDENT><DEDENT>yield res<EOL><DEDENT><DEDENT>finally:<EOL><INDENT>for future in fs:<EOL><INDENT>future.cancel()<EOL><DEDENT><DEDENT><DEDENT>return _result_iterator()<EOL>
Lazy apdaption of ThreadPoolExecutor.map.

        Unlike ThreadPoolExecutor.map:
        - iterables are prefetched lazily
        - if only a single iterable is specified, iter(iterables[0]) is used
          instead of zip(*iterables) to obtain a iterator over the arguments
          that are mapped to fn. This is to match the behavior of
          mxnet.gluon.Dataset.transform and gluonnlp.data.DataStream.transform
          which unpack argument tuples.