Function extract_worker

src/pipeline/pass_parallel.c:760–1001

}

static void extract_worker(int worker_id, void *ctx_ptr) {
    extract_ctx_t *ec = ctx_ptr;
    extract_worker_state_t *ws = &ec->workers[worker_id];

    /* Lazy gbuf creation */
    if (!ws->local_gbuf) {
        ws->local_gbuf = cbm_gbuf_new_shared_ids(ec->project_name, ec->repo_path, ec->shared_ids);
    }

    /* Pull files from shared atomic counter */
    while (SKIP_ONE) {
        int sort_pos =
            atomic_fetch_add_explicit(&ec->next_file_idx, SKIP_ONE, memory_order_relaxed);
        if (sort_pos >= ec->file_count) {
            break;
        }
        if (atomic_load_explicit(ec->cancelled, memory_order_relaxed)) {
            break;
        }

        /* Memory back-pressure (large repos): if the process is over its RSS
         * budget, reclaim this thread's freed pages and nap so peer workers can
         * finish their current file and return memory before this worker adds
         * another parse working set. Caps the concurrent extraction transient
         * near the budget instead of letting all workers parse their biggest
         * files at once. Self-disabling when the budget is unset (tests) or RSS
         * is under budget; bounded spins avoid deadlock when the resident graph
         * is itself near budget (then proceed with a soft overshoot).
         *
         * Futility latch: when a FULL nap cycle ends still over budget, the
         * resident floor — not transients — holds the memory; napping again on
         * the next pull cannot reclaim it and only idles workers (linux kernel:
         * one full cycle per pull ≈ 390 s at 79% avg CPU). Latch bp_futile and
         * proceed with the soft overshoot; the over-budget probe below re-arms
         * the gate as soon as RSS drains under budget. */
        if (cbm_mem_budget() > 0) {
            bool over = cbm_mem_over_budget();
            bool futile = atomic_load_explicit(&ec->bp_futile, memory_order_relaxed) != 0;
            if (over && !futile) {
                cbm_mem_collect();
                atomic_fetch_add_explicit(&g_bp_nap_cycles, SKIP_ONE, memory_order_relaxed);
                int bp = 0;
                for (; bp < PP_BACKPRESSURE_MAX_SPINS && cbm_mem_over_budget() &&
                       !atomic_load_explicit(ec->cancelled, memory_order_relaxed);
                     bp++) {
                    struct timespec nap = {0, PP_BACKPRESSURE_NAP_NS};
                    cbm_nanosleep(&nap, NULL);
                }
                if (bp == PP_BACKPRESSURE_MAX_SPINS && cbm_mem_over_budget()) {
                    /* Log only the 0→1 transition: all workers race into the
                     * gate before anyone latches, so a plain store would WARN
                     * once per worker (12 lines per latch event). */
                    if (atomic_exchange_explicit(&ec->bp_futile, 1, memory_order_relaxed) == 0) {
                        cbm_log_warn("mem.backpressure.futile", "action", "soft_overshoot");
                    }
                }
            } else if (!over && futile) {
                atomic_store_explicit(&ec->bp_futile, 0, memory_order_relaxed);