[ControlFlow] replace executor in run method of control flow ops with standalone_executor

paddle/fluid/framework/new_executor/interpreter/interpreter_util.cc

@@ -19,6 +19,7 @@
 #include "paddle/fluid/framework/details/nan_inf_utils.h"
 #include "paddle/fluid/framework/executor_gc_helper.h"
 #include "paddle/fluid/framework/new_executor/interpreter/data_transfer.h"
+#include "paddle/fluid/framework/new_executor/interpreter/execution_config.h"
 #include "paddle/fluid/memory/stats.h"
 #include "paddle/fluid/operators/controlflow/conditional_block_op_helper.h"
 #include "paddle/fluid/operators/controlflow/recurrent_op_helper.h"
@@ -298,7 +299,7 @@ std::tuple<VariableValueMap, VariableIdMap> BuildVariableMap(
     const VariableNameMap& var_name_map,
     VariableScope* var_scope,
     Scope* local_scope,
-    bool allow_var_not_in_program = false,
+    bool find_var_recursively = false,
     bool allow_var_not_in_scope = false) {
   VariableValueMap name2var;
   VariableIdMap name2id;
@@ -311,7 +312,7 @@ std::tuple<VariableValueMap, VariableIdMap> BuildVariableMap(
       auto* var = local_scope->FindVar(var_name);
 
       if (!var_scope->HasVar(var_name)) {
-        if (allow_var_not_in_program && var) {
+        if (find_var_recursively && var) {
           VLOG(3) << "Add " << var_name << " to var_scope";
           var_scope->AddVar(var_name, nullptr);
         } else if (allow_var_not_in_scope) {
@@ -427,17 +428,16 @@ void BuildOpFuncList(const platform::Place& place,
                      const std::set<std::string>& skip_gc_vars,
                      std::vector<OpFuncNode>* vec_func_list,
                      VariableScope* var_scope,
-                     bool use_local_scope,
-                     bool used_for_jit,
-                     bool used_for_control_flow_op) {
+                     ExecutionConfig execution_config,
+                     bool use_local_scope) {
   Scope* local_scope = use_local_scope ? var_scope->GetMutableLocalScope()
                                        : var_scope->GetMutableScope();
   std::vector<std::unique_ptr<OperatorBase>>
       ops_unique;  // its elements will be moved to vec_func_list
   // Step 1: create all ops for current block.
   CreateAllOps(block, &ops_unique);
 
-  if (!used_for_jit) {
+  if (!execution_config.used_for_jit) {
     // If gc is enabled and block size > 1
     const ProgramDesc& main_program = *block.Program();
     operators::PrepareSafeEagerDeletionOnConditionalOpAndConditionalGradOp(
@@ -490,17 +490,16 @@ void BuildOpFuncList(const platform::Place& place,
 
     // ops in the control flow block may not find its inputs or outputs
     // in VariableScope of the sub-block, so we need search it in parent scope.
-    allow_var_not_in_program =
-        used_for_control_flow_op || allow_var_not_in_program;
 
     framework::VariableNameMap& input_name_map = op->Inputs();
     VariableValueMap ins_map;
     VariableIdMap ins_name2id;
-    std::tie(ins_map, ins_name2id) = BuildVariableMap(input_name_map,
-                                                      var_scope,
-                                                      local_scope,
-                                                      allow_var_not_in_program,
-                                                      allow_var_not_in_scope);
+    std::tie(ins_map, ins_name2id) = BuildVariableMap(
+        input_name_map,
+        var_scope,
+        local_scope,
+        execution_config.used_for_control_flow_op || allow_var_not_in_program,
+        allow_var_not_in_scope);
 
     framework::VariableNameMap& output_name_map = op->Outputs();
     VariableValueMap outs_map;
@@ -509,7 +508,7 @@ void BuildOpFuncList(const platform::Place& place,
         BuildVariableMap(output_name_map,
                          var_scope,
                          local_scope,
-                         allow_var_not_in_program,
+                         execution_config.used_for_control_flow_op,
                          allow_var_not_in_scope);
 
     // step 1: build OpFuncNode

paddle/fluid/framework/new_executor/interpreter/interpreter_util.h

@@ -24,6 +24,7 @@
 
 #include "paddle/fluid/framework/executor_gc_helper.h"
 #include "paddle/fluid/framework/garbage_collector.h"
+#include "paddle/fluid/framework/new_executor/interpreter/execution_config.h"
 #include "paddle/fluid/framework/new_executor/new_executor_defs.h"
 #include "paddle/fluid/framework/new_executor/workqueue/workqueue.h"
 #include "paddle/fluid/framework/new_executor/workqueue/workqueue_utils.h"
@@ -75,9 +76,8 @@ void BuildOpFuncList(const platform::Place& place,
                      const std::set<std::string>& skip_gc_vars,
                      std::vector<OpFuncNode>* vec_func_list,
                      VariableScope* scope,
-                     bool use_local_scope = true,
-                     bool used_for_jit = false,
-                     bool used_for_control_flow_op = false);
+                     ExecutionConfig execution_config,
+                     bool use_local_scope = true);
 
 void AddFetch(const std::vector<std::string>& fetch_names,
               framework::BlockDesc* block);

paddle/fluid/framework/new_executor/interpretercore.cc

@@ -115,7 +115,8 @@ InterpreterCore::InterpreterCore(const platform::Place& place,
                                  const BlockDesc& block,
                                  const std::set<std::string>& skip_gc_vars,
                                  framework::Scope* scope,
-                                 bool used_for_jit)
+                                 bool used_for_jit,
+                                 bool used_for_control_flow_op)
     : place_(place),
       block_(block),
       execution_config_(place, block.OpSize()),
@@ -127,8 +128,10 @@ InterpreterCore::InterpreterCore(const platform::Place& place,
   completion_notifier_ = main_thread_blocker_.RegisterEvent(kTaskCompletion);
 
   execution_config_.used_for_jit = used_for_jit;
-  execution_config_.create_local_scope =
-      !used_for_jit && FLAGS_new_executor_use_local_scope;
+  execution_config_.used_for_control_flow_op = used_for_control_flow_op;
+  execution_config_.create_local_scope = !used_for_jit &&
+                                         FLAGS_new_executor_use_local_scope &&
+                                         !used_for_control_flow_op;
   execution_config_.skip_gc_vars = skip_gc_vars;
   execution_config_.Log(/*log_level=*/8);
 
@@ -251,9 +254,8 @@ paddle::framework::FetchList InterpreterCore::Run(
         execution_config_.skip_gc_vars,
         &op_func_nodes,
         &var_scope_,
-        HasLocalScope(),
-        execution_config_.used_for_jit,
-        execution_config_.used_for_control_flow_op);
+        execution_config_,
+        HasLocalScope());
     SetFeedVarsInplaceSkip(feed_names);
     // convert vec func_list to graph
     Convert(&op_func_nodes);
@@ -1121,9 +1123,8 @@ void InterpreterCore::Prepare(const std::vector<std::string>& feed_names,
         execution_config_.skip_gc_vars,
         &op_func_nodes,
         &var_scope_,
-        HasLocalScope(),
-        execution_config_.used_for_jit,
-        execution_config_.used_for_control_flow_op);
+        execution_config_,
+        HasLocalScope());
     SetFeedVarsInplaceSkip(feed_names);
     // convert vec func_list to graph
     Convert(&op_func_nodes);

paddle/fluid/framework/new_executor/interpretercore.h

@@ -47,7 +47,8 @@ class InterpreterCore {
                   const BlockDesc& block,
                   const std::set<std::string>& skip_gc_vars,
                   Scope* scope,
-                  bool used_for_jit = false);
+                  bool used_for_jit = false,
+                  bool used_for_control_flow_op = false);
 
   ~InterpreterCore();
 
@@ -74,13 +75,6 @@ class InterpreterCore {
 
   const platform::Place& GetPlace() const { return place_; }
 
-  void SetUsedForControlFlowOp(bool new_value) {
-    execution_config_.used_for_control_flow_op = new_value;
-  }
-  bool UsedForControlFlowOp() const {
-    return execution_config_.used_for_control_flow_op;
-  }
-
  private:
   // build graph
   void Convert(std::vector<paddle::framework::OpFuncNode>* op_func_nodes);

paddle/fluid/operators/controlflow/conditional_block_op.cc

@@ -49,16 +49,16 @@ static void BuildScopeForConditionalBlockOp(
     if (var_name == framework::kEmptyVarName) {
       continue;
     }
-    VLOG(10) << "[BuildScopeForConditionalBlockOp]"
-             << "start:" << var_name;
+    VLOG(5) << "[BuildScopeForConditionalBlockOp]"
+            << "start:" << var_name;
     if (var_desc->Persistable()) {
-      VLOG(10) << "[BuildScopeForConditionalBlockOp]"
-               << "Don't process persistent: " << var_name;
+      VLOG(5) << "[BuildScopeForConditionalBlockOp]"
+              << "Don't process persistent: " << var_name;
     } else {
       auto *ptr = scope->Var(var_name);
       InitializeVariable(ptr, var_desc->GetType());
-      VLOG(10) << "[BuildScopeForConditionalBlockOp]"
-               << "Not Found locally and created: " << var_name;
+      VLOG(5) << "[BuildScopeForConditionalBlockOp]"
+              << "Not Found locally and created: " << var_name;
     }
   }
 
@@ -121,6 +121,8 @@ class ConditionalBlockOp : public ConditionalOp {
         scopes->front() = &scope.NewScope();
       }
 
+      // We need to know whether the scope we cached is still valid.
+      // If not, we need to create a new one.
       if (scope.kids().size() == 0) {
         scopes->front() = &scope.NewScope();
       }
@@ -141,19 +143,19 @@ class ConditionalBlockOp : public ConditionalOp {
 
       if (FLAGS_control_flow_use_new_executor) {
         std::set<std::string> skip_gc_vars(skip_vars.begin(), skip_vars.end());
-        bool old_flag_new_executor_use_loacl_scope =
-            FLAGS_new_executor_use_local_scope;
-        FLAGS_new_executor_use_local_scope = false;
 
         if (!core || !platform::is_same_place(core->GetPlace(), dev_place) ||
             !FLAGS_control_flow_use_new_executor_cache) {
           VLOG(10) << "[interpreterCore cache]" << core.get();
           VLOG_IF(10, core)
               << platform::is_same_place(core->GetPlace(), dev_place);
           VLOG(10) << FLAGS_control_flow_use_new_executor_cache;
-          core.reset(new InterpreterCore(
-              dev_place, *block, skip_gc_vars, &cur_scope, false));
-          core->SetUsedForControlFlowOp(true);
+          core.reset(new InterpreterCore(dev_place,
+                                         *block,
+                                         skip_gc_vars,
+                                         &cur_scope,
+                                         /* used_for_jit */ false,
+                                         /* used_for_control_flow_op */ true));
           VLOG(10) << "[interpreterCore cache]"
                    << "new created:" << core;
         } else {
@@ -163,9 +165,6 @@ class ConditionalBlockOp : public ConditionalOp {
 
         core->Run({}, false);
 
-        FLAGS_new_executor_use_local_scope =
-            old_flag_new_executor_use_loacl_scope;
-
       } else {
         if (!exec || !platform::is_same_place(exec->GetPlace(), dev_place)) {
           auto &pdesc = *block->Program();
@@ -257,19 +256,19 @@ class ConditionalBlockGradOp : public ConditionalOp {
       if (FLAGS_control_flow_use_new_executor) {
         std::set<std::string> skip_gc_vars(inside_grads.begin(),
                                            inside_grads.end());
-        bool old_flag_new_executor_use_loacl_scope =
-            FLAGS_new_executor_use_local_scope;
-        FLAGS_new_executor_use_local_scope = false;
 
         if (!core || !platform::is_same_place(core->GetPlace(), dev_place) ||
             !FLAGS_control_flow_use_new_executor_cache) {
           VLOG(10) << "[interpreterCore cache]" << core.get();
           VLOG_IF(10, core)
               << platform::is_same_place(core->GetPlace(), dev_place);
           VLOG(10) << FLAGS_control_flow_use_new_executor_cache;
-          core.reset(new InterpreterCore(
-              dev_place, *block, skip_gc_vars, &cur_scope, false));
-          core->SetUsedForControlFlowOp(true);
+          core.reset(new InterpreterCore(dev_place,
+                                         *block,
+                                         skip_gc_vars,
+                                         &cur_scope,
+                                         /* used_for_jit */ false,
+                                         /* used_for_control_flow_op */ true));
           VLOG(10) << "[interpreterCore cache]"
                    << "new created:" << core;
         } else {
@@ -278,9 +277,6 @@ class ConditionalBlockGradOp : public ConditionalOp {
         }
         core->Run({}, false);
 
-        FLAGS_new_executor_use_local_scope =
-            old_flag_new_executor_use_loacl_scope;
-
       } else {
         if (!exec || !platform::is_same_place(exec->GetPlace(), dev_place)) {
           auto &pdesc = *block->Program();