code: def _copy_files(
      source_path: pathlib.Path,
      destination_path: pathlib.Path,
      fs: Optional[AbstractFileSystem] = None,
  ) -> None:
      """Copy files from one path to another.
  
      The source path must either be an existing file or folder. If the source is a folder, the destination path is
      interpreted as a folder as well. If the source is a file, the destination path is interpreted as a file too.
  
      Files in a folder are copied recursively and efficiently using multiple threads.
  
      """
      if fs is None:
          fs = _filesystem()
  
      def _copy(
          from_path: pathlib.Path, to_path: pathlib.Path
      ) -> Optional[Exception]:
          _logger.debug(f"Copying {str(from_path)} -> {str(to_path)}")
  
          try:
              # NOTE: S3 does not have a concept of directories, so we do not need to create one.
              if isinstance(fs, LocalFileSystem):
                  fs.makedirs(str(to_path.parent), exist_ok=True)
  
              fs.put(str(from_path), str(to_path), recursive=False)
          except Exception as ex:
              # Return the exception so that it can be handled in the main thread
              return ex
  
      # NOTE: Cannot use `S3FileSystem.put(recursive=True)` because it tries to access parent directories
      #       which it does not have access to.
      if source_path.is_dir():
          src = [
              file for file in source_path.rglob("*") if file.is_file()
          ]
          dst = [
              destination_path / file.relative_to(source_path)
              for file in src
          ]
  
          with concurrent.futures.ThreadPoolExecutor(
              num_workers
          ) as executor:
              results = executor.map(_copy, src, dst)
  
          # Raise the first exception found
          exception = next(
              (e for e in results if isinstance(e, Exception)), None
          )
          if exception:
              raise exception
      else:
          if isinstance(fs, LocalFileSystem):
              fs.makedirs(str(destination_path.parent), exist_ok=True)
  
          fs.put(str(source_path), str(destination_path))
  
file_path: pytorch-lightning/src/lightning/app/storage/copier.py
imports: import concurrent.futures
  import pathlib
  import threading
  from threading import Thread
  from time import time
  from typing import TYPE_CHECKING, Optional, Union
  from fsspec import AbstractFileSystem
  from fsspec.implementations.local import LocalFileSystem
  from lightning.app.core.queues import BaseQueue
  from lightning.app.storage.path import _filesystem
  from lightning.app.storage.requests import _ExistsRequest, _GetRequest
  from lightning.app.utilities.app_helpers import Logger
label_desc: Use image node_id 2 for calling a global function with example usage: _copy_files(source_path, destination_path, fs) and returns: ex