hcq/oceanbase
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

fix wait init_sqc response core

Browse Source
This commit is contained in:
sdc
2024-01-23 14:18:02 +00:00
committed by ob-robot
parent 55be7bf8fa
commit e158ab5e5f
2 changed files with 15 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
src/sql/engine/px/ob_px_sqc_async_proxy.cpp
View File

@ -251,16 +251,20 @@ void ObPxSqcAsyncProxy::fail_process() {
K(return_cb_count_), K(callbacks_.count()));
ARRAY_FOREACH_X(callbacks_, idx, count, true) {
ObSqcAsyncCB &callback = *callbacks_.at(idx);
if (!callback.is_visited() &&
!(callback.is_processed() || callback.is_timeout() || callback.is_invalid())) {
// unregister async callbacks that have not received response.
ObAsyncRespCallback *async_cb = static_cast<ObAsyncRespCallback *>(callback.low_level_cb_);
uint64_t gtid = async_cb->gtid_;
uint32_t pkt_id = async_cb->pkt_id_;
int err = 0;
if ((err = pn_terminate_pkt(gtid, pkt_id)) != 0) {
int ret = tranlate_to_ob_error(err);
LOG_WARN("terminate pkt failed", K(ret), K(err));
{
// avoid rpc thread access the callback currently.
ObThreadCondGuard guard(callback.get_cond());
if (!callback.is_visited() &&
!(callback.is_processed() || callback.is_timeout() || callback.is_invalid())) {
// unregister async callbacks that have not received response.
ObAsyncRespCallback *async_cb = static_cast<ObAsyncRespCallback *>(callback.low_level_cb_);
uint64_t gtid = async_cb->gtid_;
uint32_t pkt_id = async_cb->pkt_id_;
int err = 0;
if ((err = pn_terminate_pkt(gtid, pkt_id)) != 0) {
int ret = tranlate_to_ob_error(err);
LOG_WARN("terminate pkt failed", K(ret), K(err));
}
}
}
}