<commit_before>#!/usr/bin/env python
# -*- python -*-
#
# The _open_SmashBox Project.
#
# Author: <NAME>, CERN, 2013
# License: AGPL
#
#$Id: $
#
# %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
# Perform internal setup of the environment.
# This is a Copy/Paste logic which must stay in THIS file
def standardSetup():
   import sys, os.path
   # insert the path to cernafs based on the relative position of this scrip inside the service directory tree
   exeDir = os.path.abspath(os.path.normpath(os.path.dirname(os.path.dirname(sys.argv[0]))))
   pythonDir = os.path.join(os.path.dirname(exeDir), 'python' )
   sys.path.insert(0, pythonDir)
   import smashbox.setup
   smashbox.setup.standardSetup(sys.argv[0]) # execute a setup hook

standardSetup()
del standardSetup
# %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

# obsolete to be removed
def log(*args,**kwds):
    import time
    print time.ctime(),_smash_.process_name,(" ".join([str(s) for s in args]))%kwds


#import shelve

class _smash_:
    """ Internals of the stepper synchronization framework. This class
    is merely a namespace to avoid polluting global namespace of
    user's test scripts.
    """

    # all the rest of attributes are internal -- accessors for process_name and common_dict are defined in utilities

    process_name = None
    process_number = 0
    
    DEBUG = False

    # this is a hardcoded maximum number of steps
    N_STEPS = 100
    
    workers = []

    class SmashSharedObject:
        """ A bunch of shared attributes stored in a directory as separate files.
        """
        
        def __init__(self,d):
           self._dir = d
           self._makedir()

        def _makedir(self):
           try:
              os.makedirs(self._dir)
           except OSError,x:
              import errno
              if x.errno != errno.EEXIST:
                 raise
                       
        def __getitem__(self,key):
           self._makedir()
           import pickle
           try:
              return pickle.load(file(os.path.join(self._dir,'_attr_'+key)))
           except Exception,x:
              logger.debug(x)
              raise AttributeError(x)
        
        def __setitem__(self,key,val):
           self._makedir()
           import pickle
           import time
           tmp_name = os.path.join(self._dir,'tmp.%s.%s._attr_%s'%(os.getpid(),time.time(),key))
           dest_name = os.path.join(self._dir,'_attr_'+key)
                           
           pickle.dump(val,file(tmp_name,'w'))

           import shutil
           shutil.move(tmp_name,dest_name)

        def keys(self):
           import glob
           attrs = [os.path.basename(a)[len('_attr_'):] for a in glob.glob(os.path.join(self._dir,'_attr_*'))]
           return attrs

        def dict(self):
           keys = {}
           for a in self.keys():
              keys[a] = self[a]
           return keys

        def __str__(self):
           return repr(self.dict())
                
    all_procs = []

    @staticmethod
    def supervisor(steps):
        import time
        #print "SU",steps
        #print 'SU',[s for s in steps]

        if _smash_.DEBUG:
            log('start',_smash_.supervisor_step.value,_smash_.steps)

        while _smash_.supervisor_step.value < _smash_.N_STEPS-1:
            while 1:
                time.sleep(0.01)
                #print [s for s in steps]
                passed = all([_smash_.steps[i]>_smash_.supervisor_step.value for i in range(len(_smash_.steps))])
                #print 'passed',supervisor_step.value,passed
                if passed:
                    break

            #print "supervisor step completed:",supervisor_step.value,steps

            _smash_.supervisor_step.value += 1

        if _smash_.DEBUG:
            log('stop',_smash_.supervisor_step.value,_smash_.steps)

    @staticmethod
    def _step(i,wi,message):
        import time
        _smash_.steps[wi] = i

        def supervisor_status():
            return "(supervisor_step="+str(_smash_.supervisor_step.value)+" worker_steps="+str(_smash_.steps)+")"

        if _smash_.DEBUG:
            logger.debug('step %d waiting (wi=%d)'%(i,wi,supervisor_status()))
        while _smash_.supervisor_step.value<i:
            time.sleep(0.01)

        if _smash_.DEBUG:
            logger.debug('step %d entered (wi=%d)'%(i,wi,supervisor_status()))

        if message is not None:
            sep='*'*80
            logger.info( 'entering new step \n'+sep+'\n'+'(%d) %s:  %s\n'%(i,_smash_.process_name,message.upper())+sep)

    @staticmethod
    def worker_wrap(wi,f,fname):
        if fname is None:
            fname = f.__name__
        _smash_.process_name=fname
        _smash_.process_number = wi
        def step(i,message=""):
            _smash_._step(i,wi,message)
        try:
            try:
                f(step)
            except Exception,x:
                import traceback
                logger.fatal("Exception occured: %s \n %s", x,traceback.format_exc())
                import sys
                sys.exit(1)
        finally:
            # worker finish
            step(_smash_.N_STEPS-1,None) # don't print any message

            import smashbox.utilities
            if smashbox.utilities.reported_errors:
               logger.error('%s error(s) reported',len(smashbox.utilities.reported_errors))
                  

    @staticmethod
    def run():
        """ Lunch worker processes and the supervisor loop. Block until all is finished.
        """
        from multiprocessing import Process, Manager

        manager = Manager()

        _smash_.shared_object = _smash_.SmashSharedObject(os.path.join(config.rundir,'_shared_objects'))<commit_msg>Worker must return error code when errors detected<commit_after>#!/usr/bin/env python
# -*- python -*-
#
# The _open_SmashBox Project.
#
# Author: <NAME>, CERN, 2013
# License: AGPL
#
#$Id: $
#
# %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
# Perform internal setup of the environment.
# This is a Copy/Paste logic which must stay in THIS file
def standardSetup():
   import sys, os.path
   # insert the path to cernafs based on the relative position of this scrip inside the service directory tree
   exeDir = os.path.abspath(os.path.normpath(os.path.dirname(os.path.dirname(sys.argv[0]))))
   pythonDir = os.path.join(os.path.dirname(exeDir), 'python' )
   sys.path.insert(0, pythonDir)
   import smashbox.setup
   smashbox.setup.standardSetup(sys.argv[0]) # execute a setup hook

standardSetup()
del standardSetup
# %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

# obsolete to be removed
def log(*args,**kwds):
    import time
    print time.ctime(),_smash_.process_name,(" ".join([str(s) for s in args]))%kwds


#import shelve

class _smash_:
    """ Internals of the stepper synchronization framework. This class
    is merely a namespace to avoid polluting global namespace of
    user's test scripts.
    """

    # all the rest of attributes are internal -- accessors for process_name and common_dict are defined in utilities

    process_name = None
    process_number = 0
    
    DEBUG = False

    # this is a hardcoded maximum number of steps
    N_STEPS = 100
    
    workers = []

    class SmashSharedObject:
        """ A bunch of shared attributes stored in a directory as separate files.
        """
        
        def __init__(self,d):
           self._dir = d
           self._makedir()

        def _makedir(self):
           try:
              os.makedirs(self._dir)
           except OSError,x:
              import errno
              if x.errno != errno.EEXIST:
                 raise
                       
        def __getitem__(self,key):
           self._makedir()
           import pickle
           try:
              return pickle.load(file(os.path.join(self._dir,'_attr_'+key)))
           except Exception,x:
              logger.debug(x)
              raise AttributeError(x)
        
        def __setitem__(self,key,val):
           self._makedir()
           import pickle
           import time
           tmp_name = os.path.join(self._dir,'tmp.%s.%s._attr_%s'%(os.getpid(),time.time(),key))
           dest_name = os.path.join(self._dir,'_attr_'+key)
                           
           pickle.dump(val,file(tmp_name,'w'))

           import shutil
           shutil.move(tmp_name,dest_name)

        def keys(self):
           import glob
           attrs = [os.path.basename(a)[len('_attr_'):] for a in glob.glob(os.path.join(self._dir,'_attr_*'))]
           return attrs

        def dict(self):
           keys = {}
           for a in self.keys():
              keys[a] = self[a]
           return keys

        def __str__(self):
           return repr(self.dict())
                
    all_procs = []

    @staticmethod
    def supervisor(steps):
        import time
        #print "SU",steps
        #print 'SU',[s for s in steps]

        if _smash_.DEBUG:
            log('start',_smash_.supervisor_step.value,_smash_.steps)

        while _smash_.supervisor_step.value < _smash_.N_STEPS-1:
            while 1:
                time.sleep(0.01)
                #print [s for s in steps]
                passed = all([_smash_.steps[i]>_smash_.supervisor_step.value for i in range(len(_smash_.steps))])
                #print 'passed',supervisor_step.value,passed
                if passed:
                    break

            #print "supervisor step completed:",supervisor_step.value,steps

            _smash_.supervisor_step.value += 1

        if _smash_.DEBUG:
            log('stop',_smash_.supervisor_step.value,_smash_.steps)

    @staticmethod
    def _step(i,wi,message):
        import time
        _smash_.steps[wi] = i

        def supervisor_status():
            return "(supervisor_step="+str(_smash_.supervisor_step.value)+" worker_steps="+str(_smash_.steps)+")"

        if _smash_.DEBUG:
            logger.debug('step %d waiting (wi=%d)'%(i,wi,supervisor_status()))
        while _smash_.supervisor_step.value<i:
            time.sleep(0.01)

        if _smash_.DEBUG:
            logger.debug('step %d entered (wi=%d)'%(i,wi,supervisor_status()))

        if message is not None:
            sep='*'*80
            logger.info( 'entering new step \n'+sep+'\n'+'(%d) %s:  %s\n'%(i,_smash_.process_name,message.upper())+sep)

    @staticmethod
    def worker_wrap(wi,f,fname):
        if fname is None:
            fname = f.__name__
        _smash_.process_name=fname
        _smash_.process_number = wi
        def step(i,message=""):
            _smash_._step(i,wi,message)
        try:
            try:
                f(step)
            except Exception,x:
                import traceback
                logger.fatal("Exception occured: %s \n %s", x,traceback.format_exc())
                import sys
                sys.exit(1)
        finally:
            # worker finish
            step(_smash_.N_STEPS-1,None) # don't print any message

            import smashbox.utilities
            if smashbox.utilities.reported_errors:
               logger.error('%s error(s) reported',len(smashbox.utilities.reported_errors))
               import sys
               sys.exit(2)
                  

    @staticmethod
    def run():
        """ Lunch worker processes and the supervisor loop. Block until all is finished.
        """
        from multiprocessing import Process, Manager

        manager = Manager()

        _smash_.shared_object = _smash_.SmashSharedObject(os.path.join(config.rundir,'_shared_objects'))
{'fraction_non_alphanumeric': 0.12739678700984627, 'fraction_numerical': 0.004232164449818622, 'mean_word_length': 2.021659707724426, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2966727', 'n_tokens_mistral': 3536, 'n_tokens_neox': 3355, 'n_words': 785}