class_docstr: Instance storage stores the states of instances in the storage. It also
  allows users to subscribe to instance status changes to trigger reconciliation
  with cloud provider.
docstr: Upsert instances into the storage. If the instance already exists,
  it will be updated. Otherwise, it will be inserted. If the
  expected_storage_version is specified, the update will fail if the
  current storage version does not match the expected version.
  
  Note the version of the upserted instances will be set to the current
  storage version.
  
  Args:
      updates: A list of instances to be upserted.
      expected_storage_version: The expected storage version.
  
  Returns:
      StoreStatus: A tuple of (success, storage_version).
docstr_tok: ["Upsert","instances","into","the","storage",".","If","the","instance","already","exists",",","it","will","be","updated",".","Otherwise",",","it","will","be","inserted",".","If","the","expected_storage_version","is","specified",",","the","update","will","fail","if","the","current","storage","version","does","not","match","the","expected","version",".","Note","the","version","of","the","upserted","instances","will","be","set","to","the","current","storage","version",".","Args",":","updates",":","A","list","of","instances","to","be","upserted",".","expected_storage_version",":","The","expected","storage","version",".","Returns",":","StoreStatus",":","A","tuple","of","(","success",",","storage_version",")","."]
code: def batch_upsert_instances(
      self,
      updates: List[Instance],
      expected_storage_version: Optional[int] = None,
  ) -> StoreStatus:
      """Upsert instances into the storage. If the instance already exists,
      it will be updated. Otherwise, it will be inserted. If the
      expected_storage_version is specified, the update will fail if the
      current storage version does not match the expected version.
  
      Note the version of the upserted instances will be set to the current
      storage version.
  
      Args:
          updates: A list of instances to be upserted.
          expected_storage_version: The expected storage version.
  
      Returns:
          StoreStatus: A tuple of (success, storage_version).
      """
      mutations = {}
      version = self._storage.get_version()
      # handle version mismatch
      if (
          expected_storage_version
          and expected_storage_version != version
      ):
          return StoreStatus(False, version)
  
      for instance in updates:
          instance = copy.deepcopy(instance)
          # the instance version is set to 0, it will be
          # populated by the storage entry's verion on read
          instance.version = 0
          mutations[instance.instance_id] = instance.SerializeToString()
  
      result, version = self._storage.batch_update(
          self._table_name, mutations, {}, expected_storage_version
      )
  
      if result:
          for subscriber in self._status_change_subscribers:
              subscriber.notify(
                  [
                      InstanceUpdateEvent(
                          instance_id=instance.instance_id,
                          new_instance_status=instance.status,
                      )
                      for instance in updates
                  ],
              )
  
      return StoreStatus(result, version)
  
file_path: ray/python/ray/autoscaler/v2/instance_manager/instance_storage.py
imports: import copy
  import logging
  from abc import ABCMeta, abstractmethod
  from typing import Dict, List, Optional, Set, Tuple
  from ray.autoscaler.v2.instance_manager.storage import Storage, StoreStatus
  from ray.core.generated.instance_manager_pb2 import Instance, InstanceUpdateEvent
label_desc: Use image node_id 3 for calling the InstanceStorage obj's underlying member method code with example usage: obj.batch_upsert_instances(updates, expected_storage_version) and returns: StoreStatus, StoreStatus