There are two cases which will run into schedule_node, so we need a mutex here 1. while an input packet is added to input_stream, this is called by input_stream_manager 2. while node is in PENDING or all input streams done, it's scheduled by scheduler */
| 467 | scheduler |
| 468 | */ |
| 469 | bool Node::schedule_node() { |
| 470 | #ifndef NO_TRACE |
| 471 | TraceProcessEmitter trace_emitter = |
| 472 | TraceProcessEmitter(SCHEDULE, node_name_); |
| 473 | #endif |
| 474 | mutex_.lock(); |
| 475 | // BMFLOG_NODE(BMF_INFO, id_) << "scheduling..."; |
| 476 | schedule_node_cnt_++; |
| 477 | if (state_ == NodeState::PAUSE_DONE) { |
| 478 | mutex_.unlock(); |
| 479 | return false; |
| 480 | } |
| 481 | // we don't need schedule node if it's closed or set to pause |
| 482 | if (state_ == NodeState::CLOSED || |
| 483 | (wait_pause_ == true && pending_tasks_ != 0)) { |
| 484 | mutex_.unlock(); |
| 485 | return false; |
| 486 | } |
| 487 | // for node that need force_close, check if all downstream nodes are closed |
| 488 | // already |
| 489 | // if yes, set it to closed, and remove it from node scheduler |
| 490 | if (need_force_close()) { |
| 491 | if (all_downstream_nodes_closed()) { |
| 492 | close(); |
| 493 | mutex_.unlock(); |
| 494 | BMFLOG_NODE(BMF_INFO, id_) |
| 495 | << "scheduling failed, all downstream node closed: " << type_; |
| 496 | return false; |
| 497 | } |
| 498 | } |
| 499 | |
| 500 | bool result = false; |
| 501 | if (is_source()) { |
| 502 | Task task = Task(id_, input_stream_manager_->stream_id_list_, |
| 503 | output_stream_manager_->get_stream_id_list()); |
| 504 | if (infinity_node_) { |
| 505 | task.set_timestamp(INF_SRC); |
| 506 | } else { |
| 507 | task.set_timestamp(get_source_timestamp()); |
| 508 | } |
| 509 | callback_.scheduler_cb(task); |
| 510 | schedule_node_success_cnt_++; |
| 511 | result = true; |
| 512 | } else { |
| 513 | if (node_output_updated_) { |
| 514 | input_stream_manager_->output_stream_id_list_ = |
| 515 | output_stream_manager_->get_stream_id_list(); |
| 516 | node_output_updated_ = false; |
| 517 | } |
| 518 | result = input_stream_manager_->schedule_node(); |
| 519 | if (result) |
| 520 | schedule_node_success_cnt_++; |
| 521 | } |
| 522 | mutex_.unlock(); |
| 523 | return result; |
| 524 | } |
| 525 | |
| 526 | void Node::wait_paused() { |