partisan_plumtree_broadcast (partisan v6.0.0)

View Source

The process that runs one broadcast group — the supervised owner of a single epidemic-broadcast context: its own mailbox, spanning-tree state and outstanding-lazy table. Group identity is the handler module, so each broadcast handler runs in its own group and independent gossip streams never share a tree or a mailbox. The public entry points for broadcasting and managing groups are in partisan_broadcast.

What the shell owns, and what it delegates

Tree construction and repair are delegated to a pluggable tree engine (partisan_broadcast_engine); the shell owns everything else — the gen_server process and mailbox, the membership poll, handler dispatch, the message transport, the periodic ticks, and the rolling-upgrade router. The engine is pure and returns actions the shell executes, so an alternate engine drops in with no change here.

Two engines exist: Plumtree (partisan_plumtree_engine, the default) and Thicket (partisan_thicket_engine). The shell drives an engine in one of two dispatch modes, read from the engine's dispatch_mode/0:

  • typed — the shell decodes each inbound wire message and calls a Plumtree-shaped callback, passing the handler's verdicts (is this novel? is this i_have stale?). This is the default path.
  • raw — the shell hands the engine whole wire messages via handle_message/2 and executes its deliver/fetch actions against the group's single handler. This is how a self-describing engine such as Thicket runs.

Non-blocking delivery

A handler may keep its heavy apply off the tree process: claim/2 runs on the tree process and does only the fast, atomic novelty check, while handle_broadcast/2 runs the apply in the handler's own process, delivered asynchronously. A slow apply then never blocks the tree or the other handlers. A handler that implements only merge/2 keeps the synchronous path.

Membership

The shell reads membership from a lock-free snapshot (partisan_membership), polling its version on each tick and reconciling the engine's peer sets when it changes, rather than subscribing to a synchronous event bus that would block the membership oracle.

Rolling upgrade

Partisan's own control-plane group keeps a fixed registered name, so membership keeps converging in a mixed-version cluster, and a compatibility router forwards legacy-addressed messages for application handlers to their per-group process.

Reading guide

start_link/2 and init/1 build the group and select the engine; the handle_cast clauses dispatch inbound wire messages to the engine (typed or raw); handle_info drives the lazy/repair and exchange ticks; run_engine/2 runs an engine callback and executes the actions it returns. Broadcasting and the debug queries are near the top.

Summary

Types

exchange()

-type exchange() :: {module(), node(), reference(), pid()}.

exchanges()

-type exchanges() :: [exchange()].

info_opt()

-type info_opt() :: node_spec | metadata | distance.

nodeset()

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

selector()

-type selector() :: all | {peer, node()} | {mod, module()} | reference() | pid().

state()

-type state() ::
          #state{name :: atom(),
                 node :: node(),
                 mods :: [module()],
                 exchanges :: exchanges(),
                 members_version :: non_neg_integer(),
                 lazy_tick_period :: non_neg_integer(),
                 exchange_tick_period :: non_neg_integer(),
                 engine :: module(),
                 engine_state :: partisan_broadcast_engine:state(),
                 engine_mode :: typed | raw,
                 handler_mod :: module() | undefined,
                 channel :: partisan:channel() | undefined}.

Functions

broadcast(Broadcast, Mod)

-spec broadcast(any(), module()) -> ok.

broadcast_channel(Mod)

-spec broadcast_channel(Mod :: module()) -> partisan:channel().

broadcast_members()

-spec broadcast_members() -> nodeset().

broadcast_members(Timeout)

-spec broadcast_members(infinity | pos_integer()) -> nodeset().

cancel_exchanges(Selector)

-spec cancel_exchanges(selector()) -> exchanges().

code_change(OldVsn, State, Extra)

-spec code_change(term() | {down, term()}, state(), term()) -> {ok, state()}.

debug_get_peers(Node, Root)

-spec debug_get_peers(node(), node()) -> {nodeset(), nodeset()} | no_return().

debug_get_peers(Node, Root, Timeout)

-spec debug_get_peers(node(), node(), infinity | pos_integer()) -> {nodeset(), nodeset()} | no_return().

debug_get_peers(Node, Root, Opts, Timeout)

-spec debug_get_peers(node(), node(), [info_opt()], infinity | pos_integer()) ->
                         {nodeset(), nodeset(), Info :: map()} | no_return().

debug_get_tree(Root, Nodes)

-spec debug_get_tree(node(), [node()]) -> [{node(), {nodeset(), nodeset()} | down}].

debug_get_tree(Root, Nodes, Timeout)

-spec debug_get_tree(node(), [node()], timeout()) -> [{node(), {nodeset(), nodeset()} | down}].

debug_get_tree(Root, Nodes, Opts, Timeout)

-spec debug_get_tree(node(), [node()], [info_opt()], timeout()) ->
                        [{node(), {nodeset(), nodeset(), Info :: map()} | down}].

exchanges()

-spec exchanges() -> {ok, exchanges()}.

exchanges(Node)

-spec exchanges(node()) -> {ok, exchanges()} | {error, {badrpc, Reason :: any()}}.

exchanges(Node, Timeout)

-spec exchanges(node(), timeout()) -> {ok, exchanges()} | {error, {badrpc, Reason :: any()}}.

get_eager_peers(Root)

-spec get_eager_peers(Root :: node()) -> list().

get_lazy_peers(Root)

-spec get_lazy_peers(Root :: node()) -> list().

get_peers(Root)

-spec get_peers(Root :: node()) -> list().

get_peers(Root, Opts)

-spec get_peers(Root :: node(), Opts :: [partisan:info_opt()]) -> list().

group_channel(Name)

-spec group_channel(Name :: atom()) -> partisan:channel() | undefined.

group_name(Mod)

-spec group_name(module()) -> atom().

handle_call/3

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

handle_cast/2

-spec handle_cast(term(), state()) -> {noreply, state()}.

handle_info/2

-spec handle_info(exchange_tick | lazy_tick | {'DOWN', _, process, _, _}, state()) -> {noreply, state()}.

init/1

-spec init(list()) -> {ok, state()}.

start_link()

-spec start_link() -> {ok, pid()} | ignore | {error, term()}.

start_link(Name, Opts)

-spec start_link(Name :: atom(), Opts :: map()) -> {ok, pid()} | ignore | {error, term()}.

terminate(Reason, State)

-spec terminate(term(), state()) -> term().