mirror of
https://github.com/facebook/rocksdb.git
synced 2026-07-07 14:47:40 +08:00
CPU corruption injector: runner + db_stress preset flags (#14866)
Summary: Orchestration layer of the CPU corruption injector -- the glue between detection (https://github.com/facebook/rocksdb/pull/14852) and injection (https://github.com/facebook/rocksdb/pull/14858 ). One CPU-corruption injection says little on its own. What matters is the OUTCOME DISTRIBUTION over many injections -- how often a corruption is silently absorbed (NO_EFFECT), crashes the process (CRASH), is caught by an integrity check (CORRUPTION), or more importantly slips through as a silent data corruption (SDC) and the paths frequently leading to those outcomes. A trustworthy distribution needs a (somewhat) repeatable and reproducible harness as well as a db_stress configuration in which an injected corruption is both reachable and attributable to the chosen stress test op (i.e, write, foreground compaction or flush). Therefore this PR implements a runner that can launch N independent runs for a chosen op type (i.e, write, foreground compaction or flush). Each run picks where to inject, runs db_stress under gdb via the `injector,py` (https://github.com/facebook/rocksdb/pull/14858), and is classified into one outcome bucket (https://github.com/facebook/rocksdb/pull/14852). The runner has DB_STRESS_PRESET -- the pinned db_stress config that isolates a single injected corruption (single-threaded, integrity checks on, other fault injection off, auto-compaction off). The runner also does gdb preflight that fails fast when the build or gdb cannot support injection because, for example, the hard-coded `target_fn` has changed its name, provides parallel launching of many runs and one summary.json per campaign (the outcome distribution plus each run's record). The whole run set is reproducible from one logged base_seed (run i uses base_seed + i). **Test plan:** Build: `make DEBUG_LEVEL=0 EXTRA_CXXFLAGS="-g -fno-inline" db_stress` # 1. Preflight (`verify_injection_site`) catches a build that can't support injection Before doing any work, the runner has gdb confirm — on this exact binary — that every injection-site function resolves and gdb can read its source line. A good build logs: ``` INFO gdb check OK for op=compaction: rocksdb::CompactionJob::Run, rocksdb::CompactionIterator::NextFromInput, rocksdb::BlockBuilder::Add ``` A build that cannot support injection (functions renamed, fully inlined, or absent) fails fast with exit 2 before any run — forced here by pointing `--stress_cmd` at a non-db_stress binary: ``` $ python3 tools/cpu_corruption_injector/runner.py --op compaction --runs 1 --stress_cmd /bin/ls --report_dir /tmp/icc/preflight_demo ERROR gdb could not set a breakpoint on these functions in ls (renamed, fully inlined, or not in this build?): rocksdb::CompactionJob::Run, rocksdb::CompactionIterator::NextFromInput, rocksdb::BlockBuilder::Add Function "rocksdb::CompactionJob::Run" not defined. Function "rocksdb::CompactionIterator::NextFromInput" not defined. # exit code 2 ``` So a broken/inlined build is rejected up front instead of silently producing `NO_INJECTION` runs. # 2. Compaction op -- 100 runs ``` $ python3 tools/cpu_corruption_injector/runner.py --op compaction --runs 100 --stress_cmd /data/users/huixiao/rocksdb/db_stress --report_dir /tmp/icc/preflight_demo ``` **Runs' outcomes (`summary.json`):** | SDC | CORRUPTION | CRASH | NO_EFFECT | NO_INJECTION | ERROR | | --- | --- | --- | --- | --- | --- | | 9 | 5 | 6 | 79 | 1 | 0 | **Spread:** - target_fn x outcome: `NextFromInput` {SDC 9, CORRUPTION 3, CRASH 2, NO_EFFECT 34, NO_INJECTION 1}; `BlockBuilder::Add` {CORRUPTION 2, CRASH 4, NO_EFFECT 45}. - corruption_type x outcome: `bit_flip` {SDC 8, CORRUPTION 3, CRASH 6, NO_EFFECT 32}; `flag_flip` {SDC 1, CORRUPTION 2, NO_EFFECT 42}; `lane_bit_flip` {NO_EFFECT 5}. **Analysis:** all 9 SDCs land on the read/iterate path (`NextFromInput`); corrupting the output writer (`BlockBuilder::Add`) never produced an SDC (its blocks are checksummed -- corruption there is caught or inert). The 5 detected `CORRUPTION`s are compaction's key-order and record-count cross-checks firing (both CompactRange and CompactFiles origins appear), correctly bucketed by the fix. ### A representative compaction SDC: `run_00000` What we corrupted (`inject.json`): ```json {"op":"compaction","op_index":17,"entry_fn":"rocksdb::CompactionJob::Run","target_fn":"rocksdb::CompactionIterator::NextFromInput","injection_result":"injected","db_stress_crash_signal":null, "corruptions":[{"instruction":"mov %rdx,0x160(%r12)","register":"rdx","corruption_type":"bit_flip","before":"0x10","after":"0x18", "details":{"source":"rocksdb::CompactionIterator::NextFromInput @ db/compaction/compaction_iterator.cc:719"}}]} ``` The recorded silent corruption (`data_corruption.<tid>.json`): ```json {"kind":"wrong-value","cf":0,"key":70,"value_from_db":"010000000504070609080B0A0D0C0F0E070F105E78787878","value_from_expected":"010000000504070609080B0A0D0C0F0E","op_status":"Get: OK"} ``` **Walkthrough:** a single bit flip on `rdx` (`0x10 -> 0x18`, 16 -> 24) at `CompactionIterator::NextFromInput` (`compaction_iterator.cc:719`) -- `rdx` holds the value LENGTH stored into the iterator's record (offset `0x160`). The length reads 24 instead of 16, so compaction copies a value 8 bytes too long into the output SST, absorbing adjacent bytes. The internal key is untouched (`ParseInternalKey` passes); the over-long value is the single Slice fed to both the paranoid validator and the SST builder, so the file is self-consistent and every checksum agrees. On read-back `Get(key=70)` returns OK with the wrong bytes -- `value_from_db` is the expected value (`...0F0E`) **plus 8 trailing bytes** (`070F105E78787878`). Silent: read OK, all checks pass, the value visibly grew. `classify()` routes `kind=wrong-value` to the SDC bucket. ### A representative compaction CORRUPTION (detected): `run_00007` What we corrupted (`inject.json`): ```json {"op":"compaction","op_index":41,"entry_fn":"rocksdb::CompactionJob::Run","target_fn":"rocksdb::CompactionIterator::NextFromInput","injection_result":"injected","db_stress_crash_signal":"SIGABRT", "corruptions":[{"instruction":"mov (%rbx),%rdi","register":"rdi","corruption_type":"bit_flip","before":"0x7fffeee8c1c0","after":"0x7fffeee8c1c2", "details":{"source":"rocksdb::IterKey::SetKeyImpl @ ./db/dbformat.h:941","call_chain":["rocksdb::IterKey::SetKeyImpl @ ./db/dbformat.h:941","rocksdb::CompactionIterator::NextFromInput @ db/compaction/compaction_iterator.cc:781"]}}]} ``` The recorded detection (`data_corruption.<tid>.json`): ```json {"kind":"detected-corruption","cf":-1,"key":-1,"value_from_db":"","value_from_expected":"","op_status":"compactfiles: Corruption: Compaction sees out-of-order keys."} ``` **Walkthrough:** a bit flip on `rdi` (`...c1c0 -> ...c1c2`, a key pointer) at `IterKey::SetKeyImpl` (`dbformat.h:941`), reached from `NextFromInput`, mis-sets the iterator's key so the next emitted key is out of order. Compaction's key-order check catches it and returns `compactfiles: Corruption: Compaction sees out-of-order keys`. The op then takes `SIGABRT`, but `classify()` reads the recorded `data_corruption` result before the crash signal, so the run is correctly bucketed `CORRUPTION` (the bucketization fix; pre-fix this surfaced as `CRASH`). `classify()` routes `kind=detected-corruption` to the `CORRUPTION` bucket. # 3. Flush op -- 100 runs ``` $ python3 tools/cpu_corruption_injector/runner.py --op flush --runs 100 --stress_cmd /data/users/huixiao/rocksdb/db_stress --report_dir /tmp/icc/preflight_demo ``` **Runs' outcomes (`summary.json`):** | SDC | CORRUPTION | CRASH | NO_EFFECT | NO_INJECTION | ERROR | | --- | --- | --- | --- | --- | --- | | 2 | 12 | 11 | 74 | 1 | 0 | **Spread:** - target_fn x outcome: `BlockBuilder::Add` {SDC 1, CORRUPTION 5, CRASH 5, NO_EFFECT 49}; `NextFromInput` {SDC 1, CORRUPTION 7, CRASH 6, NO_EFFECT 25, NO_INJECTION 1}. - corruption_type x outcome: `bit_flip` {SDC 2, CORRUPTION 9, CRASH 9, NO_EFFECT 32}; `flag_flip` {CORRUPTION 3, CRASH 2, NO_EFFECT 42}. **Analysis:** flush mirrors compaction's mechanisms (shared iterator/builder). The 2 SDCs are a value/key-pointer corruption that slips past the checksums; the 12 corruptions are caught by the flush-time key-order / key-size integrity checks. ### A representative flush SDC: `run_00027` What we corrupted (`inject.json`): ```json {"op":"flush","op_index":16,"entry_fn":"rocksdb::FlushJob::Run","target_fn":"rocksdb::BlockBuilder::Add","injection_result":"injected","db_stress_crash_signal":null, "corruptions":[{"instruction":"mov (%rdi),%rax","register":"rax","corruption_type":"bit_flip","before":"0x7fffef059400","after":"0x7fffef059440", "details":{"source":"rocksdb::Slice::data @ ./include/rocksdb/slice.h:58","call_chain":["rocksdb::Slice::data @ ./include/rocksdb/slice.h:58","rocksdb::BlockBuilder::AddWithLastKeyImpl @ table/block_based/block_builder.cc:351"]}}]} ``` The recorded silent corruption (`data_corruption.<tid>.json`): ```json {"kind":"lost","cf":0,"key":763,"value_from_db":"","value_from_expected":"010000000504070609080B0A0D0C0F0E","op_status":"Get: NotFound"} ``` **Walkthrough:** a bit flip on `rax` (a key/value data pointer, `...9400 -> ...9440`) at `Slice::data` (`slice.h:58`), reached from `BlockBuilder::AddWithLastKeyImpl` while the flush builds the output block, makes the builder read key bytes from the wrong address, so key 763's entry is written wrong and the key is dropped from the flushed SST. On read-back `Get(key=763)` returns `NotFound` for a committed key -- silent. `classify()` routes `kind=lost` to the SDC bucket. ### A representative flush CORRUPTION (detected): `run_00047` What we corrupted (`inject.json`): ```json {"op":"flush","op_index":7,"entry_fn":"rocksdb::FlushJob::Run","target_fn":"rocksdb::CompactionIterator::NextFromInput","injection_result":"injected","db_stress_crash_signal":"SIGABRT", "corruptions":[{"instruction":"cmp $0x7,%rax","register":"eflags","corruption_type":"flag_flip","before":"0x216","after":"0x256", "details":{"source":"rocksdb::ParseInternalKey @ ./db/dbformat.h:523","call_chain":["rocksdb::ParseInternalKey @ ./db/dbformat.h:523","rocksdb::CompactionIterator::NextFromInput @ db/compaction/compaction_iterator.cc:731"]}}]} ``` The recorded detection (`data_corruption.<tid>.json`): ```json {"kind":"detected-corruption","cf":-1,"key":-1,"value_from_db":"","value_from_expected":"","op_status":"flush: Corruption: Corrupted Key: Internal Key too small. Size=16. "} ``` **Walkthrough:** a flag flip (`eflags 0x216 -> 0x256`) on a `cmp $0x7,%rax` branch in `ParseInternalKey` (`dbformat.h:523`), reached from `NextFromInput`, makes the parser mis-judge the internal-key size, so the flush emits a malformed key and the key-size integrity check returns `flush: Corruption: Corrupted Key: Internal Key too small`. The op takes `SIGABRT`; `classify()` reads the recorded `data_corruption` before the signal and buckets `CORRUPTION` (bucketization fix). `classify()` routes `kind=detected-corruption` to the `CORRUPTION` bucket. # 4. Write op (`MemTable::Add`) -- two key spaces ``` $ python3 tools/cpu_corruption_injector/runner.py --op write --runs 100 --stress_cmd /data/users/huixiao/rocksdb/db_stress --report_dir /tmp/icc/preflight_demo ``` A write injection corrupts a single `MemTable::Add` (a Put/Delete/DeleteRange). The corruption is reachable and attributable, but whether it surfaces as a *silent* write SDC depends heavily on the key space. A silent write SDC needs the affected/mispositioned key to have other live versions to fall through to -- which only happens in a dense, multi-version memtable. We therefore run two write campaigns: the default `max_key=1000`, then a small `max_key=8`. The contrast is what motivates randomizing `max_key` (see PR https://github.com/facebook/rocksdb/pull/14867 for `--randomize_stress_flags`). ### 4a. Default `max_key=1000` -- 100 runs (no silent write SDC) **Runs' outcomes (`summary.json`):** | SDC | CORRUPTION | CRASH | NO_EFFECT | NO_INJECTION | ERROR | | --- | --- | --- | --- | --- | --- | | 0 | 31 | 13 | 56 | 0 | 0 | With a 1000-key space almost every write touches a distinct key, so a corrupted entry has no older live version to mask it: a value/key byte flip is caught at write by the per-key checksum (`VerifyEncodedEntry` -> `CORRUPTION`), and a structural flip tends to crash (`CRASH`) rather than silently mis-read. `ERROR=0`, `NO_INJECTION=0`. No write op silently corrupted data -- every reachable corruption was caught or crashed. ### 4b. Small `max_key=8` -- 100 runs (surfaces 2 silent write SDCs) **Runs' outcomes (`summary.json`):** | SDC | CORRUPTION | CRASH | NO_EFFECT | NO_INJECTION | ERROR | | --- | --- | --- | --- | --- | --- | | 2 | 33 | 8 | 57 | 0 | 0 | Shrinking the key space makes each key hold ~125 versions (`ops_per_thread` / `max_key`), so a misplaced entry can fall through to an older version of the *same* key and be returned silently -- the per-key checksum (bytes intact) and on-seek verify cannot see a pure link-position error. ### A representative write SDC: `run_00028` (skiplist misposition -> silent stale read, flush catches) What we corrupted (`inject.json`): ```json {"op":"write","op_index":317,"entry_fn":"rocksdb::MemTable::Add","target_fn":"rocksdb::MemTable::Add","injection_result":"injected","db_stress_crash_signal":null, "corruptions":[{"instruction":"cmp %rbx,-0xb8(%rbp)","register":"eflags","corruption_type":"flag_flip","before":"0x216","after":"0x217", "details":{"source":"rocksdb::MemTable::Add @ db/memtable.cc:1319", "call_chain":["rocksdb::MemTable::Add @ db/memtable.cc:1319"]}}]} ``` The recorded silent corruption (`data_corruption.<tid>.json`): ```json {"kind":"resurrected","cf":0,"key":1,"value_from_db":"110000001514171619181B1A1D1C1F1E0100030205040706","value_from_expected":"","op_status":"Get: OK"} ``` **Walkthrough:** a flag flip (CF, `eflags 0x216 -> 0x217`) on the `cmp` that produces `KeyIsAfterNode` inside `InlineSkipList::Insert` (`inlineskiplist.h:1253`; the `@ memtable.cc:1319` in the record is inlining line-drift) inverts the key comparison, so the Delete tombstone for key 1 is linked at the wrong position. The stored bytes and per-key checksum are intact, so neither the checksum nor on-seek verify sees anything wrong -- on read-back `Get(key=1)` returns OK with key 1's live value for a key that was Deleted (`kind=resurrected`, silent). A follow-up `Flush()` in unit test repro *does* catch it: the full-scan order check returns `Corruption: Out-of-order keys found in skiplist` -- caught only after the silent read, not during it. ### A representative write CORRUPTION (detected) `max_key=1000 or 8` : `run_00018` Where `run_00028`'s pure link-position error is invisible to the per-key checksum, this run shows a byte-level corruption that the checksum *catches* at write time. What we corrupted (`inject.json`): ```json {"op":"write","op_index":106,"entry_fn":"rocksdb::MemTable::Add","target_fn":"rocksdb::MemTable::Add","injection_result":"injected","db_stress_crash_signal":null, "corruptions":[{"instruction":"mov %rsi,(%rdi)","register":"rsi","corruption_type":"bit_flip","before":"0x7fffeec2a21c","after":"0x7fffeec2a25c", "details":{"source":"rocksdb::Slice::Slice @ ./include/rocksdb/slice.h:39", "call_chain":["rocksdb::Slice::Slice @ ./include/rocksdb/slice.h:39","rocksdb::GetVarint32 @ ./util/coding.h:280","rocksdb::MemTable::VerifyEncodedEntry @ db/memtable.cc:1102","rocksdb::MemTable::Add @ db/memtable.cc:1189"]}}]} ``` The recorded detection (`data_corruption.<tid>.json`): ```json {"kind":"detected-corruption","cf":-1,"key":-1,"value_from_db":"","value_from_expected":"","op_status":"put: Corruption: ProtectionInfo mismatch"} ``` **Walkthrough:** a bit flip on `rsi` (`0x7fffeec2a21c -> 0x7fffeec2a25c`) at `Slice::Slice` (`slice.h:39`) while `MemTable::Add` re-parses the just-encoded entry through `VerifyEncodedEntry` (`memtable.cc:1102`) corrupts the Slice the verifier reads, so the recomputed per-key protection info no longer matches and the put returns `Corruption: ProtectionInfo mismatch`. Reviewed By: pdillinger Differential Revision: D108367345
This commit is contained in:
Hui Xiao
committed by
Facebook GitHub Bot
Facebook GitHub Bot
parent
3745f2e234
commit
f2b6fbf7b3
@@ -0,0 +1,220 @@
|
||||
# CPU Corruption Injector
|
||||
|
||||
## Overview
|
||||
|
||||
This tool measures the **outcome distribution** over many `db_stress` test runs.
|
||||
In each `db_stress` test run, a single CPU-register bit-flip is injected into
|
||||
exactly one `db_stress` operation (a write, flush, or compaction) running under
|
||||
`gdb`. Aggregating over many such runs answers "when the CPU corrupts one
|
||||
instruction inside that op, how often does RocksDB catch it, crash, silently
|
||||
lose/return wrong data, or absorb it with no effect?".
|
||||
|
||||
It has three layers:
|
||||
|
||||
- **Detection** (`db_stress --verify_cpu_corruption_dir`): right after the
|
||||
injected op, `db_stress` reads the whole keyspace back and compares against its
|
||||
expected state, writing any finding to a `data_corruption.*.json` file.
|
||||
- **Injection** (`injector*.py`): runs inside `gdb`, navigates to a randomly
|
||||
chosen instance of the op, stops at a random "critical" instruction -- one of
|
||||
the two kinds that can carry a fault into data: a **move** carrying a value
|
||||
between CPU and memory (a general-purpose or vector register), or a
|
||||
**branch/jump** (the `eflags` register) -- and flips one random bit in the
|
||||
register it uses. Control registers (`rsp`/`rbp`/`rip`/segment) are left alone.
|
||||
- **Orchestration** (`runner*.py`): builds the run set, launches `gdb` per run,
|
||||
classifies each outcome, and aggregates them.
|
||||
|
||||
## Requirements
|
||||
|
||||
- A `db_stress` binary built with debug info and no inlining, so `gdb` can set
|
||||
breakpoints on the injection-site functions and read their source lines:
|
||||
|
||||
```bash
|
||||
make DEBUG_LEVEL=0 EXTRA_CXXFLAGS="-g -fno-inline" db_stress
|
||||
```
|
||||
|
||||
- `gdb`. Defaults to `/usr/bin/gdb`; override with the `ICC_GDB` env var if the
|
||||
telemetry's source resolves to `@ ?:0` (a `gdb` too old to read this build's
|
||||
debug info).
|
||||
- Optional per-run timeout via the `ICC_RUN_TIMEOUT` env var (seconds, default
|
||||
`600`). One hung `gdb`+`db_stress` cannot block the rest of the set.
|
||||
|
||||
## Entry point: `runner.py`
|
||||
|
||||
A **campaign** is many `db_stress` test **runs** of one op type. Each run is a
|
||||
full single-threaded `db_stress` test that performs `ops_per_thread` ops, but the
|
||||
injector corrupts exactly **one** op instance of the chosen type in that run
|
||||
(e.g. the 51st Put of that run); the run is then classified into a single outcome
|
||||
bucket.
|
||||
|
||||
- `--op` is one of `write`, `flush`, `compaction`.
|
||||
- `--parallel N` runs `N` runs at once (each `db_stress` run is itself
|
||||
single-threaded, so raise this to use more cores).
|
||||
- `--seed N` reproduces a set: run `i` uses seed `N+i`. With `--seed 0` (the
|
||||
default) a fresh random `base_seed` is chosen and logged; pass that value back
|
||||
via `--seed` to replay, and add `--runs 1` with `--seed=<base_seed+i>` to
|
||||
replay a single run `i`.
|
||||
|
||||
```bash
|
||||
# a compaction campaign of 100 runs
|
||||
python3 tools/cpu_corruption_injector/runner.py --op compaction --runs 100 \
|
||||
--stress_cmd <db_stress> --report_dir <dir>
|
||||
|
||||
# a flush campaign of 200 runs, 8 in parallel
|
||||
python3 tools/cpu_corruption_injector/runner.py --op flush --runs 200 \
|
||||
--stress_cmd <db_stress> --report_dir <dir> --parallel 8
|
||||
|
||||
# reproduce an earlier set from its logged base_seed
|
||||
python3 tools/cpu_corruption_injector/runner.py --op write --runs 100 \
|
||||
--stress_cmd <db_stress> --report_dir <dir> --seed 12345678
|
||||
```
|
||||
|
||||
Before doing any work, `runner.py` preflights the build: it confirms `gdb` can
|
||||
set breakpoints on the op's injection-site functions and read their source
|
||||
lines, and fails fast otherwise.
|
||||
|
||||
## Outputs
|
||||
|
||||
Everything lands under `--report_dir`: two campaign-wide files at the top level,
|
||||
plus one `run_XXXXX/` subdirectory per run.
|
||||
|
||||
```text
|
||||
<report_dir>/
|
||||
├── summary.json # campaign-wide: outcome_counts + a record per run
|
||||
├── runner.log # campaign-wide: runner logging (also echoed to console)
|
||||
├── run_00000/ # one subdirectory per run
|
||||
│ ├── inject.json # what the injector did (WHERE / WHAT / HOW)
|
||||
│ ├── data_corruption.<tid>.json # the finding, only if this run flagged an SDC/corruption
|
||||
│ ├── gdb.log # this run's gdb output
|
||||
│ ├── db/ # this run's db_stress DB (working dir)
|
||||
│ └── exp/ # this run's expected-values (working dir)
|
||||
├── run_00001/
|
||||
│ └── ...
|
||||
└── ...
|
||||
```
|
||||
|
||||
So `summary.json` and `runner.log` are at the top of `--report_dir`; the
|
||||
per-run `gdb.log` / `inject.json` / `data_corruption.*.json` live inside each
|
||||
`run_XXXXX/`.
|
||||
|
||||
Outcome buckets:
|
||||
|
||||
- **SDC**: silent data corruption -- the read-back succeeded but returned the
|
||||
wrong result (a lost key, a resurrected key, or a wrong value).
|
||||
- **CORRUPTION**: `db_stress` detected a corruption via an integrity check
|
||||
(`Status::Corruption`).
|
||||
- **CRASH**: the injected corruption crashed `db_stress` (a fatal signal).
|
||||
- **NO_EFFECT**: the corruption landed but the run finished with correct data.
|
||||
- **NO_INJECTION**: no bit was flipped this run. The op instance, its
|
||||
`target_fn`, and the instruction to corrupt are all picked **randomly** up
|
||||
front, so a run can occasionally target something it never reaches -- a random
|
||||
op instance beyond the ops this run actually performed, or a
|
||||
critical-instruction index that no later `target_fn` call happens to have.
|
||||
Should be rare; a high count means the injection sites (or the random ranges)
|
||||
need revisiting.
|
||||
- **ERROR**: a harness/framework failure, **not** a `db_stress` data outcome --
|
||||
the injector hit a bug, `gdb` died before writing `inject.json`, or the run hit
|
||||
the `ICC_RUN_TIMEOUT` cap. Expected to be **0**; any non-zero count is a
|
||||
framework bug worth investigating.
|
||||
|
||||
## Reading a run's output across the 3 layers
|
||||
|
||||
Each run threads through all three layers: the injector records what it corrupted
|
||||
(`inject.json`), `db_stress` records what the post-op read-back found
|
||||
(`data_corruption.<tid>.json`), and the runner rolls every run up into
|
||||
`summary.json`. The examples below are from a real 40-run compaction campaign:
|
||||
|
||||
```bash
|
||||
python3 tools/cpu_corruption_injector/runner.py --op compaction --runs 40 \
|
||||
--seed 424242 --stress_cmd <db_stress> --report_dir <dir>
|
||||
```
|
||||
|
||||
**Orchestration layer** -- `summary.json`'s `outcome_counts` (how many of the
|
||||
campaign's runs landed in each bucket), the whole campaign at a glance:
|
||||
|
||||
| SDC | CORRUPTION | CRASH | NO_EFFECT | NO_INJECTION | ERROR |
|
||||
| --- | ---------- | ----- | --------- | ------------ | ----- |
|
||||
| 2 | 10 | 0 | 25 | 3 | 0 |
|
||||
|
||||
### A silent data corruption (SDC): `run_00023`
|
||||
|
||||
**Injection layer** -- `run_00023/inject.json`: an `eflags` flag-flip on the
|
||||
compaction iterator's key-copy path; the op did NOT crash
|
||||
(`db_stress_crash_signal: null`):
|
||||
|
||||
```json
|
||||
{
|
||||
"injection_result": "injected",
|
||||
"db_stress_crash_signal": null,
|
||||
"op": "compaction",
|
||||
"op_index": 6,
|
||||
"entry_fn": "rocksdb::CompactionJob::Run",
|
||||
"target_fn": "rocksdb::CompactionIterator::NextFromInput",
|
||||
"critical_instruction_index": 0,
|
||||
"corruptions": [
|
||||
{
|
||||
"instruction": "cmp $0x40,%rdx",
|
||||
"register": "eflags",
|
||||
"corruption_type": "flag_flip",
|
||||
"before": "0x287",
|
||||
"after": "0x286",
|
||||
"details": {
|
||||
"source": "__memmove_avx512_unaligned_erms @ ../sysdeps/x86_64/multiarch/memmove-vec-unaligned-erms.S:299",
|
||||
"call_chain": [
|
||||
"__memmove_avx512_unaligned_erms @ ../sysdeps/x86_64/multiarch/memmove-vec-unaligned-erms.S:299",
|
||||
"rocksdb::IterKey::SetKeyImpl @ ./db/dbformat.h:941",
|
||||
"rocksdb::IterKey::SetInternalKey @ ./db/dbformat.h:741",
|
||||
"rocksdb::IterKey::SetInternalKey @ ./db/dbformat.h:749",
|
||||
"rocksdb::CompactionIterator::NextFromInput @ db/compaction/compaction_iterator.cc:781"
|
||||
]
|
||||
}
|
||||
}
|
||||
],
|
||||
"ops_seen": 6,
|
||||
"critical_instructions_seen": 1
|
||||
}
|
||||
```
|
||||
|
||||
**Detection layer** -- `run_00023/data_corruption.<tid>.json`: the post-op
|
||||
read-back found a committed key missing:
|
||||
|
||||
```json
|
||||
{
|
||||
"kind": "lost",
|
||||
"cf": 0,
|
||||
"key": 202,
|
||||
"value_from_db": "",
|
||||
"value_from_expected": "010000000504070609080B0A0D0C0F0E",
|
||||
"op_status": "Get: NotFound"
|
||||
}
|
||||
```
|
||||
|
||||
Reading it: the flag flip mis-copied the compaction output key, dropping key 202
|
||||
from the output SST. The op returned OK and no checksum fired, but reading key
|
||||
202 back returns `NotFound` for a committed key, so the runner classifies the run
|
||||
`SDC` (`kind=lost`) -- the bug class this tool exists to surface.
|
||||
|
||||
## Auxiliary entry point: `injector.py`
|
||||
|
||||
`injector.py` is **not run on its own** -- there is no `python3 injector.py`
|
||||
usage. It only works inside `gdb`'s embedded Python (it does `import gdb`), so
|
||||
`runner.py` runs it by launching `gdb` with the injector as gdb's script. Each
|
||||
run's invocation looks like:
|
||||
|
||||
```bash
|
||||
gdb --batch --nx \
|
||||
-iex 'py import sys; sys.argv=[...]' \
|
||||
-x tools/cpu_corruption_injector/injector.py \
|
||||
--args <db_stress> <db_stress flags...>
|
||||
```
|
||||
|
||||
- `--batch` runs `gdb` non-interactively and exits when the script finishes;
|
||||
`--nx` skips the user's `~/.gdbinit` so their config can't change behavior.
|
||||
- `-iex 'py ... sys.argv=[...]'` runs a Python command *before* the script loads,
|
||||
seeding `injector.py`'s `sys.argv` with its parameters (`--op`, `--op_index`,
|
||||
`--entry_fn`, `--target_fn`, `--seed`, `--dir`) -- gdb otherwise hands a `-x`
|
||||
script an empty argv.
|
||||
- `-x tools/cpu_corruption_injector/injector.py` is the script `gdb` executes in
|
||||
its embedded Python.
|
||||
- `--args <db_stress> <db_stress flags...>` is the program `gdb` launches and
|
||||
debugs -- the `db_stress` binary and its flags, which `gdb` starts, stops
|
||||
mid-op, and corrupts.
|
||||
@@ -0,0 +1,216 @@
|
||||
#!/usr/bin/env python3
|
||||
# Copyright (c) Meta Platforms, Inc. and affiliates.
|
||||
# This source code is licensed under both the GPLv2 (found in the COPYING file in the root directory)
|
||||
# and the Apache 2.0 License (found in the LICENSE.Apache file in the root directory).
|
||||
|
||||
# pyre-strict
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import logging
|
||||
import os
|
||||
import random
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
from concurrent.futures import as_completed, ThreadPoolExecutor
|
||||
|
||||
# Run as a plain script (not an installed package), so this directory is not on the
|
||||
# import path; add it so the sibling runner_* imports below resolve (hence each E402).
|
||||
_HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
if _HERE not in sys.path:
|
||||
sys.path.append(_HERE)
|
||||
|
||||
from runner_execute import _get_gdb, execute # noqa: E402
|
||||
from runner_build import build_runs, INJECTION_SITES, Run # noqa: E402
|
||||
from runner_report import report # noqa: E402
|
||||
|
||||
logger: logging.Logger = logging.getLogger("runner")
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
args = parse_args(argv)
|
||||
report_dir = os.path.abspath(args.report_dir)
|
||||
stress_cmd = os.path.abspath(args.stress_cmd)
|
||||
parallel_runs = max(args.parallel, 1)
|
||||
base_seed = args.seed or random.randint(1, 2**64) # 0 -> a fresh seed each run
|
||||
|
||||
os.makedirs(report_dir, exist_ok=True)
|
||||
_setup_logging(report_dir)
|
||||
|
||||
# Fail fast on the deterministic, whole-run problems before doing any work.
|
||||
if not os.path.exists(stress_cmd):
|
||||
logger.error("stress_cmd not found at %s", stress_cmd)
|
||||
return 2
|
||||
try:
|
||||
verify_injection_site(stress_cmd, args.op)
|
||||
except RuntimeError as e:
|
||||
logger.error("%s", e)
|
||||
return 2
|
||||
|
||||
logger.info(
|
||||
"runner: op=%s runs=%d base_seed=%d report_dir=%s parallel=%d gdb=%s",
|
||||
args.op,
|
||||
args.runs,
|
||||
base_seed,
|
||||
report_dir,
|
||||
parallel_runs,
|
||||
_get_gdb(),
|
||||
)
|
||||
# One number reproduces everything: run i used seed base_seed + i, so the whole set
|
||||
# replays with --seed=base_seed, and a single run i with --seed=<base_seed+i> --runs 1.
|
||||
logger.info("reproduce with --seed=%d (run i used seed %d+i)", base_seed, base_seed)
|
||||
injector_py = os.path.join(_HERE, "injector.py")
|
||||
runs = build_runs(args.op, args.runs, report_dir, base_seed)
|
||||
execute_all(runs, injector_py, stress_cmd, parallel_runs)
|
||||
report(runs, args.op, report_dir, base_seed)
|
||||
return 0
|
||||
|
||||
|
||||
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(
|
||||
description="CPU corruption injection runner: runs db_stress under gdb and, "
|
||||
"per run, injects CPU corruption into a single db_stress operation (a write, "
|
||||
"flush, or compaction), then classifies the outcome."
|
||||
)
|
||||
parser.add_argument(
|
||||
"--op", required=True, choices=tuple(INJECTION_SITES), help="op type to inject"
|
||||
)
|
||||
parser.add_argument(
|
||||
"--runs",
|
||||
type=int,
|
||||
required=True,
|
||||
help="how many runs to do (each run injects into one op).",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--stress_cmd", required=True, help="path to the db_stress binary"
|
||||
)
|
||||
parser.add_argument(
|
||||
"--report_dir",
|
||||
required=True,
|
||||
help="output dir for ALL artifacts: runner.log, summary.json, and one "
|
||||
"run_XXXXX/ per run (gdb.log, inject.json, data_corruption.*.json).",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--parallel",
|
||||
type=int,
|
||||
default=1,
|
||||
help="how many runs to execute at once (default 1); each db_stress run is "
|
||||
"itself single-threaded, so raise this to use more cores.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--seed",
|
||||
type=int,
|
||||
default=0,
|
||||
help="master seed for the run set. 0 (default) picks a fresh random seed each "
|
||||
"invocation (logged as base_seed=N); pass that N back via --seed to reproduce "
|
||||
"the same runs. Run i uses seed N+i.",
|
||||
)
|
||||
return parser.parse_args(argv)
|
||||
|
||||
|
||||
def _setup_logging(report_dir: str) -> None:
|
||||
# Log to the console AND <report_dir>/runner.log, so runner-level logging lands
|
||||
# beside the per-run artifacts and the summary.
|
||||
fmt = logging.Formatter("%(asctime)s %(levelname)s %(message)s")
|
||||
console = logging.StreamHandler()
|
||||
console.setFormatter(fmt)
|
||||
log_file = logging.FileHandler(os.path.join(report_dir, "runner.log"))
|
||||
log_file.setFormatter(fmt)
|
||||
root_logger = logging.getLogger()
|
||||
root_logger.setLevel(logging.INFO)
|
||||
root_logger.handlers = [console, log_file]
|
||||
|
||||
|
||||
def verify_injection_site(stress_cmd: str, op: str) -> None:
|
||||
# Before running anything, confirm gdb can work with this op's injection site on this
|
||||
# db_stress build: every function we break on still resolves, and gdb can read its
|
||||
# source file:line. Checking entry_fn's source line alone catches a gdb too old to
|
||||
# read this build's debug info (which makes the telemetry's source come out "@ ?:0").
|
||||
site = INJECTION_SITES[op]
|
||||
functions = list(dict.fromkeys([site.entry_fn, *site.target_fns]))
|
||||
gdb_output = _run_gdb_check(stress_cmd, functions, site.entry_fn)
|
||||
|
||||
missing = [fn for fn in functions if _function_not_found(gdb_output, fn)]
|
||||
if missing:
|
||||
raise RuntimeError(
|
||||
f"gdb could not set a breakpoint on these functions in "
|
||||
f"{os.path.basename(stress_cmd)} (renamed, fully inlined, or not in this "
|
||||
f"build?): {', '.join(missing)}\n{gdb_output}"
|
||||
)
|
||||
if _no_source_lines(gdb_output):
|
||||
raise RuntimeError(
|
||||
f"gdb resolved the functions but cannot read their source line numbers; "
|
||||
f"this gdb ({_get_gdb()}) is likely too old for this build -- point ICC_GDB at a "
|
||||
f"newer one.\n{gdb_output}"
|
||||
)
|
||||
logger.info("gdb check OK for op=%s: %s", op, ", ".join(functions))
|
||||
|
||||
|
||||
def _run_gdb_check(stress_cmd: str, functions: list[str], entry_fn: str) -> str:
|
||||
# gdb loads db_stress only for its symbols (it never runs it, so no db_stress flags
|
||||
# are needed): break on each function, ask entry_fn's source line, then quit. --nx
|
||||
# ignores the user's ~/.gdbinit so their gdb config can't change the output.
|
||||
commands: list[str] = []
|
||||
for function in functions:
|
||||
commands += ["-ex", f"break {function}"]
|
||||
commands += ["-ex", f"info line {entry_fn}", "-ex", "quit"]
|
||||
try:
|
||||
finished = subprocess.run(
|
||||
[_get_gdb(), "--batch", "--nx", *commands, stress_cmd],
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.STDOUT,
|
||||
timeout=180,
|
||||
check=False,
|
||||
)
|
||||
except subprocess.TimeoutExpired as e:
|
||||
raise RuntimeError(f"gdb check timed out: {e}") from e
|
||||
except OSError as e:
|
||||
raise RuntimeError(f"could not launch gdb ({_get_gdb()}): {e}") from e
|
||||
return finished.stdout.decode("utf-8", "replace")
|
||||
|
||||
|
||||
def _function_not_found(gdb_output: str, function: str) -> bool:
|
||||
short_name = function.rsplit("::", 1)[-1] # "rocksdb::MemTable::Add" -> "Add"
|
||||
markers = (
|
||||
f'Function "{function}" not defined',
|
||||
f'Function "{short_name}" not defined',
|
||||
f'No symbol "{function}"',
|
||||
f'No symbol "{short_name}"',
|
||||
)
|
||||
return any(marker in gdb_output for marker in markers)
|
||||
|
||||
|
||||
def _no_source_lines(gdb_output: str) -> bool:
|
||||
# A gdb too old for this build's debug info resolves no source line (the telemetry's
|
||||
# source would come out "@ ?:0"): a Dwarf Error, or `info line` yielding no "Line N
|
||||
# of ..." at all. A function's cold-split (.cold) range legitimately has no line info,
|
||||
# so that benign "No line number information" message alone is not a failure.
|
||||
if "Dwarf Error" in gdb_output:
|
||||
return True
|
||||
return re.search(r"Line \d+ of ", gdb_output) is None
|
||||
|
||||
|
||||
def execute_all(
|
||||
runs: list[Run],
|
||||
injector_py: str,
|
||||
stress_cmd: str,
|
||||
parallel_runs: int,
|
||||
) -> None:
|
||||
with ThreadPoolExecutor(max_workers=parallel_runs) as pool:
|
||||
runs_by_future = {
|
||||
pool.submit(execute, run, injector_py, stress_cmd): run for run in runs
|
||||
}
|
||||
for future in as_completed(runs_by_future):
|
||||
run = runs_by_future[future]
|
||||
try:
|
||||
future.result()
|
||||
except Exception: # noqa: BLE001 -- one run's crash is its own ERROR, never aborts the rest
|
||||
run.outcome = "ERROR"
|
||||
logger.exception("run %d crashed", run.index)
|
||||
logger.info("run %d: %s", run.index, run.outcome)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,236 @@
|
||||
#!/usr/bin/env python3
|
||||
# Copyright (c) Meta Platforms, Inc. and affiliates.
|
||||
# This source code is licensed under both the GPLv2 (found in the COPYING file in the root directory)
|
||||
# and the Apache 2.0 License (found in the LICENSE.Apache file in the root directory).
|
||||
|
||||
# pyre-strict
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import random
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Optional
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class InjectionSite:
|
||||
# entry_fn runs once per op; we stop on its op_index-th call to pick the op to
|
||||
# corrupt. We do not single-step entry_fn itself (for flush/compaction it is huge);
|
||||
# target_fns are the smaller inner functions we single-step inside that op to find a
|
||||
# critical instruction to corrupt (one target_fn is chosen per run).
|
||||
entry_fn: str
|
||||
target_fns: list[str]
|
||||
|
||||
|
||||
INJECTION_SITES: dict[str, InjectionSite] = {
|
||||
"write": InjectionSite(
|
||||
entry_fn="rocksdb::MemTable::Add",
|
||||
target_fns=["rocksdb::MemTable::Add"],
|
||||
),
|
||||
"flush": InjectionSite(
|
||||
entry_fn="rocksdb::FlushJob::Run",
|
||||
target_fns=[
|
||||
"rocksdb::CompactionIterator::NextFromInput",
|
||||
"rocksdb::BlockBuilder::Add",
|
||||
],
|
||||
),
|
||||
"compaction": InjectionSite(
|
||||
entry_fn="rocksdb::CompactionJob::Run",
|
||||
target_fns=[
|
||||
"rocksdb::CompactionIterator::NextFromInput",
|
||||
"rocksdb::BlockBuilder::Add",
|
||||
],
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
# The complete db_stress config every run uses (the groups below merged). Together they
|
||||
# keep one injected CPU corruption isolated and observable, so they are load-bearing for
|
||||
# both correctness and detection effectiveness -- changing them weakens the signal.
|
||||
|
||||
# A mixed write/delete workload; reads and iterators are off so a single injected
|
||||
# corruption stays visible across the following ops.
|
||||
_MIXED_WORKLOAD: dict[str, object] = {
|
||||
"ops_per_thread": 1000,
|
||||
"max_key": 1000,
|
||||
"value_size_mult": 8,
|
||||
"column_families": 1,
|
||||
"nooverwritepercent": 0,
|
||||
"writepercent": 50,
|
||||
"delpercent": 40,
|
||||
"delrangepercent": 10,
|
||||
"readpercent": 0,
|
||||
"iterpercent": 0,
|
||||
"prefixpercent": 0,
|
||||
"flush_one_in": 20,
|
||||
"compact_range_one_in": 10,
|
||||
"compact_files_one_in": 10,
|
||||
# 64 KB, small on purpose: makes flush/compaction emit several SST files so there
|
||||
# are more critical instructions to corrupt.
|
||||
"target_file_size_base": 65536,
|
||||
"reopen": 0,
|
||||
"destroy_db_initially": 1,
|
||||
"clear_column_family_one_in": 0,
|
||||
"disable_wal": 0,
|
||||
}
|
||||
|
||||
# Keep the injected op the only thing running, on one thread at a time, so (1) the op +
|
||||
# its injected corruption + the post-op read-back form a clean attributable sequence, and
|
||||
# (2) no sibling thread ever sits on the single-step target breakpoint and steals gdb's
|
||||
# current frame mid-step. Each flag pins off one source of a competing thread.
|
||||
_SERIALIZED: dict[str, object] = {
|
||||
"threads": 1, # one workload thread
|
||||
"num_dbs": 1,
|
||||
"disable_auto_compactions": 1, # no surprise background compaction
|
||||
"write_buffer_size": 1 << 30, # memtable never auto-flushes
|
||||
# Compaction runs locally, not on the simulated remote compaction service's worker
|
||||
# thread (db_stress default remote_compaction_worker_threads=2). Remote runs the work
|
||||
# in CompactionServiceCompactionJob::Run, where the injector's CompactionJob::Run
|
||||
# entry_fn is off-stack, so the op is never reached. Injecting into remote compaction
|
||||
# is a planned follow-up (a second entry_fn, pinned to one worker).
|
||||
"remote_compaction_worker_threads": 0,
|
||||
"subcompactions": 1, # one compaction thread, no parallel subcompactions
|
||||
}
|
||||
|
||||
# Every integrity check on -- the high-value goal is surfacing silent data corruption
|
||||
# that would otherwise pass the read-back unnoticed.
|
||||
_SDC_PROTECTIONS: dict[str, object] = {
|
||||
"batch_protection_bytes_per_key": 8,
|
||||
"memtable_protection_bytes_per_key": 8,
|
||||
"block_protection_bytes_per_key": 8,
|
||||
"paranoid_file_checks": 1,
|
||||
"paranoid_memory_checks": 1,
|
||||
"memtable_verify_per_key_checksum_on_seek": 1,
|
||||
"detect_filter_construct_corruption": "true",
|
||||
"verify_checksum": 1,
|
||||
"compression_checksum": 1,
|
||||
"file_checksum_impl": "crc32c",
|
||||
# Extra read-time checksum verification. Caution: a "read returns OK but checksum
|
||||
# fails" outcome is confusing and can also reflect a pre-existing issue, not our
|
||||
# injection -- interpret these with care.
|
||||
"verify_checksum_one_in": 1,
|
||||
"verify_file_checksums_one_in": 1000,
|
||||
}
|
||||
|
||||
# db_stress's own fault injection off, so the injected CPU corruption is the only fault.
|
||||
_NO_OTHER_FAULTS: dict[str, object] = {
|
||||
"read_fault_one_in": 0,
|
||||
"write_fault_one_in": 0,
|
||||
"metadata_read_fault_one_in": 0,
|
||||
"metadata_write_fault_one_in": 0,
|
||||
"open_read_fault_one_in": 0,
|
||||
"open_write_fault_one_in": 0,
|
||||
"open_metadata_read_fault_one_in": 0,
|
||||
"open_metadata_write_fault_one_in": 0,
|
||||
"secondary_cache_fault_one_in": 0,
|
||||
"sync_fault_injection": 0,
|
||||
}
|
||||
|
||||
DB_STRESS_PRESET: dict[str, object] = {
|
||||
**_MIXED_WORKLOAD,
|
||||
**_SERIALIZED,
|
||||
**_SDC_PROTECTIONS,
|
||||
**_NO_OTHER_FAULTS,
|
||||
}
|
||||
|
||||
|
||||
def _int_flag(name: str) -> int:
|
||||
return int(DB_STRESS_PRESET[name]) # type: ignore[arg-type]
|
||||
|
||||
|
||||
def op_count_estimate(op: str) -> int:
|
||||
"""Rough upper bound on how many of this op a run performs; the caller uses it to
|
||||
bound which op instance to target."""
|
||||
ops = _int_flag("ops_per_thread")
|
||||
if op == "write":
|
||||
write_pct = (
|
||||
_int_flag("writepercent")
|
||||
+ _int_flag("delpercent")
|
||||
+ _int_flag("delrangepercent")
|
||||
)
|
||||
return max(ops * write_pct // 100, 1)
|
||||
if op == "flush":
|
||||
return max(ops // _int_flag("flush_one_in"), 1)
|
||||
per_range = ops // _int_flag("compact_range_one_in")
|
||||
per_files = ops // _int_flag("compact_files_one_in")
|
||||
return max(per_range + per_files, 1)
|
||||
|
||||
|
||||
@dataclass
|
||||
class Run:
|
||||
# Inputs, chosen by build_runs():
|
||||
index: int
|
||||
op: str
|
||||
op_index: int
|
||||
entry_fn: str
|
||||
target_fn: str
|
||||
seed: int
|
||||
dir: str
|
||||
stress_flags: list[str] = field(default_factory=list)
|
||||
# Outcome, set by runner_execute.execute():
|
||||
outcome: str = "" # the result bucket: SDC / CORRUPTION / CRASH / NO_EFFECT / ...
|
||||
# inject.json contents, kept because runner_report reads them for the pick spread;
|
||||
# db_stress's data_corruption.json is not kept (it only feeds `outcome`).
|
||||
inject_json: Optional[dict[str, object]] = None
|
||||
|
||||
|
||||
def build_runs(
|
||||
op: str,
|
||||
run_count: int,
|
||||
report_dir: str,
|
||||
base_seed: int,
|
||||
) -> list[Run]:
|
||||
site = INJECTION_SITES[op]
|
||||
# op_index picks which op instance the injector targets. Warmup single-steps that
|
||||
# instance's target_fn call to choose a critical-instruction index; a later instance
|
||||
# may have fewer critical instructions, so the injector retries the next instances
|
||||
# (op_index+1, +2, ...) until one reaches that index. Pick op_index in the first third
|
||||
# of the estimate so enough later instances remain to retry into.
|
||||
op_index_max = max(op_count_estimate(op) // 3, 1)
|
||||
runs: list[Run] = []
|
||||
for index in range(run_count):
|
||||
seed = (base_seed + index) % 2**64
|
||||
rng = random.Random(seed)
|
||||
op_index = rng.randint(1, op_index_max)
|
||||
target_fn = rng.choice(site.target_fns)
|
||||
run_dir = os.path.join(report_dir, f"run_{index:05d}")
|
||||
os.makedirs(run_dir, exist_ok=True)
|
||||
per_run_stress_flags = _per_run_stress_flags(run_dir, seed)
|
||||
runs.append(
|
||||
Run(
|
||||
index=index,
|
||||
op=op,
|
||||
op_index=op_index,
|
||||
entry_fn=site.entry_fn,
|
||||
target_fn=target_fn,
|
||||
seed=seed,
|
||||
dir=run_dir,
|
||||
stress_flags=_fixed_stress_flags(per_run_stress_flags),
|
||||
)
|
||||
)
|
||||
return runs
|
||||
|
||||
|
||||
def _per_run_stress_flags(run_dir: str, seed: int) -> dict[str, object]:
|
||||
# The db_stress flags unique to one run -- its own db / expected-values dirs (created
|
||||
# here), its seed, and the dir db_stress writes corruption findings to. Shared by both
|
||||
# the fixed and randomized flag paths.
|
||||
db = os.path.join(run_dir, "db")
|
||||
exp = os.path.join(run_dir, "exp")
|
||||
os.makedirs(db, exist_ok=True)
|
||||
os.makedirs(exp, exist_ok=True)
|
||||
return {
|
||||
"db": db,
|
||||
"expected_values_dir": exp,
|
||||
"seed": seed,
|
||||
"verify_cpu_corruption_dir": run_dir,
|
||||
}
|
||||
|
||||
|
||||
def _fixed_stress_flags(per_run_stress_flags: dict[str, object]) -> list[str]:
|
||||
# The fixed db_stress flag list: this run's flags plus every DB_STRESS_PRESET flag.
|
||||
return [
|
||||
f"--{flag}={value}"
|
||||
for flag, value in {**per_run_stress_flags, **DB_STRESS_PRESET}.items()
|
||||
]
|
||||
@@ -0,0 +1,179 @@
|
||||
#!/usr/bin/env python3
|
||||
# Copyright (c) Meta Platforms, Inc. and affiliates.
|
||||
# This source code is licensed under both the GPLv2 (found in the COPYING file in the root directory)
|
||||
# and the Apache 2.0 License (found in the LICENSE.Apache file in the root directory).
|
||||
|
||||
# pyre-strict
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import signal
|
||||
import subprocess
|
||||
|
||||
from runner_build import Run
|
||||
|
||||
logger: logging.Logger = logging.getLogger("runner")
|
||||
|
||||
|
||||
def _get_gdb() -> str:
|
||||
# gdb that drives db_stress. Override ICC_GDB to point at a newer gdb if the
|
||||
# telemetry's source comes out "@ ?:0" (an old gdb that cannot read this build's
|
||||
# debug info). Read on use, not at import, per Meta's lazy-import guidance.
|
||||
return os.environ.get("ICC_GDB", "/usr/bin/gdb")
|
||||
|
||||
|
||||
def _get_run_timeout_sec() -> float:
|
||||
# Hard per-run cap so one gdb+db_stress that hangs or deadlocks cannot block the
|
||||
# whole run set; single-stepping is slow, so keep it generous.
|
||||
return float(os.environ.get("ICC_RUN_TIMEOUT", "600"))
|
||||
|
||||
|
||||
# Every bucket classify() can return; runner_report iterates this so the set of outcomes
|
||||
# lives in one place.
|
||||
OUTCOMES: tuple[str, ...] = (
|
||||
"SDC",
|
||||
"CORRUPTION",
|
||||
"CRASH",
|
||||
"NO_EFFECT",
|
||||
"NO_INJECTION",
|
||||
"ERROR",
|
||||
)
|
||||
|
||||
# db_stress's read-back tags a corruption it catches with a "kind"; map each kind to our
|
||||
# result bucket (these are the only kinds db_stress emits).
|
||||
_CORRUPTION_BUCKET: dict[str, str] = {
|
||||
"lost": "SDC",
|
||||
"resurrected": "SDC",
|
||||
"wrong-value": "SDC",
|
||||
"detected-corruption": "CORRUPTION",
|
||||
}
|
||||
|
||||
|
||||
def execute(run: Run, injector_py: str, stress_cmd: str) -> None:
|
||||
try:
|
||||
_run_gdb(
|
||||
_gdb_command(injector_py, stress_cmd, run),
|
||||
os.path.join(run.dir, "gdb.log"),
|
||||
)
|
||||
except subprocess.TimeoutExpired:
|
||||
logger.warning(
|
||||
"run %d: timed out after %.0fs", run.index, _get_run_timeout_sec()
|
||||
)
|
||||
run.outcome = "ERROR"
|
||||
return
|
||||
except (
|
||||
OSError
|
||||
) as e: # gdb itself failed to launch -- a runner failure, not an outcome
|
||||
logger.warning("run %d: gdb launch failed: %s", run.index, e)
|
||||
run.outcome = "ERROR"
|
||||
return
|
||||
run.inject_json = _read_json(os.path.join(run.dir, "inject.json"))
|
||||
run.outcome = classify(run)
|
||||
|
||||
|
||||
def classify(run: Run) -> str:
|
||||
# ERROR is a runner-level failure (gdb or the injector itself), never a db_stress
|
||||
# outcome. injection_result is authoritative: only a corruption that actually landed
|
||||
# can be SDC / CORRUPTION / CRASH / NO_EFFECT.
|
||||
inject = run.inject_json
|
||||
if inject is None: # gdb died before the injector wrote inject.json
|
||||
return "ERROR"
|
||||
injection_result = inject.get("injection_result")
|
||||
if injection_result == "error":
|
||||
return "ERROR"
|
||||
if (
|
||||
injection_result != "injected"
|
||||
): # the op was never reached, nothing was corrupted
|
||||
return "NO_INJECTION"
|
||||
kind = _data_corruption_kind(run.dir)
|
||||
if kind is not None:
|
||||
return _CORRUPTION_BUCKET[kind]
|
||||
return "CRASH" if inject.get("db_stress_crash_signal") is not None else "NO_EFFECT"
|
||||
|
||||
|
||||
def _data_corruption_kind(run_dir: str) -> str | None:
|
||||
names = [
|
||||
name
|
||||
for name in os.listdir(run_dir)
|
||||
if name.startswith("data_corruption.") and name.endswith(".json")
|
||||
]
|
||||
assert (
|
||||
len(names) <= 1
|
||||
), f"expected <=1 data_corruption json in {run_dir}, got {names}"
|
||||
if not names:
|
||||
return None
|
||||
data = _read_json(os.path.join(run_dir, names[0]))
|
||||
return str(data["kind"]) if data is not None else None
|
||||
|
||||
|
||||
def _gdb_command(injector_py: str, stress_cmd: str, run: Run) -> list[str]:
|
||||
# gdb's -x runs injector.py with an empty argv, so seed its argv with -iex first;
|
||||
# --args then hands db_stress and its flags to the inferior gdb launches.
|
||||
seed_argv = "py import sys; sys.argv=" + json.dumps(_injector_argv(run))
|
||||
return [
|
||||
_get_gdb(),
|
||||
"--batch",
|
||||
"--nx",
|
||||
"-iex",
|
||||
seed_argv,
|
||||
"-x",
|
||||
injector_py,
|
||||
"--args",
|
||||
stress_cmd,
|
||||
*run.stress_flags,
|
||||
]
|
||||
|
||||
|
||||
def _injector_argv(run: Run) -> list[str]:
|
||||
return [
|
||||
"injector.py",
|
||||
"--op",
|
||||
run.op,
|
||||
"--op_index",
|
||||
str(run.op_index),
|
||||
"--entry_fn",
|
||||
run.entry_fn,
|
||||
"--target_fn",
|
||||
run.target_fn,
|
||||
"--seed",
|
||||
str(run.seed),
|
||||
"--dir",
|
||||
run.dir,
|
||||
]
|
||||
|
||||
|
||||
def _run_gdb(command: list[str], log_path: str) -> None:
|
||||
with open(log_path, "wb") as log_file:
|
||||
proc = subprocess.Popen(
|
||||
command,
|
||||
stdout=log_file,
|
||||
stderr=subprocess.STDOUT, # fold gdb's stderr into the same gdb.log
|
||||
# new session => its own process group, so a timeout kills gdb AND db_stress
|
||||
start_new_session=True,
|
||||
)
|
||||
try:
|
||||
proc.wait(timeout=_get_run_timeout_sec())
|
||||
except subprocess.TimeoutExpired:
|
||||
_kill_group(proc)
|
||||
proc.wait()
|
||||
raise
|
||||
|
||||
|
||||
def _kill_group(proc: subprocess.Popen[bytes]) -> None:
|
||||
try:
|
||||
os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
|
||||
except ProcessLookupError:
|
||||
pass # already exited
|
||||
|
||||
|
||||
def _read_json(path: str) -> dict[str, object] | None:
|
||||
# None if the file is missing or truncated (e.g. a crash mid-write); the caller reads
|
||||
# a missing inject.json as ERROR.
|
||||
try:
|
||||
with open(path) as f:
|
||||
return json.load(f)
|
||||
except (FileNotFoundError, json.JSONDecodeError):
|
||||
return None
|
||||
@@ -0,0 +1,46 @@
|
||||
#!/usr/bin/env python3
|
||||
# Copyright (c) Meta Platforms, Inc. and affiliates.
|
||||
# This source code is licensed under both the GPLv2 (found in the COPYING file in the root directory)
|
||||
# and the Apache 2.0 License (found in the LICENSE.Apache file in the root directory).
|
||||
|
||||
# pyre-strict
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
from collections import Counter
|
||||
from dataclasses import asdict
|
||||
|
||||
from runner_build import Run
|
||||
from runner_execute import OUTCOMES
|
||||
|
||||
logger: logging.Logger = logging.getLogger("runner")
|
||||
|
||||
|
||||
def report(
|
||||
runs: list[Run], op: str, report_dir: str, base_seed: int
|
||||
) -> dict[str, object]:
|
||||
counts = Counter(run.outcome for run in runs)
|
||||
summary: dict[str, object] = {
|
||||
"op": op,
|
||||
"base_seed": base_seed, # re-run with --seed=<base_seed>; run i used base_seed+i
|
||||
"outcome_counts": {bucket: counts.get(bucket, 0) for bucket in OUTCOMES},
|
||||
"runs": [asdict(run) for run in runs],
|
||||
}
|
||||
with open(os.path.join(report_dir, "summary.json"), "w", encoding="utf-8") as f:
|
||||
json.dump(summary, f, indent=2)
|
||||
|
||||
errors = counts.get("ERROR", 0)
|
||||
if errors:
|
||||
# ERROR is a per-run harness failure, not a db_stress data outcome, and separate
|
||||
# from the setup failures that fail fast in the preflight. Two causes, both rare
|
||||
# but expected: a timed-out run (a deadlock from the injected corruption, or just
|
||||
# slow under load), or gdb crashing mid-run / an injector bug (rarer, and worth
|
||||
# investigating -- it may point at something to fix). Surface the count, never fail
|
||||
# the sweep (no retry by design); dig in if it climbs.
|
||||
logger.warning(
|
||||
"%d/%d runs ended in ERROR; monitor that this stays low", errors, len(runs)
|
||||
)
|
||||
return summary
|
||||
Reference in New Issue
Block a user