ensure the order of regst producing and consuming (#1085)

oneflow/core/actor/actor.cpp

@@ -58,6 +58,7 @@ void Actor::Init(const TaskProto& task_proto, const ThreadCtx& thread_ctx) {
       writeable_produced_ctrl_regst_[regst->regst_desc_id()].push_back(regst.get());
       produced_ctrl_regst2reading_cnt_[regst.get()] = 0;
     }
+    produced_ctrl_regst2max_act_id_[pair.first] = act_id_;
   }
 
   // non ctrl regst
@@ -81,6 +82,7 @@ void Actor::Init(const TaskProto& task_proto, const ThreadCtx& thread_ctx) {
       writeable_produced_data_regst_[regst->regst_desc_id()].push_back(regst.get());
       produced_data_regst2reading_cnt_[regst.get()] = 0;
     }
+    produced_data_regst2max_act_id_[pair.first] = act_id_;
   }
   actual_writeable_produced_data_regst_desc_num_ = writeable_produced_data_regst_.size();
   writeable_produced_data_regst_desc_cnt_ = actual_writeable_produced_data_regst_desc_num_;
@@ -437,7 +439,14 @@ int Actor::ProcessWriteableCtrlRegstMsg(const ActorMsg& msg) {
   auto writeable_it = writeable_produced_ctrl_regst_.find(regst->regst_desc_id());
   CHECK(writeable_it != writeable_produced_ctrl_regst_.end());
   if (writeable_it->second.empty()) { writeable_ctrl_regst_desc_cnt_ += 1; }
-  writeable_it->second.push_back(msg.regst());
+  int64_t& max_act_id = produced_ctrl_regst2max_act_id_[regst->regst_desc_id()];
+  if (max_act_id >= 0) {
+    CHECK_GT(regst->act_id(), max_act_id);
+    max_act_id = regst->act_id();
+  } else if (regst->act_id() >= 0) {
+    max_act_id = regst->act_id();
+  }
+  writeable_it->second.push_back(regst);
   return 0;
 }
 
@@ -513,6 +522,13 @@ int Actor::TryUpdtStateAsProducedRegst(Regst* regst) {
   auto writeable_it = writeable_produced_data_regst_.find(regst->regst_desc_id());
   CHECK(writeable_it != writeable_produced_data_regst_.end());
   if (writeable_it->second.empty()) { writeable_produced_data_regst_desc_cnt_ += 1; }
+  int64_t& max_act_id = produced_data_regst2max_act_id_[regst->regst_desc_id()];
+  if (max_act_id >= 0) {
+    CHECK_GT(regst->act_id(), max_act_id);
+    max_act_id = regst->act_id();
+  } else if (regst->act_id() >= 0) {
+    max_act_id = regst->act_id();
+  }
   writeable_it->second.push_back(regst);
   return 0;
 }

oneflow/core/actor/actor.h

@@ -149,6 +149,7 @@ class Actor {
   // Status of Produced Registers
   HashMap<int64_t, std::vector<std::unique_ptr<Regst>>> produced_data_regsts_;
   HashMap<int64_t, std::deque<Regst*>> writeable_produced_data_regst_;
+  HashMap<int64_t, int64_t> produced_data_regst2max_act_id_;
   HashMap<Regst*, int64_t> produced_data_regst2reading_cnt_;
   int64_t actual_writeable_produced_data_regst_desc_num_;
   int64_t writeable_produced_data_regst_desc_cnt_;
@@ -162,6 +163,7 @@ class Actor {
   // Status of Control Registers
   HashMap<int64_t, std::vector<std::unique_ptr<Regst>>> produced_ctrl_regst_;
   HashMap<int64_t, std::deque<Regst*>> writeable_produced_ctrl_regst_;
+  HashMap<int64_t, int64_t> produced_ctrl_regst2max_act_id_;
   HashMap<Regst*, int64_t> produced_ctrl_regst2reading_cnt_;
   HashMap<int64_t, std::deque<Regst*>> consumed_ctrl_regst_;
   int64_t total_reading_ctrl_cnt_;