-module(canister_sync). -behaviour(gen_server). %% API -export([ start_link/0, send_update/2, send_update/3, send_clear/2, send_touch/2, get_nodes/0, get_node_to_resync/0 ]). %-export([up/0]). %% gen_server callbacks -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). -define(ETS_TABLE, canister_config). -define(SERVER, ?MODULE). -define(REMOTE_TIMEOUT, canister_config:remote_timeout()). -define(NODE_INTERVAL, canister_config:node_interval() * 1000). -record(state, {resync_timer}). start_link() -> gen_server:start_link({local, ?SERVER}, ?MODULE, [], []). get_nodes() -> case ets:lookup(?ETS_TABLE, nodes) of [] -> []; [{_, Nodes}] -> Nodes end. send_update(ID, Time) -> Data = canister:get_data(ID), send_update(ID, Data, Time). send_update(ID, Data, Time) -> gen_server:cast(?SERVER, {cast, {update, ID, Data, Time}}). send_clear(ID, Time) -> gen_server:cast(?SERVER, {cast, {clear, ID, Time}}). send_touch(ID, Time) -> gen_server:cast(?SERVER, {cast, {touch, ID, Time}}). init([]) -> ok = init_ets(), auto_connect_nodes(), timer:send_after(1, ?SERVER, refresh_nodes), timer:send_interval(?NODE_INTERVAL, ?SERVER, refresh_nodes), net_kernel:monitor_nodes(true), {ok, #state{}}. init_ets() -> ETSConfig = [ named_table, {read_concurrency, true}, {write_concurrency, false}, public ], try ets:new(?ETS_TABLE, ETSConfig) of _ -> ok catch error:badarg -> case lists:member(?ETS_TABLE, ets:all()) of false -> {error, could_not_init_ets_for_canister}; true -> Owner = proplists:get_value(owner, ets:info(?ETS_TABLE)), case Owner==self() of true -> ok; false -> {error, {count_not_init_ets_for_canister, table_exists_with_different_owner}} end end end. handle_call(uptime, _From, State) -> {Time, _} = erlang:statistics(wall_clock), {reply, Time, State}; handle_call(are_you_there, _From, State) -> {reply, yes, State}; handle_call({remote, Msg}, _From, State) -> Res = handle_remote(Msg), {reply, Res, State}; handle_call({last_access_time, ID}, _From, State) -> Val = canister:last_access_time(ID), {reply, {ok, Val}, State}; handle_call({last_update_time, ID}, _From, State) -> Val = canister:last_update_time(ID), {reply, {ok, Val}, State}; handle_call({deleted_time, ID}, _From, State) -> Val = canister:deleted_time(ID), {reply, {ok, Val}, State}; handle_call({get_local, ID, Key}, _From, State) -> Val = canister:get_local(ID, Key), {reply, Val, State}; handle_call(_Request, _From, State) -> Reply = ok, {reply, Reply, State}. handle_cast({update_nodes, NewNodes}, State = #state{resync_timer=OldResyncTimer}) -> OldNodes = get_nodes(), true = ets:insert(?ETS_TABLE, {nodes, NewNodes}), canister_log:info("Nodes Updated: ~p => ~p",[OldNodes, NewNodes]), case NewNodes -- OldNodes of [] -> canister_log:info("No resync necessary. Node change was only from nodes going offline."), {noreply, State}; NewlyUp -> canister_log:info("New node(s) were added (~p), scheduling a full resync in about 5 seconds", [NewlyUp]), canister_log:info("Canceling previous Resync Timer: ~p",[OldResyncTimer]), timer:cancel(OldResyncTimer), {ok, NewResyncTimer} = timer:send_after(5000 + rand:uniform(2000), full_resync), NewState = State#state{resync_timer=NewResyncTimer}, {noreply, NewState} end; handle_cast({cast, Msg}, State = #state{}) -> erlang:spawn(fun() -> Nodes = get_nodes(), cast_to_nodes(Nodes, Msg) end), {noreply, State}; handle_cast(_Msg, State) -> {noreply, State}. handle_info({nodeup, _Node}, State) -> handle_info(refresh_nodes, State); handle_info({nodedown, _Node}, State) -> handle_info(refresh_nodes, State); handle_info(refresh_nodes, State = #state{}) -> erlang:spawn(fun() -> Nodes = get_nodes(), refresh_nodes(Nodes) end), {noreply, State}; handle_info(full_resync, State = #state{}) -> case get_nodes() of [] -> do_nothing; Nodes -> erlang:spawn(fun() -> List = lists:sort([node() | Nodes]), assemble_and_requeue(List) end) end, NewState = State#state{resync_timer=undefined}, {noreply, NewState}; handle_info(_, State) -> {noreply, State}. terminate(_Reason, _State) -> ok. code_change(_OldVsn, State, _Extra) -> {ok, State}. %%%=================================================================== %%% Internal functions %%%=================================================================== handle_remote({update, ID, Data, Time}) -> canister:update(ID, Data, Time), ok; handle_remote({touch, ID, Time}) -> canister:touch_local(ID, Time), ok; handle_remote({clear, ID, Time}) -> canister:clear(ID, Time), ok. assemble_and_requeue(Nodes) -> case get_node_to_resync(Nodes) of undefined -> ok; MainNode -> canister_log:info("Will resync all on ~p",[MainNode]), IDSet = lists:foldl(fun(Node, Acc) -> try erpc:call(Node, canister, all_sessions, [], ?REMOTE_TIMEOUT) of NewIDs -> sets:union(Acc, sets:from_list(NewIDs)) catch _:_ -> Acc end end, sets:new(), Nodes), IDs = sets:to_list(IDSet), canister_resync:add_many(MainNode, IDs) end. get_node_to_resync() -> get_node_to_resync([node() | get_nodes()]). get_node_to_resync(Nodes) -> case which_nodes_are_resyncing(Nodes) of [] -> hd(Nodes); NodeNums -> canister_log:info("Eligible Servers to Resync~nAll: ~p~nFound: ~p",[Nodes, NodeNums]), {FoundNode, _FoundNum} = lists:foldl(fun({Node, Num}, {BestNode, BestNum}) -> case Num > BestNum of true -> {Node, Num}; false -> {BestNode, BestNum} end end, {node(), 0}, NodeNums), FoundNode end. cast_to_nodes(Nodes, Msg) -> ec_plists:foreach(fun(Node) -> Server = {?SERVER, Node}, case gen_server:call(Server, {remote, Msg}, ?REMOTE_TIMEOUT) of ok -> ok; Other -> canister_log:info("Failed to send ~p to ~p~nResult: ~p",[Msg, Node, Other]) end end, Nodes). refresh_nodes(OrigNodes) -> NewNodes = ec_plists:filter(fun(Node) -> is_node_up(Node) andalso is_node_canister_responding(Node) end, nodes()), Sorted = lists:sort(NewNodes), case Sorted==OrigNodes of true -> ok; false -> gen_server:cast(?SERVER, {update_nodes, Sorted}) end. which_nodes_are_resyncing(Nodes) -> lists:filtermap(fun(Node) -> case {canister_resync:is_resyncing(Node), canister_resync:num_queued(Node)} of {false, _} -> false; {true, 0} -> false; {true, Num} -> {true, {Node, Num}} end end, Nodes). is_node_up(Node) -> net_adm:ping(Node)==pong. is_node_canister_responding(Node) -> Server = {?SERVER, Node}, gen_server:call(Server, are_you_there, 1000) == yes. auto_connect_nodes() -> Nodes = canister_config:default_cluster(), auto_connect_nodes(Nodes). auto_connect_nodes([]) -> []; auto_connect_nodes([H|T]) -> case H==node() of true -> canister_log:info("Node (~p) is self(), so no need to connect", [H]); false -> case net_kernel:connect_node(H) of true -> canister_log:info("Node (~p) successfully connected", [H]); false -> canister_log:info("Node (~p) failed to connect", [H]) end end, auto_connect_nodes(T).