

def clean(self, jobCache=None):
    '\n        Function to cleanup the state of a job store after a restart.\n        Fixes jobs that might have been partially updated. Resets the try counts and removes jobs\n        that are not successors of the current root job.\n\n        :param dict[str,toil.jobWrapper.JobWrapper] jobCache: if a value it must be a dict\n               from job ID keys to JobWrapper object values. Jobs will be loaded from the cache\n               (which can be downloaded from the job store in a batch) instead of piecemeal when \n               recursed into.\n        '
    reachableFromRoot = set()
    if (jobCache is None):
        logger.warning('Cleaning jobStore recursively. This may be slow.')

    def getJob(jobId):
        if (jobCache is not None):
            return jobCache[jobId]
        else:
            return self.load(jobId)

    def haveJob(jobId):
        if (jobCache is not None):
            return jobCache.has_key(jobId)
        else:
            return self.exists(jobId)

    def getJobs():
        if (jobCache is not None):
            return jobCache.itervalues()
        else:
            return self.jobs()

    def getConnectedJobs(jobWrapper):
        if (jobWrapper.jobStoreID in reachableFromRoot):
            return
        reachableFromRoot.add(jobWrapper.jobStoreID)
        for jobs in jobWrapper.stack:
            for successorJobStoreID in map((lambda x: x[0]), jobs):
                if ((successorJobStoreID not in reachableFromRoot) and haveJob(successorJobStoreID)):
                    getConnectedJobs(getJob(successorJobStoreID))
        for jobs in jobWrapper.services:
            for serviceJobStoreID in map((lambda x: x[0]), jobs):
                assert (serviceJobStoreID not in reachableFromRoot)
                reachableFromRoot.add(serviceJobStoreID)
    logger.info('Checking job graph connectivity...')
    getConnectedJobs(self.loadRootJob())
    logger.info(('%d jobs reachable from root.' % len(reachableFromRoot)))
    for jobWrapper in getJobs():
        changed = [False]
        if (len(jobWrapper.filesToDelete) != 0):
            for fileID in jobWrapper.filesToDelete:
                logger.critical(('Removing file in job store: %s that was marked for deletion but not previously removed' % fileID))
                changed.deleteFile(fileID)
            jobWrapper.filesToDelete = []
            changed[0] = True
        if (jobWrapper.jobStoreID not in reachableFromRoot):
            logger.critical(('Removing job: %s that is not a successor of the root job in cleanup' % jobWrapper.jobStoreID))
            self.delete(jobWrapper.jobStoreID)
            continue
        if (jobWrapper.command is None):
            stackSize = sum(map(len, jobWrapper.stack))
            jobWrapper.stack = map((lambda x: filter((lambda y: self.exists(y[0])), x)), jobWrapper.stack)
            jobWrapper.stack = filter((lambda x: (len(x) > 0)), jobWrapper.stack)
            if (sum(map(len, jobWrapper.stack)) != stackSize):
                changed[0] = True

        def subFlagFile(jobStoreID, jobStoreFileID, flag):
            if self.fileExists(jobStoreFileID):
                return jobStoreFileID
            newFlag = self.getEmptyFileStoreID()
            serviceJobWrapper = getJob(jobStoreID)
            if (flag == 1):
                logger.debug('Recreating a start service flag for job: %s, flag: %s', jobStoreID, newFlag)
                serviceJobWrapper.startJobStoreID = newFlag
            elif (flag == 2):
                logger.debug('Recreating a terminate service flag for job: %s, flag: %s', jobStoreID, newFlag)
                serviceJobWrapper.terminateJobStoreID = newFlag
            else:
                logger.debug('Recreating a error service flag for job: %s, flag: %s', jobStoreID, newFlag)
                assert (flag == 3)
                serviceJobWrapper.errorJobStoreID = newFlag
            self.update(serviceJobWrapper)
            changed[0] = True
            return newFlag
        servicesSize = sum(map(len, jobWrapper.services))
        jobWrapper.services = filter((lambda z: (len(z) > 0)), map((lambda serviceJobList: map((lambda x: (x[:4] + (subFlagFile(x[0], x[4], 1), subFlagFile(x[0], x[5], 2), subFlagFile(x[0], x[6], 3)))), filter((lambda y: self.exists(y[0])), serviceJobList))), jobWrapper.services))
        if (sum(map(len, jobWrapper.services)) != servicesSize):
            changed[0] = True
        if (jobWrapper.remainingRetryCount != self._defaultTryCount()):
            jobWrapper.remainingRetryCount = self._defaultTryCount()
            changed[0] = True
        if (jobWrapper.logJobStoreFileID != None):
            self.delete(jobWrapper.logJobStoreFileID)
            jobWrapper.logJobStoreFileID = None
            changed[0] = True
        if changed[0]:
            logger.critical(('Repairing job: %s' % jobWrapper.jobStoreID))
            self.update(jobWrapper)
    logger.info('Discarding old statistics and logs...')
    self.readStatsAndLogging((lambda x: None))
    logger.info('Job store is clean')
    return self.loadRootJob()

dataset/ETHPy150Open BD2KGenomics/toil/src/toil/jobStores/abstractJobStore.py AbstractJobStore.clean/VarMisuse@375/769 `self`->`changed`