<commit_before>import json
import logging
import os
import os.path
import signal
from subprocess import call
from time import sleep
import traceback

import boto.swf.layer2 as swf
import daemon
import luigi
import luigi.configuration

from .util import default_log_format, get_class, kill_from_pid_file, \
    SingleWaitingLockPidFile, dt_from_iso


logger = logging.getLogger(__name__)


seconds = 1.


class LuigiSwfWorker(swf.ActivityWorker):
    """Implementation of boto's SWF Activity Worker

    See :class:`WorkerServer` for daemonizing this.
    """

    def run(self, identity=None):
        """Poll for and run an activity task

        This should be run in a loop. It will poll for up to 60 seconds. After
        60 seconds, it will return without running any activity tasks. The user
        should usually not need to interact with this class directly. Instead,
        :class:`WorkerServer` can be used to run the loop.

        :return: None
        """
        activity_task = self.poll(identity=identity)
        if 'activityId' not in activity_task:
            logger.debug('LuigiSwfWorker().run(), poll timed out')
            return
        task = None
        try:
            logger.info('LuigiSwfWorker().run(), %s, processing',
                        activity_task['activityId'])
            input_task = json.loads(activity_task.get('input'))
            input_params = json.loads(input_task['params'])
            task_cls = get_class(*input_task['class'])
            task_params = task_cls.get_params()
            kwargs = dict()
            for param_name, param_cls in task_params:
                if param_name == 'pool':
                    continue
                if isinstance(param_cls, luigi.DateParameter):
                    kwargs[param_name] = dt_from_iso(input_params[param_name])
                else:
                    kwargs[param_name] = input_params[param_name]
            task = task_cls(**kwargs)
            if hasattr(task, 'register_activity_worker'):
                task.register_activity_worker(self, activity_task)
            if task.complete():
                result = 'Did not run (task.complete() returned true)'
                logger.debug('LuigiSwfWorker().run(), %s, %s', result)
                self.complete(result=result)
                return
            task.run()
            if not getattr(task, 'cancel_acked', False):
                task.on_success()
                task_completed = task.complete()
            else:
                task_completed = False
        except Exception as error:
            tb = traceback.format_exc()
            if task is not None:
                try:
                    message = task.on_failure(error)
                except:
                    message = ('on_failure() failed: \n' +
                               traceback.format_exc())
            if message is None:
                message = '(no message)'
            logger.error('LuigiSwfWorker().run(), %s, error:\n%s',
                         activity_task['activityId'], tb)
            details = (tb + '\n\n\n' + str(error) + '\n\n\non_failure():\n' +
                       message)
            self.fail(reason=str(error)[:255], details=details[:32767])
            raise
        if task_completed:
            self.complete()
            logger.info('LuigiSwfWorker().run(), completed %s',
                        activity_task['activityId'])
        elif not getattr(task, 'cancel_acked', False):
            reason = 'complete() returned false after running'
            logger.error('LuigiSwfWorker().run(), %s, failed (%s)',
                         activity_task['activityId'], reason)
            self.fail(reason=reason)<commit_msg>fix worker failure handling: message variable<commit_after>import json
import logging
import os
import os.path
import signal
from subprocess import call
from time import sleep
import traceback

import boto.swf.layer2 as swf
import daemon
import luigi
import luigi.configuration

from .util import default_log_format, get_class, kill_from_pid_file, \
    SingleWaitingLockPidFile, dt_from_iso


logger = logging.getLogger(__name__)


seconds = 1.


class LuigiSwfWorker(swf.ActivityWorker):
    """Implementation of boto's SWF Activity Worker

    See :class:`WorkerServer` for daemonizing this.
    """

    def run(self, identity=None):
        """Poll for and run an activity task

        This should be run in a loop. It will poll for up to 60 seconds. After
        60 seconds, it will return without running any activity tasks. The user
        should usually not need to interact with this class directly. Instead,
        :class:`WorkerServer` can be used to run the loop.

        :return: None
        """
        activity_task = self.poll(identity=identity)
        if 'activityId' not in activity_task:
            logger.debug('LuigiSwfWorker().run(), poll timed out')
            return
        task = None
        try:
            logger.info('LuigiSwfWorker().run(), %s, processing',
                        activity_task['activityId'])
            input_task = json.loads(activity_task.get('input'))
            input_params = json.loads(input_task['params'])
            task_cls = get_class(*input_task['class'])
            task_params = task_cls.get_params()
            kwargs = dict()
            for param_name, param_cls in task_params:
                if param_name == 'pool':
                    continue
                if isinstance(param_cls, luigi.DateParameter):
                    kwargs[param_name] = dt_from_iso(input_params[param_name])
                else:
                    kwargs[param_name] = input_params[param_name]
            task = task_cls(**kwargs)
            if hasattr(task, 'register_activity_worker'):
                task.register_activity_worker(self, activity_task)
            if task.complete():
                result = 'Did not run (task.complete() returned true)'
                logger.debug('LuigiSwfWorker().run(), %s, %s', result)
                self.complete(result=result)
                return
            task.run()
            if not getattr(task, 'cancel_acked', False):
                task.on_success()
                task_completed = task.complete()
            else:
                task_completed = False
        except Exception as error:
            tb = traceback.format_exc()
            message = None
            if task is not None:
                try:
                    message = task.on_failure(error)
                except:
                    message = ('on_failure() failed: \n' +
                               traceback.format_exc())
            if message is None:
                message = '(no message)'
            logger.error('LuigiSwfWorker().run(), %s, error:\n%s',
                         activity_task['activityId'], tb)
            details = (tb + '\n\n\n' + str(error) + '\n\n\non_failure():\n' +
                       message)
            self.fail(reason=str(error)[:255], details=details[:32767])
            raise
        if task_completed:
            self.complete()
            logger.info('LuigiSwfWorker().run(), completed %s',
                        activity_task['activityId'])
        elif not getattr(task, 'cancel_acked', False):
            reason = 'complete() returned false after running'
            logger.error('LuigiSwfWorker().run(), %s, failed (%s)',
                         activity_task['activityId'], reason)
            self.fail(reason=reason)
{'fraction_non_alphanumeric': 0.09844352800319277, 'fraction_numerical': 0.003724890248769456, 'mean_word_length': 1.894878706199461, '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': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4583212', 'n_tokens_mistral': 2012, 'n_tokens_neox': 1922, 'n_words': 526}