aboutsummaryrefslogtreecommitdiff
path: root/learn-you-some-erlang/pool
diff options
context:
space:
mode:
Diffstat (limited to 'learn-you-some-erlang/pool')
-rw-r--r--learn-you-some-erlang/pool/ppool.erl25
-rw-r--r--learn-you-some-erlang/pool/ppool_nagger.erl50
-rw-r--r--learn-you-some-erlang/pool/ppool_serv.erl113
-rw-r--r--learn-you-some-erlang/pool/ppool_sup.erl17
-rw-r--r--learn-you-some-erlang/pool/ppool_supersup.erl31
-rw-r--r--learn-you-some-erlang/pool/ppool_tests.erl200
-rw-r--r--learn-you-some-erlang/pool/ppool_worker_sup.erl15
7 files changed, 451 insertions, 0 deletions
diff --git a/learn-you-some-erlang/pool/ppool.erl b/learn-you-some-erlang/pool/ppool.erl
new file mode 100644
index 0000000..b5d34c0
--- /dev/null
+++ b/learn-you-some-erlang/pool/ppool.erl
@@ -0,0 +1,25 @@
+%%% API module for the pool
+-module(ppool).
+-export([start_link/0, stop/0, start_pool/3,
+ run/2, sync_queue/2, async_queue/2, stop_pool/1]).
+
+start_link() ->
+ ppool_supersup:start_link().
+
+stop() ->
+ ppool_supersup:stop().
+
+start_pool(Name, Limit, {M,F,A}) ->
+ ppool_supersup:start_pool(Name, Limit, {M,F,A}).
+
+stop_pool(Name) ->
+ ppool_supersup:stop_pool(Name).
+
+run(Name, Args) ->
+ ppool_serv:run(Name, Args).
+
+async_queue(Name, Args) ->
+ ppool_serv:async_queue(Name, Args).
+
+sync_queue(Name, Args) ->
+ ppool_serv:sync_queue(Name, Args).
diff --git a/learn-you-some-erlang/pool/ppool_nagger.erl b/learn-you-some-erlang/pool/ppool_nagger.erl
new file mode 100644
index 0000000..903f821
--- /dev/null
+++ b/learn-you-some-erlang/pool/ppool_nagger.erl
@@ -0,0 +1,50 @@
+%% demo module, a nagger for tasks,
+%% because the previous one wasn't good enough
+%%
+%% Can take:
+%% - a time delay for which to nag,
+%% - an adress to say where the messages should be sent
+%% - a message to send in the mailbox telling you what to nag,
+%% with an id to be able to call: ->
+%% - a command to say the task is done
+-module(ppool_nagger).
+-behaviour(gen_server).
+-export([start_link/4, stop/1]).
+-export([init/1, handle_call/3, handle_cast/2,
+ handle_info/2, code_change/3, terminate/2]).
+
+start_link(Task, Delay, Max, SendTo) ->
+ gen_server:start_link(?MODULE, {Task, Delay, Max, SendTo} , []).
+
+stop(Pid) ->
+ gen_server:call(Pid, stop).
+
+init({Task, Delay, Max, SendTo}) ->
+ process_flag(trap_exit, true), % for tests & terminate too
+ {ok, {Task, Delay, Max, SendTo}, Delay}.
+
+%%% OTP Callbacks
+handle_call(stop, _From, State) ->
+ {stop, normal, ok, State};
+handle_call(_Msg, _From, State) ->
+ {noreply, State}.
+
+handle_cast(_Msg, State) ->
+ {noreply, State}.
+
+handle_info(timeout, {Task, Delay, Max, SendTo}) ->
+ SendTo ! {self(), Task},
+ if Max =:= infinity ->
+ {noreply, {Task, Delay, Max, SendTo}, Delay};
+ Max =< 1 ->
+ {stop, normal, {Task, Delay, 0, SendTo}};
+ Max > 1 ->
+ {noreply, {Task, Delay, Max-1, SendTo}, Delay}
+ end;
+handle_info(_Msg, State) ->
+ {noreply, State}.
+
+code_change(_OldVsn, State, _Extra) ->
+ {ok, State}.
+
+terminate(_Reason, _State) -> ok.
diff --git a/learn-you-some-erlang/pool/ppool_serv.erl b/learn-you-some-erlang/pool/ppool_serv.erl
new file mode 100644
index 0000000..512b49f
--- /dev/null
+++ b/learn-you-some-erlang/pool/ppool_serv.erl
@@ -0,0 +1,113 @@
+-module(ppool_serv).
+-behaviour(gen_server).
+-export([start/4, start_link/4, run/2, sync_queue/2, async_queue/2, stop/1]).
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
+ code_change/3, terminate/2]).
+
+%% The friendly supervisor is started dynamically!
+-define(SPEC(MFA),
+ {worker_sup,
+ {ppool_worker_sup, start_link, [MFA]},
+ temporary,
+ 10000,
+ supervisor,
+ [ppool_worker_sup]}).
+
+-record(state, {limit=0,
+ sup,
+ refs,
+ queue=queue:new()}).
+
+start(Name, Limit, Sup, MFA) when is_atom(Name), is_integer(Limit) ->
+ gen_server:start({local, Name}, ?MODULE, {Limit, MFA, Sup}, []).
+
+start_link(Name, Limit, Sup, MFA) when is_atom(Name), is_integer(Limit) ->
+ gen_server:start_link({local, Name}, ?MODULE, {Limit, MFA, Sup}, []).
+
+run(Name, Args) ->
+ gen_server:call(Name, {run, Args}).
+
+sync_queue(Name, Args) ->
+ gen_server:call(Name, {sync, Args}, infinity).
+
+async_queue(Name, Args) ->
+ gen_server:cast(Name, {async, Args}).
+
+stop(Name) ->
+ gen_server:call(Name, stop).
+
+%% Gen server
+init({Limit, MFA, Sup}) ->
+ %% We need to find the Pid of the worker supervisor from here,
+ %% but alas, this would be calling the supervisor while it waits for us!
+ self() ! {start_worker_supervisor, Sup, MFA},
+ {ok, #state{limit=Limit, refs=gb_sets:empty()}}.
+
+handle_call({run, Args}, _From, S = #state{limit=N, sup=Sup, refs=R}) when N > 0 ->
+ {ok, Pid} = supervisor:start_child(Sup, Args),
+ Ref = erlang:monitor(process, Pid),
+ {reply, {ok,Pid}, S#state{limit=N-1, refs=gb_sets:add(Ref,R)}};
+handle_call({run, _Args}, _From, S=#state{limit=N}) when N =< 0 ->
+ {reply, noalloc, S};
+
+handle_call({sync, Args}, _From, S = #state{limit=N, sup=Sup, refs=R}) when N > 0 ->
+ {ok, Pid} = supervisor:start_child(Sup, Args),
+ Ref = erlang:monitor(process, Pid),
+ {reply, {ok,Pid}, S#state{limit=N-1, refs=gb_sets:add(Ref,R)}};
+handle_call({sync, Args}, From, S = #state{queue=Q}) ->
+ {noreply, S#state{queue=queue:in({From, Args}, Q)}};
+
+handle_call(stop, _From, State) ->
+ {stop, normal, ok, State};
+handle_call(_Msg, _From, State) ->
+ {noreply, State}.
+
+
+handle_cast({async, Args}, S=#state{limit=N, sup=Sup, refs=R}) when N > 0 ->
+ {ok, Pid} = supervisor:start_child(Sup, Args),
+ Ref = erlang:monitor(process, Pid),
+ {noreply, S#state{limit=N-1, refs=gb_sets:add(Ref,R)}};
+handle_cast({async, Args}, S=#state{limit=N, queue=Q}) when N =< 0 ->
+ {noreply, S#state{queue=queue:in(Args,Q)}};
+
+handle_cast(_Msg, State) ->
+ {noreply, State}.
+
+handle_info({'DOWN', Ref, process, _Pid, _}, S = #state{refs=Refs}) ->
+ io:format("received down msg~n"),
+ case gb_sets:is_element(Ref, Refs) of
+ true ->
+ handle_down_worker(Ref, S);
+ false -> %% Not our responsibility
+ {noreply, S}
+ end;
+handle_info({start_worker_supervisor, Sup, MFA}, S = #state{}) ->
+ {ok, Pid} = supervisor:start_child(Sup, ?SPEC(MFA)),
+ link(Pid),
+ {noreply, S#state{sup=Pid}};
+handle_info(Msg, State) ->
+ io:format("Unknown msg: ~p~n", [Msg]),
+ {noreply, State}.
+
+code_change(_OldVsn, State, _Extra) ->
+ {ok, State}.
+
+terminate(_Reason, _State) ->
+ ok.
+
+handle_down_worker(Ref, S = #state{limit=L, sup=Sup, refs=Refs}) ->
+ case queue:out(S#state.queue) of
+ {{value, {From, Args}}, Q} ->
+ {ok, Pid} = supervisor:start_child(Sup, Args),
+ NewRef = erlang:monitor(process, Pid),
+ NewRefs = gb_sets:insert(NewRef, gb_sets:delete(Ref,Refs)),
+ gen_server:reply(From, {ok, Pid}),
+ {noreply, S#state{refs=NewRefs, queue=Q}};
+ {{value, Args}, Q} ->
+ {ok, Pid} = supervisor:start_child(Sup, Args),
+ NewRef = erlang:monitor(process, Pid),
+ NewRefs = gb_sets:insert(NewRef, gb_sets:delete(Ref,Refs)),
+ {noreply, S#state{refs=NewRefs, queue=Q}};
+ {empty, _} ->
+ {noreply, S#state{limit=L+1, refs=gb_sets:delete(Ref,Refs)}}
+ end.
diff --git a/learn-you-some-erlang/pool/ppool_sup.erl b/learn-you-some-erlang/pool/ppool_sup.erl
new file mode 100644
index 0000000..71ec31d
--- /dev/null
+++ b/learn-you-some-erlang/pool/ppool_sup.erl
@@ -0,0 +1,17 @@
+-module(ppool_sup).
+-export([start_link/3, init/1]).
+-behaviour(supervisor).
+
+start_link(Name, Limit, MFA) ->
+ supervisor:start_link(?MODULE, {Name, Limit, MFA}).
+
+init({Name, Limit, MFA}) ->
+ MaxRestart = 1,
+ MaxTime = 3000,
+ {ok, {{one_for_all, MaxRestart, MaxTime},
+ [{serv,
+ {ppool_serv, start_link, [Name, Limit, self(), MFA]},
+ permanent,
+ 5000,
+ worker,
+ [ppool_serv]}]}}.
diff --git a/learn-you-some-erlang/pool/ppool_supersup.erl b/learn-you-some-erlang/pool/ppool_supersup.erl
new file mode 100644
index 0000000..5790c5f
--- /dev/null
+++ b/learn-you-some-erlang/pool/ppool_supersup.erl
@@ -0,0 +1,31 @@
+-module(ppool_supersup).
+-behaviour(supervisor).
+-export([start_link/0, stop/0, start_pool/3, stop_pool/1]).
+-export([init/1]).
+
+start_link() ->
+ supervisor:start_link({local, ppool}, ?MODULE, []).
+
+%% technically, a supervisor can not be killed in an easy way.
+%% Let's do it brutally!
+stop() ->
+ case whereis(ppool) of
+ P when is_pid(P) ->
+ exit(P, kill);
+ _ -> ok
+ end.
+
+start_pool(Name, Limit, MFA) ->
+ ChildSpec = {Name,
+ {ppool_sup, start_link, [Name, Limit, MFA]},
+ permanent, 10500, supervisor, [ppool_sup]},
+ supervisor:start_child(ppool, ChildSpec).
+
+stop_pool(Name) ->
+ supervisor:terminate_child(ppool, Name),
+ supervisor:delete_child(ppool, Name).
+
+init([]) ->
+ MaxRestart = 6,
+ MaxTime = 3000,
+ {ok, {{one_for_one, MaxRestart, MaxTime}, []}}.
diff --git a/learn-you-some-erlang/pool/ppool_tests.erl b/learn-you-some-erlang/pool/ppool_tests.erl
new file mode 100644
index 0000000..7cec6d0
--- /dev/null
+++ b/learn-you-some-erlang/pool/ppool_tests.erl
@@ -0,0 +1,200 @@
+-module(ppool_tests).
+-include_lib("eunit/include/eunit.hrl").
+-export([test_mfa/1, wait_mfa/1]).
+
+%%% All Test Fixtures
+start_test_() ->
+ {"It should be possible to start a pool server and give it a name",
+ {setup,
+ fun find_unique_name/0,
+ fun(Name) ->
+ [start_and_test_name(Name)]
+ end}}.
+
+mfa_test_() ->
+ {"A pool process can be allocated which will be ordered "
+ "to run an MFA call determined at start time, with arguments "
+ "provided at call time",
+ {setup,
+ fun start_ppool/0,
+ fun kill_ppool/1,
+ fun(Name) ->
+ [pool_run_mfa(Name)]
+ end}
+ }.
+
+alloc_test_() ->
+ {"A pool process can be allocated which will be ordered "
+ "to run a worker, only if there are enough which "
+ "haven't been ordered to run yet.",
+ {setup,
+ fun start_ppool/0,
+ fun kill_ppool/1,
+ fun(Name) ->
+ [pool_run_alloc(Name),
+ pool_run_noalloc(Name)]
+ end}
+ }.
+
+realloc_test_() ->
+ {"When an allocated process dies, "
+ "A new one can be allocated to replace it.",
+ {setup,
+ fun start_ppool/0,
+ fun kill_ppool/1,
+ fun(Name) ->
+ [pool_run_realloc(Name)]
+ end}
+ }.
+
+queue_test_() ->
+ {"The queue function can be used to run the function as soon as possible. "
+ "If no space is available, the worker call is added to the queue.",
+ {foreach,
+ fun start_ppool/0,
+ fun kill_ppool/1,
+ [fun(Name) -> test_async_queue(Name) end,
+ fun(Name) -> test_sync_queue(Name) end]}
+ }.
+
+supervision_test_() ->
+ {"The ppool will never restart a dead child, but all children (OTP "
+ "compliant) will be shut down when closing the pool, even if they "
+ "are trapping exits",
+ {setup,
+ fun find_unique_name/0,
+ fun test_supervision/1}}.
+
+auth_test_() ->
+ {"The ppool should only dequeue tasks after receiving a down signal "
+ "from a worker and nobody else",
+ {setup,
+ fun start_ppool/0,
+ fun kill_ppool/1,
+ fun test_auth_dealloc/1}}.
+
+%%% Setups/teardowns
+find_unique_name() ->
+ ppool:start_link(),
+ Name = list_to_atom(lists:flatten(io_lib:format("~p",[now()]))),
+ ?assertEqual(undefined, whereis(Name)),
+ Name.
+
+start_ppool() ->
+ Name = find_unique_name(),
+ ppool:start_pool(Name, 2, {ppool_nagger, start_link, []}),
+ Name.
+
+kill_ppool(Name) ->
+ ppool:stop_pool(Name).
+
+%%% Actual tests
+start_and_test_name(Name) ->
+ ppool:start_pool(Name, 1, {ppool_nagger, start_link, []}),
+ A = whereis(Name),
+ ppool:stop_pool(Name),
+ timer:sleep(100),
+ B = whereis(Name),
+ [?_assert(undefined =/= A),
+ ?_assertEqual(undefined, B)].
+
+pool_run_mfa(Name) ->
+ ppool:run(Name, [i_am_running, 1, 1, self()]),
+ X = receive
+ {_Pid, i_am_running} -> ok
+ after 3000 ->
+ timeout
+ end,
+ ?_assertEqual(ok, X).
+
+pool_run_alloc(Name) ->
+ {ok, Pid} = ppool:run(Name, [i_am_running, 1, 1, self()]),
+ X = receive
+ {Pid, i_am_running} -> ok
+ after 3000 ->
+ timeout
+ end,
+ [?_assert(is_pid(Pid)),
+ ?_assertEqual(ok, X)].
+
+pool_run_noalloc(Name) ->
+ %% Init function should have set the limit to 2
+ ppool:run(Name, [i_am_running, 300, 1, self()]),
+ ppool:run(Name, [i_am_running, 300, 1, self()]),
+ X = ppool:run(Name, [i_am_running, 1, 1, self()]),
+ ?_assertEqual(noalloc, X).
+
+pool_run_realloc(Name) ->
+ %% Init function should have set the limit to 2
+ {ok, A} = ppool:run(Name, [i_am_running, 500, 1, self()]),
+ timer:sleep(100),
+ {ok, B} = ppool:run(Name, [i_am_running, 500, 1, self()]),
+ timer:sleep(600),
+ {ok, Pid} = ppool:run(Name, [i_am_running, 1, 1, self()]),
+ timer:sleep(100),
+ L = flush(),
+ [?_assert(is_pid(Pid)),
+ ?_assertEqual([{A,i_am_running}, {B,i_am_running}, {Pid,i_am_running}],
+ L)].
+
+test_async_queue(Name) ->
+ %% Still two elements max!
+ ok = ppool:async_queue(Name, [i_am_running, 2000, 1, self()]),
+ ok = ppool:async_queue(Name, [i_am_running, 2000, 1, self()]),
+ noalloc = ppool:run(Name, [i_am_running, 2000, 1, self()]),
+ ok = ppool:async_queue(Name, [i_am_running, 500, 1, self()]),
+ timer:sleep(3500),
+ L = flush(),
+ ?_assertMatch([{_, i_am_running}, {_, i_am_running}, {_, i_am_running}], L).
+
+test_sync_queue(Name) ->
+ %% Hell yase, two max
+ {ok, Pid} = ppool:sync_queue(Name, [i_am_running, 200, 1, self()]),
+ ok = ppool:async_queue(Name, [i_am_running, 200, 1, self()]),
+ ok = ppool:async_queue(Name, [i_am_running, 200, 1, self()]),
+ {ok, Pid2} = ppool:sync_queue(Name, [i_am_running, 100, 1, self()]),
+ timer:sleep(300),
+ L = flush(),
+ [?_assert(is_pid(Pid)),
+ ?_assert(is_pid(Pid2)),
+ ?_assertMatch([{_,i_am_running}, {_,i_am_running},
+ {_,i_am_running}, {_,i_am_running}],
+ L)].
+
+test_supervision(Name) ->
+ ppool:start_pool(Name, 1, {ppool_nagger, start_link, []}),
+ {ok, Pid} = ppool:run(Name, [sup, 10000, 100, self()]),
+ ppool:stop_pool(Name),
+ timer:sleep(100),
+ ?_assertEqual(undefined, process_info(Pid)).
+
+test_auth_dealloc(Name) ->
+ %% Hell yase, two max
+ {ok, _Pid} = ppool:sync_queue(Name, [i_am_running, 500, 1, self()]),
+ ok = ppool:async_queue(Name, [i_am_running, 10000, 1, self()]),
+ ok = ppool:async_queue(Name, [i_am_running, 10000, 1, self()]),
+ ok = ppool:async_queue(Name, [i_am_running, 1, 1, self()]),
+ timer:sleep(600),
+ Name ! {'DOWN', make_ref(), process, self(), normal},
+ Name ! {'DOWN', make_ref(), process, self(), normal},
+ Name ! {'DOWN', make_ref(), process, self(), normal},
+ timer:sleep(200),
+ L = flush(),
+ ?_assertMatch([{_,i_am_running}], L).
+
+
+
+flush() ->
+ receive
+ X -> [X|flush()]
+ after 0 ->
+ []
+ end.
+
+%% Exported Helper functions
+test_mfa(Pid) ->
+ Pid ! i_am_running.
+
+wait_mfa(Pid) ->
+ Pid ! i_am_running,
+ timer:sleep(3000).
diff --git a/learn-you-some-erlang/pool/ppool_worker_sup.erl b/learn-you-some-erlang/pool/ppool_worker_sup.erl
new file mode 100644
index 0000000..2467c47
--- /dev/null
+++ b/learn-you-some-erlang/pool/ppool_worker_sup.erl
@@ -0,0 +1,15 @@
+-module(ppool_worker_sup).
+-export([start_link/1, init/1]).
+-behaviour(supervisor).
+
+start_link(MFA = {_,_,_}) ->
+ supervisor:start_link(?MODULE, MFA).
+
+init({M,F,A}) ->
+ MaxRestart = 5,
+ MaxTime = 3600,
+ {ok, {{simple_one_for_one, MaxRestart, MaxTime},
+ [{ppool_worker,
+ {M,F,A},
+ temporary, 5000, worker, [M]}]}}.
+
generated by cgit v1.2.3 (git 2.39.1) at 2024-09-01 19:02:34 +0000