Refactor Future abstraction: define an abstract future and make executor specific future impls in there package

Author: darkjh

simpleflow/executor.py

@@ -14,7 +14,7 @@ def get_actual_value(value):
     """Unwrap the result of a Future or return the value.
 
     """
-    if isinstance(value, futures.Future):
+    if isinstance(value, futures.AbstractFuture):
         return value.result()
     return value
 

simpleflow/futures.py

@@ -1,6 +1,6 @@
 # -*- coding: utf-8 -*-
 
-from concurrent import futures as py_futures
+import abc
 from concurrent.futures._base import (
     PENDING,
     RUNNING,
@@ -10,7 +10,7 @@
 )
 
 
-__all__ = ['Future', 'wait']
+__all__ = ['AbstractFuture', 'wait']
 
 
 FIRST_COMPLETED = 'FIRST_COMPLETED'
@@ -27,26 +27,50 @@
     FINISHED
 ]
 
-_STATE_TO_DESCRIPTION_MAP = {
-    PENDING: "pending",
-    RUNNING: "running",
-    CANCELLED: "cancelled",
-    CANCELLED_AND_NOTIFIED: "cancelled",
-    FINISHED: "finished"
-}
-
 
 def wait(*fs):
     """Returns a list of the results of futures if there are available.
     """
     return [future.result() for future in fs]
 
 
-class Future(py_futures.Future):
-    """Patched version of ``concurrent.futures.Future``
+class AbstractFuture(object):
+    """Base future class that defines an interface for concrete impls
     """
+
+    __metaclass__ = abc.ABCMeta
+
+    @abc.abstractmethod
+    def result(self):
+        """Return the result of the underlying computation
+
+        The actual behavior (blocking etc.) depends on impl
+        """
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def exception(self):
+        """Return the exception raised (if any) by the underlying computation
+
+        The actual behavior (blocking etc.) depends on impl
+        """
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def running(self):
+        """Return True if the underlying computation is currently executing
+        """
+        raise NotImplementedError
+
+    @abc.abstractmethod
     def finished(self):
-        with self._condition:
-            if self._state == FINISHED:
-                return True
-            return False
+        """Return True if the underlying computation has finished
+        """
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def done(self):
+        """Return True if the underlying compuation is cancelled or
+        has finished
+        """
+        raise NotImplementedError

simpleflow/local/futures.py

@@ -0,0 +1,21 @@
+from simpleflow.futures import AbstractFuture
+from concurrent.futures import Future as PythonFuture
+
+
+class Future(PythonFuture, AbstractFuture):
+    """Future impl for local execution
+
+    The `concurrent.futures.Future` from python is itself a concrete impl
+    rather than an abstract class or an interface. So a multiple inheritance
+    is used to work with our interface `simpleflow.futures.AbstractFuture`.
+
+    This class inherits both the abstracts methods from `AbstractFuture` and
+    concrete methods from `concurrent.futures.Future`. The `AbstractFuture`
+    interface is designed so that python's Future 'implements' its methods.
+    Since python's Future is placed left to the `AbstractFuture` in inheritance
+    declaration, runtime method resolution will pick the concrete methods.
+
+    This Future class is interoperable with python's builtin executors.
+    """
+    def finished(self):
+        return self.done()

simpleflow/swf/futures.py

@@ -4,33 +4,93 @@
     CANCELLED,
     PENDING,
     RUNNING,
-    Future as BaseFuture
+    CANCELLED_AND_NOTIFIED,
+    AbstractFuture
 )
 
 
-class Future(BaseFuture):
-    """Future with special Simple Workflow semantics
+_STATE_TO_DESCRIPTION_MAP = {
+    PENDING: "pending",
+    RUNNING: "running",
+    CANCELLED: "cancelled",
+    CANCELLED_AND_NOTIFIED: "cancelled",
+    FINISHED: "finished"
+}
+
+
+class Future(AbstractFuture):
+    """Future impl that contains Simple Workflow specific logic
     """
+
+    def __init__(self):
+        """Represents the state of a computation.
+
+        It tries to mimics mod::`concurrent.futures` but involved some
+        adaptations to fit the Amazon SWF model.
+
+        """
+        self._state = PENDING
+        self._result = None
+        self._exception = None
+
+    def __repr__(self):
+        return '<Future at %s state=%s>' % (
+            hex(id(self)),
+            _STATE_TO_DESCRIPTION_MAP[self._state])
+
     @classmethod
     def wait(cls):
         raise exceptions.ExecutionBlocked
 
-    def result(self, timeout=None):
-        with self._condition:
-            if self._state != FINISHED:
-                return self.wait()
-            # TODO what happen if cancelled ???
-            return self._result
+    def result(self):
+        """Raise a cls::`exceptions.ExecutionBlocked` when the result is not
+        available."""
+        if self._state != FINISHED:
+            return self.wait()
 
-    def exception(self, timeout=None):
-        with self._condition:
-            if self._state != FINISHED:
-                return self.wait()
-            return self._exception
+        return self._result
 
     def cancel(self):
-        with self._condition:
-            if self._state == FINISHED:
-                return False
-            self._state = CANCELLED
-            return True
+        """Cancel a future.
+
+        Note: cannot cancel a future that is already finished.
+        It will not raise an exception but return ``False`` to notify it.
+
+        """
+        if self._state == FINISHED:
+            return False
+
+        self._state = CANCELLED
+        return True
+
+    def state(self):
+        return self._state
+
+    def exception(self):
+        """
+        Returns `None` if no exception occurred, otherwise, returns the
+        exception object that what raised by the task.
+
+        Raise a cls::`exceptions.ExecutionBlocked` when the result is not
+        available.
+
+        """
+        if self._state != FINISHED:
+            return self.wait()
+
+        return self._exception
+
+    def cancelled(self):
+        return self._state == CANCELLED
+
+    def running(self):
+        return self._state == RUNNING
+
+    def finished(self):
+        return self._state == FINISHED
+
+    def done(self):
+        return self._state in [
+            CANCELLED,
+            FINISHED
+        ]