From 8c9af3ec8c4d7f1b0b9f6f3a0ff1e85c16ebc4a6 Mon Sep 17 00:00:00 2001 From: Konstantin Date: Fri, 28 Aug 2026 15:35:58 +0200 Subject: [PATCH 1/6] keep CRAB out of the AFS home --- docs/concepts/backends.md | 14 ++++++++++++++ dsprod/crab.py | 4 +++- env.sh | 8 ++++++++ 3 files changed, 25 insertions(+), 1 deletion(-) diff --git a/docs/concepts/backends.md b/docs/concepts/backends.md index 89f71d4..2d5da59 100644 --- a/docs/concepts/backends.md +++ b/docs/concepts/backends.md @@ -172,6 +172,20 @@ the site they were assigned — CRAB cannot re-target a running task. A site you know is bad belongs in the static `blacklist` instead: that one is never lifted. +!!! note "CRAB does not write to your AFS home" + CRAB rewrites its task cache `~/.crab3` on *every* command, status queries included. With + `$HOME` on AFS that makes a long production depend on an AFS token: when the token lapses, + every status query fails at once with + `PermissionError: [Errno 13] Permission denied: '/afs/.../.crab3.'` and law reports it as + a status-query failure for all jobs. The `crab` wrapper `env.sh` installs therefore points + `HOME` at `$DSPROD_CRAB_HOME` (default: a per-user directory under `$TMPDIR`), so nothing in a + production run needs AFS. law passes `--proxy` to submit, status and kill, so CRAB never needs + `~/.globus` from the real home either. + + DSProd still renews Kerberos and the AFS token (`kinit -R` + `aklog`, hourly, from + `crab_poll_callback`) — but renewal can only extend a ticket that is still valid, so it is not + a substitute for keeping the production off AFS. + !!! note "Why `env.sh` matters for CRAB" law runs `crab` inside a CMSSW sandbox of its own, and dumps that sandbox's environment with bare `python` — which modern CMSSW no longer ships, and for which the DSProd venv's `python` diff --git a/dsprod/crab.py b/dsprod/crab.py index a117893..43e89ab 100644 --- a/dsprod/crab.py +++ b/dsprod/crab.py @@ -450,7 +450,9 @@ def crab_poll_callback(self, poll_data): if self._crab_kerberos_update is None: def renew_kerberos_ticket(): - update_kerberos_ticket(verbose=0) + # verbose: a silent renewal leaves no way to tell, after a credential failure, + # whether it had been running at all + update_kerberos_ticket(verbose=1) krenew = float(getattr(self, "krenew", 1) or 0) self._crab_kerberos_update = ( diff --git a/env.sh b/env.sh index b0692b6..167c5f5 100644 --- a/env.sh +++ b/env.sh @@ -156,6 +156,14 @@ action() { cat > "$ANALYSIS_PATH/soft/bin/crab" <<'CRABWRAP' #!/bin/bash source /cvmfs/cms.cern.ch/cmsset_default.sh +# CRAB rewrites its task cache ~/.crab3 (via ~/.crab3.) on *every* command, status queries +# included -- so with $HOME on AFS a long production dies the moment the AFS token lapses: +# PermissionError: [Errno 13] Permission denied: '/afs/cern.ch/user/x/xyz/.crab3.' +# and law reports it as a status-query failure for every job at once. DSProd keeps everything else +# off AFS, so give CRAB a home of its own too. law passes --proxy to submit/status/kill, so it +# never needs ~/.globus from the real home. +export HOME="${DSPROD_CRAB_HOME:-${TMPDIR:-/tmp}/dsprod_crab_home_$(id -u)}" +mkdir -p "$HOME" || exit 1 _c=$(ls -d "$ANALYSIS_PATH"/soft/CMSSW_*/ 2>/dev/null | sort | tail -1) [ -n "$_c" ] && { cd "$_c/src" && eval $(scramv1 runtime -sh 2>/dev/null); cd - >/dev/null; } exec /cvmfs/cms.cern.ch/common/crab "$@" From e2f1c0bdaeb8f78ec1a708baf5578e6bf1394ac9 Mon Sep 17 00:00:00 2001 From: Konstantin Date: Fri, 28 Aug 2026 15:46:51 +0200 Subject: [PATCH 2/6] make an excluded site actually absent from the CRAB whitelist --- dsprod/crab.py | 57 +++++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 56 insertions(+), 1 deletion(-) diff --git a/dsprod/crab.py b/dsprod/crab.py index 43e89ab..e0ce1c3 100644 --- a/dsprod/crab.py +++ b/dsprod/crab.py @@ -20,6 +20,7 @@ # refill_fraction: 0.2 # min wave size / free slots, as a fraction of parallel_jobs """ +import fnmatch import math import os import re @@ -192,6 +193,59 @@ def _rewrite_crab_job_file(job_file): f.writelines(new_lines) +#: CMS site names, from cvmfs — no network, and present wherever CRAB runs +_SITECONF = "/cvmfs/cms.cern.ch/SITECONF" + + +def known_sites(): + """Every CMS site name cvmfs knows about.""" + try: + return sorted( + n for n in os.listdir(_SITECONF) if re.match(r"^T\d_[A-Za-z0-9_]+$", n) + ) + except OSError: + return [] + + +def resolve_whitelist(whitelist, blacklist): + """A `Site.whitelist` from which `blacklist` is actually absent. + + CRAB gives the whitelist precedence: a site matched by both is *kept*, and it says so only in a + warning ("Since the whitelist has precedence, these sites are not considered in the blacklist"). + With the default all-tier globs that silently defeats every exclusion -- the configured + `crab.blacklist` and the automatic site quarantine alike. + + So a tier glob covering an excluded site is expanded, from the cvmfs site list, into the sites + it actually matches minus the excluded ones. Globs covering nothing excluded are left alone, + which keeps the pool wide and the expansion small: excluding one T2 lists the T2s and leaves + `T1_*` and `T3_*` as they are. The storage site is dropped as well -- it need not be a + processing site, and CRAB refuses a task whose whitelist names one. + """ + if not blacklist: + return list(whitelist) + sites = known_sites() + out = [] + for entry in whitelist: + hit = [b for b in blacklist if fnmatch.fnmatch(b, entry)] + if not hit: + out.append(entry) + continue + if not sites: + raise RuntimeError( + f"cannot exclude {', '.join(hit)}: they are covered by the whitelist entry " + f"'{entry}', and CRAB keeps a site that appears in both. Expanding it needs the " + f"site list at {_SITECONF}, which is not readable. Set `crab.whitelist` explicitly." + ) + out += [ + site + for site in sites + if fnmatch.fnmatch(site, entry) + and site not in blacklist + and site != _CRAB_DUMMY_SITE + ] + return out + + _CrabProxyBase = law.cms.CrabWorkflow.workflow_proxy_cls @@ -594,7 +648,8 @@ def crab_job_config(self, config, job_nums, branches=None): ) blacklist = list(blacklist) + quarantined - config.crab.Site.whitelist = [str(s) for s in whitelist or _CRAB_ALL_SITES] + sites = resolve_whitelist(whitelist or _CRAB_ALL_SITES, blacklist) + config.crab.Site.whitelist = [str(s) for s in sites] if blacklist: config.crab.Site.blacklist = [str(s) for s in blacklist] # Keep CMS's global blacklist of known-broken sites in force unless explicitly waived: From 4e70149155a3ad88c1cc9659034832938bf029f8 Mon Sep 17 00:00:00 2001 From: Konstantin Date: Fri, 28 Aug 2026 16:08:46 +0200 Subject: [PATCH 3/6] take the processing-site list from CRIC, not cvmfs SITECONF --- dsprod/crab.py | 85 +++++++++++++++++++++++++++++++++++--------------- 1 file changed, 59 insertions(+), 26 deletions(-) diff --git a/dsprod/crab.py b/dsprod/crab.py index e0ce1c3..bd97896 100644 --- a/dsprod/crab.py +++ b/dsprod/crab.py @@ -21,10 +21,13 @@ """ import fnmatch +import json import math import os import re import subprocess +import time +import urllib.request import uuid from collections import Counter, OrderedDict @@ -193,21 +196,51 @@ def _rewrite_crab_job_file(job_file): f.writelines(new_lines) -#: CMS site names, from cvmfs — no network, and present wherever CRAB runs -_SITECONF = "/cvmfs/cms.cern.ch/SITECONF" +#: CRIC's site table — the same source CRAB validates a whitelist against +_CRIC_URL = "https://cms-cric.cern.ch/api/cms/site/query/?json" +#: how long a cached site list is reused before CRIC is asked again +_CRIC_CACHE_SECONDS = 24 * 3600 -def known_sites(): - """Every CMS site name cvmfs knows about.""" + +def processing_sites(cache_path=None, url=_CRIC_URL, timeout=60): + """CMS site names that actually run jobs, newest-first from CRIC, cached on disk. + + `/cvmfs/cms.cern.ch/SITECONF` cannot be used for this: it also lists storage endpoints such as + `T1_US_FNAL_Disk` and `T3_CH_CERNBOX`, and a whitelist naming one gets the task refused -- + "A site name T1_US_FNAL_Disk that user specified is not in the list of known CMS Processing + Site Names". CRIC marks the difference: a site that runs jobs has `computeunits`. + """ + if cache_path and os.path.exists(cache_path): + if time.time() - os.path.getmtime(cache_path) < _CRIC_CACHE_SECONDS: + try: + with open(cache_path) as f: + return json.load(f) + except (OSError, ValueError): + pass try: - return sorted( - n for n in os.listdir(_SITECONF) if re.match(r"^T\d_[A-Za-z0-9_]+$", n) - ) - except OSError: - return [] + with urllib.request.urlopen(url, timeout=timeout) as response: + payload = json.load(response) + except Exception as exc: + if cache_path and os.path.exists(cache_path): + with open(cache_path) as f: # stale is better than nothing + return json.load(f) + raise RuntimeError(f"could not read the CMS site list from {url}: {exc}") + entries = payload.values() if isinstance(payload, dict) else payload + sites = sorted( + e["name"] for e in entries if e.get("name") and e.get("computeunits") + ) + if cache_path: + try: + os.makedirs(os.path.dirname(cache_path), exist_ok=True) + with open(cache_path, "w") as f: + json.dump(sites, f) + except OSError: + pass + return sites -def resolve_whitelist(whitelist, blacklist): +def resolve_whitelist(whitelist, blacklist, sites): """A `Site.whitelist` from which `blacklist` is actually absent. CRAB gives the whitelist precedence: a site matched by both is *kept*, and it says so only in a @@ -215,34 +248,30 @@ def resolve_whitelist(whitelist, blacklist): With the default all-tier globs that silently defeats every exclusion -- the configured `crab.blacklist` and the automatic site quarantine alike. - So a tier glob covering an excluded site is expanded, from the cvmfs site list, into the sites - it actually matches minus the excluded ones. Globs covering nothing excluded are left alone, - which keeps the pool wide and the expansion small: excluding one T2 lists the T2s and leaves - `T1_*` and `T3_*` as they are. The storage site is dropped as well -- it need not be a - processing site, and CRAB refuses a task whose whitelist names one. + So a tier glob covering an excluded site is expanded, from `sites`, into the sites it actually + matches minus the excluded ones. Globs covering nothing excluded are left alone, which keeps the + pool wide and the expansion small: excluding one T2 lists the T2s and leaves `T1_*` and `T3_*` + as they are. """ if not blacklist: return list(whitelist) - sites = known_sites() out = [] for entry in whitelist: hit = [b for b in blacklist if fnmatch.fnmatch(b, entry)] if not hit: out.append(entry) continue - if not sites: - raise RuntimeError( - f"cannot exclude {', '.join(hit)}: they are covered by the whitelist entry " - f"'{entry}', and CRAB keeps a site that appears in both. Expanding it needs the " - f"site list at {_SITECONF}, which is not readable. Set `crab.whitelist` explicitly." - ) + # an entry that is itself excluded simply disappears out += [ site for site in sites - if fnmatch.fnmatch(site, entry) - and site not in blacklist - and site != _CRAB_DUMMY_SITE + if fnmatch.fnmatch(site, entry) and site not in blacklist ] + if not out: + raise RuntimeError( + f"the blacklist {', '.join(blacklist)} excludes every site the whitelist " + f"{', '.join(whitelist)} allows" + ) return out @@ -648,7 +677,11 @@ def crab_job_config(self, config, job_nums, branches=None): ) blacklist = list(blacklist) + quarantined - sites = resolve_whitelist(whitelist or _CRAB_ALL_SITES, blacklist) + sites = resolve_whitelist( + whitelist or _CRAB_ALL_SITES, + blacklist, + processing_sites(os.path.join(self.ana_data_path(), "cms_sites.json")), + ) config.crab.Site.whitelist = [str(s) for s in sites] if blacklist: config.crab.Site.blacklist = [str(s) for s in blacklist] From 0fcb1abfdbe5e242b0d4eaa8dbc3e05e53989e84 Mon Sep 17 00:00:00 2001 From: Konstantin Date: Fri, 28 Aug 2026 17:09:43 +0200 Subject: [PATCH 4/6] run crab from its own home so it stops dropping crab.log in the production area --- .gitignore | 3 +++ env.sh | 11 ++++++++++- 2 files changed, 13 insertions(+), 1 deletion(-) diff --git a/.gitignore b/.gitignore index 364fcb8..38c34a0 100644 --- a/.gitignore +++ b/.gitignore @@ -12,3 +12,6 @@ __pycache__/ # user-specific config (see docs/configuration/settings.md) /config/user_custom.yaml + +# crab drops one wherever it runs; the wrapper keeps it in $DSPROD_CRAB_HOME, this catches strays +crab.log diff --git a/env.sh b/env.sh index 167c5f5..7983f24 100644 --- a/env.sh +++ b/env.sh @@ -166,7 +166,16 @@ export HOME="${DSPROD_CRAB_HOME:-${TMPDIR:-/tmp}/dsprod_crab_home_$(id -u)}" mkdir -p "$HOME" || exit 1 _c=$(ls -d "$ANALYSIS_PATH"/soft/CMSSW_*/ 2>/dev/null | sort | tail -1) [ -n "$_c" ] && { cd "$_c/src" && eval $(scramv1 runtime -sh 2>/dev/null); cd - >/dev/null; } -exec /cvmfs/cms.cern.ch/common/crab "$@" +# crab drops a crab.log wherever it is run from, which for law's status and kill calls is the +# production area. Run it from its own home instead. law submits with a *relative* `--config` and +# cwd set to the job-file directory, so make that path absolute before leaving the directory. +_args=(); _prev="" +for _a in "$@"; do + case "$_prev" in --config|-c) case "$_a" in /*) ;; *) _a="$PWD/$_a" ;; esac ;; esac + _args+=("$_a"); _prev="$_a" +done +cd "$HOME" || exit 1 +exec /cvmfs/cms.cern.ch/common/crab "${_args[@]}" CRABWRAP chmod +x "$ANALYSIS_PATH/soft/bin/crab" # law.contrib.cms's CMSSW sandbox runs bare `python` to dump its environment, but modern CMSSW From 7dbc73187afaed1f8f16754ddcceab54606c5a59 Mon Sep 17 00:00:00 2001 From: Konstantin Date: Fri, 28 Aug 2026 17:17:11 +0200 Subject: [PATCH 5/6] keep crab submit in its own directory --- env.sh | 21 +++++++++++---------- 1 file changed, 11 insertions(+), 10 deletions(-) diff --git a/env.sh b/env.sh index 7983f24..fe731e0 100644 --- a/env.sh +++ b/env.sh @@ -166,16 +166,17 @@ export HOME="${DSPROD_CRAB_HOME:-${TMPDIR:-/tmp}/dsprod_crab_home_$(id -u)}" mkdir -p "$HOME" || exit 1 _c=$(ls -d "$ANALYSIS_PATH"/soft/CMSSW_*/ 2>/dev/null | sort | tail -1) [ -n "$_c" ] && { cd "$_c/src" && eval $(scramv1 runtime -sh 2>/dev/null); cd - >/dev/null; } -# crab drops a crab.log wherever it is run from, which for law's status and kill calls is the -# production area. Run it from its own home instead. law submits with a *relative* `--config` and -# cwd set to the job-file directory, so make that path absolute before leaving the directory. -_args=(); _prev="" -for _a in "$@"; do - case "$_prev" in --config|-c) case "$_a" in /*) ;; *) _a="$PWD/$_a" ;; esac ;; esac - _args+=("$_a"); _prev="$_a" -done -cd "$HOME" || exit 1 -exec /cvmfs/cms.cern.ch/common/crab "${_args[@]}" +# crab drops a crab.log wherever it is run from, and law calls status/kill without setting a +# directory, so they inherited the caller's cwd -- the production area. Run those from crab's own +# home. `submit` must keep its directory: law runs it with cwd set to the job-file directory and +# the generated config names `scriptExe` and `inputFiles` relative to it, which CRAB resolves +# against the cwd ("Cannot find the file crab_wrapper_*.sh specified in the JobType.scriptExe +# configuration parameter"). Its log then stays next to the job files, under data/jobs/. +case "$1" in + submit) ;; + *) cd "$HOME" || exit 1 ;; +esac +exec /cvmfs/cms.cern.ch/common/crab "$@" CRABWRAP chmod +x "$ANALYSIS_PATH/soft/bin/crab" # law.contrib.cms's CMSSW sandbox runs bare `python` to dump its environment, but modern CMSSW From 4c5e07311c680922cafebfe0bb817133f09c74a2 Mon Sep 17 00:00:00 2001 From: Konstantin Date: Fri, 28 Aug 2026 15:56:24 +0200 Subject: [PATCH 6/6] resolve the premix file list once instead of in every job --- docs/concepts/tasks.md | 19 +++++++++ dsprod/run_step.py | 13 +++++- dsprod/tasks.py | 91 +++++++++++++++++++++++++++++++++++++++++- 3 files changed, 119 insertions(+), 4 deletions(-) diff --git a/docs/concepts/tasks.md b/docs/concepts/tasks.md index d95438e..7387e9e 100644 --- a/docs/concepts/tasks.md +++ b/docs/concepts/tasks.md @@ -67,6 +67,25 @@ that is worth a batch backend (`--workflow htcondor|crab`). unreachable from the worker. +### `PremixFileList` + +Resolves an era's premix pileup dataset to a plain file list on `fs_default` +(`/premix/.txt`), once, and `RunProd` passes it to `cmsDriver` as +`--pileup_input filelist:...`. + +Without it, `--pileup_input dbs:` makes **every job** resolve that dataset — ~38 000 files +— with its own DAS query. At a few thousand concurrent jobs the queries start returning nothing, +cmsDriver then writes a config with no secondary input, and cmsRun dies on + +``` +NoSecondaryFiles: RootEmbeddedFileSequence no input files specified for secondary input source +``` + +*after* the job has already produced its GEN-SIM. A 5000-job production lost 10 % of its jobs that +way. The list is identical to what a successful DAS query returns, so nothing about the physics +changes; it is stored in the production area (not `_test`) because it depends only on the +era, and a batch node refuses to build one — that query belongs on the submitting machine. + ### `RunProd` The core production task: a fused GEN→…→MiniAOD→NanoAOD chain for one `(era, point, seed)`, run diff --git a/dsprod/run_step.py b/dsprod/run_step.py index 0dcb7f8..070bdcb 100644 --- a/dsprod/run_step.py +++ b/dsprod/run_step.py @@ -93,6 +93,7 @@ def build_cmsdriver( gridpack=None, fragment_rel=None, n_threads=1, + pileup_filelist=None, ): """Assemble the cmsDriver.py command line for one step.""" out = fileout or f"{step}.root" @@ -118,6 +119,10 @@ def build_cmsdriver( # derives numberOfStreams from it. A single-threaded cmsRun in a multi-core slot wastes # the extra cores and does not get any faster. cmd += f" --mc -n {n_evt} --nThreads {int(step_params.get('nThreads', n_threads))}" + pileup = step_params.get("pileup_input") + if pileup and pileup_filelist and str(pileup).startswith(("dbs:", "das:")): + # resolved once by PremixFileList instead of by a DAS query in every job + pileup = f"filelist:{pileup_filelist}" # empty means "none": an era can drop a modifier that `default_step` sets for the others if step_params.get("procModifiers"): cmd += f" --procModifiers {step_params['procModifiers']}" @@ -127,8 +132,8 @@ def build_cmsdriver( cmd += f" --datamix {step_params['datamix']}" if "pileup" in step_params: cmd += f" --pileup {step_params['pileup']}" - if "pileup_input" in step_params: - cmd += f" --pileup_input \"{step_params['pileup_input']}\"" + if pileup: + cmd += f' --pileup_input "{pileup}"' "" cmd += "".join(f" --customise {x}" for x in step_params.get("customise", [])) customise_commands += step_params.get("customise_commands", []) # match central compression on the persisted (final) tier @@ -165,6 +170,7 @@ def run_step( gridpack=None, fragment_path=None, n_threads=1, + pileup_filelist=None, verbose=1, ): """Run one cmsDriver step in its CMSSW env, in work_dir.""" @@ -181,6 +187,7 @@ def run_step( gridpack=os.path.abspath(gridpack) if gridpack else None, fragment_rel=fragment_rel, n_threads=n_threads, + pileup_filelist=pileup_filelist, ) from .tools import ps_call @@ -200,6 +207,7 @@ def run_chain( fragment_path=None, previous_file=None, n_threads=1, + pileup_filelist=None, verbose=1, ): """Run the linear prod_steps[first_step..last_step] in work_dir, chaining outputs. @@ -228,6 +236,7 @@ def run_chain( gridpack=gridpack, fragment_path=fragment_path, n_threads=n_threads, + pileup_filelist=pileup_filelist, verbose=verbose, ) prev = out diff --git a/dsprod/tasks.py b/dsprod/tasks.py index d7b4d9b..8b39f97 100644 --- a/dsprod/tasks.py +++ b/dsprod/tasks.py @@ -126,6 +126,16 @@ def runprod_branches(eras, points): return out +def premix_dataset_for_era(conditions, era): + """The `dbs:` pileup dataset an era's steps resolve, or None when it has no premix step.""" + for step in conditions[era]["prod_steps"]: + params = run_step.resolve_step_params(conditions, era, step) + pileup = str(params.get("pileup_input") or "") + if pileup.startswith("dbs:") or pileup.startswith("das:"): + return pileup[4:] + return None + + def cmssw_releases_for_era(conditions, era): """Sorted unique (SCRAM_ARCH, CMSSW) needed to produce `era` (all prod_steps + NANO versions).""" releases = set() @@ -346,6 +356,13 @@ def storage_path(self, *parts): """Path of a product relative to `fs_default`: /.""" return os.path.join(self.storage_name(), *parts) + def premix_target(self, era): + """Where an era's premix file list lives on `fs_default`. + + Always the production area, also under `--test`: the list depends only on the era. + """ + return self.remote_target(self.prod_storage_name(), "premix", f"{era}.txt") + def gridpack_target(self, gridpack_name): """Where a gridpack lives on `fs_default`. @@ -734,6 +751,61 @@ def run(self): ) +class PremixFileList(Task, law.LocalWorkflow): + """Resolve an era's premix pileup dataset to a file list, once, and store it on `fs_default`. + + `cmsDriver --pileup_input dbs:` resolves the dataset with a DAS query *in the job*. + That dataset holds ~38 000 files, and a production asks every job to look it up: at a few + thousand concurrent jobs the queries start coming back empty, cmsDriver writes a config with no + secondary input, and cmsRun dies with + + NoSecondaryFiles: RootEmbeddedFileSequence no input files specified for secondary input + + after the job has already produced its GEN-SIM. Resolving once here and passing the result as + `filelist:` gives every job exactly what a successful DAS query would have, with no per-job + lookup. Always local: a worker is the last place that query should run. + """ + + def create_branch_map(self): + eras = [ + era + for era in self.prod_eras + if premix_dataset_for_era(self.conditions, era) + ] + return dict(enumerate(eras)) + + def output(self): + return self.premix_target(self.branch_data) + + def run(self): + era = self.branch_data + dataset = premix_dataset_for_era(self.conditions, era) + family = self.get_task_family().rsplit(".", 1)[-1] + if on_batch_node() and submitted_task_family() not in (None, family): + raise RuntimeError( + f"the premix file list for {era} is missing from fs_default, and resolving it " + f"needs a DAS query that must not run on a worker. Submit {family} (or any task " + "that requires it) from a machine with dasgoclient, then resubmit." + ) + _, out, _ = ps_call( + [f'dasgoclient -query="file dataset={dataset}"'], + shell=True, + catch_stdout=True, + verbose=1, + ) + files = [ + line.strip() for line in out.splitlines() if line.strip().endswith(".root") + ] + if not files: + raise RuntimeError( + f"DAS returned no files for the premix dataset {dataset}" + ) + print(f"PremixFileList[{era}]: {len(files)} files from {dataset}") + with self.output().localize("w") as out_local: + with open(out_local.abspath, "w") as f: + f.write("\n".join(files) + "\n") + + class RunProd(Task, HTCondorWorkflow, CrabWorkflow, law.LocalWorkflow): """Fused GEN->NANO production for one (era, point, seed); stages one nano per version.""" @@ -747,11 +819,14 @@ def create_branch_map(self): return dict(enumerate(runprod_branches(self.prod_eras, self.prod_points))) def workflow_requires(self): - return { + reqs = { "voms": CreateVomsProxy.req(self), "cmssw": InstallCMSSW.req(self, workflow="local"), "gridpack": MakeGridpack.req(self), } + if PremixFileList.req(self, workflow="local").get_branch_map(): + reqs["premix"] = PremixFileList.req(self, workflow="local") + return reqs def requires(self): era, pi, _ = self.branch_data @@ -759,13 +834,20 @@ def requires(self): gp_branch = self.gridpack_index()[ self.process.gridpack_name(self.prod_points[pi]) ] - return { + reqs = { "voms": CreateVomsProxy.req(self), "cmssw": InstallCMSSW.req( self, branch=self.prod_eras.index(era), workflow="local" ), "gridpack": MakeGridpack.req(self, branch=gp_branch), } + premix = PremixFileList.req(self, workflow="local") + eras_with_premix = list(premix.get_branch_map().values()) + if era in eras_with_premix: + reqs["premix"] = PremixFileList.req( + self, branch=eras_with_premix.index(era), workflow="local" + ) + return reqs def _staged_target(self, era, point, version, seed): name = self.process.point_name(point) @@ -790,6 +872,10 @@ def run(self): self.input()["gridpack"].localize("r") ).abspath work_dir, is_tmp = self.law_job_home() + premix = self.input().get("premix") + premix_list = ( + stack.enter_context(premix.localize("r")).abspath if premix else None + ) try: miniaod = run_step.run_chain( self.conditions, @@ -802,6 +888,7 @@ def run(self): gridpack=gridpack, fragment_path=fragment, n_threads=int(self.n_cpus), + pileup_filelist=premix_list, ) for version in self.nano_versions(era): nano_out = run_step.run_nano(