-
Notifications
You must be signed in to change notification settings - Fork 23
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Ensure coordinators are supervised and pooled
This serves two purposes: first, no process is left unsupervised, and second, solves the potential bottleneck a single 'gen_event' process could become at handling all the notifications.
- Loading branch information
1 parent
d4a1071
commit fcbdeda
Showing
8 changed files
with
199 additions
and
141 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,51 @@ | ||
| %% @private | ||
| %% @see amoc_coordinator | ||
| %% @copyright 2023 Erlang Solutions Ltd. | ||
| %% @doc Supervisor for a pool of handlers for a new supervision tree | ||
| -module(amoc_coordinator_sup). | ||
|
|
||
| -behaviour(supervisor). | ||
|
|
||
| -export([start_link/1, init/1]). | ||
|
|
||
| -spec start_link({amoc_coordinator:name(), amoc_coordinator:coordination_plan(), timeout()}) -> | ||
| gen_server:start_ret(). | ||
| start_link({Name, Plan, Timeout}) -> | ||
| supervisor:start_link(?MODULE, {Name, Plan, Timeout}). | ||
|
|
||
| -spec init({amoc_coordinator:name(), amoc_coordinator:coordination_plan(), timeout()}) -> | ||
| {ok, {supervisor:sup_flags(), [supervisor:child_spec()]}}. | ||
| init({Name, Plan, Timeout}) -> | ||
| %% Order matters, as we need to ensure that: | ||
| %% - all the items in the plan with NoOfUsers =:= all are executed last | ||
| %% - the timeout handler is executed first | ||
| {All, NotAll} = lists:partition(fun partitioner/1, Plan), | ||
| OrderedPlan = lists:reverse(All) ++ lists:reverse(NotAll), | ||
| OrderedChilds = [ worker_spec(Item) || Item <- OrderedPlan ], | ||
| TimeoutChild = timeout_child(Name, Timeout), | ||
| Childs = [TimeoutChild | OrderedChilds], | ||
|
|
||
| SupFlags = #{strategy => one_for_one, intensity => 1, period => 5}, | ||
| {ok, {SupFlags, Childs}}. | ||
|
|
||
| %% Helpers | ||
| timeout_child(Name, Timeout) -> | ||
| #{id => {amoc_coordinator_timeout, Name, Timeout}, | ||
| start => {amoc_coordinator_timeout, start_link, [Name, Timeout]}, | ||
| restart => permanent, | ||
| shutdown => timer:seconds(5), | ||
| type => worker, | ||
| modules => [amoc_coordinator_timeout]}. | ||
|
|
||
| worker_spec(Item) -> | ||
| #{id => {amoc_coordinator_worker, Item}, | ||
| start => {amoc_coordinator_worker, start_link, [Item]}, | ||
| restart => permanent, | ||
| shutdown => timer:seconds(5), | ||
| type => worker, | ||
| modules => [amoc_coordinator_worker]}. | ||
|
|
||
| partitioner({all, _}) -> | ||
| true; | ||
| partitioner(_) -> | ||
| false. |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,65 @@ | ||
| %% @private | ||
| %% @see amoc_coordinator | ||
| %% @copyright 2023 Erlang Solutions Ltd. | ||
| %% @doc Global supervisor for all started coordination actions | ||
| -module(amoc_coordinator_sup_sup). | ||
|
|
||
| -behaviour(supervisor). | ||
|
|
||
| -export([start_coordinator/3, stop_coordinator/1, get_workers/1]). | ||
|
|
||
| -export([start_link/0, init/1]). | ||
|
|
||
| -spec start_coordinator(amoc_coordinator:name(), amoc_coordinator:coordination_plan(), timeout()) -> | ||
| {ok, pid()} | {error, term()}. | ||
| start_coordinator(Name, Plan, Timeout) -> | ||
| case supervisor:start_child(?MODULE, [{Name, Plan, Timeout}]) of | ||
| {ok, Coordinator} -> | ||
| Children = supervisor:which_children(Coordinator), | ||
| [TimeoutWorker] = [ Pid || {{amoc_coordinator_timeout, _, _}, Pid, _, _} <- Children ], | ||
| Workers = [ Pid || {{amoc_coordinator_worker, _}, Pid, _, _} <- Children ], | ||
| store_coordinator(Name, Coordinator, TimeoutWorker, Workers), | ||
| {ok, Coordinator}; | ||
| Other -> | ||
| Other | ||
| end. | ||
|
|
||
| -spec stop_coordinator(amoc_coordinator:name()) -> | ||
| ok | {error, term()}. | ||
| stop_coordinator(Name) -> | ||
| case ets:lookup(?MODULE, Name) of | ||
| [{_, #{coordinator := Coordinator}}] -> | ||
| supervisor:terminate_child(?MODULE, Coordinator); | ||
| [] -> | ||
| {error, not_found} | ||
| end. | ||
|
|
||
| -spec get_workers(amoc_coordinator:name()) -> | ||
| {ok, pid(), [pid()]} | {error, term()}. | ||
| get_workers(Name) -> | ||
| case ets:lookup(?MODULE, Name) of | ||
| [{_, #{timeout_worker := TimeoutWorker, workers := Workers}}] -> | ||
| {ok, TimeoutWorker, Workers}; | ||
| [] -> | ||
| {error, not_found} | ||
| end. | ||
|
|
||
| store_coordinator(Name, Coordinator, TimeoutWorker, Workers) -> | ||
| Item = #{coordinator => Coordinator, timeout_worker => TimeoutWorker, workers => Workers}, | ||
| ets:insert(?MODULE, {Name, Item}). | ||
|
|
||
| -spec start_link() -> {ok, Pid :: pid()}. | ||
| start_link() -> | ||
| supervisor:start_link({local, ?MODULE}, ?MODULE, []). | ||
|
|
||
| -spec init(term()) -> {ok, {supervisor:sup_flags(), [supervisor:child_spec()]}}. | ||
| init([]) -> | ||
| ets:new(?MODULE, [named_table, ordered_set, public, {read_concurrency, true}]), | ||
| SupFlags = #{strategy => simple_one_for_one}, | ||
| AChild = #{id => amoc_coordinator_sup, | ||
| start => {amoc_coordinator_sup, start_link, []}, | ||
| restart => transient, | ||
| shutdown => infinity, | ||
| type => supervisor, | ||
| modules => [amoc_coordinator_sup]}, | ||
| {ok, {SupFlags, [AChild]}}. |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,35 @@ | ||
| -module(amoc_coordinator_timeout). | ||
|
|
||
| -export([start_link/2, init/2]). | ||
|
|
||
| -spec start_link(amoc_coordinator:name(), timeout()) -> | ||
| {ok, pid()}. | ||
| start_link(Name, Timeout) -> | ||
| proc_lib:start_link(?MODULE, init, [Name, Timeout]). | ||
|
|
||
| -spec init(amoc_coordinator:name(), timeout()) -> | ||
| no_return(). | ||
| init(Name, Timeout) -> | ||
| proc_lib:init_ack({ok, self()}), | ||
| case Timeout of | ||
| infinity -> | ||
| timeout_fn(Name, infinity, infinity); | ||
| Int when is_integer(Int), Int > 0 -> | ||
| timeout_fn(Name, timer:seconds(Timeout), infinity) | ||
| end. | ||
|
|
||
| timeout_fn(Name, CoordinationTimeout, Timeout) -> | ||
| %% coordinator_timeout | reset_coordinator | {coordinate, {pid(), term()}} | ||
| receive | ||
| coordinator_timeout -> | ||
| timeout_fn(Name, CoordinationTimeout, CoordinationTimeout); | ||
| reset_coordinator -> | ||
| timeout_fn(Name, CoordinationTimeout, CoordinationTimeout); | ||
| {coordinate, _} -> | ||
| timeout_fn(Name, CoordinationTimeout, CoordinationTimeout); | ||
| _ -> %% other, end | ||
| ok | ||
| after Timeout -> | ||
| amoc_coordinator:notify(Name, coordinator_timeout), | ||
| timeout_fn(Name, CoordinationTimeout, infinity) | ||
| end. |
Oops, something went wrong.