Merge pull request #1026 from Avaiga/feature/#865-default-max_nb_of_workers-to-2

Feature/#865 - Modify the default value of max_nb_of_workers on standalone mode to 2

taipy/config/config.pyi

@@ -851,7 +851,7 @@ class Config:
                 Possible values are: *"standalone"* (the default value) or *"development"*.
             max_nb_of_workers (Optional[int, str]): Parameter used only in default *"standalone"* mode.
                 This indicates the maximum number of jobs able to run in parallel.<br/>
-                The default value is 1.<br/>
+                The default value is 2.<br/>
                 A string can be provided to dynamically set the value using an environment
                 variable. The string must follow the pattern: `ENV[&lt;env_var&gt;]` where
                 `&lt;env_var&gt;` is the name of an environment variable.

taipy/core/_orchestrator/_dispatcher/_standalone_job_dispatcher.py

@@ -25,9 +25,11 @@ from ._task_function_wrapper import _TaskFunctionWrapper
 class _StandaloneJobDispatcher(_JobDispatcher):
     """Manages job dispatching (instances of `Job^` class) in an asynchronous way using a ProcessPoolExecutor."""
 
+    _DEFAULT_MAX_NB_OF_WORKERS = 2
+
     def __init__(self, orchestrator: _AbstractOrchestrator, subproc_initializer: Optional[Callable] = None):
         super().__init__(orchestrator)
-        max_workers = Config.job_config.max_nb_of_workers or 1
+        max_workers = Config.job_config.max_nb_of_workers or self._DEFAULT_MAX_NB_OF_WORKERS
         self._executor: Executor = ProcessPoolExecutor(
             max_workers=max_workers,
             initializer=subproc_initializer,

taipy/core/config/job_config.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 copy import copy
 from typing import Any, Dict, Optional, Union
 
@@ -35,6 +36,7 @@ class JobConfig(UniqueSection):
     _STANDALONE_MODE = "standalone"
     _DEVELOPMENT_MODE = "development"
     _DEFAULT_MODE = _DEVELOPMENT_MODE
+    _DEFAULT_MAX_NB_OF_WORKERS = 2
     _MODES = [_STANDALONE_MODE, _DEVELOPMENT_MODE]
 
     def __init__(self, mode: Optional[str] = None, **properties):
@@ -87,7 +89,7 @@ class JobConfig(UniqueSection):
                 Possible values are: *"standalone"* (the default value) or *"development"*.
             max_nb_of_workers (Optional[int, str]): Parameter used only in default *"standalone"* mode.
                 This indicates the maximum number of jobs able to run in parallel.<br/>
-                The default value is 1.<br/>
+                The default value is 2.<br/>
                 A string can be provided to dynamically set the value using an environment
                 variable. The string must follow the pattern: `ENV[&lt;env_var&gt;]` where
                 `&lt;env_var&gt;` is the name of an environment variable.
@@ -117,14 +119,12 @@ class JobConfig(UniqueSection):
         """True if the config is set to development mode"""
         return self.mode == self._DEVELOPMENT_MODE
 
-    @classmethod
-    def get_default_config(cls, mode: str) -> Dict[str, Any]:
-        if cls.is_standalone:  # type: ignore
-            return {"max_nb_of_workers": 1}
-        if cls.is_development:
+    def get_default_config(self, mode: str) -> Dict[str, Any]:
+        if self.is_standalone:
+            return {"max_nb_of_workers": self._DEFAULT_MAX_NB_OF_WORKERS}
+        if self.is_development:
             return {}
         raise ModeNotAvailable(mode)
 
-    @classmethod
-    def _create_config(cls, mode, **properties):
-        return {**cls.get_default_config(mode), **properties}
+    def _create_config(self, mode, **properties):
+        return {**self.get_default_config(mode), **properties}

tests/core/_orchestrator/_dispatcher/test_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 Future, ProcessPoolExecutor
 from unittest import mock
 from unittest.mock import call
@@ -41,7 +42,7 @@ def test_init_default():
 
     assert job_dispatcher.orchestrator == orchestrator
     assert job_dispatcher.lock == orchestrator.lock
-    assert job_dispatcher._nb_available_workers == 1
+    assert job_dispatcher._nb_available_workers == 2
     assert isinstance(job_dispatcher._executor, ProcessPoolExecutor)
 
 
@@ -81,7 +82,7 @@ def test_dispatch_job():
 
 def test_can_execute():
     dispatcher = _StandaloneJobDispatcher(_OrchestratorFactory._orchestrator)
-    assert dispatcher._nb_available_workers == 1
+    assert dispatcher._nb_available_workers == 2
     assert dispatcher._can_execute()
     dispatcher._nb_available_workers = 0
     assert not dispatcher._can_execute()
@@ -94,11 +95,11 @@ def test_can_execute():
 def test_release_worker():
     dispatcher = _StandaloneJobDispatcher(_OrchestratorFactory._orchestrator)
 
-    assert dispatcher._nb_available_workers == 1
-    dispatcher._release_worker(None)
     assert dispatcher._nb_available_workers == 2
     dispatcher._release_worker(None)
     assert dispatcher._nb_available_workers == 3
+    dispatcher._release_worker(None)
+    assert dispatcher._nb_available_workers == 4
 
 
 def test_update_job_status_from_future():

tests/core/_orchestrator/test_orchestrator_factory.py

@@ -105,11 +105,13 @@ def test_rebuild_standalone_dispatcher_and_force_restart():
 
 
 def test_build_unknown_dispatcher():
-    Config.configure_job_executions(mode="UNKNOWN")
+    with pytest.raises(ModeNotAvailable):
+        Config.configure_job_executions(mode="UNKNOWN")
+
+    Config.job_config.mode = "UNKNOWN"
     _OrchestratorFactory._build_orchestrator()
     with pytest.raises(ModeNotAvailable):
         _OrchestratorFactory._build_dispatcher()
-        assert _OrchestratorFactory._dispatcher is None
 
 
 def test_remove_dispatcher_not_built():

tests/core/config/test_config_serialization.py

@@ -113,7 +113,6 @@ def test_read_write_toml_configuration_file():
 
 [JOB]
 mode = "development"
-max_nb_of_workers = "1:int"
 
 [CORE]
 root_folder = "./taipy/"
@@ -205,7 +204,7 @@ sequence1 = [ "test_task:SECTION",]
     assert len(Config.unique_sections) == 3
 
     assert Config.unique_sections[JobConfig.name].mode == "development"
-    assert Config.unique_sections[JobConfig.name].max_nb_of_workers == 1
+    assert Config.unique_sections[JobConfig.name].max_nb_of_workers is None
 
     assert Config.unique_sections[MigrationConfig.name].migration_fcts["1.0"] == {"test_csv_dn": migrate_csv_path}
 
@@ -285,8 +284,7 @@ def test_read_write_json_configuration_file():
         """{
 "TAIPY": {},
 "JOB": {
-"mode": "development",
-"max_nb_of_workers": "1:int"
+"mode": "development"
 },
 "CORE": {
 "root_folder": "./taipy/",
@@ -407,7 +405,7 @@ def test_read_write_json_configuration_file():
     assert len(Config.unique_sections) == 3
 
     assert Config.unique_sections[JobConfig.name].mode == "development"
-    assert Config.unique_sections[JobConfig.name].max_nb_of_workers == 1
+    assert Config.unique_sections[JobConfig.name].max_nb_of_workers is None
 
     assert Config.unique_sections[MigrationConfig.name].migration_fcts["1.0"] == {"test_csv_dn": migrate_csv_path}
 

tests/core/config/test_job_config.py

@@ -15,19 +15,19 @@ from taipy.config.config import Config
 def test_job_config():
     assert Config.job_config.mode == "development"
 
-    job_c = Config.configure_job_executions(mode="standalone", max_nb_of_workers=2)
+    job_c = Config.configure_job_executions(mode="standalone", max_nb_of_workers=3)
     assert job_c.mode == "standalone"
-    assert job_c.max_nb_of_workers == 2
+    assert job_c.max_nb_of_workers == 3
 
     assert Config.job_config.mode == "standalone"
-    assert Config.job_config.max_nb_of_workers == 2
+    assert Config.job_config.max_nb_of_workers == 3
 
     Config.configure_job_executions(foo="bar")
     assert Config.job_config.foo == "bar"
 
 
 def test_clean_config():
-    job_config = Config.configure_job_executions(mode="standalone", max_nb_of_workers=2, prop="foo")
+    job_config = Config.configure_job_executions(mode="standalone", max_nb_of_workers=3, prop="foo")
 
     assert Config.job_config is job_config
 
@@ -37,5 +37,5 @@ def test_clean_config():
     assert Config.job_config is job_config
 
     assert job_config.mode == "development"
-    assert job_config._config == {"max_nb_of_workers": 1}
+    assert job_config._config == {}
     assert job_config.properties == {}

tests/core/config/test_override_config.py

@@ -83,7 +83,8 @@ max_nb_of_workers = -1
 [SCENARIO.qux]
 """
     )
-    assert Config.job_config.max_nb_of_workers == 1
+    assert Config.job_config.mode == "development"
+    assert Config.job_config.max_nb_of_workers is None
     assert len(Config.data_nodes) == 1
     assert len(Config.tasks) == 1
     assert len(Config.scenarios) == 1
@@ -110,7 +111,8 @@ max_nb_of_workers = "ENV[FOO]:int"
 start_executor = "ENV[BAR]"
 """
     )
-    assert Config.job_config.max_nb_of_workers == 1
+    assert Config.job_config.mode == "development"
+    assert Config.job_config.max_nb_of_workers is None
     assert not Config.job_config.start_executor
 
     with mock.patch.dict(os.environ, {"FOO": "6", "BAR": "TRUe"}):
@@ -198,7 +200,8 @@ max_nb_of_workers = 10
     )
 
     # Default config is applied
-    assert Config.job_config.max_nb_of_workers == 1
+    assert Config.job_config.mode == "development"
+    assert Config.job_config.max_nb_of_workers is None
 
     # Code config is applied
     Config.configure_job_executions(max_nb_of_workers=-1)
@@ -229,7 +232,8 @@ max_nb_of_workers = 10
 
     with mock.patch.dict(os.environ, {"FOO": "/data/csv", "BAR": "/baz/data/csv"}):
         # Default config is applied
-        assert Config.job_config.max_nb_of_workers == 1
+        assert Config.job_config.mode == "development"
+        assert Config.job_config.max_nb_of_workers is None
 
         # Code config is applied
         Config.configure_job_executions(max_nb_of_workers=-1)

tests/core/test_core_cli.py

@@ -473,14 +473,14 @@ def test_modified_job_configuration_dont_block_application_run(caplog, init_conf
     scenario_config = config_scenario()
     with patch("sys.argv", ["prog", "--experiment", "1.0"]):
         core = Core()
-        Config.configure_job_executions(mode="standalone", max_nb_of_workers=5)
+        Config.configure_job_executions(mode="standalone", max_nb_of_workers=3)
         core.run(force_restart=True)
         scenario = _ScenarioManager._create(scenario_config)
         jobs = taipy.submit(scenario).jobs
         assert_true_after_time(lambda: all(job.is_finished() for job in jobs))
         error_message = str(caplog.text)
         assert 'JOB "mode" was modified' in error_message
-        assert 'JOB "max_nb_of_workers" was modified' in error_message
+        assert 'JOB "max_nb_of_workers" was added' in error_message
         core.stop()
 
 
@@ -509,6 +509,7 @@ def test_modified_config_properties_without_force(caplog, init_config):
     error_message = str(caplog.text)
 
     assert 'DATA_NODE "d3" was added' in error_message
+    assert 'JOB "max_nb_of_workers" was added' in error_message
 
     assert 'DATA_NODE "d0" was removed' in error_message
 
@@ -516,7 +517,6 @@ def test_modified_config_properties_without_force(caplog, init_config):
     assert 'CORE "root_folder" was modified' in error_message
     assert 'CORE "repository_type" was modified' in error_message
     assert 'JOB "mode" was modified' in error_message
-    assert 'JOB "max_nb_of_workers" was modified' in error_message
     assert 'SCENARIO "my_scenario" has attribute "frequency" modified' in error_message
     assert 'SCENARIO "my_scenario" has attribute "tasks" modified' in error_message
     assert 'TASK "my_task" has attribute "inputs" modified' in error_message
@@ -557,7 +557,7 @@ def config_scenario_2():
         repository_type="bar",
         repository_properties={"foo": "bar"},
     )
-    Config.configure_job_executions(mode="standalone", max_nb_of_workers=5)
+    Config.configure_job_executions(mode="standalone", max_nb_of_workers=3)
     data_node_1_config = Config.configure_data_node(
         id="d1", storage_type="pickle", default_data="abc", scope=Scope.SCENARIO
     )

tests/core/test_core_cli_with_sql_repo.py

@@ -483,13 +483,13 @@ def test_modify_config_properties_without_force(caplog, init_sql_repo, init_conf
     error_message = str(caplog.text)
 
     assert 'DATA_NODE "d3" was added' in error_message
+    assert 'JOB "max_nb_of_workers" was added' in error_message
 
     assert 'DATA_NODE "d0" was removed' in error_message
 
     assert 'DATA_NODE "d2" has attribute "default_path" modified' in error_message
     assert 'CORE "root_folder" was modified' in error_message
     assert 'JOB "mode" was modified' in error_message
-    assert 'JOB "max_nb_of_workers" was modified' in error_message
     assert 'SCENARIO "my_scenario" has attribute "frequency" modified' in error_message
     assert 'SCENARIO "my_scenario" has attribute "tasks" modified' in error_message
     assert 'TASK "my_task" has attribute "inputs" modified' in error_message
@@ -517,7 +517,7 @@ def test_modify_job_configuration_dont_stop_application(caplog, init_sql_repo, i
     scenario_config = config_scenario()
 
     with patch("sys.argv", ["prog", "--experiment", "1.0"]):
-        Config.configure_job_executions(mode="standalone", max_nb_of_workers=2)
+        Config.configure_job_executions(mode="standalone", max_nb_of_workers=3)
         core = Core()
         core.run(force_restart=True)
         scenario = _ScenarioManager._create(scenario_config)
@@ -526,7 +526,7 @@ def test_modify_job_configuration_dont_stop_application(caplog, init_sql_repo, i
         assert_true_after_time(lambda: all(job.is_finished() for job in jobs))
         error_message = str(caplog.text)
         assert 'JOB "mode" was modified' in error_message
-        assert 'JOB "max_nb_of_workers" was modified' in error_message
+        assert 'JOB "max_nb_of_workers" was added' in error_message
         core.stop()
         assert_true_after_time(lambda: core._dispatcher is None)
 
@@ -558,7 +558,7 @@ def config_scenario_2():
         # Changing the "storage_folder" will fail since older versions are stored in older folder
         # storage_folder="foo_storage",
     )
-    Config.configure_job_executions(mode="standalone", max_nb_of_workers=5)
+    Config.configure_job_executions(mode="standalone", max_nb_of_workers=3)
     data_node_1_config = Config.configure_data_node(
         id="d1", storage_type="pickle", default_data="abc", scope=Scope.SCENARIO
     )

tests/core/test_taipy.py

@@ -40,7 +40,6 @@ from taipy.core import (
 from taipy.core._orchestrator._orchestrator_factory import _OrchestratorFactory
 from taipy.core._version._version_manager import _VersionManager
 from taipy.core.config.data_node_config import DataNodeConfig
-from taipy.core.config.job_config import JobConfig
 from taipy.core.config.scenario_config import ScenarioConfig
 from taipy.core.cycle._cycle_manager import _CycleManager
 from taipy.core.data._data_manager import _DataManager
@@ -594,7 +593,6 @@ class TestTaipy:
             mck.assert_called_once_with(task)
 
     def test_block_config_when_core_is_running(self):
-        Config.configure_job_executions(mode=JobConfig._STANDALONE_MODE)
         input_cfg_1 = Config.configure_data_node(id="i1", storage_type="pickle", scope=Scope.SCENARIO, default_data=1)
         output_cfg_1 = Config.configure_data_node(id="o1", storage_type="pickle", scope=Scope.SCENARIO)
         task_cfg_1 = Config.configure_task("t1", print, input_cfg_1, output_cfg_1)