Skip to content

Fix: stabilize depth-two chip run lifecycle #1987

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
high-cloud wants to merge 1 commit into
base: main
Choose a base branch
from
+1,065 −185
Draft

Fix: stabilize depth-two chip run lifecycle #1987

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
249 changes: 198 additions & 51 deletions python/simpler/worker.py
View file
Open in desktop

Large diffs are not rendered by default.

86 changes: 71 additions & 15 deletions src/a2a3/platform/onboard/host/device_runner.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@
#include "callable_protocol.h"
#include "call_config.h"
#include "chip_callable_layout.h"
#include "common/strace.h"
#include "utils/elf_build_id.h"
#include "host/host_regs.h" // Register address retrieval
#include "host/raii_scope_guard.h"
Expand Down Expand Up @@ -241,13 +242,18 @@ int DeviceRunner::abandon_native_run_resources(uint32_t pipeline_slot) {
return retire_run_aicore_stream(pipeline_slot);
}

int DeviceRunner::complete_native_run_resources(uint32_t pipeline_slot) {
return retire_run_aicore_stream_async(pipeline_slot);
}

int DeviceRunner::run(Runtime &runtime, const CallConfig &config) {
const unsigned selected_pipeline_slot = pipeline_slot();
// The AICore stream is created during native prepare so its provisioning
// can overlap the predecessor's execution. Once run() is entered, every
// exit owns retirement; the success path reports destroy failure, while
// early-error paths keep the original error and leave a failed-destroy
// handle in the slot so it cannot be reused.
// can overlap the predecessor's execution. Early-error paths retire it
// synchronously and leave a failed-destroy handle in the slot. Success
// hands ownership back to native finalize: only after output D2H completes
// may it start asynchronous retire/replenish. This prevents CANN stream
// control calls from contending with validate's rtMemcpy.
bool aicore_stream_retired = false;
auto aicore_stream_retire = RAIIScopeGuard([this, selected_pipeline_slot, &aicore_stream_retired]() {
if (!aicore_stream_retired) (void)retire_run_aicore_stream(selected_pipeline_slot);
Expand Down Expand Up @@ -300,13 +306,15 @@ int DeviceRunner::run(Runtime &runtime, const CallConfig &config) {
// before the allocs so that an alloc-failure early-return still triggers
// cleanup of previously-allocated buffers (the predicates no-op on 0).
auto regs_cleanup = RAIIScopeGuard([this]() {
STRACE("simpler_run.runner_run.cleanup.regs");
if (kernel_args_.args.regs != 0) {
mem_alloc_.free(reinterpret_cast<void *>(kernel_args_.args.regs));
kernel_args_.args.regs = 0;
}
});

auto pmu_regs_cleanup = RAIIScopeGuard([this]() {
STRACE("simpler_run.runner_run.cleanup.pmu_regs");
if (kernel_args_.args.pmu_reg_addrs != 0) {
mem_alloc_.free(reinterpret_cast<void *>(kernel_args_.args.pmu_reg_addrs));
kernel_args_.args.pmu_reg_addrs = 0;
Expand Down Expand Up @@ -401,6 +409,7 @@ int DeviceRunner::run(Runtime &runtime, const CallConfig &config) {
}

auto runtime_args_cleanup = RAIIScopeGuard([this]() {
STRACE("simpler_run.runner_run.cleanup.runtime_args");
kernel_args_.finalize_device_kernel_args();
kernel_args_.finalize_runtime_args();
});
Expand Down Expand Up @@ -455,6 +464,7 @@ int DeviceRunner::run(Runtime &runtime, const CallConfig &config) {
// Worker reused across runs (e.g. a pytest session-scoped worker pool) would
// otherwise re-enter init_l2_swimlane() with stale state still allocated.
auto perf_cleanup = RAIIScopeGuard([this]() {
STRACE("simpler_run.runner_run.cleanup.collectors");
finalize_collectors();
});

Expand Down Expand Up @@ -507,21 +517,27 @@ int DeviceRunner::run(Runtime &runtime, const CallConfig &config) {
l2_swimlane_collector_.set_core_types(core_types.data(), num_aicore);
}

rc = launch_run(runtime, num_aicore, launch_aicpu_num, selected_pipeline_slot);
{
STRACE("simpler_run.runner_run.launch_device");
rc = launch_run(runtime, num_aicore, launch_aicpu_num, selected_pipeline_slot);
}
if (rc != 0) return rc;

rc = reap_run(selected_pipeline_slot);
{
STRACE("simpler_run.runner_run.reap_device");
rc = reap_run(selected_pipeline_slot);
}
if (rc != 0) return rc;

// The run owns its AICore stream, so a destroy this run cannot complete is
// this run's failure: reporting success would leave the caller believing a
// slot is reusable that the next prepare will now refuse.
// Reap has established that no device work remains on this run's AICore
// stream. Native finalize owns its retire/replenish after output D2H.
aicore_stream_retired = true;
rc = retire_run_aicore_stream(selected_pipeline_slot);
if (rc != 0) return rc;

// Print handshake results (reads from device memory, must be before free)
print_handshake_results();
{
STRACE("simpler_run.runner_run.print_handshake");
print_handshake_results();
}

return 0;
}
Expand All @@ -543,6 +559,36 @@ int DeviceRunner::retire_run_aicore_stream(unsigned slot) {
return rc;
}

int DeviceRunner::retire_run_aicore_stream_async(unsigned slot) {
const unsigned trace_inv = simpler::strace::StraceScope::current_inv();
const uint64_t trace_hid = simpler::strace::StraceScope::current_hid();
int rc = run_stream_slots_.retire_aicore_async(
slot,
[this](std::function<void()> fn) {
return create_thread(std::move(fn));
},
[trace_inv, trace_hid](std::function<void()> fn) {
return [fn = std::move(fn), trace_inv, trace_hid]() {
STRACE_CONTEXT(trace_inv, trace_hid, 2);
const long long wall_start = STRACE_NOW_NS();
const long long cpu_start = STRACE_THREAD_CPU_NOW_NS();
{
STRACE("simpler_run.runner_run.async_destroy_aicore_stream");
fn();
}
STRACE_HOST_SPAN_AT(
"simpler_run.runner_run.async_destroy_aicore_stream.thread_cpu", wall_start,
STRACE_THREAD_CPU_NOW_NS() - cpu_start, 3
);
};
}
);
if (rc != 0) {
LOG_ERROR("async rtStreamDestroy launch (run AICore slot %u) failed: %d", slot, rc);
}
return rc;
}

int DeviceRunner::destroy_run_stream_sets() {
// No pre-destroy sync, for the reason finalize_common() documents for the
// bootstrap pair: rtStreamDestroy is the supported teardown for a stream
Expand Down Expand Up @@ -619,7 +665,11 @@ int DeviceRunner::reap_run(unsigned slot) {
LOG_ERROR("reap_run: invalid stream set %u", slot);
return -1;
}
int rc = sync_stream_pair(run_stream_slots_.aicpu(slot), run_stream_slots_.aicore(slot));
int rc = 0;
{
STRACE("simpler_run.runner_run.native_fence_wait");
rc = sync_stream_pair(run_stream_slots_.aicpu(slot), run_stream_slots_.aicore(slot));
}
if (rc != 0) {
// The pair wait surfaces the AICore op-timeout (STARS-reaped op ->
// 507000/507018/507046 at AICPU/AICore stream sync). The op-timeout
Expand All @@ -639,11 +689,17 @@ int DeviceRunner::reap_run(unsigned slot) {
return rc;
}

read_device_wall_ns();
{
STRACE("simpler_run.runner_run.read_device_wall");
read_device_wall_ns();
}

// Tear down collectors. stop() joins mgmt then collector in the only safe
// order (mgmt's final-drain pass into L2 has poll as its consumer).
teardown_shared_collectors_after_run();
{
STRACE("simpler_run.runner_run.teardown_collectors");
teardown_shared_collectors_after_run();
}

// a2a3-only dep_gen teardown: host-orch emits the graph snapshot adopted
// from the prepare thread; device-orch stops the collector, reconciles the
Expand Down
2 changes: 2 additions & 0 deletions src/a2a3/platform/onboard/host/device_runner.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -119,6 +119,7 @@ class DeviceRunner : public DeviceRunnerBase {
bool can_accept_run() const override { return !device_unusable_.load(std::memory_order_acquire); }
int provision_native_run_resources(uint32_t pipeline_slot) override;
int abandon_native_run_resources(uint32_t pipeline_slot) override;
int complete_native_run_resources(uint32_t pipeline_slot) override;

// Map/unmap a device buffer into host address space via
// halHostRegister(DEV_SVM_MAP_HOST) / halHostUnregister. The returned host
Expand Down Expand Up @@ -264,6 +265,7 @@ class DeviceRunner : public DeviceRunnerBase {
// image's instructions, so the slot must refuse the next run until finalize
// reclaims it.
int retire_run_aicore_stream(unsigned slot);
int retire_run_aicore_stream_async(unsigned slot);
int destroy_run_stream_sets();

// The kernel submission boundary is separate from the stream wait and the
Expand Down
16 changes: 8 additions & 8 deletions src/a2a3/runtime/tensormap_and_ringbuffer/host/runtime_maker.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -981,14 +981,10 @@ extern "C" int validate_runtime_impl(Runtime *runtime, const HostApi *api, int e

int rc = 0;

LOG_INFO("=== Copying Results Back to Host ===");

// Copy all recorded tensors from device back to host
TensorLease *tensor_leases = runtime->tensor_leases_.data();
int tensor_lease_count = static_cast<int>(runtime->tensor_leases_.size());

LOG_INFO("Tensor leases to process: %d", tensor_lease_count);

bool skip_tensor_copy_back = execution_rc != 0;
int32_t runtime_status = 0;
PTO2SharedMemoryHeader host_header;
Expand Down Expand Up @@ -1024,6 +1020,7 @@ extern "C" int validate_runtime_impl(Runtime *runtime, const HostApi *api, int e
if (skip_tensor_copy_back) {
LOG_WARN("Skipping tensor copy-back because execution failed");
} else {
STRACE("simpler_run.validate.copy_back");
for (int i = 0; i < tensor_lease_count; i++) {
const TensorLease &lease = tensor_leases[i];

Expand All @@ -1047,6 +1044,9 @@ extern "C" int validate_runtime_impl(Runtime *runtime, const HostApi *api, int e
continue;
}

char copy_attrs[96];
std::snprintf(copy_attrs, sizeof(copy_attrs), "index=%d bytes=%zu", i, lease.size);
STRACE_A("simpler_run.validate.copy_back.tensor", copy_attrs);
int copy_rc = api->copy_from_device(lease.host_ptr, lease.dev_ptr, lease.size);
if (copy_rc != 0) {
LOG_ERROR("Failed to copy tensor %d from device: %d", i, copy_rc);
Expand All @@ -1058,10 +1058,10 @@ extern "C" int validate_runtime_impl(Runtime *runtime, const HostApi *api, int e
}

// Cleanup device tensors
LOG_INFO("=== Cleaning Up ===");
release_tensor_leases(runtime, api);

LOG_INFO("=== Finalize Complete ===");
{
STRACE("simpler_run.validate.release_leases");
release_tensor_leases(runtime, api);
}

if (rc == 0 && runtime_status != 0) {
rc = runtime_status;
Expand Down
79 changes: 69 additions & 10 deletions src/common/log/include/common/strace.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,8 +59,10 @@

#include <atomic>
#include <chrono>
#include <cstdio>
#include <cstdint>
#include <cstdlib>
#include <ctime>

#include <unistd.h>

Expand Down Expand Up @@ -112,6 +114,62 @@ inline long strace_tid() {
#endif
}

/**
* Emit one host marker. Profiling can redirect markers to one fully-buffered
* file per process with SIMPLER_HOST_STRACE_DIR. This avoids making every
* short scope synchronously flush the shared stderr pipe -- at eight ranks the
* observer itself otherwise creates millisecond scheduling holes between
* adjacent tensor copies. The outermost marker flushes the private file once
* per invocation, so a completed invocation is always available to the parser.
*/
inline void write_span(
const char *name, long long ts_ns, long long dur_ns, int depth, unsigned inv, uint64_t hid, const char *attrs
) {
const char *directory = std::getenv("SIMPLER_HOST_STRACE_DIR");
if (directory == nullptr || directory[0] == '\0') {
LOG_TIMING(
"[STRACE] v=1 pid=%d tid=%ld inv=%u hid=%llx depth=%d name=%s ts=%lld dur=%lld %s",
static_cast<int>(getpid()), strace_tid(), inv, static_cast<unsigned long long>(hid), depth, name, ts_ns,
dur_ns, attrs
);
return;
}

static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
static FILE *stream = nullptr;
static pid_t stream_pid = -1;
pthread_mutex_lock(&mutex);
const pid_t pid = getpid();
if (stream == nullptr || stream_pid != pid) {
// Do not close a stream inherited across fork: another stdio object may
// still own the copied buffer. Each child switches to its own PID file.
char path[4096];
const int count = std::snprintf(path, sizeof(path), "%s/host-strace.%d.log", directory, pid);
if (count > 0 && static_cast<size_t>(count) < sizeof(path)) {
stream = std::fopen(path, "a");
if (stream != nullptr) {
std::setvbuf(stream, nullptr, _IOFBF, 1U << 20U);
stream_pid = pid;
}
}
}
if (stream != nullptr && stream_pid == pid) {
std::fprintf(
stream, "[STRACE] v=1 pid=%d tid=%ld inv=%u hid=%llx depth=%d name=%s ts=%lld dur=%lld %s\n",
static_cast<int>(pid), strace_tid(), inv, static_cast<unsigned long long>(hid), depth, name, ts_ns, dur_ns,
attrs
);
if (depth == 0) std::fflush(stream);
pthread_mutex_unlock(&mutex);
return;
}
pthread_mutex_unlock(&mutex);
LOG_TIMING(
"[STRACE] v=1 pid=%d tid=%ld inv=%u hid=%llx depth=%d name=%s ts=%lld dur=%lld %s", static_cast<int>(pid),
strace_tid(), inv, static_cast<unsigned long long>(hid), depth, name, ts_ns, dur_ns, attrs
);
}

class StraceScope {
public:
explicit StraceScope(const char *name, const char *attrs = "") :
Expand All @@ -131,11 +189,7 @@ class StraceScope {
// depth printed is the scope's own level (post-decrement so the
// outermost scope prints depth=0).
const int d = --depth();
LOG_TIMING(
"[STRACE] v=1 pid=%d tid=%ld inv=%u hid=%llx depth=%d name=%s ts=%lld dur=%lld %s",
static_cast<int>(getpid()), strace_tid(), inv(), static_cast<unsigned long long>(hid()), d, name_, ts, dur,
attrs_
);
write_span(name_, ts, dur, d, inv(), hid(), attrs_);
}

StraceScope(const StraceScope &) = delete;
Expand Down Expand Up @@ -208,6 +262,13 @@ inline long long strace_now_ns() {
);
}

/** Current per-thread CPU clock; diagnostic companion to wall-clock spans. */
inline long long strace_thread_cpu_now_ns() {
struct timespec value{};
if (clock_gettime(CLOCK_THREAD_CPUTIME_ID, &value) != 0) return 0;
return static_cast<long long>(value.tv_sec) * 1000000000LL + value.tv_nsec;
}

/**
* Emit a marker for a span whose duration was measured elsewhere (e.g. a device
* phase: AICPU cycles → ns). Shares the current thread's inv/hid grouping so the
Expand All @@ -220,11 +281,7 @@ inline long long strace_now_ns() {
*/
inline void
emit_span_at(const char *name, long long ts_ns, long long dur_ns, int depth, const char *attrs = "clk=dev") {
LOG_TIMING(
"[STRACE] v=1 pid=%d tid=%ld inv=%u hid=%llx depth=%d name=%s ts=%lld dur=%lld %s", static_cast<int>(getpid()),
strace_tid(), StraceScope::current_inv(), static_cast<unsigned long long>(StraceScope::current_hid()), depth,
name, ts_ns, dur_ns, attrs
);
write_span(name, ts_ns, dur_ns, depth, StraceScope::current_inv(), StraceScope::current_hid(), attrs);
}

/** Emit an explicitly timed host-domain span in the active invocation. */
Expand Down Expand Up @@ -253,6 +310,7 @@ inline void emit_host_span_at(const char *name, long long ts_ns, long long dur_n
::simpler::strace::StraceContextScope STRACE_CAT(_strace_context_, __LINE__)((inv), (hid), (depth))
/** Read the current host monotonic clock in nanoseconds. */
#define STRACE_NOW_NS() ::simpler::strace::strace_now_ns()
#define STRACE_THREAD_CPU_NOW_NS() ::simpler::strace::strace_thread_cpu_now_ns()
/** Emit a host-domain span measured across disjoint API calls. */
#define STRACE_HOST_SPAN_AT(name, ts_ns, dur_ns, depth) \
::simpler::strace::emit_host_span_at((name), (ts_ns), (dur_ns), (depth))
Expand All @@ -272,6 +330,7 @@ inline void emit_host_span_at(const char *name, long long ts_ns, long long dur_n
#define STRACE_SET_HID(h) ((void)0)
#define STRACE_CONTEXT(inv, hid, depth) ((void)0)
#define STRACE_NOW_NS() 0LL
#define STRACE_THREAD_CPU_NOW_NS() 0LL
#define STRACE_HOST_SPAN_AT(name, ts_ns, dur_ns, depth) ((void)0)
#define STRACE_HOST_SPAN_AT_A(name, ts_ns, dur_ns, depth, attrs) ((void)0)
#define STRACE_DEV_SPAN_AT(name, ts_ns, dur_ns, depth) ((void)0)
Expand Down
Loading