<commit_before># This module used to create multi worker processes shipped with a manager
# And it's only a framework which will manager the process create, quit, etc.
# It'll be used in daemon.

from __future__ import print_function, absolute_import

__all__ = ['Worker', 'Manager', 'make_log']
__version__ = '0.1'

import os
import sys
import time
import signal
import logging
from .process import pid_exists, wait_pid

default_log = logging.getLogger(__name__)

is_exit = False

class TimeoutException(Exception):
    pass

class Timeout():
    """Timeout class using ALARM signal."""

    def __init__(self, sec):
        self.sec = sec

    def __enter__(self):
        signal.signal(signal.SIGALRM, self.raise_timeout)
        signal.alarm(self.sec)

    def __exit__(self, *args):
        signal.alarm(0)    # disable alarm

    def raise_timeout(self, *args):
        raise TimeoutException("Timeout {}s".format(self.sec))

def get_memory(pid):
    # return the memory usage in MB, psutil should be 4.0 version
    from psutil import Process, __version__

    # if __version__ < '4.0.0':
    #     raise Exception('psutil module should be 4.0.0 version at least.')

    if pid_exists(pid):
        process = Process(pid)
        # mem = process.memory_full_info().uss / float(1024*1024)
        mem = process.memory_info().rss / float(1024*1024)
        return mem
    return 0

FORMAT = "[%(levelname)s - %(asctime)s - %(filename)s:%(lineno)s] - %(message)s"

def make_log(log, log_filename, format=FORMAT, datafmt=None, max_bytes=1024*1024*50,
             backup_count=5):
    import logging.handlers

    if isinstance(log, (str, unicode)):
        log = logging.getLogger(log)

    handler = logging.handlers.RotatingFileHandler(
        log_filename, maxBytes=max_bytes, backupCount=backup_count)
    fmt = logging.Formatter(format, datafmt)
    handler.setFormatter(fmt)
    log.addHandler(handler)
    return log

class Worker(object):
    _id = 1

    def __init__(self, log=None,
                 max_requests=None,
                 soft_memory_limit=200, #memory limit MB
                 hard_memory_limit=300, #memory limit MB
                 timeout=None,
                 check_point=None,
                 name=None,
                 args=None, kwargs=None):
        self.log = log or default_log
        self.max_requests = max_requests or sys.maxint
        self.soft_memory_limit = soft_memory_limit
        self.hard_memory_limit = hard_memory_limit
        self.timeout = timeout
        self.args = args or ()
        self.kwargs = kwargs or {}
        self.is_exit = None
        self.count = 0
        self.check_point = check_point
        self.name = "%s-%d" % ((name or 'Process'), self._id)
        self.__class__._id += 1

    def start(self):
        self.pid = os.getpid()

        signal.signal(signal.SIGINT, self.signal_handler)
        signal.signal(signal.SIGTERM, self.signal_handler)
        signal.signal(signal.SIGUSR1, self.signal_handler_usr1)
        signal.signal(signal.SIGUSR2, self.signal_handler_usr2)

        self.init()
        try:
            self._run()
            self.on_finished()
        except Exception as e:
            self.log.exception(e)
            self.on_exception(e)

        self.after_run()

    def init(self):
        self.log.info('%s %d created' % (self.name, self.pid))

    def run(self):
        self.log.info('%s %d running' % (self.name, self.pid))
        time.sleep(1)
        return True

    def on_exception(self, e):
        pass

    def on_finished(self):
        pass

    def _run(self):
        while (not self.max_requests or
                   (self.max_requests and self.count <= self.max_requests)) and \
                not self.is_exit:
            try:
                if self.timeout:
                    with Timeout(self.timeout):
                        ret = self.run()
                else:
                    ret = self.run()
            except TimeoutException as e:
                self.log.info('Time out')
            except Exception as e:
                self.log.exception(e)
                self.is_exit = True
            finally:
                # !important
                # count shoud be calculated by child class
                # self.count += 1<commit_msg>Improve workers enable child class overwrite _run and rewrite timeout process<commit_after># This module used to create multi worker processes shipped with a manager
# And it's only a framework which will manager the process create, quit, etc.
# It'll be used in daemon.

from __future__ import print_function, absolute_import

__all__ = ['Worker', 'Manager', 'make_log', 'Timeout']
__version__ = '0.1'

import os
import sys
import time
import signal
import logging
from .process import pid_exists, wait_pid

default_log = logging.getLogger(__name__)

is_exit = False

class TimeoutException(Exception):
    pass

class Timeout():
    """Timeout class using ALARM signal."""

    def __init__(self, sec):
        self.sec = sec

    def __enter__(self):
        signal.signal(signal.SIGALRM, self.raise_timeout)
        signal.alarm(self.sec)

    def __exit__(self, *args):
        signal.alarm(0)    # disable alarm

    def raise_timeout(self, *args):
        raise TimeoutException("Timeout {}s".format(self.sec))

def get_memory(pid):
    # return the memory usage in MB, psutil should be 4.0 version
    from psutil import Process, __version__

    # if __version__ < '4.0.0':
    #     raise Exception('psutil module should be 4.0.0 version at least.')

    if pid_exists(pid):
        process = Process(pid)
        # mem = process.memory_full_info().uss / float(1024*1024)
        mem = process.memory_info().rss / float(1024*1024)
        return mem
    return 0

FORMAT = "[%(levelname)s - %(asctime)s - %(filename)s:%(lineno)s] - %(message)s"

def make_log(log, log_filename, format=FORMAT, datafmt=None, max_bytes=1024*1024*50,
             backup_count=5):
    import logging.handlers

    if isinstance(log, (str, unicode)):
        log = logging.getLogger(log)

    handler = logging.handlers.RotatingFileHandler(
        log_filename, maxBytes=max_bytes, backupCount=backup_count)
    fmt = logging.Formatter(format, datafmt)
    handler.setFormatter(fmt)
    log.addHandler(handler)
    return log

class Worker(object):
    _id = 1

    def __init__(self, log=None,
                 max_requests=None,
                 soft_memory_limit=200, #memory limit MB
                 hard_memory_limit=300, #memory limit MB
                 timeout=None,
                 check_point=None,
                 name=None,
                 args=None, kwargs=None):
        self.log = log or default_log
        self.max_requests = max_requests or sys.maxint
        self.soft_memory_limit = soft_memory_limit
        self.hard_memory_limit = hard_memory_limit
        self.timeout = timeout
        self.args = args or ()
        self.kwargs = kwargs or {}
        self.is_exit = None
        self.count = 0
        self.check_point = check_point
        self.name = "%s-%d" % ((name or 'Process'), self._id)
        self.__class__._id += 1

    def start(self):
        self.pid = os.getpid()

        signal.signal(signal.SIGINT, self.signal_handler)
        signal.signal(signal.SIGTERM, self.signal_handler)
        signal.signal(signal.SIGUSR1, self.signal_handler_usr1)
        signal.signal(signal.SIGUSR2, self.signal_handler_usr2)

        self.init()
        try:
            self.loop()
            self.on_finished()
        except Exception as e:
            self.log.exception(e)
            self.on_exception(e)

        self.after_run()

    def init(self):
        self.log.info('%s %d created' % (self.name, self.pid))

    def _run(self):
        if self.timeout:
            with Timeout(self.timeout):
                ret = self.run()
        else:
            ret = self.run()
        return ret

    def run(self):
        self.log.info('%s %d running' % (self.name, self.pid))
        time.sleep(1)
        return True

    def on_exception(self, e):
        pass

    def on_finished(self):
        pass

    def loop(self):
        while (not self.max_requests or
                   (self.max_requests and self.count <= self.max_requests)) and \
                not self.is_exit:
            try:
                self._run()
            except TimeoutException as e:
                self.log.info('Time out')
            except Exception as e:
                self.log.exception(e)
                self.is_exit = True
            finally:
                # !important
                # count shoud be calculated by child class
                # self.count += 1
{'fraction_non_alphanumeric': 0.10033406289597972, 'fraction_numerical': 0.012440963022693237, 'mean_word_length': 2.137694253704373, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 3, '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': '6780050', 'n_tokens_mistral': 2650, 'n_tokens_neox': 2524, 'n_words': 651}