| 1004 | const char *repo_path); |
| 1005 | |
| 1006 | int cbm_parallel_extract_ex(cbm_pipeline_ctx_t *ctx, const cbm_file_info_t *files, int file_count, |
| 1007 | CBMFileResult **result_cache, _Atomic int64_t *shared_ids, |
| 1008 | int worker_count, const cbm_parallel_extract_opts_t *opts) { |
| 1009 | cbm_parallel_extract_opts_t resolved_opts = cbm_parallel_extract_resolve_opts(opts); |
| 1010 | |
| 1011 | if (file_count == 0) { |
| 1012 | return 0; |
| 1013 | } |
| 1014 | |
| 1015 | cbm_log_info("parallel.extract.start", "files", itoa_log(file_count), "workers", |
| 1016 | itoa_log(worker_count)); |
| 1017 | { |
| 1018 | size_t mb = (size_t)CBM_SZ_1K * CBM_SZ_1K; |
| 1019 | cbm_log_info("parallel.extract.retention", "retain_sources", |
| 1020 | resolved_opts.retain_sources ? "true" : "false", "total_mb", |
| 1021 | itoa_log((int)(resolved_opts.retain_total_budget_bytes / mb)), "per_file_mb", |
| 1022 | itoa_log((int)(resolved_opts.retain_per_file_max_bytes / mb))); |
| 1023 | } |
| 1024 | |
| 1025 | /* Log per-worker memory budget */ |
| 1026 | if (cbm_mem_budget() > 0) { |
| 1027 | size_t worker_budget = cbm_mem_worker_budget(worker_count); |
| 1028 | cbm_log_info("parallel.mem.budget", "total_mb", |
| 1029 | itoa_log((int)(cbm_mem_budget() / ((size_t)CBM_SZ_1K * CBM_SZ_1K))), |
| 1030 | "per_worker_mb", |
| 1031 | itoa_log((int)(worker_budget / ((size_t)CBM_SZ_1K * CBM_SZ_1K)))); |
| 1032 | } |
| 1033 | |
| 1034 | /* Sub-phase: Ensure extraction library is initialized */ |
| 1035 | CBM_PROF_START(t_init); |
| 1036 | cbm_init(); |
| 1037 | |
| 1038 | /* Slab allocator for tree-sitter (thread-safe via TLS). Destroy any |
| 1039 | * parser this thread still holds BEFORE switching the global ts |
| 1040 | * allocator: a parser created in the mimalloc epoch (sequential run, |
| 1041 | * watcher tick) must be freed by the allocator that created it, or its |
| 1042 | * teardown after the switch routes mi pointers into plain free() |
| 1043 | * (#773). */ |
| 1044 | cbm_destroy_thread_parser(); |
| 1045 | cbm_slab_install(); |
| 1046 | CBM_PROF_END("parallel_extract", "1_init_libs", t_init); |
| 1047 | |
| 1048 | /* Sub-phase: Sort files by descending size for tail-latency reduction */ |
| 1049 | CBM_PROF_START(t_sort); |
| 1050 | file_sort_entry_t *sorted = malloc((size_t)file_count * sizeof(file_sort_entry_t)); |
| 1051 | if (!sorted) { |
| 1052 | return CBM_NOT_FOUND; |
| 1053 | } |
| 1054 | for (int i = 0; i < file_count; i++) { |
| 1055 | sorted[i].idx = i; |
| 1056 | sorted[i].size = files[i].size; |
| 1057 | } |
| 1058 | qsort(sorted, file_count, sizeof(file_sort_entry_t), compare_by_size_desc); |
| 1059 | CBM_PROF_END_N("parallel_extract", "2_sort_files", t_sort, file_count); |
| 1060 | |
| 1061 | /* Allocate per-worker state (cache-line aligned via posix_memalign) */ |
| 1062 | extract_worker_state_t *workers = NULL; |
| 1063 | if (cbm_aligned_alloc((void **)&workers, CBM_CACHE_LINE, |