%% ------------------------------------------------------------------- %% %% Copyright (c) 2017 Christopher S. Meiklejohn. 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. %% %% ------------------------------------------------------------------- %% ----------------------------------------------------------------------------- %% @doc This modules implements a server that realises the %% {@link partisan_plumtree_broadcast_handler} behaviour in order to diseminate %% heartbeat messages. Partisan uses these heartbeat messages to stimulate the %% Epidemic Broadcast Tree construction. %% %% The server will schedule the sending of a `hearbeat` message periodically %% using the `broadcast_heartbeat_interval` option. %% %% Notice that this handler does not perform AAE Exchanges, as we will always %% have a periodic heartbeat. For that reason, the implementation %% of the {@link partisan_plumtree_broadcast_handler:exchange/1} %% callback always returns `ignore'. %% %% @end %% ----------------------------------------------------------------------------- -module(partisan_plumtree_backend). -author("Christopher S. Meiklejohn "). -behaviour(gen_server). -behaviour(partisan_plumtree_broadcast_handler). -include("partisan.hrl"). -include("partisan_logger.hrl"). -record(state, {}). -type state() :: #state{}. -record(broadcast, { timestamp :: timestamp() }). -type broadcast_message() :: #broadcast{}. -type timestamp() :: {node(), non_neg_integer()}. -type broadcast_id() :: timestamp(). -type broadcast_payload() :: timestamp(). %% API -export([start_link/0]). -export([start_link/1]). %% transmission callbacks -export([extract_log_type_and_payload/1]). %% partisan_plumtree_broadcast_handler callbacks -export([broadcast_channel/0]). -export([broadcast_data/1]). -export([exchange/1]). -export([graft/1]). -export([is_stale/1]). -export([merge/2]). %% gen_server callbacks -export([init/1]). -export([handle_call/3]). -export([handle_cast/2]). -export([handle_info/2]). -export([terminate/2]). -export([code_change/3]). %% ============================================================================= %% API %% ============================================================================= %% ----------------------------------------------------------------------------- %% @doc Same as start_link([]). %% @end %% ----------------------------------------------------------------------------- -spec start_link() -> {ok, pid()} | ignore | {error, term()}. start_link() -> start_link([]). %% ----------------------------------------------------------------------------- %% @doc Start and link to calling process. %% @end %% ----------------------------------------------------------------------------- -spec start_link(list())-> {ok, pid()} | ignore | {error, term()}. start_link(Opts) -> gen_server:start_link({local, ?MODULE}, ?MODULE, Opts, []). %% ----------------------------------------------------------------------------- %% @doc %% @end %% ----------------------------------------------------------------------------- extract_log_type_and_payload({prune, Root, From}) -> [{broadcast_protocol, {Root, From}}]; extract_log_type_and_payload({ignored_i_have, MessageId, _Mod, Round, Root, From}) -> [{broadcast_protocol, {MessageId, Round, Root, From}}]; extract_log_type_and_payload({graft, MessageId, _Mod, Round, Root, From}) -> [{broadcast_protocol, {MessageId, Round, Root, From}}]; extract_log_type_and_payload( {broadcast, MessageId, Timestamp, _Mod, Round, Root, From}) -> [{broadcast_protocol, {Timestamp, MessageId, Round, Root, From}}]; extract_log_type_and_payload({i_have, MessageId, _Mod, Round, Root, From}) -> [{broadcast_protocol, {MessageId, Round, Root, From}}]; extract_log_type_and_payload(Message) -> ?LOG_INFO("No match for extracted payload: ~p", [Message]), []. %% ============================================================================= %% PARTISAN_PLUMTREE_BROADCAST_HANDLER CALLBACKS %% ============================================================================= %% ----------------------------------------------------------------------------- %% @doc Returns the channel to be used when broadcasting a message %% on behalf of this handler. %% @end %% ----------------------------------------------------------------------------- -spec broadcast_channel() -> partisan:channel(). broadcast_channel() -> ?MEMBERSHIP_CHANNEL. %% ----------------------------------------------------------------------------- %% @doc Returns from the broadcast message the identifier and the payload. %% In this case a tuple where both arguments have the broadcast message %% `timestamp'. These messages are used by Partisan as a stimulus for the %% Epidemic Broadcast Tree (Plumtree) construction. %% @end %% ----------------------------------------------------------------------------- -spec broadcast_data(broadcast_message()) -> {broadcast_id(), broadcast_payload()}. broadcast_data(#broadcast{timestamp = Timestamp}) -> {Timestamp, Timestamp}. %% ----------------------------------------------------------------------------- %% @doc Perform a merge of an incoming object with an object in the %% local datastore. %% @end %% ----------------------------------------------------------------------------- -spec merge(broadcast_id(), broadcast_payload()) -> boolean(). merge(Timestamp, Timestamp) -> ?LOG_DEBUG("Heartbeat received: ~p", [Timestamp]), case is_stale(Timestamp) of true -> false; false -> gen_server:call(?MODULE, {merge, Timestamp}, infinity), true end. %% ----------------------------------------------------------------------------- %% @doc Use the clock on the object to determine if this message is %% stale or not. %% @end %% ----------------------------------------------------------------------------- -spec is_stale(broadcast_id()) -> boolean(). is_stale(Timestamp) -> gen_server:call(?MODULE, {is_stale, Timestamp}, infinity). %% ----------------------------------------------------------------------------- %% @doc Given a message identifier and a clock, return a given message. %% @end %% ----------------------------------------------------------------------------- -spec graft(broadcast_id()) -> stale | {ok, broadcast_payload()} | {error, term()}. graft(Timestamp) -> gen_server:call(?MODULE, {graft, Timestamp}, infinity). %% ----------------------------------------------------------------------------- %% @doc Returns `ignore`. %% This is because we don't need to worry about reliable delivery: we always %% know we'll have another heartbeat message to further repair during the next %% interval. %% @end %% ----------------------------------------------------------------------------- -spec exchange(node()) -> {ok, pid()} | {error, any()} | ignore. exchange(_Peer) -> ignore. %% ============================================================================= %% GEN_SERVER CALLBACKS %% ============================================================================= -spec init([]) -> {ok, state()}. init([]) -> %% Seed the random number generator. partisan_config:seed(), schedule_heartbeat(), %% Open an ETS table for tracking heartbeat messages. ets:new(?MODULE, [named_table]), {ok, #state{}}. -spec handle_call(term(), {pid(), term()}, state()) -> {reply, term(), state()}. handle_call({is_stale, Timestamp}, _From, State) -> Result = case ets:lookup(?MODULE, Timestamp) of [] -> false; _ -> true end, {reply, Result, State}; handle_call({graft, Timestamp}, _From, State) -> Result = case ets:lookup(?MODULE, Timestamp) of [] -> ?LOG_DEBUG(#{ description => "Heartbeat message not found for graft.", timestamp => Timestamp }), {error, {not_found, Timestamp}}; [{Timestamp, _}] -> {ok, Timestamp} end, {reply, Result, State}; handle_call({merge, Timestamp}, _From, State) -> true = ets:insert(?MODULE, [{Timestamp, true}]), {reply, ok, State}; handle_call(Event, _From, State) -> ?LOG_WARNING(#{description => "Unhandled call event", event => Event}), {reply, ok, State}. -spec handle_cast(term(), state()) -> {noreply, state()}. handle_cast(Event, State) -> ?LOG_WARNING(#{description => "Unhandled cast event", event => Event}), {noreply, State}. handle_info(heartbeat, State) -> %% Generate message with monotonically increasing integer. Counter = erlang:unique_integer([monotonic, positive]), %% Make sure the node prefixes the timestamp with it's own %% identifier: this means that we can have this tree %% participate in multiple trees, each rooted at a different %% node. Timestamp = {partisan:node(), Counter}, %% Insert a new message into the table. true = ets:insert(?MODULE, [{Timestamp, true}]), %% Send message with monotonically increasing integer. ok = partisan_plumtree_broadcast:broadcast( #broadcast{timestamp = Timestamp}, ?MODULE ), ?LOG_DEBUG( "Heartbeat triggered: sending ping ~p to ensure tree.", [Timestamp] ), %% Schedule report. schedule_heartbeat(), {noreply, State}; handle_info(Event, State) -> ?LOG_WARNING(#{description => "Unhandled info event", event => Event}), {noreply, State}. -spec terminate(term(), state()) -> term(). terminate(_Reason, _State) -> ok. -spec code_change(term() | {down, term()}, state(), term()) -> {ok, state()}. code_change(_OldVsn, State, _Extra) -> {ok, State}. %% ============================================================================= %% PRIVATE %% ============================================================================= %% @private schedule_heartbeat() -> case partisan_config:get(broadcast, false) of true -> Interval = partisan_config:get(broadcast_heartbeat_interval, 10000), timer:send_after(Interval, heartbeat); false -> ok end.