| 195 | } |
| 196 | |
| 197 | int Node::reset() { |
| 198 | mutex_.lock(); |
| 199 | if (enable_stat_ && module_stat_data_.process_cnts > 0 && mode_ == BmfMode::SERVER_MODE) { |
| 200 | module_stat_data_.avg_processing_time = |
| 201 | static_cast<float>( |
| 202 | static_cast<double>(module_stat_data_.total_process_time) / |
| 203 | module_stat_data_.process_cnts |
| 204 | ); |
| 205 | JsonParam user_df; |
| 206 | if (module_->report_user_df_data(user_df)) |
| 207 | module_stat_data_.user_df = user_df.dump(); |
| 208 | module_stat_data_.node_id = id_; |
| 209 | module_stat_data_.module_name = module_name_; |
| 210 | module_stat_data_.graph_uuid = graph_uuid_; |
| 211 | bmf_stat_report(std::make_shared<ModuleData>(std::move(module_stat_data_))); |
| 212 | } |
| 213 | // reset module of this node |
| 214 | module_->reset(); |
| 215 | // remove from source nodes |
| 216 | set_source(false); |
| 217 | |
| 218 | // clear remaining node task from schedule queue |
| 219 | callback_.clear_cb(id_, get_scheduler_queue_id()); |
| 220 | |
| 221 | // reset all input_streams of node |
| 222 | for (auto &input_stream : input_stream_manager_->input_streams_) { |
| 223 | if (input_stream.second->get_block()) { |
| 224 | input_stream.second->set_block(false); |
| 225 | } |
| 226 | if (!input_stream.second->is_connected()) { |
| 227 | std::shared_ptr<SafeQueue<Packet> > q = std::make_shared<SafeQueue<Packet> >(); |
| 228 | q->push(Packet::generate_eof_packet()); |
| 229 | input_stream.second->add_packets(q); |
| 230 | } |
| 231 | } |
| 232 | |
| 233 | if (!all_input_queue_empty()) { // in some case such as server mode previous |
| 234 | // blocked |
| 235 | int res = input_stream_manager_->schedule_node(); |
| 236 | if (res) |
| 237 | schedule_node_success_cnt_++; |
| 238 | } |
| 239 | |
| 240 | mutex_.unlock(); |
| 241 | return 0; |
| 242 | } |
| 243 | |
| 244 | bool Node::is_closed() { return state_ == NodeState::CLOSED; } |
| 245 | |