//   Copyright (c) 2018 PaddlePaddle Authors. All Rights Reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#include "paddle/fluid/framework/details/broadcast_op_handle.h"
#include "paddle/fluid/framework/details/container_cast.h"
#include "paddle/fluid/framework/details/variable_visitor.h"

namespace paddle {
namespace framework {
namespace details {

void BroadcastOpHandle::RunImpl() {
  if (places_.size() == 1) return;

  // The input and output may have dummy vars.
  VarHandle *in_var_handle;
  {
    auto in_var_handles = DynamicCast<VarHandle>(inputs_);
    PADDLE_ENFORCE_EQ(in_var_handles.size(), 1,
                      "The number of input should be one.");
    in_var_handle = in_var_handles[0];
  }

  auto out_var_handles = DynamicCast<VarHandle>(outputs_);

  PADDLE_ENFORCE_EQ(
      out_var_handles.size(), places_.size(),
      "The number of output should equal to the number of places.");

  // Wait input done, this Wait is asynchronous operation platform::Place
  // &in_place;
  WaitInputVarGenerated(*in_var_handle);

  std::vector<const Scope *> var_scopes;
  for (auto *s : local_scopes_) {
    var_scopes.emplace_back(s->FindVar(kLocalExecScopeName)->Get<Scope *>());
  }

  auto *in_var =
      var_scopes.at(in_var_handle->scope_idx_)->FindVar(in_var_handle->name_);
  PADDLE_ENFORCE_NOT_NULL(in_var);

  Tensor &in_tensor = VariableVisitor::GetMutableTensor(in_var);

  // NOTE(zcd): the Place of input can get from in_tensor and in_var_handle ,
  // maybe they are different, because the Place that getting from in_tensor is
  // determined at runtime, the other is determined at building SSA graph stage.
  // If they are different, DataTransform should be applied. Currently, it has
  // not been done yet.
  for (auto *out_var_handle : out_var_handles) {
    if (*out_var_handle == *in_var_handle) {
      continue;
    }
    auto &out_p = out_var_handle->place_;
    auto *out_var = var_scopes.at(out_var_handle->scope_idx_)
                        ->FindVar(out_var_handle->name_);
    PADDLE_ENFORCE_NOT_NULL(out_var);
    PADDLE_ENFORCE_EQ(
        out_p.which(), in_tensor.place().which(),
        "Currently, Places of input and output must be all on CPU "
        "or all on GPU.");
    VariableVisitor::ShareDimsAndLoD(*in_var, out_var);
    VariableVisitor::GetMutableTensor(out_var).mutable_data(out_p,
                                                            in_tensor.type());
  }

  if (platform::is_cpu_place(in_tensor.place())) {
    for (auto *out_var_handle : out_var_handles) {
      if (*out_var_handle == *in_var_handle) {
        continue;
      }

      auto &out_p = out_var_handle->place_;
      auto dev_ctx = dev_ctxes_.at(out_p);
      auto *out_var = var_scopes.at(out_var_handle->scope_idx_)
                          ->FindVar(out_var_handle->name_);

      RunAndRecordEvent(out_p, [in_tensor, out_var, dev_ctx, out_p] {
        paddle::framework::TensorCopy(
            in_tensor, out_p, *dev_ctx,
            &VariableVisitor::GetMutableTensor(out_var));
      });
    }
  } else {
#ifdef PADDLE_WITH_CUDA
    VarHandle *out_handle = nullptr;
    int root_id = boost::get<platform::CUDAPlace>(in_tensor.place()).device;
    std::vector<std::function<void()>> broadcast_calls;

    for (auto out_var_handle : out_var_handles) {
      Variable *out_var = var_scopes.at(out_var_handle->scope_idx_)
                              ->FindVar(out_var_handle->name_);

      int dst_id =
          boost::get<platform::CUDAPlace>(out_var_handle->place_).device;

      auto &nccl_ctx = nccl_ctxs_->at(dst_id);

      void *send_recv_buffer = nullptr;
      if (root_id == dst_id) {
        send_recv_buffer = const_cast<void *>(in_tensor.data<void>());
        out_handle = out_var_handle;
      } else {
        send_recv_buffer =
            VariableVisitor::GetMutableTensor(out_var).mutable_data(
                out_var_handle->place_);
      }

      int type = platform::ToNCCLDataType(in_tensor.type());
      size_t numel = static_cast<size_t>(in_tensor.numel());
      broadcast_calls.emplace_back(
          [send_recv_buffer, numel, type, root_id, &nccl_ctx] {
            PADDLE_ENFORCE(platform::dynload::ncclBcast(
                send_recv_buffer, numel, static_cast<ncclDataType_t>(type),
                root_id, nccl_ctx.comm_, nccl_ctx.stream()));
          });
    }

    this->RunAndRecordEvent([&] {
      {
        platform::NCCLGroupGuard guard;
        for (auto &call : broadcast_calls) {
          call();
        }
      }
      // TODO(zcd): Maybe the unequal operator is not appropriate here.
      if (*out_handle != *in_var_handle) {
        auto out_var = var_scopes.at(in_var_handle->scope_idx_)
                           ->FindVar(out_var_handles[0]->name_);
        paddle::framework::TensorCopy(
            in_tensor, in_var_handle->place_,
            *(dev_ctxes_.at(in_var_handle->place_)),
            &VariableVisitor::GetMutableTensor(out_var));
      }
    });
#else
    PADDLE_THROW("CUDA is not enabled.");
#endif
  }
}

void BroadcastOpHandle::WaitInputVarGenerated(const VarHandle &in_var) {
  if (in_var.generated_op_) {
    for (auto &pair : dev_ctxes_) {
      in_var.generated_op_->Wait(pair.second);
    }
  }
}

std::string BroadcastOpHandle::Name() const { return "broadcast"; }
}  // namespace details
}  // namespace framework
}  // namespace paddle