pandaserver/taskbuffer/task_split_rules.py

0001 # split rules
0002 split_rule_dict = {
0003     "allowEmptyInput": "AE",
0004     "allowIncompleteInDS": "AI",
0005     "addNthFieldToLFN": "AN",
0006     "allowPartialFinish": "AP",
0007     "altStageOut": "AT",
0008     "avoidVP": "AV",
0009     "maxCoreCount": "CC",
0010     "cloudAsVO": "CV",
0011     "ddmBackEnd": "DE",
0012     "disableAutoFinish": "DF",
0013     "disableReassign": "DI",
0014     "debugMode": "DM",
0015     "disableAutoRetry": "DR",
0016     "dynamicNumEvents": "DY",
0017     "nEsConsumers": "EC",
0018     "nEventsPerInput": "EI",
0019     "encJobParams": "EJ",
0020     "nEventsPerWorker": "ES",
0021     "firstContentsFeed": "FC",
0022     "failGoalUnreached": "FG",
0023     "fineGrainedProc": "FP",
0024     "firstEvent": "FT",
0025     "fullChain": "FU",
0026     "groupBoundaryID": "GB",
0027     "hpoWorkflow": "HO",
0028     "instantiateTmplSite": "IA",
0029     "inFilePosEvtNum": "IF",
0030     "ipStack": "IK",
0031     "allowInputLAN": "IL",
0032     "ignoreMissingInDS": "IM",
0033     "intermediateTask": "IN",
0034     "ipConnectivity": "IP",
0035     "inputPreStaging": "IS",
0036     "instantiateTmpl": "IT",
0037     "allowInputWAN": "IW",
0038     "noLoopingCheck": "LC",
0039     "useLocalIO": "LI",
0040     "limitedSites": "LS",
0041     "loadXML": "LX",
0042     "minCpuEfficiency": "MC",
0043     "messageDriven": "MD",
0044     "mergeEsOnOS": "ME",
0045     "nMaxFilesPerJob": "MF",
0046     "maxJumboPerSite": "MJ",
0047     "maxNumJobs": "MN",
0048     "mergeOutput": "MO",
0049     "multiStepExec": "MS",
0050     "maxWalltime": "MW",
0051     "maxEventsPerJob": "MX",
0052     "noExecStrCnv": "NC",
0053     "notDiscardEvents": "ND",
0054     "nEventsPerJob": "NE",
0055     "nFilesPerJob": "NF",
0056     "nGBPerJob": "NG",
0057     "noInputPooling": "NI",
0058     "nJumboJobs": "NJ",
0059     "noAutoPause": "NP",
0060     "nSitesPerJob": "NS",
0061     "nChunksToWait": "NT",
0062     "noWaitParent": "NW",
0063     "orderInputBy": "OI",
0064     "orderByLB": "OL",
0065     "onSiteMerging": "OM",
0066     "osMatching": "OS",
0067     "onlyTagsForFC": "OT",
0068     "pushStatusChanges": "PC",
0069     "pushJob": "PJ",
0070     "pfnList": "PL",
0071     "putLogToOS": "PO",
0072     "runUntilClosed": "RC",
0073     "registerDatasets": "RD",
0074     "registerEsFiles": "RE",
0075     "respectLB": "RL",
0076     "retryModuleRules": "RM",
0077     "reuseSecOnDemand": "RO",
0078     "releasePerLB": "RP",
0079     "respectSplitRule": "RR",
0080     "randomSeed": "RS",
0081     "retryRamOffset": "RX",
0082     "retryRamStep": "RY",
0083     "retryRamMax": "RZ",
0084     "resurrectConsumers": "SC",
0085     "switchEStoNormal": "SE",
0086     "stayOutputOnSite": "SO",
0087     "scoutSuccessRate": "SS",
0088     "useSecrets": "ST",
0089     "segmentedWork": "SW",
0090     "totNumJobs": "TJ",
0091     "tgtMaxOutputForNG": "TN",
0092     "t1Weight": "TW",
0093     "useBuild": "UB",
0094     "useJobCloning": "UC",
0095     "useRealNumEvents": "UE",
0096     "useFileAsSourceLFN": "UF",
0097     "usePrePro": "UP",
0098     "useScout": "US",
0099     "usePrefetcher": "UT",
0100     "useExhausted": "UX",
0101     "useZipToPin": "UZ",
0102     "writeInputToFile": "WF",
0103     "workflowHoldup": "WH",
0104     "waitInput": "WI",
0105     "maxAttemptES": "XA",
0106     "decAttOnFailedES": "XF",
0107     "maxAttemptEsJob": "XJ",
0108     "nEventsPerMergeJob": "ZE",
0109     "nFilesPerMergeJob": "ZF",
0110     "nGBPerMergeJob": "ZG",
0111     "nMaxFilesPerMergeJob": "ZM",
0112 }
0113
0114 # value→label mappings for split rule tokens with enumerated values
0115 # dict-form enums (single source of truth; referenced back by JediTaskSpec)
0116 enum_limitedSites = {"1": "with_allowlist", "2": "with_denylist", "3": "with_allow_and_denylist"}
0117 enum_ipConnectivity = {"1": "full", "2": "http"}
0118 enum_ipStack = {"4": "IPv4", "6": "IPv6"}
0119 enum_altStageOut = {"1": "on", "2": "off", "3": "force"}
0120 enum_inputLAN = {"1": "use", "2": "only"}
0121
0122 # decode dicts for tokens whose JediTaskSpec counterparts are scalar constants or class-based enums
0123 enum_useScout = {"1": "no_use", "2": "will_update_requirements", "3": "updated_requirements"}
0124 enum_usePrePro = {"1": "toPreProcess", "2": "preProcessed", "3": "postPProcess"}
0125 enum_registerDatasets = {"1": "registering", "2": "registered"}
0126 enum_inputPreStaging = {"0": "notUse", "1": "use"}  # inverted vs JediTaskSpec's label→value dict
0127 enum_firstContentsFeed = {"0": "False", "1": "True"}
0128 enum_fullChain = {"1": "only", "2": "require", "3": "capable"}
0129 enum_orderInputBy = {"1": "eventsAlignment"}
0130
0131 # maps 2-letter split rule tags to their value→label dicts (used by decode_split_rule)
0132 split_rule_value_dict = {
0133     "US": enum_useScout,
0134     "LS": enum_limitedSites,
0135     "IP": enum_ipConnectivity,
0136     "IK": enum_ipStack,
0137     "AT": enum_altStageOut,
0138     "IL": enum_inputLAN,
0139     "UP": enum_usePrePro,
0140     "RD": enum_registerDatasets,
0141     "IS": enum_inputPreStaging,
0142     "FC": enum_firstContentsFeed,
0143     "FU": enum_fullChain,
0144     "OI": enum_orderInputBy,
0145 }
0146
0147 # changeable split rules
0148 changeable_split_rule_names = [
0149     "allowIncompleteInDS",
0150     "t1Weight",
0151     "nEsConsumers",
0152     "nMaxFilesPerJob",
0153     "nGBPerJob",
0154     "noInputPooling",
0155     "nFilesPerJob",
0156     "nEventsPerWorker",
0157     "nJumboJobs",
0158     "avoidVP",
0159     "allowInputLAN",
0160     "useLocalIO",
0161     "noLoopingCheck",
0162     "maxCoreCount",
0163     "onlyTagsForFC",
0164     "useZipToPin",
0165     "ignoreMissingInDS",
0166     "noAutoPause",
0167 ]
0168
0169 changeable_split_rule_tags = [split_rule_dict[tmp_name] for tmp_name in changeable_split_rule_names]
0170
0171
0172 # extract rules
0173 def extract_rule_values(split_rules: str, rule_names: list, is_sub_rule: bool = False) -> dict:
0174     """
0175     Extract rule values from split rule string
0176
0177     :param split_rules: a comma separated string
0178     :param rule_names: a list of rule names
0179     :param is_sub_rule: True to indicate the rule is a subrule
0180     :return: dict of rule names and values
0181     """
0182     if split_rules is None:
0183         split_rules = ""
0184     ret = {}
0185     if is_sub_rule:
0186         rule_separator = "|"
0187         key_value_separator = ":"
0188     else:
0189         rule_separator = ","
0190         key_value_separator = "="
0191     for tmp_rule in split_rules.split(rule_separator):
0192         for tmp_name in rule_names:
0193             if tmp_name in split_rule_dict and tmp_rule.startswith(split_rule_dict[tmp_name] + key_value_separator):
0194                 ret[tmp_name] = tmp_rule.split(key_value_separator)[-1]
0195                 break
0196     for tmp_name in rule_names:
0197         if tmp_name not in ret:
0198             ret[tmp_name] = None
0199     return ret
0200
0201
0202 # replace a rule
0203 def replace_rule(split_rules: str, rule_name: str, rule_value: int | str, is_sub_rule=False):
0204     """
0205     Replace a rule in the split rule string
0206
0207     :param split_rules: s comma separated string
0208     :param rule_name: rule name
0209     :param rule_value: rule value
0210     :param is_sub_rule: True to indicate the rule is a subrule
0211     :return: string of split rules
0212     """
0213     if rule_name not in split_rule_dict:
0214         return split_rules
0215     if split_rules is None:
0216         split_rules = ""
0217     if is_sub_rule:
0218         rule_separator = "|"
0219         key_value_separator = ":"
0220     else:
0221         rule_separator = ","
0222         key_value_separator = "="
0223     tmp_str = ""
0224     for tmp_rule in split_rules.split(rule_separator):
0225         if tmp_rule.startswith(split_rule_dict[rule_name] + key_value_separator):
0226             continue
0227         if tmp_str != "":
0228             tmp_str += rule_separator
0229         tmp_str += tmp_rule
0230     if tmp_str != "":
0231         tmp_str += rule_separator
0232     tmp_str += split_rule_dict[rule_name] + key_value_separator + str(rule_value)
0233     return tmp_str
0234
0235
0236 # remove a rule
0237 def remove_rule(split_rules: str, rule_token: str, is_sub_rule: bool = False):
0238     """
0239     Remove a rule from the split rule string
0240
0241     :param split_rules: s comma separated string
0242     :param rule_token: rule token
0243     :param is_sub_rule: TRue to indicate the rule is a subrule
0244     :return: string of split rules
0245     """
0246     if split_rules is None:
0247         split_rules = ""
0248     if is_sub_rule:
0249         rule_separator = "|"
0250         key_value_separator = ":"
0251     else:
0252         rule_separator = ","
0253         key_value_separator = "="
0254     tmp_str = ""
0255     for tmp_rule in split_rules.split(rule_separator):
0256         if tmp_rule.startswith(rule_token + key_value_separator):
0257             continue
0258         if tmp_str != "":
0259             tmp_str += rule_separator
0260         tmp_str += tmp_rule
0261     return tmp_str
0262
0263
0264 # decode split rule string: replace two-letter tags with human-readable names
0265 def decode_split_rule(split_rules: str) -> str:
0266     """
0267     Convert a compact split-rule string into a human-readable one by replacing
0268     two-letter tags with their full names from split_rule_dict.
0269
0270     The input is a comma-separated string of ``TAG=value`` pairs, e.g.::
0271
0272         "AE=1,AN=1,2,3,AI=2"
0273
0274     Special case: ``AN`` (addNthFieldToLFN) may carry a comma-separated list
0275     as its value (e.g. ``AN=1,2,3``).  All other tags treat ``=`` as the sole
0276     separator and never embed commas in their values.
0277
0278     :param split_rules: compact comma-separated split-rule string, or None/empty
0279     :return: human-readable comma-separated string, e.g.
0280              ``"allowEmptyInput=1,addNthFieldToLFN=1,2,3,allowIncompleteInDS=2"``
0281              Unknown tags are kept as-is.
0282     """
0283     if not split_rules:
0284         return split_rules or ""
0285
0286     # build reverse map: two-letter tag -> full name
0287     tag_to_name = {tag: name for name, tag in split_rule_dict.items()}
0288
0289     # The AN tag is the only one whose value may contain commas.
0290     an_tag = split_rule_dict["addNthFieldToLFN"]  # "AN"
0291
0292     # Split on commas, then re-group tokens into (tag, value) pairs.
0293     # A token starts a new rule when it matches "XX=..." where XX is a known
0294     # two-letter tag.  Otherwise, it is appended to the current rule's value.
0295     tokens = split_rules.split(",")
0296     rules = []  # list of [tag, value_parts...]
0297     for token in tokens:
0298         # Check whether this token begins a new TAG=value entry
0299         eq_pos = token.find("=")
0300         if eq_pos == 2:
0301             candidate_tag = token[:2]
0302             if candidate_tag in tag_to_name or candidate_tag == an_tag:
0303                 # Start a new rule
0304                 rules.append([candidate_tag, token[3:]])  # token[3:] = value after "XX="
0305                 continue
0306         # Not a recognized new rule:
0307         # - if there are existing rules and the last one is AN, treat as value continuation
0308         # - otherwise keep the token verbatim as its own entry
0309         if rules and rules[-1][0] == an_tag:
0310             rules[-1].append(token)
0311         else:
0312             # unknown / non-standard token — keep as-is using a sentinel
0313             rules.append([None, token])
0314
0315     # Reconstruct using full names; rejoin multipart AN values with commas
0316     result_parts = []
0317     for parts in rules:
0318         tag = parts[0]
0319         if tag is None:
0320             # unknown / non-standard token kept verbatim
0321             result_parts.append(parts[1])
0322         else:
0323             value = "_".join(parts[1:])
0324             name = tag_to_name.get(tag, tag)  # fall back to raw tag if unknown
0325             label = split_rule_value_dict.get(tag, {}).get(value, value)
0326             result_parts.append(f"{name}={label}")
0327
0328     return ",".join(result_parts)
0329
0330
0331 # remove a rule with name
0332 def remove_rule_with_name(split_rules: str, rule_name: str, is_sub_rule: bool = False) -> str:
0333     """
0334     Remove a rule from the split rule string using the rule name
0335
0336     :param split_rules: a comma separated string
0337     :param rule_name: rule name
0338     :param is_sub_rule: True to indicate the rule is a subrule
0339     :return: string of split rules
0340     """
0341     split_rules = remove_rule(split_rules, split_rule_dict[rule_name], is_sub_rule)
0342     return split_rules