Avoid data race in grpc alarm (#10124) - #11030
Conversation
Signed-off-by: ti-chi-bot <ti-community-prow-bot@tidb.io>
ti-chi-bot
added
do-not-merge/hold
|
This cherry pick PR is for a release branch and has not yet been approved by triage owners. To merge this cherry pick:
DetailsInstructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. |
|
@solotzg This PR has conflicts, I have hold it. |
|
@ti-chi-bot: ## If you want to know how to resolve it, please read the guide in TiDB Dev Guide. DetailsInstructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the ti-community-infra/tichi repository. |
|
[APPROVALNOTIFIER] This PR is NOT APPROVED This pull-request has been approved by: The full list of commands accepted by this bot can be found here. DetailsNeeds approval from an approver in each of these files:Approvers can indicate their approval by writing |
📝 WalkthroughWalkthroughThe change adds a call-owned gRPC alarm accessor. MPP gather tracking stores references to these alarms and uses them during tunnel waiting and alarm cancellation. The waiting-tunnel block contains unresolved merge-conflict markers. ChangesAlarm lifetime handling
Estimated code review effort: 3 (Moderate) | ~20 minutes Poem
🚥 Pre-merge checks | ✅ 3 | ❌ 2❌ Failed checks (2 warnings)
✅ Passed checks (3 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@dbms/src/Flash/Mpp/MPPTaskManager.cpp`:
- Around line 250-259: Resolve the conflict in the MPP task setup by removing
all merge markers and retaining the race-safe alarm retrieval via
EstablishCallData::getAlarm() and std::ref(alarm). Update EstablishCallData’s
public API in EstablishCall.h so the call uses the existing public
setToWaitingTunnelState() behavior, or otherwise exposes an equivalent valid
transition without referencing private WAIT_TUNNEL or unavailable
setCallStateAndUpdateMetrics().
- Line 258: Resolve the remaining merge-conflict markers in findAsyncTunnel(),
preserving the intended gather_task_set->alarms insertion involving
sender_task_id, receiver_task_id, and alarm. Remove all conflict markers and
ensure the resulting C++ block is syntactically valid and compilable.
In `@dbms/src/Flash/Mpp/MPPTaskManager.h`:
- Line 46: Update the alarms storage in MPPTaskManager and the abortMPPGather
cancellation flow so each grpc::Alarm remains owned, along with its
EstablishCallData owner, until the completion queue removes the alarm; do not
erase a non-owning reference while the alarm may still be used. Preserve
cancellation behavior and run the ComputeServerRunner_testErrorMessage
ThreadSanitizer reproduction after resolving the merge conflict.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 4960dcca-c719-43ec-a1a3-67cb32d02089
📒 Files selected for processing (4)
dbms/src/Flash/EstablishCall.cppdbms/src/Flash/EstablishCall.hdbms/src/Flash/Mpp/MPPTaskManager.cppdbms/src/Flash/Mpp/MPPTaskManager.h
| <<<<<<< HEAD | ||
| auto & alarm = gather_task_set->alarms[sender_task_id][receiver_task_id]; | ||
| call_data->setToWaitingTunnelState(); | ||
| ======= | ||
| auto & alarm = call_data->getAlarm(); | ||
| call_data->setCallStateAndUpdateMetrics( | ||
| EstablishCallData::WAIT_TUNNEL, | ||
| GET_METRIC(tiflash_establish_calldata_count, type_wait_tunnel_calldata)); | ||
| gather_task_set->alarms[sender_task_id].emplace(receiver_task_id, std::ref(alarm)); | ||
| >>>>>>> 262b942077 (Avoid data race in grpc alarm (#10124)) |
There was a problem hiding this comment.
🎯 Functional Correctness | 🔴 Critical | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
file="dbms/src/Flash/Mpp/MPPTaskManager.cpp"
if rg -n '^(<<<<<<<|=======|>>>>>>>)' "$file"; then
echo "Unresolved merge conflict markers found" >&2
exit 1
fi
rg -n -C 3 \
'setCallStateAndUpdateMetrics|WAIT_TUNNEL|setToWaitingTunnelState|getAlarm|alarms.*emplace' \
dbms/src/Flash/EstablishCall.h \
"$file"Repository: pingcap/tiflash
Length of output: 286
🏁 Script executed:
#!/bin/bash
set -u
echo "== file section =="
sed -n '230,280p' dbms/src/Flash/Mpp/MPPTaskManager.cpp 2>/dev/null || true
echo "== EstablishCall.h related section =="
sed -n '1,220p' dbms/src/Flash/EstablishCall.h 2>/dev/null || true
echo "== other relevant files =="
git ls-files | rg 'dbms/src/Flash/.*\.(h|cpp|hpp)$' | sed -n '1,120p'
echo "== search for API names =="
rg -n 'setCallStateAndUpdateMetrics|WAIT_TUNNEL|setToWaitingTunnelState|getAlarm|type_wait_tunnel_calldata|tiflash_establish_calldata_count' dbms/src/Flash 2>/dev/null || trueRepository: pingcap/tiflash
Length of output: 14831
🏁 Script executed:
#!/bin/bash
set -u
echo "== MPPTaskManager outline == "
ast-grep outline dbms/src/Flash/Mpp/MPPTaskManager.cpp --view compact 2>/dev/null || true
echo "== alarms declarations and usages in MPPTaskManager.cpp =="
rg -n -C 2 'gather_task_set->alarms|sender_task_id|receiver_task_id|\.Set\(|std::ref\(alarm\)|type_wait_tunnel_calldata' dbms/src/Flash/Mpp/MPPTaskManager.cpp
echo "== EstablishCall related symbols =="
rg -n -C 2 'WAIT_TUNNEL|getAlarm|setToWaitingTunnelState|setCallStateAndUpdateMetrics|type_wait_tunnel_calldata|tiflash_establish_calldata_count|GET_METRIC' dbms/src/Flash/EstablishCall.h dbms/src/Flash/EstablishCall.cpp dbms/src/Metrics 2>/dev/null || true
echo "== changed conflict metadata =="
git diff --stat || true
git diff --name-status || trueRepository: pingcap/tiflash
Length of output: 6465
Resolve the unresolved merge conflict at MPPTaskManager.cpp:250-259.
The conflict markers remain in the build file, so MPPTaskManager.cpp does not compile. The incoming branch also calls setCallStateAndUpdateMetrics() and EstablishCallData::WAIT_TUNNEL, but dbms/src/Flash/EstablishCall.h exposes setToWaitingTunnelState() and keeps WAIT_TUNNEL private; getAlarm() is available, so keep std::ref(alarm) as intended and update the public API instead.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@dbms/src/Flash/Mpp/MPPTaskManager.cpp` around lines 250 - 259, Resolve the
conflict in the MPP task setup by removing all merge markers and retaining the
race-safe alarm retrieval via EstablishCallData::getAlarm() and std::ref(alarm).
Update EstablishCallData’s public API in EstablishCall.h so the call uses the
existing public setToWaitingTunnelState() behavior, or otherwise exposes an
equivalent valid transition without referencing private WAIT_TUNNEL or
unavailable setCallStateAndUpdateMetrics().
| call_data->setCallStateAndUpdateMetrics( | ||
| EstablishCallData::WAIT_TUNNEL, | ||
| GET_METRIC(tiflash_establish_calldata_count, type_wait_tunnel_calldata)); | ||
| gather_task_set->alarms[sender_task_id].emplace(receiver_task_id, std::ref(alarm)); |
There was a problem hiding this comment.
🩺 Stability & Availability | 🔴 Critical | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
rg -n -C 8 \
'EstablishMPPConnection|findAsyncTunnel|sender_meta\(\).*task_id|receiver_meta\(\).*task_id' .Repository: pingcap/tiflash
Length of output: 50026
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== MPPTaskManager relevant sections =="
sed -n '170,255p;360,395p' dbms/src/Flash/Mpp/MPPTaskManager.cpp
echo
echo "== MPPTaskManager.h gather_task_set / alarm declarations =="
sed -n '180,230p' dbms/src/Flash/Mpp/MPPTaskManager.h
echo
echo "== Find MPPGatherTaskSet definition and alarm methods =="
rg -n -C 4 'struct MPPGatherTaskSet|class MPPGatherTaskSet|alarms|hasAlarm|notifyAlarm|Cancel\\(' dbms/src/Flash/MppRepository: pingcap/tiflash
Length of output: 6992
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== MPPTaskManager.h declarations =="
sed -n '1,140p' dbms/src/Flash/Mpp/MPPTaskManager.h
sed -n '140,190p' dbms/src/Flash/Mpp/MPPTaskManager.h
sed -n '190,270p' dbms/src/Flash/Mpp/MPPTaskManager.h
echo
echo "== findAsyncTunnel full and concurrent merge markers =="
sed -n '204,300p' dbms/src/Flash/Mpp/MPPTaskManager.cpp
echo
echo "== alarm-related symbols and Cancel references =="
rg -n -C 3 'alarms|hasAlarm|notifyAlarm|setAlarm|Cancel\(' dbms/src/Flash dbms/src/Core 2>/dev/null || true
echo
echo "== compile marker check =="
python3 - <<'PY'
from pathlib import Path
p=Path('dbms/src/Flash/Mpp/MPPTaskManager.cpp')
for i,line in enumerate(p.read_text().splitlines(),1):
if line.startswith('<'):
print(i, line)
PYRepository: pingcap/tiflash
Length of output: 25014
Resolve the merge conflict in findAsyncTunnel().
dbms/src/Flash/Mpp/MPPTaskManager.cpp:250-259 still contains unresolved conflict markers, so this translation unit will not compile.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@dbms/src/Flash/Mpp/MPPTaskManager.cpp` at line 258, Resolve the remaining
merge-conflict markers in findAsyncTunnel(), preserving the intended
gather_task_set->alarms insertion involving sender_task_id, receiver_task_id,
and alarm. Remove all conflict markers and ensure the resulting C++ block is
syntactically valid and compilable.
| String error_message; | ||
| /// <sender_task_id, <receiver_task_id, alarm>> | ||
| std::unordered_map<Int64, std::unordered_map<Int64, grpc::Alarm>> alarms; | ||
| std::unordered_map<Int64, std::unordered_map<Int64, std::reference_wrapper<grpc::Alarm>>> alarms; |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
rg -n -C 8 \
'delete this|EstablishCallData::~|getAlarm\(\)|alarms|cancelAlarmsBySenderTaskId|abortMPPGather|Cancel\(\)' \
dbms/src/Flash/EstablishCall.h \
dbms/src/Flash/EstablishCall.cpp \
dbms/src/Flash/Mpp/MPPTaskManager.h \
dbms/src/Flash/Mpp/MPPTaskManager.cpp
rg -n -C 4 \
'ComputeServerRunner_testErrorMessage|testErrorMessage|ThreadSanitizer|TSAN' .Repository: pingcap/tiflash
Length of output: 31996
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect the relevant class definitions, methods, and exact call sites using small targeted reads.
printf '--- MPPTaskManager.h outline ---\n'
ast-grep outline dbms/src/Flash/Mpp/MPPTaskManager.h --view compact || true
printf '--- MPPTaskManager.h relevant lines ---\n'
sed -n '1,110p' dbms/src/Flash/Mpp/MPPTaskManager.h
sed -n '240,275p' dbms/src/Flash/Mpp/MPPTaskManager.h
printf '--- MPPTaskManager.cpp relevant lines ---\n'
sed -n '35,70p' dbms/src/Flash/Mpp/MPPTaskManager.cpp
sed -n '230,310p' dbms/src/Flash/Mpp/MPPTaskManager.cpp
sed -n '344,405p' dbms/src/Flash/Mpp/MPPTaskManager.cpp
sed -n '488,525p' dbms/src/Flash/Mpp/MPPTaskManager.cpp
printf '--- EstablishCall.h relevant lines ---\n'
sed -n '1,130p' dbms/src/Flash/EstablishCall.h
printf '--- EstablishCall.cpp relevant lines ---\n'
sed -n '62,145p' dbms/src/Flash/EstablishCall.cpp
sed -n '140,190p' dbms/src/Flash/EstablishCall.cpp
sed -n '270,340p' dbms/src/Flash/EstablishCall.cpp
printf '--- Tests relevant lines ---\n'
sed -n '1354,1430p' dbms/src/Flash/tests/gtest_compute_server.cpp
printf '--- Deterministic control-flow extraction ---\n'
python3 - <<'PY'
from pathlib import Path
for path in [
Path('dbms/src/Flash/Mpp/MPPTaskManager.cpp'),
Path('dbms/src/Flash/Mpp/MPPTaskManager.h'),
Path('dbms/src/Flash/EstablishCall.cpp'),
Path('dbms/src/Flash/EstablishCall.h'),
Path('dbms/src/Flash/tests/gtest_compute_server.cpp'),
]:
text = path.read_text()
print(f'## {path}')
for needle in [
'std::reference_wrapper<grpc::Alarm> alarms',
'cancelAlarmsBySenderTaskId',
'markTaskAsFinishedOrFailed',
'abortMPPGather',
'alarm.second.get().Cancel()',
'gather_task_set->alarms.clear()',
'gather_task_set->alarms[sender_task_id].emplace(receiver_task_id, std::ref(alarm))',
'call_data->getAlarm()',
'delete this;',
'tryConnectTunnel()',
'async_tunnel_sender->setResult',
'ComputeServerRunner',
'testErrorMessage',
]:
idx = text.find(needle)
if idx == -1:
continue
line = text[:idx].count('\n') + 1
print(f'line {line}: {needle}')
PYRepository: pingcap/tiflash
Length of output: 244
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect the relevant class definitions, methods, and exact call sites using small targeted reads.
printf '%s\n' '--- MPPTaskManager.h outline ---'
ast-grep outline dbms/src/Flash/Mpp/MPPTaskManager.h --view compact || true
printf '%s\n' '--- MPPTaskManager.h relevant lines ---'
sed -n '1,110p' dbms/src/Flash/Mpp/MPPTaskManager.h
sed -n '240,275p' dbms/src/Flash/Mpp/MPPTaskManager.h
printf '%s\n' '--- MPPTaskManager.cpp relevant lines ---'
sed -n '35,70p' dbms/src/Flash/Mpp/MPPTaskManager.cpp
sed -n '230,310p' dbms/src/Flash/Mpp/MPPTaskManager.cpp
sed -n '344,405p' dbms/src/Flash/Mpp/MPPTaskManager.cpp
sed -n '488,525p' dbms/src/Flash/Mpp/MPPTaskManager.cpp
printf '% s\n' '--- EstablishCall.h relevant lines ---'
sed -n '1,130p' dbms/src/Flash/EstablishCall.h
printf '%s\n' '--- EstablishCall.cpp relevant lines ---'
sed -n '62,145p' dbms/src/Flash/EstablishCall.cpp
sed -n '140,190p' dbms/src/Flash/EstablishCall.cpp
sed -n '270,340p' dbms/src/Flash/EstablishCall.cpp
printf '% s\n' '--- Tests relevant lines ---'
sed -n '1354,1430p' dbms/src/Flash/tests/gtest_compute_server.cpp
printf '%s\n' '--- Deterministic control-flow extraction ---'
python3 - <<'PY'
from pathlib import Path
for path in [
Path('dbms/src/Flash/Mpp/MPPTaskManager.cpp'),
Path('dbms/src/Flash/Mpp/MPPTaskManager.h'),
Path('dbms/src/Flash/EstablishCall.cpp'),
Path('dbms/src/Flash/EstablishCall.h'),
Path('dbms/src/Flash/tests/gtest_compute_server.cpp'),
]:
text = path.read_text()
print(f'## {path}')
for needle in [
'std::reference_wrapper<grpc::Alarm> alarms',
'cancelAlarmsBySenderTaskId',
'markTaskAsFinishedOrFailed',
'abortMPPGather',
'alarm.second.get().Cancel()',
'gather_task_set->alarms.clear()',
'gather_task_set->alarms[sender_task_id].emplace(receiver_task_id, std::ref(alarm))',
'call_data->getAlarm()',
'delete this;',
'tryConnectTunnel()',
'async_tunnel_sender->setResult',
'ComputeServerRunner',
'testErrorMessage',
]:
idx = text.find(needle)
if idx == -1:
continue
line = text[:idx].count('\n') + 1
print(f'line {line}: {needle}')
PYRepository: pingcap/tiflash
Length of output: 30566
🏁 Script executed:
#!/bin/bash
set -euo pipesse
printf '%s\n' '--- EstablishCall.h tail members ---'
sed -n '130,185p' dbms/src/Flash/EstablishCall.h
printf '%s\n' '--- EstablishCall.cpp writeDone ---'
sed -n '310,380p' dbms/src/Flash/EstablishCall.cpp
printf '%s\n' '--- MPPTaskManager.cpp getMPPQueryAndGatherTaskSet ---'
sed -n '310,350p' dbms/src/Flash/Mpp/MPPTaskManager.cpp
sed -n '525,580p' dbms/src/Flash/Mpp/MPPTaskManager.cpp
printf '%s\n' '--- MPPTaskManager.h gather definitions ---'
sed -n '53,180p' dbms/src/Flash/Mpp/MPPTaskManager.h
printf '%s\n' '--- cancelGather calls ---'
rg -n -C 6 'cancelGather' .Repository: pingcap/tiflash
Length of output: 201
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '%s\n' '--- EstablishCall.h tail members ---'
sed -n '130,185p' dbms/src/Flash/EstablishCall.h
printf '%s\n' '--- EstablishCall.cpp writeDone ---'
sed -n '310,380p' dbms/src/Flash/EstablishCall.cpp
printf '%s\n' '--- MPPTaskManager.cpp cleanup helpers ---'
sed -n '310,350p' dbms/src/Flash/Mpp/MPPTaskManager.cpp
sed -n '525,580p' dbms/src/Flash/Mpp/MPPTaskManager.cpp
printf '%s\n' '--- MPPTaskManager.h gather definitions ---'
sed -n '53,180p' dbms/src/Flash/Mpp/MPPTaskManager.h
printf '%s\n' '--- cancelGather calls ---'
rg -n -C 6 'cancelGather' .Repository: pingcap/tiflash
Length of output: 32258
Keep the stored grpc::Alarm references owned while cancelling.
std::reference_wrapper does not extend the EstablishCallData lifetime, and MpPTaskManager::abortMPPGather() cancels and erases the wrappers even while the owned EstablishCallData::alarm can still be used by the completion queue. Store the alarms/owner until the alarm is removed, or add an object lifetime handle such that a wrapper can reach the alarm after abortMPPGather() erases it. Run the ComputeServerRunner_testErrorMessage ThreadSanitizer reproduction after resolving the merge conflict.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@dbms/src/Flash/Mpp/MPPTaskManager.h` at line 46, Update the alarms storage in
MPPTaskManager and the abortMPPGather cancellation flow so each grpc::Alarm
remains owned, along with its EstablishCallData owner, until the completion
queue removes the alarm; do not erase a non-owning reference while the alarm may
still be used. Preserve cancellation behavior and run the
ComputeServerRunner_testErrorMessage ThreadSanitizer reproduction after
resolving the merge conflict.
|
@ti-chi-bot: The following tests failed, say
Full PR test history. Your PR dashboard. DetailsInstructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. I understand the commands that are listed here. |
3 participants
This is an automated cherry-pick of #10124
What problem does this PR solve?
Issue Number: close #10107
Problem Summary:
In current implementation,
Alarmis hold byMPPGatherTaskSet, and inMPPTaskManager::abortMPPGather, all theAlarmwill be deconstructed since it callgather_task_set->alarms.clear();But when
gather_task_set->alarms.clear();is called, theEstablishCallDatamay still inside grpc's core, and it hold a raw pointer ofAlarmImpl, although insideAlarmImpl, it use atomic to try to make it thread-safeBut
Unref/Refis not thread safe because in grpc's implementation, ifEstablishCallDatais put back to grpc's core, it does not callRefimmediately, instead, looks like grpc only callRefif some event happens:So there is a case that 2 threads try to delete the AlarmImpl concurrently.
gpr_unref(&refs_)gpr_unref(&refs_)return true(refs is 0)Ref()delete thisUnref(), and it will also try to deletethisWhat is changed and how it works?
This pr let
EstablishCallDatato hold the alarm, so it will never be constructed whenEstablishCallDatais inside grpc's core.Check List
Tests
Side effects
Documentation
Release note
Summary by CodeRabbit