In current design, this function will not be called in parallel since one node is only scheduled by one scheduler queue
| 331 | // In current design, this function will not be called in parallel |
| 332 | // since one node is only scheduled by one scheduler queue |
| 333 | int Node::process_node(Task &task) { |
| 334 | if (state_ == NodeState::CLOSED || state_ == NodeState::PAUSE_DONE) { |
| 335 | dec_pending_task(); |
| 336 | return 0; |
| 337 | } |
| 338 | |
| 339 | #ifndef NO_TRACE |
| 340 | TraceProcessEmitter trace_emitter = |
| 341 | TraceProcessEmitter(PROCESSING, node_name_); |
| 342 | #endif |
| 343 | |
| 344 | // TODO check the task is valid |
| 345 | |
| 346 | last_timestamp_ = task.timestamp(); |
| 347 | // std::cout<<"timestamp="<<last_timestamp_<<std::endl; |
| 348 | // task timestamp is eof means all inputs are done, but at this moment |
| 349 | // the outputs might not be done, so we can schedule the node as source |
| 350 | //(no inputs, like source node) until all outputs done |
| 351 | if (not is_source() and task.timestamp() == BMF_EOF) { |
| 352 | BMFLOG_NODE(BMF_INFO, id_) << "process eof, add node to scheduler"; |
| 353 | // become source node |
| 354 | set_source(true); |
| 355 | // add to scheduler |
| 356 | // callback_.throttled_cb(id_, true); |
| 357 | // callback_.sched_required(id_, false); |
| 358 | } |
| 359 | // used only for SERVER mode |
| 360 | // receive eos means that the node is ready to close |
| 361 | // before close we should propagate eos pkt to the downstream |
| 362 | else if (task.timestamp() == EOS) { |
| 363 | dec_pending_task(); |
| 364 | BMFLOG_NODE(BMF_INFO, id_) << "meet eos, close the node and propagate " |
| 365 | "eos pkt instead of node process"; |
| 366 | for (auto &iter : task.get_outputs()) { |
| 367 | iter.second->push(Packet::generate_eos_packet()); |
| 368 | } |
| 369 | output_stream_manager_->post_process(task); |
| 370 | close(); |
| 371 | return 0; |
| 372 | } |
| 373 | |
| 374 | auto cleanup = [this] { |
| 375 | this->task_processed_cnt_++; |
| 376 | this->dec_pending_task(); |
| 377 | BMFLOG_NODE(BMF_ERROR, this->id_) << "Process node failed, will exit."; |
| 378 | }; |
| 379 | |
| 380 | int result = 0; |
| 381 | try { |
| 382 | opt_reset_mutex_.lock(); |
| 383 | if (need_opt_reset_) { |
| 384 | module_->dynamic_reset(reset_option_); |
| 385 | need_opt_reset_ = false; |
| 386 | } |
| 387 | opt_reset_mutex_.unlock(); |
| 388 | |
| 389 | BMF_TRACE_PROCESS(module_name_.c_str(), "process", START); |
| 390 | state_ = NodeState::RUNNING; |