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:
obdev
2024-02-10 00:31:53 +00:00
committed by ob-robot
parent 081db61e75
commit e76674237a
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())); K(return_cb_count_), K(callbacks_.count()));
ARRAY_FOREACH_X(callbacks_, idx, count, true) { ARRAY_FOREACH_X(callbacks_, idx, count, true) {
ObSqcAsyncCB &callback = *callbacks_.at(idx); ObSqcAsyncCB &callback = *callbacks_.at(idx);
if (!callback.is_visited() && {
!(callback.is_processed() || callback.is_timeout() || callback.is_invalid())) { // avoid rpc thread access the callback currently.
// unregister async callbacks that have not received response. ObThreadCondGuard guard(callback.get_cond());
ObAsyncRespCallback *async_cb = static_cast<ObAsyncRespCallback *>(callback.low_level_cb_); if (!callback.is_visited() &&
uint64_t gtid = async_cb->gtid_; !(callback.is_processed() || callback.is_timeout() || callback.is_invalid())) {
uint32_t pkt_id = async_cb->pkt_id_; // unregister async callbacks that have not received response.
int err = 0; ObAsyncRespCallback *async_cb = static_cast<ObAsyncRespCallback *>(callback.low_level_cb_);
if ((err = pn_terminate_pkt(gtid, pkt_id)) != 0) { uint64_t gtid = async_cb->gtid_;
int ret = tranlate_to_ob_error(err); uint32_t pkt_id = async_cb->pkt_id_;
LOG_WARN("terminate pkt failed", K(ret), K(err)); 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));
}
} }
} }
} }

1
src/sql/engine/px/ob_px_sqc_async_proxy.h
View File

@ -62,6 +62,7 @@ public:
const ObPxUserErrorMsg get_ret_code() const { return rcode_; } const ObPxUserErrorMsg get_ret_code() const { return rcode_; }
const common::ObAddr &get_dst() const { return dst_; } const common::ObAddr &get_dst() const { return dst_; }
int64_t get_timeout() const { return timeout_; } int64_t get_timeout() const { return timeout_; }
ObThreadCond &get_cond() { return cond_; }
// to string // to string
TO_STRING_KV("dst", get_dst(), "timeout", get_timeout(), "ret_code", TO_STRING_KV("dst", get_dst(), "timeout", get_timeout(), "ret_code",
get_ret_code(), "result", get_result(), "is_visited", get_ret_code(), "result", get_result(), "is_visited",