Add asynchrone job handler (#752)

* Add first draft asynchrone_job_handler

* Adjust background job manager to use asynchronous job handler

* Fix test

* Use google sytle docstrings

* Use threads instead of processes

* fix typos

Author: Alexander-Dubrawski

hyrisecockpit/database_manager/asynchronous_job_handler.py

@@ -0,0 +1,134 @@
+"""Asynchronous job handler."""
+
+from multiprocessing import Value
+from threading import Thread
+from typing import Dict
+
+from hyrisecockpit.database_manager.cursor import ConnectionFactory
+
+from .job.activate_plugin import activate_plugin as activate_plugin_job
+from .job.deactivate_plugin import deactivate_plugin as deactivate_plugin_job
+from .job.delete_tables import delete_tables as delete_tables_job
+from .job.load_tables import load_tables as load_tables_job
+
+
+class AsynchronousJobHandler:
+    """Handles asynchronous jobs.
+
+    All jobs in this class are executed with python threads in the background.
+    If a method from this class is called it will only response if the job
+    was started successfully or not. All jobs in this class do not need to return a
+    result.
+    """
+
+    def __init__(
+        self,
+        database_blocked: Value,
+        connection_factory: ConnectionFactory,
+        workload_drivers: Dict,
+    ):
+        """Initialize asynchronous job handler object.
+
+        Args:
+            database_blocked: Flag stored in a shared memory map. This flag
+                stores if the Hyrise instance is blocked or not.
+            connection_factory: An object to create a connection to the Hyrise
+                database. All connection relevant information (port, host) is
+                saved in this object.
+            workload_drivers: A dictionary containing all workload drivers (TPCC,...)
+        """
+        self._database_blocked: Value = database_blocked
+        self._connection_factory: ConnectionFactory = connection_factory
+        self._workload_drivers: Dict = workload_drivers
+
+    def load_tables(self, workload_type: str, scalefactor: float) -> bool:
+        """Start load tabled job.
+
+        This function will check if the database is blocked, if not it will
+        create a thread that is executing the job that will load the tables.
+
+        Returns:
+            bool: True if job was successful started, False if database was
+                blocked and job couldn't be started.
+        """
+        if not self._database_blocked.value:
+            self._database_blocked.value = True
+            job_thread = Thread(
+                target=load_tables_job,
+                args=(
+                    self._database_blocked,
+                    workload_type,
+                    scalefactor,
+                    self._connection_factory,
+                    self._workload_drivers,
+                ),
+            )
+            job_thread.start()
+            return True
+        else:
+            return False
+
+    def delete_tables(self, workload_type: str, scalefactor: float) -> bool:
+        """Start delete tabled job.
+
+        This function will check if the database is blocked. If not it will
+        create a thread that is executing the job that will delete the tables.
+
+        Returns:
+            bool: True if job was successful started, False if database was
+                blocked and job couldn't be started.
+        """
+        if not self._database_blocked.value:
+            self._database_blocked.value = True
+            job_thread = Thread(
+                target=delete_tables_job,
+                args=(
+                    self._database_blocked,
+                    workload_type,
+                    scalefactor,
+                    self._connection_factory,
+                    self._workload_drivers,
+                ),
+            )
+            job_thread.start()
+            return True
+        else:
+            return False
+
+    def activate_plugin(self, plugin: str) -> bool:
+        """Start activate plug-in job.
+
+        This function will check if the database is blocked. If not it will
+        create a thread that is executing the job that will activate the plug-in.
+
+        Returns:
+            bool: True if job was successful started, False if database was
+                blocked and job couldn't be started.
+        """
+        if not self._database_blocked.value:
+            job_thread = Thread(
+                target=activate_plugin_job, args=(self._connection_factory, plugin,)
+            )
+            job_thread.start()
+            return True
+        else:
+            return False
+
+    def deactivate_plugin(self, plugin: str) -> bool:
+        """Start deactivate plug-in job.
+
+        This function will check if the database is blocked. If not it will
+        create a thread that is executing the job that will deactivate the plug-in.
+
+        Returns:
+            bool: True if job was successful started, False if database was
+                blocked and job couldn't be started.
+        """
+        if not self._database_blocked.value:
+            job_thread = Thread(
+                target=deactivate_plugin_job, args=(self._connection_factory, plugin,)
+            )
+            job_thread.start()
+            return True
+        else:
+            return False

hyrisecockpit/database_manager/background_scheduler.py

@@ -4,11 +4,8 @@
 
 from apscheduler.schedulers.background import BackgroundScheduler
 
+from .asynchronous_job_handler import AsynchronousJobHandler
 from .cursor import ConnectionFactory, StorageConnectionFactory
-from .job.activate_plugin import activate_plugin as activate_plugin_job
-from .job.deactivate_plugin import deactivate_plugin as deactivate_plugin_job
-from .job.delete_tables import delete_tables as delete_tables_job
-from .job.load_tables import load_tables as load_tables_job
 from .job.ping_hyrise import ping_hyrise
 from .job.update_chunks_data import update_chunks_data
 from .job.update_plugin_log import update_plugin_log
@@ -54,6 +51,9 @@ def __init__(
         self._previous_chunk_data = {
             "value": None,
         }
+        self._asynchronous_job_handler = AsynchronousJobHandler(
+            self._database_blocked, self._connection_factory, self._workload_drivers,
+        )
 
         self._init_jobs()
 
@@ -162,56 +162,16 @@ def close(self) -> None:
 
     def load_tables(self, workload_type: str, scalefactor: float) -> bool:
         """Load tables."""
-        if not self._database_blocked.value:
-            self._database_blocked.value = True
-            self._scheduler.add_job(
-                func=load_tables_job,
-                args=(
-                    self._database_blocked,
-                    workload_type,
-                    scalefactor,
-                    self._connection_factory,
-                    self._workload_drivers,
-                ),
-            )
-            return True
-        else:
-            return False
+        return self._asynchronous_job_handler.load_tables(workload_type, scalefactor)
 
     def delete_tables(self, workload_type: str, scalefactor: float) -> bool:
         """Delete tables."""
-        if not self._database_blocked.value:
-            self._database_blocked.value = True
-            self._scheduler.add_job(
-                func=delete_tables_job,
-                args=(
-                    self._database_blocked,
-                    workload_type,
-                    scalefactor,
-                    self._connection_factory,
-                    self._workload_drivers,
-                ),
-            )
-            return True
-        else:
-            return False
+        return self._asynchronous_job_handler.delete_tables(workload_type, scalefactor)
 
     def activate_plugin(self, plugin: str) -> bool:
         """Activate plugin."""
-        if not self._database_blocked.value:
-            self._scheduler.add_job(
-                func=activate_plugin_job, args=(self._connection_factory, plugin,)
-            )
-            return True
-        else:
-            return False
+        return self._asynchronous_job_handler.activate_plugin(plugin)
 
     def deactivate_plugin(self, plugin: str) -> bool:
         """Dectivate plugin."""
-        if not self._database_blocked.value:
-            self._scheduler.add_job(
-                func=deactivate_plugin_job, args=(self._connection_factory, plugin,)
-            )
-            return True
-        else:
-            return False
+        return self._asynchronous_job_handler.deactivate_plugin(plugin)

hyrisecockpit/database_manager/job/execute_queries_parallel.py

@@ -1,5 +1,5 @@
 """This job execute queries parallel in separate processes."""
-from multiprocessing import Process
+from threading import Thread
 from typing import Any, List, Optional, Tuple
 
 from psycopg2 import DatabaseError, InterfaceError, ProgrammingError
@@ -36,12 +36,11 @@ def _execute_table_query(
 
 def execute_queries_parallel(queries, connection_factory: ConnectionFactory) -> None:
     """Start processes for query execution."""
-    processes: List[Process] = [
-        Process(target=_execute_table_query, args=(query, connection_factory),)
+    threads: List[Thread] = [
+        Thread(target=_execute_table_query, args=(query, connection_factory),)
         for query in queries
     ]
-    for process in processes:
-        process.start()
-    for process in processes:
-        process.join()
-        process.terminate()
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join()

tests/database_manager/job/test_execute_queries_parallel.py

@@ -105,16 +105,16 @@ def raise_exception(*args):
             ("keep", "hyriseDown", "keep",),
         )
 
-    @patch("hyrisecockpit.database_manager.job.execute_queries_parallel.Process")
+    @patch("hyrisecockpit.database_manager.job.execute_queries_parallel.Thread")
     @patch(
         "hyrisecockpit.database_manager.job.execute_queries_parallel._execute_table_query"
     )
     def test_execute_queries_parallel(
-        self, mock_execute_table_query: MagicMock, mock_process_constructor: MagicMock
+        self, mock_execute_table_query: MagicMock, mock_thread_constructor: MagicMock
     ) -> None:
         """Test start processes for table loading queries in parallel."""
         mock_process: MagicMock = MagicMock()
-        mock_process_constructor.return_value = mock_process
+        mock_thread_constructor.return_value = mock_process
         mock_cursor = MagicMock()
         mock_connection_factory = MagicMock()
         mock_connection_factory.create_cursor.return_value.__enter__.return_value = (
@@ -126,7 +126,7 @@ def test_execute_queries_parallel(
 
         execute_queries_parallel(queries, mock_connection_factory)
 
-        mock_process_constructor.assert_called_once_with(
+        mock_thread_constructor.assert_called_once_with(
             target=mock_execute_table_query,
             args=(
                 "Why does dragons like the sun? Because they are scared of (k)nights.",
@@ -135,4 +135,3 @@ def test_execute_queries_parallel(
         )
         mock_process.start.assert_called_once()
         mock_process.join.assert_called_once()
-        mock_process.terminate.assert_called_once()