From 588a775fafe29a17be54ff219decce9b358b1684 Mon Sep 17 00:00:00 2001 From: Henry Ruhs Date: Tue, 3 Nov 2026 16:54:05 +0100 Subject: [PATCH] move process lifecycle to the job (#1266) Claude-Session: https://claude.ai/code/session_01Cd6JuxiVXYNPCDii9go2eh Co-authored-by: Claude Opus 5.5 --- facefusion/apis/endpoints/jobs.py | 31 +++++++- facefusion/apis/jobs_helper.py | 10 --- facefusion/benchmarker.py | 6 +- facefusion/core.py | 1 - facefusion/download.py | 6 +- facefusion/ffmpeg.py | 3 +- facefusion/jobs/job_runner.py | 6 ++ facefusion/process_manager.py | 9 +-- .../workflows/modules/audio_to_image.py | 5 -- .../modules/audio_to_image_as_frames.py | 5 -- .../workflows/modules/image_to_image.py | 5 -- .../workflows/modules/image_to_video.py | 6 +- .../modules/image_to_video_as_frames.py | 5 -- tests/apis/endpoints/test_jobs.py | 79 ++++++++++++++++--- tests/jobs/test_job_runner.py | 10 ++- tests/test_download.py | 21 +++++ tests/test_process_manager.py | 10 ++- 17 files changed, 157 insertions(+), 61 deletions(-) diff --git a/facefusion/apis/endpoints/jobs.py b/facefusion/apis/endpoints/jobs.py index b4b3cadb..9a021e82 100644 --- a/facefusion/apis/endpoints/jobs.py +++ b/facefusion/apis/endpoints/jobs.py @@ -8,7 +8,7 @@ from starlette.responses import JSONResponse from starlette.status import HTTP_200_OK, HTTP_201_CREATED, HTTP_202_ACCEPTED, HTTP_400_BAD_REQUEST, HTTP_404_NOT_FOUND, HTTP_409_CONFLICT import facefusion.core -from facefusion import args_helper, state_manager, translator +from facefusion import args_helper, process_manager, state_manager, translator from facefusion.apis import jobs_helper from facefusion.apis.job_validator import validate_status from facefusion.filesystem import create_directory, get_file_extension, is_directory @@ -108,7 +108,8 @@ async def update_jobs(request : Request) -> JSONResponse: if action == 'cancel': if job_manager.cancel_jobs(state_manager.get_item('halt_on_error')): - jobs_helper.stop_fork_session() + if not process_manager.is_pending(): + process_manager.stop() return JSONResponse( { @@ -121,6 +122,12 @@ async def update_jobs(request : Request) -> JSONResponse: }, status_code = HTTP_400_BAD_REQUEST) if action == 'run': + if not process_manager.is_pending(): + return JSONResponse( + { + 'message': translator.get('job_all_not_started', 'facefusion.apis') + }, status_code = HTTP_409_CONFLICT) + if job_manager.find_job_ids('queued'): run_jobs_task = BackgroundTask(partial(job_runner.run_jobs, facefusion.core.process_step, state_manager.get_item('halt_on_error'))) @@ -135,6 +142,12 @@ async def update_jobs(request : Request) -> JSONResponse: }, status_code = HTTP_400_BAD_REQUEST) if action == 'retry': + if not process_manager.is_pending(): + return JSONResponse( + { + 'message': translator.get('job_all_not_retried', 'facefusion.apis') + }, status_code = HTTP_409_CONFLICT) + if job_manager.find_job_ids('queued'): return JSONResponse( { @@ -187,7 +200,7 @@ async def update_job(request : Request) -> JSONResponse: if job_manager.cancel_job(job_id): if is_started: - jobs_helper.stop_fork_session() + process_manager.stop() return JSONResponse( { @@ -200,6 +213,12 @@ async def update_job(request : Request) -> JSONResponse: }, status_code = HTTP_400_BAD_REQUEST) if action == 'run': + if not process_manager.is_pending(): + return JSONResponse( + { + 'message': translator.get('job_not_started', 'facefusion.apis') + }, status_code = HTTP_409_CONFLICT) + if job_id in job_manager.find_job_ids('queued'): run_job_tasks = BackgroundTasks() run_job_tasks.add_task(partial(job_runner.run_job, job_id, facefusion.core.process_step)) @@ -216,6 +235,12 @@ async def update_job(request : Request) -> JSONResponse: }, status_code = HTTP_400_BAD_REQUEST) if action == 'retry': + if not process_manager.is_pending(): + return JSONResponse( + { + 'message': translator.get('job_not_retried', 'facefusion.apis') + }, status_code = HTTP_409_CONFLICT) + if job_manager.find_job_ids('queued'): return JSONResponse( { diff --git a/facefusion/apis/jobs_helper.py b/facefusion/apis/jobs_helper.py index b13b6fc8..281e9174 100644 --- a/facefusion/apis/jobs_helper.py +++ b/facefusion/apis/jobs_helper.py @@ -1,4 +1,3 @@ -from facefusion import process_manager, session_context, session_manager from facefusion.apis import asset_store from facefusion.filesystem import is_file from facefusion.jobs import job_manager @@ -14,12 +13,3 @@ def create_output_asset(job_id : JobId) -> None: if output_path and is_file(output_path): asset_store.create_asset('output', output_path) - -def stop_fork_session() -> None: - owner_id = session_manager.resolve_owner_id() - fork_id = session_manager.find_fork_session_id(owner_id) - - if fork_id: - session_context.set_session_id(fork_id) - process_manager.stop() - session_context.set_session_id(owner_id) diff --git a/facefusion/benchmarker.py b/facefusion/benchmarker.py index 15c0d859..1ff7be03 100644 --- a/facefusion/benchmarker.py +++ b/facefusion/benchmarker.py @@ -6,7 +6,7 @@ from time import perf_counter from typing import Iterator, List import facefusion.choices -from facefusion import content_analyser, core, face_store, state_manager +from facefusion import content_analyser, core, face_store, process_manager, state_manager from facefusion.cli_helper import render_table from facefusion.download import conditional_download, resolve_download_url from facefusion.filesystem import get_file_extension @@ -56,6 +56,8 @@ def cycle(cycle_count : int) -> BenchmarkCycleSet: video_frame_total = count_video_frame_total(state_manager.get_item('target_path')) state_manager.init_item('output_video_fps', detect_video_fps(state_manager.get_item('target_path'))) + process_manager.start() + if state_manager.get_item('benchmark_mode') == 'warm': core.conditional_process() @@ -70,6 +72,8 @@ def cycle(cycle_count : int) -> BenchmarkCycleSet: end_time = perf_counter() process_times.append(end_time - start_time) + process_manager.end() + average_run = round(statistics.mean(process_times), 2) fastest_run = round(min(process_times), 2) slowest_run = round(max(process_times), 2) diff --git a/facefusion/core.py b/facefusion/core.py index bb9602e8..b19b6b0f 100755 --- a/facefusion/core.py +++ b/facefusion/core.py @@ -342,7 +342,6 @@ def process_step(job_id : JobId, step_index : StepIndex, step_args : Args) -> Er [ state_manager.STATE_SET, content_store.CONTENT_STORE, - process_manager.PROCESS_STORE, video_manager.VIDEO_POOL_STORE ] diff --git a/facefusion/download.py b/facefusion/download.py index 19d6b322..f3385c03 100644 --- a/facefusion/download.py +++ b/facefusion/download.py @@ -74,6 +74,7 @@ def ping_static_url(url : str) -> bool: def conditional_download_files(file_set : DownloadSet) -> bool: + process_state = process_manager.get_state() process_manager.check() for file in file_set.values(): @@ -84,7 +85,9 @@ def conditional_download_files(file_set : DownloadSet) -> bool: conditional_download(os.path.dirname(file_path), [ file_url ]) is_valid = conditional_validate_files(file_set) - process_manager.end() + + if process_manager.is_checking(): + process_manager.set_state(process_state) return is_valid @@ -118,6 +121,7 @@ def resolve_download_url(base_name : str, file_name : str) -> Optional[str]: for download_provider in download_providers: download_url = resolve_download_url_by_provider(download_provider, base_name, file_name) + if download_url: return download_url diff --git a/facefusion/ffmpeg.py b/facefusion/ffmpeg.py index de3a8c7c..0e917c1d 100644 --- a/facefusion/ffmpeg.py +++ b/facefusion/ffmpeg.py @@ -346,8 +346,7 @@ def concat_video(output_path : str, temp_output_paths : List[str]) -> bool: ffmpeg_builder.set_faststart(output_video_format), ffmpeg_builder.force_output(output_path) ) - process = run_ffmpeg(commands) - process.communicate() + process = await_process(run_ffmpeg(commands)) remove_file(concat_video_path) return process.returncode == 0 diff --git a/facefusion/jobs/job_runner.py b/facefusion/jobs/job_runner.py index 78259eed..defc8e36 100644 --- a/facefusion/jobs/job_runner.py +++ b/facefusion/jobs/job_runner.py @@ -1,5 +1,6 @@ import os +from facefusion import process_manager from facefusion.ffmpeg import concat_video from facefusion.filesystem import are_images, are_videos, copy_file, create_directory, is_directory, is_file, move_directory, move_file, remove_directory, remove_file, resolve_file_paths from facefusion.jobs import job_helper, job_manager @@ -10,11 +11,16 @@ def run_job(job_id : JobId, process_step : ProcessStep) -> bool: queued_job_ids = job_manager.find_job_ids('queued') if job_id in queued_job_ids: + process_manager.start() + if run_steps(job_id, process_step) and job_id in job_manager.find_job_ids('queued') and finalize_steps(job_id): clean_steps(job_id) + process_manager.end() return job_manager.move_job_file(job_id, 'completed') clean_steps(job_id) + process_manager.end() + if job_id in job_manager.find_job_ids('queued'): job_manager.move_job_file(job_id, 'failed') return False diff --git a/facefusion/process_manager.py b/facefusion/process_manager.py index 17a580cc..ed8e02d4 100644 --- a/facefusion/process_manager.py +++ b/facefusion/process_manager.py @@ -12,14 +12,13 @@ def init() -> None: def get_state() -> ProcessState: - session_id = get_session_id() owner_id = resolve_owner_id() local_id = resolve_local_id() - if store_creator.get_content(PROCESS_STORE, owner_id) == 'stopping' or store_creator.get_content(PROCESS_STORE, local_id) == 'stopping': + if store_creator.get_content(PROCESS_STORE, local_id) == 'stopping': return 'stopping' - return store_creator.get_content(PROCESS_STORE, session_id) + return store_creator.get_content(PROCESS_STORE, owner_id) def is_checking() -> bool: @@ -39,8 +38,8 @@ def is_pending() -> bool: def set_state(process_state : ProcessState) -> None: - session_id = get_session_id() - store_creator.set_content(PROCESS_STORE, session_id, process_state) + owner_id = resolve_owner_id() + store_creator.set_content(PROCESS_STORE, owner_id, process_state) def check() -> None: diff --git a/facefusion/workflows/modules/audio_to_image.py b/facefusion/workflows/modules/audio_to_image.py index a5d1e33c..df4ab48f 100644 --- a/facefusion/workflows/modules/audio_to_image.py +++ b/facefusion/workflows/modules/audio_to_image.py @@ -1,6 +1,5 @@ from functools import partial -from facefusion import process_manager from facefusion.types import ErrorCode from facefusion.workflows.as_frames import create_temp_frames from facefusion.workflows.core import clear, process_frames, setup @@ -22,14 +21,10 @@ def process(start_time : float) -> ErrorCode: clear ] - process_manager.start() - for task in tasks: error_code = task() #type:ignore[operator] if error_code > 0: - process_manager.end() return error_code - process_manager.end() return 0 diff --git a/facefusion/workflows/modules/audio_to_image_as_frames.py b/facefusion/workflows/modules/audio_to_image_as_frames.py index 9f14bf25..aa8295ae 100644 --- a/facefusion/workflows/modules/audio_to_image_as_frames.py +++ b/facefusion/workflows/modules/audio_to_image_as_frames.py @@ -1,6 +1,5 @@ from functools import partial -from facefusion import process_manager from facefusion.types import ErrorCode from facefusion.workflows.as_frames import copy_temp_frames, create_temp_frames, finalize_frames from facefusion.workflows.core import clear, process_frames, setup @@ -20,14 +19,10 @@ def process(start_time : float) -> ErrorCode: clear ] - process_manager.start() - for task in tasks: error_code = task() #type:ignore[operator] if error_code > 0: - process_manager.end() return error_code - process_manager.end() return 0 diff --git a/facefusion/workflows/modules/image_to_image.py b/facefusion/workflows/modules/image_to_image.py index ed9f3654..d84a5265 100644 --- a/facefusion/workflows/modules/image_to_image.py +++ b/facefusion/workflows/modules/image_to_image.py @@ -1,6 +1,5 @@ from functools import partial -from facefusion import process_manager from facefusion.types import ErrorCode from facefusion.workflows.core import clear, setup from facefusion.workflows.to_image import analyse_image, finalize_image, prepare_image, process_image @@ -18,14 +17,10 @@ def process(start_time : float) -> ErrorCode: clear ] - process_manager.start() - for task in tasks: error_code = task() #type:ignore[operator] if error_code > 0: - process_manager.end() return error_code - process_manager.end() return 0 diff --git a/facefusion/workflows/modules/image_to_video.py b/facefusion/workflows/modules/image_to_video.py index 011f8bdc..1410d723 100644 --- a/facefusion/workflows/modules/image_to_video.py +++ b/facefusion/workflows/modules/image_to_video.py @@ -1,6 +1,6 @@ from functools import partial -from facefusion import process_manager, state_manager +from facefusion import state_manager from facefusion.types import ErrorCode from facefusion.workflows.core import clear, process_frames, setup from facefusion.workflows.to_video import analyse_video, create_temp_frames, finalize_video, merge_frames, process_memory_frames, restore_audio @@ -32,14 +32,10 @@ def process(start_time : float) -> ErrorCode: clear ]) - process_manager.start() - for task in tasks: error_code = task() #type:ignore[operator] if error_code > 0: - process_manager.end() return error_code - process_manager.end() return 0 diff --git a/facefusion/workflows/modules/image_to_video_as_frames.py b/facefusion/workflows/modules/image_to_video_as_frames.py index 70729d7d..e11aa850 100644 --- a/facefusion/workflows/modules/image_to_video_as_frames.py +++ b/facefusion/workflows/modules/image_to_video_as_frames.py @@ -1,6 +1,5 @@ from functools import partial -from facefusion import process_manager from facefusion.types import ErrorCode from facefusion.workflows.as_frames import copy_temp_frames, finalize_frames from facefusion.workflows.core import clear, process_frames, setup @@ -20,14 +19,10 @@ def process(start_time : float) -> ErrorCode: clear ] - process_manager.start() - for task in tasks: error_code = task() #type:ignore[operator] if error_code > 0: - process_manager.end() return error_code - process_manager.end() return 0 diff --git a/tests/apis/endpoints/test_jobs.py b/tests/apis/endpoints/test_jobs.py index 7b1eb6eb..eeffcc43 100644 --- a/tests/apis/endpoints/test_jobs.py +++ b/tests/apis/endpoints/test_jobs.py @@ -355,9 +355,8 @@ def test_cancel_jobs(test_client : TestClient) -> None: { 'Authorization': 'Bearer ' + access_token }) - fork_id = session_manager.fork_session() - process_manager.start() session_context.set_session_id(session_id) + process_manager.start() cancel_jobs_response = test_client.patch('/jobs?action=cancel', headers = { @@ -367,11 +366,10 @@ def test_cancel_jobs(test_client : TestClient) -> None: assert cancel_jobs_body.get('message') == 'ok' assert find_job_ids('drafted') == [ 'job-test-cancel-jobs' ] - assert store_creator.get_content(process_manager.PROCESS_STORE, fork_id) == 'stopping' + assert store_creator.get_content(process_manager.PROCESS_STORE, session_id) == 'stopping' assert cancel_jobs_response.status_code == 200 - session_manager.clear_cli_session(fork_id) - store_creator.delete_content(process_manager.PROCESS_STORE, fork_id) + process_manager.end() def test_cancel_job(test_client : TestClient) -> None: @@ -410,9 +408,8 @@ def test_cancel_job(test_client : TestClient) -> None: { 'Authorization': 'Bearer ' + access_token }) - fork_id = session_manager.fork_session() - process_manager.start() session_context.set_session_id(session_id) + process_manager.start() cancel_job_response = test_client.patch('/jobs/job-test-cancel-job?action=cancel', headers = { @@ -422,7 +419,7 @@ def test_cancel_job(test_client : TestClient) -> None: assert cancel_job_body.get('message') == 'ok' assert find_job_ids('drafted') == [ 'job-test-cancel-job' ] - assert store_creator.get_content(process_manager.PROCESS_STORE, fork_id) == 'processing' + assert store_creator.get_content(process_manager.PROCESS_STORE, session_id) == 'processing' assert cancel_job_response.status_code == 200 test_client.patch('/jobs/job-test-cancel-job?action=submit', headers = @@ -439,11 +436,10 @@ def test_cancel_job(test_client : TestClient) -> None: assert cancel_job_body.get('message') == 'ok' assert find_job_ids('drafted') == [ 'job-test-cancel-job' ] - assert store_creator.get_content(process_manager.PROCESS_STORE, fork_id) == 'stopping' + assert store_creator.get_content(process_manager.PROCESS_STORE, session_id) == 'stopping' assert cancel_job_response.status_code == 200 - session_manager.clear_cli_session(fork_id) - store_creator.delete_content(process_manager.PROCESS_STORE, fork_id) + process_manager.end() def test_run_jobs(test_client : TestClient) -> None: @@ -482,6 +478,28 @@ def test_run_jobs(test_client : TestClient) -> None: 'Authorization': 'Bearer ' + access_token }) + process_manager.start() + + run_jobs_response = test_client.patch('/jobs?action=run', headers = + { + 'Authorization': 'Bearer ' + access_token + }) + run_jobs_body = run_jobs_response.json() + + assert run_jobs_body.get('message') == 'jobs not started' + assert run_jobs_response.status_code == 409 + + process_manager.check() + + run_jobs_response = test_client.patch('/jobs?action=run', headers = + { + 'Authorization': 'Bearer ' + access_token + }) + + assert run_jobs_response.status_code == 409 + + process_manager.end() + with patch('facefusion.jobs.job_runner.run_jobs', return_value = True) as run_jobs_mock: run_jobs_response = test_client.patch('/jobs?action=run', headers = { @@ -531,6 +549,19 @@ def test_run_job(test_client : TestClient) -> None: 'Authorization': 'Bearer ' + access_token }) + process_manager.start() + + run_job_response = test_client.patch('/jobs/job-test-run-job?action=run', headers = + { + 'Authorization': 'Bearer ' + access_token + }) + run_job_body = run_job_response.json() + + assert run_job_body.get('message') == 'job not started' + assert run_job_response.status_code == 409 + + process_manager.end() + with patch('facefusion.jobs.job_runner.run_job', return_value = True) as run_job_mock: run_job_response = test_client.patch('/jobs/job-test-run-job?action=run', headers = { @@ -581,6 +612,19 @@ def test_retry_jobs(test_client : TestClient) -> None: set_steps_status('job-test-retry-jobs', 'failed') move_job_file('job-test-retry-jobs', 'failed') + process_manager.start() + + retry_jobs_response = test_client.patch('/jobs?action=retry', headers = + { + 'Authorization': 'Bearer ' + access_token + }) + retry_jobs_body = retry_jobs_response.json() + + assert retry_jobs_body.get('message') == 'jobs not retried' + assert retry_jobs_response.status_code == 409 + + process_manager.end() + with patch('facefusion.jobs.job_runner.retry_jobs', return_value = True) as retry_jobs_mock: retry_jobs_response = test_client.patch('/jobs?action=retry', headers = { @@ -642,6 +686,19 @@ def test_retry_job(test_client : TestClient) -> None: set_steps_status('job-test-retry-job', 'failed') move_job_file('job-test-retry-job', 'failed') + process_manager.start() + + retry_job_response = test_client.patch('/jobs/job-test-retry-job?action=retry', headers = + { + 'Authorization': 'Bearer ' + access_token + }) + retry_job_body = retry_job_response.json() + + assert retry_job_body.get('message') == 'job not retried' + assert retry_job_response.status_code == 409 + + process_manager.end() + with patch('facefusion.jobs.job_runner.retry_job', return_value = True) as retry_job_mock: retry_job_response = test_client.patch('/jobs/job-test-retry-job?action=retry', headers = { diff --git a/tests/jobs/test_job_runner.py b/tests/jobs/test_job_runner.py index b0cbfaa1..4d4400e6 100644 --- a/tests/jobs/test_job_runner.py +++ b/tests/jobs/test_job_runner.py @@ -110,8 +110,15 @@ def test_run_job() -> None: assert run_job('job-test-run-job', process_step) is False submit_job('job-test-run-job') + process_manager.stop() - assert run_job('job-test-run-job', process_step) is True + with patch('facefusion.process_manager.start', wraps = process_manager.start) as start_mock: + with patch('facefusion.process_manager.end', wraps = process_manager.end) as end_mock: + assert run_job('job-test-run-job', process_step) is True + + assert start_mock.call_count == 1 + assert end_mock.call_count == 1 + assert process_manager.get_state() == 'pending' create_job('job-test-run-job-cancel') add_step('job-test-run-job-cancel', args_1) @@ -319,6 +326,7 @@ def test_finalize_steps() -> None: create_directory(temp_directory_2) copy_file(args_4.get('target_path'), os.path.join(temp_directory_1, '00000001.jpg')) copy_file(args_5.get('target_path'), os.path.join(temp_directory_2, '00000002.jpg')) + process_manager.start() assert finalize_steps('job-test-finalize-steps') is True assert is_test_output_file('output-1.mp4') is True diff --git a/tests/test_download.py b/tests/test_download.py index c2b53995..81842932 100644 --- a/tests/test_download.py +++ b/tests/test_download.py @@ -1,8 +1,17 @@ +from unittest.mock import patch + +import pytest + from facefusion import process_manager from facefusion.download import conditional_download_files, conditional_validate_files, get_static_download_size, ping_static_url, resolve_download_url_by_provider, validate_file from facefusion.types import DownloadSet +@pytest.fixture(scope = 'module', autouse = True) +def before_all() -> None: + process_manager.clear() + + def test_get_static_download_size() -> None: assert get_static_download_size('https://github.com/facefusion/facefusion-assets/releases/download/models-3.4.0/yunet_2023_mar.onnx') == 232475 assert get_static_download_size('https://huggingface.co/facefusion/models-3.4.0/resolve/main/yunet_2023_mar.onnx') == 232475 @@ -49,6 +58,18 @@ def test_conditional_download_files() -> None: assert conditional_download_files(invalid_set) is False assert process_manager.is_pending() is True + process_manager.start() + + assert conditional_download_files(hash_set) is True + assert process_manager.get_state() == 'processing' + + with patch('facefusion.process_manager.check', side_effect = process_manager.stop): + assert conditional_download_files(hash_set) is True + + assert process_manager.get_state() == 'stopping' + + process_manager.end() + def test_conditional_validate_files() -> None: file_set : DownloadSet =\ diff --git a/tests/test_process_manager.py b/tests/test_process_manager.py index 9d336d96..8cb3b481 100644 --- a/tests/test_process_manager.py +++ b/tests/test_process_manager.py @@ -42,8 +42,16 @@ def test_get_state() -> None: assert get_state() == 'processing' fork_id = session_manager.fork_session() - clear() + set_state('stopping') + + assert get_state() == 'stopping' + + session_context.set_session_id('session-a') + + assert get_state() == 'stopping' + set_state('processing') + session_context.set_session_id(fork_id) assert get_state() == 'processing'