docstr: Once all the workers have written their own index, the merge function is responsible to read and merge them
  into a single index.
docstr_tok: ["Once","all","the","workers","have","written","their","own","index",",","the","merge","function","is","responsible","to","read","and","merge","them","into","a","single","index","."]
code: def merge(
      self, num_workers: int = 1, node_rank: Optional[int] = None
  ) -> None:
      """Once all the workers have written their own index, the merge function is responsible to read and merge them
      into a single index."""
      num_workers = num_workers or 1
  
      # Only for non rank 0
      if self.rank != 0:
          while not os.path.exists(
              os.path.join(self._cache_dir, _INDEX_FILENAME)
          ):
              sleep(0.01)
          return
  
      # Wait for all indexes to be available
      is_done = False
      while not is_done:
          files = os.listdir(self._cache_dir)
  
          # Return if the index already exists
          if _INDEX_FILENAME in files:
              return
  
          index_files = [
              f for f in files if f.endswith(_INDEX_FILENAME)
          ]
  
          # When using the Data Optimizer, we don't use multi processes.
          is_done = (
              len(index_files)
              == self._distributed_env.world_size * num_workers
          )
          sleep(0.01)
  
      self._merge_no_wait(node_rank=node_rank)
  
imports: import json
  import os
  import warnings
  from dataclasses import dataclass
  from time import sleep
  from typing import Any, Dict, List, Optional, Tuple, Union
  import numpy
  import torch
  from lightning.data.streaming.compression import _COMPRESSORS, Compressor
  from lightning.data.streaming.constants import _INDEX_FILENAME, _TORCH_GREATER_EQUAL_2_1_0
  from lightning.data.streaming.serializers import Serializer, _get_serializers
  from lightning.data.utilities.env import _DistributedEnv, _WorkerEnv
  from lightning.data.utilities.format import _convert_bytes_to_int, _human_readable_bytes
label_desc: Use image node_id 16 for calling the BinaryWriter obj's underlying member method code with example usage: obj.merge(num_workers, node_rank) without return types