diff --git a/pandajedi/jedicore/JediTaskBuffer.py b/pandajedi/jedicore/JediTaskBuffer.py
index 4aeb45389..0e5c85791 100644
--- a/pandajedi/jedicore/JediTaskBuffer.py
+++ b/pandajedi/jedicore/JediTaskBuffer.py
@@ -6,7 +6,7 @@
from pandacommon.pandalogger.PandaLogger import PandaLogger
from pandajedi.jediconfig import jedi_config
-from pandaserver.taskbuffer import ParseJobXML, TaskBuffer
+from pandaserver.taskbuffer import TaskBuffer
from pandaserver.taskbuffer.InputChunk import InputChunk
from pandaserver.taskbuffer.JediCacheSpec import JediCacheSpec
from pandaserver.taskbuffer.JediDatasetSpec import JediDatasetSpec
@@ -72,7 +72,6 @@ def insertFilesForDataset_JEDI(
nChunksForScout: Any,
includePatt: list[str],
excludePatt: list[str],
- xmlConfig: Any,
noWaitParent: bool,
parent_tid: int | None,
pid: str | None,
@@ -111,7 +110,6 @@ def insertFilesForDataset_JEDI(
nChunksForScout,
includePatt,
excludePatt,
- xmlConfig,
noWaitParent,
parent_tid,
pid,
@@ -353,7 +351,6 @@ def getOutputFiles_JEDI(
instantiatedSite: str | None = None,
isUnMerging: bool = False,
isPrePro: bool = False,
- xmlConfigJob: ParseJobXML.dom_job | None = None,
siteDsMap: dict[Any, Any] | None = None,
middleName: str = "",
registerDatasets: bool = False,
@@ -372,7 +369,6 @@ def getOutputFiles_JEDI(
instantiatedSite,
isUnMerging,
isPrePro,
- xmlConfigJob,
siteDsMap,
middleName,
registerDatasets,
diff --git a/pandajedi/jediorder/ContentsFeeder.py b/pandajedi/jediorder/ContentsFeeder.py
index b69430d84..add2efeae 100644
--- a/pandajedi/jediorder/ContentsFeeder.py
+++ b/pandajedi/jediorder/ContentsFeeder.py
@@ -188,11 +188,7 @@ def feed_contents_to_tasks(self, task_ds_list: list[tuple[int, list[JediDatasetS
nFilesPerJob = taskSpec.getNumFilesPerJob()
# the number of chunks used by scout
nChunksForScout = 10
- # load XML
- if taskSpec.useLoadXML():
- xmlConfig = taskParamMap["loadXML"]
- else:
- xmlConfig = None
+
# skip files used by another task
if "skipFilesUsedBy" in taskParamMap:
skipFilesUsedBy = taskParamMap["skipFilesUsedBy"]
@@ -616,7 +612,6 @@ def feed_contents_to_tasks(self, task_ds_list: list[tuple[int, list[JediDatasetS
nChunksForScout,
includePatt,
excludePatt,
- xmlConfig,
noWaitParent,
taskSpec.parent_tid,
self.pid,
diff --git a/pandajedi/jediorder/JobGenerator.py b/pandajedi/jediorder/JobGenerator.py
index 3d0853ecf..d2494c5d2 100644
--- a/pandajedi/jediorder/JobGenerator.py
+++ b/pandajedi/jediorder/JobGenerator.py
@@ -28,7 +28,7 @@
from pandaserver.dataservice import DataServiceUtils
from pandaserver.dataservice.DataServiceUtils import select_scope
from pandaserver.srvcore import CoreUtils
-from pandaserver.taskbuffer import EventServiceUtils, JobUtils, ParseJobXML
+from pandaserver.taskbuffer import EventServiceUtils, JobUtils
from pandaserver.taskbuffer.FileSpec import FileSpec
from pandaserver.taskbuffer.InputChunk import InputChunk
from pandaserver.taskbuffer.JediDatasetSpec import JediDatasetSpec
@@ -1019,11 +1019,6 @@ def doGenerate(
# register datasets
registerDatasets = taskSpec.toRegisterDatasets()
try:
- # load XML
- xmlConfig = None
- if taskSpec.useLoadXML():
- loadXML = taskParamMap["loadXML"]
- xmlConfig = ParseJobXML.dom_parser(xmlStr=loadXML)
# using grouping with boundaryID
useBoundary = taskSpec.useGroupWithBoundaryID()
# loop over all sub chunks
@@ -1096,12 +1091,7 @@ def doGenerate(
fetched_out_sub_chunks = {}
fetched_serial_numbers = {}
fetched_parallel_out_map = {}
- if (
- xmlConfig is None
- and (useBoundary is None or not useBoundary["outMap"])
- and not taskSpec.on_site_merging()
- and taskSpec.getFieldNumToLFN() is None
- ):
+ if (useBoundary is None or not useBoundary["outMap"]) and not taskSpec.on_site_merging() and taskSpec.getFieldNumToLFN() is None:
for site_name in n_jobs_per_site:
# set site name if the chunks instantiate template datasets
if to_produce_outputs_merged_later or (taskSpec.instantiateTmpl() and taskSpec.instantiateTmplSite()):
@@ -1123,7 +1113,6 @@ def doGenerate(
site_to_instantiate,
to_produce_outputs_merged_later,
False,
- None,
siteDsMap,
"",
registerDatasets,
@@ -1699,17 +1688,7 @@ def doGenerate(
except Exception:
jobSpec.resource_type = "Undefined"
tmpLog.error(f"set resource_type excepted with {traceback.format_exc()}")
- # XML config
- xmlConfigJob = None
- if xmlConfig is not None:
- if boundaryID is None:
- tmpLog.error(f"failed to get XML config for N={boundaryID}")
- return failedRet
- try:
- xmlConfigJob = xmlConfig.jobs[boundaryID]
- except Exception:
- tmpLog.error(f"failed to get XML config for N={boundaryID}")
- return failedRet
+
# num of output files per job
if taskSpec.on_site_merging():
readStat, taskParamMap = self.readTaskParams(taskSpec, taskParamMap, tmpLog)
@@ -1740,7 +1719,6 @@ def doGenerate(
instantiatedSite,
isUnMerging,
False,
- xmlConfigJob,
siteDsMap,
middleName,
registerDatasets,
@@ -1822,10 +1800,7 @@ def doGenerate(
tmpBuildFileSpec = copy.copy(buildFileSpec)
jobSpec.addFile(tmpBuildFileSpec)
paramList.append(("LIB", buildFileSpec.lfn))
- # placeholders for XML config
- if xmlConfigJob is not None:
- paramList.append(("XML_OUTMAP", xmlConfigJob.get_outmap_str(outSubChunk)))
- paramList.append(("XML_EXESTR", xmlConfigJob.exec_string_enc()))
+
# middle name
paramList.append(("MIDDLENAME", middleName))
# segment ID
diff --git a/pandajedi/jediorder/JobSplitter.py b/pandajedi/jediorder/JobSplitter.py
index 293b4f132..ca359dc24 100644
--- a/pandajedi/jediorder/JobSplitter.py
+++ b/pandajedi/jediorder/JobSplitter.py
@@ -52,7 +52,7 @@ def doSplit(
nEventsPerJob = taskSpec.getNumEventsPerJob()
# number of files per job if defined
nFilesPerJob = taskSpec.getNumFilesPerJob()
- if nFilesPerJob is None and nEventsPerJob is None and inputChunk.useScout() and not taskSpec.useLoadXML() and not taskSpec.respectSplitRule():
+ if nFilesPerJob is None and nEventsPerJob is None and inputChunk.useScout() and not taskSpec.respectSplitRule():
nFilesPerJob = 1
# grouping with boundaryID
useBoundary = taskSpec.useGroupWithBoundaryID()
diff --git a/pandajedi/jedirefine/TaskRefinerBase.py b/pandajedi/jedirefine/TaskRefinerBase.py
index e8c570886..76a5abd1d 100644
--- a/pandajedi/jedirefine/TaskRefinerBase.py
+++ b/pandajedi/jedirefine/TaskRefinerBase.py
@@ -341,9 +341,6 @@ def extractCommon(
self.setSplitRule(taskParamMap, "maxJumboPerSite", JediTaskSpec.splitRuleToken["maxJumboPerSite"])
if "minCpuEfficiency" in taskParamMap:
self.setSplitRule(taskParamMap, "minCpuEfficiency", JediTaskSpec.splitRuleToken["minCpuEfficiency"])
- if "loadXML" in taskParamMap:
- self.setSplitRule(None, 3, JediTaskSpec.splitRuleToken["loadXML"])
- self.setSplitRule(None, 4, JediTaskSpec.splitRuleToken["groupBoundaryID"])
if "pfnList" in taskParamMap:
self.setSplitRule(None, 1, JediTaskSpec.splitRuleToken["pfnList"])
if "noWaitParent" in taskParamMap and taskParamMap["noWaitParent"] is True:
diff --git a/pandaserver/taskbuffer/JediTaskSpec.py b/pandaserver/taskbuffer/JediTaskSpec.py
index 2b4efb85f..5a8dc8fed 100644
--- a/pandaserver/taskbuffer/JediTaskSpec.py
+++ b/pandaserver/taskbuffer/JediTaskSpec.py
@@ -905,10 +905,6 @@ def setErrDiag(self, diag: str | None, append: bool | None = False, prepend: boo
else:
self.errorDialog = diag
- # use loadXML
- def useLoadXML(self) -> bool:
- return self.check_split_rule("loadXML")
-
# make VOMS FQANs
def makeFQANs(self) -> list[str]:
# no working group
diff --git a/pandaserver/taskbuffer/ParseJobXML.py b/pandaserver/taskbuffer/ParseJobXML.py
deleted file mode 100644
index 3241e119a..000000000
--- a/pandaserver/taskbuffer/ParseJobXML.py
+++ /dev/null
@@ -1,366 +0,0 @@
-#!/usr/bin/env python
-
-import sys
-import xml.dom.minidom
-from typing import TYPE_CHECKING, Any
-from urllib.parse import quote
-
-if TYPE_CHECKING:
- from pandaserver.taskbuffer.FileSpec import FileSpec
-
-
-class dom_job:
- """infiles[inds]=[file1,file2...]
- outfiles = [file1,file2...]
- command - script that will be executed on the grid
- prepend - list of (option,value) prepended to output file name
- forward - list of (option,value) forwarded to the grid job
- """
-
- def __init__(
- s,
- domjob: xml.dom.minidom.Element | None = None,
- primaryds: str | None = None,
- defaultcmd: str | None = None,
- defaultout: list[str] = [],
- ) -> None:
- """Loads from xml file.
- If primaryds is set, makes sure it is present in job spec"""
- s.infiles: dict[str, Any] = {}
- s.outfiles = []
- s.command = defaultcmd
- s.prepend = []
- s.forward = []
- if not domjob:
- return
- # script executed on the grid node for this job
- if len(domjob.getElementsByTagName("command")) > 0:
- s.command = dom_parser.text(domjob.getElementsByTagName("command")[0])
- # input files
- for inds in domjob.getElementsByTagName("inds"):
- name = dom_parser.text(inds.getElementsByTagName("name")[0])
- files = inds.getElementsByTagName("file")
- if len(files) == 0:
- continue
- s.infiles[name] = []
- for file in files:
- s.infiles[name].append(dom_parser.text(file))
- if primaryds and primaryds not in s.infiles.keys():
- print(f"ERROR: primaryds={primaryds} must be present in each job")
- sys.exit(0)
- # output files (also, drop duplicates within this job)
- outfiles = set(defaultout)
- outfiles.update(dom_parser.text(v) for v in domjob.getElementsByTagName("output"))
- s.outfiles = list(outfiles)
- # gearing options
- for o in domjob.getElementsByTagName("option"):
- name = o.attributes["name"].value
- value = dom_parser.text(o)
- prepend = dom_parser.true(o.attributes["prepend"].value)
- forward = dom_parser.true(o.attributes["forward"].value)
- if prepend:
- s.prepend.append((name, value))
- if forward:
- s.forward.append((name, value))
-
- def to_dom(s) -> xml.dom.minidom.Element:
- """Converts this job to a dom tree branch"""
- x = xml.dom.minidom.Document()
- job = x.createElement("job")
- # appendChild returns the element it appended, so each new node is named rather
- # than reached back through childNodes[-1]: a child list holds text and comments
- # as well as elements, so the last entry of one is not an element to a reader
- for inds in s.infiles.keys():
- inds_node = job.appendChild(x.createElement("inds"))
- name_node = inds_node.appendChild(x.createElement("name"))
- name_node.appendChild(x.createTextNode(inds))
- for file in s.infiles[inds]:
- file_node = inds_node.appendChild(x.createElement("file"))
- file_node.appendChild(x.createTextNode(file))
- for outfile in s.outfiles:
- output_node = job.appendChild(x.createElement("output"))
- output_node.appendChild(x.createTextNode(outfile))
- if s.command:
- command_node = job.appendChild(x.createElement("command"))
- command_node.appendChild(x.createTextNode(s.command))
- for option in s.prepend + list(set(s.prepend + s.forward) - set(s.prepend)):
- option_node = job.appendChild(x.createElement("option"))
- option_node.setAttribute("name", str(option[0]))
- option_node.setAttribute("forward", "true" if option in s.forward else "false")
- option_node.setAttribute("prepend", "true" if option in s.prepend else "false")
- option_node.appendChild(x.createTextNode(str(option[1])))
- return job
-
- def files_in_DS(s, DS: str) -> list[str]:
- """Returns a list of files used in a given job in a given dataset"""
- if DS in s.infiles:
- files: list[str] = s.infiles[DS]
- return files
- else:
- return []
-
- def forward_opts(s) -> str:
- """passable string of forward options"""
- return " ".join([f"{v[0]}={v[1]}" for v in s.forward])
-
- def prepend_string(s) -> str:
- """a tag string prepended to output files"""
- return "_".join([f"{v[0]}{v[1]}" for v in s.prepend])
-
- def exec_string(s) -> str:
- """exec string for prun.
- If user requested to run script run.sh (via run.sh), it will return
- opt1=value1 opt2=value2 opt3=value3 run.sh
- This way, all options will be set inside run.sh
- """
- return f"{s.forward_opts()} {s.command}"
-
- def exec_string_enc(s) -> str:
- """exec string for prun.
- If user requested to run script run.sh (via run.sh), it will return
- opt1=value1 opt2=value2 opt3=value3 run.sh
- This way, all options will be set inside run.sh
- """
- comStr = f"{s.forward_opts()} {s.command}"
- return quote(comStr)
-
- def get_outmap_str(s, outMap: dict[str, "FileSpec"]) -> str:
- """return mapping of original and new filenames"""
- newMap = {}
- for oldLFN, fileSpec in outMap.items():
- newMap[oldLFN] = str(fileSpec.lfn)
- return str(newMap)
-
- def outputs_list(s, prepend: bool = False) -> list[str]:
- """python list with finalized output file names"""
- if prepend and s.prepend_string():
- return [s.prepend_string() + "." + o for o in s.outfiles]
- else:
- return [o for o in s.outfiles]
-
- def outputs(s, prepend: bool = False) -> str:
- """Comma-separated list of output files accepted by prun"""
- return ",".join(s.outputs_list(prepend))
-
-
-class dom_parser:
- def __init__(s, fname: str | None = None, xmlStr: str | None = None) -> None:
- """creates a dom object out of a text file (if provided)"""
- s.fname = fname
- s.dom: xml.dom.minidom.Document | None = None
- s.title: str | None = None
- s.tag: str | None = None
- s.command: str | None = None
- s.outds: str | None = None
- # input dataset name -> the stream it is read as
- s.inds: dict[str, str] = {}
- s.global_outfiles: list[str] = []
- s.jobs: list["dom_job"] = []
- s.primaryds: str | None = None
- if fname:
- s.dom = xml.dom.minidom.parse(fname)
- s.parse()
- s.check()
- if xmlStr is not None:
- s.dom = xml.dom.minidom.parseString(xmlStr)
- s.parse()
- s.check()
-
- @staticmethod
- def true(v: str) -> bool:
- """define True"""
- return v in ("1", "true", "True", "TRUE", "yes", "Yes", "YES")
-
- @staticmethod
- def text(pnode: xml.dom.minidom.Node) -> str:
- """extracts the value stored in the node"""
- rc = []
- for node in pnode.childNodes:
- if node.nodeType == node.TEXT_NODE:
- rc.append(str(node.data).strip())
- return "".join(rc)
-
- def parse(s) -> None:
- """loads submission configuration from an xml file"""
- if s.dom is None:
- # __init__ parses the document before it calls this
- raise RuntimeError("parse() called before a document was loaded")
- try:
- # general settings
- if len(s.dom.getElementsByTagName("title")) > 0:
- s.title = dom_parser.text(s.dom.getElementsByTagName("title")[0])
- else:
- s.title = "Default title"
- if len(s.dom.getElementsByTagName("tag")) > 0:
- s.tag = dom_parser.text(s.dom.getElementsByTagName("tag")[0])
- else:
- s.tag = "default_tag"
- s.command = None # can be overridden in subjobs
- for elm in s.dom.getElementsByTagName("submission")[0].childNodes:
- if elm.nodeName != "command":
- continue
- s.command = dom_parser.text(elm)
- break
- s.global_outfiles = [] # subjobs can append *additional* outputs
- for elm in s.dom.getElementsByTagName("submission")[0].childNodes:
- if elm.nodeName != "output":
- continue
- s.global_outfiles.append(dom_parser.text(elm))
- s.outds = dom_parser.text(s.dom.getElementsByTagName("outds")[0])
- # declaration of all input datasets
- primarydss = []
- for elm in s.dom.getElementsByTagName("submission")[0].childNodes:
- # a child list holds text and comments too, and only an element has
- # attributes or elements of its own; the name check alone does not say so
- if not isinstance(elm, xml.dom.minidom.Element) or elm.nodeName != "inds":
- continue
- if "primary" in elm.attributes.keys() and dom_parser.true(elm.attributes["primary"].value):
- primary = True
- else:
- primary = False
- stream = dom_parser.text(elm.getElementsByTagName("stream")[0])
- name = dom_parser.text(elm.getElementsByTagName("name")[0])
- s.inds[name] = stream
- if primary:
- primarydss.append(name)
- # see if one of the input datasets was explicitly labeled as inDS
- if len(primarydss) == 1:
- s.primaryds = primarydss[0]
- else:
- s.primaryds = None
- for job in s.dom.getElementsByTagName("job"):
- s.jobs.append(dom_job(job, primaryds=s.primaryds, defaultcmd=s.command, defaultout=s.global_outfiles))
- except Exception:
- print(f"ERROR: failed to parse {s.fname}")
- raise
-
- def to_dom(s) -> xml.dom.minidom.Element:
- """Converts this submission to a dom tree branch"""
- x = xml.dom.minidom.Document()
- submission = x.createElement("submission")
- # named rather than reached back through childNodes[-1], for the reason given in
- # dom_job.to_dom above
- if s.title:
- title_node = submission.appendChild(x.createElement("title"))
- title_node.appendChild(x.createTextNode(s.title))
- if s.tag:
- tag_node = submission.appendChild(x.createElement("tag"))
- tag_node.appendChild(x.createTextNode(s.tag))
- for name, stream in s.inds.items():
- inds_node = submission.appendChild(x.createElement("inds"))
- inds_node.setAttribute("primary", "true" if name == s.primaryds else "false")
- stream_node = inds_node.appendChild(x.createElement("stream"))
- stream_node.appendChild(x.createTextNode(stream))
- name_node = inds_node.appendChild(x.createElement("name"))
- name_node.appendChild(x.createTextNode(name))
- if s.command:
- command_node = submission.appendChild(x.createElement("command"))
- command_node.appendChild(x.createTextNode(s.command))
- for outfile in s.global_outfiles:
- output_node = submission.appendChild(x.createElement("output"))
- output_node.appendChild(x.createTextNode(outfile))
- outds_node = submission.appendChild(x.createElement("outds"))
- # parse() always sets outds, from the document or from the default
- outds_node.appendChild(x.createTextNode(s.outds or ""))
- for job in s.jobs:
- submission.appendChild(job.to_dom())
- return submission
-
- def check(s) -> None:
- """checks that all output files have unique qualifiers"""
- quals = []
- for j in s.jobs:
- quals += j.outputs_list(True)
- if len(list(set(quals))) != len(quals):
- print("ERROR: found non-unique output file names across the jobs")
- print("(you likely need to review xml options with prepend=true)")
- sys.exit(0)
-
- def input_datasets(s) -> list[str]:
- """returns a list of all used input datasets"""
- DSs = set()
- for j in s.jobs:
- for ds in j.infiles.keys():
- DSs.add(ds)
- return list(DSs)
-
- def inDS(s) -> str:
- """chooses a dataset we'll call inDS; others will become secondaryDS"""
- # user manually labeled one of datasets as primary, so make it inDS:
- if s.primaryds:
- return s.primaryds
- # OR: choose inDS dataset randomly
- else:
- return s.input_datasets()[0]
-
- def secondaryDSs(s) -> list[str]:
- """returns all secondaryDSs. This excludes inDS, unless inDS is managed by prun"""
- return [d for d in s.input_datasets() if d != s.inDS()]
-
- def writeInputToTxt(s) -> str:
- """Prepares prun option --writeInputToTxt
- comma-separated list of STREAM:STREAM.files.dat
- """
- out = []
- DSs = s.secondaryDSs()
- for i, DS in enumerate(DSs):
- if DS in s.inds:
- stream = s.inds[DS]
- else:
- stream = "IN%d" % (i + 1,)
- out.append(f"{stream}:{stream}.files.dat")
- out.append("IN:IN.files.dat")
- return ",".join(out)
-
- def files_in_DS(s, DS: str, regex: bool = False) -> str | list[str]:
- """Returns a list of all files from a given dataset
- that will be used in at least one job in this submission
- If regex==True, the list is converted to a regex string
- """
- assert DS in s.input_datasets(), f"ERROR: dataset {DS} was not requested in the xml file"
- files = []
- for j in s.jobs:
- if DS in j.infiles.keys():
- files += j.infiles[DS]
- if regex:
- return "|".join(sorted(list(set(files))))
- else:
- return sorted(list(set(files)))
-
- def nJobs(s) -> int:
- return len(s.jobs)
-
- def dump(s, verbose: bool = True) -> None:
- """prints a summary of this submission"""
-
- def P(key: str, value: Any = "") -> None:
- if value == "":
- print(key)
- else:
- # str() because the counts below are numbers, which the concatenation this
- # replaces could not take
- print((key + ":").ljust(14) + " " + str(value))
-
- P("XML FILE LOADED", s.fname)
- P("Title", s.title)
- P("Command", s.command)
- P("InDS", s.inDS())
- P("Output DS", s.outds)
- P("njobs", s.nJobs())
- if verbose:
- for i, job in enumerate(s.jobs):
- P("===============> JOB%d" % i)
- P("command", job.exec_string())
- P("outfiles", job.outputs())
- P("INPUTS:")
- j = 0
- for dsname, files in job.infiles.items():
- P(" Dataset%d" % j, dsname)
- for k, fname in enumerate(files):
- P(" File%d" % k, fname)
- j += 1
-
-
-if __name__ == "__main__":
- p = dom_parser("./job.xml")
- p.dump()
diff --git a/pandaserver/taskbuffer/db_proxy_mods/task_complex_module.py b/pandaserver/taskbuffer/db_proxy_mods/task_complex_module.py
index 54b83580a..97c829088 100644
--- a/pandaserver/taskbuffer/db_proxy_mods/task_complex_module.py
+++ b/pandaserver/taskbuffer/db_proxy_mods/task_complex_module.py
@@ -18,7 +18,7 @@
from pandaserver.config import panda_config
from pandaserver.srvcore import CoreUtils
-from pandaserver.taskbuffer import EventServiceUtils, JobUtils, ParseJobXML
+from pandaserver.taskbuffer import EventServiceUtils, JobUtils
from pandaserver.taskbuffer.db_proxy_mods.base_module import BaseModule, varNUMBER
from pandaserver.taskbuffer.db_proxy_mods.job_complex_module import (
get_job_complex_module,
@@ -205,7 +205,6 @@ def insertFilesForDataset_JEDI(
nChunksForScout: Any,
includePatt: list[str],
excludePatt: list[str],
- xmlConfig: Any,
noWaitParent: bool,
parent_tid: int | None,
pid: str | None,
@@ -231,7 +230,7 @@ def insertFilesForDataset_JEDI(
tmpLog.debug(f"useFilesWithNewAttemptNr={useFilesWithNewAttemptNr} nFilesPerJob={nFilesPerJob} nEventsPerRange={nEventsPerRange}")
tmpLog.debug(f"useScout={useScout} nChunksForScout={nChunksForScout} userRealEventNumber={useRealNumEvents}")
tmpLog.debug(f"includePatt={str(includePatt)} excludePatt={str(excludePatt)}")
- tmpLog.debug(f"xmlConfig={type(xmlConfig)} noWaitParent={noWaitParent} parent_tid={parent_tid}")
+ tmpLog.debug(f"noWaitParent={noWaitParent} parent_tid={parent_tid}")
tmpLog.debug(f"len(fileMap)={len(fileMap)} pid={pid}")
tmpLog.debug(f"datasetState={datasetState} dataset.state={datasetSpec.state}")
tmpLog.debug(f"respectLB={respectLB} tgtNumEventsPerJob={tgtNumEventsPerJob} skipFilesUsedBy={skipFilesUsedBy} ramCount={ramCount}")
@@ -363,24 +362,12 @@ def insertFilesForDataset_JEDI(
if fileVal["lfn"] in givenFileMap:
newFileMap[guid] = fileVal
fileMap = newFileMap
- # XML config
- if xmlConfig is not None:
- try:
- xmlConfig = ParseJobXML.dom_parser(xmlStr=xmlConfig)
- except Exception as e:
- tmpErrStr = f"failed to load XML config with {type(e).__name__}:{e}"
- raise RuntimeError(tmpErrStr)
- newFileMap = {}
- for guid, fileVal in fileMap.items():
- if fileVal["lfn"] in xmlConfig.files_in_DS(datasetSpec.datasetName):
- newFileMap[guid] = fileVal
- fileMap = newFileMap
+
# make map with LFN as key
filelValMap = {}
for guid, fileVal in fileMap.items():
filelValMap[fileVal["lfn"]] = (guid, fileVal)
# make LFN list
- listBoundaryID = []
if order_by == "eventsAlignment" and nEventsPerJob:
aligned = []
unaligned = dict()
@@ -394,24 +381,10 @@ def insertFilesForDataset_JEDI(
# not be the name the sorted list is bound to
unaligned_lfns = sorted(unaligned, key=lambda i: unaligned[i], reverse=True)
lfnList = aligned + unaligned_lfns
- elif xmlConfig is None:
+ else:
# sort by LFN
lfnList = sorted(filelValMap.keys())
- else:
- # sort as described in XML
- tmpBoundaryID = 0
- lfnList = []
- for tmpJobXML in xmlConfig.jobs:
- for tmpLFN in tmpJobXML.files_in_DS(datasetSpec.datasetName):
- # check if the file is available
- if tmpLFN not in filelValMap:
- diagMap["errMsg"] = f"{tmpLFN} is not found in {datasetSpec.datasetName}"
- tmpLog.error(diagMap["errMsg"])
- return failedRet
- lfnList.append(tmpLFN)
- listBoundaryID.append(tmpBoundaryID)
- # increment boundaryID
- tmpBoundaryID += 1
+
# truncate if necessary
if datasetSpec.isSeqNumber():
offsetVal = 0
@@ -502,11 +475,8 @@ def insertFilesForDataset_JEDI(
if datasetSpec.toKeepTrack():
fileSpec.keepTrack = 1
tmpFileSpecList = []
- if xmlConfig is not None:
- # splitting with XML
- fileSpec.boundaryID = listBoundaryID[tmpIdx]
- tmpFileSpecList.append(fileSpec)
- elif (
+
+ if (
((nEventsPerJob is None or nEventsPerJob <= 0) and (tgtNumEventsPerJob is None or tgtNumEventsPerJob <= 0))
or fileSpec.nEvents is None
or fileSpec.nEvents <= 0
@@ -520,7 +490,7 @@ def insertFilesForDataset_JEDI(
else:
# event-level splitting
splitStartEvent = 0
- # the elif above ruled out an unset nEvents, and the copies made below
+ # the if above ruled out an unset nEvents, and the copies made below
# inherit it, so read it once here where that is still visible
numEventsInFile = fileSpec.nEvents
# change nEventsPerJob if target number is specified
@@ -1199,9 +1169,6 @@ def insertFilesForDataset_JEDI(
varMap[":nFilesTobeUsed"] = total_pending_files_to_activate + nUsed
elif datasetSpec.isMaster() and useScout and (set([taskStatus, taskSpec.oldStatus]) & set(["scouting", "ready", "assigning"])):
varMap[":nFilesTobeUsed"] = nFilesToUseDS
- elif xmlConfig is not None:
- # disable scout for --loadXML
- varMap[":nFilesTobeUsed"] = nReady + nUsed
elif (
(set([taskStatus, taskSpec.oldStatus]) & set(["defined", "ready", "scouting", "assigning"]))
and useScout
@@ -2899,7 +2866,7 @@ def _read_unprocessed_inputs(
max_files_to_read_for_this_dataset = n_master_files_to_read
else:
# for secondaries
- if task_spec.useLoadXML() or tmp_dataset_spec.isNoSplit() or tmp_dataset_spec.getEventRatio() is not None:
+ if tmp_dataset_spec.isNoSplit() or tmp_dataset_spec.getEventRatio() is not None:
max_files_to_read_for_this_dataset = max_secondary_files_to_read_with_event_ratio
elif tmp_dataset_spec.getNumFilesPerJob() is not None:
max_files_to_read_for_this_dataset = n_master_files_to_read * tmp_dataset_spec.getNumFilesPerJob()
@@ -2922,10 +2889,9 @@ def _read_unprocessed_inputs(
order_by_policy = "fileID"
elif task_spec.respectLumiblock() or task_spec.orderByLB():
order_by_policy = "lumiBlockNr,lfn"
- elif not task_spec.useLoadXML():
- order_by_policy = "fileID"
else:
- order_by_policy = "boundaryID"
+ order_by_policy = "fileID"
+
# repeat file reading to duplicate secondly file records if necessary
tmp_num_already_read_files = 0
tmp_num_waiting_files = 0
@@ -3084,13 +3050,7 @@ def _read_unprocessed_inputs(
tmp_num_waiting_files += 1
# escape the duplication loop since it is not requested
- if (
- tmp_dataset_spec.isMaster()
- or task_spec.useLoadXML()
- or tmp_dataset_spec.isNoSplit()
- or tmp_dataset_spec.toMerge()
- or input_chunk.ramCount not in (None, 0)
- ):
+ if tmp_dataset_spec.isMaster() or tmp_dataset_spec.isNoSplit() or tmp_dataset_spec.toMerge() or input_chunk.ramCount not in (None, 0):
break
# escape since enough files were read
if tmp_num_already_read_files >= num_files_to_read_for_the_chunk and tmp_dataset_spec.getEventRatio() is None:
@@ -6044,7 +6004,6 @@ def getOutputFiles_JEDI(
instantiatedSites: str | None,
isUnMerging: bool,
isPrePro: bool,
- xmlConfigJob: ParseJobXML.dom_job | None,
siteDsMap: dict[Any, Any] | None,
middleName: str,
registerDatasets: bool,
@@ -6066,7 +6025,7 @@ def getOutputFiles_JEDI(
else:
tmpLog = self.create_tagged_logger(comment, f"jediTaskID={jediTaskID}")
tmpLog.debug(f"start with simul={simul} instantiateTmpl={instantiateTmpl} instantiatedSites={instantiatedSites}")
- tmpLog.debug(f"isUnMerging={isUnMerging} isPrePro={isPrePro} provenanceID={provenanceID} xmlConfigJob={type(xmlConfigJob)}")
+ tmpLog.debug(f"isUnMerging={isUnMerging} isPrePro={isPrePro} provenanceID={provenanceID}")
tmpLog.debug(f"middleName={middleName} registerDatasets={registerDatasets} idPool={len(fileIDPool)}")
tmpLog.debug(f"n_files_per_chunk={n_files_per_chunk} bulk_fetch={bulk_fetch_for_multiple_jobs}")
try:
@@ -6258,18 +6217,8 @@ def getOutputFiles_JEDI(
for resR in resTmpList:
# make FileSpec
outTempID, datasetID, fileNameTemplate, serialNr, outType, streamName = resR
- if xmlConfigJob is None or outType.endswith("log"):
- fileNameTemplateList = [(fileNameTemplate, streamName)]
- else:
- fileNameTemplateList = []
- # get output filenames from XML config
- for tmpFileName in xmlConfigJob.outputs().split(","):
- # ignore empty
- if tmpFileName == "":
- continue
- newStreamName = tmpFileName
- newFileNameTemplate = fileNameTemplate + "." + xmlConfigJob.prepend_string() + "." + newStreamName
- fileNameTemplateList.append((newFileNameTemplate, newStreamName))
+ fileNameTemplateList = [(fileNameTemplate, streamName)]
+
if bulk_fetch_for_multiple_jobs:
nFileLoop = n_files_per_chunk
else:
diff --git a/pandaserver/taskbuffer/task_split_rules.py b/pandaserver/taskbuffer/task_split_rules.py
index 36c1bf2a5..5dea882f1 100644
--- a/pandaserver/taskbuffer/task_split_rules.py
+++ b/pandaserver/taskbuffer/task_split_rules.py
@@ -40,7 +40,6 @@
"noLoopingCheck": "LC",
"useLocalIO": "LI",
"limitedSites": "LS",
- "loadXML": "LX",
"minCpuEfficiency": "MC",
"messageDriven": "MD",
"mergeEsOnOS": "ME",