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"]
code: def task_pre_step(
      self,
      step_name,
      task_datastore,
      metadata,
      run_id,
      task_id,
      flow,
      graph,
      retry_count,
      max_retries,
      ubf_context,
      inputs,
  ):
      self.metadata = metadata
      self.task_datastore = task_datastore
  
      # current.tempdir reflects the value of METAFLOW_TEMPDIR (the current working
      # directory by default), or the value of tmpfs_path if tmpfs_tempdir=False.
      if not self.attributes["tmpfs_tempdir"]:
          current._update_env(
              {"tempdir": self.attributes["tmpfs_path"]}
          )
  
      # task_pre_step may run locally if fallback is activated for @catch
      # decorator. In that scenario, we skip collecting AWS Batch execution
      # metadata. A rudimentary way to detect non-local execution is to
      # check for the existence of AWS_BATCH_JOB_ID environment variable.
  
      if "AWS_BATCH_JOB_ID" in os.environ:
          meta = {}
          meta["aws-batch-job-id"] = os.environ["AWS_BATCH_JOB_ID"]
          meta["aws-batch-job-attempt"] = os.environ[
              "AWS_BATCH_JOB_ATTEMPT"
          ]
          meta["aws-batch-ce-name"] = os.environ["AWS_BATCH_CE_NAME"]
          meta["aws-batch-jq-name"] = os.environ["AWS_BATCH_JQ_NAME"]
          meta["aws-batch-execution-env"] = os.environ[
              "AWS_EXECUTION_ENV"
          ]
  
          # Capture AWS Logs metadata. This is best-effort only since
          # only V4 of the metadata uri for the ECS container hosts this
          # information, and it is quite likely that not all consumers of
          # Metaflow would be running the container agent compatible with
          # version V4.
          # https://docs.aws.amazon.com/AmazonECS/latest/developerguide/task-metadata-endpoint.html
          try:
              logs_meta = (
                  requests.get(
                      url=os.environ["ECS_CONTAINER_METADATA_URI_V4"]
                  )
                  .json()
                  .get("LogOptions", {})
              )
              meta["aws-batch-awslogs-group"] = logs_meta.get(
                  "awslogs-group"
              )
              meta["aws-batch-awslogs-region"] = logs_meta.get(
                  "awslogs-region"
              )
              meta["aws-batch-awslogs-stream"] = logs_meta.get(
                  "awslogs-stream"
              )
          except:
              pass
  
          instance_meta = get_ec2_instance_metadata()
          meta.update(instance_meta)
  
          entries = [
              MetaDatum(
                  field=k,
                  value=v,
                  type=k,
                  tags=["attempt_id:{0}".format(retry_count)],
              )
              for k, v in meta.items()
          ]
          # Register book-keeping metadata for debugging.
          metadata.register_metadata(
              run_id, step_name, task_id, entries
          )
  
          self._save_logs_sidecar = Sidecar("save_logs_periodically")
          self._save_logs_sidecar.start()
  
      num_parallel = int(os.environ.get("AWS_BATCH_JOB_NUM_NODES", 0))
      if num_parallel >= 1 and ubf_context == UBF_CONTROL:
          # UBF handling for multinode case
          control_task_id = current.task_id
          top_task_id = control_task_id.replace(
              "control-", ""
          )  # chop "-0"
          mapper_task_ids = [control_task_id] + [
              "%s-node-%d" % (top_task_id, node_idx)
              for node_idx in range(1, num_parallel)
          ]
          flow._control_mapper_tasks = [
              "%s/%s/%s" % (run_id, step_name, mapper_task_id)
              for mapper_task_id in mapper_task_ids
          ]
          flow._control_task_is_mapper_zero = True
  
      if num_parallel >= 1:
          _setup_multinode_environment()
  
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 6 for calling the BatchDecorator obj's underlying member method code with example usage: obj.task_pre_step(step_name, task_datastore, metadata, run_id, task_id, flow, graph, retry_count, max_retries, ubf_context, inputs) without return types