diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6941fa6..c66c7ad 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -10,7 +10,7 @@ on: jobs: test: - name: "Erlang Test" + name: "OTP ${{ matrix.otp }}" runs-on: ubuntu-latest strategy: matrix: @@ -19,6 +19,14 @@ jobs: container: image: erlang:${{ matrix.otp }} steps: - - uses: actions/checkout@v3 + - uses: actions/checkout@v7 - name: Test run: make + - name: Upload CT logs + if: failure() + uses: actions/upload-artifact@v7 + with: + name: ct-logs-otp-${{ matrix.otp }} + path: _build/test/logs + if-no-files-found: ignore + retention-days: 14 diff --git a/.gitignore b/.gitignore index 95ff3ca..e67eb14 100644 --- a/.gitignore +++ b/.gitignore @@ -9,4 +9,3 @@ eqc/*.beam */#*# erl_crash.dump gproc_dist*@* -rebar.lock diff --git a/Makefile b/Makefile index 569d78a..e5a3860 100755 --- a/Makefile +++ b/Makefile @@ -21,7 +21,7 @@ ## DEALINGS IN THE SOFTWARE. REBAR3=$(shell which rebar3 || echo ./rebar3) -.PHONY: all compile clean eunit test doc check dialyzer +.PHONY: all compile clean eunit ct test doc check dialyzer DIRS=src @@ -35,10 +35,18 @@ compile: clean: $(REBAR3) clean +# Local / non-distributed eunit only. gproc_dist lives under CT +# (see gproc_dist_SUITE); the old dist eunit generator is disabled. eunit: $(REBAR3) eunit -test: eunit +# Distributed suite: locks_leader peers + per-node disk logs. +ct: + @epmd -daemon 2>/dev/null || true + $(REBAR3) ct --suite test/gproc_dist_SUITE + +# Full local gate (matches intended CI split). +test: eunit ct doc: $(REBAR3) as edown edoc diff --git a/rebar.config b/rebar.config index 756f6f0..ebad4a8 100644 --- a/rebar.config +++ b/rebar.config @@ -2,12 +2,11 @@ {erl_opts, [debug_info]}. {deps, [ - {gen_leader, ".*", - {git, "https://github.com/garret-smith/gen_leader_revival.git", "HEAD"}} + {locks, {git, "https://github.com/uwiger/locks.git", {ref, "ed69e18"}}} ]}. {dialyzer, [{plt_apps, all_deps}, - {plt_extra_apps, [mnesia, runtime_tools, gen_leader]}, + {plt_extra_apps, [mnesia, runtime_tools, locks]}, {warnings, [no_unused, no_improper_lists, no_fun_app, no_match, no_opaque, no_fail_call, diff --git a/rebar.config.script b/rebar.config.script deleted file mode 100644 index 8679268..0000000 --- a/rebar.config.script +++ /dev/null @@ -1,33 +0,0 @@ -%% -*- erlang -*- -Script = fun(D,S,Vs) -> - Scr = filename:join(D, S), - case file:script(Scr, orddict:store('SCRIPT', Scr, Vs)) of - {ok, Res} -> Res; - {error,_} = Err -> - io:fwrite("Error evaluating script ~s~n", [S]), - Err - end - end. -CFG1 = case os:getenv("REBAR_DEPS") of - false -> CONFIG; - [] -> CONFIG; - Dir -> - lists:keystore(deps_dir, 1, CONFIG, {deps_dir, Dir}) - end. -Priv = filename:join(filename:dirname(SCRIPT), "priv"). -CFG2 = case os:getenv("GPROC_DIST") of - "true" -> - Sh = proplists:get_value(shell, CFG1, []), - case lists:keymember(config, 1, Sh) of - true -> - CFG1; - false -> - lists:keystore( - shell, 1, CFG1, - {shell, [{config, "dist_shell.config"}|Sh]}) - end; - F when F=="false"; F==false -> - Script(Priv, "remove_deps.script", - [{'CONFIG', CFG1}, {'DEPS', [gen_leader]}]) - end. -Script(Priv, "check_edown.script", [{'CONFIG', CFG2}]). diff --git a/rebar.lock b/rebar.lock new file mode 100644 index 0000000..d2627fd --- /dev/null +++ b/rebar.lock @@ -0,0 +1,12 @@ +{"1.2.0", +[{<<"locks">>, + {git,"https://github.com/uwiger/locks.git", + {ref,"ed69e1864cc2f8a323f592ca23e3a8e792116ee9"}}, + 0}, + {<<"plain_fsm">>,{pkg,<<"plain_fsm">>,<<"1.4.9">>},1}]}. +[ +{pkg_hash,[ + {<<"plain_fsm">>, <<"047B015FD70B533562DEC3FF2861E2465E67CD6F31034C5D7954B56ABBB839DD">>}]}, +{pkg_hash_ext,[ + {<<"plain_fsm">>, <<"A9630F78C05599E45AFCF9B423F2A30752F72A527F3C17A24F10E387BBCB91B8">>}]} +]. diff --git a/src/gproc.app.src b/src/gproc.app.src index 570f16d..8af0aff 100644 --- a/src/gproc.app.src +++ b/src/gproc.app.src @@ -8,7 +8,11 @@ {vsn, "zomp"}, {id, "GPROC"}, {registered, [ ] }, - {applications, [ kernel, stdlib ] }, + %% locks is optional: listed in both applications and optional_applications + %% so the app controller starts it automatically when present, but gproc + %% still loads if locks is absent (local-only installs). + {applications, [ kernel, stdlib, locks ] }, + {optional_applications, [ locks ] }, {mod, {gproc_app, []} }, {licenses, ["Apache-2.0"]}, {links, [{"Github", "https://github.com/uwiger/gproc"}]} diff --git a/src/gproc.erl b/src/gproc.erl index 730a9b7..f6e3a8e 100644 --- a/src/gproc.erl +++ b/src/gproc.erl @@ -41,6 +41,13 @@ %% starting the `gproc' and `gproc_dist' servers. Default is `[]'. It is %% likely that `{priority, high | max}' and/or increasing `min_heap_size' %% will improve performance. +%% * `{gproc_dist, all | [node()] | {[node()], list()}}' - When `locks' is +%% available and `gproc_dist' is started, an explicit node list causes +%% `net_kernel:connect_node/1' for each peer before joining the society. +%% `all' / `true' do not connect (already-connected nodes are discovered +%% via locks_pg). Under gen_leader this env also controlled whether dist +%% started; now dist starts whenever `locks' is running +%% (see optional_applications). %% %% @end @@ -191,6 +198,7 @@ ok end). + -define(PID_IS_DEAD(Pid), (node(Pid) == node() andalso is_process_alive(Pid) == false)). @@ -706,7 +714,6 @@ await(Node, Key, Timeout) when is_atom(Node) -> ?CATCH_GPROC_ERROR(await1(Node, Key, Timeout), [Node, Key, Timeout]). await1({T,g,_} = Key, Timeout) when T=:=n; T=:=a; T=:=rc -> - ?CHK_DIST, request_wait(Key, Timeout); await1({T,l,_} = Key, Timeout) when T=:=n; T=:=a; T=:=rc -> case ets:lookup(?TAB, {Key, T}) of @@ -860,7 +867,6 @@ nb_wait(Node, Key) -> ?CATCH_GPROC_ERROR(nb_wait1(Node, Key), [Node, Key]). nb_wait1({T,g,_} = Key) when T=:=n; T=:=a; T=:=rc -> - ?CHK_DIST, call({await, Key, self()}, g); nb_wait1({T,l,_} = Key) when T=:=n; T=:=a; T=:=rc -> call({await, Key, self()}, l); @@ -901,7 +907,6 @@ cancel_wait(N, Key, Ref) -> cancel_wait1({_,g,_} = Key, Ref) -> - ?CHK_DIST, cast({cancel_wait, self(), Key, Ref}, g), ok; cancel_wait1({_,l,_} = Key, Ref) -> @@ -917,7 +922,6 @@ cancel_wait_or_monitor(Key) -> ?CATCH_GPROC_ERROR(cancel_wait_or_monitor1(Key), [Key]). cancel_wait_or_monitor1({_,g,_} = Key) -> - ?CHK_DIST, cast({cancel_wait_or_monitor, self(), Key}, g), ok; cancel_wait_or_monitor1({_,l,_} = Key) -> @@ -2610,12 +2614,14 @@ call(Req) -> call(Req, l) -> chk_reply(gen_server:call(?MODULE, Req)); call(Req, g) -> + ?CHK_DIST, chk_reply(gproc_dist:leader_call(Req)). call(N, Req, l) -> chk_reply(gen_server:call({?MODULE, N}, Req)); call(undefined, Req, g) -> %% we always call the leader + ?CHK_DIST, chk_reply(gproc_dist:leader_call(Req)). @@ -2632,6 +2638,7 @@ cast(Msg) -> cast(Msg, l) -> gen_server:cast(?MODULE, Msg); cast(Msg, g) -> + ?CHK_DIST, gproc_dist:leader_cast(Msg). cast(N, Msg, l) -> diff --git a/src/gproc_app.erl b/src/gproc_app.erl index 0a587da..ff58033 100644 --- a/src/gproc_app.erl +++ b/src/gproc_app.erl @@ -40,6 +40,11 @@ start() -> start(normal, []). start(_Type, StartArgs) -> + %% locks is optional_applications: ensure_all_started(gproc) will start it + %% as a dep, but ensure_started/start(gproc) will not. Pull locks in here + %% when it is loadable so gproc_sup can start gproc_dist; ignore failure + %% when locks is not present (local-only install). + _ = application:ensure_all_started(locks), case gproc_sup:start_link(StartArgs) of {ok, Pid} -> {ok, Pid}; diff --git a/src/gproc_dist.erl b/src/gproc_dist.erl index 8fe5c35..46d4b1d 100644 --- a/src/gproc_dist.erl +++ b/src/gproc_dist.erl @@ -24,7 +24,7 @@ -module(gproc_dist). -vsn("1.3.0"). -%% -behaviour(gen_leader). % to avoid unnecessary warnings +-behaviour(locks_leader). % to avoid unnecessary warnings -export([start_link/0, start_link/1, reg/1, reg/4, unreg/1, @@ -50,6 +50,15 @@ sync/0, get_leader/0]). +%% Lifecycle event type for gproc_ps (local scope). Tests and operators can +%% subscribe with gproc_ps:subscribe(l, gproc_dist) or subscribe_remote/2. +%% Messages: {gproc_ps_event, gproc_dist, RoleMsg} +%% where RoleMsg is {elected, node()} | {following, LeaderNode} +%% | {sync_done, node()}. +%% The durable {p,l,gproc_dist_role} property holds only elected|following; +%% sync_done is published but does not overwrite the role property. +-export([lifecycle_event/0]). + %%% internal exports -export([init/1, handle_cast/3, @@ -69,6 +78,8 @@ -include("gproc.hrl"). -define(SERVER, ?MODULE). +-define(LC_EVENT, gproc_dist). +-define(LC_ROLE, gproc_dist_role). -record(state, { always_broadcast = false, @@ -77,24 +88,54 @@ sync_requests = []}). -include("gproc_trace.hrl"). + +lifecycle_event() -> + ?LC_EVENT. %% ========================================================== %% Start functions +%% @doc Start gproc_dist using `{gproc, gproc_dist}' application env (if set). +%% +%% Env values (connectivity only; locks_leader syncs with peers where it is +%% already running once those nodes are connected): +%% +%% @end start_link() -> - start_link({[node()|nodes()], []}). - + start_link(application:get_env(gproc, gproc_dist)). + +start_link(undefined) -> + start_link_(); +start_link(false) -> + start_link_(); +start_link({ok, Env}) -> + start_link(Env); +start_link(true) -> + start_link_(); start_link(all) -> - Workers = case application:get_env(gproc_dist_workers) of - {ok, [_|_] = WorkersList} -> WorkersList; - _ -> [] - end, - start_link({[node()|nodes()], [{bcast_type, all}, {workers, Workers}]}); + start_link_(); start_link(Nodes) when is_list(Nodes) -> - start_link({Nodes, []}); -start_link({Nodes, Opts}) -> - SpawnOpts = gproc_lib:valid_opts(server_options, []), - gen_leader:start_link( - ?SERVER, Nodes, Opts, ?MODULE, [], [{spawn_opt, SpawnOpts}]). + connect_nodes(Nodes), + start_link_(); +start_link({Nodes, _Opts}) when is_list(Nodes) -> + %% Opts (workers, bcast_type, …) were gen_leader knobs; not used here. + connect_nodes(Nodes), + start_link_(). + +start_link_() -> + locks_leader:start_link(?SERVER, ?MODULE, [], [{role, candidate}]). + +connect_nodes(Nodes) -> + _ = [net_kernel:connect_node(N) || N <- Nodes, + is_atom(N), + N =/= node()], + ok. %% ========================================================== %% API @@ -267,7 +308,7 @@ reset_counter(_) -> %% during an ongoing sync, the call will fail with a timeout exception. %% (Actually, it should be a `leader_died' exception; more study needed to find %% out why gen_leader times out in this situation, rather than reporting that -%% the leader died.) +%% the leader died.) NOTE: switching to locks_leader, we need to revisit this. %% @end %% sync() -> @@ -278,8 +319,7 @@ sync() -> %% @doc Returns the node of the current gproc leader. %% @end get_leader() -> - GenLeader = gen_leader, - GenLeader:call(?MODULE, get_leader). + locks_leader:call(?MODULE, get_leader). %% ========================================================== %% Server-side @@ -288,7 +328,7 @@ handle_cast(_Msg, S, _) -> {stop, unknown_cast, S}. handle_call(get_leader, _, S, E) -> - {reply, gen_leader:leader_node(E), S}; + {reply, locks_leader:leader_node(E), S}; handle_call(sync, From, S, E) -> {noreply, initiate_sync(From, S, E)}; handle_call(_, _, S, _) -> @@ -308,23 +348,19 @@ handle_info(Msg, S, _E) -> elected(S, _E) -> - {ok, {globals,globs()}, S#state{is_leader = true}}. + S1 = S#state{is_leader = true}, + notify_role({elected, node()}), + {ok, {globals, globs()}, S1}. -elected(S, E, undefined) -> +elected(S, _E, undefined) -> %% I have become leader; full synch - {ok, {globals, globs()}, - maybe_reinitiate_sync(S#state{is_leader = true}, E)}; -elected(S, E, _Node) -> - Synch = {globals, globs()}, - if not S#state.always_broadcast -> - %% Another node recognized us as the leader. - %% Don't broadcast all data to everyone else - {reply, Synch, maybe_reinitiate_sync(S, E)}; - true -> - %% Main reason for doing this is if we are using a gen_leader - %% that doesn't support the 'reply' return value - {ok, Synch, maybe_reinitiate_sync(S, E)} - end. + S1 = S#state{is_leader = true}, + notify_role({elected, node()}), + {ok, {globals, globs()}, S1}; +elected(S, _E, _Node) -> + %% Still leader; a peer is joining — re-assert role for late subscribers. + notify_role({elected, node()}), + {reply, {globals, globs()}, S}. globs() -> Gs = ets:select(?TAB, [{{{{'_',g,'_'},'_'},'_','_'},[],['$_']}]), @@ -335,25 +371,40 @@ globs() -> surrendered(#state{is_leader = true} = S, {globals, Globs}, E) -> %% Leader conflict! surrendered_1(Globs), - {ok, maybe_reinitiate_sync(S#state{is_leader = false}, E)}; + S1 = S#state{is_leader = false}, + notify_role({following, locks_leader:leader_node(E)}), + {ok, maybe_reinitiate_sync(S1, E)}; surrendered(S, {globals, Globs}, E) -> %% globals from this node should be more correct in our table than %% in the leader's surrendered_1(Globs), - {ok, maybe_reinitiate_sync(S#state{is_leader = false}, E)}. + S1 = S#state{is_leader = false}, + notify_role({following, locks_leader:leader_node(E)}), + {ok, maybe_reinitiate_sync(S1, E)}. + +%% locks_leader passes the dead candidate/worker pid (gen_leader used a node). +handle_DOWN(Pid, S, E) when is_pid(Pid) -> + handle_DOWN_node(node(Pid), S, E); +handle_DOWN(Node, S, E) when is_atom(Node) -> + handle_DOWN_node(Node, S, E). -handle_DOWN(Node, S, E) -> +handle_DOWN_node(Node, S, E) -> S1 = check_sync_requests(Node, S, E), Head = {{{'_',g,'_'},'_'},'$1','_'}, Gs = [{'==', {node,'$1'},Node}], Globs = ets:select(?TAB, [{Head, Gs, [{{{element,1,{element,1,'$_'}}, {element,2,'$_'}}}]}]), + %% process_globals/1 already mutates the local tab. Only the leader + %% should rebroadcast insert/notify ops (followers used to crash with + %% not_leader inside locks_leader:apply_cb/2). case process_globals(Globs) of [] -> {ok, S1}; - Broadcast -> - {ok, Broadcast, S1} + Broadcast when S1#state.is_leader -> + {ok, Broadcast, S1}; + _Broadcast -> + {ok, S1} end. check_sync_requests(Node, #state{sync_requests = SReqs} = S, E) -> @@ -683,12 +734,13 @@ handle_leader_call(_, _, S, _E) -> {reply, badarg, S}. handle_leader_cast({initiate_sync, Ref}, S, E) -> - case gen_leader:alive(E) -- [node()] of + case other_alive_nodes(E) of [] -> %% ??? {noreply, send_sync_complete(Ref, S, E)}; Alive -> - gen_leader:broadcast({from_leader, {sync, Ref}}, Alive, E), + %% locks_leader wraps as from_leader — pass payload only. + locks_leader:broadcast({sync, Ref}, E), {noreply, S#state{sync_requests = [{Ref, Alive}|S#state.sync_requests]}} end; @@ -866,7 +918,7 @@ terminate(_Reason, _S) -> ok. from_leader({sync, Ref}, S, _E) -> - gen_leader:leader_cast(?MODULE, {sync_reply, node(), Ref}), + locks_leader:leader_cast(?MODULE, {sync_reply, node(), Ref}), {ok, S}; from_leader({sync_complete, Ref}, S, _E) -> case Ref of @@ -941,7 +993,7 @@ ets_key(K, Pid) -> {K, Pid}. leader_call(Req) -> - case gen_leader:leader_call(?MODULE, Req) of + case locks_leader:leader_call(?MODULE, Req) of badarg -> ?THROW_GPROC_ERROR(badarg); Reply -> Reply end. @@ -953,7 +1005,7 @@ leader_call(Req) -> %% end. leader_cast(Msg) -> - gen_leader:leader_cast(?MODULE, Msg). + locks_leader:leader_cast(?MODULE, Msg). init(Opts) -> S0 = #state{}, @@ -1183,14 +1235,13 @@ regged_new(ensure) -> new. initiate_sync(From, #state{is_leader = true} = S, E) -> - case gen_leader:alive(E) -- [node()] of + case other_alive_nodes(E) of [] -> %% I'm alone - sync is trivial gen_server:reply(From, true), S; Alive -> - gen_leader:broadcast( - {from_leader, {sync, From}}, Alive, E), + locks_leader:broadcast({sync, From}, E), S#state{sync_requests = [{From, Alive}|S#state.sync_requests]} end; @@ -1209,16 +1260,44 @@ maybe_reinitiate_sync(#state{sync_clients = Cs} = S, E) -> send_sync_complete({From, _} = Ref, S, _E) when node(From) == node() -> reply_to_sync_client(Ref, S); send_sync_complete({From, _} = Ref, S, E) -> - %% Notify the node that initiated the sync - %% 'broadcasting' to exactly one node. - gen_leader:broadcast( - {from_leader, {sync_complete, Ref}}, [node(From)], E), + %% Notify the node that initiated the sync (payload only; leader wraps). + Targets = pids_on_nodes(E, [node(From)]), + locks_leader:broadcast({sync_complete, Ref}, Targets, E), S#state{sync_requests = lists:keydelete(Ref, 1, S#state.sync_requests)}. +%% locks_leader:alive/1 returns candidate/worker pids, not nodes. +other_alive_nodes(E) -> + lists:usort([node(P) || P <- locks_leader:alive(E), node(P) =/= node()]). + +pids_on_nodes(E, Nodes) -> + [P || P <- locks_leader:alive(E), lists:member(node(P), Nodes)]. + reply_to_sync_client(Ref, S) -> gen_server:reply(Ref, true), + %% Publish only — do not overwrite the elected/following role property. + notify_event({sync_done, node()}), S#state{sync_clients = S#state.sync_clients -- [Ref], sync_requests = lists:keydelete(Ref, 1, S#state.sync_requests)}. + +%% Local gproc_ps + durable role property ({elected|following, Node}). +%% Never let notification failure take down the leader process. +notify_role(Msg) -> + try + _ = gproc:ensure_reg({p, l, ?LC_ROLE}, Msg), + _ = gproc_ps:publish(l, ?LC_EVENT, Msg), + ok + catch + _:_ -> ok + end. + +%% Transient lifecycle events (e.g. sync_done) — pub/sub only. +notify_event(Msg) -> + try + _ = gproc_ps:publish(l, ?LC_EVENT, Msg), + ok + catch + _:_ -> ok + end. diff --git a/src/gproc_sup.erl b/src/gproc_sup.erl index 3178613..526a1cd 100644 --- a/src/gproc_sup.erl +++ b/src/gproc_sup.erl @@ -54,26 +54,34 @@ init(_Args) -> %% Child_spec = [Name, {M, F, A}, %% Restart, Shutdown_time, Type, Modules_used] - GProc = - {gproc, {gproc, start_link, []}, - permanent, 2000, worker, [gproc]}, - - Dist = case application:get_env(gproc_dist) of - undefined -> []; - {ok, false} -> []; - {ok, Env} -> - [{gproc_dist, {gproc_dist, start_link, [Env]}, - permanent, 2000, worker, [gproc_dist]}] - end, + Gproc = {gproc, {gproc, start_link, []}, + permanent, 2000, worker, [gproc]}, Mon = {gproc_monitor, {gproc_monitor, start_link, []}, permanent, 2000, worker, [gproc_monitor]}, BCast = {gproc_bcast, {gproc_bcast, start_link, []}, permanent, 2000, worker, [gproc_bcast]}, Pool = {gproc_pool, {gproc_pool, start_link, []}, permanent, 2000, worker, [gproc_pool]}, - {ok,{{one_for_one, 15, 60}, [GProc| Dist] ++ [Mon, BCast, Pool]}}. + Children = [Gproc] ++ maybe_dist() ++ [Mon, BCast, Pool], + {ok, {{one_for_one, 15, 60}, Children}}. %%%---------------------------------------------------------------------- %%% Internal functions %%%---------------------------------------------------------------------- + +%% gproc_dist needs locks. With locks as an optional_application, the app +%% controller starts it when present; if it is missing (or failed to start), +%% we stay local-only. +%% +%% `{gproc, gproc_dist}' env is passed through to start_link/1 so configured +%% peer node names are connected (gen_leader used to do that implicitly). +maybe_dist() -> + case whereis(locks_server) of + Pid when is_pid(Pid) -> + Env = application:get_env(gproc, gproc_dist), + [{gproc_dist, {gproc_dist, start_link, [Env]}, + permanent, 2000, worker, [gproc_dist]}]; + undefined -> + [] + end. diff --git a/test/gproc_dist_SUITE.erl b/test/gproc_dist_SUITE.erl new file mode 100644 index 0000000..dab3e93 --- /dev/null +++ b/test/gproc_dist_SUITE.erl @@ -0,0 +1,219 @@ +%% -*- erlang-indent-level: 4; indent-tabs-mode: nil -*- +%% +%% Common Test port of gproc_dist_tests (locks_leader era). +%% Peer nodes get a per-node disk logger under the CT log dir so post-mortems +%% do not depend on the old error_logger / eunit console noise. +-module(gproc_dist_SUITE). + +-export([all/0, groups/0, suite/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([ + simple_reg/1, + simple_reg_other/1, + simple_ensure/1, + simple_ensure_other/1, + simple_reg_or_locate/1, + simple_counter/1, + simple_r_counter/1, + simple_n_counter/1, + aggr_counter/1, + awaited_aggr_counter/1, + simple_resource_count/1, + wild_resource_count/1, + wild_key_in_resource/1, + awaited_resource_count/1, + resource_count_on_zero/1, + update_counters/1, + update_r_counters/1, + update_n_counters/1, + shared_counter/1, + prop/1, + mreg/1, + await_reg/1, + await_self/1, + await_reg_exists/1, + give_away/1, + sync/1, + monitor/1, + standby_monitor/1, + standby_monitor_unreg/1, + follow_monitor/1, + monitor_demonitor/1, + subscribe/1, + sync_cand_dies/1, + fail_node/1, + master_dies/1 + ]). + +-include_lib("common_test/include/ct.hrl"). + +-define(NODE_NAMES, [gproc_d1, gproc_d2, gproc_d3]). + +suite() -> + [{timetrap, {minutes, 5}}]. + +all() -> + [{group, dist3}]. + +groups() -> + %% sequence keeps registry state isolated between cases; the final + %% fault-injection cases still run even if an earlier basic case flakes. + [{dist3, [sequence], + [simple_reg, + simple_reg_other, + simple_ensure, + simple_ensure_other, + simple_reg_or_locate, + simple_counter, + simple_r_counter, + simple_n_counter, + aggr_counter, + awaited_aggr_counter, + simple_resource_count, + wild_resource_count, + wild_key_in_resource, + awaited_resource_count, + resource_count_on_zero, + update_counters, + update_r_counters, + update_n_counters, + shared_counter, + prop, + mreg, + await_reg, + await_self, + await_reg_exists, + give_away, + sync, + monitor, + standby_monitor, + standby_monitor_unreg, + follow_monitor, + monitor_demonitor, + subscribe, + sync_cand_dies, + fail_node, + master_dies + ]}]. + +init_per_suite(Config) -> + gproc_test_lib:ensure_dist(), + %% Prefer logger over legacy error_logger on the controller too. + _ = logger:set_primary_config(level, info), + Config. + +end_per_suite(_Config) -> + ok. + +init_per_group(dist3, Config) -> + LogDir = filename:join(?config(priv_dir, Config), "peer_logs"), + ok = filelib:ensure_dir(filename:join(LogDir, "dummy")), + ct:log("Peer disk logs under ~s", [LogDir]), + Ns = gproc_test_lib:start_nodes(?NODE_NAMES, #{log_dir => LogDir}), + ok = gproc_test_lib:start_gproc(Ns), + Leader = gproc_test_lib:wait_gproc_leader(Ns), + ct:log("gproc_dist leader = ~p on nodes ~p", [Leader, Ns]), + [{nodes, Ns}, {peer_log_dir, LogDir} | Config]; +init_per_group(_, Config) -> + Config. + +end_per_group(dist3, Config) -> + case ?config(nodes, Config) of + Ns when is_list(Ns) -> + gproc_test_lib:stop_nodes(Ns); + _ -> + ok + end, + ok; +end_per_group(_, _) -> + ok. + +init_per_testcase(_Case, Config) -> + Config. + +end_per_testcase(Case, Config) -> + case ?config(peer_log_dir, Config) of + undefined -> ok; + LogDir -> + ct:log("Peer logs for ~p: ~s", [Case, LogDir]) + end, + ok. + +%% ---- cases (thin wrappers; bodies live in gproc_dist_tests) ---- + +simple_reg(Config) -> + gproc_dist_tests:t_simple_reg(ns(Config)). +simple_reg_other(Config) -> + gproc_dist_tests:t_simple_reg_other(ns(Config)). +simple_ensure(Config) -> + gproc_dist_tests:t_simple_ensure(ns(Config)). +simple_ensure_other(Config) -> + gproc_dist_tests:t_simple_ensure_other(ns(Config)). +simple_reg_or_locate(Config) -> + gproc_dist_tests:t_simple_reg_or_locate(ns(Config)). +simple_counter(Config) -> + gproc_dist_tests:t_simple_counter(ns(Config)). +simple_r_counter(Config) -> + gproc_dist_tests:t_simple_r_counter(ns(Config)). +simple_n_counter(Config) -> + gproc_dist_tests:t_simple_n_counter(ns(Config)). +aggr_counter(Config) -> + gproc_dist_tests:t_aggr_counter(ns(Config)). +awaited_aggr_counter(Config) -> + gproc_dist_tests:t_awaited_aggr_counter(ns(Config)). +simple_resource_count(Config) -> + gproc_dist_tests:t_simple_resource_count(ns(Config)). +wild_resource_count(Config) -> + gproc_dist_tests:t_wild_resource_count(ns(Config)). +wild_key_in_resource(Config) -> + gproc_dist_tests:t_wild_key_in_resource(ns(Config)). +awaited_resource_count(Config) -> + gproc_dist_tests:t_awaited_resource_count(ns(Config)). +resource_count_on_zero(Config) -> + gproc_dist_tests:t_resource_count_on_zero(ns(Config)). +update_counters(Config) -> + gproc_dist_tests:t_update_counters(ns(Config)). +update_r_counters(Config) -> + gproc_dist_tests:t_update_r_counters(ns(Config)). +update_n_counters(Config) -> + gproc_dist_tests:t_update_n_counters(ns(Config)). +shared_counter(Config) -> + gproc_dist_tests:t_shared_counter(ns(Config)). +prop(Config) -> + gproc_dist_tests:t_prop(ns(Config)). +mreg(Config) -> + gproc_dist_tests:t_mreg(ns(Config)). +await_reg(Config) -> + gproc_dist_tests:t_await_reg(ns(Config)). +await_self(Config) -> + gproc_dist_tests:t_await_self(ns(Config)). +await_reg_exists(Config) -> + gproc_dist_tests:t_await_reg_exists(ns(Config)). +give_away(Config) -> + gproc_dist_tests:t_give_away(ns(Config)). +sync(Config) -> + gproc_dist_tests:t_sync(ns(Config)). +monitor(Config) -> + gproc_dist_tests:t_monitor(ns(Config)). +standby_monitor(Config) -> + gproc_dist_tests:t_standby_monitor(ns(Config)). +standby_monitor_unreg(Config) -> + gproc_dist_tests:t_standby_monitor_unreg(ns(Config)). +follow_monitor(Config) -> + gproc_dist_tests:t_follow_monitor(ns(Config)). +monitor_demonitor(Config) -> + gproc_dist_tests:t_monitor_demonitor(ns(Config)). +subscribe(Config) -> + gproc_dist_tests:t_subscribe(ns(Config)). +sync_cand_dies(Config) -> + gproc_dist_tests:t_sync_cand_dies(ns(Config)). +fail_node(Config) -> + gproc_dist_tests:t_fail_node(ns(Config)). +master_dies(Config) -> + gproc_dist_tests:t_master_dies(ns(Config)). + +ns(Config) -> + ?config(nodes, Config). diff --git a/test/gproc_dist_tests.erl b/test/gproc_dist_tests.erl index ef6047f..39b761c 100644 --- a/test/gproc_dist_tests.erl +++ b/test/gproc_dist_tests.erl @@ -19,118 +19,55 @@ %% -module(gproc_dist_tests). +%% Case bodies shared by the eunit wrapper (legacy) and gproc_dist_SUITE. -ifdef(TEST). -include_lib("eunit/include/eunit.hrl"). -export([t_spawn/1, t_spawn_reg/2]). +-export([ + t_simple_reg/1, + t_simple_reg_other/1, + t_simple_ensure/1, + t_simple_ensure_other/1, + t_simple_reg_or_locate/1, + t_simple_counter/1, + t_simple_r_counter/1, + t_simple_n_counter/1, + t_aggr_counter/1, + t_awaited_aggr_counter/1, + t_simple_resource_count/1, + t_wild_resource_count/1, + t_wild_key_in_resource/1, + t_awaited_resource_count/1, + t_resource_count_on_zero/1, + t_update_counters/1, + t_update_r_counters/1, + t_update_n_counters/1, + t_shared_counter/1, + t_prop/1, + t_mreg/1, + t_await_reg/1, + t_await_self/1, + t_await_reg_exists/1, + t_give_away/1, + t_sync/1, + t_monitor/1, + t_standby_monitor/1, + t_standby_monitor_unreg/1, + t_follow_monitor/1, + t_monitor_demonitor/1, + t_subscribe/1, + t_sync_cand_dies/1, + t_fail_node/1, + t_master_dies/1 + ]). + +%% Case bodies below are shared with gproc_dist_SUITE. +%% +%% The old eunit generator was named dist_test_/0 (eunit auto-discovery). +%% It is intentionally not defined: distributed coverage is Common Test only +%% (`rebar3 ct --suite gproc_dist_SUITE` / `make ct`), which meshes peers, +%% waits for a leader, and writes per-node logger disk logs. --define(f(E), fun() -> ?debugVal(E) end). - -dist_test_() -> - {timeout, 120, - [ - %% {setup, - %% fun dist_setup/0, - %% fun dist_cleanup/1, - %% fun(skip) -> []; - %% (Ns) when is_list(Ns) -> - %% {inorder, basic_tests(Ns)} - %% end - %% }, - {foreach, - fun dist_setup/0, - fun dist_cleanup/1, - [ - fun(Ns) -> - [{inorder, basic_tests(Ns)}] - end, - fun(Ns) -> - tests(Ns, [?f(t_sync_cand_dies(Ns))]) - end, - fun(Ns) -> - tests(Ns, [?f(t_fail_node(Ns))]) - end, - fun(Ns) -> - tests(Ns, [{timeout, 15, ?f(t_master_dies(Ns))}]) - end - ]} - ]}. - -tests(skip, _) -> - []; -tests(_, L) -> - L. - -basic_tests(skip) -> - []; -basic_tests(Ns) -> - [ - ?f(t_simple_reg(Ns)), - ?f(t_simple_reg_other(Ns)), - ?f(t_simple_ensure(Ns)), - ?f(t_simple_ensure_other(Ns)), - ?f(t_simple_reg_or_locate(Ns)), - ?f(t_simple_counter(Ns)), - ?f(t_simple_r_counter(Ns)), - ?f(t_simple_n_counter(Ns)), - ?f(t_aggr_counter(Ns)), - ?f(t_awaited_aggr_counter(Ns)), - ?f(t_simple_resource_count(Ns)), - ?f(t_wild_resource_count(Ns)), - ?f(t_wild_key_in_resource(Ns)), - ?f(t_awaited_resource_count(Ns)), - ?f(t_resource_count_on_zero(Ns)), - ?f(t_update_counters(Ns)), - ?f(t_update_r_counters(Ns)), - ?f(t_update_n_counters(Ns)), - ?f(t_shared_counter(Ns)), - ?f(t_prop(Ns)), - ?f(t_mreg(Ns)), - ?f(t_await_reg(Ns)), - ?f(t_await_self(Ns)), - ?f(t_await_reg_exists(Ns)), - ?f(t_give_away(Ns)), - ?f(t_sync(Ns)), - ?f(t_monitor(Ns)), - ?f(t_standby_monitor(Ns)), - ?f(t_standby_monitor_unreg(Ns)), - ?f(t_follow_monitor(Ns)), - ?f(t_monitor_demonitor(Ns)), - ?f(t_subscribe(Ns)) - ]. - -dist_setup() -> - case run_dist_tests() of - true -> - Ns = gproc_test_lib:start_nodes([dist_test_n1, dist_test_n2, dist_test_n3]), - ?assertMatch({[ok,ok,ok],[]}, - rpc:multicall(Ns, application, set_env, - [gproc, gproc_dist, Ns])), - ?assertMatch({[ok,ok,ok],[]}, - rpc:multicall( - Ns, application, start, [gproc])), - Ns; - false -> - skip - end. - -dist_cleanup(skip) -> - ok; -dist_cleanup(Ns) -> - ok = gproc_test_lib:stop_nodes(Ns), - ok. - -run_dist_tests() -> - case os:getenv("GPROC_DIST") of - "true" -> true; - "false" -> false; - false -> - case code:ensure_loaded(gen_leader) of - {error, nofile} -> - false; - _ -> - true - end - end. -define(T_NAME, {n, g, {?MODULE, ?LINE, os:timestamp()}}). -define(T_KVL, [{foo, "foo"}, {bar, "bar"}]). @@ -640,7 +577,7 @@ t_subscribe([A,B|_] = Ns) -> %% Verify that the gproc_dist:sync() call returns true even if a candidate dies %% while the sync is underway. This test makes use of sys:suspend() to ensure that %% the other candidate doesn't respond too quickly. -t_sync_cand_dies([A,B,C]) -> +t_sync_cand_dies([A,B,C] = Ns) -> Leader = rpc:call(A, gproc_dist, get_leader, []), Other = case Leader of A -> B; @@ -656,7 +593,12 @@ t_sync_cand_dies([A,B,C]) -> exit(P, kill), %% The leader should detect that the other candidate died and respond %% immediately. Therefore, we should have our answer well within 1 sec. - ?assertMatch({value, true}, rpc:nb_yield(Key, 1000)). + ?assertMatch({value, true}, rpc:nb_yield(Key, 1000)), + %% Restore the killed candidate so later suite cases still have a full mesh. + ok = rpc:call(Other, application, stop, [gproc]), + ok = gproc_test_lib:start_gproc([Other]), + _ = gproc_test_lib:wait_gproc_leader(Ns), + ok. %% Verify that the registry updates consistently if a non-leader node @@ -669,9 +611,11 @@ t_fail_node(Ns) -> Pa = t_spawn_reg(A, Na), Pb = t_spawn_reg(B, Nb), ?assertMatch(ok, rpc:call(A, application, stop, [gproc])), - ?assertMatch(ok, t_lookup_everywhere(Na, Ns -- [A], undefined)), - ?assertMatch(ok, t_lookup_everywhere(Nb, Ns -- [A], Pb)), - ?assertMatch(ok, rpc:call(A, application, start, [gproc])), + Rest = Ns -- [A], + ?assertMatch(ok, t_lookup_everywhere(Na, Rest, undefined)), + ?assertMatch(ok, t_lookup_everywhere(Nb, Rest, Pb)), + ok = gproc_test_lib:start_gproc([A]), + _ = gproc_test_lib:wait_gproc_leader(Ns), ?assertMatch(ok, t_lookup_everywhere(Na, Ns, undefined)), ?assertMatch(ok, t_lookup_everywhere(Nb, Ns, Pb)), ?assertMatch(ok, t_call(Pa, die)), @@ -718,12 +662,37 @@ try_sync(N, Ns) -> true end. +%% Always sync through the current leader (non-leaders forward via cast and +%% historically raced under locks_leader). Retry briefly while membership settles. +sync_via_leader(Ns) -> + sync_via_leader(Ns, 30). + +sync_via_leader(Ns, 0) -> + error({sync_failed, Ns, + [{N, rpc:call(N, gproc_dist, get_leader, [])} || N <- Ns]}); +sync_via_leader(Ns, I) -> + case rpc:call(hd(Ns), gproc_dist, get_leader, []) of + Leader when is_atom(Leader), Leader =/= undefined -> + case rpc:call(Leader, gproc_dist, sync, []) of + true -> + true; + _ -> + timer:sleep(200), + sync_via_leader(Ns, I - 1) + end; + _ -> + timer:sleep(200), + sync_via_leader(Ns, I - 1) + end. + t_sleep() -> timer:sleep(500). t_lookup_everywhere(Key, Nodes, Exp) -> - true = rpc:call(hd(Nodes), gproc_dist, sync, []), - t_lookup_everywhere(Key, Nodes, Exp, 3). + true = sync_via_leader(Nodes), + %% Patient under CI load (OTP 27 containers): 3×500ms was too tight when + %% followers were still joining the leader's synced set. + t_lookup_everywhere(Key, Nodes, Exp, 20). t_lookup_everywhere(Key, _, Exp, 0) -> {lookup_failed, Key, Exp}; @@ -742,8 +711,8 @@ t_lookup_everywhere(Key, Nodes, Exp, I) -> end. t_read_everywhere(Key, Pid, Nodes, Exp) -> - true = rpc:call(hd(Nodes), gproc_dist, sync, []), - t_read_everywhere(Key, Pid, Nodes, Exp, 3). + true = sync_via_leader(Nodes), + t_read_everywhere(Key, Pid, Nodes, Exp, 20). t_read_everywhere(Key, _, _, Exp, 0) -> {read_failed, Key, Exp}; diff --git a/test/gproc_remote_tests.erl b/test/gproc_remote_tests.erl index c514cd4..8248af8 100644 --- a/test/gproc_remote_tests.erl +++ b/test/gproc_remote_tests.erl @@ -25,20 +25,20 @@ -define(t(E), ?_test(?debugVal(E))). remote_test_() -> - %% dbg:tracer(), - %% dbg:tpl(?MODULE,x), - %% dbg:tpl(gproc_test_lib,x), - %% dbg:tp(slave,x), - %% dbg:p(all,[c]), StartedNet = gproc_test_lib:ensure_dist(), N = gproc_test_lib:node_name(remote_n1), {setup, fun() -> - gproc_test_lib:start_node(N), - rpc:call(N, application, start, [gproc]), - N + Node = gproc_test_lib:start_node(N), + %% gproc starts locks via ensure_all_started in gproc_app; + %% use ensure_all_started here too so a missing locks beam + %% fails clearly rather than as noproc on {gproc, Node}. + {ok, _} = rpc:call(Node, application, ensure_all_started, [gproc]), + true = is_pid(rpc:call(Node, erlang, whereis, [gproc])), + Node end, fun(Node) -> + _ = rpc:call(Node, application, stop, [gproc]), gproc_test_lib:stop_node(Node), case StartedNet of true -> diff --git a/test/gproc_test_lib.erl b/test/gproc_test_lib.erl index c9a0bc3..16058ca 100644 --- a/test/gproc_test_lib.erl +++ b/test/gproc_test_lib.erl @@ -23,39 +23,182 @@ t_spawn_mreg/3, t_call/2, t_loop/0, t_loop/1, - t_pool_contains_atleast/2, + t_pool_contains_atleast/2, got_msg/1, got_msg/2, no_msg/2]). -export([ensure_dist/0, - start_nodes/1, - start_node/1, + start_nodes/1, start_nodes/2, + start_node/1, start_node/2, stop_nodes/1, stop_node/1, - node_name/1]). + node_name/1, + mesh_connect/1, + start_gproc/1, + wait_gproc_leader/1, + subscribe_dist_events/1, + wait_dist_ready/1, + setup_peer_logger/2]). -include_lib("eunit/include/eunit.hrl"). +%% ============================================================ +%% Node lifecycle (peer) +%% ============================================================ + start_nodes(Ns) -> - [H|T] = Nodes = [start_node(N) || N <- Ns], - _ = [rpc:call(H, net_adm, ping, [N]) || N <- T], + start_nodes(Ns, #{}). + +%% Opts: +%% log_dir => dirname() | undefined — enable disk logger on each peer +start_nodes(Ns, Opts) when is_map(Opts) -> + Nodes = [start_node(N, Opts) || N <- Ns], + mesh_connect(Nodes), + case maps:get(log_dir, Opts, undefined) of + undefined -> ok; + LogDir -> [setup_peer_logger(N, LogDir) || N <- Nodes] + end, Nodes. start_node(Name0) -> + start_node(Name0, #{}). + +start_node(Name0, Opts) when is_map(Opts) -> {Name, _} = eunit_lib:split_node(Name0), ensure_dist(), {Pa, Pz} = paths(), Paths = lists:append([["-pa", "./", "-pz", "../ebin"]] ++ [["-pa", Path] || Path <- Pa] ++ [["-pz", Path] || Path <- Pz]), - Args = ["-kernel", "prevent_overlapping_partitions", "false" | Paths], + Args = ["-kernel", "prevent_overlapping_partitions", "false", + "-kernel", "logger_level", "debug" + | Paths], + %% standard_io control so intentional dist drops do not kill the peer. {ok, Pid, Node} = peer:start(#{ name => Name , host => host_string() - , args => Args }), + , args => Args + , connection => standard_io }), save_controlling_pid(Node, Pid), + case maps:get(log_dir, Opts, undefined) of + undefined -> ok; + LogDir -> setup_peer_logger(Node, LogDir) + end, Node. +%% Full mesh among peers. +mesh_connect(Nodes) -> + [begin + true = rpc:call(A, net_kernel, connect_node, [B]) + end || A <- Nodes, B <- Nodes, A =/= B], + ok. + +%% Start gproc (and thus locks + gproc_dist) on every peer; wait until +%% gproc_dist role properties agree on a leader (see wait_dist_ready/1). +start_gproc(Ns) -> + {Results, Bad} = rpc:multicall(Ns, application, ensure_all_started, [gproc]), + [] = Bad, + [case R of + {ok, _} -> ok; + {error, {already_started, _}} -> ok; + Other -> error({gproc_start_failed, Other}) + end || R <- Results], + %% Remote subscribe needs gproc already running on the peer. + %% Role properties are authoritative if we miss early events. + ok = subscribe_dist_events(Ns), + wait_gproc_leader(Ns), + ok. + +wait_gproc_leader(Ns) -> + wait_dist_ready(Ns). + +%% Controller: remote-subscribe to local lifecycle events on each peer. +%% Messages: {gproc_ps_event, gproc_dist, RoleMsg}. +subscribe_dist_events(Ns) -> + Ev = gproc_dist:lifecycle_event(), + lists:foreach(fun(N) -> true = gproc_ps:subscribe_remote(N, Ev) end, Ns), + ok. + +%% Wait until every node has a gproc_dist_role property of +%% {elected, L} or {following, L} for the same L. Drain pub/sub events while +%% waiting (useful diagnostics; property is authoritative if we subscribed late). +wait_dist_ready(Ns) -> + wait_dist_ready(Ns, 100). + +wait_dist_ready(Ns, 0) -> + error({dist_not_ready, Ns, [{N, role_on(N)} || N <- Ns]}); +wait_dist_ready(Ns, I) -> + _ = drain_lifecycle_events(), + case roles_agree(Ns) of + {ok, Leader} -> + Leader; + _ -> + timer:sleep(50), + wait_dist_ready(Ns, I - 1) + end. + +drain_lifecycle_events() -> + Ev = gproc_dist:lifecycle_event(), + receive + {gproc_ps_event, Ev, _Msg} -> + drain_lifecycle_events() + after 0 -> + ok + end. + +roles_agree(Ns) -> + Roles = [{N, role_on(N)} || N <- Ns], + %% undefined can appear if leader_node/1 was read before the election + %% opaque had leader set; never treat it as agreement. + Leaders = + [L || {_, {elected, L}} <- Roles, is_atom(L), L =/= undefined] ++ + [L || {_, {following, L}} <- Roles, is_atom(L), L =/= undefined], + case {lists:usort(Leaders), length(Roles) =:= length(Ns)} of + {[Leader], true} -> + case lists:all( + fun({_, {elected, L}}) -> L =:= Leader; + ({_, {following, L}}) -> L =:= Leader; + (_) -> false + end, Roles) of + true -> {ok, Leader}; + false -> error + end; + _ -> + error + end. + +role_on(N) -> + case rpc:call(N, gproc, lookup_values, [{p, l, gproc_dist_role}]) of + [{_Pid, {elected, _} = R}] -> R; + [{_Pid, {following, _} = R}] -> R; + [] -> undefined; + {badrpc, _} = E -> E; + Other -> Other + end. + +%% Disk log per peer under LogDir/.log (logger_std_h). +setup_peer_logger(Node, LogDir) -> + ok = filelib:ensure_dir(filename:join(LogDir, "dummy")), + File = filename:join(LogDir, atom_to_list(Node) ++ ".log"), + ok = rpc:call(Node, logger, set_primary_config, [level, debug]), + %% Replace default handler so everything hits the file (and keep stderr + %% free of CT noise). Fail soft if already configured. + _ = rpc:call(Node, logger, remove_handler, [default]), + case rpc:call(Node, logger, add_handler, + [default, logger_std_h, + #{level => debug, + config => #{file => File}, + formatter => + {logger_formatter, + #{template => [time, " ", level, " ", + {pid, ["[", pid, "] "], []}, + msg, "\n"]}}}]) of + ok -> ok; + {error, {already_exist, _}} -> ok; + Other -> error({peer_logger_failed, Node, Other}) + end, + ok. + stop_nodes(Ns) -> [stop_node(N) || N <- Ns], ok. @@ -68,7 +211,7 @@ stop_node(N) -> end. save_controlling_pid(Node, Pid) -> - persistent_term:put({?MODULE,peer_ref,Node}, Pid). + persistent_term:put({?MODULE, peer_ref, Node}, Pid). get_controlling_pid(Node) -> persistent_term:get({?MODULE, peer_ref, Node}). @@ -80,17 +223,14 @@ paths() -> Path = code:get_path(), {ok, [[Root]]} = init:get_argument(root), {Pas, Rest} = lists:splitwith(fun(P) -> - not lists:prefix(Root, P) - end, Path), + not lists:prefix(Root, P) + end, Path), {_, Pzs} = lists:splitwith(fun(P) -> - lists:prefix(Root, P) - end, Rest), + lists:prefix(Root, P) + end, Rest), {Pas, Pzs}. -%% host() -> -%% list_to_atom(host_string()). - host_string() -> [_Name, Host] = re:split(atom_to_list(node()), "@", [{return, list}]), Host. @@ -109,17 +249,21 @@ ensure_dist() -> false end. +%% ============================================================ +%% Spawned helpers on peer nodes +%% ============================================================ + t_spawn(Node) -> t_spawn(Node, false). t_spawn(Node, Selective) when is_boolean(Selective) -> Me = self(), P = spawn(Node, fun() -> - Me ! {self(), ok}, - t_loop(Selective) - end), + Me ! {self(), ok}, + t_loop(Selective) + end), receive - {P, ok} -> P + {P, ok} -> P after 1000 -> erlang:error({timeout, t_spawn, [Node, Selective]}) end. @@ -135,7 +279,7 @@ t_spawn_reg(Node, Name, Value) -> t_loop() end), receive - {P, ok} -> + {P, ok} -> P after 1000 -> erlang:error({timeout, t_spawn_reg, [Node, Name, Value]}) @@ -149,7 +293,7 @@ t_spawn_reg(Node, Name, Value, Attrs) -> t_loop() end), receive - {P, ok} -> + {P, ok} -> P after 1000 -> erlang:error({timeout, t_spawn_reg, [Node, Name, Value]}) @@ -166,7 +310,7 @@ t_spawn_mreg(Node, T, KVL) -> t_loop() end), receive - {P, ok} -> + {P, ok} -> P after 1000 -> error({timeout, t_spawn_mreg, [Node, T, KVL]}) @@ -181,7 +325,7 @@ t_spawn_reg_shared(Node, Name, Value) -> t_loop() end), receive - {P, ok} -> P + {P, ok} -> P after 1000 -> erlang:error({timeout, t_spawn_reg_shared, [Node,Name,Value]}) end. @@ -193,11 +337,11 @@ t_call(P, Req) -> Ref = erlang:monitor(process, P), P ! {self(), Ref, Req}, receive - {P, Ref, Res} -> - erlang:demonitor(Ref, [flush]), - Res; - {'DOWN', Ref, _, _, Error} -> - erlang:error({'DOWN', P, Error}) + {P, Ref, Res} -> + erlang:demonitor(Ref, [flush]), + Res; + {'DOWN', Ref, _, _, Error} -> + erlang:error({'DOWN', P, Error}) after 1000 -> erlang:error({timeout,t_call,[P,Req]}) end. @@ -207,20 +351,20 @@ t_loop() -> t_loop(Selective) when is_boolean(Selective) -> receive - {From, Ref, die} -> - From ! {self(), Ref, ok}; - {From, Ref, {selective, Bool}} when is_boolean(Bool) -> - From ! {self(), Ref, ok}, - t_loop(Bool); - {From, Ref, {apply, M, F, A}} -> - From ! {self(), Ref, apply(M, F, A)}, - t_loop(Selective); - {From, Ref, {apply_fun, F}} -> - From ! {self(), Ref, F()}, - t_loop(Selective); - Other when not Selective -> - ?debugFmt("got unknown msg: ~p~n", [Other]), - exit({unknown_msg, Other}) + {From, Ref, die} -> + From ! {self(), Ref, ok}; + {From, Ref, {selective, Bool}} when is_boolean(Bool) -> + From ! {self(), Ref, ok}, + t_loop(Bool); + {From, Ref, {apply, M, F, A}} -> + From ! {self(), Ref, apply(M, F, A)}, + t_loop(Selective); + {From, Ref, {apply_fun, F}} -> + From ! {self(), Ref, F()}, + t_loop(Selective); + Other when not Selective -> + ?debugFmt("got unknown msg: ~p~n", [Other]), + exit({unknown_msg, Other}) end. got_msg(Pb) -> @@ -235,15 +379,15 @@ got_msg(Pb) -> got_msg(Pb, Tag) -> t_call(Pb, - {apply_fun, - fun() -> - receive - M when element(1, M) == Tag -> - M - after 1000 -> - erlang:error({timeout, got_msg, [Pb, Tag]}) - end - end}). + {apply_fun, + fun() -> + receive + M when element(1, M) == Tag -> + M + after 1000 -> + erlang:error({timeout, got_msg, [Pb, Tag]}) + end + end}). no_msg(Pb, Timeout) -> t_call(Pb, diff --git a/test/gproc_tests.erl b/test/gproc_tests.erl index 7da97fa..c1d9696 100755 --- a/test/gproc_tests.erl +++ b/test/gproc_tests.erl @@ -41,7 +41,7 @@ conf_test_() -> t_server_opts() -> H = 10000, application:set_env(gproc, server_options, [{min_heap_size, H}]), - ?assertMatch(ok, application:start(gproc)), + ?assertMatch({ok,_}, application:ensure_all_started(gproc)), {min_heap_size, H1} = process_info(whereis(gproc), min_heap_size), ?assert(is_integer(H1) andalso H1 > H). @@ -66,11 +66,12 @@ t_ets_opts() -> reg_test_() -> {setup, fun() -> - application:start(gproc), - application:start(mnesia) + {ok,As} = application:ensure_all_started(gproc), + application:start(mnesia), + As end, - fun(_) -> - application:stop(gproc), + fun(As) -> + [application:stop(A) || A <- lists:reverse(As)], application:stop(mnesia) end, [ @@ -539,8 +540,48 @@ t_is_clean() -> T = ets:tab2list(gproc), Tm = ets:tab2list(gproc_monitor), ?assertMatch([], Tm), - ?assertMatch([], T -- [{{whereis(gproc_monitor), l}}, - {{self(), l}}]). + %% Infrastructure that may remain while gproc/locks/gproc_dist are up. + Allowed = infrastructure_regs(), + Rest = [E || E <- T, not is_allowed_reg(E, Allowed)], + ?assertMatch([], Rest). + +infrastructure_regs() -> + Mon = whereis(gproc_monitor), + Dist = whereis(gproc_dist), + [{{Mon, l}}, {{self(), l}}] + ++ case Dist of + Pid when is_pid(Pid) -> + [{{Pid, l}}, + {{Pid, {p, l, gproc_dist_role}}, []}, + {{{p, l, gproc_dist_role}, Pid}, Pid, + {elected, node()}}, + {{{p, l, gproc_dist_role}, Pid}, Pid, + {following, node()}}]; + _ -> + [] + end. + +is_allowed_reg(E, Allowed) -> + lists:member(E, Allowed) + orelse is_dist_role_reg(E). + +%% Role value may mention a remote leader node; match on key shape only. +is_dist_role_reg({{Pid, {p, l, gproc_dist_role}}, []}) when is_pid(Pid) -> + Pid =:= whereis(gproc_dist); +is_dist_role_reg({{{p, l, gproc_dist_role}, Pid}, Pid, _Val}) + when is_pid(Pid) -> + Pid =:= whereis(gproc_dist); +is_dist_role_reg(_) -> + false. + +%% gproc:table/1,2 entries for the dist lifecycle role property. +without_dist_role(L) -> + [E || E <- L, not is_dist_role_entry(E)]. + +is_dist_role_entry({{p, l, gproc_dist_role}, _Pid, _Val}) -> + true; +is_dist_role_entry(_) -> + false. t_simple_mreg() -> P = self(), @@ -727,7 +768,7 @@ t_qlc() -> ?assertEqual(Exp3, qlc:e(qlc:q([N || {_,_,x} = N <- gproc:table(all)]))), - %% match all + %% match all (ignore gproc_dist lifecycle role property) Exp4 = [{{a,l,{c,1}},self(),1}, {{c,l,{c,1}},self(),1}, {{n,l,{n,1}},self(),x}, @@ -736,7 +777,8 @@ t_qlc() -> {{p,l,{p,2}},self(),y} ], ?assertEqual(Exp4, - qlc:e(qlc:q([X || X <- gproc:table(all)]))), + without_dist_role( + qlc:e(qlc:q([X || X <- gproc:table(all)])))), %% match on pid ?assertEqual(Exp4, qlc:e(qlc:q([{K,P,V} || {K,P,V} <- @@ -795,12 +837,13 @@ t_qlc_dead() -> qlc:e(qlc:q([N || {_,_,y} = N <- gproc:table(all)]))), - %% match all + %% match all (ignore gproc_dist lifecycle role property) Exp4 = [{{n,l,{n,1}},self(),x}, {{p,l,{p,1}},self(),x}], ?assertEqual(Exp4, - qlc:e(qlc:q([X || X <- - gproc:table(all, [check_pids])]))), + without_dist_role( + qlc:e(qlc:q([X || X <- + gproc:table(all, [check_pids])])))), %% match on pid ?assertEqual(Exp4, qlc:e(qlc:q([{K,P,V} || {K,P,V} <-