Skip to content

metal: pread pool dispatch stats and IO-tier pinning for streaming experts #570

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

Open
jasontitus wants to merge 2 commits into
base: main
Choose a base branch
from
Open

metal: pread pool dispatch stats and IO-tier pinning for streaming experts #570

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a1afb82
metal: report pread pool dispatch stats in the timing summary
jasontitus Jul 10, 2026
66ca6ef
metal: keep expert pread threads on the important IO tier
jasontitus Jul 10, 2026
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
76 changes: 76 additions & 0 deletions ds4_metal.m
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,8 @@
#include <limits.h>
#include <time.h>
#include <pthread.h>
#include <pthread/qos.h>
#include <sys/resource.h>
#include <unistd.h>
#include <sys/mman.h>
#include <sys/sysctl.h>
Expand Down Expand Up @@ -283,6 +285,7 @@ static int ds4_gpu_stream_expert_cache_note_expert_size(
static void ds4_gpu_stream_expert_pending_load_clear(void);
static void ds4_gpu_stream_expert_pread_pool_shutdown(void);
static int ds4_gpu_stream_expert_timing_summary_enabled(void);
static void ds4_gpu_stream_expert_pread_pool_stats_print(void);
static int ds4_gpu_stream_expert_cache_entry_protected(
uint32_t layer,
uint32_t expert,
Expand Down Expand Up @@ -2789,6 +2792,7 @@ void ds4_gpu_print_memory_report(const char *label) {
ds4_gpu_stream_expert_timing_print("delta", delta);
g_stream_expert_timing_last_report = total;
}
ds4_gpu_stream_expert_pread_pool_stats_print();
}
if (getenv("DS4_METAL_STREAMING_EXPERT_LAYER_STATS") != NULL) {
fprintf(stderr, "ds4: streaming expert cache per-layer stats:\n");
Expand Down Expand Up @@ -7819,9 +7823,12 @@ static int ds4_gpu_stream_expert_pread_into(
return 1;
}

static void ds4_gpu_stream_expert_pread_thread_qos(void);

static void *ds4_gpu_stream_expert_pread_worker(void *arg) {
ds4_gpu_stream_expert_pread_worker_args *wa =
(ds4_gpu_stream_expert_pread_worker_args *)arg;
ds4_gpu_stream_expert_pread_thread_qos();
for (uint32_t i = wa->worker_index; i < wa->n_tasks; i += wa->n_workers) {
ds4_gpu_stream_expert_pread_task *task = &wa->tasks[i];
task->ok = ds4_gpu_stream_expert_pread_into(task->offset,
Expand All @@ -7847,14 +7854,37 @@ static int ds4_gpu_stream_expert_pread_into(
static int g_stream_expert_pread_pool_initialized;
static int g_stream_expert_pread_pool_stopping;

/* Dispatch stats for the timing summary: qd_avg = task_ms / wall_ms is
* the effective pread concurrency the pool actually sustains. */
static uint64_t g_stream_expert_pread_pool_stat_dispatches;
static uint64_t g_stream_expert_pread_pool_stat_tasks;
static uint64_t g_stream_expert_pread_pool_stat_workers;
static uint64_t g_stream_expert_pread_pool_stat_bytes;
static double g_stream_expert_pread_pool_stat_task_ms;
static double g_stream_expert_pread_pool_stat_wall_ms;
static double g_stream_expert_pread_pool_stat_t0;
static const ds4_gpu_stream_expert_pread_task *g_stream_expert_pread_pool_stat_task_ptr;
static uint32_t g_stream_expert_pread_pool_stat_task_n;
static int g_stream_expert_pread_pool_stat_pending;

static int ds4_gpu_stream_expert_pread_pool_enabled(void) {
const char *env = getenv("DS4_METAL_STREAMING_EXPERT_PREAD_POOL");
return !(env && strcmp(env, "0") == 0);
}

/* Expert reads are on the token critical path: exempt reader threads
* from the IO throttling / background QoS a demoted parent would hand
* them (plain nice does not throttle IO, taskpolicy -b does). */
static void ds4_gpu_stream_expert_pread_thread_qos(void) {
if (getenv("DS4_METAL_DISABLE_STREAMING_EXPERT_PREAD_QOS") != NULL) return;
(void)pthread_set_qos_class_self_np(QOS_CLASS_USER_INITIATED, 0);
(void)setiopolicy_np(IOPOL_TYPE_DISK, IOPOL_SCOPE_THREAD, IOPOL_IMPORTANT);
}

static void *ds4_gpu_stream_expert_pread_pool_worker(void *arg) {
const uint32_t worker_index = (uint32_t)(uintptr_t)arg;
uint64_t seen_generation = 0;
ds4_gpu_stream_expert_pread_thread_qos();

for (;;) {
pthread_mutex_lock(&g_stream_expert_pread_pool_mutex);
Expand Down Expand Up @@ -7987,6 +8017,11 @@ static int ds4_gpu_stream_expert_pread_pool_begin(
g_stream_expert_pread_pool_active_workers = n_workers;
g_stream_expert_pread_pool_remaining_workers = n_workers;
g_stream_expert_pread_pool_generation++;
g_stream_expert_pread_pool_stat_t0 = ds4_gpu_now_ms();
g_stream_expert_pread_pool_stat_task_ptr = tasks;
g_stream_expert_pread_pool_stat_task_n = n_tasks;
g_stream_expert_pread_pool_stat_workers += n_workers;
g_stream_expert_pread_pool_stat_pending = 1;
pthread_cond_broadcast(&g_stream_expert_pread_pool_start_cond);
pthread_mutex_unlock(&g_stream_expert_pread_pool_mutex);
return 1;
Expand All @@ -8000,10 +8035,51 @@ static int ds4_gpu_stream_expert_pread_pool_wait(void) {
pthread_cond_wait(&g_stream_expert_pread_pool_done_cond,
&g_stream_expert_pread_pool_mutex);
}
if (g_stream_expert_pread_pool_stat_pending) {
const ds4_gpu_stream_expert_pread_task *st =
g_stream_expert_pread_pool_stat_task_ptr;
g_stream_expert_pread_pool_stat_wall_ms +=
ds4_gpu_now_ms() - g_stream_expert_pread_pool_stat_t0;
g_stream_expert_pread_pool_stat_dispatches++;
g_stream_expert_pread_pool_stat_tasks +=
g_stream_expert_pread_pool_stat_task_n;
for (uint32_t i = 0; i < g_stream_expert_pread_pool_stat_task_n; i++) {
g_stream_expert_pread_pool_stat_task_ms += st[i].ms;
g_stream_expert_pread_pool_stat_bytes += st[i].read_bytes;
}
g_stream_expert_pread_pool_stat_task_ptr = NULL;
g_stream_expert_pread_pool_stat_task_n = 0;
g_stream_expert_pread_pool_stat_pending = 0;
}
pthread_mutex_unlock(&g_stream_expert_pread_pool_mutex);
return 1;
}

static void ds4_gpu_stream_expert_pread_pool_stats_print(void) {
pthread_mutex_lock(&g_stream_expert_pread_pool_mutex);
const uint64_t dispatches = g_stream_expert_pread_pool_stat_dispatches;
const uint64_t tasks = g_stream_expert_pread_pool_stat_tasks;
const uint64_t workers = g_stream_expert_pread_pool_stat_workers;
const uint64_t bytes = g_stream_expert_pread_pool_stat_bytes;
const double task_ms = g_stream_expert_pread_pool_stat_task_ms;
const double wall_ms = g_stream_expert_pread_pool_stat_wall_ms;
pthread_mutex_unlock(&g_stream_expert_pread_pool_mutex);
if (dispatches == 0) return;
fprintf(stderr,
"ds4: streaming pread pool dispatches=%llu tasks=%llu "
"tasks_avg=%.1f workers_avg=%.1f bytes=%.2f GiB wall_avg=%.3f ms "
"qd_avg=%.2f pool_gbps=%.2f task_gbps=%.2f\n",
(unsigned long long)dispatches,
(unsigned long long)tasks,
(double)tasks / (double)dispatches,
(double)workers / (double)dispatches,
ds4_gpu_gib(bytes),
wall_ms / (double)dispatches,
wall_ms > 0.0 ? task_ms / wall_ms : 0.0,
wall_ms > 0.0 ? (double)bytes / 1e6 / wall_ms : 0.0,
task_ms > 0.0 ? (double)bytes / 1e6 / task_ms : 0.0);
}

static int ds4_gpu_stream_expert_pread_pool_dispatch(
ds4_gpu_stream_expert_pread_task *tasks,
uint32_t n_tasks,
Expand Down