fix bug: parallel_ctx init

Former-commit-id: 76cc22d3

oneflow/core/actor/actor.cpp

@@ -4,6 +4,9 @@ namespace oneflow {
 
 void Actor::Init(const TaskProto& task_proto, const ThreadCtx& thread_ctx) {
   actor_id_ = task_proto.task_id();
+  if (task_proto.has_parallel_ctx()) {
+    parallel_ctx_.reset(new ParallelContext(task_proto.parallel_ctx()));
+  }
   for (const ExecNodeProto& node : task_proto.exec_sequence().exec_node()) {
     ExecKernel ek;
     ek.kernel =

oneflow/core/actor/actor.h

@@ -36,7 +36,7 @@ class Actor {
 
   // Util
   Actor() = default;
-  virtual const ParallelContext* parallel_ctx() const { return nullptr; }
+  const ParallelContext* parallel_ctx() const { return parallel_ctx_.get(); }
   DeviceType GetDeviceType() const;
   virtual void VirtualActorInit(const TaskProto&) {}
   int64_t RegstDescId4Name(const std::string& name) const;
@@ -90,6 +90,7 @@ class Actor {
 
  private:
   int64_t actor_id_;
+  std::unique_ptr<ParallelContext> parallel_ctx_;
   std::vector<ExecKernel> exec_kernel_vec_;
   HashMap<int64_t, std::vector<std::unique_ptr<Regst>>> produced_regsts_;
   HashMap<std::string, int64_t> name2regst_desc_id_;

oneflow/core/actor/compute_actor.h

@@ -15,17 +15,10 @@ class CompActor : public Actor {
 
   virtual void VirtualCompActorInit(const TaskProto& task_proto) {}
 
-  const ParallelContext* parallel_ctx() const override {
-    return &parallel_ctx_;
-  }
-
  private:
   void VirtualActorInit(const TaskProto& task_proto) override {
-    parallel_ctx_ = task_proto.parallel_ctx();
     VirtualCompActorInit(task_proto);
   }
-
-  ParallelContext parallel_ctx_;
 };
 
 inline int64_t GetLastPieceIdForModelVersionId(int64_t model_version_id) {