(self)
return ret
Sends the tensors through outfeed and runs the host_fn on CPU.

    The tensors are concatenated along dimension 0 to form a global tensor
    across all shards. The concatenated function is passed to the host_fn and
    executed on the first host.

    Returns:
      A dictionary mapping name to the return type of the host_call by that
      name.

    Raises:
      RuntimeError: If outfeed tensor is scalar.
Sends the tensors through outfeed and runs the host_fn on CPU.
Sends the tensors through outfeed and runs the host_fn on CPU .
def create_tpu_hostcall(self):
    """Sends the tensors through outfeed and runs the host_fn on CPU.

    The tensors are concatenated along dimension 0 to form a global tensor
    across all shards. The concatenated function is passed to the host_fn and
    executed on the first host.

    Returns:
      A dictionary mapping name to the return type of the host_call by that
      name.

    Raises:
      RuntimeError: If outfeed tensor is scalar.
    """
    if not self._names:
      return {}

    ret = {}
    # For each i, dequeue_ops[i] is a list containing the tensors from all
    # shards. This list is concatenated later.
    dequeue_ops = []
    tensor_dtypes = []
    tensor_shapes = []
    for name in self._names:
      for _ in self._tensors[name]:
        dequeue_ops.append([])
      for dtype in self._tensor_dtypes[name]:
        tensor_dtypes.append(dtype)
      for shape in self._tensor_shapes[name]:
        tensor_shapes.append(shape)

    # Outfeed ops execute on each replica's first logical core. Note: we must
    # constraint it such that we have at most one outfeed dequeue and enqueue
    # per replica.
    for i in xrange(self._ctx.num_replicas):
      host_device, ordinal_id = self._ctx.device_for_replica(i)
      with ops.device(host_device):
        outfeed_tensors = tpu_ops.outfeed_dequeue_tuple(
            dtypes=tensor_dtypes,
            shapes=tensor_shapes,
            device_ordinal=ordinal_id)
        for j, item in enumerate(outfeed_tensors):
          dequeue_ops[j].append(item)

    # Deconstruct dequeue ops.
    dequeue_ops_by_name = {}
    pos = 0
    for name in self._names:
      dequeue_ops_by_name[name] = dequeue_ops[pos:pos +
                                              len(self._tensors[name])]
      pos += len(self._tensors[name])

    # It is assumed evaluation always happens on single host TPU system. So,
    # place all ops on tpu host if possible.
    #
    # TODO(jhseu): Evaluate whether this is right for summaries.
    with ops.device(self._ctx.tpu_host_placement_function(replica_id=0)):
      for name in self._names:
        dequeue_ops = dequeue_ops_by_name[name]
        for i, item in enumerate(dequeue_ops):
          if dequeue_ops[i][0].shape.ndims == 0:
            raise RuntimeError(
                'All tensors outfed from TPU should preserve batch size '
                'dimension, but got scalar {}'.format(dequeue_ops[i][0]))
          # TODO(xiejw): Allow users to specify the axis for batch size
          # dimension.
          dequeue_ops[i] = array_ops.concat(dequeue_ops[i], axis=0)

        if self._tensor_keys[name] is not None:
          # The user-provided eval_metrics[1] is a dict.
          dequeue_ops = dict(zip(self._tensor_keys[name], dequeue_ops))
          try:
            ret[name] = self._host_fns[name](**dequeue_ops)
          except TypeError as e:
            logging.warning(
                'Exception while calling %s: %s. It is likely the tensors '
                '(%s[1]) do not match the '
                'function\'s arguments', name, e, name)
            raise e
        else:
          ret[name] = self._host_fns[name](*dequeue_ops)

    return ret
def create_tpu_hostcall ( self ) : if not self . _names : return { } ret = { } # For each i, dequeue_ops[i] is a list containing the tensors from all # shards. This list is concatenated later. dequeue_ops = [ ] tensor_dtypes = [ ] tensor_shapes = [ ] for name in self . _names : for _ in self . _tensors [ name ] : dequeue_ops . append ( [ ] ) for dtype in self . _tensor_dtypes [ name ] : tensor_dtypes . append ( dtype ) for shape in self . _tensor_shapes [ name ] : tensor_shapes . append ( shape ) # Outfeed ops execute on each replica's first logical core. Note: we must # constraint it such that we have at most one outfeed dequeue and enqueue # per replica. for i in xrange ( self . _ctx . num_replicas ) : host_device , ordinal_id = self . _ctx . device_for_replica ( i ) with ops . device ( host_device ) : outfeed_tensors = tpu_ops . outfeed_dequeue_tuple ( dtypes = tensor_dtypes , shapes = tensor_shapes , device_ordinal = ordinal_id ) for j , item in enumerate ( outfeed_tensors ) : dequeue_ops [ j ] . append ( item ) # Deconstruct dequeue ops. dequeue_ops_by_name = { } pos = 0 for name in self . _names : dequeue_ops_by_name [ name ] = dequeue_ops [ pos : pos + len ( self . _tensors [ name ] ) ] pos += len ( self . _tensors [ name ] ) # It is assumed evaluation always happens on single host TPU system. So, # place all ops on tpu host if possible. # # TODO(jhseu): Evaluate whether this is right for summaries. with ops . device ( self . _ctx . tpu_host_placement_function ( replica_id = 0 ) ) : for name in self . _names : dequeue_ops = dequeue_ops_by_name [ name ] for i , item in enumerate ( dequeue_ops ) : if dequeue_ops [ i ] [ 0 ] . shape . ndims == 0 : raise RuntimeError ( 'All tensors outfed from TPU should preserve batch size ' 'dimension, but got scalar {}' . format ( dequeue_ops [ i ] [ 0 ] ) ) # TODO(xiejw): Allow users to specify the axis for batch size # dimension. dequeue_ops [ i ] = array_ops . concat ( dequeue_ops [ i ] , axis = 0 ) if self . _tensor_keys [ name ] is not None : # The user-provided eval_metrics[1] is a dict. dequeue_ops = dict ( zip ( self . _tensor_keys [ name ] , dequeue_ops ) ) try : ret [ name ] = self . _host_fns [ name ] ( * * dequeue_ops ) except TypeError as e : logging . warning ( 'Exception while calling %s: %s. It is likely the tensors ' '(%s[1]) do not match the ' 'function\'s arguments' , name , e , name ) raise e else : ret [ name ] = self . _host_fns [ name ] ( * dequeue_ops ) return ret