<commit_before>import os
import sys
import time
import datetime
import json
import copy
import threading
import multiprocessing
import socket
import importlib
import traceback
import signal

from pandacommon.pandalogger import logger_utils
from pandaserver.config import panda_config, daemon_config


# list of signals accepted to end the main process
END_SIGNALS = [
        signal.SIGINT,
        signal.SIGHUP,
        signal.SIGTERM,
    ]

# mandatory attributes and thier type of daemon
MANDATORY_ATTRS = [
        ('module', str),
        ('period', int),
        ('arguments', list),
    ]

# command to send in pipe to stop daemon worker processes
CMD_STOP = '__STOP'

# epoch datetime
EPOCH = datetime.datetime.fromtimestamp(0)


# worker process loop of daemon
def daemon_loop(dem_config, msg_queue, pipe_conn):
    # pid of the worker
    my_pid = os.getpid()
    my_full_pid = '{0}-{1}-{2}'.format(socket.getfqdn().split('.')[0], os.getpgrp(), my_pid)
    # logger to log in file
    base_logger = logger_utils.setup_logger('daemons')
    tmp_log = logger_utils.make_logger(base_logger, 'worker_pid={pid}'.format(pid=my_pid))
    tmp_log.info('daemon worker start')
    # signal handler
    def got_end_sig(sig, frame):
        tmp_log.warning('(got signal {sig})'.format(sig=sig))
    for sig in END_SIGNALS:
        signal.signal(sig, got_end_sig)
    # dict of all daemons and their script module object
    module_map = {}
    # package of daemon scripts
    mod_package = getattr(daemon_config, 'package')
    # initialize cx_Oracle using dummy connection
    try:
        from pandaserver.taskbuffer.Initializer import initializer
        initializer.init()
    except Exception as e:
        tmp_log.error('failed to launch initializer with {err} ; terminated'.format(
                            err='{0}: {1}'.format(e.__class__.__name__, e)))
        return
    # taskBuffer object
    try:
        from pandaserver.taskbuffer.TaskBuffer import taskBuffer as tbuf
        tbuf.init(panda_config.dbhost, panda_config.dbpasswd, nDBConnection=1)
        tmp_log.debug('taskBuffer initialized')
    except Exception as e:
        tmp_log.error('failed to initialize taskBuffer with {err} ; terminated'.format(
                            err='{0}: {1}'.format(e.__class__.__name__, e)))
        return
    # import module of all daemons
    for dem_name, attrs in dem_config.items():
        mod_name = attrs['module']
        try:
            the_module = importlib.import_module('.{mod}'.format(mod=mod_name), mod_package)
            module_map[dem_name] = the_module
        except Exception as e:
            tmp_log.warning('for daemon {dem}, failed to import {mod} with {err} ; skipped it'.format(
                                dem=dem_name, mod=mod_name, err='{0}: {1}'.format(e.__class__.__name__, e)))
        else:
            module_map[dem_name] = the_module
    tmp_log.debug('initialized, running')
    # loop
    while True:
        # get command from pipe
        if pipe_conn.poll():
            cmd = pipe_conn.recv()
            if cmd == CMD_STOP:
                # got stop command, stop the process
                tmp_log.debug('got stop command, stop this process')
                break
            else:
                tmp_log.debug('got invalid command "{cmd}" ; skipped it'.format(cmd=cmd))
        # get a message from queue
        tmp_log.debug('waiting for message...')
        one_msg = msg_queue.get()
        # process message
        if one_msg in module_map:
            # got a daemon name, get the module object and corresponding attributes
            dem_name = one_msg
            tmp_log.debug('got message of {dem}'.format(dem=dem_name))
            the_module = module_map[dem_name]
            attrs = dem_config[dem_name]
            mod_args = attrs['arguments']
            mod_argv = tuple([__file__] + mod_args)
            dem_period = attrs['period']
            dem_period_in_minute = dem_period/60.
            is_sync = attrs['sync']
            # initialize variables
            to_run_daemon = False
            has_run = False
            last_run_start_ts = 0
            last_run_end_ts = 0
            # component name in lock table
            component = 'pandaD.{dem}'.format(dem=dem_name)
            # whether the daemon shoule be synchronized among nodes
            if is_sync:
                # sychronized daemon, check process lock in DB
                ret_val, locked_time = tbuf.checkProcessLock_PANDA(component=component, pid=my_full_pid, time_limit=dem_period_in_minute)
                if ret_val:
                    # locked by some process on other nodes
                    last_run_start_ts = int((locked_time - EPOCH).total_seconds())
                else:
                    # try to get the lock
                    got_lock = tbuf.lockProcess_PANDA(component=component, pid=my_full_pid, time_limit=dem_period_in_minute)
                    if got_lock:
                        # got the lock
                        to_run_daemon = True
                    else:
                        # did not get lock, skip
                        pass
            else:
                to_run_daemon = True
            # run daemon
            if to_run_daemon:
                last_run_start_ts = int(time.time())
                try:
                    # execute the module script with arguments
                    tmp_log.info('{dem} start'.format(dem=dem_name))
                    the_module.main(argv=mod_argv, tbuf=tbuf)
                    tmp_log.info('{dem} finish'.format(dem=dem_name))
                except Exception as e:
                    tb = traceback.format_exc()
                    tmp_log.error('failed to run daemon {dem} with {err} ; skipped it'.format(
                                    dem=dem_name, err='{0}: {1}\n{2}\n'.format(e.__class__.__name__, e, tb)))
                # daemon has run
                last_run_end_ts = int(time.time())
                has_run = True
            # send daemon status back to master
            status_tuple = (dem_name, has_run, last_run_start_ts, last_run_end_ts)
            pipe_conn.send(status_tuple)
        else:
            # got invalid message
            tmp_log.error('got invalid message "{msg}", skipped it'.format(msg=one_msg))
        # sleep
        time.sleep(2**-5)


# worker class of daemon process for PanDA server
class DaemonWorker(object):

    __slots__ = (
            'wid',
            'parent_conn',<commit_msg>panda daemon: spawn new worker when exception<commit_after>import os
import sys
import time
import datetime
import json
import copy
import threading
import multiprocessing
import socket
import importlib
import traceback
import signal

from pandacommon.pandalogger import logger_utils
from pandaserver.config import panda_config, daemon_config


# list of signals accepted to end the main process
END_SIGNALS = [
        signal.SIGINT,
        signal.SIGHUP,
        signal.SIGTERM,
    ]

# mandatory attributes and thier type of daemon
MANDATORY_ATTRS = [
        ('module', str),
        ('period', int),
        ('arguments', list),
    ]

# command to send in pipe to stop daemon worker processes
CMD_STOP = '__STOP'

# epoch datetime
EPOCH = datetime.datetime.fromtimestamp(0)


# worker process loop of daemon
def daemon_loop(dem_config, msg_queue, pipe_conn):
    # pid of the worker
    my_pid = os.getpid()
    my_full_pid = '{0}-{1}-{2}'.format(socket.getfqdn().split('.')[0], os.getpgrp(), my_pid)
    # logger to log in file
    base_logger = logger_utils.setup_logger('daemons')
    tmp_log = logger_utils.make_logger(base_logger, 'worker_pid={pid}'.format(pid=my_pid))
    tmp_log.info('daemon worker start')
    # signal handler
    def got_end_sig(sig, frame):
        tmp_log.warning('(got signal {sig})'.format(sig=sig))
    for sig in END_SIGNALS:
        signal.signal(sig, got_end_sig)
    # dict of all daemons and their script module object
    module_map = {}
    # package of daemon scripts
    mod_package = getattr(daemon_config, 'package')
    # initialize cx_Oracle using dummy connection
    try:
        from pandaserver.taskbuffer.Initializer import initializer
        initializer.init()
    except Exception as e:
        tmp_log.error('failed to launch initializer with {err} ; terminated'.format(
                            err='{0}: {1}'.format(e.__class__.__name__, e)))
        return
    # taskBuffer object
    try:
        from pandaserver.taskbuffer.TaskBuffer import taskBuffer as tbuf
        tbuf.init(panda_config.dbhost, panda_config.dbpasswd, nDBConnection=1)
        tmp_log.debug('taskBuffer initialized')
    except Exception as e:
        tmp_log.error('failed to initialize taskBuffer with {err} ; terminated'.format(
                            err='{0}: {1}'.format(e.__class__.__name__, e)))
        return
    # import module of all daemons
    for dem_name, attrs in dem_config.items():
        mod_name = attrs['module']
        try:
            the_module = importlib.import_module('.{mod}'.format(mod=mod_name), mod_package)
            module_map[dem_name] = the_module
        except Exception as e:
            tmp_log.warning('for daemon {dem}, failed to import {mod} with {err} ; skipped it'.format(
                                dem=dem_name, mod=mod_name, err='{0}: {1}'.format(e.__class__.__name__, e)))
        else:
            module_map[dem_name] = the_module
    tmp_log.debug('initialized, running')
    # loop
    while True:
        # get command from pipe
        if pipe_conn.poll():
            cmd = pipe_conn.recv()
            if cmd == CMD_STOP:
                # got stop command, stop the process
                tmp_log.info('got stop command, stop this worker')
                break
            else:
                tmp_log.debug('got invalid command "{cmd}" ; skipped it'.format(cmd=cmd))
        # get a message from queue
        tmp_log.debug('waiting for message...')
        one_msg = msg_queue.get()
        # process message
        if one_msg in module_map:
            # got a daemon name, get the module object and corresponding attributes
            dem_name = one_msg
            tmp_log.debug('got message of {dem}'.format(dem=dem_name))
            the_module = module_map[dem_name]
            attrs = dem_config[dem_name]
            mod_args = attrs['arguments']
            mod_argv = tuple([__file__] + mod_args)
            dem_period = attrs['period']
            dem_period_in_minute = dem_period/60.
            is_sync = attrs['sync']
            # initialize variables
            to_run_daemon = False
            has_run = False
            last_run_start_ts = 0
            last_run_end_ts = 0
            # component name in lock table
            component = 'pandaD.{dem}'.format(dem=dem_name)
            # whether the daemon shoule be synchronized among nodes
            if is_sync:
                # sychronized daemon, check process lock in DB
                ret_val, locked_time = tbuf.checkProcessLock_PANDA(component=component, pid=my_full_pid, time_limit=dem_period_in_minute)
                if ret_val:
                    # locked by some process on other nodes
                    last_run_start_ts = int((locked_time - EPOCH).total_seconds())
                else:
                    # try to get the lock
                    got_lock = tbuf.lockProcess_PANDA(component=component, pid=my_full_pid, time_limit=dem_period_in_minute)
                    if got_lock:
                        # got the lock
                        to_run_daemon = True
                    else:
                        # did not get lock, skip
                        pass
            else:
                to_run_daemon = True
            # run daemon
            if to_run_daemon:
                last_run_start_ts = int(time.time())
                try:
                    # execute the module script with arguments
                    tmp_log.info('{dem} start'.format(dem=dem_name))
                    the_module.main(argv=mod_argv, tbuf=tbuf)
                    tmp_log.info('{dem} finish'.format(dem=dem_name))
                except Exception as e:
                    tb = traceback.format_exc()
                    tmp_log.error('failed to run daemon {dem} with {err} ; stop this worker'.format(
                                    dem=dem_name, err='{0}: {1}\n{2}\n'.format(e.__class__.__name__, e, tb)))
                    break
                # daemon has run
                last_run_end_ts = int(time.time())
                has_run = True
            # send daemon status back to master
            status_tuple = (dem_name, has_run, last_run_start_ts, last_run_end_ts)
            pipe_conn.send(status_tuple)
        else:
            # got invalid message
            tmp_log.warning('got invalid message "{msg}", skipped it'.format(msg=one_msg))
        # sleep
        time.sleep(2**-5)


# worker class of daemon process for PanDA server
class DaemonWorker(object):

    __slots__ = (
            'wid',
            'parent_conn',
{'fraction_non_alphanumeric': 0.08943027340766764, 'fraction_numerical': 0.0032075759890025967, 'mean_word_length': 2.009652953344059, 'pattern_counts': {'":': 0, '<': 3, '<?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': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '638378', 'n_tokens_mistral': 3780, 'n_tokens_neox': 3701, 'n_words': 967}