Merge branch 'develop' into opt_fetch

Author: wanghuancoder

cmake/configure.cmake

@@ -20,10 +20,6 @@ if(WITH_TESTING)
     add_definitions(-DPADDLE_WITH_TESTING)
 endif(WITH_TESTING)
 
-if(WITH_INFERENCE_API_TEST)
-    add_definitions(-DPADDLE_WITH_INFERENCE_API_TEST)
-endif(WITH_INFERENCE_API_TEST)
-
 if(NOT WITH_PROFILER)
     add_definitions(-DPADDLE_DISABLE_PROFILER)
 endif(NOT WITH_PROFILER)

paddle/fluid/framework/executor_gc_helper.cc

@@ -31,49 +31,6 @@
 namespace paddle {
 namespace framework {
 
-struct OpInOutInfo {
- public:
-  void Build(const OperatorBase *op) {
-    is_built_ = true;
-    auto &inferer = op->Info().NoNeedBufferVarsInferer();
-    if (inferer) {
-      no_need_buffer_ins_ = inferer(op->Inputs(), op->Outputs(), op->Attrs());
-
-      if (no_need_buffer_ins_.empty()) return;
-
-      for (auto &in_name_pair : op->Inputs()) {
-        if (no_need_buffer_ins_.count(in_name_pair.first) != 0) {
-          continue;
-        }
-
-        for (auto &in_arg_name : in_name_pair.second) {
-          other_args_set_.insert(in_arg_name);
-        }
-      }
-
-      for (auto &out_name_pair : op->Outputs()) {
-        for (auto &out_arg_name : out_name_pair.second) {
-          other_args_set_.insert(out_arg_name);
-        }
-      }
-    }
-  }
-
-  bool IsBuilt() const { return is_built_; }
-
-  bool IsInArgBufferNeeded(const std::string &in_arg_name) const {
-    return no_need_buffer_ins_.empty() ||
-           other_args_set_.count(in_arg_name) != 0;
-  }
-
- private:
-  // A set to record unused buffer input vars of op
-  std::unordered_set<std::string> no_need_buffer_ins_;
-  // A set to record other args of op (including in, out)
-  std::unordered_set<std::string> other_args_set_;
-  bool is_built_{false};
-};
-
 static bool VarCanBeDeleted(const std::string &name, const BlockDesc &block,
                             const std::unordered_set<std::string> &skip_vars) {
   if (skip_vars.count(name) != 0) {

paddle/fluid/framework/executor_gc_helper.h

@@ -31,6 +31,49 @@ class GarbageCollector;
 class OperatorBase;
 class Scope;
 
+struct OpInOutInfo {
+ public:
+  void Build(const OperatorBase *op) {
+    is_built_ = true;
+    auto &inferer = op->Info().NoNeedBufferVarsInferer();
+    if (inferer) {
+      no_need_buffer_ins_ = inferer(op->Inputs(), op->Outputs(), op->Attrs());
+
+      if (no_need_buffer_ins_.empty()) return;
+
+      for (auto &in_name_pair : op->Inputs()) {
+        if (no_need_buffer_ins_.count(in_name_pair.first) != 0) {
+          continue;
+        }
+
+        for (auto &in_arg_name : in_name_pair.second) {
+          other_args_set_.insert(in_arg_name);
+        }
+      }
+
+      for (auto &out_name_pair : op->Outputs()) {
+        for (auto &out_arg_name : out_name_pair.second) {
+          other_args_set_.insert(out_arg_name);
+        }
+      }
+    }
+  }
+
+  bool IsBuilt() const { return is_built_; }
+
+  bool IsInArgBufferNeeded(const std::string &in_arg_name) const {
+    return no_need_buffer_ins_.empty() ||
+           other_args_set_.count(in_arg_name) != 0;
+  }
+
+ private:
+  // A set to record unused buffer input vars of op
+  std::unordered_set<std::string> no_need_buffer_ins_;
+  // A set to record other args of op (including in, out)
+  std::unordered_set<std::string> other_args_set_;
+  bool is_built_{false};
+};
+
 std::unordered_map<const OperatorBase *, std::vector<std::string>>
 GetUnusedVars(const BlockDesc &block,
               const std::vector<std::unique_ptr<OperatorBase>> &ops,

paddle/fluid/framework/new_executor/CMakeLists.txt

@@ -1,9 +1,8 @@
+cc_library(workqueue SRCS workqueue.cc)
 cc_library(interpretercore SRCS interpretercore.cc DEPS op_registry
             device_context scope framework_proto data_feed_proto heter_service_proto trainer_desc_proto glog
             lod_rank_table fs shell fleet_wrapper heter_wrapper ps_gpu_wrapper box_wrapper lodtensor_printer feed_fetch_method
-            graph_to_program_pass variable_helper timer monitor)
+            graph_to_program_pass variable_helper timer monitor workqueue device_event device_event_gpu)
 cc_library(standalone_executor SRCS standalone_executor.cc DEPS interpretercore)
-cc_library(workqueue SRCS workqueue.cc)
-
 cc_test(workqueue_test SRCS workqueue_test.cc DEPS workqueue)
 # cc_binary(standalone_executor_test SRCS standalone_executor_test.cc DEPS interpretercore standalone_executor operator op_registry executor ${GLOB_OP_LIB} ${GLOB_OPERATOR_DEPS} profiler)

paddle/fluid/framework/new_executor/interpretercore.cc

@@ -12,6 +12,13 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 #include "paddle/fluid/framework/new_executor/interpretercore.h"
+#include "paddle/fluid/framework/executor_gc_helper.h"
+#include "paddle/fluid/framework/new_executor/interpretercore_gc_helper.h"
+
+#if defined(PADDLE_WITH_CUDA)
+using ::paddle::platform::kCUDA;
+USE_EVENT(kCUDA);
+#endif
 
 #include <unordered_set>
 
@@ -145,6 +152,12 @@ InterpreterCore::InterpreterCore(const platform::Place& place,
       d2h_ctx_pool_({place}),
       h2d_ctx_pool_({place}) {
   is_build_ = false;
+
+  garbages_.reset(new GarbageQueue());
+  max_memory_size_ = static_cast<size_t>(GetEagerDeletionThreshold());
+  cur_memory_size_ = 0;
+  gc_queue_ = CreateSingleThreadedWorkQueue();
+
   feed_names_ = feed_names;
 
   // Step1: add feedop and fetchop to main_program
@@ -215,11 +228,24 @@ void InterpreterCore::Convert() {
     temp_inst.input_index_ = vec_func_list_[i].input_index;
     temp_inst.output_index_ = vec_func_list_[i].output_index;
 
+    OpInOutInfo info;
+
     std::vector<size_t> gc_check_input_list;
     for (auto& item : vec_func_list_[i].input_index) {
       for (auto id : item.second) {
         input_var2op_info_[id].push_back(i);
-        gc_check_input_list.push_back(id);
+        // var can be gc-ed
+        if (!info.IsBuilt()) {
+          info.Build(op_list_[i]);
+        }
+        if (global_scope_->vec_meta_info_[id].vardesc_) {
+          if (info.IsInArgBufferNeeded(
+                  global_scope_->vec_meta_info_[id].vardesc_->Name())) {
+            gc_check_input_list.push_back(id);
+          }
+        } else {
+          gc_check_input_list.push_back(id);
+        }
       }
     }
     std::sort(gc_check_input_list.begin(), gc_check_input_list.end());
@@ -236,6 +262,13 @@ void InterpreterCore::Convert() {
   }
 
   for (size_t i = 0; i < vec_instruction_.size(); ++i) {
+#if defined(PADDLE_WITH_CUDA)
+    int device_type = static_cast<int>(paddle::platform::DeviceType::CUDA);
+    paddle::platform::DeviceOption dev_opt(
+        device_type, BOOST_GET_CONST(platform::CUDAPlace, place_).device);
+    gc_event_.emplace_back(dev_opt);
+#endif
+
     std::vector<size_t> vec_temp;
     for (auto& item : vec_instruction_[i].output_index_) {
       for (auto id : item.second) {
@@ -365,11 +398,8 @@ void InterpreterCore::ExecuteInstructionList(
     }
 
     // GC infomation
-
-    auto& gc_check_list = instr_node.gc_check_var_list;
-    for (auto var_id : gc_check_list) {
-      --working_var_ref[var_id].var_ref_count_;
-    }
+    CheckGC(instr_id, instr_node.gc_check_var_list, var_scope, place,
+            working_var_ref);
   }
 
   for (size_t i = 0; i < working_var_ref.size(); ++i) {
@@ -379,6 +409,87 @@ void InterpreterCore::ExecuteInstructionList(
   }
 }
 
+void InterpreterCore::CheckGC(size_t instr_id,
+                              const std::vector<size_t>& gc_check_list,
+                              const VariableScope& var_scope,
+                              const platform::Place& place,
+                              std::vector<VariableMetaInfo>& working_var_ref) {
+  for (auto var_id : gc_check_list) {
+    --working_var_ref[var_id].var_ref_count_;
+    if (var_scope.vec_meta_info_[var_id].vardesc_ &&
+        !var_scope.vec_meta_info_[var_id].vardesc_->Persistable() &&
+        working_var_ref[var_id].var_ref_count_ == 0) {
+      Variable* var = var_scope.var_list[var_id];
+      if (var->IsType<LoDTensor>()) {
+        garbages_->emplace_back(
+            var->GetMutable<LoDTensor>()->MoveMemoryHolder());
+        if (garbages_->back()) {
+          cur_memory_size_ += garbages_->back()->size();
+        }
+      } else if (var->IsType<SelectedRows>()) {
+        garbages_->emplace_back(var->GetMutable<SelectedRows>()
+                                    ->mutable_value()
+                                    ->MoveMemoryHolder());
+        if (garbages_->back()) {
+          cur_memory_size_ += garbages_->back()->size();
+        }
+      } else if (var->IsType<LoDTensorArray>()) {
+        auto* tensor_arr = var->GetMutable<LoDTensorArray>();
+        for (auto& t : *tensor_arr) {
+          garbages_->emplace_back(t.MoveMemoryHolder());
+          if (garbages_->back()) {
+            cur_memory_size_ += garbages_->back()->size();
+          }
+        }
+      } else {
+        PADDLE_THROW(platform::errors::Unimplemented(
+            "The variable(%s) is not supported in eager deletion.",
+            framework::ToTypeName(var->Type())));
+      }
+    }
+  }
+
+  if (!garbages_->empty()) {
+    if (max_memory_size_ <= 1) {
+#if defined(PADDLE_WITH_CUDA)
+      auto* dev_ctx = reinterpret_cast<platform::CUDADeviceContext*>(
+          platform::DeviceContextPool::Instance().Get(place));
+      gc_event_[instr_id].Record(place, dev_ctx);
+      gc_queue_->AddTask(
+          [ container = garbages_.release(), event = &gc_event_[instr_id] ]() {
+            while (!event->Query()) {
+              continue;
+            }
+            delete container;
+          });
+      garbages_.reset(new GarbageQueue());
+#else
+      delete garbages_.release();
+      garbages_.reset(new GarbageQueue());
+#endif
+    } else if (cur_memory_size_ >= max_memory_size_) {
+#if defined(PADDLE_WITH_CUDA)
+      auto* dev_ctx = reinterpret_cast<platform::CUDADeviceContext*>(
+          platform::DeviceContextPool::Instance().Get(place));
+      gc_event_[instr_id].Record(place, dev_ctx);
+      gc_queue_->AddTask(
+          [ container = garbages_.release(), event = &gc_event_[instr_id] ]() {
+            while (!event->Query()) {
+              continue;
+            }
+            delete container;
+          });
+      garbages_.reset(new GarbageQueue());
+      cur_memory_size_ = 0;
+#else
+      delete garbages_.release();
+      garbages_.reset(new GarbageQueue());
+      cur_memory_size_ = 0;
+#endif
+    }
+  }
+}
+
 std::vector<size_t> InterpreterCore::MergeVector(
     const std::vector<size_t>& first, const std::vector<size_t>& second) {
   std::vector<size_t> out(first.size() + second.size());
@@ -407,6 +518,11 @@ void InterpreterCore::BuildVariableScope(const framework::ProgramDesc& pdesc,
       auto v = new Variable();
       InitializeVariable(v, var->GetType());
       var_scope->var_list.push_back(v);
+
+      VariableMetaInfo info;
+      info.var_ref_count_ = 0;
+      info.vardesc_ = var;
+      var_scope->vec_meta_info_.push_back(info);
     }
   }
 }
@@ -419,6 +535,7 @@ void InterpreterCore::BuildOpFuncList(const platform::Place& place,
   auto& global_block = pdesc.Block(0);
   auto& all_op_kernels = OperatorWithKernel::AllOpKernels();
 
+  std::vector<OperatorBase*> ops;
   for (auto& op : global_block.AllOps()) {
     VLOG(3) << "Build OpFuncNode from : " << op->Type();
 
@@ -434,6 +551,20 @@ void InterpreterCore::BuildOpFuncList(const platform::Place& place,
     // step 1. Prepare VariableValueMap of input/output
     auto op_base =
         info.Creator()(op->Type(), inputs_names, outputs_names, op_attr_map);
+    ops.push_back(op_base);
+  }
+
+  auto unused_var_map = get_unused_vars(global_block, ops);
+
+  size_t ops_index = 0;
+  for (auto& op : global_block.AllOps()) {
+    VLOG(3) << op->Type();
+    // << op->Type() << endl;
+
+    auto op_base = ops[ops_index++];
+
+    auto inputs_names = op->Inputs();
+    auto outputs_names = op->Outputs();
 
     VariableValueMap ins_map;
     std::map<std::string, std::vector<int>> ins_name2id;
@@ -542,6 +673,11 @@ void InterpreterCore::BuildOpFuncList(const platform::Place& place,
           var_scope->name2id[new_var_name] = var_scope->var_list.size();
           var_scope->var_list.push_back(v);
 
+          VariableMetaInfo info;
+          info.var_ref_count_ = 0;
+          info.vardesc_ = nullptr;
+          var_scope->vec_meta_info_.push_back(info);
+
           VariableNameMap copy_in_map;
           auto x_iter = inputs_names.find(var_name_item.first);
           copy_in_map["X"] = {x_iter->second[i]};
@@ -647,6 +783,47 @@ void InterpreterCore::BuildOpFuncList(const platform::Place& place,
     op_func_node.kernel_func_ = OpKernelComputeFunc(kernel_iter->second);
     op_func_node.kernel_func_(exec_ctx);
     vec_func_list->push_back(op_func_node);
+
+    // gc---------------------------------------------------------------------------
+    auto iter = unused_var_map.find(op_base);
+    if (iter == unused_var_map.end()) {
+      continue;
+    }
+
+    auto& delete_vars = iter->second;
+    std::deque<std::shared_ptr<memory::Allocation>>* garbages =
+        new std::deque<std::shared_ptr<memory::Allocation>>();
+
+    for (auto& var_name : delete_vars) {
+      auto it = var_scope->name2id.find(var_name);
+      assert(it != var_scope->name2id.end());
+      auto* var = var_scope->var_list[it->second];
+      if (var == nullptr) {
+        continue;
+      }
+
+      VLOG(2) << "Erase variable " << var_name;
+      if (var->IsType<LoDTensor>()) {
+        garbages->emplace_back(
+            var->GetMutable<LoDTensor>()->MoveMemoryHolder());
+      } else if (var->IsType<SelectedRows>()) {
+        garbages->emplace_back(var->GetMutable<SelectedRows>()
+                                   ->mutable_value()
+                                   ->MoveMemoryHolder());
+      } else if (var->IsType<LoDTensorArray>()) {
+        auto* lod_tensor_arr = var->GetMutable<LoDTensorArray>();
+        for (auto& t : *lod_tensor_arr) {
+          garbages->emplace_back(t.MoveMemoryHolder());
+        }
+      } else {
+        PADDLE_THROW(platform::errors::Unimplemented(
+            "Type %s of variable %s is not supported eager deletion.",
+            framework::ToTypeName(var->Type()), var_name));
+      }
+    }
+
+    delete garbages;  // free mem
+
     VLOG(3) << "run " << op_base->Type() << " done.";
   }
 }