Fix - Add wait when stop dispatcher (#801)

* fix: add wait when stop dispatcher to make sure the thread is fully terminated

* fix: remove unnecessary calls in tests

* fix: remove internal imports from the assert_true_after_time()

* fix: stop the dispatcher in the core conftest

taipy/core/_core.py

@@ -66,16 +66,19 @@ def run(self, force_restart=False):
 
         self.__start_dispatcher(force_restart)
 
-    def stop(self):
+    def stop(self, wait: bool = True, timeout: Optional[float] = None):
         """
         Stop the Core service.
-
         This function stops the dispatcher and unblock the Config for update.
+
+        Parameters:
+            wait (bool): If True, the method will wait for the dispatcher to stop.
+            timeout (Optional[float]): The maximum time to wait. If None, the method will wait indefinitely.
         """
         Config.unblock_update()
 
         if self._dispatcher:
-            self._dispatcher = _OrchestratorFactory._remove_dispatcher()
+            self._dispatcher = _OrchestratorFactory._remove_dispatcher(wait, timeout)
             self.__logger.info("Core service has been stopped.")
 
         with self.__class__.__lock_is_running:

taipy/core/_orchestrator/_dispatcher/_development_job_dispatcher.py

@@ -8,6 +8,9 @@
 # Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
 # an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
 # specific language governing permissions and limitations under the License.
+
+from typing import Optional
+
 from ...job.job import Job
 from .._abstract_orchestrator import _AbstractOrchestrator
 from ._job_dispatcher import _JobDispatcher
@@ -26,7 +29,7 @@ def start(self):
     def is_running(self) -> bool:
         return True
 
-    def stop(self):
+    def stop(self, wait: bool = True, timeout: Optional[float] = None):
         raise NotImplementedError
 
     def run(self):

taipy/core/_orchestrator/_dispatcher/_job_dispatcher.py

@@ -12,7 +12,7 @@
 import threading
 from abc import abstractmethod
 from queue import Empty
-from typing import Dict
+from typing import Dict, Optional
 
 from taipy.config.config import Config
 from taipy.logger._taipy_logger import _TaipyLogger
@@ -47,12 +47,20 @@ def is_running(self) -> bool:
         """Return True if the dispatcher is running"""
         return self.is_alive()
 
-    def stop(self):
-        """Stop the dispatcher"""
+    def stop(self, wait: bool = True, timeout: Optional[float] = None):
+        """Stop the dispatcher.
+
+        Parameters:
+            wait (bool): If True, the method will wait for the dispatcher to stop.
+            timeout (Optional[float]): The maximum time to wait. If None, the method will wait indefinitely.
+        """
         self._STOP_FLAG = True
+        if wait and self.is_alive():
+            self._logger.debug("Waiting for the dispatcher thread to stop...")
+            self.join(timeout=timeout)
 
     def run(self):
-        _TaipyLogger._get_logger().info("Start job dispatcher...")
+        self._logger.info("Start job dispatcher...")
         while not self._STOP_FLAG:
             try:
                 if self._can_execute():
@@ -64,7 +72,7 @@ def run(self):
             except Empty:  # In case the last job of the queue has been removed.
                 pass
             except Exception as e:
-                _TaipyLogger._get_logger().exception(e)
+                self._logger.exception(e)
                 pass
         self._logger.info("Job dispatcher stopped.")
 

taipy/core/_orchestrator/_dispatcher/_standalone_job_dispatcher.py

@@ -8,6 +8,7 @@
 # Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
 # an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
 # specific language governing permissions and limitations under the License.
+
 from concurrent.futures import Executor, ProcessPoolExecutor
 from functools import partial
 from typing import Callable, Optional
@@ -29,7 +30,7 @@ def __init__(self, orchestrator: _AbstractOrchestrator, subproc_initializer: Opt
         max_workers = Config.job_config.max_nb_of_workers or 1
         self._executor: Executor = ProcessPoolExecutor(
             max_workers=max_workers,
-            initializer=subproc_initializer
+            initializer=subproc_initializer,
         )  # type: ignore
         self._nb_available_workers = self._executor._max_workers  # type: ignore
 

taipy/core/_orchestrator/_orchestrator_factory.py

@@ -61,9 +61,9 @@ def _build_dispatcher(cls, force_restart=False) -> Optional[_JobDispatcher]:
         return cls._dispatcher
 
     @classmethod
-    def _remove_dispatcher(cls) -> Optional[_JobDispatcher]:
+    def _remove_dispatcher(cls, wait: bool = True, timeout: Optional[float] = None) -> None:
         if cls._dispatcher is not None and not isinstance(cls._dispatcher, _DevelopmentJobDispatcher):
-            cls._dispatcher.stop()
+            cls._dispatcher.stop(wait, timeout)
         cls._dispatcher = None
         return cls._dispatcher
 

tests/core/_orchestrator/test_orchestrator.py

@@ -57,7 +57,7 @@ def test_submit_task_multithreading_multiple_task():
     task_1 = _create_task(partial(lock_multiply, lock_1))
     task_2 = _create_task(partial(lock_multiply, lock_2))
 
-    _OrchestratorFactory._build_dispatcher()
+    _OrchestratorFactory._build_dispatcher(force_restart=True)
 
     with lock_1:
         with lock_2:
@@ -104,7 +104,7 @@ def test_submit_submittable_multithreading_multiple_task():
 
     scenario = Scenario("scenario_config", [task_1, task_2], {})
 
-    _OrchestratorFactory._build_dispatcher()
+    _OrchestratorFactory._build_dispatcher(force_restart=True)
 
     with lock_1:
         with lock_2:
@@ -146,7 +146,7 @@ def test_submit_task_multithreading_multiple_task_in_sync_way_to_check_job_statu
     task_1 = _create_task(partial(lock_multiply, lock_1))
     task_2 = _create_task(partial(lock_multiply, lock_2))
 
-    _OrchestratorFactory._build_dispatcher()
+    _OrchestratorFactory._build_dispatcher(force_restart=True)
 
     with lock_0:
         submission_0 = _Orchestrator.submit_task(task_0)
@@ -215,7 +215,7 @@ def test_blocked_task():
     bar_cfg = Config.configure_data_node("bar")
     baz_cfg = Config.configure_data_node("baz")
 
-    _OrchestratorFactory._build_dispatcher()
+    _OrchestratorFactory._build_dispatcher(force_restart=True)
 
     dns = _DataManager._bulk_get_or_create([foo_cfg, bar_cfg, baz_cfg])
     foo = dns[foo_cfg]
@@ -273,7 +273,7 @@ def test_blocked_submittable():
     bar_cfg = Config.configure_data_node("bar")
     baz_cfg = Config.configure_data_node("baz")
 
-    _OrchestratorFactory._build_dispatcher()
+    _OrchestratorFactory._build_dispatcher(force_restart=True)
 
     dns = _DataManager._bulk_get_or_create([foo_cfg, bar_cfg, baz_cfg])
     foo = dns[foo_cfg]

tests/core/conftest.py

@@ -61,7 +61,6 @@
 from taipy.core.task.task import Task
 
 current_time = datetime.now()
-_OrchestratorFactory._build_orchestrator()
 
 
 @pytest.fixture(scope="function")
@@ -369,9 +368,11 @@ def _init_managers():
 @pytest.fixture
 def init_orchestrator():
     def _init_orchestrator():
+        _OrchestratorFactory._remove_dispatcher()
+
         if _OrchestratorFactory._orchestrator is None:
             _OrchestratorFactory._build_orchestrator()
-        _OrchestratorFactory._build_dispatcher()
+        _OrchestratorFactory._build_dispatcher(force_restart=True)
         _OrchestratorFactory._orchestrator.jobs_to_run = Queue()
         _OrchestratorFactory._orchestrator.blocked_jobs = []
 
@@ -392,7 +393,7 @@ def sql_engine():
 
 
 @pytest.fixture
-def init_sql_repo(tmp_sqlite):
+def init_sql_repo(tmp_sqlite, init_managers):
     Config.configure_core(repository_type="sql", repository_properties={"db_location": tmp_sqlite})
 
     # Clean SQLite database
@@ -401,4 +402,6 @@ def init_sql_repo(tmp_sqlite):
         _SQLConnection._connection = None
     _SQLConnection.init_db()
 
+    init_managers()
+
     return tmp_sqlite

tests/core/cycle/test_cycle_manager.py

@@ -14,8 +14,6 @@
 from taipy.config.common.frequency import Frequency
 from taipy.config.common.scope import Scope
 from taipy.config.config import Config
-from taipy.core._orchestrator._orchestrator_factory import _OrchestratorFactory
-from taipy.core.config.job_config import JobConfig
 from taipy.core.cycle._cycle_manager import _CycleManager
 from taipy.core.cycle.cycle import Cycle
 from taipy.core.cycle.cycle_id import CycleId
@@ -190,8 +188,6 @@ def test_get_cycle_start_date_and_end_date():
 
 
 def test_hard_delete_shared_entities():
-    Config.configure_job_executions(mode=JobConfig._DEVELOPMENT_MODE)
-
     dn_config_1 = Config.configure_data_node("my_input_1", "pickle", scope=Scope.SCENARIO, default_data="testing")
     dn_config_2 = Config.configure_data_node("my_input_2", "pickle", scope=Scope.SCENARIO, default_data="testing")
     dn_config_3 = Config.configure_data_node("my_input_3", "pickle", scope=Scope.CYCLE, default_data="testing")
@@ -219,8 +215,6 @@ def test_hard_delete_shared_entities():
     scenario_config_2 = Config.configure_scenario("scenario_config_2", [task_config_2, task_config_3])
     scenario_config_2.add_sequences({"sequence_3": [task_config_3]})
 
-    _OrchestratorFactory._build_dispatcher()
-
     scenario_1 = _ScenarioManager._create(scenario_config_1)
     scenario_2 = _ScenarioManager._create(scenario_config_1)
     scenario_3 = _ScenarioManager._create(scenario_config_2)

tests/core/cycle/test_cycle_manager_with_sql_repo.py

@@ -14,8 +14,6 @@
 from taipy.config.common.frequency import Frequency
 from taipy.config.common.scope import Scope
 from taipy.config.config import Config
-from taipy.core._orchestrator._orchestrator_factory import _OrchestratorFactory
-from taipy.core.config.job_config import JobConfig
 from taipy.core.cycle._cycle_manager import _CycleManager
 from taipy.core.cycle._cycle_manager_factory import _CycleManagerFactory
 from taipy.core.cycle.cycle import Cycle
@@ -199,7 +197,6 @@ def test_get_cycle_start_date_and_end_date(init_sql_repo):
 
 
 def test_hard_delete_shared_entities(init_sql_repo):
-    Config.configure_job_executions(mode=JobConfig._DEVELOPMENT_MODE)
     _ScenarioManager._repository = _ScenarioManagerFactory._build_repository()
 
     dn_config_1 = Config.configure_data_node("my_input_1", "in_memory", scope=Scope.SCENARIO, default_data="testing")
@@ -228,8 +225,6 @@ def test_hard_delete_shared_entities(init_sql_repo):
     )  # No Frequency so cycle attached to scenarios
     scenario_config_2.add_sequences({"sequence_3": [task_config_3]})
 
-    _OrchestratorFactory._build_dispatcher()
-
     scenario_1 = _ScenarioManager._create(scenario_config_1)
     scenario_2 = _ScenarioManager._create(scenario_config_1)
     scenario_3 = _ScenarioManager._create(scenario_config_2)

tests/core/data/test_data_manager_with_sql_repo.py

@@ -19,7 +19,6 @@
 from taipy.core._version._version_manager import _VersionManager
 from taipy.core.config.data_node_config import DataNodeConfig
 from taipy.core.data._data_manager import _DataManager
-from taipy.core.data._data_manager_factory import _DataManagerFactory
 from taipy.core.data.csv import CSVDataNode
 from taipy.core.data.data_node_id import DataNodeId
 from taipy.core.data.in_memory import InMemoryDataNode
@@ -30,14 +29,8 @@ def file_exists(file_path: str) -> bool:
     return os.path.exists(file_path)
 
 
-def init_managers():
-    _DataManagerFactory._build_manager()._delete_all()
-
-
 class TestDataManager:
     def test_create_data_node_and_modify_properties_does_not_modify_config(self, init_sql_repo):
-        init_managers()
-
         dn_config = Config.configure_data_node(id="name", foo="bar")
         dn = _DataManager._create_and_set(dn_config, None, None)
         assert dn_config.properties.get("foo") == "bar"
@@ -51,23 +44,17 @@ def test_create_data_node_and_modify_properties_does_not_modify_config(self, ini
         assert dn.properties.get("baz") == "qux"
 
     def test_create_raises_exception_with_wrong_type(self, init_sql_repo):
-        init_managers()
-
         wrong_type_dn_config = DataNodeConfig(id="foo", storage_type="bar", scope=DataNodeConfig._DEFAULT_SCOPE)
         with pytest.raises(InvalidDataNodeType):
             _DataManager._create_and_set(wrong_type_dn_config, None, None)
 
     def test_create_from_same_config_generates_new_data_node_and_new_id(self, init_sql_repo):
-        init_managers()
-
         dn_config = Config.configure_data_node(id="foo", storage_type="in_memory")
         dn = _DataManager._create_and_set(dn_config, None, None)
         dn_2 = _DataManager._create_and_set(dn_config, None, None)
         assert dn_2.id != dn.id
 
     def test_create_uses_overridden_attributes_in_config_file(self, init_sql_repo):
-        init_managers()
-
         Config.override(os.path.join(pathlib.Path(__file__).parent.resolve(), "data_sample/config.toml"))
 
         csv_dn_cfg = Config.configure_data_node(id="foo", storage_type="csv", path="bar", has_header=True)
@@ -85,14 +72,10 @@ def test_create_uses_overridden_attributes_in_config_file(self, init_sql_repo):
         assert csv_dn.has_header
 
     def test_get_if_not_exists(self, init_sql_repo):
-        init_managers()
-
         with pytest.raises(ModelNotFound):
             _DataManager._repository._load("test_data_node_2")
 
     def test_get_all(self, init_sql_repo):
-        init_managers()
-
         _DataManager._delete_all()
         assert len(_DataManager._get_all()) == 0
         dn_config_1 = Config.configure_data_node(id="foo", storage_type="in_memory")
@@ -106,8 +89,6 @@ def test_get_all(self, init_sql_repo):
         assert len([dn for dn in _DataManager._get_all() if dn.config_id == "baz"]) == 2
 
     def test_get_all_on_multiple_versions_environment(self, init_sql_repo):
-        init_managers()
-
         # Create 5 data nodes with 2 versions each
         # Only version 1.0 has the data node with config_id = "config_id_1"
         # Only version 2.0 has the data node with config_id = "config_id_6"
@@ -143,8 +124,6 @@ def test_get_all_on_multiple_versions_environment(self, init_sql_repo):
         assert len(_DataManager._get_all_by(filters=[{"version": "2.0", "config_id": "config_id_6"}])) == 1
 
     def test_set(self, init_sql_repo):
-        init_managers()
-
         dn = InMemoryDataNode(
             "config_id",
             Scope.SCENARIO,
@@ -171,7 +150,6 @@ def test_set(self, init_sql_repo):
         assert _DataManager._get(dn.id).config_id == "foo"
 
     def test_delete(self, init_sql_repo):
-        init_managers()
         _DataManager._delete_all()
 
         dn_1 = InMemoryDataNode("config_id", Scope.SCENARIO, id="id_1")
@@ -198,8 +176,6 @@ def test_get_or_create(self, init_sql_repo):
         def _get_or_create_dn(config, *args):
             return _DataManager._bulk_get_or_create([config], *args)[config]
 
-        init_managers()
-
         global_dn_config = Config.configure_data_node(
             id="test_data_node", storage_type="in_memory", scope=Scope.GLOBAL, data="In memory Data Node"
         )
@@ -259,8 +235,6 @@ def _get_or_create_dn(config, *args):
         assert cycle_dn_4.id == cycle_dn_5.id
 
     def test_get_data_nodes_by_config_id(self, init_sql_repo):
-        init_managers()
-
         dn_config_1 = Config.configure_data_node("dn_1", scope=Scope.SCENARIO)
         dn_config_2 = Config.configure_data_node("dn_2", scope=Scope.SCENARIO)
         dn_config_3 = Config.configure_data_node("dn_3", scope=Scope.SCENARIO)
@@ -290,8 +264,6 @@ def test_get_data_nodes_by_config_id(self, init_sql_repo):
         assert sorted([dn_3_1.id]) == sorted([sequence.id for sequence in dn_3_datanodes])
 
     def test_get_data_nodes_by_config_id_in_multiple_versions_environment(self, init_sql_repo):
-        init_managers()
-
         dn_config_1 = Config.configure_data_node("dn_1", scope=Scope.SCENARIO)
         dn_config_2 = Config.configure_data_node("dn_2", scope=Scope.SCENARIO)
 

tests/core/data/test_data_node.py

@@ -20,8 +20,6 @@
 from taipy.config import Config
 from taipy.config.common.scope import Scope
 from taipy.config.exceptions.exceptions import InvalidConfigurationId
-from taipy.core._orchestrator._orchestrator_factory import _OrchestratorFactory
-from taipy.core.config.job_config import JobConfig
 from taipy.core.data._data_manager import _DataManager
 from taipy.core.data.data_node import DataNode
 from taipy.core.data.data_node_id import DataNodeId
@@ -355,8 +353,6 @@ def test_is_up_to_date_across_scenarios(self, current_datetime):
         assert not dn_3.is_up_to_date
 
     def test_do_not_recompute_data_node_valid_but_continue_sequence_execution(self):
-        Config.configure_job_executions(mode=JobConfig._DEVELOPMENT_MODE)
-
         a = Config.configure_data_node("A", "pickle", default_data="A")
         b = Config.configure_data_node("B", "pickle")
         c = Config.configure_data_node("C", "pickle")
@@ -367,8 +363,6 @@ def test_do_not_recompute_data_node_valid_but_continue_sequence_execution(self):
         task_b_d = Config.configure_task("task_b_d", funct_b_d, input=b, output=d)
         scenario_cfg = Config.configure_scenario("scenario", [task_a_b, task_b_c, task_b_d])
 
-        _OrchestratorFactory._build_dispatcher()
-
         scenario = tp.create_scenario(scenario_cfg)
         scenario.submit()
         assert scenario.A.read() == "A"