dht_join.erl

-module(dht_join).
-export([join/1, join/2, get_server/1, return_server/0]).

-define(SERVER_MODULE, dht_server).


join(InitNode) ->
    join(InitNode, whereis(?SERVER_MODULE)).

join(InitNode, NewServer) ->
    case get_server(InitNode) of
        {ok, InitServer} ->
            gen_server:call(InitServer, {join, NewServer});
        {error, Reason} ->
            io:format("** join error: ~p~n", [Reason]),
            {error, Reason}
    end.


get_server(Node) ->
    get_server(Node, net_adm:ping(Node)).

get_server(pong, Node) ->
    Server = rpc:call(Node, dht_join, return_server, []),
    {ok, Server};
get_server(pang, Node) ->
    {error, {pang, Node}}.



%%====================================================================
%% called by other nodes
%%====================================================================

return_server() ->
    whereis(?SERVER_MODULE).

dht_server.erl

-module(dht_server).
-behaviour(gen_server).

-export([start/1, start/2, lookup/1, lookup/2]).
-export([init/1, handle_call/3, handle_cast/2,
         terminate/2, handle_info/2, code_change/3]).


-record(state, {
        hash,
        mvector,
        pairs_table,
        replica_table,
        b_neighbor,    %% bigger hash neighbor list
        s_neighbor    %% smaller hash neighbor list
    }).

-define(MAX_LEVEL, 8).
-define(SERVER_MODULE, dht_server).
-define(TIMEOUT, 1000).
-define(DEBUG_OPTION, trace).



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

start(InitNode) ->
    gen_server:start_link({local, ?MODULE}, ?MODULE, [InitNode], [{timeout, ?TIMEOUT}]).

start(InitNode, 'debugging-mode') ->
    gen_server:start_link({local, ?MODULE}, ?MODULE, [InitNode], [{timeout, ?TIMEOUT}, {debug, ?DEBUG_OPTION}]).


new_table() ->
    Return = self(),
    Ref  = make_ref(),

    spawn(fun() ->
                ets:new(pairs_table, [set, named_table]),
                ets:new(replica_table, [set, named_table]),
                Return ! {Ref, {done, new_table}},
                timer:sleep(infinity)
        end),

    receive
        {Ref, {done, new_table}} ->
            {whereis(pairs_table), whereis(replica_table)}
    end.


new_state() ->
    {A1, A2, A3} = now(),
    random:seed(A1, A2, A3),

    {PairsTable, ReplicaTable} = new_table(),
    #state{
        hash          = erlang:phash2(random:uniform()),
        mvector       = util_mvector:make_mvector(),
        pairs_table   = PairsTable,
        replica_table = ReplicaTable,
        b_neighbor    = lists:duplicate(?MAX_LEVEL, '__none__'),
        s_neighbor    = lists:duplicate(?MAX_LEVEL, '__none__')
    }.


init([InitNode]) ->
    dht_join:join(InitNode, self()),
    {ok, new_state()}.


handle_call(Any, From, State) ->
    io:format("Received Unknown Message: ~p~n", [Any]),
    {ok, {error, unknown_message}, State}.


handle_cast(_, State) ->
    {ok, State}.


handle_info(_, State) ->
    {noreply, State}.


code_change(_, State, _) ->
    {ok, State}.


terminate(_Reason, State) ->
    ok.



%%====================================================================
%% interfaces
%%====================================================================

lookup(Hash) ->
    lookup('hash', Hash).

lookup('key', Key) ->
    gen_server:call(?MODULE, {lookup_with_key, Key, erlang:phash2(Key)});
lookup('hash', Hash) ->
    gen_server:call(?MODULE, {lookup, Hash}).

util_mvector.erl

-module(util_mvector).
-export([is_equal/2, is_equal/3, make_mvector/0]).

-define(MAX_LEVEL, 8).
%-define(MAX_LEVEL, 16).
%-define(MAX_LEVEL, 32).
%-define(MAX_LEVEL, 64).
%-define(MAX_LEVEL, 128).


is_equal(MVector1, MVector2) ->
    MVector1 == MVector2.

is_equal(MVector1, MVector2, Level) ->
    Tail = ?MAX_LEVEL - Level - 1,
    <<_:Level, B1:1, _:Tail>> = MVector1,
    <<_:Level, B2:1, _:Tail>> = MVector2,

    B1 == B2.


make_mvector() ->
    {A1, A2, A3} = now(),
    random:seed(A1, A2, A3),

    FirstEightBit = lsb_zero(random:uniform(256) - 1 rem 2),
    make_mvector(FirstEightBit, ?MAX_LEVEL / 8).

make_mvector(MVector, 0.0) ->
    MVector;
make_mvector(MVector, N) ->
    make_mvector(<<(random:uniform(256) - 1):8, MVector/binary>>, N - 1).


%% 最下位ビットを0に変換する
lsb_to_zero(0) ->
    make_mvector(<<N>>, ?MAX_LEVEL / 8 - 1);
lsb_to_zero(1) ->
    M = N - 1,
    make_mvector(<<M>>, ?MAX_LEVEL / 8 - 1).