Changeset 27802 for lang/erlang

Timestamp:

01/03/09 04:31:13 (4 years ago)

Author:

kuenishi

Message:

lang/erlang/qnesia make almost done...

Files:

• lang/erlang/qnesia/src/qnesia.erl (modified) (8 diffs)

lang/erlang/qnesia/src/qnesia.erl

@@ -27 +27 @@
         ]).
 
-% master process
-% erl -sname a@localhost -s qnesia start_test
-start_test()->
-    qnesia:start([node(), 'b@localhost']).
 
 write(RemoteNode, {Key, Value})->
@@ -60 +56 @@
     end.
 
+% master process
+% erl -sname a@localhost -s qnesia start_test
+start_test()->
+    qnesia:start([node(), 'b@localhost']).
+
 start([])->    start( [node()] );
+
 start(Nodes)-> 
     N = erlang:length( Nodes ),
@@ -67 +69 @@
     register( coordinator, Pid ),
     % start mnesia_wrapper
-    spawn( ?MODULE, mnesia_wrapper, [] ),
+    spawn( ?MODULE, replica, [] ),
     Pid.
 
@@ -80 +82 @@
                 case Req of
                     write->
-                        Result = recv_ack( N, W, N-W+1 ),
+                        Result = recv_ack( W, N-W+1 ),
                         From ! Result;
                     read->
@@ -95 +97 @@
     end.
 
-
+%% @breif  データを書き込んだ結果をFromに返信する。トランザクションのコーディネーター。
+%% @spec   tx_coordinator(N, W, Members, Data, From )-> {ok, transaction_done} | {error, Reason }    
+tx_coordinator(N, W, Members, Data, From)->
+    ok = send_tx( write, Data, Members ),               
+    case recv_ack( W, N-W+1 ) of
+        {ok, Remain}->
+            ok = send_tx( ok, do_commit, Members),
+            From ! {self(), {ok, transaction_done}},
+            ok = recv_ack( N+Remain, 1 ),
+            {ok, transaction_done};
+        {error, quorum_nodes_failed}->
+            From ! {self(), {error, quorum_nodes_failed}},
+            {error, quorum_nodes_failed};
+        {error, Other} ->
+            io:format("unknown error: ~p~n", [Other]),
+            From ! {self(), {error, Other}},
+            {error, Other}
+    end.
+
+%% @breif Membersのノード(のslaveというプロセス)全てにDataとリクエストを投げつける。 
+%%        {Req, Data} = {write, Data}|{ok,do_commit}
+%% @spec  send_tx(Req, Data, Nodes)-> ok
 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}->
+    try
+       {slave, Receiver} ! {self(), {Req, Data}}
+    catch
+        _->ok
+    after
+        send_tx(Req, Data, Members)
+    end.
+
+%% @breif Q個のackを待つ。Remainが0になったら諦める。
+%% @spec  recv_ack( Q, Remain ) -> {ok, Remain} | {error, quorum_nodes_failed}
+recv_ack(_Q, 0)->      {error, quorum_nodes_failed};
+recv_ack( 0, Remain)-> {ok, Remain};
+recv_ack( Q, Remain)->
+    receive
+        {ok, _Result}->
 %           io:format( "ok: ~p.~n", [Result]),
-            recv_ack(N, Q-1, Remain);
+            recv_ack(Q-1, Remain);
         {error, Reason}->
             io:format( "failed: ~p.~n", [Reason]),
-            recv_ack(N, Q, Remain-1)
+            recv_ack(Q, Remain-1)
 %    after T->
 %           {error, timeout}
@@ -128 +158 @@
 
 replica()->
+    replica(3, slave).
+
+%% spawn N workers...
+%% replica(N) -> ok
+replica(N, ProcessName) when N > 1 ->
+    true=register(ProcessName, self()),
     mnesia_wrapper_init(),
-    WriteTx = fun(Data)->
-                      mnesia:transaction(
-                        fun()->
-                                mnesia:write(Data)
-                        end) end,
-    ReadTx = fun(Key)->
-                     mnesia:transaction(
-                       fun()->
-                               mnesia:read(data, Key)
-                       end) end,
-    mnesia_wrapper_loop(WriteTx, ReadTx),
-    mnesia:stop().
+    io:format("entering ~p loop: process name is '~p'.~n", [?MODULE, ProcessName]),
+    io:format("if you want to stop: ~p ! { self(), stop }.~n", [ProcessName]),
+    Pids=spawn_mnesia_wrapper_loop( N ),
+    receive
+        {_From, stop}->
+            stop_mnesia_wrapper_loop( Pids ),
+            mnesia:stop(),
+            true=unregister(ProcessName),
+            ok
+    end.
 
 %% mnesia wrapper
 %erl -sname b@localhost  -s qnesia mnesia_wrapper  -noshell -s init stop
 mnesia_wrapper_init()->
-    ok=mnesia:create_schema([node()]),
+    case mnesia:create_schema([node()]) of
+        {error, Reason}->
+            io:format( "create_schema: ~p~n", [Reason]);
+        ok -> ok;
+        _ -> ng
+    end,
     ok=mnesia:start(),
-    {atomic,ok}=mnesia:create_table( data, [
+    case mnesia:create_table( data, [
                                 {disc_copies, [node()]},
-                                {attributes, record_info(fields, data)}
-                                ]),
-    true=register(slave, self()),%, slave),
-    io:format("entering ~p loop.~n", [?MODULE]).
+                                {attributes, record_info(fields, data)},
+                                {type, set}
+                               ]) of
+        {error, Reason2}->
+            io:format( "create_table: ~p~n~p~n", [Reason2, mnesia:error_description(Reason2)]);
+        ok -> ok;
+        _ -> ng
+    end.
+
+spawn_mnesia_wrapper_loop( 0 )-> [];
+spawn_mnesia_wrapper_loop( N ) when N > 0 ->
+    Pid = spawn_link( fun()->mnesia_wrapper_loop_init() end ),
+    [Pid | spawn_mnesia_wrapper_loop( N-1 )].
+
+stop_mnesia_wrapper_loop( [] )->
+    {ok,done};
+stop_mnesia_wrapper_loop( [Pid|Pids] ) ->
+    Pid ! {self(), stop},
+    receive
+        {_From, {ok, stopped}}->
+            stop_mnesia_wrapper_loop( Pids );
+        {_From, _Other} -> %意味ねー
+            stop_mnesia_wrapper_loop( Pids )
+    end.
+
+mnesia_wrapper_loop_init()->
+    WriteTx = fun(Data)->
+                      mnesia:transaction( fun()->mnesia:write(Data)end )
+              end,
+    ReadTx  = fun(Key)->
+                      mnesia:transaction( fun()->mnesia:read(data, Key)end )
+              end,
+    mnesia_wrapper_loop(WriteTx, ReadTx).
 
 mnesia_wrapper_loop(WriteTx, ReadTx)->
@@ -158 +226 @@
         {From, stop} ->
             io:format("stopping ~p loop.~n", [?MODULE]),
-            From ! {ok, stopped};
+            From ! {self(), {ok, stopped}};
         {From, {write, Data}}->
             case mnesia:transaction(% Fun, Argv, 3 ) of
@@ -167 +235 @@
                                    From ! {self(), {ok, atomic}},
                                    receive
-                                       {From, {error, do_rollback}}-> mnesia:abort(rollback_done);
-                                       {From, {ok, do_commit}}    -> ok;
-                                       _->      mnesia:abort(unknown_error)
+                                       {From, {error, do_rollback}}->
+                                           mnesia:abort(rollback_done);
+                                       {From, {ok, do_commit}}-> 
+                                           ok;
+                                       _->     
+                                           mnesia:abort(unknown_error)
                                    end;
                                {aborted, Reason} ->