code: def save_bytes(
      self, path_and_bytes_iter, overwrite=False, len_hint=0
  ):
      tmpdir = None
      try:
          tmpdir = mkdtemp(
              dir=ARTIFACT_LOCALROOT, prefix="metaflow.gs.save_bytes."
          )
          futures = []
          for path, byte_stream in path_and_bytes_iter:
              metadata = None
              # bytes_stream could actually be (bytes_stream, metadata) instead.
              # Read the small print on DatastoreStorage.save_bytes()
              if isinstance(byte_stream, tuple):
                  byte_stream, metadata = byte_stream
              tmp_filename = os.path.join(tmpdir, str(uuid.uuid4()))
              with open(tmp_filename, "wb") as f:
                  f.write(byte_stream.read())
              # Fully finish writing the file, before submitting work. Careful with indentation.
  
              futures.append(
                  self._executor.submit(
                      self.root_client.save_bytes_single,
                      (path, tmp_filename, metadata),
                      overwrite=overwrite,
                  )
              )
          for future in as_completed(futures):
              future.result()
      finally:
          # *Future* improvement: We could clean up individual tmp files as each future completes
          if tmpdir and os.path.exists(tmpdir):
              shutil.rmtree(tmpdir)
  
imports: import json
  import os
  import shutil
  import uuid
  from concurrent.futures import as_completed
  from tempfile import mkdtemp
  from metaflow.datastore.datastore_storage import DataStoreStorage, CloseAfterUse
  from metaflow.exception import MetaflowInternalError
  from metaflow.metaflow_config import DATASTORE_SYSROOT_GS, ARTIFACT_LOCALROOT, GS_STORAGE_WORKLOAD_TYPE
  from metaflow.plugins.gcp.gs_storage_client_factory import get_gs_storage_client
  from metaflow.plugins.gcp.gs_utils import check_gs_deps, parse_gs_full_path, process_gs_exception
  from metaflow.plugins.storage_executor import StorageExecutor, handle_executor_exceptions
label_desc: Use image node_id 8 for calling the GSStorage obj's underlying member method code with example usage: obj.save_bytes(path_and_bytes_iter, overwrite, len_hint) without return types