Repository navigation
[Core] Add whole-hook BatchQueue/SliceQueue handoff, pageable staging pool, and capture leases - #173
Conversation
|
Independent audit of PR #173 at 1. [P1]
|
4362f6f to
dc06342
Compare
7f98428 to
c1d242b
Compare
|
@Samfisheryu thanks for the audit and the self-contained harnesses. All four findings are fixed in c1d242b, rebased onto the revised #172. The default-off pool observation is addressed as well. 1. BatchQueue delivers HF batched captures
2. Retention is charged by backing allocation
3.
|
|
Independent re-review of The previous HF fallback, allocation-capacity charging, pending-context ownership, and end-to-end deadline findings are addressed at the CPU/source evidence level used in the original review. The three new CPU native regression tests pass locally. The pool=0 path now bypasses pooling and uses exact-sized allocation. Real CUDA teardown, staging-saturation timing and performance remain unverified; no GPU work was run. [P1] New pool context is deleted twice when tensor construction throws Location:
With the actual CPU ATen implementation, a single injected allocation failure at the 96-byte StorageImpl allocation produced: Without ASan this run exited with SIGSEGV. The old #173 revision, tested at the same 96-byte ATen allocation, safely propagated Fix: make context ownership singular through the entire construction. Do not manually delete a context already transferred to TensorMaker. Also cover the pre-transfer allocation with RAII. Add a construction-failure test that proves one release and safe exception propagation. Actual CPU allocation-failure reproducer (Torch 2.13.0+cu130, no GPU initialization)Save as #include <ATen/ATen.h>
#include <atomic>
#include <cstdio>
#include <cstdlib>
#include <new>
#include "ring/buffer_pool.h"
// Deterministically simulate one CPU allocation failure in the real ATen
// TensorMaker. No replacement/stub of TensorMaker or pool code is used.
static std::atomic<int> fail_after{-1};
void* operator new(std::size_t n) {
const int count = fail_after.load();
if (count > 0 && fail_after.fetch_sub(1) == 1) {
fail_after.store(-1);
std::fprintf(stderr, "injected std::bad_alloc size=%zu\n", n);
throw std::bad_alloc();
}
if (void* p = std::malloc(n)) return p;
throw std::bad_alloc();
}
void operator delete(void* p) noexcept { std::free(p); }
void operator delete(void* p, std::size_t) noexcept { std::free(p); }
int main(int argc, char** argv) {
at::zeros({1}, at::kByte); // initialize ATen before failure injection
ring::BufferPool pool(0); // released block is freed, never cached
fail_after.store(argc > 1 ? std::atoi(argv[1]) : 2);
try {
auto t = pool.acquire(32);
fail_after.store(-1);
std::puts("acquire succeeded");
} catch (const std::bad_alloc&) {
fail_after.store(-1);
std::puts("std::bad_alloc propagated safely");
}
}Use [P2] The allocation ledger's first-owner rank breaks eviction decisions for shared views Locations: fixed rank stored in the ledger, rank-byte accounting, and eviction preflight. The allocation is charged once, but its entire capacity stays assigned to the rank of the first queued view until the last view leaves. That rank can cease to describe which items keep the allocation alive.
Executed unchanged C++ queue code: Fix: base admission/preflight on the allocation references that would remain after the proposed evictions, including surviving higher-priority aliases and the incoming item's reference. A permanent first-owner bucket cannot support the required decision. Compute the victim set without mutating the queue; commit it only if it admits the new item. Complete CPU shared-view eviction reproducerSave as #include <ATen/ATen.h>
#include <iostream>
#include "ring/batch_queue.h"
void push(ring_py::BatchQueue& q, const char* act, const at::Tensor& t) {
ring_py::HookBatchItem item;
item.act_name = act;
item.layer_no = 0;
item.tensor = t;
item.nbytes = t.nbytes();
item.req_ids = {act};
q.push(std::move(item));
}
void show(ring_py::BatchQueue& q, const char* label) {
std::cout << label << " size=" << q.size() << " retained=" << q.retained_bytes()
<< " dropped=" << q.dropped() << " evicted=" << q.evicted() << " items=";
for (const auto& item : q.drain(100, 0)) std::cout << item.act_name << ",";
std::cout << " retained_after_drain=" << q.retained_bytes() << '\n';
}
int main() {
// A formerly urgent view drains; its shared allocation is now pinned only
// by low priority. Evicting that low item would admit the new urgent item.
{
ring_py::BatchQueue q(100);
q.set_layer_class_map({{"urgent", 0, 0}, {"newurgent", 0, 0}}, 2);
auto shared = at::zeros({80}, at::kByte);
push(q, "urgent", shared.narrow(0, 0, 40));
push(q, "low", shared.narrow(0, 40, 40));
q.drain(1, 0);
push(q, "urgent", at::zeros({10}, at::kByte));
push(q, "newurgent", at::zeros({80}, at::kByte));
show(q, "underestimated_evictable_capacity");
}
// Conversely, low and urgent items share one allocation charged to low.
// Evicting low cannot free it. The contract requires no futile eviction.
{
ring_py::BatchQueue q(100);
q.set_layer_class_map({{"urgent", 0, 0}, {"newurgent", 0, 0}}, 2);
auto shared = at::zeros({80}, at::kByte);
push(q, "low", shared.narrow(0, 0, 40));
push(q, "urgent", shared.narrow(0, 40, 40));
push(q, "newurgent", at::zeros({80}, at::kByte));
show(q, "overestimated_evictable_capacity");
}
}Portable CPU compile/run commands for both reproducersFrom the fixed PR checkout, with the two source files in the current directory and a Torch-enabled Python environment: export CUDA_VISIBLE_DEVICES=''
export OMP_NUM_THREADS=1 MKL_NUM_THREADS=1 OPENBLAS_NUM_THREADS=1
ulimit -c 0
python - <<'PY'
from pathlib import Path
import subprocess
import torch
root = Path(torch.__file__).resolve().parent
includes = [Path("native/csrc"), root/"include",
root/"include/torch/csrc/api/include"]
libs = root/"lib"
for name in ("shared_eviction", "pool_failure"):
flags = (["-fsanitize=address", "-fno-omit-frame-pointer", "-g"]
if name == "pool_failure" else [])
cmd = ["g++", "-std=c++20", "-O0", "-pthread", *flags,
"-D_GLIBCXX_USE_CXX11_ABI=" + str(int(torch.compiled_with_cxx11_abi())),
*["-I" + str(p) for p in includes], name + ".cpp",
"-L" + str(libs), "-Wl,-rpath," + str(libs),
"-Wl,--disable-new-dtags", "-ltorch_cpu", "-lc10", "-o", name]
subprocess.run(cmd, check=True)
PY
./shared_eviction
ASAN_OPTIONS=detect_leaks=0 ./pool_failure 2The fault run is expected to fail on this revision; use the ASan diagnostic, not the process exit code alone, to identify the second deletion. The audit ran the same source and compiler/library options through an isolated runner with core dumps disabled. No production source was edited. [P2] The new CPU test cannot run on the configured CPU CI image Location: This test is marked CPU but skips without CUDA headers. The existing CPU CI image does not install those headers, and its skip gate intentionally permits absent hardware only. This is now an observed CI failure on the cumulative head that inherits this test: Fix: make the host-only harness build without a CUDA toolkit (for example with the required declaration-only boundary), or explicitly provision its header dependency in the CPU job. Do not merely whitelist the skip and silently lose this coverage. Keep the compiler standard aligned with the project's Torch-header requirements. These new failures mean this PR still needs changes, despite the original four findings having targeted fixes. The actual CUDA deadline/teardown and end-to-end performance checks remain pending. |
c1d242b to
21f58b1
Compare
dc06342 to
22a6482
Compare
|
@Samfisheryu thanks for the re-review and the harnesses. All three items are fixed in 21f58b1, rebased on the revised #172. [P1]
|
|
Independent third review of R2-173-pool — allocation cleanup: the context now has one owner through construction and transfer to TensorMaker. Against the real CPU ATen implementation, injected allocation failures 1–4 safely propagate the exception; positions 5–6 succeed. The positive test confirms one block return and subsequent reuse at every failed construction stage. The previous revision remains the contrary control at the same 96-byte allocation point. This also passed CPU allocation checking. R2-173-eviction — shared backing references: both original examples now produce the intended results: An independent reference model recomputing allocation sets, without using BackingLedger, passed 50,000 mixed operations: shared views, partial drain, rank-map changes, eviction enabled/disabled, zero caps and oversized first items. It compared item order, retained bytes and the admission/drop/eviction counters after each operation. R2-173-ci — host build dependency: the harness now always uses the checked-in declaration stubs and C++20. Independent compiler dependency tracing confirmed that both P2P translation units build without toolkit headers. All 6 CPU consumer-fix tests passed, without skips. The latest CI is fully successful, including the CPU skip gate and native test link step. The original HF handoff, lease cleanup and deadline changes remain intact. No additional confirmed issue was found in the revised ownership and admission logic. GPU teardown, staging-saturation timing and end-to-end performance remain deferred; no GPU work was performed. |
Samfisheryu
left a comment
Samfisheryu
left a comment
There was a problem hiding this comment.
Approved at 21f58b10a3a84952f53fed1f18ae67adaf08749c following the completed independent code and CPU/host review. The issues raised for this PR have been addressed and independently rechecked. The reviewed commit's CI is successful.
Review record: #173 (comment)
GPU execution and end-to-end performance benchmarks remain a separate follow-up, pending the owner's explicit instruction as agreed. This approval records the completed code/CPU review and does not claim those measurements.
22a6482 to
6d7e741
Compare
… pool, and capture leases Deliver legacy-ring captures to in-process consumers without the P2P thread taking the GIL, account for every capture a consumer does not receive, and give integrations an end-to-end delivery barrier. Native: - SliceQueue (one item per request x hook x step) and BatchQueue (one item per hook x step: the whole packed tensor, an int32 segments table and the request ids), both pushed in pure C++. BatchQueue is bounded by retained bytes and evicts strictly less urgent queued items first under an optional (act_name, layer) -> rank map; without a map it is FIFO drop-the-newcomer. - Drop notices: every shed item names the requests that lost data; a saturated notice buffer raises a sticky flag for one poll window. - Capture leases: an integration-defined identity attached to a step and delivered with every slice, batch and drop notice; its reference count tells when a configuration's payloads are all consumed. A lease may carry a shared drop-rank map. - The P2P thread copies staging into pooled pageable buffers (warm pages); pool state is reference-counted by the delivered tensors, so a tensor that outlives the engine stays valid and frees its block when dropped. - Batch handoff in the P2P thread for packed steps (RingConfig.batch_submit), with one step_seq per step and final_logits rows at the original position. - drain_to_sink_and_wait(): flush, then wait until every flushed task has been through the consumer. Consumer exceptions are counted (submit_exceptions, last_sink_error); set_abort_on_sink_error makes the next prepare_step raise. - capture_ts_us on every delivery: when the drain first saw the task published. request_flush(): flush the tasks reserved so far as soon as they are published, whatever the thresholds say. Python: - MonitoringEngine(ring_consumer=...) for a callable, SliceQueue or BatchQueue (switching batch_submit on); a ring built with no consumer warns (RingWithoutConsumerWarning) instead of silently dropping captures. - RingTransport/StepContext carry a capture lease to the next metadata publication only. Delivery, retention and lifecycle details: - BatchQueue consumers receive HF batched captures: with only a batch sink installed, HF token [batch, q_len, ...] and attention [batch, heads, q, kv] hooks are delivered as one-request items (one per request slice, segments row (0, db_start, db_end, orig_position), is_attn for attention) instead of being dropped silently; consumer exceptions are counted. - Queue retention charges the allocation an item keeps alive: pool blocks carry their capacity in the DataPtr context, and BatchQueue/SliceQueue charge each distinct allocation once at max(logical, allocated) bytes. SliceQueue gains an optional max_bytes cap and retained_bytes. p2p_pool_cap_bytes == 0 now allocates exact-size buffers with no pool. - TensorMetaFifo owns its pending StepContexts (one per step): clearing or destroying it releases them, so their capture leases turn idle. - drain_to_sink_and_wait(timeout_ms) takes one deadline at entry and bounds producer-stream completion (event polling), the flush and the delivery wait with it.
21f58b1 to
d91a545
Compare
Depends on #172.
Summary
In-process consumers without the GIL on the drain path: a
SliceQueue(per-request slices) or
BatchQueue(one whole tensor per HookPoint per stepplus a segments table), pooled pageable buffers, capture leases that track
when a step's data has been consumed, and a delivery barrier.
Implementation details
capture_lease.h,drop_notice.h,slice_queue.h,batch_queue.h(rank-ordered eviction that evicts nothing unless evicting every less
urgent item admits the newcomer),
buffer_pool.h(reference-counted poolstate, so delivered tensors may outlive the engine; bounded size classes),
batch submit in the P2P thread with bounds-checked segment rows, capture
timestamps,
request_flush,drain_to_sink_and_wait(returns once theconsumer thread has exited), consumer-failure accounting
(
submit_exceptions,last_sink_error, opt-in abort).MonitoringEngine(ring_consumer=...)(aBatchQueueselects the batchhandoff);
RingWithoutConsumerWarning.docs/ring-consumers.md.Compatibility and scope
Existing consumers are unchanged.
Validation
test_ring_consumer_queues.py(11 passed, including eviction that cannotadmit the newcomer),
test_ring_consumers_gpu.py(9 passed,including tensors outliving the engine).
test_ring_engine317 assertions, including adelivery barrier that returns after the consumer thread exits (it hangs
without the fix).