Change executor to thread pool executor

Author: MDUYN

investing_algorithm_framework/core/executors/executor.py

@@ -1,117 +1,49 @@
-from queue import Queue
-from typing import List, Dict
-from wrapt import synchronized
-from abc import abstractmethod, ABC
+from typing import List
+from abc import ABC
+from concurrent.futures import ThreadPoolExecutor
 
 from investing_algorithm_framework.core.workers import Worker
-from investing_algorithm_framework.core.exceptions import OperationalException
-from investing_algorithm_framework.core.utils import StoppableThread
 from investing_algorithm_framework.core.events.observer import Observer
 from investing_algorithm_framework.core.events.observable import Observable
 from investing_algorithm_framework.configuration.config_constants import \
     DEFAULT_MAX_WORKERS
 
 
-class Executor(Observable, Observer, ABC):
+class Executor(Observable, ABC):
     """
-    Executor class: functions as an abstract class that will handle the
+    Executor class: functions as a thread executor that will handle the
     executions of workers in asynchronous order.
+
+    It will make use of the concurrent library for execution of the workers.
+    Also the executor functions as an observable instance.
     """
 
-    def __init__(self,  max_workers: int = DEFAULT_MAX_WORKERS):
+    def __init__(
+            self,
+            workers: List[Worker],
+            max_concurrent_workers: int = DEFAULT_MAX_WORKERS
+    ) -> None:
         super(Executor, self).__init__()
 
-        self._max_workers = max_workers
-        self._pending_workers: Queue = None
-        self._running_threads: Dict[Worker, StoppableThread] = {}
+        self._workers = workers
+        self._max_concurrent_workers = max_concurrent_workers
 
     def start(self) -> None:
         """
-        Main entry for the executor.
-        """
-
-        self._initialize()
-        self.run_jobs()
-
-    def stop(self) -> None:
-        """
-        Function that will stop all running workers.
-        """
-
-        for worker in self._running_threads:
-            self.stop_running_worker(worker)
-
-        self.clean_up()
-
-    def clean_up(self):
-        """
-        Clean ups the resources.
-        """
-
-        self._pending_workers: Queue = None
-        self._running_threads: Dict[Worker, StoppableThread] = {}
-
-    def _initialize(self):
-        """
-        Functions that initializes the pending workers.
-        """
-
-        workers = self.create_workers()
-
-        if not workers or len(workers) == 0:
-            raise OperationalException("There where no workers initialized "
-                                       "for the executor instance")
-
-        self._pending_workers = Queue()
-
-        for worker in workers:
-            self._pending_workers.put(worker)
-
-    @abstractmethod
-    def create_workers(self) -> List[Worker]:
-        """
-        Abstract function that will create the workers.
-        """
-        pass
+        Main entry for the executor. The executor creates a ThreadPoolExecutor
+        with the given amount of max_workers.
 
-    def run_jobs(self) -> None:
+        It will then pass all the workers to the ThreadPoolExecutor. When
+        finished it will update all its observers
         """
-        Function that will start all the workers.
-        """
-
-        worker_iteration = self._max_workers - len(
-            self._running_threads.keys()
-        )
 
-        while worker_iteration > 0 and not self._pending_workers.empty():
-            worker = self._pending_workers.get()
-            worker_iteration -= 1
-            thread = StoppableThread(target=worker.start)
-            worker.add_observer(self)
-            self._running_threads[worker] = thread
-            thread.start()
+        with ThreadPoolExecutor(max_workers=self._max_concurrent_workers) as \
+                executor:
 
-    @synchronized
-    def update(self, observable, **kwargs) -> None:
-        """
-        Observer implementation.
-        """
+            for worker in self.workers:
+                executor.submit(worker.start)
 
-        if observable in self._running_threads:
-            del self._running_threads[observable]
-
-        if not self.processing:
-            self.notify_observers()
-        else:
-            self.run_jobs()
-
-    def stop_running_worker(self, worker: Worker) -> None:
-        """
-        Function that will stop a running worker.
-        """
-
-        thread = self._running_threads[worker]
-        thread.kill()
+        self.notify_observers()
 
     def add_observer(self, observer: Observer) -> None:
         super(Executor, self).add_observer(observer)
@@ -120,12 +52,5 @@ def remove_observer(self, observer: Observer) -> None:
         super(Executor, self).remove_observer(observer)
 
     @property
-    def processing(self) -> bool:
-        """
-        Property that will show if the executor is running.
-        """
-
-        return (self._pending_workers is not None
-                and not self._pending_workers.empty()) or \
-               (self._running_threads is not None
-                and len(self._running_threads.keys()) > 0)
+    def workers(self) -> List[Worker]:
+        return self._workers

investing_algorithm_framework/core/state/state.py

@@ -6,9 +6,9 @@
 
 class State(ABC):
     """
-    Represents a state of the Bot, these state are use by the BotContext.
-    Each implemented state represents a work mode for the
-    investing_algorithm_framework.
+    Represents a state of the context, these state are use by the Context.
+    Each implemented state represents a work mode for a application created
+    with investing_algorithm_framework.
     """
 
     # Transition state for the next BotState
@@ -24,7 +24,7 @@ def __init__(self, context) -> None:
     def start(self):
 
         # Will stop the state if pre-conditions are not met
-        if not self.validate_state():
+        if not self.validate_state(pre_state=True):
             return
 
         while True:

investing_algorithm_framework/core/workers/__init__.py

@@ -1,5 +1,7 @@
 from investing_algorithm_framework.core.workers.worker import Worker
 from investing_algorithm_framework.core.workers.scheduled_worker \
     import ScheduledWorker
+from investing_algorithm_framework.core.workers.relational_worker import \
+    RelationalWorker
 
-__all__ = ['Worker', 'ScheduledWorker']
+__all__ = ['Worker', 'ScheduledWorker', 'RelationalWorker']

investing_algorithm_framework/core/workers/relational_worker.py

@@ -0,0 +1,39 @@
+from abc import ABC
+from typing import Dict, Any
+
+from investing_algorithm_framework.core.workers.worker import Worker
+from investing_algorithm_framework.core.exceptions import OperationalException
+
+
+class RelationalWorker(Worker, ABC):
+    """
+    RelationalWorker will start after it's relational worker has run.
+
+    It will check if the related worked had run, and if this is
+    true it will start itself. Use this worker if you want to create
+    chains of workers that are depended on each other.
+    """
+    run_after: Worker
+
+    def start(self, **kwargs: Dict[str, Any]) -> None:
+
+        # Only run if the last time this worker stared is before
+        # the last time the 'run_after' worker had finished.
+        if not self.run_after:
+            raise OperationalException(
+                'The run_after worker is not set, make sure you set this '
+                'attribute to let the RelationalWorker run properly.'
+            )
+
+        if self.last_run is not None:
+
+            if self.run_after.last_run is None:
+                raise OperationalException(
+                    "Relational Worker has run before 'run_after' worker."
+                )
+
+            if self.run_after.last_run > self.last_run:
+                super(RelationalWorker, self).start()
+
+        elif self.run_after.last_run is not None:
+            super(RelationalWorker, self).start()

investing_algorithm_framework/core/workers/scheduled_worker.py

@@ -1,24 +1,58 @@
 from abc import ABC
+from datetime import datetime
+from typing import Dict, Any
 
 from investing_algorithm_framework.core.utils import TimeUnit
 from investing_algorithm_framework.core.workers.worker import Worker
 
 
 class ScheduledWorker(Worker, ABC):
+    time_unit: TimeUnit = None
+    time_interval: int = None
+
+    def start(self, **kwargs: Dict[str, Any]) -> None:
+
+        # If the worker has never run, run it
+        if self.last_run is None:
+            super(ScheduledWorker, self).start()
+
+        else:
+            # Get the current time
+            elapsed_time = datetime.now() - self.last_run
+
+            # Second evaluation
+            if self.get_time_unit() is TimeUnit.SECOND:
+                seconds = elapsed_time.total_seconds()
+
+                if seconds > self.get_time_interval():
+                    super(ScheduledWorker, self).start()
+
+            # Minute evaluation
+            elif self.get_time_unit() is TimeUnit.MINUTE:
+                minutes = divmod(elapsed_time.total_seconds(), 60)
+
+                if minutes > self.get_time_interval():
+                    super(ScheduledWorker, self).start()
+
+            # Hour evaluation
+            elif self.get_time_unit() is TimeUnit.HOUR:
+                hours = divmod(elapsed_time.total_seconds(), 3600)
+
+                if hours > self.get_time_interval():
+                    super(ScheduledWorker, self).start()
 
     def get_time_unit(self) -> TimeUnit:
         assert getattr(self, 'time_unit', None) is not None, (
             "{} should either include a time_unit attribute, or override the "
-            "`get_time_unit()`, method.".format(self.__class__.__name__)
+            "`get_time_unit()` method.".format(self.__class__.__name__)
         )
 
         return getattr(self, 'time_unit')
 
     def get_time_interval(self) -> int:
         assert getattr(self, 'time_interval', None) is not None, (
-            "{} should either include a time_interval attribute, or "
-            "override the `get_time_interval()`, "
-            "method.".format(self.__class__.__name__)
+            "{} should either include a time_interval attribute, or override "
+            "the `get_time_interval()` method.".format(self.__class__.__name__)
         )
 
         return getattr(self, 'time_interval')

investing_algorithm_framework/templates/data_providers/__init__.py

@@ -1,4 +1,4 @@
 from investing_algorithm_framework.templates.data_providers.data_provider \
-    import DataProvider
+    import DataProvider, RelationalDataProvider, ScheduledDataProvider
 
-__all__ = ['DataProvider']
+__all__ = ['DataProvider', 'RelationalDataProvider', 'ScheduledDataProvider']

investing_algorithm_framework/templates/data_providers/data_provider.py

@@ -1,43 +1,60 @@
-import logging
 from typing import Dict, Any
-from abc import abstractmethod
+from abc import abstractmethod, ABC
 
-from investing_algorithm_framework.core.workers import ScheduledWorker
+from investing_algorithm_framework.core.workers import ScheduledWorker, \
+    Worker, RelationalWorker
 
-logger = logging.getLogger(__name__)
+
+class DataProviderInterface:
+    """
+    Class DataProviderInterface: interface for data provider implementation,
+    this interface can be used to implement a data provider. A client then
+    knows which method to call when presented with a 'data provider'
+    """
+    @abstractmethod
+    def provide_data(self) -> None:
+        pass
 
 
-class DataProviderException(Exception):
+class DataProvider(DataProviderInterface, Worker, ABC):
     """
-    Should be raised when an data_provider related error occurs, for
-    example if an authorization for an API fails,
-    i.e.: raise DataProviderException('Provided api token is false')
+    Class DataProvider: makes use of the abstract Worker class and inherits the
+    interface of of the DataProviderInterface.
+
+    This is a Worker instance.
     """
 
-    def __init__(self, message: str) -> None:
-        super().__init__(self)
-        self.message = message
+    def work(self, **kwargs: Dict[str, Any]) -> None:
 
-    def __str__(self) -> str:
-        return self.message
+        # Call DataProviderInterface
+        self.provide_data()
 
 
-class DataProvider(ScheduledWorker):
+class ScheduledDataProvider(DataProviderInterface, ScheduledWorker, ABC):
     """
-    Class DataProvider: An entity which responsibility is to provide data
-    from an external source. Where a data_providers source is defined as
-    any third party service that provides data e.g  cloud storage,
-    REST API, or website.
+    Class ScheduledDataProvider: makes use of the abstract ScheduledWorker
+    class and inherits the interface of of the DataProviderInterface.
 
-    A data_provider must always be run with the start function from it´s
-    super class. Otherwise depended
-    observers will not be updated.
+    This is a ScheduledWorker instance, and therefore you must set the
+    'time_unit' class attribute and the 'time_interval' class attribute.
     """
 
-    @abstractmethod
-    def provide_data(self) -> None:
-        pass
+    def work(self, **kwargs: Dict[str, Any]) -> None:
+
+        # Call DataProviderInterface
+        self.provide_data()
+
+
+class RelationalDataProvider(RelationalWorker, DataProviderInterface, ABC):
+    """
+    Class RelationalDataProvider: makes use of the abstract RelationalWorker
+    class and inherits the interface of of the DataProviderInterface.
+
+    This is a RelationalWorker instance, and therefore you must link it to
+    another worker instance, by setting the 'run_after' class attribute.
+    """
 
     def work(self, **kwargs: Dict[str, Any]) -> None:
-        logger.info("Starting data provider {}".format(self.get_id()))
+
+        # Call DataProviderInterface
         self.provide_data()