code: def sort_main(args: Args):
      parts = _load_manifest(args, constants.INPUT_MANIFEST_FILE)
      assert len(parts) == args.num_mappers
      boundaries = sortlib.get_boundaries(args.num_reducers)
  
      # The exception of 'ValueError("Resource quantities >1 must be whole numbers.")'
      # will be raised if the `num_cpus` > 1 and not an integer.
      num_cpus = os.cpu_count() / args.num_concurrent_rounds
      if num_cpus > 1.0:
          num_cpus = int(num_cpus)
      mapper_opt = {
          "num_returns": args.num_reducers,
          "num_cpus": num_cpus,
      }  # Load balance across worker nodes by setting `num_cpus`.
      merge_results = np.empty(
          (args.num_rounds, args.num_reducers), dtype=object
      )
  
      part_id = 0
      with worker_placement_groups(args) as pgs:
          for round in range(args.num_rounds):
              # Limit the number of in-flight rounds.
              num_extra_rounds = round - args.num_concurrent_rounds + 1
              if num_extra_rounds > 0:
                  ray.wait(
                      [
                          f
                          for f in merge_results.flatten()
                          if f is not None
                      ],
                      num_returns=num_extra_rounds * args.num_reducers,
                  )
  
              # Submit map tasks.
              mapper_results = np.empty(
                  (args.num_mappers_per_round, args.num_reducers),
                  dtype=object,
              )
              for _ in range(args.num_mappers_per_round):
                  _, node, path = parts[part_id]
                  m = part_id % args.num_mappers_per_round
                  mapper_results[m, :] = mapper.options(
                      **mapper_opt
                  ).remote(args, part_id, boundaries, path)
                  part_id += 1
  
              # Submit merge tasks.
              merge_results[round, :] = [
                  merge_mapper_blocks.options(
                      scheduling_strategy=PlacementGroupSchedulingStrategy(
                          placement_group=pgs[r]
                      )
                  ).remote(
                      args, r, round, *mapper_results[:, r].tolist()
                  )
                  for r in range(args.num_reducers)
              ]
  
              # Delete local references to mapper results.
              mapper_results = None
  
          # Submit second-stage reduce tasks.
          reducer_results = [
              final_merge.options(
                  scheduling_strategy=PlacementGroupSchedulingStrategy(
                      placement_group=pgs[r]
                  )
              ).remote(args, r, *merge_results[:, r].tolist())
              for r in range(args.num_reducers)
          ]
          reducer_results = ray.get(reducer_results)
  
      if not args.skip_output:
          with open(constants.OUTPUT_MANIFEST_FILE, "w") as fout:
              writer = csv.writer(fout)
              writer.writerows(reducer_results)
  
      logging.info(
          ray._private.internal_api.memory_summary(stats_only=True)
      )
  
imports: import argparse
  import contextlib
  import csv
  import logging
  import os
  import random
  import subprocess
  import tempfile
  from typing import Callable, Dict, Iterable, List
  import numpy
  import ray
  from ray.experimental.raysort import constants, logging_utils, sortlib, tracing_utils
  from ray.experimental.raysort.types import BlockInfo, ByteCount, PartId, PartInfo, Path, RecordCount
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
label_desc: Use image node_id 17 for calling a global function with example usage: sort_main(args) without return types