class_docstr: Specifies that this step should execute on [AWS Batch](https://aws.amazon.com/batch/).
  
  Parameters
  ----------
  cpu : int, default: 1
      Number of CPUs required for this step. If `@resources` is
      also present, the maximum value from all decorators is used.
  gpu : int, default: 0
      Number of GPUs required for this step. If `@resources` is
      also present, the maximum value from all decorators is used.
  memory : int, default: 4096
      Memory size (in MB) required for this step. If
      `@resources` is also present, the maximum value from all decorators is
      used.
  image : str, optional
      Docker image to use when launching on AWS Batch. If not specified, and
      METAFLOW_BATCH_CONTAINER_IMAGE is specified, that image is used. If
      not, a default Docker image mapping to the current version of Python is used.
  queue : str, default: METAFLOW_BATCH_JOB_QUEUE
      AWS Batch Job Queue to submit the job to.
  iam_role : str, default: METAFLOW_ECS_S3_ACCESS_IAM_ROLE
      AWS IAM role that AWS Batch container uses to access AWS cloud resources.
  execution_role : str, default: METAFLOW_ECS_FARGATE_EXECUTION_ROLE
      AWS IAM role that AWS Batch can use [to trigger AWS Fargate tasks]
      (https://docs.aws.amazon.com/batch/latest/userguide/execution-IAM-role.html).
  shared_memory : int, optional
      The value for the size (in MiB) of the /dev/shm volume for this step.
      This parameter maps to the `--shm-size` option in Docker.
  max_swap : int, optional
      The total amount of swap memory (in MiB) a container can use for this
      step. This parameter is translated to the `--memory-swap` option in
      Docker where the value is the sum of the container memory plus the
      `max_swap` value.
  swappiness : int, optional
      This allows you to tune memory swappiness behavior for this step.
      A swappiness value of 0 causes swapping not to happen unless absolutely
      necessary. A swappiness value of 100 causes pages to be swapped very
      aggressively. Accepted values are whole numbers between 0 and 100.
  use_tmpfs: bool, default: False
      This enables an explicit tmpfs mount for this step.
  tmpfs_tempdir: bool, default: True
      sets METAFLOW_TEMPDIR to tmpfs_path if set for this step.
  tmpfs_size: int, optional
      The value for the size (in MiB) of the tmpfs mount for this step.
      This parameter maps to the `--tmpfs` option in Docker. Defaults to 50% of the
      memory allocated for this step.
  tmpfs_path: string, optional
      Path to tmpfs mount for this step. Defaults to /metaflow_temp.
  inferentia : int, default: 0
      Number of Inferentia chips required for this step.
  efa: int, default: 0
      Number of elastic fabric adapter network devices to attach to container
class_docstr_tok: ["Specifies","that","this","step","should","execute","on","[","AWS","Batch","]","(","https",":","\/\/aws.amazon.com\/batch\/",")",".","Parameters","--","--","--","--","--","cpu",":","int",",","default",":","1","Number","of","CPUs","required","for","this","step",".","If","`","@","resources","`","is","also","present",",","the","maximum","value","from","all","decorators","is","used",".","gpu",":","int",",","default",":","0","Number","of","GPUs","required","for","this","step",".","If","`","@","resources","`","is","also","present",",","the","maximum","value","from","all","decorators","is","used",".","memory",":","int",",","default",":","4096","Memory","size","(","in","MB",")","required","for","this","step",".","If","`","@","resources","`","is","also","present",",","the","maximum","value","from","all","decorators","is","used",".","image",":","str",",","optional","Docker","image","to","use","when","launching","on","AWS","Batch",".","If","not","specified",",","and","METAFLOW_BATCH_CONTAINER_IMAGE","is","specified",",","that","image","is","used",".","If","not",",","a","default","Docker","image","mapping","to","the","current","version","of","Python","is","used",".","queue",":","str",",","default",":","METAFLOW_BATCH_JOB_QUEUE","AWS","Batch","Job","Queue","to","submit","the","job","to",".","iam_role",":","str",",","default",":","METAFLOW_ECS_S3_ACCESS_IAM_ROLE","AWS","IAM","role","that","AWS","Batch","container","uses","to","access","AWS","cloud","resources",".","execution_role",":","str",",","default",":","METAFLOW_ECS_FARGATE_EXECUTION_ROLE","AWS","IAM","role","that","AWS","Batch","can","use","[","to","trigger","AWS","Fargate","tasks","]","(","https",":","\/\/docs.aws.amazon.com\/batch\/latest\/userguide\/execution-IAM-role.html",")",".","shared_memory",":","int",",","optional","The","value","for","the","size","(","in","MiB",")","of","the","\/dev\/shm","volume","for","this","step",".","This","parameter","maps","to","the","`","--","shm-size","`","option","in","Docker",".","max_swap",":","int",",","optional","The","total","amount","of","swap","memory","(","in","MiB",")","a","container","can","use","for","this","step",".","This","parameter","is","translated","to","the","`","--","memory-swap","`","option","in","Docker","where","the","value","is","the","sum","of","the","container","memory","plus","the","`","max_swap","`","value",".","swappiness",":","int",",","optional","This","allows","you","to","tune","memory","swappiness","behavior","for","this","step",".","A","swappiness","value","of","0","causes","swapping","not","to","happen","unless","absolutely","necessary",".","A","swappiness","value","of","100","causes","pages","to","be","swapped","very","aggressively",".","Accepted","values","are","whole","numbers","between","0","and","100.","use_tmpfs",":","bool",",","default",":","False","This","enables","an","explicit","tmpfs","mount","for","this","step",".","tmpfs_tempdir",":","bool",",","default",":","True","sets","METAFLOW_TEMPDIR","to","tmpfs_path","if","set","for","this","step",".","tmpfs_size",":","int",",","optional","The","value","for","the","size","(","in","MiB",")","of","the","tmpfs","mount","for","this","step",".","This","parameter","maps","to","the","`","--","tmpfs","`","option","in","Docker",".","Defaults","to","50","%","of","the","memory","allocated","for","this","step",".","tmpfs_path",":","string",",","optional","Path","to","tmpfs","mount","for","this","step",".","Defaults","to","\/metaflow_temp",".","inferentia",":","int",",","default",":","0","Number","of","Inferentia","chips","required","for","this","step",".","efa",":","int",",","default",":","0","Number","of","elastic","fabric","adapter","network","devices","to","attach","to","container"]
docstr: When launching multinode task with UBF, need to wait for the secondary
  tasks to finish cleanly and produce their output before exiting the
  main task. Otherwise, the main task finishing will cause secondary nodes
  to terminate immediately, and possibly prematurely.
docstr_tok: ["When","launching","multinode","task","with","UBF",",","need","to","wait","for","the","secondary","tasks","to","finish","cleanly","and","produce","their","output","before","exiting","the","main","task",".","Otherwise",",","the","main","task","finishing","will","cause","secondary","nodes","to","terminate","immediately",",","and","possibly","prematurely","."]
code: def _wait_for_mapper_tasks(self, flow, step_name):
      """
      When launching multinode task with UBF, need to wait for the secondary
      tasks to finish cleanly and produce their output before exiting the
      main task. Otherwise, the main task finishing will cause secondary nodes
      to terminate immediately, and possibly prematurely.
      """
      from metaflow import Step  # avoid circular dependency
  
      TIMEOUT = 600
      last_completion_timeout = time.time() + TIMEOUT
      print("Waiting for batch secondary tasks to finish")
      while last_completion_timeout > time.time():
          time.sleep(2)
          try:
              step_path = "%s/%s/%s" % (
                  flow.name,
                  current.run_id,
                  step_name,
              )
              tasks = [task for task in Step(step_path)]
              if len(tasks) == len(flow._control_mapper_tasks):
                  if all(
                      task.finished_at is not None for task in tasks
                  ):  # for some reason task.finished fails
                      return True
              else:
                  print(
                      "Waiting for all parallel tasks to finish. Finished: {}/{}".format(
                          len(tasks),
                          len(flow._control_mapper_tasks),
                      )
                  )
          except Exception as e:
              pass
      raise Exception(
          "Batch secondary workers did not finish in %s seconds"
          % TIMEOUT
      )
  
imports: import os
  import sys
  import platform
  import requests
  import time
  from metaflow import util
  from metaflow import R, current
  from metaflow.decorators import StepDecorator
  from metaflow.plugins.resources_decorator import ResourcesDecorator
  from metaflow.plugins.timeout_decorator import get_run_time_limit_for_task
  from metaflow.metadata import MetaDatum
  from metaflow.metadata.util import sync_local_metadata_to_datastore
  from metaflow.metaflow_config import ECS_S3_ACCESS_IAM_ROLE, BATCH_JOB_QUEUE, BATCH_CONTAINER_IMAGE, BATCH_CONTAINER_REGISTRY, ECS_FARGATE_EXECUTION_ROLE, DATASTORE_LOCAL_DIR
  from metaflow.sidecar import Sidecar
  from metaflow.unbounded_foreach import UBF_CONTROL
  from .batch import BatchException
  from ..aws_utils import compute_resource_attributes, get_docker_registry, get_ec2_instance_metadata
label_desc: Use image node_id 8 for calling the BatchDecorator obj's underlying member method code with example usage: obj._wait_for_mapper_tasks(flow, step_name) and returns: True