| 722 | } |
| 723 | |
| 724 | static void extract_worker(int worker_id, void *ctx_ptr) { |
| 725 | extract_ctx_t *ec = ctx_ptr; |
| 726 | extract_worker_state_t *ws = &ec->workers[worker_id]; |
| 727 | |
| 728 | /* Lazy gbuf creation */ |
| 729 | if (!ws->local_gbuf) { |
| 730 | ws->local_gbuf = cbm_gbuf_new_shared_ids(ec->project_name, ec->repo_path, ec->shared_ids); |
| 731 | } |
| 732 | |
| 733 | /* Pull files from shared atomic counter */ |
| 734 | while (SKIP_ONE) { |
| 735 | int sort_pos = |
| 736 | atomic_fetch_add_explicit(&ec->next_file_idx, SKIP_ONE, memory_order_relaxed); |
| 737 | if (sort_pos >= ec->file_count) { |
| 738 | break; |
| 739 | } |
| 740 | cbm_scale_tick(&ec->scale, sort_pos); |
| 741 | if (atomic_load_explicit(ec->cancelled, memory_order_relaxed)) { |
| 742 | break; |
| 743 | } |
| 744 | |
| 745 | /* Memory back-pressure (large repos): if the process is over its RSS |
| 746 | * budget, reclaim this thread's freed pages and nap so peer workers can |
| 747 | * finish their current file and return memory before this worker adds |
| 748 | * another parse working set. Caps the concurrent extraction transient |
| 749 | * near the budget instead of letting all workers parse their biggest |
| 750 | * files at once. Self-disabling when the budget is unset (tests) or RSS |
| 751 | * is under budget; bounded spins avoid deadlock when the resident graph |
| 752 | * is itself near budget (then proceed with a soft overshoot). |
| 753 | * |
| 754 | * Futility latch: when a FULL nap cycle ends still over budget, the |
| 755 | * resident floor — not transients — holds the memory; napping again on |
| 756 | * the next pull cannot reclaim it and only idles workers (linux kernel: |
| 757 | * one full cycle per pull ≈ 390 s at 79% avg CPU). Latch bp_futile and |
| 758 | * proceed with the soft overshoot; the over-budget probe below re-arms |
| 759 | * the gate as soon as RSS drains under budget. */ |
| 760 | if (cbm_mem_budget() > 0) { |
| 761 | bool over = cbm_mem_over_budget(); |
| 762 | bool futile = atomic_load_explicit(&ec->bp_futile, memory_order_relaxed) != 0; |
| 763 | if (over && !futile) { |
| 764 | cbm_mem_collect(); |
| 765 | atomic_fetch_add_explicit(&g_bp_nap_cycles, SKIP_ONE, memory_order_relaxed); |
| 766 | int bp = 0; |
| 767 | for (; bp < PP_BACKPRESSURE_MAX_SPINS && cbm_mem_over_budget() && |
| 768 | !atomic_load_explicit(ec->cancelled, memory_order_relaxed); |
| 769 | bp++) { |
| 770 | struct timespec nap = {0, PP_BACKPRESSURE_NAP_NS}; |
| 771 | cbm_nanosleep(&nap, NULL); |
| 772 | } |
| 773 | if (bp == PP_BACKPRESSURE_MAX_SPINS && cbm_mem_over_budget()) { |
| 774 | /* Log only the 0→1 transition: all workers race into the |
| 775 | * gate before anyone latches, so a plain store would WARN |
| 776 | * once per worker (12 lines per latch event). */ |
| 777 | if (atomic_exchange_explicit(&ec->bp_futile, 1, memory_order_relaxed) == 0) { |
| 778 | cbm_log_warn("mem.backpressure.futile", "action", "soft_overshoot"); |
| 779 | } |
| 780 | } |
| 781 | } else if (!over && futile) { |
nothing calls this directly
no test coverage detected