<commit_before>import logging
import time
import json
import uuid

from boto.sqs.message import Message

from nymms import resources

logger = logging.getLogger(__name__)


class SQSScheduler(object):
    def __init__(self, connection, queue_name, node_backend):
        self.node_backend = node_backend
        self.connection = connection
        self.queue_name = queue_name
        self.tasks_sent = 0
        self.create_queue()

    def create_queue(self):
        logger.debug("Creating queue '%s'." % (self.queue_name))
        self.queue = self.connection.create_queue(self.queue_name)

    def send_task(self, task, delay=None):
        logger.debug("Sending task to queue '%s'." % (self.queue_name))
        m = Message()
        m.set_body(json.dumps(task))
        return self.queue.write(m, delay_seconds=delay)

    def get_tasks(self):
        tasks = {}
        self.node_backend.load_nodes()
        nodes = resources.Node.registry
        for node_name, node in nodes.iteritems():
            tasks[node_name] = node.tasks
        return tasks

    def task_url(self, task):
        url = "nymms://{address}/{monitor[name]}/"
        return url.format(**task) + "?timestamp=%f" % (time.time())

    def task_uuid(self, url):
        _uuid = uuid.uuid5(uuid.NAMESPACE_URL, url)
        logger.debug("Task %s uuid %s" % (url, _uuid))
        return _uuid

    def run(self):
        while True:
            logger.info("Tasks sent: %d" % (self.tasks_sent))
            pass_count = 0
            tasks = self.get_tasks()
            while True:
                working_index = tasks.keys()
                if not working_index:
                    break
                for node in working_index:
                    try:
                        task = tasks[node].pop()
                    except IndexError:
                        del(tasks[node])
                        continue
                    # Give task a unique UUID
                    url = self.task_url(task)
                    task['_url'] = url
                    task['_uuid'] = self.task_uuid(url).hex
                    task['_attempt'] = 0
                    self.send_task(task, pass_count * 5)
                    self.tasks_sent += 1
                pass_count += 1
            time.sleep(10)


class YamlNodeBackend(object):
    def __init__(self, path):
        self.path = path

    def load_nodes(self):
        logger.debug("Loading node config from %s" % (self.path))
        resources.load_nodes(self.path, reset=True)<commit_msg>Integrate delay and remove uuid<commit_after>import logging
import time
import json

from boto.sqs.message import Message

from nymms import resources
from nymms.config import config

logger = logging.getLogger(__name__)


class SQSScheduler(object):
    def __init__(self, connection, queue_name, node_backend):
        self.node_backend = node_backend
        self.connection = connection
        self.queue_name = queue_name
        self.tasks_sent = 0
        self.create_queue()

    def create_queue(self):
        logger.debug("Creating queue '%s'." % (self.queue_name))
        self.queue = self.connection.create_queue(self.queue_name)

    def send_task(self, task, delay=None):
        logger.debug("Sending task to queue '%s'." % (self.queue_name))
        m = Message()
        m.set_body(json.dumps(task))
        return self.queue.write(m, delay_seconds=delay)

    def get_tasks(self):
        tasks = {}
        self.node_backend.load_nodes()
        nodes = resources.Node.registry
        for node_name, node in nodes.iteritems():
            tasks[node_name] = node.tasks
        return tasks

    def task_url(self, task):
        url = "nymms://{address}/{monitor[name]}/"
        return url.format(**task) + "?timestamp=%f" % (time.time())

    def run(self):
        while True:
            start = time.time()
            logger.info("Tasks sent: %d" % (self.tasks_sent))
            pass_count = 0
            tasks = self.get_tasks()
            while True:
                working_index = tasks.keys()
                if not working_index:
                    break
                for node in working_index:
                    try:
                        task = tasks[node].pop()
                    except IndexError:
                        del(tasks[node])
                        continue
                    url = self.task_url(task)
                    task['_url'] = url
                    task['_attempt'] = 0
                    self.send_task(task, pass_count * 5)
                    self.tasks_sent += 1
                pass_count += 1
            run_time = time.time() - start
            logger.debug("Scheduler iteration took %d seconds." % (run_time,))
            sleep_time = (config.settings['scheduler']['interval'] -
                          max(run_time, 0))
            logger.debug("Scheduler sleeping for %d seconds." % (sleep_time,))
            time.sleep(sleep_time)


class YamlNodeBackend(object):
    def __init__(self, path):
        self.path = path

    def load_nodes(self):
        logger.debug("Loading node config from %s" % (self.path))
        resources.load_nodes(self.path, reset=True)
{'fraction_non_alphanumeric': 0.10644850818094322, 'fraction_numerical': 0.0030798845043310875, 'mean_word_length': 1.8116883116883118, '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': '2112746', 'n_tokens_mistral': 1496, 'n_tokens_neox': 1432, 'n_words': 335}