code: def split_numpy_chunk_pool(
      input_path,
      output_path,
      pretrain_data,
      worker,
      dupe_factor,
      seq_len,
      file_name,
  ):
      if os.path.exists(os.path.join(output_path, f"{file_name}.h5")):
          print(f"{file_name}.h5 exists")
          return
  
      documents = []
      instances = []
  
      s = time.time()
      with open(input_path, "r", encoding="utf-8") as fd:
          document = []
          for i, line in enumerate(tqdm(fd)):
              line = line.strip()
              if (
                  len(line) > 0 and line[:2] == "]]"
              ):  # This is end of document
                  documents.append(document)
                  document = []
              elif len(line) >= 2:
                  document.append(line)
          if len(document) > 0:
              documents.append(document)
      print(
          f"read_file cost {time.time() - s}, length is {len(documents)}"
      )
  
      ans = []
      s = time.time()
      pool = multiprocessing.Pool(worker)
      encoded_doc = pool.imap_unordered(
          pretrain_data.tokenize, documents, 100
      )
      for index, res in tqdm(
          enumerate(encoded_doc, start=1),
          total=len(documents),
          colour="cyan",
      ):
          ans.append(res)
      pool.close()
      print((time.time() - s) / 60)
      del documents
  
      instances = []
      for a in tqdm(ans, colour="MAGENTA"):
          raw_ins = get_raw_instance(a, max_sequence_length=seq_len)
          instances.extend(raw_ins)
      del ans
  
      print("len instance", len(instances))
  
      new_instances = []
      for _ in range(dupe_factor):
          for ins in instances:
              new_instances.append(ins)
  
      shuffle(new_instances)
      instances = new_instances
      print("after dupe_factor, len instance", len(instances))
  
      sentence_num = len(instances)
      input_ids = np.zeros([sentence_num, seq_len], dtype=np.int32)
      input_mask = np.zeros([sentence_num, seq_len], dtype=np.int32)
      segment_ids = np.zeros([sentence_num, seq_len], dtype=np.int32)
      masked_lm_output = np.zeros(
          [sentence_num, seq_len], dtype=np.int32
      )
  
      s = time.time()
      pool = multiprocessing.Pool(worker)
      encoded_docs = pool.imap_unordered(
          pretrain_data.create_training_instance, instances, 32
      )
      for index, mask_dict in tqdm(
          enumerate(encoded_docs), total=len(instances), colour="blue"
      ):
          input_ids[index] = mask_dict[0]
          input_mask[index] = mask_dict[1]
          segment_ids[index] = mask_dict[2]
          masked_lm_output[index] = mask_dict[3]
      pool.close()
      print((time.time() - s) / 60)
  
      with h5py.File(
          os.path.join(output_path, f"{file_name}.h5"), "w"
      ) as hf:
          hf.create_dataset("input_ids", data=input_ids)
          hf.create_dataset("input_mask", data=input_mask)
          hf.create_dataset("segment_ids", data=segment_ids)
          hf.create_dataset(
              "masked_lm_positions", data=masked_lm_output
          )
  
      del instances
  
file_path: ColossalAI/examples/community/roberta/preprocessing/tokenize_mask.py
imports: import argparse
  import multiprocessing
  import os
  import time
  from random import shuffle
  import h5py
  import numpy
  import psutil
  from get_mask import PreTrainingDataset
  from tqdm import tqdm
  from transformers import AutoTokenizer
label_desc: Use image node_id 3 for calling a global function with example usage: split_numpy_chunk_pool(input_path, output_path, pretrain_data, worker, dupe_factor, seq_len, file_name) without return types