code: def worker_loop(
      dataset,
      cols_to_retain,
      collate_fn,
      collate_fn_args,
      columns_to_np_types,
      columns_to_ranks,
      string_columns,
      indices,
      extra_batch,
      worker_name,
      array_ready_event,
      array_loaded_event,
  ):
      os.environ["TF_CPP_MIN_LOG_LEVEL"] = "3"
  
      if config.TF_AVAILABLE:
          import tensorflow as tf
      else:
          raise ImportError(
              "Called a Tensorflow-specific function but Tensorflow is not installed."
          )
  
      tf.config.set_visible_devices(
          [], "GPU"
      )  # Make sure workers don't try to allocate GPU memory
  
      def send_batch_to_parent(indices):
          batch = np_get_batch(
              indices=indices,
              dataset=dataset,
              cols_to_retain=cols_to_retain,
              collate_fn=collate_fn,
              collate_fn_args=collate_fn_args,
              columns_to_np_types=columns_to_np_types,
              return_dict=True,
          )
  
          # Now begins the fun part where we start shovelling shared memory at the parent process
          out_arrays = {}
          with SharedMemoryContext() as batch_shm_ctx:
              # The batch shared memory context exists only as long as it takes for the parent process
              # to read everything, after which it cleans everything up again
              for col, cast_dtype in columns_to_np_types.items():
                  # Everything has to be np.array for this to work, even if the collate_fn is giving us tf.Tensor
                  array = batch[col]
                  if col in string_columns:
                      # We can't send unicode arrays over shared memory, so we convert to single chars ("U1")
                      # which have a fixed width of 4 bytes. The parent process will convert these back to strings.
                      array = array.view("U1").reshape(
                          array.shape + (-1,)
                      )
                  shape_arrays[col][:] = array.shape
                  out_arrays[col] = batch_shm_ctx.get_array(
                      f"{worker_name}_{col}",
                      shape=array.shape,
                      dtype=cast_dtype,
                      create=True,
                  )
                  out_arrays[col][:] = array
  
              array_ready_event.set()
              array_loaded_event.wait()
              array_loaded_event.clear()
  
      with SharedMemoryContext() as shm_ctx:
          shape_arrays = {
              col: shm_ctx.get_array(
                  f"{worker_name}_{col}_shape",
                  shape=(rank,),
                  dtype=np.int64,
                  create=False,
              )
              for col, rank in columns_to_ranks.items()
          }
  
          for batch in indices:
              send_batch_to_parent(batch)
          if extra_batch is not None:
              send_batch_to_parent(extra_batch)
          # Now we send a batsignal to the parent process that we're done
          for col, array in shape_arrays.items():
              array[:] = -1
          array_ready_event.set()
  
imports: import os
  import warnings
  from functools import partial
  from math import ceil
  from uuid import uuid4
  import numpy
  import pyarrow
  from multiprocess import get_context
  from ..None import config
label_desc: Use image node_id 4 for calling the NumpyMultiprocessingGenerator obj's underlying member method code with example usage: obj.worker_loop(dataset, cols_to_retain, collate_fn, collate_fn_args, columns_to_np_types, columns_to_ranks, string_columns, indices, extra_batch, worker_name, array_ready_event, array_loaded_event) without return types