code: def _mixed_dataset_process(
      config_path: str,
      q_cmd: multiprocessing.Queue,
      q_cmd_out: multiprocessing.Queue,
      q_data: multiprocessing.Queue,
      rank: int,
      world_size: int,
      packer: _MixedDatasetBatchPacker,
  ):
      # ignore SIGINT
      import signal
  
      signal.signal(signal.SIGINT, signal.SIG_IGN)
      config_base_path = os.path.dirname(os.path.abspath(config_path))
  
      def _convert_to_abs_path(transform_path: str):
          if transform_path.startswith("/"):
              return transform_path
          else:
              return os.path.join(config_base_path, transform_path)
  
      def _build_sample_weights(config: List[_MixedDatasetConfig]):
          if len(config) == 0:
              return np.array([], dtype=np.float32)
          weights = [c["weight"] * c["lines"] for c in config]
          weights = np.array(weights, dtype=np.float32)
          sm_weight = weights.sum()
          if sm_weight > 0:
              weights = weights / sm_weight
              return weights
          else:
              raise RuntimeError("Empty datasets")
  
      cfg_mgr = _MixedDatasetConfigMananger(config_path)
      config = cfg_mgr.get_config()
  
      for c in config:
          ds = DistributedDataset(
              _convert_to_abs_path(c["path"]),
              rank,
              world_size,
          )
  
          c["lines"] = ds._nlines
          c["dataset"] = ds
          if "weight" not in c:
              c["weight"] = 1.0
          if "transforms" not in c:
              c["transforms"] = []
          elif isinstance(c["transforms"], str):
              c["transforms"] = _convert_to_abs_path(c["transforms"])
          if "incontext_weight" not in c:
              c["incontext_weight"] = [1.0]
  
      weights = _build_sample_weights(config)
  
      should_stop = False
      should_start = False
  
      while not should_stop:
          # update config first
          if cfg_mgr.changed():
              path_ds_map: Dict[str, _MixedDatasetConfig] = {}
              nw_path_set: Set[str] = set()
  
              # load new config
              nw_config = cfg_mgr.get_config()
  
              # build path -> dataset map
              for c in config:
                  path_ds_map[_dataset_identity(c)] = c
  
              # add new datasets
              for c in nw_config:
                  if _dataset_identity(c) in path_ds_map:
                      # update values only
                      if "weight" in c:
                          path_ds_map[_dataset_identity(c)][
                              "weight"
                          ] = c["weight"]
                      if "transform" in c:
                          if isinstance(c["transforms"], str):
                              path_ds_map[_dataset_identity(c)][
                                  "transforms"
                              ] = _convert_to_abs_path(c["transforms"])
                          else:
                              path_ds_map[_dataset_identity(c)][
                                  "transforms"
                              ] = c["transforms"]
                      if "incontext_weight" in c:
                          path_ds_map[_dataset_identity(c)][
                              "incontext_weight"
                          ] = c["incontext_weight"]
                  else:
                      # new dataset
                      ds = DistributedDataset(
                          _convert_to_abs_path(c["path"]),
                          rank,
                          world_size,
                      )
                      c["lines"] = ds._nlines
                      c["dataset"] = ds
                      if "weight" not in c:
                          c["weight"] = 1.0
                      if "transforms" not in c:
                          c["transforms"] = []
                      elif isinstance(c["transforms"], str):
                          c["transforms"] = _convert_to_abs_path(
                              c["transforms"]
                          )
                      if "incontext_weight" not in c:
                          c["incontext_weight"] = [1.0]
                      path_ds_map[_dataset_identity(c)] = c
                  nw_path_set.add(_dataset_identity(c))
  
              # remove unused datasets
              for c in config:
                  if _dataset_identity(c) not in nw_path_set:
                      del path_ds_map[_dataset_identity(c)]
  
              config: List[_MixedDatasetConfig] = []
              for c in nw_config:
                  config.append(path_ds_map[_dataset_identity(c)])
              del path_ds_map
              del nw_path_set
              del nw_config
  
              weights = _build_sample_weights(config)
  
          # get cmds
          while True:
              try:
                  cmd = q_cmd.get_nowait()
              except Empty:
                  break
              if cmd == "stop":
                  should_stop = True
                  q_cmd_out.put(True)
                  break
              elif cmd == "state_dict":
                  ret = OrderedDict()
                  for c in config:
                      ds_name = _dataset_identity(c)
                      ret[ds_name] = c["dataset"]._state_dict()
                  q_cmd_out.put(ret)
              elif cmd == "load_state_dict":
                  state_dict = q_cmd.get()
                  missing = []
                  for c in config:
                      ds_name = _dataset_identity(c)
                      if ds_name in state_dict:
                          c["dataset"].load_state_dict(
                              state_dict[ds_name], strict=False
                          )
                      else:
                          # new dataset
                          missing.append(ds_name)
                  q_cmd_out.put(missing)
              elif cmd == "start":
                  should_start = True
                  q_cmd_out.put(True)
              else:
                  raise RuntimeError("Unknown command: {}".format(cmd))
  
          if should_stop:
              break
  
          if not should_start:
              # wait for start cmd
              time.sleep(1)
              continue
  
          if len(config) == 0:
              # no dataset available
              time.sleep(1)
              continue
  
          if q_data.full():
              # queue full
              time.sleep(1)
              continue
  
          # sample a dataset
          ds_id: int = 0
  
          while True:
              ds_id = np.random.choice(weights.shape[0], p=weights)
              if (
                  config[ds_id]["dataset"]._nlines
                  != config[ds_id]["lines"]
              ):
                  # dataset size changed
                  for c in config:
                      c["lines"] = c["dataset"]._nlines
                  weights = _build_sample_weights(config)
                  continue
              else:
                  break
  
          batch = packer.add_data(config[ds_id])
          if batch is not None:
              # new batch comming
              q_data.put(batch)
  
      # clean queue
      while True:
          try:
              q_data.get_nowait()
          except Empty:
              break
  
imports: from collections import OrderedDict
  import multiprocessing
  import os
  from queue import Empty
  from typing import Any, Callable, Dict, List, Optional, Set, Tuple, Union
  from typing_extensions import TypedDict
  from ...dataset import DistributedDataset
  from ...tokenizers import CPMBeeTokenizer
  from ...utils.config import load_dataset_config
  import numpy
  import time
  from numpy.typing import NDArray
  import torch
  import bmtrain
  import importlib.machinery
  import importlib.util
  import types
  import random
label_desc: Use image node_id 4 for calling a global function with example usage: _mixed_dataset_process(config_path, q_cmd, q_cmd_out, q_data, rank, world_size, packer) and returns: transform_path, os, np, weights