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

fix bug: runtime filter core because of data racing

Browse Source
This commit is contained in:
obdev
2024-02-10 06:53:44 +00:00
committed by ob-robot
parent f8c2bab01d
commit 81270342f7
2 changed files with 5 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
src/sql/engine/px/p2p_datahub/ob_p2p_dh_mgr.cpp
View File

@ -418,6 +418,7 @@ int ObP2PDatahubManager::P2PMsgSetCall::operator() (const common::hash::HashMapP
K(dh_key_), K(dh_msg_.get_dm_cb_node_seq_id()));
if (OB_FAIL(dh_msg_.regenerate())) {
LOG_WARN("failed to do regen_call", K(dh_key_));
} else if (FALSE_IT(dh_msg_.check_finish_receive())) {
}
}
if (OB_FAIL(ret)) {

7
src/sql/engine/px/p2p_datahub/ob_p2p_dh_msg.cpp
View File

@ -126,6 +126,9 @@ int ObP2PDatahubMsgBase::process_msg_internal(bool &need_free)
ObP2PDatahubMsgGuard guard(this);
bool need_merge = true;
// to avoid data racing in the process of check_finish_receive, protect it in hashmap lock
// if set succ, check_finish_receive() in P2PMsgSetCall
// if set failed(with OB_HASH_EXIST, need merge), check_finish_receive() in P2PMsgMergeCall
if (OB_FAIL(map.set_refactored(dh_key, this, 0/*flag*/, 0/*broadcast*/, 0/*overwrite_key*/, &set_call))) {
if (OB_HASH_EXIST == ret) {
ret = OB_SUCCESS;
@ -145,9 +148,7 @@ int ObP2PDatahubMsgBase::process_msg_internal(bool &need_free)
}
}
}
if (OB_SUCC(ret) && !need_merge) {
(void)check_finish_receive();
}
if (need_free) {
// msg not in map, dec ref count
guard.dec_msg_ref_count();