pandaserver/taskbuffer/ProcessGroups.py

0001 from pandaserver.taskbuffer import JobUtils
0002
0003 processGroups = [
0004     ("others", []),
0005     ("evgen", ["evgen"]),
0006     ("simul", ["simul"]),
0007     ("reprocessing", ["reprocessing"]),
0008     ("test", ["prod_test", "validation"] + JobUtils.list_ptest_prod_sources),
0009     ("mcore", ["mcore"]),
0010     ("group", ["group"]),
0011     ("deriv", ["deriv"]),
0012     ("pile", ["pile"]),
0013     ("merge", ["merge"]),
0014 ]
0015
0016 # ('evgensimul',   ['evgen','simul']),
0017
0018 # maximum number of debug jobs per user
0019 maxDebugJobs = 3
0020
0021 # maximum number of debug jobs for prod role
0022 maxDebugProdJobs = 30
0023
0024 # maximum number of debug jobs for working group
0025 maxDebugWgJobs = 10
0026
0027 # extension level for GP
0028 extensionLevel_1 = 1
0029
0030
0031 # get corresponding group
0032 def getProcessGroup(valGroup):
0033     tmpGroup = None
0034     for tmpKey, tmpList in processGroups:
0035         # set default
0036         if tmpGroup is None:
0037             tmpGroup = tmpKey
0038             continue
0039         if valGroup in tmpList:
0040             tmpGroup = tmpKey
0041             break
0042     # return
0043     return tmpGroup
0044
0045
0046 # convert cloud and processingType for extended PG
0047 def converCPTforEPG(cloud, processingType, coreCount, workingGroup=None):
0048     if coreCount in [0, 1, None]:
0049         # use group queue for GP jobs
0050         if workingGroup is not None and workingGroup.startswith("GP_"):
0051             return cloud, "group"
0052         return cloud, processingType
0053     else:
0054         # use MCORE queue for MPC jobs in all clouds
0055         return cloud, "mcore"
0056
0057
0058 # count the number of jobs per group
0059 def countJobsPerGroup(valMap):
0060     ret = {}
0061     # loop over all clouds
0062     for cloud in valMap:
0063         cloudVal = valMap[cloud]
0064         # add cloud
0065         ret.setdefault(cloud, {})
0066         # loop over all sites
0067         for site in cloudVal:
0068             siteVal = cloudVal[site]
0069             # add site
0070             ret[cloud].setdefault(site, {})
0071             # loop over all types
0072             for pType in siteVal:
0073                 typeVal = siteVal[pType]
0074                 # get process group
0075                 tmpGroup = getProcessGroup(pType)
0076                 # add group
0077                 ret[cloud][site].setdefault(tmpGroup, {})
0078                 # loop over all status
0079                 for jobStatus in typeVal:
0080                     statVal = typeVal[jobStatus]
0081                     ret[cloud][site][tmpGroup].setdefault(jobStatus, 0)
0082                     # add
0083                     ret[cloud][site][tmpGroup][jobStatus] += statVal
0084     # return
0085     return ret
0086
0087
0088 # count the number of jobs per group for analysis
0089 def countJobsPerGroupForAnal(valMap):
0090     ret = {}
0091     # loop over all sites
0092     for site in valMap:
0093         siteVal = valMap[site]
0094         # add site
0095         ret.setdefault(site, {})
0096         # loop over all types
0097         for pType in siteVal:
0098             typeVal = siteVal[pType]
0099             # get process group
0100             tmpGroup = getProcessGroup(pType)
0101             # add group
0102             if tmpGroup not in ret[site]:
0103                 ret[site][tmpGroup] = {}
0104             # loop over all status
0105             for jobStatus in typeVal:
0106                 statVal = typeVal[jobStatus]
0107                 ret[site][tmpGroup].setdefault(jobStatus, 0)
0108                 # add
0109                 ret[site][tmpGroup][jobStatus] += statVal
0110     # return
0111     return ret