code: def test_map(client):
      with client.get_executor() as e:
          N = 10
          it = e.map(inc, range(N))
          expected = set(range(1, N + 1))
          for x in it:
              expected.remove(x)
          assert not expected
  
      with client.get_executor(pure=False) as e:
          N = 10
          it = e.map(slowinc, range(N), [0.3] * N, timeout=1.2)
          results = []
          with pytest.raises(TimeoutError):
              for x in it:
                  results.append(x)
          assert 2 <= len(results) < 7
  
      with client.get_executor(pure=False) as e:
          N = 10
          # Not consuming the iterator will cancel remaining tasks
          it = e.map(slowinc, range(N), [0.3] * N)
          for _ in take(2, it):
              pass
          # Some tasks still processing
          assert number_of_processing_tasks(client) > 0
          # Garbage collect the iterator => remaining tasks are cancelled
          del it
          sleep(0.5)
          assert number_of_processing_tasks(client) == 0
  
imports: from __future__ import annotations
  import random
  from concurrent.futures import FIRST_COMPLETED, FIRST_EXCEPTION, Future, TimeoutError, as_completed, wait
  from time import sleep
  import pytest
  from tlz import take
  from distributed.event import Event
  from distributed.metrics import time
  from distributed.utils import CancelledError
  from distributed.utils_test import inc, slowadd, slowinc, throws, varying
label_desc: Use image node_id 8 for calling a global function with example usage: test_map(client) without return types