Method SendFileSlice

tensorflow/core/kernels/file_slice_sendrecv_ops.cc:179–221

}

Status FileSliceSendOp::SendFileSlice(OpKernelContext* ctx,
                                      const FrameAndIter& frame_iter,
                                      const Tensor& file_path_t,
                                      const uint64 element_bytes) {
  Rendezvous::Args args;
  args.device_context = ctx->op_device_context();
  args.alloc_attrs = AllocatorAttributes();
  Rendezvous::ParsedKey parsed_key;

  std::unique_ptr<RandomAccessFile> file;
  Env* env = Env::Default();
  const string& file_path = file_path_t.scalar<tstring>()();
  TF_RETURN_IF_ERROR(env->NewRandomAccessFile(file_path, &file));

  // Slice Send.
  int64 slice_num = element_bytes / slice_size_;
  if (element_bytes % slice_size_ != 0) {
    slice_num += 1;
  }
  Tensor data_t;
  for (int64 i = 0; i < slice_num; i++) {
    TF_RETURN_IF_ERROR(ctx->allocate_temp(DT_STRING, TensorShape({}), &data_t));
    uint64 start = i * slice_size_;
    uint64 copy_size = slice_size_;
    if (start > element_bytes - slice_size_) {
      copy_size = element_bytes - start;
    }
    TF_RETURN_IF_ERROR(ReadFileSlice(file, start, copy_size, data_t));
    std::string tensor_name_suffix = \
      strings::StrCat("_slice_transfer_data_", std::to_string(0), "_",
                      std::to_string(i));
    slice_sendrecv::GetSliceRendezvousKey(key_prefix_, tensor_name_suffix,
                                          frame_iter, &parsed_key.buf_);
    VLOG(2) << "FileSliceSend " << parsed_key.buf_;
    TF_RETURN_IF_ERROR(Rendezvous::ParseKey(parsed_key.buf_, &parsed_key));
    TF_RETURN_IF_ERROR(
      ctx->rendezvous()->FlowControlSend(tensor_name_, parsed_key, args, data_t,
                                         ctx->is_input_dead()));
  }


  return Status::OK();
}

Status FileSliceSendOp::ReadFileSlice(
                          const std::unique_ptr<RandomAccessFile>& file,