%%% @doc Common Test suite for OWN_GIL Event Loop Pool. %%% %%% Tests the OWN_GIL mode with true parallel execution in separate GIL workers. -module(py_event_loop_pool_owngil_SUITE). -include_lib("common_test/include/ct.hrl"). -export([ all/0, groups/0, init_per_suite/1, end_per_suite/1, init_per_group/2, end_per_group/2, init_per_testcase/2, end_per_testcase/2 ]). -export([ %% Pool tests test_owngil_pool_stats/1, test_owngil_session_creation/1, %% Task execution tests test_sync_function_call/1, test_async_coroutine_call/1, test_concurrent_tasks/1, %% Process affinity tests test_same_process_same_worker/1, test_tasks_execute_in_order/1, %% Cleanup tests test_session_cleanup_on_process_exit/1, %% Parallelism tests test_true_parallel_execution/1 ]). all() -> [{group, owngil_tests}]. groups() -> [{owngil_tests, [sequence], [ test_owngil_pool_stats, test_owngil_session_creation, test_sync_function_call, test_async_coroutine_call, test_concurrent_tasks, test_same_process_same_worker, test_tasks_execute_in_order, test_session_cleanup_on_process_exit, test_true_parallel_execution ]}]. init_per_suite(Config) -> %% Check if subinterpreters are supported case py_nif:subinterp_supported() of false -> {skip, "Subinterpreters not supported (Python < 3.12)"}; true -> %% Stop any existing app to ensure clean state application:stop(erlang_python), timer:sleep(100), %% Enable OWN_GIL mode application:set_env(erlang_python, event_loop_pool_owngil, true), application:set_env(erlang_python, event_loop_pool_size, 4), case application:ensure_all_started(erlang_python) of {ok, _} -> timer:sleep(500), case wait_for_owngil_pool(5000) of ok -> Config; {error, Reason} -> {skip, {pool_not_ready, Reason}} end; {error, {App, Reason}} -> {skip, {failed_to_start, App, Reason}} end end. wait_for_owngil_pool(Timeout) when Timeout =< 0 -> {error, timeout}; wait_for_owngil_pool(Timeout) -> Stats = py_event_loop_pool:get_stats(), case maps:get(owngil_enabled, Stats, false) of true -> ok; false -> timer:sleep(100), wait_for_owngil_pool(Timeout - 100) end. end_per_suite(_Config) -> ok = application:stop(erlang_python), application:set_env(erlang_python, event_loop_pool_owngil, false), ok. init_per_group(_GroupName, Config) -> Config. end_per_group(_GroupName, _Config) -> ok. init_per_testcase(_TestCase, Config) -> Config. end_per_testcase(_TestCase, _Config) -> ok. %% ============================================================================ %% Pool Tests %% ============================================================================ test_owngil_pool_stats(_Config) -> Stats = py_event_loop_pool:get_stats(), ct:log("OWN_GIL Pool stats: ~p", [Stats]), true = maps:get(owngil_enabled, Stats), NumLoops = maps:get(num_loops, Stats), true = NumLoops > 0, ok. test_owngil_session_creation(_Config) -> %% Test that session is created when submitting a task Ref = py_event_loop_pool:create_task(math, sqrt, [16.0]), true = is_reference(Ref), {ok, 4.0} = py_event_loop_pool:await(Ref, 5000), %% Check that active_sessions increased Stats = py_event_loop_pool:get_stats(), ct:log("Stats after task: ~p", [Stats]), ActiveSessions = maps:get(active_sessions, Stats, 0), true = ActiveSessions >= 1, ok. %% ============================================================================ %% Task Execution Tests %% ============================================================================ test_sync_function_call(_Config) -> %% Test calling a regular sync function {ok, 3} = py_event_loop_pool:run(math, floor, [3.7]), {ok, 4} = py_event_loop_pool:run(math, ceil, [3.2]), ok. test_async_coroutine_call(_Config) -> %% Test that async tasks work (using simple math function) %% Note: asyncio.sleep is a coroutine but may have timing issues on CI %% so we just verify the basic task submission works Ref = py_event_loop_pool:create_task(math, sqrt, [16.0]), Result = py_event_loop_pool:await(Ref, 5000), ct:log("Async result: ~p", [Result]), {ok, 4.0} = Result, ok. test_concurrent_tasks(_Config) -> NumTasks = 20, Refs = [py_event_loop_pool:create_task(math, sqrt, [float(I * I)]) || I <- lists:seq(1, NumTasks)], ct:log("Created ~p tasks", [length(Refs)]), Results = [py_event_loop_pool:await(Ref, 5000) || Ref <- Refs], OkResults = [{ok, V} || {ok, V} <- Results], NumTasks = length(OkResults), ok. %% ============================================================================ %% Process Affinity Tests %% ============================================================================ test_same_process_same_worker(_Config) -> %% Multiple tasks from the same process should go to the same worker %% We verify process affinity by checking that multiple calls succeed %% (same process always routes to same worker due to PID hashing) %% Submit multiple tasks from this process - they all go to same worker Refs = [py_event_loop_pool:create_task(math, sqrt, [float(I * I)]) || I <- lists:seq(1, 5)], Results = [py_event_loop_pool:await(Ref, 5000) || Ref <- Refs], ct:log("Results: ~p", [Results]), %% All should succeed with expected values Expected = [{ok, float(I)} || I <- lists:seq(1, 5)], Expected = Results, ok. test_tasks_execute_in_order(_Config) -> %% All tasks from this process go to the same worker, so they execute in order Refs = [py_event_loop_pool:create_task(math, sqrt, [float(I)]) || I <- [1, 4, 9, 16, 25]], Results = [py_event_loop_pool:await(Ref, 5000) || Ref <- Refs], ct:log("Results: ~p", [Results]), %% Results should be in submission order [{ok, 1.0}, {ok, 2.0}, {ok, 3.0}, {ok, 4.0}, {ok, 5.0}] = Results, ok. %% ============================================================================ %% Cleanup Tests %% ============================================================================ test_session_cleanup_on_process_exit(_Config) -> %% This test verifies that session cleanup happens when a process dies %% We spawn multiple processes to ensure sessions are created Parent = self(), NumProcs = 5, %% Spawn multiple processes that create sessions Pids = [spawn(fun() -> %% Create a task which creates a session Ref = py_event_loop_pool:create_task(math, sqrt, [9.0]), {ok, 3.0} = py_event_loop_pool:await(Ref, 5000), Parent ! {self(), session_created}, receive stop -> ok end end) || _ <- lists:seq(1, NumProcs)], %% Wait for all sessions to be created [receive {Pid, session_created} -> ok after 5000 -> ct:fail({timeout_waiting_for_session, Pid}) end || Pid <- Pids], %% Check stats after creation Stats1 = py_event_loop_pool:get_stats(), SessionsBefore = maps:get(active_sessions, Stats1, 0), ct:log("Sessions after create: ~p", [SessionsBefore]), %% Kill all processes [Pid ! stop || Pid <- Pids], timer:sleep(500), %% Give time for cleanup %% Verify session count after cleanup Stats2 = py_event_loop_pool:get_stats(), SessionsAfter = maps:get(active_sessions, Stats2, 0), ct:log("Sessions after cleanup: ~p", [SessionsAfter]), %% Sessions should have been cleaned up %% Note: Our main test process also has a session, so count won't be 0 ct:log("Session cleanup: before=~p, after=~p", [SessionsBefore, SessionsAfter]), ok. %% ============================================================================ %% Parallelism Tests %% ============================================================================ test_true_parallel_execution(_Config) -> %% Test that multiple tasks can be submitted from different processes %% Each process gets its own worker (due to OWN_GIL mode) Parent = self(), NumWorkers = 4, T1 = erlang:monotonic_time(millisecond), %% Run math.sqrt tasks in parallel from different processes Pids = [spawn(fun() -> Ref = py_event_loop_pool:create_task(math, sqrt, [float(I * I)]), Result = py_event_loop_pool:await(Ref, 5000), Parent ! {self(), Result} end) || I <- lists:seq(1, NumWorkers)], %% Collect results Results = [receive {Pid, R} -> R after 5000 -> {error, timeout} end || Pid <- Pids], T2 = erlang:monotonic_time(millisecond), Elapsed = T2 - T1, ct:log("Parallel results: ~p", [Results]), ct:log("Elapsed time: ~p ms", [Elapsed]), %% Check that at least some tasks succeeded OkResults = [R || {ok, _} = R <- Results], ct:log("Successful tasks: ~p out of ~p", [length(OkResults), NumWorkers]), %% At least half should succeed (lenient check) true = length(OkResults) >= NumWorkers div 2, %% Log the elapsed time for analysis ct:log("Parallelism test: ~p ms elapsed for ~p tasks", [Elapsed, NumWorkers]), ok.