<commit_before>import re
import sys
import traceback
import time
import fcntl
import random
import datetime
import uuid
from pandaserver.brokerage import ErrorCode
from pandaserver.taskbuffer import ProcessGroups
from pandaserver.dataservice import DataServiceUtils
from pandaserver.dataservice.DataServiceUtils import select_scope
from pandaserver.dataservice.DDM import rucioAPI
from pandaserver.config import panda_config

from pandacommon.pandalogger.PandaLogger import PandaLogger
_log = PandaLogger().getLogger('broker')

try:
    long
except NameError:
    long = int

# all known sites
_allSites = []


# non LRC checking
_disableLRCcheck = []

# lock for uuidgen
_lockGetUU   = open(panda_config.lockfile_getUU, 'w')

# short-long mapping
shortLongMap = {'ANALY_BNL_SHORT': 'ANALY_BNL_LONG'}

# processingType to skip brokerage
skipBrokerageProTypes = ['prod_test']

# comparison function for sort
def _compFunc(jobA,jobB):
    # append site if not in list
    if not jobA.computingSite in _allSites:
        _allSites.append(jobA.computingSite)
    if not jobB.computingSite in _allSites:
        _allSites.append(jobB.computingSite)
    # compare
    indexA = _allSites.index(jobA.computingSite) 
    indexB = _allSites.index(jobB.computingSite) 
    if indexA > indexB:
        return 1
    elif indexA < indexB:
        return -1
    else:
        return 0


# release checker
def _checkRelease(jobRels,siteRels):
    # all on/off
    if "True" in siteRels:
        return True
    if "False" in siteRels:
        return False
    # loop over all releases
    for tmpRel in jobRels.split('\n'):
        relVer = re.sub('^Atlas-','',tmpRel)
        # not available releases
        if not relVer in siteRels:
            return False
    return True


# get list of files which already exist at the site
def _getOkFiles(v_ce,v_files,v_guids,allLFNs,allGUIDs,allOkFilesMap,prodsourcelabel,tmpLog=None,
                scopeList=None,allScopeList=None):

    scope_association_input, scope_association_output  = select_scope(v_ce, prodsourcelabel)
    dq2IDs = v_ce.setokens_input[scope_association_input].values()
    try:
        dq2IDs.remove('')
    except Exception:
        pass
    dq2IDs.sort()
    if dq2IDs == []:
        dq2ID = v_ce.ddm_input[scope_association_input]
    else:
        dq2ID = ''
        for tmpID in dq2IDs:
            dq2ID += '%s,' % tmpID
        dq2ID = dq2ID[:-1]    
    # set LFC and SE name 
    dq2URL = 'rucio://atlas-rucio.cern.ch:/grid/atlas'
    tmpSE = v_ce.ddm_endpoints_input[scope_association_input].getAllEndPoints()
    if tmpLog is not None:
        tmpLog.debug('getOkFiles for %s with dq2ID:%s,LFC:%s,SE:%s' % (v_ce.sitename,dq2ID,dq2URL,str(tmpSE)))
    anyID = 'any'
    # use bulk lookup
    if allLFNs != []:
        # get all replicas 
        if not dq2URL in allOkFilesMap:
            allOkFilesMap[dq2URL] = {}
            tmpStat,tmpAvaFiles = rucioAPI.listFileReplicas(allScopeList,allLFNs,tmpSE)
            if not tmpStat and tmpLog is not None:
                tmpLog.debug('getOkFile failed to get file replicas')
                tmpAvaFiles = {}
            allOkFilesMap[dq2URL][anyID] = tmpAvaFiles
        # get files for each dq2ID
        if not dq2ID in allOkFilesMap[dq2URL]:
            allOkFilesMap[dq2URL][dq2ID] = allOkFilesMap[dq2URL][anyID]
        # make return map
        retMap = {}
        for tmpLFN in v_files:
            if tmpLFN in allOkFilesMap[dq2URL][dq2ID]:
                retMap[tmpLFN] = allOkFilesMap[dq2URL][dq2ID][tmpLFN]
        tmpLog.debug('getOkFiles done')
        # return
        return retMap
    else:
        # old style
        tmpLog.debug('getOkFiles old')
        return {}


# check reprocessing or not
def _isReproJob(tmpJob):
    if tmpJob is not None:
        if tmpJob.processingType in ['reprocessing']:
            return True
        if tmpJob.transformation in ['csc_cosmics_trf.py','csc_BSreco_trf.py','BStoESDAODDPD_trf.py']:
            return True
    return False

    
# set 'ready' if files are already there
def _setReadyToFiles(tmpJob, okFiles, siteMapper, tmpLog):
    tmpLog.debug(str(okFiles))
    allOK = True
    tmpSiteSpec = siteMapper.getSite(tmpJob.computingSite)
    tmpSrcSpec  = siteMapper.getSite(siteMapper.getCloud(tmpJob.getCloud())['source'])
    scope_association_site_input, scope_association_site_output = select_scope(tmpSiteSpec, tmpJob.prodSourceLabel)
    scope_association_src_input, scope_association_src_output = select_scope(tmpSrcSpec, tmpJob.prodSourceLabel)
    tmpTapeEndPoints = tmpSiteSpec.ddm_endpoints_input[scope_association_site_input].getTapeEndPoints()
    # direct usage of remote SE
    if tmpSiteSpec.ddm_input[scope_association_site_input] != tmpSrcSpec.ddm_input[scope_association_src_input] \
            and tmpSrcSpec.ddm_input[scope_association_src_input] in tmpSiteSpec.setokens_input[scope_association_site_input].values():
        tmpSiteSpec = tmpSrcSpec
        tmpLog.debug('%s uses remote SiteSpec of %s for %s' % (tmpJob.PandaID,tmpSrcSpec.sitename,tmpJob.computingSite))
    for tmpFile in tmpJob.Files:
        if tmpFile.type == 'input':
            if tmpFile.status == 'ready':
                tmpFile.dispatchDBlock = 'NULL'
            elif DataServiceUtils.isCachedFile(tmpFile.dataset,tmpSiteSpec):
                # cached file
                tmpFile.status = 'cached'
                tmpFile.dispatchDBlock = 'NULL'
            elif tmpJob.computingSite == siteMapper.getCloud(tmpJob.getCloud())['source'] or \
                    tmpSiteSpec.ddm_input[scope_association_site_input] == tmpSrcSpec.ddm_input[scope_association_src_input]:
                # use DDM prestage only for on-tape files
                if len(tmpTapeEndPoints) > 0 and tmpFile.lfn in okFiles:
                    tapeOnly = True
                    tapeCopy = False
                    for tmpSE in okFiles[tmpFile.lfn]:
                        if tmpSE not in tmpTapeEndPoints:
                            tapeOnly = False
                        else:
                            # there is a tape copy
                            tapeCopy = True
                    # trigger prestage when disk copy doesn't exist or token is TAPE
                    if tapeOnly or (tapeCopy and tmpFile.dispatchDBlockToken in ['<PASSWORD>','<PASSWORD>']):
                        allOK = False
                    else:
                        # set ready                        
                        tmpFile.status = 'ready'
                        tmpFile.dispatchDBlock = 'NULL'
                else:
                    # set ready anyway even if LFC is down. i.e. okFiles doesn't contain the file
                    tmpFile.status = 'ready'
                    tmpFile.dispatchDBlock = 'NULL'                                
            else:
                # set ready if the file exists and the site doesn't use prestage
                tmpFile.status = 'ready'
                tmpFile.dispatchDBlock = 'NULL'
    # unset disp dataset
    if allOK:
        tmpJob.dispatchDBlock = 'NULL'
        
    

# check number/size of inputs
def _isTooManyInput(nFilesPerJob,inputSizePerJob):
    # the number of inputs is larger than 5 or
    # size of inputs is larger than 500MB
    if nFilesPerJob > 5 or inputSizePerJob > 500*1024*1024:
        return True
    return False


# send analysis brokerage info to logger
def sendMsgToLogger(message):
    _log.debug(message)


# send analysis brokerage info to logger with HTTP
def sendMsgToLoggerHTTP(msgList,job):
    try:
        # logging
        iMsg = 0
        # message type
        msgType = 'analy_brokerage'
        # make header
        if not job.jobsetID in [None,'NULL']:
            msgHead = "dn='%s' : jobset=%s jobdef=%s" % (job.prodUserName,job.jobsetID,job.jobDefinitionID)
        else:
            msgHead = "dn='%s' : jobdef=%s" % (job.prodUserName,job.jobDefinitionID)
        for msgBody in msgList:
            # make message
            message = msgHead + ' : ' + msgBody
            # dump locally
            _log.debug(message)
            # get logger
            _pandaLogger = PandaLogger()            
            _pandaLogger.lock()
            _pandaLogger.setParams({'Type':msgType})
            logger = _pandaLogger.getHttpLogger(panda_config.loggername)
            # add message
            logger.info(message)
            # release HTTP handler
            _pandaLogger.release()
            # sleep
            iMsg += 1
            if iMsg % 5 == 0:
                time.sleep(1)
    except Exception:
        errType,errValue = sys.exc_info()[:2]
        _log.error("sendMsgToLoggerHTTP : %s %s" % (errType,errValue))
    

# get T2 candidates when files are missing at T2
def getT2CandList(tmpJob,siteMapper,t2FilesMap):
    if tmpJob is None:
        return []
    # no cloud info
    if tmpJob.getCloud() not in t2FilesMap:
        return []
    # loop over all files
    tmpCandT2s = None
    for tmpFile in tmpJob.Files:
        if tmpFile.type == 'input' and tmpFile.status == 'missing':
            # no dataset info
            if tmpFile.dataset not in t2FilesMap[tmpJob.getCloud()]:
                return []
            # initial candidates
            if tmpCandT2s is None:
                tmpCandT2s = t2FilesMap[tmpJob.getCloud()][tmpFile.dataset]['sites']
            # check all candidates
            newCandT2s = []
            for tmpCandT2 in tmpCandT2s:
                # site doesn't have the dataset
                if tmpCandT2 not in t2FilesMap[tmpJob.getCloud()][tmpFile.dataset]['sites']:
                    continue
                # site has the file
                if tmpFile.lfn in t2FilesMap[tmpJob.getCloud()][tmpFile.dataset]['sites'][tmpCandT2]:
                    if not tmpCandT2 in newCandT2s:
                        newCandT2s.append(tmpCandT2)
            # set new candidates
            tmpCandT2s = newCandT2s
            if tmpCandT2s == []:
                break
    # return [] if no missing files         
    if tmpCandT2s is None:
        return []
    # return
    tmpCandT2s.sort() 
    return tmpCandT2s 


# get hospital queues
def getHospitalQueues(siteMapper, forAnalysis):
    retMap = {}
    # hospital words
    goodWordList = ['CORE$','VL$','MEM$','MP\d+$','LONG$','_HIMEM','_\d+$']

    # the prodSourceLabel is needed to know the def-vs-anal scope of the PQ-RSE association
    prodSourceLabel = 'managed'
    if forAnalysis:
        prodSourceLabel = 'user'

    # loop over all clouds
    for tmpCloudName in siteMapper.getCloudList():
        # get cloud
        tmpCloudSpec = siteMapper.getCloud(tmpCloudName)
        # get T1
        tmpT1Name = tmpCloudSpec['source']
        tmpT1Spec = siteMapper.getSite(tmpT1Name)
        scope_association_t1_input, scope_association_t1_output = select_scope(tmpT1Spec, prodSourceLabel)
        # skip if DDM is undefined
        if not tmpT1Spec.ddm_input[scope_association_t1_input]:
            continue
        # loop over all sites
        for tmpSiteName in tmpCloudSpec['sites']:
            # skip T1 defined in cloudconfig
            if tmpSiteName == tmpT1Name:
                continue
            # check hospital words
            checkHospWord = False
            for tmpGoodWord in goodWordList:
                if re.search(tmpGoodWord,tmpSiteName) is not None:
                    checkHospWord = True
                    break
            if not checkHospWord:
                continue
            # check site
            if not siteMapper.checkSite(tmpSiteName):
                continue
            tmpSiteSpec = siteMapper.getSite(tmpSiteName)
            scope_association_site_input, scope_association_site_output = select_scope(tmpSiteSpec, prodSourceLabel)
            # check DDM
            if tmpT1Spec.ddm_input[scope_association_t1_input] == tmpSiteSpec.ddm_input[scope_association_site_input]:
                # append
                if tmpCloudName not in retMap:
                    retMap[tmpCloudName] = []
                if not tmpSiteName in retMap[tmpCloudName]:
                    retMap[tmpCloudName].append(tmpSiteName)
    _log.debug('hospital queues : %s' % str(retMap))
    # return
    return retMap


# make compact dialog message
def makeCompactDiagMessage(header,results):
    # limit
    maxSiteList  = 5
    # types for compact format
    compactTypeList = ['status','cpucore']
    # message mapping
    messageMap = {'rel'          : 'missing rel/cache',
                  'pilot'        : 'no pilot', 
                  'status'       : 'not online',
                  'disk'         : 'SE full',
                  'memory'       : 'RAM shortage',
                  'transferring' : 'many transferring',
                  'share'        : 'zero share',
                  'maxtime'      : 'short walltime',
                  'cpucore'      : 'CPU core mismatch',
                  'scratch'      : 'small scratch disk'
                  }
    # put header
    if header in ['',None]:
        retStr = 'No candidate - '
    else:
        retStr = 'special brokerage for %s - ' % header
    # count number of sites per type
    numTypeMap = {}
    for resultType in results:
        resultList = results[resultType]
        # ignore empty
        if len(resultList) == 0:
            continue
        # add
        nSites = len(resultList)
        if nSites not in numTypeMap:
            numTypeMap[nSites] = []
        numTypeMap[nSites].append(resultType)    
    # sort
    numTypeKeys = list(numTypeMap)
    numTypeKeys.sort()
    # use compact format for largest one
    largeTypes = None
    if len(numTypeKeys) > 0:
        largeTypes = numTypeMap[numTypeKeys[-1]]
    # loop over all types
    for numTypeKey in numTypeKeys:
        for resultType in numTypeMap[numTypeKey]:
            # label
            if resultType in messageMap:
                retStr += '%s at ' % messageMap[resultType]
            else:
                retStr += '%s at' % resultType
            # use comact format or not
            if (resultType in compactTypeList+largeTypes \
               or len(results[resultType]) >= maxSiteList) \
               and header in ['',None,'reprocessing'] :
                if len(results[resultType]) == 1:
                    retStr += '%s site' % len(results[resultType])
                else:
                    retStr += '%s sites' % len(results[resultType])
            else:
                for tmpSite in results[resultType]:
                    retStr += '%s,' % tmpSite
                retStr = retStr[:-1]
            retStr += '. '
    retStr = retStr[:-2]
    # return
    return retStr


# message class
class MsgWrapper:
    def __init__(self):
        self.timestamp = datetime.datetime.utcnow().isoformat('/')

    def info(self,msg):
        _log.info(self.timestamp + ' ' + msg)

    def debug(self,msg):
        _log.debug(self.timestamp + ' ' + msg)

    def error(self,msg):
        _log.error(self.timestamp + ' ' + msg)

    def warning(self,msg):
        _log.warning(self.timestamp + ' ' + msg)

        

# schedule
def schedule(jobs,taskBuffer,siteMapper,forAnalysis=False,setScanSiteList=[],trustIS=False,
             distinguishedName=None,specialWeight={},getWeight=False,sizeMapForCheck={},
             datasetSize=0,replicaMap={},pd2pT1=False,reportLog=False,minPriority=None,
             t2FilesMap={},preferredCountries=[],siteReliability=None):
    # make a message instance
    tmpLog = MsgWrapper()
    try:
        tmpLog.debug('start %s %s %s %s minPrio=%s pref=%s siteRel=%s' % (forAnalysis,str(setScanSiteList),trustIS,
                                                                          distinguishedName,minPriority,
                                                                          str(preferredCountries),
                                                                          siteReliability))
        if specialWeight != {}:
            tmpLog.debug('PD2P weight : %s' % str(specialWeight))
        tmpLog.debug('replicaMap : %s' % str(replicaMap))
        # no jobs
        if len(jobs) == 0:
            tmpLog.debug('finished : no jobs')        
            return
        allOkFilesMap = {}

        nJob  = 20
        iJob  = 0
        nFile = 20
        fileList  = []
        guidList  = []
        scopeList = []
        okFiles   = {}
        prioInterval = 50
        totalNumInputs = 0
        totalInputSize = 0
        chosen_ce      = None
        prodDBlock     = None
        computingSite  = None
        dispatchDBlock = None
        previousCloud  = None
        prevRelease    = None
        prevMemory     = None
        prevCmtConfig  = None
        prevProType    = None
        prevSourceLabel= None
        prevDiskCount  = None
        prevHomePkg    = None
        prevDirectAcc  = None
        prevCoreCount  = None
        prevIsJEDI     = None
        prevDDM        = None
        prevBrokergageSiteList = None
        prevManualPreset = None
        prevGoToT2Flag   = None
        prevWorkingGroup = None
        prevMaxCpuCount  = None
        prevBrokerageNote = None
        prevPriority      = None

        nWNmap = {}
        indexJob = 0

        diskThresholdT1   = 20 * 1024
        diskThresholdT2   = 200
        diskThresholdAna  = 200
        diskThresholdPD2P = 1024 * 3
        manyInputsThr     = 20
        weightUsedByBrokerage = {}
        prestageSites = []

        # check if only JEDI
        onlyJEDI = True
        for tmpJob in jobs:
            if tmpJob.lockedby != 'jedi':
                onlyJEDI = False
                break

        # get statistics
        faresharePolicy = {}
        newJobStatWithPrio = {}
        jobStatBrokerClouds = {}
        jobStatBrokerCloudsWithPrio = {}
        if len(jobs) > 0 and (jobs[0].processingType.startswith('gangarobot') or \
                              jobs[0].processingType.startswith('hammercloud') or \
                              jobs[0].processingType in ['pandamover','usermerge'] or \
                              onlyJEDI):
            # disable redundant counting for HC
            jobStatistics = {}
            jobStatBroker = {}
            jobStatBrokerClouds = {}
            nRunningMap = {}
            hospitalQueueMap = {}
        else:
            jobStatistics = taskBuffer.getJobStatistics(forAnal=forAnalysis)
            if not forAnalysis:
                jobStatBroker = {}
                jobStatBrokerClouds = taskBuffer.getJobStatisticsBrokerage()
                faresharePolicy = taskBuffer.getFaresharePolicy()
            else:
                if minPriority is None:
                    jobStatBroker = taskBuffer.getJobStatisticsAnalBrokerage()
                else:
                    jobStatBroker = taskBuffer.getJobStatisticsAnalBrokerage(minPriority=minPriority)                    
                nRunningMap   = taskBuffer.getnRunningInSiteData()
            hospitalQueueMap = getHospitalQueues(siteMapper, forAnalysis)
        # sort jobs by siteID. Some jobs may already define computingSite
        jobs.sort(_compFunc)
        # brokerage for analysis 
        candidateForAnal = True
        relCloudMap      = {}
        loggerMessages   = []
        # get all input files for bulk LFC lookup  
        allLFNs   = []
        allGUIDs  = []
        allScopes = []
        for tmpJob in jobs:<commit_msg>removed deprecated hospital queue stuff<commit_after>import re
import sys
import traceback
import time
import fcntl
import random
import datetime
import uuid
from pandaserver.brokerage import ErrorCode
from pandaserver.taskbuffer import ProcessGroups
from pandaserver.dataservice import DataServiceUtils
from pandaserver.dataservice.DataServiceUtils import select_scope
from pandaserver.dataservice.DDM import rucioAPI
from pandaserver.config import panda_config

from pandacommon.pandalogger.PandaLogger import PandaLogger
_log = PandaLogger().getLogger('broker')

try:
    long
except NameError:
    long = int

# all known sites
_allSites = []


# non LRC checking
_disableLRCcheck = []

# lock for uuidgen
_lockGetUU   = open(panda_config.lockfile_getUU, 'w')

# short-long mapping
shortLongMap = {'ANALY_BNL_SHORT': 'ANALY_BNL_LONG'}

# processingType to skip brokerage
skipBrokerageProTypes = ['prod_test']

# comparison function for sort
def _compFunc(jobA,jobB):
    # append site if not in list
    if not jobA.computingSite in _allSites:
        _allSites.append(jobA.computingSite)
    if not jobB.computingSite in _allSites:
        _allSites.append(jobB.computingSite)
    # compare
    indexA = _allSites.index(jobA.computingSite) 
    indexB = _allSites.index(jobB.computingSite) 
    if indexA > indexB:
        return 1
    elif indexA < indexB:
        return -1
    else:
        return 0


# release checker
def _checkRelease(jobRels,siteRels):
    # all on/off
    if "True" in siteRels:
        return True
    if "False" in siteRels:
        return False
    # loop over all releases
    for tmpRel in jobRels.split('\n'):
        relVer = re.sub('^Atlas-','',tmpRel)
        # not available releases
        if not relVer in siteRels:
            return False
    return True


# get list of files which already exist at the site
def _getOkFiles(v_ce,v_files,v_guids,allLFNs,allGUIDs,allOkFilesMap,prodsourcelabel,tmpLog=None,
                scopeList=None,allScopeList=None):

    scope_association_input, scope_association_output  = select_scope(v_ce, prodsourcelabel)
    dq2IDs = v_ce.setokens_input[scope_association_input].values()
    try:
        dq2IDs.remove('')
    except Exception:
        pass
    dq2IDs.sort()
    if dq2IDs == []:
        dq2ID = v_ce.ddm_input[scope_association_input]
    else:
        dq2ID = ''
        for tmpID in dq2IDs:
            dq2ID += '%s,' % tmpID
        dq2ID = dq2ID[:-1]    
    # set LFC and SE name 
    dq2URL = 'rucio://atlas-rucio.cern.ch:/grid/atlas'
    tmpSE = v_ce.ddm_endpoints_input[scope_association_input].getAllEndPoints()
    if tmpLog is not None:
        tmpLog.debug('getOkFiles for %s with dq2ID:%s,LFC:%s,SE:%s' % (v_ce.sitename,dq2ID,dq2URL,str(tmpSE)))
    anyID = 'any'
    # use bulk lookup
    if allLFNs != []:
        # get all replicas 
        if not dq2URL in allOkFilesMap:
            allOkFilesMap[dq2URL] = {}
            tmpStat,tmpAvaFiles = rucioAPI.listFileReplicas(allScopeList,allLFNs,tmpSE)
            if not tmpStat and tmpLog is not None:
                tmpLog.debug('getOkFile failed to get file replicas')
                tmpAvaFiles = {}
            allOkFilesMap[dq2URL][anyID] = tmpAvaFiles
        # get files for each dq2ID
        if not dq2ID in allOkFilesMap[dq2URL]:
            allOkFilesMap[dq2URL][dq2ID] = allOkFilesMap[dq2URL][anyID]
        # make return map
        retMap = {}
        for tmpLFN in v_files:
            if tmpLFN in allOkFilesMap[dq2URL][dq2ID]:
                retMap[tmpLFN] = allOkFilesMap[dq2URL][dq2ID][tmpLFN]
        tmpLog.debug('getOkFiles done')
        # return
        return retMap
    else:
        # old style
        tmpLog.debug('getOkFiles old')
        return {}


# check reprocessing or not
def _isReproJob(tmpJob):
    if tmpJob is not None:
        if tmpJob.processingType in ['reprocessing']:
            return True
        if tmpJob.transformation in ['csc_cosmics_trf.py','csc_BSreco_trf.py','BStoESDAODDPD_trf.py']:
            return True
    return False

    
# set 'ready' if files are already there
def _setReadyToFiles(tmpJob, okFiles, siteMapper, tmpLog):
    tmpLog.debug(str(okFiles))
    allOK = True
    tmpSiteSpec = siteMapper.getSite(tmpJob.computingSite)
    tmpSrcSpec  = siteMapper.getSite(siteMapper.getCloud(tmpJob.getCloud())['source'])
    scope_association_site_input, scope_association_site_output = select_scope(tmpSiteSpec, tmpJob.prodSourceLabel)
    scope_association_src_input, scope_association_src_output = select_scope(tmpSrcSpec, tmpJob.prodSourceLabel)
    tmpTapeEndPoints = tmpSiteSpec.ddm_endpoints_input[scope_association_site_input].getTapeEndPoints()
    # direct usage of remote SE
    if tmpSiteSpec.ddm_input[scope_association_site_input] != tmpSrcSpec.ddm_input[scope_association_src_input] \
            and tmpSrcSpec.ddm_input[scope_association_src_input] in tmpSiteSpec.setokens_input[scope_association_site_input].values():
        tmpSiteSpec = tmpSrcSpec
        tmpLog.debug('%s uses remote SiteSpec of %s for %s' % (tmpJob.PandaID,tmpSrcSpec.sitename,tmpJob.computingSite))
    for tmpFile in tmpJob.Files:
        if tmpFile.type == 'input':
            if tmpFile.status == 'ready':
                tmpFile.dispatchDBlock = 'NULL'
            elif DataServiceUtils.isCachedFile(tmpFile.dataset,tmpSiteSpec):
                # cached file
                tmpFile.status = 'cached'
                tmpFile.dispatchDBlock = 'NULL'
            elif tmpJob.computingSite == siteMapper.getCloud(tmpJob.getCloud())['source'] or \
                    tmpSiteSpec.ddm_input[scope_association_site_input] == tmpSrcSpec.ddm_input[scope_association_src_input]:
                # use DDM prestage only for on-tape files
                if len(tmpTapeEndPoints) > 0 and tmpFile.lfn in okFiles:
                    tapeOnly = True
                    tapeCopy = False
                    for tmpSE in okFiles[tmpFile.lfn]:
                        if tmpSE not in tmpTapeEndPoints:
                            tapeOnly = False
                        else:
                            # there is a tape copy
                            tapeCopy = True
                    # trigger prestage when disk copy doesn't exist or token is TAPE
                    if tapeOnly or (tapeCopy and tmpFile.dispatchDBlockToken in ['AT<PASSWORD>','<PASSWORD>']):
                        allOK = False
                    else:
                        # set ready                        
                        tmpFile.status = 'ready'
                        tmpFile.dispatchDBlock = 'NULL'
                else:
                    # set ready anyway even if LFC is down. i.e. okFiles doesn't contain the file
                    tmpFile.status = 'ready'
                    tmpFile.dispatchDBlock = 'NULL'                                
            else:
                # set ready if the file exists and the site doesn't use prestage
                tmpFile.status = 'ready'
                tmpFile.dispatchDBlock = 'NULL'
    # unset disp dataset
    if allOK:
        tmpJob.dispatchDBlock = 'NULL'
        
    

# check number/size of inputs
def _isTooManyInput(nFilesPerJob,inputSizePerJob):
    # the number of inputs is larger than 5 or
    # size of inputs is larger than 500MB
    if nFilesPerJob > 5 or inputSizePerJob > 500*1024*1024:
        return True
    return False


# send analysis brokerage info to logger
def sendMsgToLogger(message):
    _log.debug(message)


# send analysis brokerage info to logger with HTTP
def sendMsgToLoggerHTTP(msgList,job):
    try:
        # logging
        iMsg = 0
        # message type
        msgType = 'analy_brokerage'
        # make header
        if not job.jobsetID in [None,'NULL']:
            msgHead = "dn='%s' : jobset=%s jobdef=%s" % (job.prodUserName,job.jobsetID,job.jobDefinitionID)
        else:
            msgHead = "dn='%s' : jobdef=%s" % (job.prodUserName,job.jobDefinitionID)
        for msgBody in msgList:
            # make message
            message = msgHead + ' : ' + msgBody
            # dump locally
            _log.debug(message)
            # get logger
            _pandaLogger = PandaLogger()            
            _pandaLogger.lock()
            _pandaLogger.setParams({'Type':msgType})
            logger = _pandaLogger.getHttpLogger(panda_config.loggername)
            # add message
            logger.info(message)
            # release HTTP handler
            _pandaLogger.release()
            # sleep
            iMsg += 1
            if iMsg % 5 == 0:
                time.sleep(1)
    except Exception:
        errType,errValue = sys.exc_info()[:2]
        _log.error("sendMsgToLoggerHTTP : %s %s" % (errType,errValue))
    

# get T2 candidates when files are missing at T2
def getT2CandList(tmpJob,siteMapper,t2FilesMap):
    if tmpJob is None:
        return []
    # no cloud info
    if tmpJob.getCloud() not in t2FilesMap:
        return []
    # loop over all files
    tmpCandT2s = None
    for tmpFile in tmpJob.Files:
        if tmpFile.type == 'input' and tmpFile.status == 'missing':
            # no dataset info
            if tmpFile.dataset not in t2FilesMap[tmpJob.getCloud()]:
                return []
            # initial candidates
            if tmpCandT2s is None:
                tmpCandT2s = t2FilesMap[tmpJob.getCloud()][tmpFile.dataset]['sites']
            # check all candidates
            newCandT2s = []
            for tmpCandT2 in tmpCandT2s:
                # site doesn't have the dataset
                if tmpCandT2 not in t2FilesMap[tmpJob.getCloud()][tmpFile.dataset]['sites']:
                    continue
                # site has the file
                if tmpFile.lfn in t2FilesMap[tmpJob.getCloud()][tmpFile.dataset]['sites'][tmpCandT2]:
                    if not tmpCandT2 in newCandT2s:
                        newCandT2s.append(tmpCandT2)
            # set new candidates
            tmpCandT2s = newCandT2s
            if tmpCandT2s == []:
                break
    # return [] if no missing files         
    if tmpCandT2s is None:
        return []
    # return
    tmpCandT2s.sort() 
    return tmpCandT2s


# make compact dialog message
def makeCompactDiagMessage(header,results):
    # limit
    maxSiteList  = 5
    # types for compact format
    compactTypeList = ['status','cpucore']
    # message mapping
    messageMap = {'rel'          : 'missing rel/cache',
                  'pilot'        : 'no pilot', 
                  'status'       : 'not online',
                  'disk'         : 'SE full',
                  'memory'       : 'RAM shortage',
                  'transferring' : 'many transferring',
                  'share'        : 'zero share',
                  'maxtime'      : 'short walltime',
                  'cpucore'      : 'CPU core mismatch',
                  'scratch'      : 'small scratch disk'
                  }
    # put header
    if header in ['',None]:
        retStr = 'No candidate - '
    else:
        retStr = 'special brokerage for %s - ' % header
    # count number of sites per type
    numTypeMap = {}
    for resultType in results:
        resultList = results[resultType]
        # ignore empty
        if len(resultList) == 0:
            continue
        # add
        nSites = len(resultList)
        if nSites not in numTypeMap:
            numTypeMap[nSites] = []
        numTypeMap[nSites].append(resultType)    
    # sort
    numTypeKeys = list(numTypeMap)
    numTypeKeys.sort()
    # use compact format for largest one
    largeTypes = None
    if len(numTypeKeys) > 0:
        largeTypes = numTypeMap[numTypeKeys[-1]]
    # loop over all types
    for numTypeKey in numTypeKeys:
        for resultType in numTypeMap[numTypeKey]:
            # label
            if resultType in messageMap:
                retStr += '%s at ' % messageMap[resultType]
            else:
                retStr += '%s at' % resultType
            # use comact format or not
            if (resultType in compactTypeList+largeTypes \
               or len(results[resultType]) >= maxSiteList) \
               and header in ['',None,'reprocessing'] :
                if len(results[resultType]) == 1:
                    retStr += '%s site' % len(results[resultType])
                else:
                    retStr += '%s sites' % len(results[resultType])
            else:
                for tmpSite in results[resultType]:
                    retStr += '%s,' % tmpSite
                retStr = retStr[:-1]
            retStr += '. '
    retStr = retStr[:-2]
    # return
    return retStr


# message class
class MsgWrapper:
    def __init__(self):
        self.timestamp = datetime.datetime.utcnow().isoformat('/')

    def info(self,msg):
        _log.info(self.timestamp + ' ' + msg)

    def debug(self,msg):
        _log.debug(self.timestamp + ' ' + msg)

    def error(self,msg):
        _log.error(self.timestamp + ' ' + msg)

    def warning(self,msg):
        _log.warning(self.timestamp + ' ' + msg)

        

# schedule
def schedule(jobs,taskBuffer,siteMapper,forAnalysis=False,setScanSiteList=[],trustIS=False,
             distinguishedName=None,specialWeight={},getWeight=False,sizeMapForCheck={},
             datasetSize=0,replicaMap={},pd2pT1=False,reportLog=False,minPriority=None,
             t2FilesMap={},preferredCountries=[],siteReliability=None):
    # make a message instance
    tmpLog = MsgWrapper()
    try:
        tmpLog.debug('start %s %s %s %s minPrio=%s pref=%s siteRel=%s' % (forAnalysis,str(setScanSiteList),trustIS,
                                                                          distinguishedName,minPriority,
                                                                          str(preferredCountries),
                                                                          siteReliability))
        if specialWeight != {}:
            tmpLog.debug('PD2P weight : %s' % str(specialWeight))
        tmpLog.debug('replicaMap : %s' % str(replicaMap))
        # no jobs
        if len(jobs) == 0:
            tmpLog.debug('finished : no jobs')        
            return
        allOkFilesMap = {}

        nJob  = 20
        iJob  = 0
        nFile = 20
        fileList  = []
        guidList  = []
        scopeList = []
        okFiles   = {}
        prioInterval = 50
        totalNumInputs = 0
        totalInputSize = 0
        chosen_ce      = None
        prodDBlock     = None
        computingSite  = None
        dispatchDBlock = None
        previousCloud  = None
        prevRelease    = None
        prevMemory     = None
        prevCmtConfig  = None
        prevProType    = None
        prevSourceLabel= None
        prevDiskCount  = None
        prevHomePkg    = None
        prevDirectAcc  = None
        prevCoreCount  = None
        prevIsJEDI     = None
        prevDDM        = None
        prevBrokergageSiteList = None
        prevManualPreset = None
        prevGoToT2Flag   = None
        prevWorkingGroup = None
        prevMaxCpuCount  = None
        prevBrokerageNote = None
        prevPriority      = None

        nWNmap = {}
        indexJob = 0

        diskThresholdT1   = 20 * 1024
        diskThresholdT2   = 200
        diskThresholdAna  = 200
        diskThresholdPD2P = 1024 * 3
        manyInputsThr     = 20
        weightUsedByBrokerage = {}
        prestageSites = []

        # check if only JEDI
        onlyJEDI = True
        for tmpJob in jobs:
            if tmpJob.lockedby != 'jedi':
                onlyJEDI = False
                break

        # get statistics
        faresharePolicy = {}
        newJobStatWithPrio = {}
        jobStatBrokerClouds = {}
        jobStatBrokerCloudsWithPrio = {}
        hospitalQueueMap = {}
        if len(jobs) > 0 and (jobs[0].processingType.startswith('gangarobot') or \
                              jobs[0].processingType.startswith('hammercloud') or \
                              jobs[0].processingType in ['pandamover','usermerge'] or \
                              onlyJEDI):
            # disable redundant counting for HC
            jobStatistics = {}
            jobStatBroker = {}
            jobStatBrokerClouds = {}
            nRunningMap = {}

        else:
            jobStatistics = taskBuffer.getJobStatistics(forAnal=forAnalysis)
            if not forAnalysis:
                jobStatBroker = {}
                jobStatBrokerClouds = taskBuffer.getJobStatisticsBrokerage()
                faresharePolicy = taskBuffer.getFaresharePolicy()
            else:
                if minPriority is None:
                    jobStatBroker = taskBuffer.getJobStatisticsAnalBrokerage()
                else:
                    jobStatBroker = taskBuffer.getJobStatisticsAnalBrokerage(minPriority=minPriority)                    
                nRunningMap   = taskBuffer.getnRunningInSiteData()
        # sort jobs by siteID. Some jobs may already define computingSite
        jobs.sort(_compFunc)
        # brokerage for analysis 
        candidateForAnal = True
        relCloudMap      = {}
        loggerMessages   = []
        # get all input files for bulk LFC lookup  
        allLFNs   = []
        allGUIDs  = []
        allScopes = []
        for tmpJob in jobs:
{'fraction_non_alphanumeric': 0.08330833628752965, 'fraction_numerical': 0.0074991137411033245, 'mean_word_length': 1.882565634334224, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 18, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5372616', 'n_tokens_mistral': 10919, 'n_tokens_neox': 10489, 'n_words': 2470}