code: def create_node(
      self,
      node_config: Dict[str, Any],
      tags: Dict[str, str],
      count: int,
  ) -> Optional[Dict[str, Any]]:
      filter_tags = [
          {
              "Key": TAG_RAY_CLUSTER_NAME,
              "Value": self.cluster_name,
          },
          {"Key": TAG_RAY_NODE_KIND, "Value": tags[TAG_RAY_NODE_KIND]},
          {
              "Key": TAG_RAY_USER_NODE_TYPE,
              "Value": tags[TAG_RAY_USER_NODE_TYPE],
          },
          {
              "Key": TAG_RAY_LAUNCH_CONFIG,
              "Value": tags[TAG_RAY_LAUNCH_CONFIG],
          },
          {"Key": TAG_RAY_NODE_NAME, "Value": tags[TAG_RAY_NODE_NAME]},
      ]
  
      reused_nodes_dict = {}
      if self.cache_stopped_nodes:
          reuse_nodes_candidate = self.acs.describe_instances(
              tags=filter_tags
          )
          if reuse_nodes_candidate:
              with cli_logger.group("Stopping instances to reuse"):
                  reuse_node_ids = []
                  for node in reuse_nodes_candidate:
                      node_id = node.get("InstanceId")
                      status = node.get("Status")
                      if status != STOPPING and status != STOPPED:
                          continue
                      if status == STOPPING:
                          # wait for node stopped
                          while (
                              self.acs.describe_instances(
                                  instance_ids=[node_id]
                              )[0].get("Status")
                              == STOPPING
                          ):
                              logging.info("wait for %s stop" % node_id)
                              time.sleep(STOPPING_NODE_DELAY)
                      # logger.info("reuse %s" % node_id)
                      reuse_node_ids.append(node_id)
                      reused_nodes_dict[node.get("InstanceId")] = node
                      self.acs.start_instance(node_id)
                      self.tag_cache[node_id] = node.get("Tags")
                      self.set_node_tags(node_id, tags)
                      if len(reuse_node_ids) == count:
                          break
              count -= len(reuse_node_ids)
  
      created_nodes_dict = {}
      if count > 0:
          filter_tags.append(
              {
                  "Key": TAG_RAY_NODE_STATUS,
                  "Value": tags[TAG_RAY_NODE_STATUS],
              }
          )
          instance_id_sets = self.acs.run_instances(
              instance_type=node_config["InstanceType"],
              image_id=node_config["ImageId"],
              tags=filter_tags,
              amount=count,
              vswitch_id=self.provider_config["v_switch_id"],
              security_group_id=self.provider_config[
                  "security_group_id"
              ],
              key_pair_name=self.provider_config["key_name"],
          )
          instances = self.acs.describe_instances(
              instance_ids=instance_id_sets
          )
  
          if instances is not None:
              for instance in instances:
                  created_nodes_dict[
                      instance.get("InstanceId")
                  ] = instance
  
      all_created_nodes = reused_nodes_dict
      all_created_nodes.update(created_nodes_dict)
      return all_created_nodes
  
imports: import logging
  import random
  import threading
  import time
  from collections import defaultdict
  from typing import Any, Dict, List, Optional
  from ray.autoscaler._private.aliyun.config import PENDING, RUNNING, STOPPED, STOPPING, bootstrap_aliyun
  from ray.autoscaler._private.aliyun.utils import AcsClient
  from ray.autoscaler._private.cli_logger import cli_logger
  from ray.autoscaler._private.constants import BOTO_MAX_RETRIES
  from ray.autoscaler._private.log_timer import LogTimer
  from ray.autoscaler.node_provider import NodeProvider
  from ray.autoscaler.tags import TAG_RAY_CLUSTER_NAME, TAG_RAY_LAUNCH_CONFIG, TAG_RAY_NODE_KIND, TAG_RAY_NODE_NAME, TAG_RAY_NODE_STATUS, TAG_RAY_USER_NODE_TYPE
label_desc: Use image node_id 11 for calling the AliyunNodeProvider obj's underlying member method code with example usage: obj.create_node(node_config, tags, count) and returns: all_created_nodes