code: def test_for_each_concur_sync(ray_start_regular_shared):
      main_wait = Semaphore.remote(value=0)
      test_wait = Semaphore.remote(value=0)
  
      def task(x):
          i, main_wait, test_wait = x
          ray.get(main_wait.release.remote())
          ray.get(test_wait.acquire.remote())
          return i + 10
  
      @ray.remote(num_cpus=0.01)
      def to_list(it):
          return list(it)
  
      it = from_items(
          [(i, main_wait, test_wait) for i in range(8)], num_shards=2
      )
      it = it.for_each(
          task, max_concurrency=2, resources={"num_cpus": 0.01}
      )
  
      list_promise = to_list.remote(it.gather_sync())
  
      for i in range(4):
          assert i in [0, 1, 2, 3]
          ray.get(main_wait.acquire.remote())
  
      # There should be exactly 4 tasks executing at this point.
      assert (
          ray.get(main_wait.locked.remote()) is True
      ), "Too much parallelism"
  
      for i in range(8):
          ray.get(test_wait.release.remote())
  
      assert (
          repr(it)
          == "ParallelIterator[from_items[tuple, 8, shards=2].for_each()]"
      )
      result_list = ray.get(list_promise)
      assert set(result_list) == set(range(10, 18))
  
imports: import sys
  import time
  import collections
  from collections import Counter
  import pytest
  import ray
  from ray.util.iter import from_items, from_iterators, from_range, from_actors, ParallelIteratorWorker, LocalIterator
  from ray._private.test_utils import Semaphore
label_desc: Use image node_id 14 for calling a global function with example usage: test_for_each_concur_sync(ray_start_regular_shared) and returns: unknown, list