code: def test_concurrency(shutdown_only):
      ray.init(num_cpus=6)
      ds = ray.data.range(10, parallelism=10)
  
      def udf(x):
          return x
  
      class UDFClass:
          def __call__(self, x):
              return x
  
      # Test function and class.
      for fn in [udf, UDFClass]:
          # Test concurrency with None, single integer and a tuple of integers.
          for concurrency in [2, (2, 4)]:
              result = ds.map(fn, concurrency=concurrency).take_all()
              assert sorted(extract_values("id", result)) == list(
                  range(10)
              ), result
  
      # Test concurrency with an illegal value.
      error_message = "``concurrency`` is expected to be set a"
      for concurrency in ["dummy", (1, 3, 5)]:
          with pytest.raises(ValueError, match=error_message):
              ds.map(UDFClass, concurrency=concurrency).take_all()
  
      # Test concurrency not set.
      result = ds.map(udf).take_all()
      assert sorted(extract_values("id", result)) == list(
          range(10)
      ), result
      error_message = "``concurrency`` must be specified when using a callable class."
      with pytest.raises(ValueError, match=error_message):
          ds.map(UDFClass).take_all()
  
imports: import itertools
  import math
  import os
  import signal
  import threading
  import time
  from typing import Iterator
  import numpy
  import pandas
  import pyarrow
  import pyarrow.parquet
  import pytest
  import ray
  from ray._private.test_utils import wait_for_condition
  from ray.data.block import BlockAccessor
  from ray.data.context import DataContext
  from ray.data.tests.conftest import *
  from ray.data.tests.util import column_udf, column_udf_class, extract_values
  from ray.tests.conftest import *
label_desc: Use image node_id 5 for calling a global function with example usage: test_concurrency(shutdown_only) and returns: x, x