code: def _format_in_threadpool(
      batch_iter: Iterator[Batch],
      stats: DatasetStats,
      batch_format: Optional[str],
      collate_fn: Optional[Callable[[DataBatch], Any]],
      num_threadpool_workers: int,
  ) -> Iterator[Batch]:
      """Executes the batching, formatting, and collation logic in a threadpool.
  
      Args:
          logical_batch_iterator: An iterator over logical batches.
          stats: DatasetStats object to record timing and other statistics.
          batch_format: The format in which to return each batch.
              Specify "default" to use the current block format (promoting
              Arrow to pandas automatically), "pandas" to
              select ``pandas.DataFrame`` or "pyarrow" to select
              ``pyarrow.Table``, or None to use entire blocks
              as batches.
          collate_fn: A function to apply to each data batch before returning it.
          num_threadpool_workers: The number of threads to use in the threadpool.
      """
  
      def threadpool_computations_format_collate(
          batch_iter: Iterator[Batch],
      ) -> Iterator[Batch]:
          # Step 4a: Format the batches.
          formatted_batch_iter = format_batches(
              batch_iter, batch_format=batch_format, stats=stats
          )
  
          # Step 4b: Apply the collate function if applicable.
          if collate_fn is not None:
              formatted_batch_iter = collate(
                  formatted_batch_iter,
                  collate_fn=collate_fn,
                  stats=stats,
              )
          yield from formatted_batch_iter
  
      if num_threadpool_workers > 0:
          collated_iter = make_async_gen(
              base_iterator=batch_iter,
              fn=threadpool_computations_format_collate,
              num_workers=num_threadpool_workers,
          )
      else:
          collated_iter = threadpool_computations_format_collate(
              batch_iter
          )
      return collated_iter
  
file_path: ray/python/ray/data/_internal/block_batching/iter_batches.py
imports: import collections
  from contextlib import nullcontext
  from typing import Any, Callable, Dict, Iterator, Optional, Tuple
  import ray
  from ray.data._internal.block_batching.interfaces import Batch, BlockPrefetcher
  from ray.data._internal.block_batching.util import ActorBlockPrefetcher, WaitBlockPrefetcher, blocks_to_batches, collate, extract_data_from_batch, finalize_batches, format_batches, resolve_block_refs
  from ray.data._internal.memory_tracing import trace_deallocation
  from ray.data._internal.stats import DatasetStats
  from ray.data._internal.util import make_async_gen
  from ray.data.block import Block, BlockMetadata, DataBatch
  from ray.data.context import DataContext
  from ray.types import ObjectRef
label_desc: Use image node_id 2 for calling a global function with example usage: _format_in_threadpool(batch_iter, stats, batch_format, collate_fn, num_threadpool_workers) and returns: collated_iter