class_docstr: Basic implementation of StorageWriter using FFspec.
  
  This implementation makes the following assumptions and simplifications:
  
  * The checkpoint path is an empty or non-existing directory.
  * File creation is atomic
  
  The checkpoint consist of one file per write request plus
  a `.metadata` file with the serialized metadata.
class_docstr_tok: ["Basic","implementation","of","StorageWriter","using","FFspec",".","This","implementation","makes","the","following","assumptions","and","simplifications",":","*","The","checkpoint","path","is","an","empty","or","non-existing","directory",".","*","File","creation","is","atomic","The","checkpoint","consist","of","one","file","per","write","request","plus","a","`",".metadata","`","file","with","the","serialized","metadata","."]
code: def write_data(
      self,
      plan: SavePlan,
      planner: SavePlanner,
  ) -> Future[List[WriteResult]]:
      storage_plan: _StoragePrefix = plan.storage_data
      file_count = 0
  
      def gen_file():
          nonlocal file_count
          file_name = (
              f"{storage_plan.prefix}{file_count}{DEFAULT_SUFFIX}"
          )
          file_count += 1
          return file_name
  
      file_queue: queue.Queue = queue.Queue()
      if self.single_file_per_rank:
          for bucket in _split_by_size_and_type(
              self.thread_count, plan.items
          ):
              file_name = gen_file()
              file_path = os.path.join(self.path, file_name)
              file_queue.put((file_path, file_name, bucket))
      else:
          for item in plan.items:
              file_name = gen_file()
              file_path = os.path.join(self.path, file_name)
              file_queue.put((file_path, file_name, [item]))
  
      result_queue: queue.Queue = queue.Queue()
  
      threads = []
      for _ in range(1, self.thread_count):
          t = threading.Thread(
              target=_write_files_from_queue,
              args=(
                  file_queue,
                  result_queue,
                  planner,
                  self.per_thread_copy_ahead,
                  self.fs,
              ),
          )
          t.start()
          threads.append(t)
  
      _write_files_from_queue(
          file_queue=file_queue,
          result_queue=result_queue,
          planner=planner,
          inflight_threshhold=self.per_thread_copy_ahead,
          fs=self.fs,
      )
  
      for t in threads:
          t.join()
  
      res = []
      try:
          while True:
              res += result_queue.get_nowait()
      except queue.Empty:
          pass
  
          fut: Future[List[WriteResult]] = Future()
          fut.set_result(res)
          return fut
  
imports: import collections
  import dataclasses
  import io
  import os
  import pickle
  import queue
  import threading
  from abc import ABC, abstractmethod
  from dataclasses import dataclass
  from typing import Callable, cast, Dict, Iterable, Iterator, List, Optional, Tuple, Union
  import fsspec
  from fsspec import AbstractFileSystem
  from fsspec.core import url_to_fs
  import torch
  from torch import Tensor
  from torch._utils import _get_device_module
  from torch.distributed._shard._utils import narrow_tensor_by_index
  from torch.distributed.checkpoint.metadata import Metadata, MetadataIndex
  from torch.distributed.checkpoint.planner import LoadItemType, LoadPlan, LoadPlanner, ReadItem, SavePlan, SavePlanner, WriteItem, WriteItemType
  from torch.distributed.checkpoint.storage import StorageReader, StorageWriter, WriteResult
  from torch.distributed.checkpoint.utils import _create_file_view
  from torch.futures import Future
label_desc: Use image node_id 5 for calling the FsspecWriter obj's underlying member method code with example usage: obj.write_data(plan, planner) and returns: file_name, fut