%%% @doc Stress tests for No-GIL Safe Mode. %%% %%% This suite tests race conditions, concurrent context operations, %%% and callback storms to verify the atomic state machine and %%% proper resource management under load. -module(py_stress_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([ context_churn_test/1, concurrent_calls_test/1, concurrent_evals_test/1, callback_storm_test/1, finalize_race_test/1, counter_balance_test/1, parallel_context_operations_test/1, rapid_start_stop_test/1 ]). all() -> [{group, stress_tests}]. groups() -> [{stress_tests, [parallel], [ context_churn_test, concurrent_calls_test, concurrent_evals_test, callback_storm_test, counter_balance_test, parallel_context_operations_test ]}, {sequential_tests, [], [ finalize_race_test, rapid_start_stop_test ]}]. init_per_suite(Config) -> {ok, _} = application:ensure_all_started(erlang_python), case py:contexts_started() of true -> ok; false -> {ok, _} = py:start_contexts() end, Config. end_per_suite(_Config) -> ok = application:stop(erlang_python), ok. init_per_group(stress_tests, Config) -> Config; init_per_group(sequential_tests, Config) -> Config; init_per_group(_Group, Config) -> Config. end_per_group(_Group, _Config) -> ok. init_per_testcase(_TestCase, Config) -> %% Get initial counter state Counters = py_nif:get_debug_counters(), [{initial_counters, Counters} | Config]. end_per_testcase(_TestCase, Config) -> %% Allow time for cleanup timer:sleep(100), %% Force GC to trigger destructors erlang:garbage_collect(), timer:sleep(100), Config. %%% ============================================================================ %%% Stress Test Cases %%% ============================================================================ %% @doc Rapidly create and destroy contexts to stress resource management context_churn_test(_Config) -> NumIterations = 100, AvailableContexts = py_context_router:num_contexts(), NumContexts = min(10, AvailableContexts), Self = self(), %% Spawn processes that rapidly create/use/destroy contexts Pids = [spawn_link(fun() -> context_churn_worker(NumIterations, N), Self ! {done, self()} end) || N <- lists:seq(1, NumContexts)], %% Wait for all workers to complete [receive {done, Pid} -> ok after 30000 -> ct:fail(timeout) end || Pid <- Pids], ok. context_churn_worker(0, _N) -> ok; context_churn_worker(Iterations, N) -> Ctx = py:context(N), {ok, _} = py:eval(Ctx, <<"1 + 1">>), context_churn_worker(Iterations - 1, N). %% @doc Concurrent py:call operations concurrent_calls_test(_Config) -> NumProcesses = 50, NumCalls = 20, Self = self(), Pids = [spawn_link(fun() -> concurrent_call_worker(NumCalls), Self ! {done, self()} end) || _ <- lists:seq(1, NumProcesses)], [receive {done, Pid} -> ok after 30000 -> ct:fail(timeout) end || Pid <- Pids], ok. concurrent_call_worker(0) -> ok; concurrent_call_worker(N) -> {ok, _} = py:call(math, sqrt, [N * N]), concurrent_call_worker(N - 1). %% @doc Concurrent py:eval operations concurrent_evals_test(_Config) -> NumProcesses = 50, NumEvals = 20, Self = self(), Pids = [spawn_link(fun() -> concurrent_eval_worker(NumEvals), Self ! {done, self()} end) || _ <- lists:seq(1, NumProcesses)], [receive {done, Pid} -> ok after 30000 -> ct:fail(timeout) end || Pid <- Pids], ok. concurrent_eval_worker(0) -> ok; concurrent_eval_worker(N) -> Expr = list_to_binary(io_lib:format("~p * ~p", [N, N])), {ok, _} = py:eval(Expr), concurrent_eval_worker(N - 1). %% @doc Storm of callbacks to stress suspension/resume callback_storm_test(_Config) -> %% Use unique name for callback to avoid parallel test conflicts UniqueId = erlang:unique_integer([positive]), CallbackName = list_to_atom("stress_cb_" ++ integer_to_list(UniqueId)), %% Register the callback function py:register_function(CallbackName, fun([_Arg]) -> 42 end), %% Use py:eval with inline code - no stored state needed %% This avoids __main__ namespace conflicts in parallel tests NumCallbacks = 50, Code = list_to_binary(io_lib:format( "(lambda n, erl: len([erl.call('~s', i) for i in range(n)]))(~p, __import__('erlang'))", [atom_to_list(CallbackName), NumCallbacks])), {ok, NumCallbacks} = py:eval(Code), %% Cleanup py:unregister_function(CallbackName), ok. %% @doc Test finalize behavior when operations are in flight finalize_race_test(Config) -> %% This test requires stopping and restarting the application %% Skip if we're running in parallel with other tests case ?config(parallel_running, Config) of true -> {skip, "Cannot run finalize test in parallel mode"}; _ -> do_finalize_race_test() end. do_finalize_race_test() -> %% Start the application fresh _ = application:stop(erlang_python), timer:sleep(100), {ok, _} = application:ensure_all_started(erlang_python), {ok, _} = py:start_contexts(), %% Start some concurrent operations Self = self(), Pids = [spawn(fun() -> %% Each worker does some work try lists:foreach(fun(_) -> py:eval(<<"1 + 1">>) end, lists:seq(1, 10)) catch _:_ -> ok end, Self ! {done, self()} end) || _ <- lists:seq(1, 5)], %% Give workers a bit of time to start timer:sleep(10), %% Stop the application while work is in flight ok = application:stop(erlang_python), %% Wait for workers to finish (they should handle shutdown gracefully) [receive {done, Pid} -> ok after 5000 -> ok end || Pid <- Pids], %% Restart for subsequent tests {ok, _} = application:ensure_all_started(erlang_python), {ok, _} = py:start_contexts(), ok. %% @doc Verify counter invariants counter_balance_test(_Config) -> %% Do some operations Ctx = py:context(1), lists:foreach(fun(_) -> {ok, _} = py:eval(Ctx, <<"1 + 1">>) end, lists:seq(1, 100)), %% Force GC erlang:garbage_collect(), timer:sleep(200), %% Check counters Counters = py_nif:get_debug_counters(), %% GIL ensure/release should be balanced GilEnsure = maps:get(gil_ensure, Counters, 0), GilRelease = maps:get(gil_release, Counters, 0), ct:log("GIL ensure: ~p, release: ~p", [GilEnsure, GilRelease]), %% Context created/destroyed should be balanced after GC CtxCreated = maps:get(ctx_created, Counters, 0), CtxDestroyed = maps:get(ctx_destroyed, Counters, 0), ct:log("Context created: ~p, destroyed: ~p", [CtxCreated, CtxDestroyed]), %% Suspended states should be balanced SuspCreated = maps:get(suspended_created, Counters, 0), SuspDestroyed = maps:get(suspended_destroyed, Counters, 0), ct:log("Suspended created: ~p, destroyed: ~p", [SuspCreated, SuspDestroyed]), %% Enqueue should match complete + rejected Enqueued = maps:get(enqueue_count, Counters, 0), Completed = maps:get(complete_count, Counters, 0), Rejected = maps:get(rejected_count, Counters, 0), ct:log("Enqueued: ~p, completed: ~p, rejected: ~p", [Enqueued, Completed, Rejected]), %% Verify no rejected operations during normal operation 0 = Rejected, ok. %% @doc Parallel operations on multiple contexts parallel_context_operations_test(_Config) -> AvailableContexts = py_context_router:num_contexts(), NumContexts = min(10, AvailableContexts), NumOps = 50, %% Get available contexts Contexts = [py:context(N) || N <- lists:seq(1, NumContexts)], %% Spawn workers for each context - use py:eval with inline expressions %% to avoid __main__ namespace conflicts Self = self(), Pids = [spawn_link(fun() -> Results = [py:eval(Ctx, list_to_binary(io_lib:format("~p * ~p", [I, I]))) || I <- lists:seq(1, NumOps)], AllOk = lists:all(fun({ok, _}) -> true; (_) -> false end, Results), Self ! {done, self(), AllOk} end) || Ctx <- Contexts], %% Collect results Results = [receive {done, Pid, Ok} -> Ok after 30000 -> false end || Pid <- Pids], true = lists:all(fun(X) -> X end, Results), ok. %% @doc Rapid application start/stop cycles rapid_start_stop_test(Config) -> case ?config(parallel_running, Config) of true -> {skip, "Cannot run start/stop test in parallel mode"}; _ -> do_rapid_start_stop_test() end. do_rapid_start_stop_test() -> NumCycles = 5, lists:foreach(fun(_) -> %% Stop ok = application:stop(erlang_python), timer:sleep(50), %% Start {ok, _} = application:ensure_all_started(erlang_python), {ok, _} = py:start_contexts(), %% Quick operation {ok, 2} = py:eval(<<"1 + 1">>), timer:sleep(50) end, lists:seq(1, NumCycles)), ok.