cauchy1988 opened a new pull request, #2407:
URL: https://github.com/apache/incubator-pegasus/pull/2407
## Problem
`replication_ddl_client::call_rpcs_sync`
(`src/client/replication_ddl_client.h`) fans out an RPC to each replica server
and collects the replies into the `rpcs`/`resps` maps. It is used by
`query_disk_info`, `detect_hotkey`, and `add_new_disk`.
The reply callbacks run **concurrently on multiple IO threads** (one per
outstanding RPC), but the two `std::map`s were mutated without any
synchronization. Concurrent `erase`/`emplace` corrupts the red-black tree,
causing:
- intermittent crashes, or
- a **100%-CPU infinite loop** (the reporter's pegasus_shell spinning in
`_Rb_tree_increment`)
Reported in #1103 with frequent use of the legacy `disk_capacity -d`.
## Root Cause
```cpp
for (auto &rpc : rpcs) {
rpc.second.call(rpc.first.resolve(), &tracker,
[&err, &resps, &rpcs, &rpc](error_code code) mutable { // two bugs
err = code;
if (err == dsn::ERR_OK) {
resps.emplace(rpc.first, std::move(rpc.second.response()));
rpcs.erase(rpc.first); //
concurrent erase, no lock
} else {
resps.emplace(rpc.first, std::move(error_s::make(err,
"...")));
}
});
}
```
1. **Unsynchronized concurrent map mutation** — `resps.emplace(...)` and
`rpcs.erase(...)` execute from concurrent IO-thread callbacks with no lock →
red-black tree corruption.
2. **Dangling reference capture** — `&rpc` captures the range-for loop
variable, a single object reused across iterations. All callbacks fire later
(during `wait_outstanding_tasks`) and alias the *last* element, so responses
were keyed under the wrong `host_port`.
## Fix
- Serialize all access to `rpcs`/`resps` with a `std::mutex`.
- Capture the `host_port` key **by value**; look up the map element under
the lock by that key.
- Remove the dead, racy `error_code err` local (written by every callback,
never read).
```cpp
std::mutex mtx;
for (auto &rpc : rpcs) {
const dsn::host_port hp = rpc.first;
rpc.second.call(rpc.first.resolve(), &tracker,
[&mtx, &rpcs, &resps, hp](error_code code) {
std::lock_guard<std::mutex> guard(mtx);
if (code == dsn::ERR_OK) {
auto it = rpcs.find(hp);
if (it != rpcs.end()) {
resps.emplace(hp, std::move(it->second.response()));
rpcs.erase(it);
}
} else {
resps.emplace(hp, error_s::make(code, "unable to send rpc to
server"));
}
});
}
```
## Testing
- Builds cleanly (GCC, all targets including `pegasus_shell`, server,
function tests).
- The fix is a standard correctness fix (mutex + capture-by-value); the
original race is non-deterministic and the legacy repro command (`disk_capacity
-d`) has since been removed from the shell, so deterministic reproduction via
the shell is no longer possible. The shared client library is exercised
end-to-end by the existing function-test suite, which links and builds with
this change.
Closes #1103
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
