add bp_actor_id in fw_data_comp_actor

oneflow/core/actor/fw_data_comp_actor.cpp

@@ -29,6 +29,7 @@ void FwDataCompActor::Init(const TaskProto& task_proto,
         1 + (model_regst_desc_id_ != -1) + (model_tmp_regst_desc_id_ != -1);
     OF_SET_MSG_HANDLE(&FwDataCompActor::HandleNormal);
   }
+  bp_actor_id_ = IDMgr::Singleton()->ActorId4TaskId(task_proto.bp_task_id());
 }
 
 bool FwDataCompActor::IsReadReady() {

oneflow/core/actor/fw_data_comp_actor.h

@@ -32,6 +32,7 @@ class FwDataCompActor final : public CompActor {
   std::queue<std::shared_ptr<RegstWrapper>> in_;
   HashMap<int64_t, std::shared_ptr<RegstWrapper>> ready_in_regst_;
   KernelCtx kernel_ctx_;
+  int64_t bp_actor_id_;
 };
 
 }  // namespace oneflow

oneflow/core/graph/data_comp_task_node.h

@@ -16,8 +16,12 @@ class DataCompTaskNode final : public CompTaskNode {
     proto->set_parallel_policy(chain_node()->parallel_desc()->policy());
     proto->set_parallel_id(parallel_id());
     proto->set_parallel_num(chain_node()->parallel_desc()->parallel_num());
-    if (IsFwNode() && GetBpNode()) {
-      proto->set_bp_task_id(GetBpNode()->task_id());
+    if (IsFwNode()) {
+      if (TaskNode* bp_node = GetBpNode()) {
+        proto->set_bp_task_id(bp_node->task_id());
+      } else {
+        proto->set_bp_task_id(-1);
+      }
     }
   }