<commit_before>from pandajedi.jedicore import Interaction
from pandajedi.jedicore.InputChunk import InputChunk
from pandajedi.jedicore.MsgWrapper import MsgWrapper

# logger
from pandacommon.pandalogger.PandaLogger import PandaLogger
logger = PandaLogger().getLogger(__name__.split('.')[-1])


# class to split job
class JobSplitter:

    # constructor
    def __init__(self):
        self.sizeGradientsPerInSizeForMerge = 1.2
        self.interceptsMerginForMerge = 500 * 1024 * 1024

        

    # split
    def doSplit(self,taskSpec,inputChunk,siteMapper):
        # return for failure
        retFatal    = self.SC_FATAL,[]
        retTmpError = self.SC_FAILED,[]
        # make logger
        tmpLog = MsgWrapper(logger,'<jediTaskID={0} datasetID={1}>'.format(taskSpec.jediTaskID,inputChunk.masterIndexName))
        tmpLog.debug('start')
        if not inputChunk.isMerging:
            # set maxNumFiles using taskSpec if specified
            maxNumFiles = taskSpec.getMaxNumFilesPerJob()
            # set fsize gradients using taskSpec
            sizeGradients  = taskSpec.getOutDiskSize()
            # set fsize intercepts using taskSpec                
            sizeIntercepts = taskSpec.getWorkDiskSize()
            # walltime
            if not taskSpec.useHS06():
                walltimeGradient = taskSpec.walltime
            else:
                walltimeGradient = taskSpec.cpuTime
            # number of events per job if defined
            nEventsPerJob = taskSpec.getNumEventsPerJob()
            # number of files per job if defined
            if not taskSpec.dynamicNumEvents():
                nFilesPerJob = taskSpec.getNumFilesPerJob()
            else:
                nFilesPerJob = None
            if nFilesPerJob == None and nEventsPerJob == None and inputChunk.useScout() \
                    and not taskSpec.useLoadXML() and not taskSpec.respectSplitRule():
                nFilesPerJob = 1
            # grouping with boundaryID
            useBoundary = taskSpec.useGroupWithBoundaryID()
            # fsize intercepts per input size
            sizeGradientsPerInSize = None
            # max primay output size
            maxOutSize = None
            # max size per job
            maxSizePerJob = taskSpec.getMaxSizePerJob()
            if maxSizePerJob is not None:
                maxSizePerJob += InputChunk.defaultOutputSize
            # dynamic number of events
            dynNumEvents = taskSpec.dynamicNumEvents()
            # max number of event ranges
            maxNumEventRanges = None
            # multiplicity of jobs
            if taskSpec.useJobCloning():
                multiplicity = 1
            else:
                multiplicity = taskSpec.getNumEventServiceConsumer()
            # split with fields
            if taskSpec.getFieldNumToLFN() != None and taskSpec.useFileAsSourceLFN():
                splitByFields = taskSpec.getFieldNumToLFN()
            else:
                splitByFields = None
        else:
            # set parameters for merging
            maxNumFiles = taskSpec.getMaxNumFilesPerMergeJob()
            sizeGradients = 0
            walltimeGradient = 0
            nFilesPerJob = taskSpec.getNumFilesPerMergeJob()
            nEventsPerJob = taskSpec.getNumEventsPerMergeJob()
            maxSizePerJob = None
            useBoundary = {'inSplit':3}
            dynNumEvents = False
            maxNumEventRanges = None
            multiplicity = None
            # gradients per input size is 1 + margin
            sizeGradientsPerInSize = self.sizeGradientsPerInSizeForMerge
            # intercepts for libDS
            sizeIntercepts = taskSpec.getWorkDiskSize()
            # mergein of 500MB
            interceptsMergin = self.interceptsMerginForMerge
            if sizeIntercepts < interceptsMergin:
                sizeIntercepts = interceptsMergin
            maxOutSize = taskSpec.getMaxSizePerMergeJob()
            if maxOutSize == None:
                # max output size is 5GB for merging by default
                maxOutSize = 5 * 1024 * 1024 * 1024
            # split with fields
            if taskSpec.getFieldNumToLFN() != None and taskSpec.useFileAsSourceLFN():
                splitByFields = range(4+1,4+1+len(taskSpec.getFieldNumToLFN()))
            else:
                splitByFields = None
        # LB
        respectLB = taskSpec.respectLumiblock()
        # dump
        tmpLog.debug('maxNumFiles={0} sizeGradients={1} sizeIntercepts={2} useBoundary={3}'.format(maxNumFiles,
                                                                                                   sizeGradients,
                                                                                                   sizeIntercepts,
                                                                                                   useBoundary))
        tmpLog.debug('walltimeGradient={0} nFilesPerJob={1} nEventsPerJob={2}'.format(walltimeGradient,
                                                                                        nFilesPerJob,
                                                                                        nEventsPerJob))
        tmpLog.debug('sizeGradientsPerInSize={0} maxOutSize={1} respectLB={2} dynNumEvents={3}'.format(sizeGradientsPerInSize,
                                                                                                       maxOutSize,
                                                                                                       respectLB,
                                                                                                       dynNumEvents))
        tmpLog.debug('multiplicity={0} splitByFields={1} nFiles={2}'.format(multiplicity,str(splitByFields),
                                                                            inputChunk.getNumFilesInMaster()))
        # split
        returnList = []
        subChunks  = []
        iSubChunks = 0
        nSubChunks = 25
        subChunk   = None
        while True:
            # change site
            if iSubChunks % nSubChunks == 0 or subChunk == []:
                # append to return map
                if subChunks != []:
                    # get site names for parallel execution
                    if taskSpec.getNumSitesPerJob() > 1 and not inputChunk.isMerging and inputChunk.useJumbo != 'fake':
                        siteName = inputChunk.getParallelSites(taskSpec.getNumSitesPerJob(),
                                                               nSubChunks,[siteName])
                    returnList.append({'siteName':siteName,
                                       'subChunks':subChunks,
                                       'siteCandidate':siteCandidate,
                                       })
                    tmpLog.debug('split to %s subchunks' % len(subChunks))
                    # reset
                    subChunks = []
                # skip unavailable files in distributed datasets
                nSkip = inputChunk.skipUnavailableFiles()
                tmpLog.debug('skipped {0} files'.format(nSkip))
                # new candidate
                siteCandidate = inputChunk.getOneSiteCandidate(nSubChunks)
                if siteCandidate == None:
                    break
                siteName = siteCandidate.siteName
                siteSpec = siteMapper.getSite(siteName)
                # directIO
                if taskSpec.useLocalIO() or not siteSpec.isDirectIO() or taskSpec.allowInputLAN() is None \
                        or inputChunk.isMerging:
                    useDirectIO = False
                else:
                    useDirectIO = True
                # get maxSize if it is set in taskSpec
                maxSize = maxSizePerJob
                if maxSize == None:
                    # use maxwdir as the default maxSize
                    if not useDirectIO:
                        maxSize = siteSpec.maxwdir * 1024 * 1024
                    else:
                        maxSize = max(20, siteSpec.maxwdir) * 1024 * 1024
                else:
                    # add offset
                    maxSize += sizeIntercepts
                # max disk size
                maxDiskSize = siteSpec.maxwdir * 1024 * 1024
                # max walltime
                maxWalltime = None
                if not inputChunk.isMerging:
                    maxWalltime = taskSpec.getMaxWalltime()
                if maxWalltime is None:
                    maxWalltime = siteSpec.maxtime
                # core count
                if siteSpec.coreCount > 0:
                    coreCount = siteSpec.coreCount
                else:
                    coreCount = 1
                # core power
                corePower = siteSpec.corepower
                # max num of event ranges for dynNumEvents
                if dynNumEvents:
                    maxNumEventRanges = int(siteSpec.get_n_sim_events() / taskSpec.get_min_granularity())
                    if maxNumEventRanges == 0:
                        maxNumEventRanges = 1
                tmpLog.debug('chosen {0}'.format(siteName))
                tmpLog.debug('new weight {0}'.format(siteCandidate.weight))
                tmpLog.debug('maxSize={0} maxWalltime={1} coreCount={2} corePower={3} maxNumEventRanges={4} maxDisk={5}'.format(maxSize,maxWalltime,
                                                                                                                                coreCount,corePower,
                                                                                                                                maxNumEventRanges,
                                                                                                                                maxDiskSize))
                tmpLog.debug('useDirectIO={0} label={1}'.format(useDirectIO, taskSpec.prodSourceLabel))<commit_msg>maxSize for direct<commit_after>from pandajedi.jedicore import Interaction
from pandajedi.jedicore.InputChunk import InputChunk
from pandajedi.jedicore.MsgWrapper import MsgWrapper

# logger
from pandacommon.pandalogger.PandaLogger import PandaLogger
logger = PandaLogger().getLogger(__name__.split('.')[-1])


# class to split job
class JobSplitter:

    # constructor
    def __init__(self):
        self.sizeGradientsPerInSizeForMerge = 1.2
        self.interceptsMerginForMerge = 500 * 1024 * 1024

        

    # split
    def doSplit(self,taskSpec,inputChunk,siteMapper):
        # return for failure
        retFatal    = self.SC_FATAL,[]
        retTmpError = self.SC_FAILED,[]
        # make logger
        tmpLog = MsgWrapper(logger,'<jediTaskID={0} datasetID={1}>'.format(taskSpec.jediTaskID,inputChunk.masterIndexName))
        tmpLog.debug('start')
        if not inputChunk.isMerging:
            # set maxNumFiles using taskSpec if specified
            maxNumFiles = taskSpec.getMaxNumFilesPerJob()
            # set fsize gradients using taskSpec
            sizeGradients  = taskSpec.getOutDiskSize()
            # set fsize intercepts using taskSpec                
            sizeIntercepts = taskSpec.getWorkDiskSize()
            # walltime
            if not taskSpec.useHS06():
                walltimeGradient = taskSpec.walltime
            else:
                walltimeGradient = taskSpec.cpuTime
            # number of events per job if defined
            nEventsPerJob = taskSpec.getNumEventsPerJob()
            # number of files per job if defined
            if not taskSpec.dynamicNumEvents():
                nFilesPerJob = taskSpec.getNumFilesPerJob()
            else:
                nFilesPerJob = None
            if nFilesPerJob == None and nEventsPerJob == None and inputChunk.useScout() \
                    and not taskSpec.useLoadXML() and not taskSpec.respectSplitRule():
                nFilesPerJob = 1
            # grouping with boundaryID
            useBoundary = taskSpec.useGroupWithBoundaryID()
            # fsize intercepts per input size
            sizeGradientsPerInSize = None
            # max primay output size
            maxOutSize = None
            # max size per job
            maxSizePerJob = taskSpec.getMaxSizePerJob()
            if maxSizePerJob is not None:
                maxSizePerJob += InputChunk.defaultOutputSize
            # dynamic number of events
            dynNumEvents = taskSpec.dynamicNumEvents()
            # max number of event ranges
            maxNumEventRanges = None
            # multiplicity of jobs
            if taskSpec.useJobCloning():
                multiplicity = 1
            else:
                multiplicity = taskSpec.getNumEventServiceConsumer()
            # split with fields
            if taskSpec.getFieldNumToLFN() != None and taskSpec.useFileAsSourceLFN():
                splitByFields = taskSpec.getFieldNumToLFN()
            else:
                splitByFields = None
        else:
            # set parameters for merging
            maxNumFiles = taskSpec.getMaxNumFilesPerMergeJob()
            sizeGradients = 0
            walltimeGradient = 0
            nFilesPerJob = taskSpec.getNumFilesPerMergeJob()
            nEventsPerJob = taskSpec.getNumEventsPerMergeJob()
            maxSizePerJob = None
            useBoundary = {'inSplit':3}
            dynNumEvents = False
            maxNumEventRanges = None
            multiplicity = None
            # gradients per input size is 1 + margin
            sizeGradientsPerInSize = self.sizeGradientsPerInSizeForMerge
            # intercepts for libDS
            sizeIntercepts = taskSpec.getWorkDiskSize()
            # mergein of 500MB
            interceptsMergin = self.interceptsMerginForMerge
            if sizeIntercepts < interceptsMergin:
                sizeIntercepts = interceptsMergin
            maxOutSize = taskSpec.getMaxSizePerMergeJob()
            if maxOutSize == None:
                # max output size is 5GB for merging by default
                maxOutSize = 5 * 1024 * 1024 * 1024
            # split with fields
            if taskSpec.getFieldNumToLFN() != None and taskSpec.useFileAsSourceLFN():
                splitByFields = range(4+1,4+1+len(taskSpec.getFieldNumToLFN()))
            else:
                splitByFields = None
        # LB
        respectLB = taskSpec.respectLumiblock()
        # dump
        tmpLog.debug('maxNumFiles={0} sizeGradients={1} sizeIntercepts={2} useBoundary={3}'.format(maxNumFiles,
                                                                                                   sizeGradients,
                                                                                                   sizeIntercepts,
                                                                                                   useBoundary))
        tmpLog.debug('walltimeGradient={0} nFilesPerJob={1} nEventsPerJob={2}'.format(walltimeGradient,
                                                                                        nFilesPerJob,
                                                                                        nEventsPerJob))
        tmpLog.debug('sizeGradientsPerInSize={0} maxOutSize={1} respectLB={2} dynNumEvents={3}'.format(sizeGradientsPerInSize,
                                                                                                       maxOutSize,
                                                                                                       respectLB,
                                                                                                       dynNumEvents))
        tmpLog.debug('multiplicity={0} splitByFields={1} nFiles={2}'.format(multiplicity,str(splitByFields),
                                                                            inputChunk.getNumFilesInMaster()))
        # split
        returnList = []
        subChunks  = []
        iSubChunks = 0
        nSubChunks = 25
        subChunk   = None
        while True:
            # change site
            if iSubChunks % nSubChunks == 0 or subChunk == []:
                # append to return map
                if subChunks != []:
                    # get site names for parallel execution
                    if taskSpec.getNumSitesPerJob() > 1 and not inputChunk.isMerging and inputChunk.useJumbo != 'fake':
                        siteName = inputChunk.getParallelSites(taskSpec.getNumSitesPerJob(),
                                                               nSubChunks,[siteName])
                    returnList.append({'siteName':siteName,
                                       'subChunks':subChunks,
                                       'siteCandidate':siteCandidate,
                                       })
                    tmpLog.debug('split to %s subchunks' % len(subChunks))
                    # reset
                    subChunks = []
                # skip unavailable files in distributed datasets
                nSkip = inputChunk.skipUnavailableFiles()
                tmpLog.debug('skipped {0} files'.format(nSkip))
                # new candidate
                siteCandidate = inputChunk.getOneSiteCandidate(nSubChunks)
                if siteCandidate == None:
                    break
                siteName = siteCandidate.siteName
                siteSpec = siteMapper.getSite(siteName)
                # directIO
                if taskSpec.useLocalIO() or not siteSpec.isDirectIO() or taskSpec.allowInputLAN() is None \
                        or inputChunk.isMerging:
                    useDirectIO = False
                else:
                    useDirectIO = True
                # get maxSize if it is set in taskSpec
                maxSize = maxSizePerJob
                if maxSize == None:
                    # use maxwdir as the default maxSize
                    if not useDirectIO:
                        maxSize = siteSpec.maxwdir * 1024 * 1024
                    else:
                        maxSize = max(20000, siteSpec.maxwdir) * 1024 * 1024
                else:
                    # add offset
                    maxSize += sizeIntercepts
                # max disk size
                maxDiskSize = siteSpec.maxwdir * 1024 * 1024
                # max walltime
                maxWalltime = None
                if not inputChunk.isMerging:
                    maxWalltime = taskSpec.getMaxWalltime()
                if maxWalltime is None:
                    maxWalltime = siteSpec.maxtime
                # core count
                if siteSpec.coreCount > 0:
                    coreCount = siteSpec.coreCount
                else:
                    coreCount = 1
                # core power
                corePower = siteSpec.corepower
                # max num of event ranges for dynNumEvents
                if dynNumEvents:
                    maxNumEventRanges = int(siteSpec.get_n_sim_events() / taskSpec.get_min_granularity())
                    if maxNumEventRanges == 0:
                        maxNumEventRanges = 1
                tmpLog.debug('chosen {0}'.format(siteName))
                tmpLog.debug('new weight {0}'.format(siteCandidate.weight))
                tmpLog.debug('maxSize={0} maxWalltime={1} coreCount={2} corePower={3} maxNumEventRanges={4} maxDisk={5}'.format(maxSize,maxWalltime,
                                                                                                                                coreCount,corePower,
                                                                                                                                maxNumEventRanges,
                                                                                                                                maxDiskSize))
                tmpLog.debug('useDirectIO={0} label={1}'.format(useDirectIO, taskSpec.prodSourceLabel))
{'fraction_non_alphanumeric': 0.06295399515738499, 'fraction_numerical': 0.01074455205811138, 'mean_word_length': 1.2247783638199978, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 110, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1538217', 'n_tokens_mistral': 4731, 'n_tokens_neox': 4402, 'n_words': 999}