%% Copyright (c) 2016, Takeru Ohta %% %% This software is released under the MIT License. %% See the LICENSE file in the project root for full license information. %% %% @doc Plumtree based Process Group -module(ppg). %%---------------------------------------------------------------------------------------------------------------------- %% Exported API %%---------------------------------------------------------------------------------------------------------------------- -export([default_join_options/0]). -export([create/1]). -export([delete/1]). -export([which_groups/0]). -export([get_members/1]). -export([get_local_members/1]). -export([get_closest_member/1]). -export([join/2, join/3]). -export([leave/1]). -export([broadcast/2]). -export_type([name/0]). -export_type([member/0]). -export_type([channel/0]). -export_type([message/0]). -export_type([join_options/0, join_option/0]). -export_type([plumtree_option/0]). -export_type([hyparview_option/0]). %%---------------------------------------------------------------------------------------------------------------------- %% Types %%---------------------------------------------------------------------------------------------------------------------- -type name() :: term(). %% Group Name -type member() :: pid(). %% A member process %% %% This process receives messages which are broadcasted to the belonging group -type channel() :: pid(). %% A broadcast channel -type message() :: term(). %% A broadcast message -type join_options() :: [join_option()]. -type join_option() :: {plumtree, [plumtree_option()]} | {hyparview, [hyparview_option()]}. -type plumtree_option() :: {gossip_wait_timeout, timeout()} | {ihave_retention_period, timeout()} | {wehave_retention_period, timeout()} | {ticktime, timeout()}. -type hyparview_option() :: {active_view_size, pos_integer()} | {passive_view_size, pos_integer()} | {active_random_walk_length, pos_integer()} | {passive_random_walk_length, pos_integer()} | {shuffle_count, pos_integer()} | {shuffle_interval, timeout()}. %%---------------------------------------------------------------------------------------------------------------------- %% Exported Functions %%---------------------------------------------------------------------------------------------------------------------- %% @doc Returns the default join options -spec default_join_options() -> join_options(). default_join_options() -> [ {plumtree, ppg_plumtree:default_options()}, {hyparview, ppg_hyparview:default_options()} ]. %% @doc Creates a new group %% %% If the group already exists on the node, nothing happens. %% %% Unlike the pg2 module, every nodes which are interested in the group must call this function. -spec create(name()) -> ok. create(Group) -> case ppg_group_sup:start_child(Group) of {ok, _} -> ok; {error, {already_started, _}} -> ok; Other -> error({badresult, Other}, [Group]) end. %% @doc Deletes the group from the local node -spec delete(name()) -> ok. delete(Group) -> ppg_group_sup:stop_child(Group). %% @doc Returns the list of locally known groups -spec which_groups() -> [name()]. which_groups() -> [Group || {Group, _} <- ppg_group_sup:which_children()]. %% @doc Returns all processes in the group `Group' %% %% This function is provided for debugging purposes only. -spec get_members(name()) -> {ok, [{member(), channel()}]} | {error, {no_such_group, name()}}. get_members(Group) -> case get_closest_member(Group) of {error, Reason} -> {error, Reason}; {ok, {_, Peer0}} -> Members = [{Member, Peer1} || {Peer1, Member, _} <- ppg_peer:get_graph(Peer0, 5000)], {ok, Members} end. %% @doc Returns a process on the local node, if such a process exist. Otherwise, it returns the contact service process. -spec get_closest_member(name()) -> {ok, {member(), channel()}} | {error, Reason} when Reason :: {no_such_group, name()} | {no_reachable_member, name()}. get_closest_member(Group) -> case ppg_group_sup:find_child(Group) of error -> {error, {no_such_group, Group}}; {ok, Sup} -> case ppg_peer_sup:which_children(Sup) of [] -> case ppg_contact_service:find_peer(ppg_contact_service:new(Group)) of error -> {error, {no_reachable_member, Group}}; {ok, Peer} -> {ok, Peer} end; Peers -> Peer = lists:nth(rand:uniform(length(Peers)), Peers), {ok, {ppg_peer:get_member(Peer), Peer}} end end. %% @doc Returns all processes running on the local node in the group `Group' -spec get_local_members(name()) -> {ok, [{member(), channel()}]} | {error, {no_such_group, name()}}. get_local_members(Group) -> case ppg_group_sup:find_child(Group) of error -> {error, {no_such_group, Group}}; {ok, Sup} -> Members = [{ppg_peer:get_member(Peer), Peer} || Peer <- ppg_peer_sup:which_children(Sup)], {ok, Members} end. %% @equiv join(Group, Member, default_join_options()) -spec join(name(), ppg:member()) -> {ok, channel()} | {error, {no_such_group, name()}}. join(Group, Member) -> join(Group, Member, default_join_options()). %% @doc Joins the process `Member' to the group `Group' %% %% After the joining, `Member' can broadcast messages via `Channel'. %% %% If you are interested in the disconnection of the channel, %% you should apply {@link erlang:monitor/2} or {@link erlang:link/1} to the channel. -spec join(name(), ppg:member(), join_options()) -> {ok, Channel::channel()} | {error, {no_such_group, name()}}. join(Group, Member, Options) -> _ = is_list(Options) orelse error(badarg, [Group, Member, Options]), case ppg_peer_sup:start_child(Group, Member, Options) of {ok, Pid} -> {ok, Pid}; {error, {no_such_group, _}} -> {error, {no_such_group, Group}}; Other -> error({badresult, Other}, [Group, Member, Options]) end. %% @doc Leaves the group associated with `Channel' -spec leave(channel()) -> ok. leave(Channel) -> ppg_peer:stop(Channel). %% @doc Broadcasts `Message' to the group associated with `Channel' -spec broadcast(channel(), message()) -> ok. broadcast(Channel, Message) -> ppg_peer:broadcast(Channel, Message).