code: def test_make_async_gen():
      """Tests that make_async_gen overlaps compute."""
  
      num_items = 10
  
      def gen(base_iterator):
          for i in base_iterator:
              time.sleep(2)
              yield i
  
      def sleep_udf(item):
          time.sleep(3)
          return item
  
      iterator = make_async_gen(
          base_iterator=iter(range(num_items)), fn=gen, num_workers=1
      )
  
      start_time = time.time()
      outputs = []
      for item in iterator:
          print(item)
          outputs.append(sleep_udf(item))
      end_time = time.time()
  
      assert outputs == list(range(num_items))
  
      # Three second buffer.
      assert end_time - start_time < num_items * 3 + 3
  
imports: import threading
  import time
  import numpy
  import pandas
  import pyarrow
  import pytest
  import ray
  from ray.data._internal.block_batching.interfaces import Batch
  from ray.data._internal.block_batching.util import _calculate_ref_hits, blocks_to_batches, collate, finalize_batches, format_batches, resolve_block_refs
  from ray.data._internal.util import Queue, make_async_gen
label_desc: Use image node_id 8 for calling a global function with example usage: test_make_async_gen() and returns: item