root/lang/erlang/qnesia.erl @ 27118

-module(qnesia).
-author('kuenishi@gmail.com').
%%
%% qnesia
%% @brief  Mnesia wrapper with Quorum-based replication
%% @author $Author$
%%  $Id$
%%

-record( data, {key, value}).

-export([ start/1, stop/1,
         write/2, read/1 ]).

-export([
         mnesia_wrapper/0, coordinator/4,
         start_test/0
        ]).

% master process
% erl -sname a@localhost -s qnesia start_test
start_test()->
    qnesia:start([node(), 'b@localhost']).

% write test :  n> qnesia:write_test(). 
% @spec   write({Key, Value}) -> ok | ng
write(Key, Value)->
    Data = #data{key=Key, value=Value},
    coordinator ! {write, Data , self() },
    receive
        Result->
            io:format("write test: ~p~n", [Result]) %とりあえず失敗する
    end.

% @spec   read(Key) -> DataList
read(Key)->
    coordinator ! {read, Key, self()},
    receive
        {ok, DataList}->
            io:format("read test: ~p~n", [DataList]), %とりあえず失敗する
            DataList;
        Other -> Other
    end.

start([])->    start( [node()] );
start(Nodes)-> 
    N = erlang:length( Nodes ),
    % start coordinator 
    Pid = spawn_link( ?MODULE, coordinator, [ N, N, N, Nodes ] ),
    register( coordinator, Pid ),
    % start mnesia_wrapper
    spawn( ?MODULE, mnesia_wrapper, [] ),
    Pid.

stop(Pid) ->
    Pid ! { stop, hand_command, self() }.

%% @breif coordinator process dedicated to 1 transaction
%% @spec
coordinator(N, R, W, Members)->
    receive 
        {stop, _Reason, _From} -> 
            slave ! {self(), stop}; % send stop command to all members here.
        {Req,  Data, From}->
            ok = send_tx( Req, Data, Members ),         
            case Req of
                write->
                    Result = recv_ack( N, W, N-W+1 ),
                    From ! Result;
                read->
                    Result = recv_data( N, R, N-R+1, [] ),
                    From ! Result
            end,
            coordinator( N, R, W, Members )
    end.


send_tx(_Req, _Data, [])->      ok;
send_tx( Req,  Data, [Receiver|Members])->
    {slave, Receiver} ! {self(), {Req, Data}},
    send_tx(Req, Data, Members).

recv_ack(_N,_Q, 0)->      {error, quorum_nodes_failed};
recv_ack(_N, 0, Remain)-> {ok, Remain};
recv_ack( N, Q, Remain)->
    receive
        {ok, Result}->
            io:format( "ok: ~p.~n", [Result]),
            recv_ack(N, Q-1, Remain);
        {error, Reason}->
            io:format( "failed: ~p.~n", [Reason]),
            recv_ack(N, Q, Remain-1)
%    after T->
%           {error, timeout}
    end.

recv_data(_N,_Q, 0,      DataList)-> {error, DataList};
recv_data(_N, 0,_Remain, DataList)-> {ok, DataList};
recv_data( N, Q, Remain, DataList)->
    receive
        {ok, Data}->
            io:format( "recv data: ~p.~n", [Data]),
            recv_data(N, Q-1, Remain, [Data|DataList]);
        {error, Reason} ->
            io:format( "failed: ~p.~n", [Reason]),
            recv_data(N, Q, Remain-1, DataList)
    end.

%% mnesia wrapper
%erl -sname b@localhost  -s qnesia mnesia_wrapper  -noshell -s init stop
mnesia_wrapper()->
    ok=mnesia:create_schema([node()]),
    ok=mnesia:start(),
    {atomic,ok}=mnesia:create_table( data, [
                                {disc_copies, [node()]},
                                {attributes, record_info(fields, data)}
                                ]),
    true=register(slave, self()),%, slave),
    mnesia_wrapper_loop(),
    mnesia:stop().

mnesia_wrapper_loop()->
    io:format("entering ~p loop.~n", [?MODULE]),
    receive 
        {From, stop} ->
            io:format("stopping ~p loop.~n", [?MODULE]),
            From ! {ok, stopped};
        {From, {write, Data}}->
            case mnesia:transaction(% Fun, Argv, 3 ) of
                   fun()->
                           mnesia:write(Data)
                   end) of
                {atomic, ok}->
                    io:format("ok, atomic ~p.~n", [Data]),
                    From ! {ok, atomic};
                {aborted, Reason} ->
                    io:format("error ~p.~n", [Reason]),
                    From ! {error, Reason}
            end,
            mnesia_wrapper_loop();
        {From, {read, Key}}-> % when read, 
            case mnesia:dirty_read( data, Key ) of
                []->
                    io:format("error: no data for ~p.~n", [Key]),
                    From ! {error, no_data};
                [E]->
                    io:format("ok, atomic ~p.~n", [E]),
                    From ! {ok, E};
                Other->
                    io:format("error: too many data for ~p: ~p.~n", [Key, Other]),
                    From ! {error, too_many_data}
            end,
            mnesia_wrapper_loop();
%       {From, be_master}->
        {From, _Other} ->
            io:format("error... illegal_message ~p.~n", [_Other]),
            From ! {error, illegal_message},
            mnesia_wrapper_loop()
    end.