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

PeiwenGaoMS · 2024-03-06T11:04:50Z

Description

Background

After splitting the executor and runtime into two separate containers, the implementation of batch run in the executor server also requires a process pool to handle execution requests line by line. Therefore, we need the line process pool to have the capability to process individual lines. As a process pool, it should also have the ability to submit one or multiple tasks at a time, consistent with the python process pool interface.

Usages

Use with context

with LineExecutionProcessPool(...) as pool:
      line_results = await pool.run(zip(line_number, batch_inputs))

Use method

pool = LineExecutionProcessPool(...)
pool.start()
line_results = await pool.run(zip(line_number, batch_inputs))
pool.close()

Public Functions

start: Create task queue, input/output queues, and start processes and monitor thread pool.
close: Send terminate signal to monitor threads, end processes and close thread pool.
run: Put all line inputs to task queue and get the line results list.
submit: Put one line input to task queue and get one line result.

Implementation Process

Main Differences from Previous Implementation

Start the monitor threads at the beginning instead of doing it in the run method.
The monitor thread will not exit due to an empty task queue. It exits only when the batch run times out or when a termination signal is received from the task queue.
Update run to async function.

Code Changes Sunmary

This pull request includes changes to various parts of the promptflow system, with the main focus being the addition of new functionalities and the refactoring of existing code for better performance and readability. The most significant changes include the addition of a new function to convert multimedia data to string, the introduction of a method to determine the maximum number of workers that can be created, the conversion of some functions to asynchronous, and the implementation of new exception classes.

New functionalities:

src/promptflow/promptflow/_utils/multimedia_utils.py: Added a new function convert_multimedia_data_to_string to convert multimedia data to string.
src/promptflow/promptflow/_utils/process_utils.py: Introduced a new function get_available_max_worker_count to determine the maximum number of workers that can be created.
src/promptflow/promptflow/executor/_errors.py: Implemented new exception classes ThreadCrashError and ProcessCrashError to handle thread and process crashes.

Refactoring:

src/promptflow/promptflow/batch/_batch_engine.py, src/promptflow/promptflow/batch/_python_executor_proxy.py: Converted some functions to asynchronous for better performance. [1] [2] [3]
src/promptflow/promptflow/executor/_process_manager.py: Refactored the AbstractProcessManager and SpawnProcessManager classes, added new methods for process management, and made changes to improve readability. [1] [2] [3] [4] [5] [6] [7] [8] [9] [10]
src/promptflow/tests/executor/e2etests/test_batch_timeout.py, src/promptflow/tests/executor/unittests/_utils/test_process_utils.py, src/promptflow/tests/executor/unittests/executor/test_line_execution_process_pool.py: Updated test cases to reflect the changes made in the codebase. [1] [2] [3] [4]

Additions:

src/promptflow/promptflow/storage/_queue_run_storage.py: Added a new storage class QueueRunStorage for storing run information in a queue.

All Promptflow Contribution checklist:

The pull request does not introduce [breaking changes].
CHANGELOG is updated for new features, bug fixes or other significant changes.
I have read the contribution guidelines.
Create an issue and link to the pull request to get dedicated review from promptflow team. Learn more: suggested workflow.

General Guidelines and Best Practices

Title of the pull request is clear and informative.
There are a small number of commits, each of which have an informative message. This means that previously merged commits do not appear in the history of the PR. For more information on cleaning up the commits in your PR, see this page.

Testing Guidelines

Pull request includes test coverage for the included changes.

…evs/peiwen/refactor_line_process_pool

github-actions · 2024-03-06T11:07:31Z