[Internal][Executor] Refine the LineExecutionProcessPool to align more closely with a standard process pool

src/promptflow/promptflow/_utils/multimedia_utils.py

@@ -178,6 +178,11 @@ def convert_multimedia_data_to_base64(value: Any, with_type=False, dict_type=Fal
     return _process_recursively(value, process_funcs=to_base64_funcs)
 
 
+def convert_multimedia_data_to_string(value: Any, inplace=False):
+    serialization_funcs = {Image: partial(Image.serialize, **{"encoder": None})}
+    return _process_recursively(value, process_funcs=serialization_funcs, inplace=inplace)
+
+
 # TODO: Move this function to a more general place and integrate serialization to this function.
 def _process_recursively(value: Any, process_funcs: Dict[type, Callable] = None, inplace: bool = False) -> dict:
     if process_funcs:

src/promptflow/promptflow/_utils/process_utils.py

@@ -2,23 +2,63 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # ---------------------------------------------------------
 
+import logging
+import os
 import signal
 
+import psutil
+
+from promptflow._utils.logger_utils import bulk_logger
+
 
 def block_terminate_signal_to_parent():
-    # In uvicorn app, the main process listens for requests and handles graceful shutdowns through
-    # signal listeners set up at initialization. These listeners use a file descriptor for event notifications.
+    """
+    In uvicorn app, the main process listens for requests and handles graceful shutdowns through
+    signal listeners set up at initialization. These listeners use a file descriptor for event notifications.
 
-    # However, when a child process is forked within the application, it inherits this file descriptor,
-    # leading to an issue where signals sent to terminate the child process are also intercepted by the main process,
-    # causing an unintended shutdown of the entire application.
+    However, when a child process is forked within the application, it inherits this file descriptor,
+    leading to an issue where signals sent to terminate the child process are also intercepted by the main process,
+    causing an unintended shutdown of the entire application.
 
-    # To avoid this, we should return the default behavior of signal handlers for child process and call
-    # signal.set_wakeup_fd(-1) in the child process to prevent it from using the parent's file descriptor
-    # and avoiding unintended shutdowns of the main process.
+    To avoid this, we should return the default behavior of signal handlers for child process and call
+    signal.set_wakeup_fd(-1) in the child process to prevent it from using the parent's file descriptor
+    and avoiding unintended shutdowns of the main process.
 
-    # References: https://github.com/tiangolo/fastapi/discussions/7442
+    References: https://github.com/tiangolo/fastapi/discussions/7442
+    """
     signal.set_wakeup_fd(-1)
 
     signal.signal(signal.SIGTERM, signal.SIG_DFL)
     signal.signal(signal.SIGINT, signal.SIG_DFL)
+
+
+def get_available_max_worker_count(logger: logging.Logger = bulk_logger):
+    """
+    When creating processes using the spawn method, it consumes certain resources.
+    So we can use this method to determine how many workers can be maximally created.
+    """
+    pid = os.getpid()
+    mem_info = psutil.virtual_memory()
+    available_memory = mem_info.available / (1024 * 1024)  # in MB
+    process = psutil.Process(pid)
+    process_memory_info = process.memory_info()
+    process_memory = process_memory_info.rss / (1024 * 1024)  # in MB
+    estimated_available_worker_count = int(available_memory // process_memory)
+    if estimated_available_worker_count < 1:
+        # TODO: For the case of vector db, Optimize execution logic
+        # 1. Let the main process not consume memory because it does not actually invoke
+        # 2. When the degree of parallelism is 1, main process executes the task directly
+        #    and not create the child process
+        logger.warning(
+            f"Current system's available memory is {available_memory}MB, less than the memory "
+            f"{process_memory}MB required by the process. The maximum available worker count is 1."
+        )
+        estimated_available_worker_count = 1
+    else:
+        logger.info(
+            f"Current system's available memory is {available_memory}MB, "
+            f"memory consumption of current process is {process_memory}MB, "
+            f"estimated available worker count is {available_memory}/{process_memory} "
+            f"= {estimated_available_worker_count}"
+        )
+    return estimated_available_worker_count

src/promptflow/promptflow/batch/_batch_engine.py

@@ -346,7 +346,7 @@ async def _exec(
         # execute lines
         is_timeout = False
         if isinstance(self._executor_proxy, PythonExecutorProxy):
-            results, is_timeout = self._executor_proxy._exec_batch(
+            results, is_timeout = await self._executor_proxy._exec_batch(
                 inputs_to_run,
                 output_dir,
                 run_id,

src/promptflow/promptflow/batch/_python_executor_proxy.py

@@ -44,7 +44,7 @@ async def exec_aggregation_async(
         with self._flow_executor._run_tracker.node_log_manager:
             return self._flow_executor._exec_aggregation(batch_inputs, aggregation_inputs, run_id=run_id)
 
-    def _exec_batch(
+    async def _exec_batch(
         self,
         batch_inputs: List[Mapping[str, Any]],
         output_dir: Path,
@@ -79,7 +79,7 @@ def _exec_batch(
                 worker_count=worker_count,
             ) as pool:
                 line_number = [batch_input["line_number"] for batch_input in batch_inputs]
-                line_results = pool.run(zip(line_number, batch_inputs))
+                line_results = await pool.run(zip(line_number, batch_inputs))
 
             # For bulk run, currently we need to add line results to run_tracker
             self._flow_executor._add_line_results(line_results, run_tracker)

src/promptflow/promptflow/executor/_errors.py

@@ -189,6 +189,12 @@ def __init__(self, line_number, timeout):
         )
 
 
+class ThreadCrashError(SystemErrorException):
+    """Exception raised when thread crashed."""
+
+    pass
+
+
 class ProcessCrashError(UserErrorException):
     """Exception raised when process crashed."""