diff --git a/Analysis/HistProducerFromNTuple.py b/Analysis/HistProducerFromNTuple.py index 679e9021..d46e10b2 100644 --- a/Analysis/HistProducerFromNTuple.py +++ b/Analysis/HistProducerFromNTuple.py @@ -11,7 +11,6 @@ import FLAF.Common.HistHelper as HistHelper import FLAF.Common.Utilities as Utilities from FLAF.Common.Setup import Setup -from FLAF.RunKit.run_tools import ps_call def find_keys(inFiles_list): @@ -36,6 +35,10 @@ def SaveHist(key_tuple, outFile, hist_list, hist_name, unc, scale, verbose=0): dir_ptr = Utilities.mkdir(outFile, dir_name) merged_hist = model.GetHistogram().Clone() + # Detach from the current ROOT directory: the histogram is persisted explicitly via + # WriteTObject below, so it must not also be auto-flushed into the output file's root + # (which would leave one stray, unnamed histogram per call when writing directly). + merged_hist.SetDirectory(0) N_bins = ( unit_hist.GetNbins() if hasattr(unit_hist, "GetNbins") @@ -65,14 +68,21 @@ def SaveHist(key_tuple, outFile, hist_list, hist_name, unc, scale, verbose=0): dir_ptr.WriteTObject(merged_hist, final_hist_name, "Overwrite") -def GetUnitBinHist(rdf, var, filter_to_apply, weight_name, unc, scale): +def BookUnitHist(rdf_filtered, var, weight_name): + """Book the unit-bin histogram for ``var`` on an ALREADY-filtered RDataFrame node. + + The selection filter is applied once by the caller and the resulting node is shared + across all variables, so the (compound) channel/region/category cut is evaluated once + per event instead of once per variable. + """ var_entry = HistHelper.findBinEntry(hist_cfg_dict, var) dims = ( 1 if not hist_cfg_dict[var_entry].get("var_list", False) else len(hist_cfg_dict[var_entry]["var_list"]) ) - + if dims < 1 or dims > 3: + raise RuntimeError("Only 1D, 2D and 3D histograms are supported") model, unit_bin_model = HistHelper.GetModel( hist_cfg_dict, var, dims, return_unit_bin_model=True ) @@ -81,106 +91,76 @@ def GetUnitBinHist(rdf, var, filter_to_apply, weight_name, unc, scale): if dims > 1 else [f"{var}_bin"] ) - - rdf_filtered = rdf.Filter(filter_to_apply) - if dims >= 1 and dims <= 3: - mkhist_fn = getattr(rdf_filtered, f"Histo{dims}D") - unit_hist = mkhist_fn(unit_bin_model, *var_bin_list, weight_name) - else: - raise RuntimeError("Only 1D, 2D and 3D histograms are supported") + mkhist_fn = getattr(rdf_filtered, f"Histo{dims}D") + unit_hist = mkhist_fn(unit_bin_model, *var_bin_list, weight_name) return model, unit_hist -def SaveSingleHistSet( - all_trees, - var, - filter_expr, - unc, - scale, - key, - outFile, - is_shift_unc, - treeName, - further_cut_name=None, -): - hist_list = [] - if is_shift_unc: - tree_prefix = f"Events__{unc}__{scale}" - rdf_shift = all_trees[tree_prefix] - model, unit_hist = GetUnitBinHist( - rdf_shift, var, filter_expr, "weight_Central", unc, scale - ) - hist_list.append((model, unit_hist, rdf_shift)) - else: - weight_name = f"weight_{unc}_{scale}" if unc != "Central" else "weight_Central" - rdf_central = all_trees[treeName] - model, unit_hist = GetUnitBinHist( - rdf_central, var, filter_expr, weight_name, unc, scale - ) - hist_list.append((model, unit_hist, rdf_central)) - +def _make_save_fn(key_tuple, outFile, model, unit_hist, rdf, var, unc, scale): def save_fn(): - if hist_list: - key_tuple = key - if further_cut_name: - key_tuple = key + (further_cut_name,) - SaveHist(key_tuple, outFile, hist_list, var, unc, scale) + SaveHist(key_tuple, outFile, [(model, unit_hist, rdf)], var, unc, scale) return save_fn -def BuildHistActions( - tmp_file_root, +def BuildAllHistActions( uncs_to_compute, unc_cfg_dict, all_trees, - var, + vars_to_process, key_filter_dict, further_cuts, treeName, + var_tmp_files, ): - """Register all histogram actions for var against the open TFile. + """Register histogram actions for every variable, sharing one filtered RDataFrame node + per (uncertainty, scale, selection-key, further-cut) across ALL variables. + + The dominant per-event cost is evaluating the compound channel/region/category filter, + not filling the (precomputed unit-bin) histogram. Booking each variable on its own + ``rdf.Filter(...)`` re-evaluated the identical selection once per variable; sharing the + filtered node collapses ~Nvar redundant filter passes into a single one. Histograms are + identical -- only the RDF graph is smaller and the single event loop does far less work. - Returns a list of save_fn callables without triggering the RDF event loop - or closing the file. Call all returned functions after collecting actions - for every variable so ROOT can execute them in a single event-loop pass. + Returns a list of save callables; invoke them after booking so ROOT runs one event loop. """ save_fns = [] + cut_names = list(further_cuts.keys()) if further_cuts else [None] for unc, scales in uncs_to_compute.items(): is_shift_unc = unc in unc_cfg_dict["shape"].keys() for scale in scales: + if is_shift_unc: + rdf_base = all_trees[f"Events__{unc}__{scale}"] + weight_name = "weight_Central" + else: + rdf_base = all_trees[treeName] + weight_name = ( + f"weight_{unc}_{scale}" if unc != "Central" else "weight_Central" + ) for key, filter_to_apply_base in key_filter_dict.items(): - if further_cuts: - for further_cut_name in further_cuts.keys(): - filter_to_apply_final = ( - f"{filter_to_apply_base} && {further_cut_name}" - ) - save_fn = SaveSingleHistSet( - all_trees, - var, - filter_to_apply_final, - unc, - scale, - key, - tmp_file_root, - is_shift_unc, - treeName, - further_cut_name, - ) - save_fns.append(save_fn) - else: - save_fn = SaveSingleHistSet( - all_trees, - var, - filter_to_apply_base, - unc, - scale, - key, - tmp_file_root, - is_shift_unc, - treeName, + for further_cut_name in cut_names: + filter_to_apply_final = ( + f"{filter_to_apply_base} && {further_cut_name}" + if further_cut_name + else filter_to_apply_base ) - save_fns.append(save_fn) + rdf_filtered = rdf_base.Filter(filter_to_apply_final) + key_tuple = key + (further_cut_name,) if further_cut_name else key + for var in vars_to_process: + model, unit_hist = BookUnitHist(rdf_filtered, var, weight_name) + _, tmp_root_file = var_tmp_files[var] + save_fns.append( + _make_save_fn( + key_tuple, + tmp_root_file, + model, + unit_hist, + rdf_filtered, + var, + unc, + scale, + ) + ) return save_fns @@ -277,37 +257,37 @@ def BuildHistActions( os.makedirs(args.outDir, exist_ok=True) if all_trees: - # Open a tmp ROOT file per variable and register all histogram actions. - # Collecting actions for all variables before triggering lets ROOT execute - # them in a single event-loop pass over the input files. + # Open a tmp ROOT file per variable, then register all histogram actions sharing one + # filtered RDataFrame node per selection across variables (see BuildAllHistActions). + # Collecting every action before triggering lets ROOT execute them in a single + # event-loop pass over the input files. + # Write each variable's histograms directly into its final, compressed output file. + # SaveHist persists objects via WriteTObject as the actions run, so once the single + # event loop has executed we just close the files -- no per-variable hadd recompress + # pass (209 == LZMA level 9, matching the previous `hadd -f209` output compression). var_tmp_files = {} - all_save_fns = [] for var in vars_to_process: - tmp_path = os.path.join(args.outDir, f"tmp_{var}.root") - tmp_root_file = ROOT.TFile(tmp_path, "RECREATE") - var_tmp_files[var] = (tmp_path, tmp_root_file) - fns = BuildHistActions( - tmp_root_file, - uncs_to_compute, - unc_cfg_dict, - all_trees, - var, - key_filter_dict, - further_cuts, - treeName, - ) - all_save_fns.extend(fns) + out_path = os.path.join(args.outDir, f"{var}.root") + out_root_file = ROOT.TFile(out_path, "RECREATE", "", 209) + var_tmp_files[var] = (out_path, out_root_file) + + all_save_fns = BuildAllHistActions( + uncs_to_compute, + unc_cfg_dict, + all_trees, + vars_to_process, + key_filter_dict, + further_cuts, + treeName, + var_tmp_files, + ) for fn in all_save_fns: fn() for var in vars_to_process: - tmp_path, tmp_root_file = var_tmp_files[var] - tmp_root_file.Close() - out_path = os.path.join(args.outDir, f"{var}.root") - hadd_str = f"hadd -f209 -j -O {out_path} {tmp_path}" - ps_call([hadd_str], True) - os.remove(tmp_path) + _, out_root_file = var_tmp_files[var] + out_root_file.Close() time_elapsed = time.time() - start print(f"execution time = {time_elapsed} ") diff --git a/Analysis/tasks.py b/Analysis/tasks.py index 36d5f896..a949ebc1 100644 --- a/Analysis/tasks.py +++ b/Analysis/tasks.py @@ -1,11 +1,8 @@ import law import os import contextlib -import hashlib import luigi -import queue import shutil -import threading from concurrent.futures import ThreadPoolExecutor @@ -506,16 +503,14 @@ class HistFromNtupleProducerTask(Task, HTCondorWorkflow, law.LocalWorkflow): max_runtime = copy_param(HTCondorWorkflow.max_runtime, 10.0) n_cpus = copy_param(HTCondorWorkflow.n_cpus, 2) variables = luigi.Parameter(default="") - # Fixed NUMBER of variable batches (not batch size): branch indices stay stable when - # the set of variables changes. n_var_batches=1 => one job per dataset (the #257 extreme). - n_var_batches = luigi.IntParameter(default=10) - - @staticmethod - def _var_batch_id(var_name, n_batches): - # Deterministic across processes (unlike the salted built-in hash()), so a given - # variable always lands in the same batch regardless of which other variables exist. - digest = hashlib.md5(var_name.encode("utf-8")).hexdigest() - return int(digest, 16) % n_batches + # Number of input ntuple files processed per job. A job reads its chunk of files once + # and fills the histograms for ALL active variables in a single RDataFrame event-loop + # pass (one traversal fills every booked histogram, so the dominant per-event cost -- + # I/O + filter/define evaluation -- is shared across variables). Work is therefore split + # by FILES, never by variable: a per-variable split re-reads the same events once per + # variable batch, which is what made big datasets (e.g. TTtoLNu2Q with ~500 files read + # 10x) exceed the wall-time limit. Big datasets are parallelized by chunking their files. + n_files_per_job = luigi.IntParameter(default=20) @property def bundle_flavours(self): @@ -557,7 +552,7 @@ def workflow_requires(self): ) return reqs branch_set = set() - for br_idx, (dataset_name, prod_br_list, var_batch) in self.branch_map.items(): + for br_idx, (dataset_name, prod_br_list, chunk_id) in self.branch_map.items(): branch_set.update(prod_br_list) branches = tuple(sorted(branch_set)) reqs["HistTupleProducerTask"] = HistTupleProducerTask.req( @@ -568,7 +563,7 @@ def workflow_requires(self): return reqs def requires(self): - dataset_name, prod_br_list, var_batch = self.branch_data + dataset_name, prod_br_list, chunk_id = self.branch_data return [ HistTupleProducerTask.req( self, @@ -607,70 +602,49 @@ def _build_branch_map(self): ) in HistTupleBranchMap.items(): dataset_to_branches.setdefault(histTuple_dataset_name, []).append(prod_br) - # Each (dataset, variable-batch) is its own branch/job: a job localizes the dataset - # inputs once and produces the histograms for just its batch of variables (avoids the - # old nDatasets x nVariables re-localization while keeping per-batch parallelism). - # - # The number of batches is FIXED (n_var_batches) and a variable's batch is a - # deterministic function of its name, so branch indices - # index = dataset_position * n_var_batches + batch_id - # are stable when the set of variables changes: adding/removing a variable only - # makes the affected (dataset, batch) branches incomplete, without renumbering any - # other branch (so a running production's HTCondor job map stays aligned). Empty - # batches yield an empty output() and are trivially complete (never submitted). - n_batches = max(1, self.n_var_batches) - batched_vars = [[] for _ in range(n_batches)] - for var_name in ( - v["name"] if isinstance(v, dict) else v for v in self.active_variables - ): - batched_vars[self._var_batch_id(var_name, n_batches)].append(var_name) - batched_vars = [sorted(b) for b in batched_vars] - - for ds_pos, (dataset_name, prod_br_list) in enumerate( - sorted(dataset_to_branches.items()) - ): - for batch_id in range(n_batches): - idx = ds_pos * n_batches + batch_id + # Each (dataset, file-chunk) is its own branch/job: a job localizes its chunk of + # input files once and produces the histograms for ALL active variables in a single + # event-loop pass. Variables are NOT split across jobs (the event loop reads each + # event once and fills every booked histogram in that pass, so adding variables is + # nearly free; splitting them would instead re-read the same events per batch). Big + # datasets are parallelized by chunking their files into groups of n_files_per_job, + # so no single job has to read all ~500 files of e.g. TTtoLNu2Q. + n_files = max(1, self.n_files_per_job) + idx = 0 + for dataset_name, prod_br_list in sorted(dataset_to_branches.items()): + prod_br_list = sorted(prod_br_list) + for chunk_id, start in enumerate(range(0, len(prod_br_list), n_files)): branches[idx] = ( dataset_name, - sorted(prod_br_list), - batched_vars[batch_id], + prod_br_list[start : start + n_files], + chunk_id, ) + idx += 1 return branches - def _empty_batch_placeholder(self): - # Existing local placeholder used as the output of an empty variable bucket, so the - # branch is always complete and is never submitted to HTCondor. Created idempotently - # (only ever evaluated on the submit side, where the data path is writable). - path = self.local_path("empty_batch.placeholder") - if not os.path.exists(path): - os.makedirs(os.path.dirname(path), exist_ok=True) - open(path, "a").close() - return law.LocalFileTarget(path) - @workflow_condition.output def output(self): - dataset_name, prod_br_list, var_batch = self.branch_data - if not var_batch: - return self._empty_batch_placeholder() + dataset_name, prod_br_list, chunk_id = self.branch_data outputs = {} - for var_name in var_batch: + for var in self.active_variables: + var_name = var["name"] if isinstance(var, dict) else var output_path = os.path.join( self.version, "Hists_split", self.period, var_name, - f"{dataset_name}.root", + dataset_name, + f"chunk_{chunk_id}.root", ) outputs[var_name] = self.remote_target(output_path, fs=self.fs_HistTuple) return outputs def run(self): - dataset_name, prod_br_list, var_batch = self.branch_data - if not var_batch: - # Empty bucket: output is the placeholder (already complete); nothing to do. - return + dataset_name, prod_br_list, chunk_id = self.branch_data + var_names = [ + v["name"] if isinstance(v, dict) else v for v in self.active_variables + ] job_home, remove_job_home = self.law_job_home() customisation_dict = getCustomisationSplit(self.customisations) channels = ( @@ -690,22 +664,38 @@ def run(self): ) nMT = self.n_cpus * 2 if self.effective_workflow == "htcondor" else 8 - # Determine which variables of this batch still need to be produced. + # Determine which variables still need to be produced for this (dataset, chunk). outputs = self.output() - vars_to_run = [v for v in var_batch if not outputs[v].exists()] + vars_to_run = [v for v in var_names if not outputs[v].exists()] if not vars_to_run: print( f"All outputs already exist for dataset {dataset_name} " - f"batch {var_batch}, skipping" + f"chunk {chunk_id}, skipping" ) if remove_job_home: shutil.rmtree(job_home) return - inputs = list(self.input()) - n_inputs = len(inputs) + # Localize this chunk's input files concurrently, then run the producer once over all + # of them. A chunk is small by construction (n_files_per_job), and the producer fills + # every variable in a single event-loop pass, writing one .root per variable -- + # so there is nothing to stage in waves or merge afterwards. + max_dl = max(1, int(self.global_params.get("max_simultaneous_downloads", 8))) + out_dir = os.path.join(job_home, "hists") + os.makedirs(out_dir, exist_ok=True) + + def _localize(inp): + file_stack = contextlib.ExitStack() + return file_stack.enter_context(inp.localize("r")).abspath, file_stack + + with contextlib.ExitStack() as stack: + with ThreadPoolExecutor(max_workers=max_dl) as executor: + localized = list(executor.map(_localize, self.input())) + local_inputs = [] + for abspath, file_stack in localized: + stack.callback(file_stack.close) + local_inputs.append(abspath) - def _run_producer(out_dir, files): cmd = [ "python3", HistFromNtupleProducer, @@ -737,118 +727,13 @@ def _run_producer(out_dir, files): cmd.extend(["--customisations", self.customisations]) if self.user_custom: cmd.extend(["--user-custom", self.user_custom]) - cmd.extend(files) + cmd.extend(local_inputs) ps_call(cmd, verbose=1) - round_outputs = {var: [] for var in vars_to_run} - - # ---- Queue-based download/process pipeline ---- - # A bounded pool of downloader threads localizes the input files concurrently and - # pushes each ready local path onto `ready_q`; the single processing loop (the producer - # is itself multi-threaded via nMT) consumes them in waves -- block for >=1 file, then - # take everything already downloaded -- and hadds the per-wave partials at the end, so - # download and processing overlap. Two limits from the global config bound resource use: - # * max_simultaneous_downloads (default 4): concurrent transfers. - # * max_total_download_size_gb (default 10): total size of localized-but-not-yet- - # processed files. A downloader waits when the budget is exhausted and resumes once a - # processed wave releases its files. At least one file is always admitted, so a single - # file larger than the whole budget cannot deadlock (the cap is then soft). - max_parallel_dl = max( - 1, int(self.global_params.get("max_simultaneous_downloads", 4)) - ) - max_download_bytes = ( - float(self.global_params.get("max_total_download_size_gb", 10)) * 1024**3 - ) - - ready_q = queue.Queue() - errors = [] - budget_cond = threading.Condition() - downloaded_bytes = [0.0] - - def _download_one(inp): - try: - with budget_cond: - # Wait for budget, but always admit at least one file (downloaded_bytes==0). - while ( - downloaded_bytes[0] >= max_download_bytes - and downloaded_bytes[0] > 0 - ): - budget_cond.wait() - stack = contextlib.ExitStack() - abspath = stack.enter_context(inp.localize("r")).abspath - size = os.path.getsize(abspath) - with budget_cond: - downloaded_bytes[0] += size - ready_q.put((abspath, stack, size)) - except Exception as e: # noqa: BLE001 - surfaced to the main thread - errors.append(e) - ready_q.put(("__error__", None, 0)) - - def _release(stack, size): - if stack is not None: - stack.close() - with budget_cond: - downloaded_bytes[0] -= size - budget_cond.notify_all() - - executor = ThreadPoolExecutor(max_workers=max_parallel_dl) - for inp in inputs: - executor.submit(_download_one, inp) - - try: - remaining = n_inputs - round_idx = 0 - while remaining > 0: - wave = [] - # Block for the first ready file, then take everything already downloaded. - item = ready_q.get() - if item[0] == "__error__": - raise errors[0] - wave.append(item) - while len(wave) < remaining: - try: - item = ready_q.get_nowait() - except queue.Empty: - break - if item[0] == "__error__": - raise errors[0] - wave.append(item) - - remaining -= len(wave) - round_dir = os.path.join(job_home, "rounds", str(round_idx)) - os.makedirs(round_dir, exist_ok=True) - _run_producer(round_dir, [abspath for abspath, _, _ in wave]) - for var in vars_to_run: - round_outputs[var].append(os.path.join(round_dir, f"{var}.root")) - round_idx += 1 - # Release this wave's local copies and free their download budget. - for _, stack, size in wave: - _release(stack, size) - finally: - executor.shutdown(wait=True) - # Drain and release anything downloaded after an exception. - while True: - try: - _, stack, size = ready_q.get_nowait() - except queue.Empty: - break - _release(stack, size) - - if errors: - raise errors[0] - - # Merge per-wave partial outputs per variable and upload. + # Upload the single per-variable output file. for var in vars_to_run: - files = round_outputs[var] - if len(files) == 1: - with outputs[var].localize("w") as tmp_out: - shutil.move(files[0], tmp_out.abspath) - else: - merged = os.path.join(job_home, f"merged_{var}.root") - hadd_cmd = f"hadd -f209 -j -O {merged} " + " ".join(files) - ps_call([hadd_cmd], True) - with outputs[var].localize("w") as tmp_out: - shutil.move(merged, tmp_out.abspath) + with outputs[var].localize("w") as tmp_out: + shutil.move(os.path.join(out_dir, f"{var}.root"), tmp_out.abspath) if remove_job_home: shutil.rmtree(job_home) @@ -938,22 +823,21 @@ def _build_branch_map(self): ).create_branch_map() if not hfn_branch_map: return {} - # HFN now branches per (dataset, variable-batch). Map each variable to the HFN - # branch (one per dataset) whose batch produced it, so this merger branch only - # depends on the HFN jobs that actually wrote its variable. - var_to_dataset_branch = {} # var_name -> {dataset_name: hfn_br_idx} - for br_idx, (dataset_name, _prod_br_list, var_batch) in sorted( + # HFN branches per (dataset, file-chunk), and every chunk produces every active + # variable. So each merger branch (one per variable) depends on all HFN branches; + # we record the dataset name aligned with each HFN branch so the merger can map each + # input file to its process type (multiple chunks of the same dataset are summed). + hfn_br_indices = [] + dataset_names = [] + for br_idx, (dataset_name, _prod_br_list, _chunk_id) in sorted( hfn_branch_map.items() ): - for v in var_batch: - var_to_dataset_branch.setdefault(v, {})[dataset_name] = br_idx + hfn_br_indices.append(br_idx) + dataset_names.append(dataset_name) # One HistMerger branch per active variable. branches = {} for k, var in enumerate(self.active_variables): var_name = var["name"] if isinstance(var, dict) else var - ds_to_br = var_to_dataset_branch.get(var_name, {}) - dataset_names = sorted(ds_to_br) - hfn_br_indices = [ds_to_br[ds] for ds in dataset_names] branches[k] = (var_name, hfn_br_indices, dataset_names) return branches @@ -1007,12 +891,34 @@ def run(self): all_datasets = [] local_inputs = [] with contextlib.ExitStack() as stack: - for inp in self.input(): - # Each inp is a dict {var_name: FileTarget} from HistFromNtupleProducerTask. - var_file = inp[var_name] - dataset_name = os.path.basename(var_file.abspath).split(".")[0] + # `datasets` is aligned with self.input() (both follow requires()/br_indices + # order); multiple file-chunks of the same dataset appear as repeated entries. + # Each input is a tiny remote histogram file, but every davs localize pays a fixed + # connection cost, so localizing the O(100s) of (dataset, chunk) inputs serially + # dominated the merge. Download them concurrently -- each in its own (thread-safe) + # ExitStack registered for cleanup; ThreadPoolExecutor.map preserves order so + # local_inputs stays aligned with all_datasets (mapped to process types below). + var_targets = [ + (inp[var_name], dataset_name) + for inp, dataset_name in zip(self.input(), datasets) + ] + max_dl = max( + 1, int(self.global_params.get("max_simultaneous_downloads", 8)) + ) + + def _localize(var_file): + file_stack = contextlib.ExitStack() + abspath = file_stack.enter_context(var_file.localize("r")).abspath + return abspath, file_stack + + with ThreadPoolExecutor(max_workers=max_dl) as executor: + localized = list(executor.map(lambda t: _localize(t[0]), var_targets)) + for (var_file, dataset_name), (abspath, file_stack) in zip( + var_targets, localized + ): + stack.callback(file_stack.close) all_datasets.append(dataset_name) - local_inputs.append(stack.enter_context(var_file.localize("r")).abspath) + local_inputs.append(abspath) dataset_names = ",".join(smpl for smpl in all_datasets) all_outputs_merged = [] if len(uncNames) == 1: diff --git a/docs/reference/tasks.md b/docs/reference/tasks.md index f8cac150..32ac7804 100644 --- a/docs/reference/tasks.md +++ b/docs/reference/tasks.md @@ -39,10 +39,13 @@ producers"), writing **histTuples**. ### `HistFromNtupleProducerTask` Fills **histograms** of the requested variables from the histTuples, including systematic -variations. **Branches over variables.** +variations. **Branches over (dataset, file-chunk):** each job reads its chunk of input files +once and fills **all** active variables in a single event-loop pass. Large datasets are +parallelized by splitting their files into chunks (work is never split per variable, which +would re-read the same events once per variable). -- **Parameters:** `--variables` (string; restrict which variables), `--n-var-batches` (int, - default `10`; how variables are grouped into branches). +- **Parameters:** `--variables` (string; restrict which variables), `--n-files-per-job` (int, + default `20`; input files processed per branch). ### `HistMergerTask` Merges the per-piece histograms into per-process histograms ready for plotting and fitting. diff --git a/docs/workflow/walkthrough.md b/docs/workflow/walkthrough.md index 797558ae..77bbf7a1 100644 --- a/docs/workflow/walkthrough.md +++ b/docs/workflow/walkthrough.md @@ -76,11 +76,12 @@ law run FLAF.Analysis.tasks.HistTupleProducerTask --period $ERA --version $VER - ## Stage 3 — Fill and merge histograms `HistFromNtupleProducerTask` fills **histograms** of the requested variables from the histTuples — -**one branch per variable** — including systematic variations. `HistMergerTask` merges the pieces -into per-process histograms ready for plotting and fitting. +**one branch per (dataset, file-chunk)**, filling all variables in a single pass — including +systematic variations. `HistMergerTask` merges the pieces into per-process histograms ready for +plotting and fitting. ```sh -# Fill histograms (restrict variables with --variables, batch with --n-var-batches): +# Fill histograms (restrict variables with --variables, set files/job with --n-files-per-job): law run FLAF.Analysis.tasks.HistFromNtupleProducerTask --period $ERA --version $VER --workflow local # Merge them: diff --git a/run_tools/law_customizations.py b/run_tools/law_customizations.py index 0c6e9dc3..e3e13adc 100644 --- a/run_tools/law_customizations.py +++ b/run_tools/law_customizations.py @@ -197,7 +197,7 @@ def _branch_map_cache_key(self): getattr(self, "producer_to_run", None), getattr(self, "producer_to_aggregate", None), getattr(self, "variables", None), - getattr(self, "n_var_batches", None), + getattr(self, "n_files_per_job", None), ) def cached_branch_map(self, build_fn):