| 333 | } |
| 334 | |
| 335 | int Actor::HandlerNormal(const ActorMsg& msg) { |
| 336 | if (msg.msg_type() == ActorMsgType::kEordMsg) { |
| 337 | remaining_eord_cnt_ -= 1; |
| 338 | CHECK(eord_regst_desc_ids_.insert(msg.eord_regst_desc_id()).second); |
| 339 | if (naive_consumed_rs_.HasRegstDescId(msg.eord_regst_desc_id())) { |
| 340 | is_naive_consumed_eord_ = true; |
| 341 | } else if (inplace_consumed_rs_.HasRegstDescId(msg.eord_regst_desc_id())) { |
| 342 | is_inplace_consumed_eord_ = true; |
| 343 | } else { |
| 344 | NormalProcessCustomizedEordMsg(msg); |
| 345 | } |
| 346 | } else if (msg.msg_type() == ActorMsgType::kRegstMsg) { |
| 347 | if (msg.SrcMachineId() == GlobalProcessCtx::Rank()) { |
| 348 | Regst* regst = msg.regst(); |
| 349 | if (naive_consumed_rs_.HasRegstDescId(regst->regst_desc_id())) { |
| 350 | CHECK_EQ(0, naive_consumed_rs_.TryPushBackRegst(regst)); |
| 351 | const auto& rdeq = naive_consumed_rs_.RegstDeq4RegstDescId(regst->regst_desc_id()); |
| 352 | CHECK(rdeq.empty() == false); |
| 353 | if (rdeq.front()->regst_desc()->regst_desc_type().has_data_regst_desc()) { |
| 354 | NormalProcessNaiveReadableDataRegstMsg(rdeq); |
| 355 | } |
| 356 | } else if (inplace_consumed_rs_.HasRegstDescId(regst->regst_desc_id())) { |
| 357 | CHECK_EQ(0, inplace_consumed_rs_.TryPushBackRegst(regst)); |
| 358 | } else if (TryUpdtStateAsProducedRegst(regst) == 0) { |
| 359 | // do nothing |
| 360 | } else { |
| 361 | NormalProcessCustomizedReadableRegstMsg(msg); |
| 362 | } |
| 363 | } else { |
| 364 | if (NormalTryProcessReadableMsgFromOtherMachine(msg) == false) { |
| 365 | // process ctrl msg from other rank |
| 366 | if (IsConsumedCtrlRegstDescId(msg.regst_desc_id())) { |
| 367 | Regst* regst = msg.regst(); |
| 368 | CHECK(naive_consumed_rs_.HasRegstDescId(msg.regst_desc_id())); |
| 369 | CHECK(Singleton<RegstMgr>::Get()->HasProducerTaskId4RegstDescId(msg.regst_desc_id())); |
| 370 | CHECK_EQ(0, naive_consumed_rs_.TryPushBackRegst(regst, msg.regst_desc_id())); |
| 371 | const auto& rdeq = naive_consumed_rs_.RegstDeq4RegstDescId(msg.regst_desc_id()); |
| 372 | CHECK(rdeq.empty() == false); |
| 373 | } else { |
| 374 | CHECK_EQ(TryUpdtStateAsProducedRegst(msg.regst()), 0); |
| 375 | } |
| 376 | } |
| 377 | } |
| 378 | |
| 379 | if (debug_) { |
| 380 | LOG(INFO) << " Actor: " << actor_id_ << " op: " << op_name_ << " in act_cnt: [ " << act_cnt_ |
| 381 | << " ] , Recv ActorMsg from: " << msg.src_actor_id() |
| 382 | << " to: " << msg.dst_actor_id() << " with regst: " << msg.regst_desc_id(); |
| 383 | } |
| 384 | ActUntilFail(); |
| 385 | } else if (msg.msg_type() == ActorMsgType::kCmdMsg) { |
| 386 | CHECK_EQ(msg.actor_cmd(), ActorCmd::kStart); |
| 387 | ActUntilFail(); |
| 388 | } else { |
| 389 | UNIMPLEMENTED(); |
| 390 | } |
| 391 | // handler halts |
| 392 | bool has_naive_or_inplace = naive_consumed_rs_.total_regst_desc_cnt() != 0 |
nothing calls this directly
no test coverage detected