Avoid selective receive in code-purger process

The code purger process handles vast amounts of messages when
there are lots of processes alive. A single message in the
message queue that does not match will in such cases cause
lots of extra work. The code purger process now always picks
the first message in the message queue, and by this avoid
this extra work.

3 files changed, 187 insertions, 135 deletions

diff --git a/erts/emulator/test/code_SUITE.erl b/erts/emulator/test/code_SUITE.erl
index 82607424b5..8427bb134d 100644
--- a/erts/emulator/test/code_SUITE.erl
+++ b/erts/emulator/test/code_SUITE.erl
@@ -22,7 +22,7 @@
 -export([all/0, suite/0, init_per_suite/1, end_per_suite/1, 
          versions/1,new_binary_types/1, call_purged_fun_code_gone/1,
 	 call_purged_fun_code_reload/1, call_purged_fun_code_there/1,
-         t_check_old_code/1,
+         multi_proc_purge/1, t_check_old_code/1,
          external_fun/1,get_chunk/1,module_md5/1,make_stub/1,
          make_stub_many_funs/1,constant_pools/1,constant_refc_binaries/1,
          false_dependency/1,coverage/1,fun_confusion/1,
@@ -36,7 +36,7 @@ suite() -> [{ct_hooks,[ts_install_cth]}].
 all() -> 
     [versions, new_binary_types, call_purged_fun_code_gone,
      call_purged_fun_code_reload, call_purged_fun_code_there,
-     t_check_old_code, external_fun, get_chunk,
+     multi_proc_purge, t_check_old_code, external_fun, get_chunk,
      module_md5, make_stub, make_stub_many_funs,
      constant_pools, constant_refc_binaries, false_dependency,
      coverage, fun_confusion, t_copy_literals, t_copy_literals_frags].
@@ -275,6 +275,64 @@ call_purged_fun_test(Priv, Data, Type) ->
     end,
     ok.
 
+multi_proc_purge(Config) when is_list(Config) ->
+    %%
+    %% Make sure purge requests aren't lost when
+    %% purger process is working.
+    %%
+    Priv = proplists:get_value(priv_dir, Config),
+    Data = proplists:get_value(data_dir, Config),
+    File1 = filename:join(Data, "my_code_test"),
+    File2 = filename:join(Data, "my_code_test2"),
+    
+    {ok,my_code_test} = c:c(File1, [{outdir,Priv}]),
+    {ok,my_code_test2} = c:c(File2, [{outdir,Priv}]),
+    erlang:delete_module(my_code_test),
+    erlang:delete_module(my_code_test2),
+
+    Self = self(),
+
+    Fun1 = fun () ->
+		   erts_code_purger:purge(my_code_test),
+		   Self ! {self(), done}
+	   end,
+    Fun2 = fun () ->
+		   erts_code_purger:soft_purge(my_code_test2),
+		   Self ! {self(), done}
+	   end,
+    Fun3 = fun () ->
+		   erts_code_purger:purge('__nonexisting_module__'),
+		   Self ! {self(), done}
+	   end,
+    Fun4 = fun () ->
+		   erts_code_purger:soft_purge('__another_nonexisting_module__'),
+		   Self ! {self(), done}
+	   end,
+
+    Pid1 = spawn_link(Fun1),
+    Pid2 = spawn_link(Fun2),
+    Pid3 = spawn_link(Fun3),
+    Pid4 = spawn_link(Fun4),
+    Pid5 = spawn_link(Fun1),
+    Pid6 = spawn_link(Fun2),
+    Pid7 = spawn_link(Fun3),
+    receive after 50 -> ok end,
+    Pid8 = spawn_link(Fun4),
+    Pid9 = spawn_link(Fun1),
+    Pid10 = spawn_link(Fun2),
+    Pid11 = spawn_link(Fun3),
+    Pid12 = spawn_link(Fun4),
+    Pid13 = spawn_link(Fun1),
+    receive after 50 -> ok end,
+    Pid14 = spawn_link(Fun2),
+    Pid15 = spawn_link(Fun3),
+    Pid16 = spawn_link(Fun4),
+
+    lists:foreach(fun (P) -> receive {P, done} -> ok end end,
+		  [Pid1, Pid2, Pid3, Pid4, Pid5, Pid6, Pid7, Pid8,
+		   Pid9, Pid10, Pid11, Pid12, Pid13, Pid14, Pid15, Pid16]),
+    ok.
+
 body(F, Fakes) ->
     receive
         jog ->
diff --git a/erts/preloaded/ebin/erts_code_purger.beam b/erts/preloaded/ebin/erts_code_purger.beam
index 2dfed3df93..a012c46396 100644
--- a/erts/preloaded/ebin/erts_code_purger.beam
+++ b/erts/preloaded/ebin/erts_code_purger.beam
diff --git a/erts/preloaded/src/erts_code_purger.erl b/erts/preloaded/src/erts_code_purger.erl
index ad51815229..a48aebe4e7 100644
--- a/erts/preloaded/src/erts_code_purger.erl
+++ b/erts/preloaded/src/erts_code_purger.erl
@@ -28,24 +28,29 @@
 start() ->
     register(erts_code_purger, self()),
     process_flag(trap_exit, true),
-    loop().
-
-loop() ->
-    _ = receive
-	{purge,Mod,From,Ref} when is_atom(Mod), is_pid(From) ->
-	    Res = do_purge(Mod),
-	    From ! {reply, purge, Res, Ref};
-
-	{soft_purge,Mod,From,Ref} when is_atom(Mod), is_pid(From) ->
-	    Res = do_soft_purge(Mod),
-	    From ! {reply, soft_purge, Res, Ref};
-
-	{test_purge, Mod, From, Type, Ref} when is_atom(Mod), is_pid(From) ->
-	     do_test_purge(Mod, From, Type, Ref);
-
-	_Other -> ignore
-    end,
-    loop().
+    wait_for_request().
+
+wait_for_request() ->
+    handle_request(receive Msg -> Msg end, []).
+
+handle_request({purge, Mod, From, Ref}, Reqs) when is_atom(Mod), is_pid(From) ->
+    {Res, NewReqs} = do_purge(Mod, Reqs),
+    From ! {reply, purge, Res, Ref},
+    check_requests(NewReqs);
+handle_request({soft_purge, Mod, From, Ref}, Reqs) when is_atom(Mod), is_pid(From) ->
+    {Res, NewReqs} = do_soft_purge(Mod, Reqs),
+    From ! {reply, soft_purge, Res, Ref},
+    check_requests(NewReqs);
+handle_request({test_purge, Mod, From, Type, Ref}, Reqs) when is_atom(Mod), is_pid(From) ->
+    NewReqs = do_test_purge(Mod, From, Type, Ref, Reqs),
+    check_requests(NewReqs);
+handle_request(_Garbage, Reqs) ->
+    check_requests(Reqs).
+
+check_requests([]) ->
+    wait_for_request();
+check_requests([R|Rs]) ->
+    handle_request(R, Rs).
 
 %%
 %% Processes that tries to call a fun that belongs to
@@ -93,14 +98,15 @@ purge(Mod) when is_atom(Mod) ->
 	    Result
     end.
 
-do_purge(Mod) ->
+do_purge(Mod, Reqs) ->
     case erts_internal:purge_module(Mod, prepare) of
 	false ->
-	    {false, false};
+	    {{false, false}, Reqs};
 	true ->
-	    DidKill = check_proc_code(erlang:processes(), Mod, true),
+	    {DidKill, NewReqs} = check_proc_code(erlang:processes(),
+						 Mod, true, Reqs),
 	    true = erts_internal:purge_module(Mod, complete),
-	    {true, DidKill}
+	    {{true, DidKill}, NewReqs}
     end.
 
 %% soft_purge(Module)
@@ -116,29 +122,26 @@ soft_purge(Mod) ->
 	    Result
     end.
 
-do_soft_purge(Mod) ->
+do_soft_purge(Mod, Reqs) ->
     case erts_internal:purge_module(Mod, prepare) of
 	false ->
-	    true;
+	    {true, Reqs};
 	true ->
-	    Res = check_proc_code(erlang:processes(), Mod, false),
-	    erts_internal:purge_module(Mod,
-				       case Res of
-					   false -> abort;
-					   true -> complete
-				       end)
+	    {PurgeOp, NewReqs} = check_proc_code(erlang:processes(),
+						 Mod, false, Reqs),
+	    {erts_internal:purge_module(Mod, PurgeOp), NewReqs}
     end.
 
 %%
-%% check_proc_code(Pids, Mod, Hard) - Send asynchronous
+%% check_proc_code(Pids, Mod, Hard, Preqs) - Send asynchronous
 %%   requests to all processes to perform a check_process_code
 %%   operation. Each process will check their own state and
 %%   reply with the result. If 'Hard' equals
 %%   - true, processes that refer 'Mod' will be killed. If
 %%     any processes were killed true is returned; otherwise,
 %%     false.
-%%   - false, and any processes refer 'Mod', false will
-%%     returned; otherwise, true.
+%%   - false, and any processes refer 'Mod', 'abort' will
+%%     be returned; otherwise, 'complete'.
 %%
 %%   We only allow ?MAX_CPC_NO_OUTSTANDING_KILLS
 %%   outstanding kills. This both in order to avoid flooding
@@ -149,99 +152,91 @@ do_soft_purge(Mod) ->
 
 -define(MAX_CPC_NO_OUTSTANDING_KILLS, 10).
 
--record(cpc_static, {hard, module, tag}).
+-record(cpc_static, {hard, module, tag, purge_requests}).
 
 -record(cpc_kill, {outstanding = [],
 		   no_outstanding = 0,
 		   waiting = [],
 		   killed = false}).
 
-check_proc_code(Pids, Mod, Hard) ->
+check_proc_code(Pids, Mod, Hard, PReqs) ->
     Tag = erlang:make_ref(),
     CpcS = #cpc_static{hard = Hard,
 		       module = Mod,
-		       tag = Tag},
-    check_proc_code(CpcS, cpc_init(CpcS, Pids, 0), #cpc_kill{}, true).
-
-check_proc_code(#cpc_static{hard = true}, 0,
-		#cpc_kill{outstanding = [], waiting = [], killed = Killed},
-		true) ->
-    %% No outstanding requests. We did a hard check, so result is whether or
-    %% not we killed any processes...
-    Killed;
-check_proc_code(#cpc_static{hard = false}, 0, _KillState, Success) ->
-    %% No outstanding requests and we did a soft check...
-    Success;
-check_proc_code(#cpc_static{hard = false, tag = Tag} = CpcS, NoReq, _KillState, false) ->
-    %% Failed soft check; just cleanup the remaining replies corresponding
-    %% to the requests we've sent...
-    receive {check_process_code, {Tag, _P}, _Res} -> ok end,
-    check_proc_code(CpcS, NoReq-1, _KillState, false);
-check_proc_code(#cpc_static{tag = Tag} = CpcS, NoReq0, KillState0, Success) ->
-
-    %% Wait for a cpc reply or 'DOWN' message
-    {NoReq1, Pid, Result, KillState1} = cpc_recv(Tag, NoReq0, KillState0),
-
-    %% Check the result of the reply
-    case Result of
-	false ->
+		       tag = Tag,
+		       purge_requests = PReqs},
+    cpc_receive(CpcS, cpc_init(CpcS, Pids, 0), #cpc_kill{}, []).
+
+cpc_receive(#cpc_static{hard = true} = CpcS,
+	    0,
+	    #cpc_kill{outstanding = [], waiting = [], killed = Killed},
+	    PReqs) ->
+    %% No outstanding cpc requests. We did a hard check, so result is
+    %% whether or not we killed any processes...
+    cpc_result(CpcS, PReqs, Killed);
+cpc_receive(#cpc_static{hard = false} = CpcS, 0, _KillState, PReqs) ->
+    %% No outstanding cpc requests and we did a soft check that succeeded...
+    cpc_result(CpcS, PReqs, complete);
+cpc_receive(#cpc_static{tag = Tag} = CpcS, NoReq, KillState0, PReqs) ->
+    receive
+	{check_process_code, {Tag, _Pid}, false} ->
 	    %% Process not referring the module; done with this process...
-	    check_proc_code(CpcS, NoReq1, KillState1, Success);
-	true ->
+	    cpc_receive(CpcS, NoReq-1, KillState0, PReqs);
+	{check_process_code, {Tag, Pid}, true} ->
 	    %% Process referring the module...
 	    case CpcS#cpc_static.hard of
 		false ->
 		    %% ... and soft check. The whole operation failed so
-		    %% no point continuing; clean up and fail...
-		    check_proc_code(CpcS, NoReq1, KillState1, false);
+		    %% no point continuing; fail straight away. Garbage
+		    %% messages from this session will be ignored
+		    %% by following sessions...
+		    cpc_result(CpcS, PReqs, abort);
 		true ->
 		    %% ... and hard check; schedule kill of it...
-		    check_proc_code(CpcS, NoReq1, cpc_sched_kill(Pid, KillState1), Success)
+		    KillState1 = cpc_sched_kill(Pid, KillState0),
+		    cpc_receive(CpcS, NoReq-1, KillState1, PReqs)
 	    end;
-	'DOWN' ->
-	    %% Handled 'DOWN' message
-	    check_proc_code(CpcS, NoReq1, KillState1, Success)
+	{'DOWN', MonRef, process, _, _} ->
+	    KillState1 = cpc_handle_down(MonRef, KillState0),
+	    cpc_receive(CpcS, NoReq, KillState1, PReqs);
+	PReq when element(1, PReq) == purge;
+		  element(1, PReq) == soft_purge;
+		  element(1, PReq) == test_purge ->
+	    %% A new purge request; save it until later...
+	    cpc_receive(CpcS, NoReq, KillState0, [PReq | PReqs]);
+	_Garbage ->
+	    %% Garbage message; ignore it...
+	    cpc_receive(CpcS, NoReq, KillState0, PReqs)
     end.
 
-cpc_recv(Tag, NoReq, #cpc_kill{outstanding = []} = KillState) ->
-    receive
-	{check_process_code, {Tag, Pid}, Res} ->
-	    cpc_handle_cpc(NoReq, Pid, Res, KillState)
-    end;
-cpc_recv(Tag, NoReq,
-	 #cpc_kill{outstanding = [R0, R1, R2, R3, R4 | _]} = KillState) ->
-    receive
-	{'DOWN', R, process, _, _} when R == R0;
-					R == R1;
-					R == R2;
-					R == R3;
-					R == R4 ->
-	    cpc_handle_down(NoReq, R, KillState);
-	{check_process_code, {Tag, Pid}, Res} ->
-	    cpc_handle_cpc(NoReq, Pid, Res, KillState)
-    end;
-cpc_recv(Tag, NoReq, #cpc_kill{outstanding = [R|_]} = KillState) ->
-    receive
-	{'DOWN', R, process, _, _} ->
-	    cpc_handle_down(NoReq, R, KillState);
-	{check_process_code, {Tag, Pid}, Res} ->
-	    cpc_handle_cpc(NoReq, Pid, Res, KillState)
+cpc_result(#cpc_static{purge_requests = PReqs}, NewPReqs, Res) ->
+    {Res, PReqs ++ cpc_reverse(NewPReqs)}.
+
+cpc_reverse([_] = L) -> L;
+cpc_reverse(Xs) -> cpc_reverse(Xs, []).
+
+cpc_reverse([], Ys) -> Ys;
+cpc_reverse([X|Xs], Ys) -> cpc_reverse(Xs, [X|Ys]).
+
+cpc_handle_down(R, #cpc_kill{outstanding = Rs,
+			     no_outstanding = N} = KillState0) ->
+    try
+	NewOutst = cpc_list_rm(R, Rs),
+	KillState1 = KillState0#cpc_kill{outstanding = NewOutst,
+					 no_outstanding = N-1},
+	cpc_sched_kill_waiting(KillState1)
+    catch
+	throw : undefined -> %% Triggered by garbage message...
+	    KillState0
     end.
 
-cpc_handle_down(NoReq, R, #cpc_kill{outstanding = Rs,
-				    no_outstanding = N} = KillState) ->
-    {NoReq, undefined, 'DOWN',
-     cpc_sched_kill_waiting(KillState#cpc_kill{outstanding = cpc_list_rm(R, Rs),
-					       no_outstanding = N-1})}.
-
+cpc_list_rm(_R, []) ->
+    throw(undefined);
 cpc_list_rm(R, [R|Rs]) ->
     Rs;
 cpc_list_rm(R0, [R1|Rs]) ->
     [R1|cpc_list_rm(R0, Rs)].
 
-cpc_handle_cpc(NoReq, Pid, Res, KillState) ->
-    {NoReq-1, Pid, Res, KillState}.
-
 cpc_sched_kill_waiting(#cpc_kill{waiting = []} = KillState) ->
     KillState;
 cpc_sched_kill_waiting(#cpc_kill{outstanding = Rs,
@@ -283,64 +278,63 @@ cpc_init(CpcS, [Pid|Pids], NoReqs) ->
 %% as usual, but the tester can control when to enter the
 %% specific phases.
 %%
-do_test_purge(Mod, From, Type, Ref) when Type == true; Type == false ->
-    Mon = erlang:monitor(process, From),
-    Res = case Type of
-	      true -> do_test_hard_purge(Mod, From, Ref, Mon);
-	      false -> do_test_soft_purge(Mod, From, Ref, Mon)
-	  end,
+do_test_purge(Mod, From, true, Ref, Reqs) ->
+    {Res, NewReqs} = do_test_hard_purge(Mod, From, Ref, Reqs),
+    From ! {test_purge, Res, Ref},
+    NewReqs;
+do_test_purge(Mod, From, false, Ref, Reqs) ->
+    {Res, NewReqs} = do_test_soft_purge(Mod, From, Ref, Reqs),
     From ! {test_purge, Res, Ref},
-    erlang:demonitor(Mon, [flush]),
-    ok;
-do_test_purge(_, _, _, _) ->
-    ok.
+    NewReqs;
+do_test_purge(_, _, _, _, Reqs) ->
+    Reqs.
 
-do_test_soft_purge(Mod, From, Ref, Mon) ->
+do_test_soft_purge(Mod, From, Ref, Reqs) ->
     PrepRes = erts_internal:purge_module(Mod, prepare),
-    TestRes = test_progress(started, From, Mon, Ref, ok),
+    TestRes = test_progress(started, From, Ref, ok),
     case PrepRes of
 	false ->
-	    _ = test_progress(continued, From, Mon, Ref, TestRes),
-	    true;
+	    _ = test_progress(continued, From, Ref, TestRes),
+	    {true, Reqs};
 	true ->
-	    Res = check_proc_code(erlang:processes(), Mod, false),
-	    _ = test_progress(continued, From, Mon, Ref, TestRes),
-	    erts_internal:purge_module(Mod,
-				       case Res of
-					   false -> abort;
-					   true -> complete
-				       end)
+	    {PurgeOp, NewReqs} = check_proc_code(erlang:processes(),
+						 Mod, false, Reqs),
+	    _ = test_progress(continued, From, Ref, TestRes),
+	    {erts_internal:purge_module(Mod, PurgeOp), NewReqs}
     end.
 
-do_test_hard_purge(Mod, From, Ref, Mon) ->
+do_test_hard_purge(Mod, From, Ref, Reqs) ->
     PrepRes = erts_internal:purge_module(Mod, prepare),
-    TestRes = test_progress(started, From, Mon, Ref, ok),
+    TestRes = test_progress(started, From, Ref, ok),
     case PrepRes of
 	false ->
-	    _ = test_progress(continued, From, Mon, Ref, TestRes),
-	    {false, false};
+	    _ = test_progress(continued, From, Ref, TestRes),
+	    {{false, false}, Reqs};
 	true ->
-	    DidKill = check_proc_code(erlang:processes(), Mod, true),
-	    _ = test_progress(continued, From, Mon, Ref, TestRes),
+	    {DidKill, NewReqs} = check_proc_code(erlang:processes(),
+						 Mod, true, Reqs),
+	    _ = test_progress(continued, From, Ref, TestRes),
 	    true = erts_internal:purge_module(Mod, complete),
-	    {true, DidKill}
+	    {{true, DidKill}, NewReqs}
     end.
 
-test_progress(_State, _From, _Mon, _Ref, died) ->
+test_progress(_State, _From, _Ref, died) ->
     %% Test process died; continue so we wont
     %% leave the system in an inconsistent
     %% state...
     died;
-test_progress(started, From, Mon, Ref, ok) ->
+test_progress(started, From, Ref, ok) ->
     From ! {started, Ref},
+    Mon = erlang:monitor(process, From),
     receive
 	{'DOWN', Mon, process, From, _} -> died;
-	{continue, Ref} -> ok
+	{continue, Ref} -> erlang:demonitor(Mon, [flush]), ok
     end;
-test_progress(continued, From, Mon, Ref, ok) ->
+test_progress(continued, From, Ref, ok) ->
     From ! {continued, Ref},
+    Mon = erlang:monitor(process, From),
     receive
 	{'DOWN', Mon, process, From, _} -> died;
-	{complete, Ref} -> ok
+	{complete, Ref} -> erlang:demonitor(Mon, [flush]), ok
     end.