graph.cc

/* 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 <algorithm>
#include <unordered_set>

#include "paddle/fluid/framework/ir/graph.h"
#include "paddle/fluid/framework/op_proto_maker.h"
#include "paddle/fluid/framework/program_desc.h"
#include "paddle/fluid/framework/var_desc.h"

namespace paddle {
namespace framework {
namespace ir {

std::vector<std::string> FindDistTrainSendVars(
    const std::vector<ir::Node *> &nodes) {
  std::vector<std::string> send_vars;
  // since parameters are all in block 0,
  // it's enough to only scan send ops in block 0
  for (auto &node : nodes) {
    auto op_vars = node->Op()->InputArgumentNames();
    send_vars.reserve(send_vars.size() +
                      std::distance(op_vars.begin(), op_vars.end()));
    send_vars.insert(send_vars.end(), op_vars.begin(), op_vars.end());
  }
  return send_vars;
}

std::vector<std::string> FindDistTrainRecvVars(
    const std::vector<ir::Node *> &nodes) {
  std::vector<std::string> recv_vars;
  for (auto &node : nodes) {
    auto op_vars = node->Op()->OutputArgumentNames();
    recv_vars.reserve(recv_vars.size() +
                      std::distance(op_vars.begin(), op_vars.end()));
    recv_vars.insert(recv_vars.end(), op_vars.begin(), op_vars.end());
  }
  return recv_vars;
}

bool IsDistTrainOp(ir::Node *node, const std::vector<std::string> &send_vars,
                   const std::vector<std::string> &recv_vars) {
  if (send_vars.size() == 0 || recv_vars.size() == 0) {
    return false;
  }

  /**
   * Check any of opvars contains `.block` and in sendvars
   */
  auto checker = [](const std::vector<std::string> &opvars,
                    const std::vector<std::string> &rpc_vars) -> bool {
    for (auto &var : opvars) {
      // a variable name with the suffix `.block` means it's a splited
      // variable by (DistributeTranspiler)
      // [python/paddle/fluid/transpiler/distribute_transpiler.py]
      if (var.find(".block") != std::string::npos &&
          std::find(rpc_vars.begin(), rpc_vars.end(), var) != rpc_vars.end()) {
        return true;
      }
    }
    return false;
  };

  std::vector<std::string> input_var_names;
  std::vector<std::string> output_var_names;
  for (ir::Node *input : node->inputs) {
    input_var_names.push_back(input->Name());
  }
  for (ir::Node *output : node->outputs) {
    output_var_names.push_back(output->Name());
  }

  return checker(output_var_names, send_vars) ||
         checker(input_var_names, recv_vars);
}

Graph::Graph(const ProgramDesc &program) : program_(program) {
  VLOG(3) << "block in program:" << program_.Size();
  std::unordered_map<std::string, VarDesc *> all_vars;
  for (auto *var : program.Block(0).AllVars()) {
    all_vars.emplace(var->Name(), var);
  }

  std::map<std::string, std::vector<ir::Node *>> var_nodes;
  for (auto *op : program.Block(0).AllOps()) {
    ir::Node *node = CreateOpNode(op);
    // For input args, reuse the same var name if it was created before.
    // Otherwise, create a new one.
    for (auto &each_var_name : op->InputArgumentNames()) {
      ir::Node *var = nullptr;
      if (var_nodes.find(each_var_name) != var_nodes.end()) {
        var = var_nodes.at(each_var_name).back();
      } else if (all_vars.count(each_var_name) != 0) {
        var = CreateVarNode(all_vars.at(each_var_name));
        var_nodes[each_var_name].push_back(var);
      } else {
        // Operation input var can be optional (dispensable). Which means
        // the operation doesn't really need the var at runtime. In this
        // case, the no-existed var is ready at the beginning.
        var = CreateEmptyNode(each_var_name, ir::Node::Type::kVariable);
        var_nodes[each_var_name].push_back(var);
      }
      node->inputs.push_back(var);
      var->outputs.push_back(node);
    }
    // For output args, always create a new var.
    for (auto &each_var_name : op->OutputArgumentNames()) {
      ir::Node *var = CreateVarNode(all_vars.at(each_var_name));
      var_nodes[each_var_name].push_back(var);
      node->outputs.push_back(var);
      var->inputs.push_back(node);
    }
  }

  std::vector<ir::Node *> send_ops;
  ir::Node *send_bar = nullptr;
  std::vector<ir::Node *> recv_ops;
  ir::Node *fetch_bar = nullptr;
  for (ir::Node *node : Nodes()) {
    if (node->Name() == "send") {
      send_ops.push_back(node);
    } else if (node->Name() == "send_barrier") {
      PADDLE_ENFORCE(!send_bar, "only has one send barrier");
      send_bar = node;
    } else if (node->Name() == "recv") {
      recv_ops.push_back(node);
    } else if (node->Name() == "fetch_barrier") {
      PADDLE_ENFORCE(!fetch_bar, "only has one fetch barrier");
      fetch_bar = node;
    }
  }
  if (send_bar) {
    for (ir::Node *send : send_ops) {
      ir::Node *dep_var = CreateControlDepVar();
      send->outputs.push_back(dep_var);
      dep_var->inputs.push_back(send);
      send_bar->inputs.push_back(dep_var);
      dep_var->outputs.push_back(send_bar);
    }
    for (ir::Node *recv : recv_ops) {
      ir::Node *dep_var = CreateControlDepVar();
      recv->inputs.push_back(dep_var);
      dep_var->outputs.push_back(recv);
      send_bar->outputs.push_back(dep_var);
      dep_var->inputs.push_back(send_bar);
    }
  }
  if (fetch_bar) {
    for (ir::Node *recv : recv_ops) {
      ir::Node *dep_var = CreateControlDepVar();
      recv->outputs.push_back(dep_var);
      dep_var->inputs.push_back(recv);
      fetch_bar->inputs.push_back(dep_var);
      dep_var->outputs.push_back(fetch_bar);
    }
  }

  std::vector<std::string> send_vars = FindDistTrainSendVars(send_ops);
  std::vector<std::string> recv_vars = FindDistTrainRecvVars(recv_ops);
  for (ir::Node *node : Nodes()) {
    if (IsDistTrainOp(node, send_vars, recv_vars)) {
      if (fetch_bar && node->Name() == "concat") {
        ir::Node *dep_var = CreateControlDepVar();
        fetch_bar->outputs.push_back(dep_var);
        dep_var->inputs.push_back(fetch_bar);
        node->inputs.push_back(dep_var);
        dep_var->outputs.push_back(node);
      }
    }
  }

  /**
   * We only handle write after read(WAR), since it should not have a write
   * after write in program. If there are write after write operators, we need
   * prune them.
   *
   * https://en.wikipedia.org/wiki/Hazard_(computer_architecture)#Write_after_read_(WAR)
   */

  for (auto &var : var_nodes) {
    auto &versions = var.second;
    if (versions.size() <= 1) continue;

    auto it_new = versions.rbegin();
    auto it_old = versions.rbegin();
    ++it_old;
    for (; it_old != versions.rend(); it_new = it_old, ++it_old) {
      ir::Node *write_op =
          (*it_new)->inputs.empty() ? nullptr : (*it_new)->inputs[0];
      const auto &read_ops = (*it_old)->outputs;

      for (auto *read_op : read_ops) {
        // Manually add a dependency var from read_op to write_op;
        if (read_op == write_op) {
          // Read Write is the same op.
          continue;
        }
        // 2 ops might have been connected via other vars.
        bool has_dep = false;
        for (ir::Node *r_out : read_op->outputs) {
          for (ir::Node *w_in : write_op->inputs) {
            if (r_out == w_in) {
              has_dep = true;
              break;
            }
          }
        }
        if (has_dep) continue;

        ir::Node *dep_var = CreateControlDepVar();
        read_op->outputs.push_back(dep_var);
        dep_var->inputs.push_back(read_op);
        write_op->inputs.push_back(dep_var);
        dep_var->outputs.push_back(write_op);
      }
    }
  }
}

bool IsControlDepVar(const ir::Node &var) {
  return var.Name().find(ir::Node::kControlDepVarName) != std::string::npos;
}
}  // namespace ir
}  // namespace framework
}  // namespace paddle