Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
11 changes: 11 additions & 0 deletions src/a5/runtime/tensormap_and_ringbuffer/runtime/orchestrator.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -648,6 +648,17 @@ static bool prepare_task(
out->slot_state->logical_block_num = block_num;
out->slot_state->active_mask = active_mask;
out->slot_state->task_attrs = task_attrs;
if (task_attrs.requires_sync_start()) {
// Ordered before the wiring publish below, which is the first read of
// this slot by any scheduler thread, so the latch is set before the
// task can be routed to ready_sync_queues.
//
// Tier-0 priority is best-effort for an entry that becomes visible
// mid-iteration: a thread already past its Tier-0 checkpoint picks that
// entry up on its next pass, whether the checkpoint is this latch or
// the six queue probes it stands in for.
orch->scheduler->sync_task_seen.store(1, std::memory_order_release);
}
// fanin_count is set during Orch-side wiring
scope_tasks_push(orch, out->slot_state);

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -537,6 +537,19 @@ struct SchedulerState {
// stop-the-world drain, per-core MIX placement, head-start spacing).
ChipReadyQueue ready_sync_queues[NUM_RESOURCE_SHAPES];

// Set when the orchestrator submits a task with requires_sync_start(), and
// stays set for the rest of the epoch. Only such a task reaches
// ready_sync_queues[], so while this is clear those queues hold nothing and
// the dispatch loop skips the whole Tier-0 staging order -- six shape probes
// per iteration, each a load on a line every scheduler thread writes.
//
// Release here, acquire on the dispatch loop's load: observing it set also
// makes visible the submission that set it.
//
// Own cache line: written once per epoch, read once per dispatch iteration,
// so it never joins the lines the scheduler threads already contend for.
alignas(64) std::atomic<uint32_t> sync_task_seen;

// Dependency-only tasks (active_mask is empty, shape == DUMMY). Drained by
// the dispatch loop and completed inline -- never goes to AICore.
ChipReadyQueue dummy_ready_queue;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -528,19 +528,26 @@ void SchedulerContext::dispatch_ready_tasks(
// (sync_start > MIX > C/V within the normal source). Same order and machinery,
// fed from ready_sync_queues; an oversized cohort arms the stop-the-world drain
// (entered_drain), which also short-circuits the regular tier below.
run_staging_order(
thread_idx, pmu_active,
[&](ResourceShape shape, CoreTracker::DispatchPhase phase) {
dispatch_shape(
thread_idx, sched_->ready_sync_queues, shape, phase, tracker, entered_drain, made_progress, try_pushed
);
return entered_drain;
},
[&] {
return has_residual_sync_mix();
}
);
if (entered_drain) return;
// Skip Tier-0 entirely until the orchestrator has submitted a sync_start
// task. Without one the staging order below can only probe six empty queues
// per iteration. The acquire pairs with the release in prepare_task(), so
// observing the latch set also makes that task's submission visible.
if (sched_->sync_task_seen.load(std::memory_order_acquire) != 0) {
run_staging_order(
thread_idx, pmu_active,
[&](ResourceShape shape, CoreTracker::DispatchPhase phase) {
dispatch_shape(
thread_idx, sched_->ready_sync_queues, shape, phase, tracker, entered_drain, made_progress,
try_pushed
);
return entered_drain;
},
[&] {
return has_residual_sync_mix();
}
);
if (entered_drain) return;
}

// Tier 1: regular ready work.
run_staging_order(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -168,6 +168,7 @@ bool SchedulerState::init_data_from_layout(const SchedulerLayout &layout, Device
sched->advance_pending_mask.store(0, std::memory_order_relaxed);
sched->publication_request_mask.store(0, std::memory_order_relaxed);
sched->publication_ack_mask.store(0, std::memory_order_relaxed);
sched->sync_task_seen.store(0, std::memory_order_relaxed);
#if SIMPLER_SCHED_PROFILING
sched->tasks_completed.store(0, std::memory_order_relaxed);
sched->tasks_consumed.store(0, std::memory_order_relaxed);
Expand Down Expand Up @@ -232,6 +233,7 @@ void SchedulerState::reset_for_reuse(const SchedulerLayout &layout, void *sm_dev
sched->advance_pending_mask.store(0, std::memory_order_relaxed);
sched->publication_request_mask.store(0, std::memory_order_relaxed);
sched->publication_ack_mask.store(0, std::memory_order_relaxed);
sched->sync_task_seen.store(0, std::memory_order_relaxed);
#if SIMPLER_SCHED_PROFILING
sched->tasks_completed.store(0, std::memory_order_relaxed);
sched->tasks_consumed.store(0, std::memory_order_relaxed);
Expand Down
Loading