Merge remote-tracking branch 'openms/main'

Author: t0mdavid-m

settings.json

@@ -26,5 +26,9 @@
         "source_dirs": [
             "example-data/workspaces"
         ]
+    },
+    "max_threads": {
+        "local": 4,
+        "online": 2
     }
 }

src/common/common.py

@@ -32,6 +32,29 @@
 OS_PLATFORM = sys.platform
 
 
+def get_max_threads() -> int:
+    """
+    Get max threads for current deployment mode.
+
+    In local mode, checks for UI override in session state.
+    In online mode, uses the configured value directly.
+
+    Returns:
+        int: Maximum number of threads to use for parallel processing.
+    """
+    settings = st.session_state.get("settings", {})
+    max_threads_config = settings.get("max_threads", {"local": 4, "online": 2})
+
+    if settings.get("online_deployment", False):
+        return max_threads_config.get("online", 2)
+    else:
+        # Local mode: check for UI override, fallback to config default
+        return st.session_state.get(
+            "max_threads_override",
+            max_threads_config.get("local", 4)
+        )
+
+
 def is_safe_workspace_name(name: str) -> bool:
     """
     Check if a workspace name is safe (no path traversal characters).

src/workflow/CommandExecutor.py

@@ -32,8 +32,9 @@ def run_multiple_commands(
         Executes multiple shell commands concurrently in separate threads.
 
         This method leverages threading to run each command in parallel, improving
-        efficiency for batch command execution. Execution time and command results are
-        logged if specified.
+        efficiency for batch command execution. The number of concurrent commands
+        is limited by the max_threads setting, which is distributed between
+        parallel command execution and per-tool thread allocation.
 
         Args:
             commands (list[str]): A list where each element is a list representing
@@ -42,17 +43,26 @@ def run_multiple_commands(
         Returns:
             bool: True if all commands succeeded, False if any failed.
         """
+        from src.common.common import get_max_threads
+
+        # Get thread settings and calculate distribution
+        max_threads = get_max_threads()
+        num_commands = len(commands)
+        parallel_commands = min(num_commands, max_threads)
+
         # Log the start of command execution
-        self.logger.log(f"Running {len(commands)} commands in parallel...", 1)
+        self.logger.log(f"Running {num_commands} commands (max {parallel_commands} parallel, {max_threads} total threads)...", 1)
         start_time = time.time()
 
         results = []
         lock = threading.Lock()
+        semaphore = threading.Semaphore(parallel_commands)
 
         def run_and_track(cmd):
-            success = self.run_command(cmd)
-            with lock:
-                results.append(success)
+            with semaphore:
+                success = self.run_command(cmd)
+                with lock:
+                    results.append(success)
 
         # Initialize a list to keep track of threads
         threads = []
@@ -69,7 +79,7 @@ def run_and_track(cmd):
 
         # Calculate and log the total execution time
         end_time = time.time()
-        self.logger.log(f"Total time to run {len(commands)} commands: {end_time - start_time:.2f} seconds", 1)
+        self.logger.log(f"Total time to run {num_commands} commands: {end_time - start_time:.2f} seconds", 1)
 
         return all(results)
 
@@ -210,6 +220,12 @@ def run_topp(self, tool: str, input_output: dict, custom_params: dict = {}) -> b
         else:
             n_processes = max(io_lengths)
 
+        # Calculate threads per command based on max_threads setting
+        from src.common.common import get_max_threads
+        max_threads = get_max_threads()
+        parallel_commands = min(n_processes, max_threads)
+        threads_per_command = max(1, max_threads // parallel_commands)
+
         commands = []
 
         # Load parameters for non-defaults
@@ -253,6 +269,8 @@ def run_topp(self, tool: str, input_output: dict, custom_params: dict = {}) -> b
                         command += [str(x) for x in v]
                     else:
                         command += [str(v)]
+            # Add threads parameter for TOPP tools
+            command += ["-threads", str(threads_per_command)]
             commands.append(command)
 
             # check if a ini file has been written, if yes use it (contains custom defaults)

src/workflow/StreamlitUI.py

@@ -1108,6 +1108,18 @@ def file_upload_section(self, custom_upload_function) -> None:
     def parameter_section(self, custom_parameter_function) -> None:
         st.toggle("Show advanced parameters", value=False, key="advanced")
 
+        # Display threads configuration for local mode only
+        if not st.session_state.settings.get("online_deployment", False):
+            max_threads_config = st.session_state.settings.get("max_threads", {})
+            default_threads = max_threads_config.get("local", 4)
+            st.number_input(
+                "Threads",
+                min_value=1,
+                value=default_threads,
+                key="max_threads_override",
+                help="Maximum threads for parallel processing. Threads are distributed between parallel commands and per-tool thread allocation."
+            )
+
         # Display preset buttons if presets are available for this workflow
         self.preset_buttons()