<commit_before>%%%-------------------------------------------------------------------
%%% File:      untitled.erl
%%% @author    <NAME> <> []
%%% @copyright 2008 <NAME>
%%% @doc  Membership process keeps track of dynomite node membership.  Maintains a version history.
%%%
%%% @end  
%%%
%%% @since 2008-03-30 by <NAME>
%%%-------------------------------------------------------------------
-module(membership).
-author('<NAME>').

-behaviour(gen_server).
-define(VIRTUALNODES, 100).

%% API
-export([start_link/1, join_node/2, nodes_for_key/1, partitions/0, nodes/0, state/0, state/1, old_partitions/0, partitions_for_node/2, fire_gossip/1, partition_for_key/1, stop/0, range/1]).

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

-record(membership, {config, partitions, version, nodes, old_partitions}).

-include("config.hrl").

-define(power_2(N), (2 bsl (N-1))).

-ifdef(TEST).
-include("etest/membership_test.erl").
-endif.

%%====================================================================
%% API
%%====================================================================
%%--------------------------------------------------------------------
%% @spec start_link() -> {ok,Pid} | ignore | {error,Error}
%% @doc Starts the server
%% @end 
%%--------------------------------------------------------------------
start_link(Config) ->
  gen_server:start_link({local, membership}, ?MODULE, Config, []).

join_node(JoinTo, Me) ->
	gen_server:call({membership, JoinTo}, {join_node, Me}).
	
nodes_for_key(Key) ->
  gen_server:call(membership, {nodes_for_key, Key}).
  
nodes() ->
  gen_server:call(membership, nodes).
  
state() ->
  gen_server:call(membership, state).
  
state(State) ->
  gen_server:call(membership, {state, State}).
  
partitions() ->
  gen_server:call(membership, partitions).
  
partitions_for_node(Node, Option) ->
  gen_server:call(membership, {partitions_for_node, Node, Option}).
  
partition_for_key(Key) ->
  gen_server:call(membership, {partition_for_key, Key}).

range(Partition) ->
  gen_server:call(membership, {range, Partition}).

old_partitions() ->
  gen_server:call(membership, old_partitions).

stop() ->
  gen_server:call(membership, stop).
	
fire_gossip({A1, A2, A3}) ->
  random:seed(A1, A2, A3),
  State = state(),
  Config = State#membership.config,
  Nodes = lists:filter(fun(E) -> E /= node() end, membership:nodes()),
  ModState = if
    length(Nodes) > 0 ->
      Node = random_node(Nodes),
        % error_logger:info_msg("firing gossip at ~p~n", [Node]),
      RemoteState = gen_server:call({membership, Node}, {share, State}),
      merge_states(RemoteState, State);
    true -> State
  end,
  membership:state(ModState#membership{config=Config}),
  reload_storage_servers(State, ModState),
  save_state(ModState),
  timer:apply_after(random:uniform(5000) + 5000, membership, fire_gossip, [random:seed()]).

%%====================================================================
%% gen_server callbacks
%%====================================================================

%%--------------------------------------------------------------------
%% @spec init(Args) -> {ok, State} |
%%                         {ok, State, Timeout} |
%%                         ignore               |
%%                         {stop, Reason}
%% @doc Initiates the server
%% @end 
%%--------------------------------------------------------------------
init(ConfigIn) ->
  process_flag(trap_exit, true),
  Nodes = erlang:nodes(),
  {ok, State} = case load_state(ConfigIn) of
    {ok, Value} -> 
      error_logger:info_msg("loading membership from disk~n", []),
      configuration:set_config(Value#membership.config),
      {ok, Value};
    _ -> if
		  length(Nodes) > 0 -> 
  		  Node = random_node(Nodes),
  		  error_logger:info_msg("joining node ~p~n", [Node]),
  		  join_node(Node, node());
  		true -> {ok, create_initial_state(ConfigIn)}
  	end
  end,
	reload_storage_servers(empty, State),
  timer:apply_after(random:uniform(1000) + 1000, membership, fire_gossip, [random:seed()]),
	{ok, State#membership{config=configuration:get_config()}}.

%%--------------------------------------------------------------------
%% @spec 
%% handle_call(Request, From, State) -> {reply, Reply, State} |
%%                                      {reply, Reply, State, Timeout} |
%%                                      {noreply, State} |
%%                                      {noreply, State, Timeout} |
%%                                      {stop, Reason, Reply, State} |
%%                                      {stop, Reason, State}
%% @doc Handling call messages
%% @end 
%%--------------------------------------------------------------------

handle_call({join_node, Node}, {_, _From}, State = #membership{config=Config}) ->
  error_logger:info_msg("~p is joining the cluster.~n", [node(_From)]),
  NewState = int_join_node(Node, State),
  reload_storage_servers(State, NewState),
  save_state(NewState),
	{reply, {ok, NewState}, NewState#membership{config=Config}};
	
handle_call({share, NewState}, _From, State = #membership{config=Config}) ->
  % error_logger:info_msg("sharing state ~p ~p~n", [State#membership.config, NewState#membership.config]),
  case vector_clock:compare(State#membership.version, NewState#membership.version) of
    less -> 
      reload_storage_servers(State, NewState),
      save_state(NewState),
      {reply, NewState, NewState};
    greater -> {reply, State, State};
    equal -> {reply, State, State};
    concurrent -> 
      Merged = merge_states(NewState, State),
      reload_storage_servers(State, Merged),
      save_state(Merged),
      {reply, Merged, Merged#membership{config=State#membership.config}}
  end;
	
handle_call(nodes, _From, State = #membership{nodes=Nodes}) ->
  {reply, Nodes, State};
  
handle_call(state, _From, State) -> {reply, State, State};

handle_call({state, NewState}, _From, State = #membership{config=Config}) -> {reply, ok, NewState#membership{config=Config}};
	
handle_call(old_partitions, _From, State = #membership{old_partitions=P}) when is_list(P) ->
  {reply, P, State};
  
handle_call(old_partitions, _From, State) -> {reply, [], State};
	
handle_call(partitions, _From, State) -> {reply, State#membership.partitions, State};
	
handle_call({range, Partition}, _From, State) ->
  {reply, int_range(Partition, State#membership.config), State};
	
handle_call({nodes_for_key, Key}, _From, State) -><commit_msg>save state on force<commit_after>%%%-------------------------------------------------------------------
%%% File:      untitled.erl
%%% @author    <NAME> <> []
%%% @copyright 2008 <NAME>
%%% @doc  Membership process keeps track of dynomite node membership.  Maintains a version history.
%%%
%%% @end  
%%%
%%% @since 2008-03-30 by <NAME>
%%%-------------------------------------------------------------------
-module(membership).
-author('<NAME>').

-behaviour(gen_server).
-define(VIRTUALNODES, 100).

%% API
-export([start_link/1, join_node/2, nodes_for_key/1, partitions/0, nodes/0, state/0, state/1, old_partitions/0, partitions_for_node/2, fire_gossip/1, partition_for_key/1, stop/0, range/1]).

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

-record(membership, {config, partitions, version, nodes, old_partitions}).

-include("config.hrl").

-define(power_2(N), (2 bsl (N-1))).

-ifdef(TEST).
-include("etest/membership_test.erl").
-endif.

%%====================================================================
%% API
%%====================================================================
%%--------------------------------------------------------------------
%% @spec start_link() -> {ok,Pid} | ignore | {error,Error}
%% @doc Starts the server
%% @end 
%%--------------------------------------------------------------------
start_link(Config) ->
  gen_server:start_link({local, membership}, ?MODULE, Config, []).

join_node(JoinTo, Me) ->
	gen_server:call({membership, JoinTo}, {join_node, Me}).
	
nodes_for_key(Key) ->
  gen_server:call(membership, {nodes_for_key, Key}).
  
nodes() ->
  gen_server:call(membership, nodes).
  
state() ->
  gen_server:call(membership, state).
  
state(State) ->
  gen_server:call(membership, {state, State}).
  
partitions() ->
  gen_server:call(membership, partitions).
  
partitions_for_node(Node, Option) ->
  gen_server:call(membership, {partitions_for_node, Node, Option}).
  
partition_for_key(Key) ->
  gen_server:call(membership, {partition_for_key, Key}).

range(Partition) ->
  gen_server:call(membership, {range, Partition}).

old_partitions() ->
  gen_server:call(membership, old_partitions).

stop() ->
  gen_server:call(membership, stop).
	
fire_gossip({A1, A2, A3}) ->
  random:seed(A1, A2, A3),
  State = state(),
  Config = State#membership.config,
  Nodes = lists:filter(fun(E) -> E /= node() end, membership:nodes()),
  ModState = if
    length(Nodes) > 0 ->
      Node = random_node(Nodes),
        % error_logger:info_msg("firing gossip at ~p~n", [Node]),
      RemoteState = gen_server:call({membership, Node}, {share, State}),
      merge_states(RemoteState, State);
    true -> State
  end,
  membership:state(ModState#membership{config=Config}),
  reload_storage_servers(State, ModState),
  save_state(ModState),
  timer:apply_after(random:uniform(5000) + 5000, membership, fire_gossip, [random:seed()]).

%%====================================================================
%% gen_server callbacks
%%====================================================================

%%--------------------------------------------------------------------
%% @spec init(Args) -> {ok, State} |
%%                         {ok, State, Timeout} |
%%                         ignore               |
%%                         {stop, Reason}
%% @doc Initiates the server
%% @end 
%%--------------------------------------------------------------------
init(ConfigIn) ->
  process_flag(trap_exit, true),
  Nodes = erlang:nodes(),
  {ok, State} = case load_state(ConfigIn) of
    {ok, Value} -> 
      error_logger:info_msg("loading membership from disk~n", []),
      configuration:set_config(Value#membership.config),
      {ok, Value};
    _ -> if
		  length(Nodes) > 0 -> 
  		  Node = random_node(Nodes),
  		  error_logger:info_msg("joining node ~p~n", [Node]),
  		  join_node(Node, node());
  		true -> {ok, create_initial_state(ConfigIn)}
  	end
  end,
	reload_storage_servers(empty, State),
  timer:apply_after(random:uniform(1000) + 1000, membership, fire_gossip, [random:seed()]),
	{ok, State#membership{config=configuration:get_config()}}.

%%--------------------------------------------------------------------
%% @spec 
%% handle_call(Request, From, State) -> {reply, Reply, State} |
%%                                      {reply, Reply, State, Timeout} |
%%                                      {noreply, State} |
%%                                      {noreply, State, Timeout} |
%%                                      {stop, Reason, Reply, State} |
%%                                      {stop, Reason, State}
%% @doc Handling call messages
%% @end 
%%--------------------------------------------------------------------

handle_call({join_node, Node}, {_, _From}, State = #membership{config=Config}) ->
  error_logger:info_msg("~p is joining the cluster.~n", [node(_From)]),
  NewState = int_join_node(Node, State),
  reload_storage_servers(State, NewState),
  save_state(NewState),
	{reply, {ok, NewState}, NewState#membership{config=Config}};
	
handle_call({share, NewState}, _From, State = #membership{config=Config}) ->
  % error_logger:info_msg("sharing state ~p ~p~n", [State#membership.config, NewState#membership.config]),
  case vector_clock:compare(State#membership.version, NewState#membership.version) of
    less -> 
      reload_storage_servers(State, NewState),
      save_state(NewState),
      {reply, NewState, NewState};
    greater -> {reply, State, State};
    equal -> {reply, State, State};
    concurrent -> 
      Merged = merge_states(NewState, State),
      reload_storage_servers(State, Merged),
      save_state(Merged),
      {reply, Merged, Merged#membership{config=State#membership.config}}
  end;
	
handle_call(nodes, _From, State = #membership{nodes=Nodes}) ->
  {reply, Nodes, State};
  
handle_call(state, _From, State) -> {reply, State, State};

handle_call({state, NewState}, _From, State = #membership{config=Config}) -> 
  save_state(NewState#membership{config=Config}),
  {reply, ok, NewState#membership{config=Config}};
	
handle_call(old_partitions, _From, State = #membership{old_partitions=P}) when is_list(P) ->
  {reply, P, State};
  
handle_call(old_partitions, _From, State) -> {reply, [], State};
	
handle_call(partitions, _From, State) -> {reply, State#membership.partitions, State};
	
handle_call({range, Partition}, _From, State) ->
  {reply, int_range(Partition, State#membership.config), State};
	
handle_call({nodes_for_key, Key}, _From, State) ->
{'fraction_non_alphanumeric': 0.2690744574802546, 'fraction_numerical': 0.009355110804386167, 'mean_word_length': 2.7563364055299537, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 93, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5731833', 'n_tokens_mistral': 4087, 'n_tokens_neox': 3913, 'n_words': 756}