code: def _create_node(self, node_config, tags, count):
      created_nodes_dict = {}
      exception_happened = False
      frozen_vm_obj = self.create_new_or_fetch_existing_frozen_vms(
          node_config
      )
  
      # The nodes are named as follows:
      # ray-<cluster-name>-head-<uuid> for the head node
      # ray-<cluster-name>-worker-<uuid> for the worker nodes
      vm_names = [
          "{}-{}".format(tags[TAG_RAY_NODE_NAME], str(uuid.uuid4()))
          for _ in range(count)
      ]
  
      requested_gpu_num = 0
      if "resources" in node_config:
          resources = node_config["resources"]
          requested_gpu_num = resources.get("GPU", 0)
      vm_2_gpu_cards_map = {}
      gpu_cards_map_array = []
  
      is_dynamic = is_dynamic_passthrough(node_config)
  
      if requested_gpu_num > 0:
          # Fetch all available frozen-vm + gpu-cards info into
          # `get_vm_2_gpu_cards_map`
          if "resource_pool" in node_config["frozen_vm"]:
              # This means that we have multiple frozen_vms, and we need to gather
              # the information of the GPUs of each frozen VM's ESXi host.
              vm_2_gpu_cards_map = get_vm_2_gpu_cards_map(
                  self.get_pyvmomi_sdk_provider(),
                  node_config["frozen_vm"]["resource_pool"],
                  requested_gpu_num,
                  is_dynamic,
              )
          else:
              # This means that we have only one frozen VM, we just need to put the
              # information of the only ESXi host's GPU info into the map
              gpu_cards = get_gpu_cards_from_vm(
                  frozen_vm_obj, requested_gpu_num, is_dynamic
              )
              vm_2_gpu_cards_map[frozen_vm_obj.name] = gpu_cards
  
          # Split `vm_2_gpu_ids_map` for nodes, check the comments inside function
          # split_vm_2_gpu_ids_map to get to know why we need do this.
          gpu_cards_map_array = split_vm_2_gpu_cards_map(
              vm_2_gpu_cards_map, requested_gpu_num
          )
          if len(gpu_cards_map_array) < count:
              logger.warning(
                  f"The GPU card number cannot fulfill {count} Ray nodes, "
                  f"but can fulfill {len(gpu_cards_map_array)} Ray nodes. "
                  f"gpu_cards_map_array: {gpu_cards_map_array}"
              )
              # Avoid invalid index when accessing gpu_cards_map_array[i]
              set_gpu_placeholder(
                  gpu_cards_map_array, count - len(gpu_cards_map_array)
              )
      else:
          # CPU node: Avoid invalid index when accessing gpu_cards_map_array[i]
          set_gpu_placeholder(gpu_cards_map_array, count)
  
      def get_frozen_vm_name():
          if self.frozen_vm_scheduler:
              return self.frozen_vm_scheduler.next_frozen_vm().name
          else:
              return frozen_vm_obj.name
  
      with ThreadPoolExecutor(max_workers=count) as executor:
          futures = [
              executor.submit(
                  self.create_instant_clone_node,
                  get_frozen_vm_name(),
                  vm_names[i],
                  node_config,
                  tags,
                  gpu_cards_map_array[i],
              )
              for i in range(count)
          ]
      failed_vms_index = []
      for i in range(count):
          future = futures[i]
          try:
              vm = future.result()
              k = Constants.VSPHERE_NODE_STATUS
              v = Constants.VsphereNodeStatus.CREATED.value
              vsphere_node_created_tag = {k: v}
              # if create succeed, we add a "created" tag
              self.set_node_tags(vm.vm, vsphere_node_created_tag)
              created_nodes_dict[vm.name] = vm
              logger.info(f"VM {vm.name} is created.")
          except Exception as e:
              logger.error(
                  "Exception occurred while creating or tagging VMs {}".format(
                      e
                  )
              )
              exception_happened = True
              failed_vms_index.append(i)
              logger.error(f"Failed creating VM {vm_names[i]}")
  
      # We clean up the created VMs if any exception occurs to them
      if exception_happened:
          with ThreadPoolExecutor(max_workers=count) as executor:
              futures = [
                  executor.submit(
                      self.get_vsphere_sdk_provider().delete_vm_by_name,
                      vm_names[i],
                  )
                  for i in failed_vms_index
              ]
          for future in futures:
              _ = future.result()
  
          if len(failed_vms_index) == count:
              raise RuntimeError("Failed creating all VMs, exiting!")
  
      return created_nodes_dict
  
imports: import copy
  import logging
  import threading
  import uuid
  from concurrent.futures import ThreadPoolExecutor
  from typing import Any, Dict
  from pyVmomi import vim
  from ray.autoscaler._private.vsphere.config import bootstrap_vsphere, is_dynamic_passthrough
  from ray.autoscaler._private.vsphere.gpu_utils import get_gpu_cards_from_vm, get_vm_2_gpu_cards_map, plug_gpu_cards_to_vm, set_gpu_placeholder, split_vm_2_gpu_cards_map
  from ray.autoscaler._private.vsphere.pyvmomi_sdk_provider import PyvmomiSdkProvider
  from ray.autoscaler._private.vsphere.scheduler import SchedulerFactory
  from ray.autoscaler._private.vsphere.utils import Constants, now_ts
  from ray.autoscaler._private.vsphere.vsphere_sdk_provider import VsphereSdkProvider
  from ray.autoscaler.node_provider import NodeProvider
  from ray.autoscaler.tags import TAG_RAY_CLUSTER_NAME, TAG_RAY_NODE_NAME
label_desc: Use image node_id 19 for calling the VsphereNodeProvider obj's underlying member method code with example usage: obj._create_node(node_config, tags, count) and returns: created_nodes_dict, self, frozen_vm_obj