Fix graph processnode destruction bug

[Discovery-Kingdom]

以下代码存在析构bug
void Node::Stop() {
if (!m_run) {
return;
}
m_run = false;
std::for_each(m_input_buffers.begin(), m_input_buffers.end(),
[&](const auto &item) { item.second->clear(); });
std::for_each(m_output_buffers.begin(), m_output_buffers.end(),
[&](const auto &item) { item.second->clear(); });
m_base_cond->notify_all();
if (m_worker.joinable()) {
m_worker.join();
}
if (exit_cb) {
exit_cb(getName(), StatusCode::NodeExit, "节点线程退出");
}
std::cout << getName() << " exit" << std::endl;
}

void Node::worker() {
while (m_run) {
auto datas = get_input_datas();
if (!datas->batch_data.empty()) {
if (batch_data_handler_hooker) {
auto res = batch_data_handler_hooker(datas);
send_output_datas(res);
continue;
}
auto &batch_data = datas->batch_data;
for (auto &data : batch_data) {
if (data_handler_hooker) {
data = data_handler_hooker(data);
} else {
data = handle_data(data);
}
send_output_data(data);
}
} else {
std::unique_lockstd::mutex lk(m_base_mutex);
m_base_cond->wait(lk);
continue;
}
}
}

修复方式
void Node::Stop() {
if (!m_run) {
return;
}
{
std::unique_lockstd::mutex lk(m_base_mutex);
m_run = false;
}
std::for_each(m_input_buffers.begin(), m_input_buffers.end(),
[&](const auto &item) { item.second->clear(); });
std::for_each(m_output_buffers.begin(), m_output_buffers.end(),
[&](const auto &item) { item.second->clear(); });
m_base_cond->notify_all();
if (m_worker.joinable()) {
m_worker.join();
}
if (exit_cb) {
exit_cb(getName(), StatusCode::NodeExit, "节点线程退出");
}
std::cout << getName() << " exit" << std::endl;
}

void Node::worker() {
while (m_run) {
auto datas = get_input_datas();
if (!datas->batch_data.empty()) {
if (batch_data_handler_hooker) {
auto res = batch_data_handler_hooker(datas);
send_output_datas(res);
continue;
}
auto &batch_data = datas->batch_data;
for (auto &data : batch_data) {
if (data_handler_hooker) {
data = data_handler_hooker(data);
} else {
data = handle_data(data);
}
send_output_data(data);
}
} else {
std::unique_lockstd::mutex lk(m_base_mutex);
if(!m_run)
{
break;
}
m_base_cond->wait(lk);
continue;
}
}
}