(
    ctx, input_fn, inputs_structure_recorder, batch_axis, device, host_id)
return enqueue_ops_fn, captured_infeed_queue, dataset_initializer
Generates infeed enqueue ops for per-host input_fn on a single host.
Generates infeed enqueue ops for per-host input_fn on a single host.
Generates infeed enqueue ops for per - host input_fn on a single host .
def generate_per_host_enqueue_ops_fn_for_host(
    ctx, input_fn, inputs_structure_recorder, batch_axis, device, host_id):
  """Generates infeed enqueue ops for per-host input_fn on a single host."""
  captured_infeed_queue = _CapturedObject()

  dataset_initializer = None

  with ops.device(device):
    user_context = tpu_context.TPUContext(
        internal_ctx=ctx, input_device=device, invocation_index=host_id)
    inputs = _Inputs.from_input_fn(input_fn(user_context))

    is_dataset = inputs.is_dataset
    if ctx.mode == model_fn_lib.ModeKeys.PREDICT:
      if not is_dataset:
        raise TypeError(
            'For mode PREDICT, `input_fn` must return `Dataset` instead of '
            '`features` and `labels`.')
      if batch_axis is not None:
        raise TypeError('For mode PREDICT, batch_axis is not supported yet.')
      inputs = _InputsWithStoppingSignals(
          dataset=inputs.dataset,
          batch_size=ctx.batch_size_for_input_fn,
          add_padding=True)

    if is_dataset:
      dataset_initializer = inputs.dataset_initializer()

    tpu_ordinal_function_impl = ctx.tpu_ordinal_function(host_id)

  def enqueue_ops_fn():
    """A Fn returning the TPU infeed enqueue ops.

    By providing as a Fn, it can be invoked inside the tf.while_loop such that
    the input pipeline for multiple iterations can be executed by one
    Session.run call.

    Returns:
      list of dict of ops.
    """
    with ops.device(device):
      num_of_replicas_per_host = ctx.num_of_replicas_per_host
      # Convert user input to features and labels.  If the user returns a
      # dataset, it is initialized and the features and labels extracted via
      # `dataset.iterator.get_next()`
      features, labels = inputs.features_and_labels()
      signals = inputs.signals()

      inputs_structure_recorder.validate_and_record_structure(features, labels)
      unsharded_tensor_list = (
          inputs_structure_recorder.flatten_features_and_labels(
              features, labels, signals))

      infeed_queue = tpu_feed.InfeedQueue(
          tuple_types=[t.dtype for t in unsharded_tensor_list],
          tuple_shapes=[t.shape for t in unsharded_tensor_list],
          shard_dimensions=batch_axis)
      captured_infeed_queue.capture(infeed_queue)
      infeed_queue.set_number_of_shards(num_of_replicas_per_host)
      per_host_enqueue_ops = (
          infeed_queue.split_inputs_and_generate_enqueue_ops(
              unsharded_tensor_list,
              placement_function=lambda x: device,
              tpu_ordinal_function=tpu_ordinal_function_impl))
      if signals is None:
        return per_host_enqueue_ops
      else:
        return {
            'ops': per_host_enqueue_ops,
            'signals': signals,
        }

  return enqueue_ops_fn, captured_infeed_queue, dataset_initializer
def generate_per_host_enqueue_ops_fn_for_host ( ctx , input_fn , inputs_structure_recorder , batch_axis , device , host_id ) : captured_infeed_queue = _CapturedObject ( ) dataset_initializer = None with ops . device ( device ) : user_context = tpu_context . TPUContext ( internal_ctx = ctx , input_device = device , invocation_index = host_id ) inputs = _Inputs . from_input_fn ( input_fn ( user_context ) ) is_dataset = inputs . is_dataset if ctx . mode == model_fn_lib . ModeKeys . PREDICT : if not is_dataset : raise TypeError ( 'For mode PREDICT, `input_fn` must return `Dataset` instead of ' '`features` and `labels`.' ) if batch_axis is not None : raise TypeError ( 'For mode PREDICT, batch_axis is not supported yet.' ) inputs = _InputsWithStoppingSignals ( dataset = inputs . dataset , batch_size = ctx . batch_size_for_input_fn , add_padding = True ) if is_dataset : dataset_initializer = inputs . dataset_initializer ( ) tpu_ordinal_function_impl = ctx . tpu_ordinal_function ( host_id ) def enqueue_ops_fn ( ) : """A Fn returning the TPU infeed enqueue ops.

    By providing as a Fn, it can be invoked inside the tf.while_loop such that
    the input pipeline for multiple iterations can be executed by one
    Session.run call.

    Returns:
      list of dict of ops.
    """ with ops . device ( device ) : num_of_replicas_per_host = ctx . num_of_replicas_per_host # Convert user input to features and labels.  If the user returns a # dataset, it is initialized and the features and labels extracted via # `dataset.iterator.get_next()` features , labels = inputs . features_and_labels ( ) signals = inputs . signals ( ) inputs_structure_recorder . validate_and_record_structure ( features , labels ) unsharded_tensor_list = ( inputs_structure_recorder . flatten_features_and_labels ( features , labels , signals ) ) infeed_queue = tpu_feed . InfeedQueue ( tuple_types = [ t . dtype for t in unsharded_tensor_list ] , tuple_shapes = [ t . shape for t in unsharded_tensor_list ] , shard_dimensions = batch_axis ) captured_infeed_queue . capture ( infeed_queue ) infeed_queue . set_number_of_shards ( num_of_replicas_per_host ) per_host_enqueue_ops = ( infeed_queue . split_inputs_and_generate_enqueue_ops ( unsharded_tensor_list , placement_function = lambda x : device , tpu_ordinal_function = tpu_ordinal_function_impl ) ) if signals is None : return per_host_enqueue_ops else : return { 'ops' : per_host_enqueue_ops , 'signals' : signals , } return enqueue_ops_fn , captured_infeed_queue , dataset_initializer