<commit_before>#!/usr/bin/env python
""" executor is the process that executes bashtasks
"""
import signal
import argparse
import subprocess
import sys
import json
import time
import os
import threading
from socket import gethostname
from time import sleep
import logging
import pika

from bashtasks.rabbit_util import connect_and_declare, close_channel_and_conn
from bashtasks.constants import DestinationNames, TASK_REQUESTS_POOL, TASK_RESPONSES_POOL
from bashtasks.logger import get_logger

channels = []  # stores all executor thread channels.
stop = False  # False until the executor is asked to stop
MB_10 = 10485760

DEFAULT_DESTINATION = DestinationNames.get_for(TASK_REQUESTS_POOL)


def curr_module_name():
    return os.path.splitext(os.path.basename(__file__))[0]


def currtimemillis():
    return int(round(time.time() * 1000))


def get_executor_name():
    return ":".join((gethostname(), str(os.getpid()), threading.current_thread().name))


def get_thread_name(worker):
    return 'worker_th_' + str(worker)


def start_executors(workers=1, host='127.0.0.1', usr='guest', pas='guest', queue=DEFAULT_DESTINATION,
                    tasks_nr=1, max_retries=0, verbose=False, custom_callback=None):
    worker_ths = []
    for worker in range(0, workers):
        worker_th = threading.Thread(target=start_executor,
                                     kwargs=({'host': host, 'usr': usr, 'pas': pas,
                                              'queue': queue, 'tasks_nr': tasks_nr,
                                              'max_retries': max_retries, 'verbose': verbose,
                                              'custom_callback':custom_callback}),
                                     name=get_thread_name(worker))
        worker_th.daemon = True

        worker_th.start()
        worker_ths.append(worker_th)

    while not stop:
        sleep(1)


def start_executor(host='127.0.0.1', usr='guest', pas='<PASSWORD>', queue=DEFAULT_DESTINATION,
                   tasks_nr=1, max_retries=0, verbose=False, custom_callback=None):
    curr_th_name = threading.current_thread().name
    logger = get_logger(name=curr_module_name())
    logger.info(">> Starting executor %s connecting to rabbitmq: %s:%s@%s for executing %d tasks.",
                curr_th_name, usr, pas, host, tasks_nr)

    ch = connect_and_declare(host=host, usr=usr, pas=pas, destinations=queue)
    ch.basic_qos(prefetch_count=1)  # consume msgs one at a time

    channels.append(ch)

    def create_response_for(msg):
        resp = {}
        resp.update(msg)
        resp['executor_name'] = get_executor_name()
        resp['retries'] = msg.get('retries', 0)
        resp['max_retries'] = msg.get('max_retries', 0)
        return resp

    def should_retry(response_msg):
        is_error = response_msg['returncode'] != 0
        returncode = response_msg['returncode']
        is_retriable =  returncode not in response_msg['non_retriable']
        current_retries = response_msg.get('retries', 0)
        msg_max_retries = response_msg.get('max_retries', max_retries)
        retries_pending = current_retries < msg_max_retries
        return is_error and is_retriable and retries_pending

    def send_response(response_msg):
        if should_retry(response_msg):
            logger.debug('---- retrying msg correlation_id: %d current_retries: %d of %d',
                         response_msg['correlation_id'], response_msg['retries'],
                         response_msg['max_retries'])
            tgt_exch = TASK_REQUESTS_POOL
            response_msg['retries'] += 1
        else:
            tgt_exch = TASK_RESPONSES_POOL

        response_str = json.dumps(response_msg)
        props = pika.BasicProperties(
                             delivery_mode = 2, # make message persistent
                          )
        ch.basic_publish(exchange=tgt_exch, routing_key='', body=response_str, properties=props)

    def tasks_nr_generator(tasks_nr):
        tasks_nr_gen = tasks_nr<commit_msg>executor_impl: make pep8 happier<commit_after>#!/usr/bin/env python
""" executor is the process that executes bashtasks
"""
import signal
import argparse
import subprocess
import sys
import json
import time
import os
import threading
from socket import gethostname
from time import sleep
import logging
import pika

from bashtasks.rabbit_util import connect_and_declare, close_channel_and_conn
from bashtasks.constants import DestinationNames, TASK_REQUESTS_POOL, TASK_RESPONSES_POOL
from bashtasks.logger import get_logger

channels = []  # stores all executor thread channels.
stop = False  # False until the executor is asked to stop
MB_10 = 10485760

DEFAULT_DESTINATION = DestinationNames.get_for(TASK_REQUESTS_POOL)


def curr_module_name():
    return os.path.splitext(os.path.basename(__file__))[0]


def currtimemillis():
    return int(round(time.time() * 1000))


def get_executor_name():
    return ":".join((gethostname(), str(os.getpid()), threading.current_thread().name))


def get_thread_name(worker):
    return 'worker_th_' + str(worker)


def start_executors(workers=1, host='127.0.0.1', usr='guest', pas='guest',
                    queue=DEFAULT_DESTINATION, tasks_nr=1, max_retries=0, verbose=False,
                    custom_callback=None):
    worker_ths = []
    for worker in range(0, workers):
        worker_th = threading.Thread(target=start_executor,
                                     kwargs=({'host': host, 'usr': usr, 'pas': pas,
                                              'queue': queue, 'tasks_nr': tasks_nr,
                                              'max_retries': max_retries, 'verbose': verbose,
                                              'custom_callback': custom_callback}),
                                     name=get_thread_name(worker))
        worker_th.daemon = True

        worker_th.start()
        worker_ths.append(worker_th)

    while not stop:
        sleep(1)


def start_executor(host='127.0.0.1', usr='guest', pas='<PASSWORD>', queue=DEFAULT_DESTINATION,
                   tasks_nr=1, max_retries=0, verbose=False, custom_callback=None):
    curr_th_name = threading.current_thread().name
    logger = get_logger(name=curr_module_name())
    logger.info(">> Starting executor %s connecting to rabbitmq: %s:%s@%s for executing %d tasks.",
                curr_th_name, usr, pas, host, tasks_nr)

    ch = connect_and_declare(host=host, usr=usr, pas=pas, destinations=queue)
    ch.basic_qos(prefetch_count=1)  # consume msgs one at a time

    channels.append(ch)

    def create_response_for(msg):
        resp = {}
        resp.update(msg)
        resp['executor_name'] = get_executor_name()
        resp['retries'] = msg.get('retries', 0)
        resp['max_retries'] = msg.get('max_retries', 0)
        return resp

    def should_retry(response_msg):
        is_error = response_msg['returncode'] != 0
        returncode = response_msg['returncode']
        is_retriable = returncode not in response_msg['non_retriable']
        current_retries = response_msg.get('retries', 0)
        msg_max_retries = response_msg.get('max_retries', max_retries)
        retries_pending = current_retries < msg_max_retries
        return is_error and is_retriable and retries_pending

    def send_response(response_msg):
        if should_retry(response_msg):
            logger.debug('---- retrying msg correlation_id: %d current_retries: %d of %d',
                         response_msg['correlation_id'], response_msg['retries'],
                         response_msg['max_retries'])
            tgt_exch = TASK_REQUESTS_POOL
            response_msg['retries'] += 1
        else:
            tgt_exch = TASK_RESPONSES_POOL

        response_str = json.dumps(response_msg)
        props = pika.BasicProperties(
                             delivery_mode=2,  # make message persistent
                          )
        ch.basic_publish(exchange=tgt_exch, routing_key='', body=response_str, properties=props)

    def tasks_nr_generator(tasks_nr):
        tasks_nr_gen = tasks_nr
{'fraction_non_alphanumeric': 0.10072031793343268, 'fraction_numerical': 0.010307998012916045, 'mean_word_length': 2.7860836859426423, 'pattern_counts': {'":': 2, '<': 7, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 4, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 16, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5300432', 'n_tokens_mistral': 2531, 'n_tokens_neox': 2421, 'n_words': 518}