code: def cluster_setup(ray_start_cluster_head: Cluster):
      # Sets up a cluster with 4 nodes: head node + 3 workers
      cluster = ray_start_cluster_head
      nodes = []
      nodes.append(
          cluster.add_node(resources={"worker1": 1, "coordinator": 1})
      )
      nodes.append(cluster.add_node(resources={"worker2": 1, "cpu": 1}))
      nodes.append(cluster.add_node(resources={"worker3": 1, "cpu": 1}))
      cluster.wait_for_nodes()
  
      @ray.remote
      def get_node_id():
          return ray.get_runtime_context().get_node_id()
  
      worker1_node_id = ray.get(
          get_node_id.options(resources={"worker1": 1}).remote()
      )
      worker2_node_id = ray.get(
          get_node_id.options(resources={"worker2": 1}).remote()
      )
      worker3_node_id = ray.get(
          get_node_id.options(resources={"worker3": 1}).remote()
      )
      wait_for_condition(
          lambda: len(
              {
                  node["NodeID"]
                  for node in ray.nodes()
                  if (node["Alive"])
              }
          )
          == 4
      )
  
      yield cluster, nodes, [
          worker1_node_id,
          worker2_node_id,
          worker3_node_id,
      ]
  
imports: import gc
  import threading
  import time
  from tempfile import TemporaryDirectory
  import pytest
  import ray
  from ray import train, tune
  from ray._private.test_utils import wait_for_condition
  from ray._raylet import GcsClient
  from ray.cluster_utils import Cluster
  from ray.core.generated import autoscaler_pb2
  from ray.train import Checkpoint, FailureConfig, RunConfig, ScalingConfig
  from ray.train.data_parallel_trainer import DataParallelTrainer
  from ray.train.trainer import BaseTrainer, TrainingFailedError
  from ray.tune import Tuner, TuneConfig, TuneError
  from ray.tests.conftest import *
  from ray.train.tests.util import create_dict_checkpoint, load_dict_checkpoint
label_desc: Use image node_id 3 for calling a global function with example usage: cluster_setup(ray_start_cluster_head) and returns: ray