<commit_before>%% -------------------------------------------------------------------
%%
%% Copyright (c) 2013 Basho Technologies, Inc.  All Rights Reserved.
%%
%% This file is provided to you under the Apache License,
%% Version 2.0 (the "License"); you may not use this file
%% except in compliance with the License.  You may obtain
%% a copy of the License at
%%
%%   http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing,
%% software distributed under the License is distributed on an
%% "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
%% KIND, either express or implied.  See the License for the
%% specific language governing permissions and limitations
%% under the License.
%%
%% -------------------------------------------------------------------
-module(partisan_plumtree_broadcast).

-behaviour(gen_server).

-include("partisan.hrl").
-include("partisan_logger.hrl").

%% API
-export([start_link/0,
         start_link/5,
         broadcast/2,
         update/1,
         broadcast_members/0,
         broadcast_members/1,
         exchanges/0,
         exchanges/1,
         cancel_exchanges/1]).

%% Debug API
-export([debug_get_peers/2,
         debug_get_peers/3,
         debug_get_tree/2]).

%% gen_server callbacks
-export([init/1,
         handle_call/3,
         handle_cast/2,
         handle_info/2,
         terminate/2,
         code_change/3]).

-define(SERVER, ?MODULE).

-type message_id()      :: any().
-type message_round()   :: non_neg_integer().
%% Lazy messages that have not been acked. Messages are added to
%% this set when a node is sent a lazy message (or when it should be
%% sent one sometime in the future). Messages are removed when the lazy
%% pushes are acknowleged via graft or ignores. Entries are keyed by their
%% destination
%% These are stored in the ?PLUMTREE_OUTSTANDING ets table under using nodename
%% as key.
%% PLUMTREE_OUTSTANDING is created and owned by partisan_sup
-type outstanding()     :: {message_id(), module(), message_round(), node()}.
-type exchange()        :: {module(), node(), reference(), pid()}.
-type exchanges()       :: [exchange()].
-type selector()        ::  all
                            | {peer, node()}
                            | {mod, module()}
                            | reference()
                            | pid().


-record(state, {
    %% Initially trees rooted at each node are the same.
    %% Portions of that tree belonging to this node are
    %% shared in this set.
    common_eagers :: nodeset() | undefined,

    %% Initially trees rooted at each node share the same lazy links.
    %% Typically this set will contain a single element. However, it may
    %% contain more in large clusters and may be empty for clusters with
    %% less than three nodes.
    common_lazys  :: nodeset() | undefined,

    %% A mapping of sender node (root of each broadcast tree)
    %% to this node's portion of the tree. Elements are
    %% added to this structure as messages rooted at a node
    %% propogate to this node. Nodes that are never the
    %% root of a message will never have a key added to
    %% `eager_sets'
    eager_sets    :: #{node() := nodeset()} | undefined,

    %% A Mapping of sender node (root of each spanning tree)
    %% to this node's set of lazy peers. Elements are added
    %% to this structure as messages rooted at a node
    %% propogate to this node. Nodes that are never the root
    %% of a message will never have a key added to `lazy_sets'
    lazy_sets     :: #{node() := nodeset()} | undefined,

    %% Set of registered modules that may handle messages that
    %% have been broadcast
    mods          :: [module()],

    %% List of outstanding exchanges
    exchanges     :: exchanges(),

    %% Set of all known members. Used to determine
    %% which members have joined and left during a membership update
    all_members   :: nodeset() | undefined,

    %% Lazy tick period in milliseconds. On every tick all outstanding
    %% lazy pushes are sent out
    lazy_tick_period :: non_neg_integer(),

    %% Exchange tick period in milliseconds that may or may not occur
    exchange_tick_period :: non_neg_integer()

}).

-type nodeset()         :: ordsets:ordset(node()).

-type state()           :: #state{}.



%% =============================================================================
%% API
%% =============================================================================



%% -----------------------------------------------------------------------------
%% @doc Starts the broadcast server on this node. The initial membership list is
%% fetched from the ring. If the node is a singleton then the initial eager and
%% lazy sets are empty. If there are two nodes, each will be in the others
%% eager set and the lazy sets will be empty. When number of members is less
%% than 5, each node will initially have one other node in its eager set and
%% lazy set. If there are more than five nodes each node will have at most two
%% other nodes in its eager set and one in its lazy set, initially.
%% In addition, after the broadcast server is started, a callback is registered
%% with ring_events to generate membership updates as the ring changes.
%% @end
%% -----------------------------------------------------------------------------
-spec start_link() -> {ok, pid()} | ignore | {error, term()}.

start_link() ->
    LazyTickPeriod = partisan_config:get(
        lazy_tick_period, ?DEFAULT_LAZY_TICK_PERIOD
    ),
    ExchangeTickPeriod = partisan_config:get(
        exchange_tick_period, ?DEFAULT_EXCHANGE_TICK_PERIOD
    ),
    Opts = #{
        lazy_tick_period => LazyTickPeriod,
        exchange_tick_period => ExchangeTickPeriod
    },

    {ok, Members} = partisan_peer_service:members(),

    ?LOG_DEBUG("Peer sampling service members: ~p", [Members]),

    %% The peer service has already sampled the members, we start off
    %% with pure gossip (ie. all members are in the eager push list and lazy
    %% list is empty)
    InitEagers = Members,
    InitLazys = [],

    ?LOG_DEBUG("Init peers, eager: ~p, lazy: ~p", [InitEagers, InitLazys]),

    Mods = partisan_config:get(broadcast_mods, []),

    Res = start_link(Members, InitEagers, InitLazys, Mods, Opts),

    %% We subscribe to the membership change events
    partisan_peer_service:add_sup_callback(fun ?MODULE:update/1),

    Res.


%% -----------------------------------------------------------------------------
%% @doc Starts the broadcast server on this node. `InitMembers' must be a list
%% of all members known to this node when starting the broadcast server.
%% `InitEagers' are the initial peers of this node for all broadcast trees.
%% `InitLazys' is a list of random peers not in `InitEagers' that will be used
%% as the initial lazy peer shared by all trees for this node. If the number
%% of nodes in the cluster is less than 3, `InitLazys' should be an empty list.
%% `InitEagers' and `InitLazys' must also be subsets of `InitMembers'. `Mods' is
%% a list of modules that may be handlers for broadcasted messages. All modules
%% in `Mods' should implement the `partisan_plumtree_broadcast_handler'
%% behaviour.
%% `Opts' is a proplist or map with the following possible options:
%%  <ul>
%%  <li> `lazy_tick_period :: non_neg_integer()' - Flush all outstanding lazy pushes period (in milliseconds)</li>
%%  <li> `exchange_tick_period :: non_neg_integer()' - Possibly perform an exchange period (in milliseconds)</li>
%% </ul>
%%
%% NOTE: When starting the server using start_link/2 no automatic membership update from
%% ring_events is registered. Use start_link/0.
%% @end
%% -----------------------------------------------------------------------------

-spec start_link(
    InitMembers :: [node()],
    InitEagers :: [node()],
    InitLazys :: [node()],
    Mods :: [module()],
    Opts :: proplists:proplist() | map()) ->
    {ok, pid()} | ignore | {error, term()}.

start_link(InitMembers, InitEagers, InitLazys, Mods, Opts) when is_list(Opts) ->
    start_link(InitMembers, InitEagers, InitLazys, Mods, maps:from_list(Opts));

start_link(InitMembers, InitEagers, InitLazys, Mods, Opts) when is_map(Opts) ->
    StartOpts = [
        {spawn_opt, ?PARALLEL_SIGNAL_OPTIMISATION([])}
    ],
    gen_server:start_link(
        {local, ?SERVER}, ?MODULE,
        [InitMembers, InitEagers, InitLazys, Mods, Opts],
        StartOpts
    ).


%% -----------------------------------------------------------------------------
%% @doc Broadcasts a message originating from this node. The message will be
%% delivered to each node at least once. The `Mod' passed is responsible for
%% handling the message on remote nodes as well as providing some other
%% information both locally and and on other nodes.
%% `Mod' must be loaded on all members of the clusters and implement the
%% `partisan_plumtree_broadcast_handler' behaviour.
%% @end
%% -----------------------------------------------------------------------------
-spec broadcast(any(), module()) -> ok.

broadcast(Broadcast, Mod) ->
    {MessageId, Payload} = Mod:broadcast_data(Broadcast),
    gen_server:cast(?SERVER, {broadcast, MessageId, Payload, Mod}).


%% -----------------------------------------------------------------------------
%% @doc Notifies broadcast server of membership update
%% @end
%% -----------------------------------------------------------------------------
-spec update([node()]) -> ok.

update(LocalState0) ->
    LocalState = partisan_peer_service:decode(LocalState0),
    gen_server:cast(?SERVER, {update, LocalState}).


%% -----------------------------------------------------------------------------
%% @doc Returns the broadcast servers view of full cluster membership.
%% Wait indefinitely for a response is returned from the process.
%% @end
%% -----------------------------------------------------------------------------
-spec broadcast_members() -> nodeset().

broadcast_members() ->
    broadcast_members(infinity).


%% -----------------------------------------------------------------------------
%% @doc Returns the broadcast servers view of full cluster membership.
%% Waits `Timeout' ms for a response from the server.
%% @end
%% -----------------------------------------------------------------------------
-spec broadcast_members(infinity | pos_integer()) -> nodeset().

broadcast_members(Timeout) ->
    gen_server:call(?SERVER, broadcast_members, Timeout).


%% -----------------------------------------------------------------------------
%% @doc return a list of exchanges, started by broadcast on thisnode, that are
%% running.
%% @end
%% -----------------------------------------------------------------------------
-spec exchanges() -> exchanges().

exchanges() ->
    exchanges(partisan:node()).


%% -----------------------------------------------------------------------------
%% @doc returns a list of exchanges, started by broadcast on `Node', that are
%% running.
%% @end
%% -----------------------------------------------------------------------------
-spec exchanges(node()) -> partisan_plumtree_broadcast:exchanges().

exchanges(Node) ->
    gen_server:call({?SERVER, Node}, exchanges, infinity).


%% -----------------------------------------------------------------------------
%% @doc cancel exchanges started by this node.
%% @end
%% -----------------------------------------------------------------------------
-spec cancel_exchanges(selector()) -> exchanges().

cancel_exchanges(WhichExchanges) ->
    gen_server:call(?SERVER, {cancel_exchanges, WhichExchanges}, infinity).



%% =============================================================================
%% DEBUG API
%% =============================================================================



%% @doc return the peers for `Node' for the tree rooted at `Root'.
%% Wait indefinitely for a response is returned from the process
-spec debug_get_peers(node(), node()) ->
    {nodeset(), nodeset()}.

debug_get_peers(Node, Root) ->
    debug_get_peers(Node, Root, infinity).


%% @doc return the peers for `Node' for the tree rooted at `Root'.
%% Waits `Timeout' ms for a response from the server
-spec debug_get_peers(node(), node(), infinity | pos_integer()) ->
    {nodeset(), nodeset()}.

debug_get_peers(Node, Root, Timeout) ->
    gen_server:call({?SERVER, Node}, {get_peers, Root}, Timeout).


%% @doc return peers for all `Nodes' for tree rooted at `Root'
%% Wait indefinitely for a response is returned from the process
-spec debug_get_tree(node(), [node()]) ->
    [{node(), {nodeset(), nodeset()}}].

debug_get_tree(Root, Nodes) ->
    [begin
         Peers = try debug_get_peers(Node, Root)
                 catch _:_ -> down
                 end,
         {Node, Peers}
     end || Node <- Nodes].



%% =============================================================================
%% GEN_SERVER CALLBACKS
%% =============================================================================



-spec init([[any()], ...]) -> {ok, state()}.

init([Members, InitEagers0, InitLazys0, Mods, Opts]) ->
    LazyTickPeriod = maps:get(lazy_tick_period, Opts),
    ExchangeTickPeriod = maps:get(exchange_tick_period, Opts),
    schedule_lazy_tick(LazyTickPeriod),
    schedule_exchange_tick(ExchangeTickPeriod),
    State1 =  #state{
        mods = lists:usort(Mods),
        exchanges = [],
        lazy_tick_period = LazyTickPeriod,
        exchange_tick_period = ExchangeTickPeriod
    },

    AllMembers = ordsets:from_list(Members),
    InitEagers = ordsets:from_list(InitEagers0),
    InitLazys = ordsets:from_list(InitLazys0),

    State2 = reset_peers(AllMembers, InitEagers, InitLazys, State1),
    {ok, State2}.


-spec handle_call(term(), {pid(), term()}, state()) -> {reply, term(), state()}.

handle_call({get_peers, Root}, _From, State) ->
    EagerPeers = all_peers(
        Root, State#state.eager_sets, State#state.common_eagers
    ),
    LazyPeers = all_peers(
        Root, State#state.lazy_sets, State#state.common_lazys
    ),
    {reply, {EagerPeers, LazyPeers}, State};
<commit_msg>move event handler setup to the broadcast server process<commit_after>%% -------------------------------------------------------------------
%%
%% Copyright (c) 2013 Basho Technologies, Inc.  All Rights Reserved.
%%
%% This file is provided to you under the Apache License,
%% Version 2.0 (the "License"); you may not use this file
%% except in compliance with the License.  You may obtain
%% a copy of the License at
%%
%%   http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing,
%% software distributed under the License is distributed on an
%% "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
%% KIND, either express or implied.  See the License for the
%% specific language governing permissions and limitations
%% under the License.
%%
%% -------------------------------------------------------------------
-module(partisan_plumtree_broadcast).

-behaviour(gen_server).

-include("partisan.hrl").
-include("partisan_logger.hrl").

%% API
-export([start_link/0,
         start_link/5,
         broadcast/2,
         update/1,
         broadcast_members/0,
         broadcast_members/1,
         exchanges/0,
         exchanges/1,
         cancel_exchanges/1]).

%% Debug API
-export([debug_get_peers/2,
         debug_get_peers/3,
         debug_get_tree/2]).

%% gen_server callbacks
-export([init/1,
         handle_call/3,
         handle_cast/2,
         handle_info/2,
         terminate/2,
         code_change/3]).

-define(SERVER, ?MODULE).

-type message_id()      :: any().
-type message_round()   :: non_neg_integer().
%% Lazy messages that have not been acked. Messages are added to
%% this set when a node is sent a lazy message (or when it should be
%% sent one sometime in the future). Messages are removed when the lazy
%% pushes are acknowleged via graft or ignores. Entries are keyed by their
%% destination
%% These are stored in the ?PLUMTREE_OUTSTANDING ets table under using nodename
%% as key.
%% PLUMTREE_OUTSTANDING is created and owned by partisan_sup
-type outstanding()     :: {message_id(), module(), message_round(), node()}.
-type exchange()        :: {module(), node(), reference(), pid()}.
-type exchanges()       :: [exchange()].
-type selector()        ::  all
                            | {peer, node()}
                            | {mod, module()}
                            | reference()
                            | pid().


-record(state, {
    %% Initially trees rooted at each node are the same.
    %% Portions of that tree belonging to this node are
    %% shared in this set.
    common_eagers :: nodeset() | undefined,

    %% Initially trees rooted at each node share the same lazy links.
    %% Typically this set will contain a single element. However, it may
    %% contain more in large clusters and may be empty for clusters with
    %% less than three nodes.
    common_lazys  :: nodeset() | undefined,

    %% A mapping of sender node (root of each broadcast tree)
    %% to this node's portion of the tree. Elements are
    %% added to this structure as messages rooted at a node
    %% propogate to this node. Nodes that are never the
    %% root of a message will never have a key added to
    %% `eager_sets'
    eager_sets    :: #{node() := nodeset()} | undefined,

    %% A Mapping of sender node (root of each spanning tree)
    %% to this node's set of lazy peers. Elements are added
    %% to this structure as messages rooted at a node
    %% propogate to this node. Nodes that are never the root
    %% of a message will never have a key added to `lazy_sets'
    lazy_sets     :: #{node() := nodeset()} | undefined,

    %% Set of registered modules that may handle messages that
    %% have been broadcast
    mods          :: [module()],

    %% List of outstanding exchanges
    exchanges     :: exchanges(),

    %% Set of all known members. Used to determine
    %% which members have joined and left during a membership update
    all_members   :: nodeset() | undefined,

    %% Lazy tick period in milliseconds. On every tick all outstanding
    %% lazy pushes are sent out
    lazy_tick_period :: non_neg_integer(),

    %% Exchange tick period in milliseconds that may or may not occur
    exchange_tick_period :: non_neg_integer()

}).

-type nodeset()         :: ordsets:ordset(node()).

-type state()           :: #state{}.



%% =============================================================================
%% API
%% =============================================================================



%% -----------------------------------------------------------------------------
%% @doc Starts the broadcast server on this node. The initial membership list is
%% fetched from the ring. If the node is a singleton then the initial eager and
%% lazy sets are empty. If there are two nodes, each will be in the others
%% eager set and the lazy sets will be empty. When number of members is less
%% than 5, each node will initially have one other node in its eager set and
%% lazy set. If there are more than five nodes each node will have at most two
%% other nodes in its eager set and one in its lazy set, initially.
%% In addition, after the broadcast server is started, a callback is registered
%% with ring_events to generate membership updates as the ring changes.
%% @end
%% -----------------------------------------------------------------------------
-spec start_link() -> {ok, pid()} | ignore | {error, term()}.

start_link() ->
    LazyTickPeriod = partisan_config:get(
        lazy_tick_period, ?DEFAULT_LAZY_TICK_PERIOD
    ),
    ExchangeTickPeriod = partisan_config:get(
        exchange_tick_period, ?DEFAULT_EXCHANGE_TICK_PERIOD
    ),
    Opts = #{
        lazy_tick_period => LazyTickPeriod,
        exchange_tick_period => ExchangeTickPeriod
    },

    {ok, Members} = partisan_peer_service:members(),

    ?LOG_DEBUG("Peer sampling service members: ~p", [Members]),

    %% The peer service has already sampled the members, we start off
    %% with pure gossip (ie. all members are in the eager push list and lazy
    %% list is empty)
    InitEagers = Members,
    InitLazys = [],

    ?LOG_DEBUG("Init peers, eager: ~p, lazy: ~p", [InitEagers, InitLazys]),

    Mods = partisan_config:get(broadcast_mods, []),

    start_link(Members, InitEagers, InitLazys, Mods, Opts).


%% -----------------------------------------------------------------------------
%% @doc Starts the broadcast server on this node. `InitMembers' must be a list
%% of all members known to this node when starting the broadcast server.
%% `InitEagers' are the initial peers of this node for all broadcast trees.
%% `InitLazys' is a list of random peers not in `InitEagers' that will be used
%% as the initial lazy peer shared by all trees for this node. If the number
%% of nodes in the cluster is less than 3, `InitLazys' should be an empty list.
%% `InitEagers' and `InitLazys' must also be subsets of `InitMembers'. `Mods' is
%% a list of modules that may be handlers for broadcasted messages. All modules
%% in `Mods' should implement the `partisan_plumtree_broadcast_handler'
%% behaviour.
%% `Opts' is a proplist or map with the following possible options:
%%  <ul>
%%  <li> `lazy_tick_period :: non_neg_integer()' - Flush all outstanding lazy pushes period (in milliseconds)</li>
%%  <li> `exchange_tick_period :: non_neg_integer()' - Possibly perform an exchange period (in milliseconds)</li>
%% </ul>
%%
%% NOTE: When starting the server using start_link/2 no automatic membership update from
%% ring_events is registered. Use start_link/0.
%% @end
%% -----------------------------------------------------------------------------

-spec start_link(
    InitMembers :: [node()],
    InitEagers :: [node()],
    InitLazys :: [node()],
    Mods :: [module()],
    Opts :: proplists:proplist() | map()) ->
    {ok, pid()} | ignore | {error, term()}.

start_link(InitMembers, InitEagers, InitLazys, Mods, Opts) when is_list(Opts) ->
    start_link(InitMembers, InitEagers, InitLazys, Mods, maps:from_list(Opts));

start_link(InitMembers, InitEagers, InitLazys, Mods, Opts) when is_map(Opts) ->
    StartOpts = [
        {spawn_opt, ?PARALLEL_SIGNAL_OPTIMISATION([])}
    ],
    gen_server:start_link(
        {local, ?SERVER}, ?MODULE,
        [InitMembers, InitEagers, InitLazys, Mods, Opts],
        StartOpts
    ).


%% -----------------------------------------------------------------------------
%% @doc Broadcasts a message originating from this node. The message will be
%% delivered to each node at least once. The `Mod' passed is responsible for
%% handling the message on remote nodes as well as providing some other
%% information both locally and and on other nodes.
%% `Mod' must be loaded on all members of the clusters and implement the
%% `partisan_plumtree_broadcast_handler' behaviour.
%% @end
%% -----------------------------------------------------------------------------
-spec broadcast(any(), module()) -> ok.

broadcast(Broadcast, Mod) ->
    {MessageId, Payload} = Mod:broadcast_data(Broadcast),
    gen_server:cast(?SERVER, {broadcast, MessageId, Payload, Mod}).


%% -----------------------------------------------------------------------------
%% @doc Notifies broadcast server of membership update
%% @end
%% -----------------------------------------------------------------------------
-spec update([node()]) -> ok.

update(LocalState0) ->
    LocalState = partisan_peer_service:decode(LocalState0),
    gen_server:cast(?SERVER, {update, LocalState}).


%% -----------------------------------------------------------------------------
%% @doc Returns the broadcast servers view of full cluster membership.
%% Wait indefinitely for a response is returned from the process.
%% @end
%% -----------------------------------------------------------------------------
-spec broadcast_members() -> nodeset().

broadcast_members() ->
    broadcast_members(infinity).


%% -----------------------------------------------------------------------------
%% @doc Returns the broadcast servers view of full cluster membership.
%% Waits `Timeout' ms for a response from the server.
%% @end
%% -----------------------------------------------------------------------------
-spec broadcast_members(infinity | pos_integer()) -> nodeset().

broadcast_members(Timeout) ->
    gen_server:call(?SERVER, broadcast_members, Timeout).


%% -----------------------------------------------------------------------------
%% @doc return a list of exchanges, started by broadcast on thisnode, that are
%% running.
%% @end
%% -----------------------------------------------------------------------------
-spec exchanges() -> exchanges().

exchanges() ->
    exchanges(partisan:node()).


%% -----------------------------------------------------------------------------
%% @doc returns a list of exchanges, started by broadcast on `Node', that are
%% running.
%% @end
%% -----------------------------------------------------------------------------
-spec exchanges(node()) -> partisan_plumtree_broadcast:exchanges().

exchanges(Node) ->
    gen_server:call({?SERVER, Node}, exchanges, infinity).


%% -----------------------------------------------------------------------------
%% @doc cancel exchanges started by this node.
%% @end
%% -----------------------------------------------------------------------------
-spec cancel_exchanges(selector()) -> exchanges().

cancel_exchanges(WhichExchanges) ->
    gen_server:call(?SERVER, {cancel_exchanges, WhichExchanges}, infinity).



%% =============================================================================
%% DEBUG API
%% =============================================================================



%% @doc return the peers for `Node' for the tree rooted at `Root'.
%% Wait indefinitely for a response is returned from the process
-spec debug_get_peers(node(), node()) ->
    {nodeset(), nodeset()}.

debug_get_peers(Node, Root) ->
    debug_get_peers(Node, Root, infinity).


%% @doc return the peers for `Node' for the tree rooted at `Root'.
%% Waits `Timeout' ms for a response from the server
-spec debug_get_peers(node(), node(), infinity | pos_integer()) ->
    {nodeset(), nodeset()}.

debug_get_peers(Node, Root, Timeout) ->
    gen_server:call({?SERVER, Node}, {get_peers, Root}, Timeout).


%% @doc return peers for all `Nodes' for tree rooted at `Root'
%% Wait indefinitely for a response is returned from the process
-spec debug_get_tree(node(), [node()]) ->
    [{node(), {nodeset(), nodeset()}}].

debug_get_tree(Root, Nodes) ->
    [begin
         Peers = try debug_get_peers(Node, Root)
                 catch _:_ -> down
                 end,
         {Node, Peers}
     end || Node <- Nodes].



%% =============================================================================
%% GEN_SERVER CALLBACKS
%% =============================================================================



-spec init([[any()], ...]) -> {ok, state()}.

init([Members, InitEagers0, InitLazys0, Mods, Opts]) ->
    %% We subscribe to the membership change events
    partisan_peer_service:add_sup_callback(fun ?MODULE:update/1),

    LazyTickPeriod = maps:get(lazy_tick_period, Opts),
    ExchangeTickPeriod = maps:get(exchange_tick_period, Opts),
    schedule_lazy_tick(LazyTickPeriod),
    schedule_exchange_tick(ExchangeTickPeriod),

    State1 =  #state{
        mods = lists:usort(Mods),
        exchanges = [],
        lazy_tick_period = LazyTickPeriod,
        exchange_tick_period = ExchangeTickPeriod
    },

    AllMembers = ordsets:from_list(Members),
    InitEagers = ordsets:from_list(InitEagers0),
    InitLazys = ordsets:from_list(InitLazys0),

    State2 = reset_peers(AllMembers, InitEagers, InitLazys, State1),
    {ok, State2}.


-spec handle_call(term(), {pid(), term()}, state()) -> {reply, term(), state()}.

handle_call({get_peers, Root}, _From, State) ->
    EagerPeers = all_peers(
        Root, State#state.eager_sets, State#state.common_eagers
    ),
    LazyPeers = all_peers(
        Root, State#state.lazy_sets, State#state.common_lazys
    ),
    {reply, {EagerPeers, LazyPeers}, State};

{'fraction_non_alphanumeric': 0.24175356458821026, 'fraction_numerical': 0.002908420231254877, 'mean_word_length': 2.304617909048289, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 79, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4651478', 'n_tokens_mistral': 7511, 'n_tokens_neox': 7194, 'n_words': 2578}