<commit_before>'''
setup dataset

'''

import re
import sys
import datetime
import commands
import threading

from config import panda_config
from pandalogger.PandaLogger import PandaLogger
from pandalogger.LogWrapper import LogWrapper
from taskbuffer import EventServiceUtils

# logger
_logger = PandaLogger().getLogger('Setupper')

panda_config.setupPlugin()


# main class
class Setupper (threading.Thread):
    # constructor
    def __init__(self,taskBuffer,jobs,resubmit=False,pandaDDM=False,ddmAttempt=0,forkRun=False,onlyTA=False,
                 resetLocation=False,firstSubmission=True):
        threading.Thread.__init__(self)
        self.jobs       = jobs
        self.taskBuffer = taskBuffer
        # resubmission or not
        self.resubmit = resubmit
        # use PandaDDM
        self.pandaDDM = pandaDDM
        # priority for ddm job
        self.ddmAttempt = ddmAttempt
        # fork another process because python doesn't release memory
        self.forkRun = forkRun
        # run task assignment only
        self.onlyTA = onlyTA
        # reset locations
        self.resetLocation = resetLocation
        # first submission
        self.firstSubmission = firstSubmission

        
        
    # main
    def run(self):
        try:
            # make a message instance
            tmpLog = LogWrapper(_logger,None)
            # run main procedure in the same process
            if not self.forkRun:
                tmpLog.debug('main start')
                tmpLog.debug('firstSubmission={0}'.format(self.firstSubmission))
                # group jobs per VO
                voJobsMap = {}
                ddmFreeJobs = []
                tmpLog.debug('{0} jobs in total'.format(len(self.jobs)))
                for tmpJob in self.jobs:
                    # set VO=local for DDM free 
                    if tmpJob.destinationSE == 'local':
                        tmpVO = 'local'
                    else:
                        tmpVO = tmpJob.VO
                    # make map
                    if not voJobsMap.has_key(tmpVO):
                        voJobsMap[tmpVO] = []
                    voJobsMap[tmpVO].append(tmpJob)
                # loop over all VOs
                for tmpVO,tmpJobList in voJobsMap.iteritems():
                    tmpLog.debug('vo={0} has {1} jobs'.format(tmpVO,len(tmpJobList)))
                    # get plugin
                    setupperPluginClass = panda_config.getPlugin('setupper_plugins',tmpVO)
                    if setupperPluginClass == None:
                        # use ATLAS plug-in by default
                        from SetupperAtlasPlugin import SetupperAtlasPlugin
                        setupperPluginClass = SetupperAtlasPlugin
                    tmpLog.debug('plugin name -> {0}'.format(setupperPluginClass.__name__))
                    try:
                        # make plugin
                        setupperPlugin = setupperPluginClass(self.taskBuffer,self.jobs,tmpLog,
                                                             resubmit=self.resubmit,
                                                             pandaDDM=self.pandaDDM,
                                                             ddmAttempt=self.ddmAttempt,
                                                             onlyTA=self.onlyTA,
                                                             firstSubmission=self.firstSubmission)
                        # run plugin
                        tmpLog.debug('run plugin')
                        setupperPlugin.run()
                        # go forward if not TA
                        if not self.onlyTA:
                            # update jobs
                            tmpLog.debug('update jobs')
                            self.updateJobs(setupperPlugin.jobs+setupperPlugin.jumboJobs,tmpLog)
                            # execute post process
                            tmpLog.debug('post execute plugin')
                            setupperPlugin.postRun()
                        tmpLog.debug('done plugin')
                    except:
                        errtype,errvalue = sys.exc_info()[:2]
                        tmpLog.error('plugin failed with {0}:{1}'.format(errtype, errvalue))
                tmpLog.debug('main end')
            else:
                tmpLog.debug('fork start')
                # write jobs to file
                import os
                import cPickle as pickle
                outFileName = '%s/set.%s_%s' % (panda_config.logdir,self.jobs[0].PandaID,commands.getoutput('uuidgen'))
                outFile = open(outFileName,'w')
                pickle.dump(self.jobs,outFile)
                outFile.close()
                # run main procedure in another process because python doesn't release memory
                com =  'cd %s > /dev/null 2>&1; export HOME=%s; ' % (panda_config.home_dir_cwd,panda_config.home_dir_cwd)
                com += 'source %s; ' % panda_config.glite_source
                com += 'env PYTHONPATH=%s:%s %s/python -Wignore %s/dataservice/forkSetupper.py -i %s' % \
                       (panda_config.pandaCommon_dir,panda_config.pandaPython_dir,panda_config.native_python,
                        panda_config.pandaPython_dir,outFileName)
                if self.onlyTA:
                    com += " -t"
                if not self.firstSubmission:
                    com += " -f"
                tmpLog.debug(com)
                # exeute
                status,output = self.taskBuffer.processLimiter.getstatusoutput(com)
                tmpLog.debug("return from main process: %s %s" % (status,output))                
                tmpLog.debug('fork end')
        except:
            errtype,errvalue = sys.exc_info()[:2]
            tmpLog.error('master failed with {0}:{1}'.format(errtype,errvalue))



    #  update jobs
    def updateJobs(self,jobList,tmpLog):
        updateJobs   = []
        failedJobs   = []
        activateJobs = []
        waitingJobs  = []
        closeJobs  = []
        # sort out jobs
        for job in jobList:
            # failed jobs
            if job.jobStatus in ['failed','cancelled']:
                failedJobs.append(job)
            # waiting
            elif job.jobStatus == 'waiting':
                waitingJobs.append(job)
            # no input jobs
            elif job.dispatchDBlock=='NULL':
                activateJobs.append(job)
            # normal jobs
            else:
                # change status
                job.jobStatus = "assigned"
                updateJobs.append(job)
        # update DB
        tmpLog.debug('# of activated jobs : {0}'.format(len(activateJobs)))
        self.taskBuffer.activateJobs(activateJobs)
        tmpLog.debug('# of updated jobs : {0}'.format(len(updateJobs)))
        self.taskBuffer.updateJobs(updateJobs,True)
        tmpLog.debug('# of failed jobs : {0}'.format(len(failedJobs)))
        self.taskBuffer.updateJobs(failedJobs,True)
        tmpLog.debug('# of waiting jobs : {0}'.format(len(waitingJobs)))
        self.taskBuffer.keepJobs(waitingJobs)
        # to trigger merge generation if all events are done
        finishedJobs = []
        for job in activateJobs:
            if job.notDiscardEvents() and job.allOkEvents() and not EventServiceUtils.isEventServiceMerge(job):
                # change status
                job.jobStatus = "finished"
                finishedJobs.append(job)
        tmpLog.debug('# of finished jobs in activated : {0}'.format(len(finishedJobs)))
        self.taskBuffer.updateJobs(finishedJobs, False)
        finishedJobs = []
        for job in updateJobs:
            if job.notDiscardEvents() and job.allOkEvents() and not EventServiceUtils.isEventServiceMerge(job):
                # change status
                job.jobStatus = "finished"
                finishedJobs.append(job)
        tmpLog.debug('# of finished jobs in defined : {0}'.format(len(finishedJobs)))
        self.taskBuffer.updateJobs(finishedJobs, True)
        # delete local values
        del updateJobs
        del failedJobs
        del activateJobs
        del waitingJobs<commit_msg>fixed jobs with all events done not to be killed before generating merge jobs<commit_after>'''
setup dataset

'''

import re
import sys
import datetime
import commands
import threading

from config import panda_config
from pandalogger.PandaLogger import PandaLogger
from pandalogger.LogWrapper import LogWrapper
from taskbuffer import EventServiceUtils

# logger
_logger = PandaLogger().getLogger('Setupper')

panda_config.setupPlugin()


# main class
class Setupper (threading.Thread):
    # constructor
    def __init__(self,taskBuffer,jobs,resubmit=False,pandaDDM=False,ddmAttempt=0,forkRun=False,onlyTA=False,
                 resetLocation=False,firstSubmission=True):
        threading.Thread.__init__(self)
        self.jobs       = jobs
        self.taskBuffer = taskBuffer
        # resubmission or not
        self.resubmit = resubmit
        # use PandaDDM
        self.pandaDDM = pandaDDM
        # priority for ddm job
        self.ddmAttempt = ddmAttempt
        # fork another process because python doesn't release memory
        self.forkRun = forkRun
        # run task assignment only
        self.onlyTA = onlyTA
        # reset locations
        self.resetLocation = resetLocation
        # first submission
        self.firstSubmission = firstSubmission

        
        
    # main
    def run(self):
        try:
            # make a message instance
            tmpLog = LogWrapper(_logger,None)
            # run main procedure in the same process
            if not self.forkRun:
                tmpLog.debug('main start')
                tmpLog.debug('firstSubmission={0}'.format(self.firstSubmission))
                # group jobs per VO
                voJobsMap = {}
                ddmFreeJobs = []
                tmpLog.debug('{0} jobs in total'.format(len(self.jobs)))
                for tmpJob in self.jobs:
                    # set VO=local for DDM free 
                    if tmpJob.destinationSE == 'local':
                        tmpVO = 'local'
                    else:
                        tmpVO = tmpJob.VO
                    # make map
                    if not voJobsMap.has_key(tmpVO):
                        voJobsMap[tmpVO] = []
                    voJobsMap[tmpVO].append(tmpJob)
                # loop over all VOs
                for tmpVO,tmpJobList in voJobsMap.iteritems():
                    tmpLog.debug('vo={0} has {1} jobs'.format(tmpVO,len(tmpJobList)))
                    # get plugin
                    setupperPluginClass = panda_config.getPlugin('setupper_plugins',tmpVO)
                    if setupperPluginClass == None:
                        # use ATLAS plug-in by default
                        from SetupperAtlasPlugin import SetupperAtlasPlugin
                        setupperPluginClass = SetupperAtlasPlugin
                    tmpLog.debug('plugin name -> {0}'.format(setupperPluginClass.__name__))
                    try:
                        # make plugin
                        setupperPlugin = setupperPluginClass(self.taskBuffer,self.jobs,tmpLog,
                                                             resubmit=self.resubmit,
                                                             pandaDDM=self.pandaDDM,
                                                             ddmAttempt=self.ddmAttempt,
                                                             onlyTA=self.onlyTA,
                                                             firstSubmission=self.firstSubmission)
                        # run plugin
                        tmpLog.debug('run plugin')
                        setupperPlugin.run()
                        # go forward if not TA
                        if not self.onlyTA:
                            # update jobs
                            tmpLog.debug('update jobs')
                            self.updateJobs(setupperPlugin.jobs+setupperPlugin.jumboJobs,tmpLog)
                            # execute post process
                            tmpLog.debug('post execute plugin')
                            setupperPlugin.postRun()
                        tmpLog.debug('done plugin')
                    except:
                        errtype,errvalue = sys.exc_info()[:2]
                        tmpLog.error('plugin failed with {0}:{1}'.format(errtype, errvalue))
                tmpLog.debug('main end')
            else:
                tmpLog.debug('fork start')
                # write jobs to file
                import os
                import cPickle as pickle
                outFileName = '%s/set.%s_%s' % (panda_config.logdir,self.jobs[0].PandaID,commands.getoutput('uuidgen'))
                outFile = open(outFileName,'w')
                pickle.dump(self.jobs,outFile)
                outFile.close()
                # run main procedure in another process because python doesn't release memory
                com =  'cd %s > /dev/null 2>&1; export HOME=%s; ' % (panda_config.home_dir_cwd,panda_config.home_dir_cwd)
                com += 'source %s; ' % panda_config.glite_source
                com += 'env PYTHONPATH=%s:%s %s/python -Wignore %s/dataservice/forkSetupper.py -i %s' % \
                       (panda_config.pandaCommon_dir,panda_config.pandaPython_dir,panda_config.native_python,
                        panda_config.pandaPython_dir,outFileName)
                if self.onlyTA:
                    com += " -t"
                if not self.firstSubmission:
                    com += " -f"
                tmpLog.debug(com)
                # exeute
                status,output = self.taskBuffer.processLimiter.getstatusoutput(com)
                tmpLog.debug("return from main process: %s %s" % (status,output))                
                tmpLog.debug('fork end')
        except:
            errtype,errvalue = sys.exc_info()[:2]
            tmpLog.error('master failed with {0}:{1}'.format(errtype,errvalue))



    #  update jobs
    def updateJobs(self,jobList,tmpLog):
        updateJobs   = []
        failedJobs   = []
        activateJobs = []
        waitingJobs  = []
        closeJobs  = []
        # sort out jobs
        for job in jobList:
            # failed jobs
            if job.jobStatus in ['failed','cancelled']:
                failedJobs.append(job)
            # waiting
            elif job.jobStatus == 'waiting':
                waitingJobs.append(job)
            # no input jobs
            elif job.dispatchDBlock=='NULL':
                activateJobs.append(job)
            # normal jobs
            else:
                # change status
                job.jobStatus = "assigned"
                updateJobs.append(job)
        # trigger merge generation if all events are done
        newActivateJobs = []
        nFinished = 0
        for job in activateJobs:
            if job.notDiscardEvents() and job.allOkEvents() and not EventServiceUtils.isEventServiceMerge(job):
                self.taskBuffer.activateJobs([job])
                # change status
                job.jobStatus = "finished"
                self.taskBuffer.updateJobs([job], False)
                nFinished += 1
            else:
                newActivateJobs.append(job)
        activateJobs = newActivateJobs
        tmpLog.debug('# of finished jobs in activated : {0}'.format(nFinished))
        newUpdateJobs = []
        nFinished = 0
        for job in updateJobs:
            if job.notDiscardEvents() and job.allOkEvents() and not EventServiceUtils.isEventServiceMerge(job):
                self.taskBuffer.updateJobs([job], True)
                # change status
                job.jobStatus = "finished"
                self.taskBuffer.updateJobs([job], True)
                nFinished += 1
            else:
                newUpdateJobs.append(job)
        updateJobs = newUpdateJobs
        tmpLog.debug('# of finished jobs in defined : {0}'.format(nFinished))
        # update DB
        tmpLog.debug('# of activated jobs : {0}'.format(len(activateJobs)))
        self.taskBuffer.activateJobs(activateJobs)
        tmpLog.debug('# of updated jobs : {0}'.format(len(updateJobs)))
        self.taskBuffer.updateJobs(updateJobs,True)
        tmpLog.debug('# of failed jobs : {0}'.format(len(failedJobs)))
        self.taskBuffer.updateJobs(failedJobs,True)
        tmpLog.debug('# of waiting jobs : {0}'.format(len(waitingJobs)))
        self.taskBuffer.keepJobs(waitingJobs)
        # delete local values
        del updateJobs
        del failedJobs
        del activateJobs
        del waitingJobs
{'fraction_non_alphanumeric': 0.08443002223424073, 'fraction_numerical': 0.0027642569557117963, 'mean_word_length': 1.492063492063492, 'pattern_counts': {'":': 0, '<': 3, '<?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': 20, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1347920', 'n_tokens_mistral': 4426, 'n_tokens_neox': 4191, 'n_words': 918}