Refactor to use concurrent.futures

CHANGELOG.md

@@ -0,0 +1,13 @@
+# Change Log
+
+
+All notable changes to this project will be documented in this file.
+
+The format is based on [Keep a Changelog](http://keepachangelog.com/)
+and this project adheres to [Semantic Versioning](http://semver.org/).
+
+
+## [Unreleased]
+### Added
+- the first pass of the small queueing system. 
+- An InMemClient that spawns thread-based workers, and stores jobs in-memory.

src/barbequeue/client.py

@@ -1,4 +1,3 @@
-import importlib
 import uuid
 
 from barbequeue.common.classes import Job
@@ -16,50 +15,56 @@ def __init__(self, app, namespace, **config):
     def schedule(self, func, *args, **kwargs):
         """
         Schedules a function func for execution.
-        """
 
-        updates_progress = kwargs.pop('updates_progress', False)
+        The only other special parameter is track_progress. If passed in and not None, the func will be passed in a
+        keyword parameter called update_progress:
+
+        def update_progress(progress, total_progress, stage=""):
+
+        The running function can call the update_progress function to notify interested parties of the function's
+        current progress.
+
+        All other parameters are directly passed to the function when it starts running.
+
+        :type func: callable or str
+        :param func: A callable object that will be scheduled for running.
+        :return: a string representing the job_id.
+        """
 
-        # turn our function object into its fully qualified name if needed
-        if callable(func):
-            funcname = self.stringify_func(func)
+        # if the func is already a job object, just schedule that directly.
+        if isinstance(func, Job):
+            job = func
+        # else, turn it into a job first.
         else:
-            funcname = func
+            job = Job(func, *args, **kwargs)
 
-        job = Job(funcname, *args, **kwargs)
+        job.track_progress = kwargs.pop('track_progress', False)
         job_id = self.storage.schedule_job(job)
         return job_id
 
     def cancel(self, job_id):
         """
         Mark a job as canceled and remove it from the list of jobs to be executed.
         Send a message to our workers to stop a job.
+
+        :param job_id: the job_id of the Job to cancel.
         """
         self.storage.cancel_job(job_id)
 
     def status(self, job_id):
         """
-        Gets the status of a job given by job_id.
-        """
-        return self.storage.get_job(job_id)
-
-    @staticmethod
-    def stringify_func(func):
-        assert callable(func), "function {} passed to stringify_func isn't a function!".format(func)
+        Returns a Job object corresponding to the job_id. From there, you can query for the following attributes:
 
-        fqn = "{module}.{funcname}".format(module=func.__module__, funcname=func.__name__)
-        return fqn
+        - function string to run
+        - its current state (see Job.State for the list of states)
+        - progress (returning an int), total_progress (returning an int), and percentage_progress
+        (derived from running job.progress/total_progress)
+        - the job.exception and job.traceback, if the job's function returned an error
 
-    @staticmethod
-    def import_stringified_func(funcstring):
-        assert isinstance(funcstring, str)
-
-        modulestring, funcname = funcstring.rsplit('.', 1)
-
-        mod = importlib.import_module(modulestring)
-
-        func = getattr(mod, funcname)
-        return func
+        :param job_id: the job_id to get the Job object for
+        :return: the Job object corresponding to the job_id
+        """
+        return self.storage.get_job(job_id)
 
 
 class InMemClient(Client):
@@ -80,13 +85,22 @@ def __init__(self, app, namespace, *args, **kwargs):
         self._storage = storage_inmem.Backend(app, namespace)
         self._messaging = messaging_inmem.Backend()
         self._workers = inmem.Backend(incoming_message_mailbox=self.worker_mailbox_name,
-                                      outgoing_message_mailbox=self.scheduler_mailbox_name)
-        self._scheduler = Scheduler(self._storage, self._messaging, self._workers,
+                                      outgoing_message_mailbox=self.scheduler_mailbox_name,
+                                      msgbackend=self._messaging)
+        self._scheduler = Scheduler(self._storage, self._messaging,
                                     worker_mailbox=self.worker_mailbox_name,
                                     incoming_mailbox=self.scheduler_mailbox_name)
 
         super(InMemClient, self).__init__(app, namespace, storage_backend=storage_inmem, *args, **kwargs)
 
     def shutdown(self):
+        """
+        Shutdown the client and all of its managed resources:
+
+        - the workers
+        - the scheduler threads
+
+        :return: None
+        """
         self._scheduler.shutdown()
         self._workers.shutdown()

src/barbequeue/common/classes.py

@@ -1,113 +1,114 @@
-import abc
 import enum
-import importlib
 import logging
-import threading
 from collections import namedtuple
+from functools import partial
 
-from barbequeue import humanhash
+from barbequeue.common.utils import import_stringified_func, stringify_func
 
 logger = logging.getLogger(__name__)
 
 
 class Job(object):
+    """
+    Job represents a function whose execution has been deferred through the Client's schedule function.
+
+    Jobs are stored on the storage backend for persistence through restarts, and are scheduled for running
+    to the workers.
+    """
     class State(enum.Enum):
+        """
+        the Job.State object enumerates a Job's possible valid states.
+
+        SCHEDULED means the Job has been accepted by the client, but has not been
+        sent to the workers for running.
+
+        QUEUED means the Job has been sent to the workers for running, but has not
+        been run yet (to our knowledge).
+
+        RUNNING means that one of the workers has started running the job, but is not
+        complete yet. If the job has been set to track progress, then the job's progress
+        and total_progress fields should be continuously updated.
+
+        FAILED means that the job's function has raised an exception during runtime.
+        The job's exception and traceback fields should be set.
+
+        CANCELED means that the job has been canceled from running.
+
+        COMPLETED means that the function has run to completion. The job's result field
+        should be set with the function's return value.
+        """
+
         SCHEDULED = 0
-        # STARTED = 1
+        QUEUED = 1
         RUNNING = 2
         FAILED = 3
         CANCELED = 4
         COMPLETED = 5
 
-    def __init__(self, func_string, *args, **kwargs):
+    def __init__(self, func, *args, **kwargs):
+        """
+        Create a new Job that will run func given the arguments passed to Job(). If the track_progress keyword parameter
+        is given, the worker will pass an update_progress function to update interested parties about the function's
+        progress. See Client.__doc__ for update_progress's function parameters.
+
+        :param func: func can be a callable object, in which case it is turned into an importable string,
+        or it can be an importable string already.
+        """
         self.job_id = kwargs.pop('job_id', None)
         self.state = kwargs.pop('state', self.State.SCHEDULED)
-        self.func = func_string
-        self.traceback = kwargs.pop('traceback', '')
-        self.exception = kwargs.pop('exception', '')
+        self.traceback = ""
+        self.exception = None
+        self.track_progress = kwargs.pop('track_progress', False)
+        self.progress = 0
+        self.total_progress = 0
         self.args = args
         self.kwargs = kwargs
 
-    def get_lambda_to_execute(self):
-        fqn = self.func
-        modulename, funcname = fqn.rsplit('.', 1)
-        mod = importlib.import_module(modulename)
-        assert hasattr(
-            mod, funcname), \
-            "Module {} does not have attribute {}".format(
-                mod, funcname)
-
-        func = getattr(mod, funcname)
-
-        y = lambda: func(*self.args, **self.kwargs)
-        return y
+        if callable(func):
+            funcstring = stringify_func(func)
+        elif isinstance(func, str):
+            funcstring = func
+        else:
+            raise Exception(
+                "Error in creating job. We do not know how to "
+                "handle a function of type {}".format(type(func)))
 
-    def serialize(self):
-        pass
-
-
-class ProgressData(namedtuple("_ProgressData", ["id", "order", "data"])):
-    pass
+        self.func = funcstring
 
+    def get_lambda_to_execute(self):
+        """
+        return a function that executes the function assigned to this job.
+
+        If job.track_progress is None (the default), the returned function accepts no argument
+        and simply needs to be called. If job.track_progress is True, an update_progress function
+        is passed in that can be used by the function to provide feedback progress back to the
+        job scheduling system.
+
+        :return: a function that executes the original function assigned to this job.
+        """
+        func = import_stringified_func(self.func)
 
-class Function(namedtuple("_Function", ["module", "funcname"])):
-    def serialize(self):
-        # Since this is all in memory, there is no need to serialize anything.
-        raise NotImplementedError()
+        if self.track_progress:
+            y = lambda p: func(update_progress=partial(p, self.job_id), *self.args, **self.kwargs)
+        else:
+            y = lambda: func(*self.args, **self.kwargs)
 
+        return y
 
-class BaseCloseableThread(threading.Thread):
-    """
-    A base class for a thread that monitors an Event as a signal for shutting down, and a main_loop that otherwise loop
-     until the shutdown event is received.
-    """
-    __metaclass__ = abc.ABCMeta
-
-    DEFAULT_TIMEOUT_SECONDS = 0.2
-
-    def __init__(self, shutdown_event, thread_name, *args, **kwargs):
-        assert isinstance(shutdown_event, threading.Event)
-
-        self.shutdown_event = shutdown_event
-        self.thread_name = thread_name
-        self.thread_id = self._generate_thread_id()
-        self.logger = logging.getLogger("{module}.{name}[{id}]".format(module=__name__,
-                                                                       name=self.thread_name,
-                                                                       id=self.thread_id))
-        self.full_thread_name = "{thread_name}-{thread_id}".format(thread_name=self.thread_name,
-                                                                   thread_id=self.thread_id)
-        super(BaseCloseableThread, self).__init__(name=self.full_thread_name, *args, **kwargs)
-
-    def run(self):
-        self.logger.info("Started new {name} thread ID#{id}".format(name=self.thread_name,
-                                                                    id=self.thread_id))
-
-        while True:
-            if self.shutdown_event.wait(self.DEFAULT_TIMEOUT_SECONDS):
-                self.logger.warning("{name} shut down event received; closing.".format(name=self.thread_name))
-                self.shutdown()
-                break
-            else:
-                self.main_loop(timeout=self.DEFAULT_TIMEOUT_SECONDS)
-                continue
-
-    @abc.abstractmethod
-    def main_loop(self, timeout):
+    @property
+    def percentage_progress(self):
         """
-        The main loop of a thread. Run this loop if we haven't received any shutdown events in the last 
-        timeout seconds. Normally this is used to read from a queue; you are encouraged to return from 
-        this function if the timeout parameter has elapsed, to allow the thread to continue to check
-        for the shutdown event.
-        :param timeout: a parameter determining how long you can wait for a timeout.
-        :return: None
+        Returns a float between 0 and 1, representing the current job's progress in its task.
+
+        :return: float corresponding to the total percentage progress of the job.
         """
-        pass
-
-    @staticmethod
-    def _generate_thread_id():
-        uid, _ = humanhash.uuid()
-        return uid
-
-    def shutdown(self):
-        # stub method, override if you need a more complex shut down procedure.
-        pass
+        return float(self.progress) / self.total_progress
+
+    def __repr__(self):
+        return "<Job id: {id} state: {state} progress: {p}/{total} func: {func}>".format(
+            id=self.job_id,
+            state=self.state.name,
+            func=self.func,
+            p=self.progress,
+            total=self.total_progress
+        )