dirty_query

Author: kjnilsson

.gitignore

@@ -10,6 +10,7 @@ erl_crash.dump
 /logs/
 /plugins/
 /xrefr
+elvis
 
 rabbitmq_rtopic_exchange.d
 *.plt

elvis.config

@@ -0,0 +1,26 @@
+[
+ {
+   elvis,
+   [
+    {config,
+     [#{dirs => ["src"],
+        filter => "*.erl",
+        ruleset => erl_files
+       },
+      #{dirs => ["."],
+        filter => "Makefile",
+        ruleset => makefiles
+       },
+      #{dirs => ["."],
+        filter => "rebar.config",
+        ruleset => rebar_config
+       },
+      #{dirs => ["."],
+        filter => "elvis.config",
+        ruleset => elvis_config
+       }
+     ]
+    }
+   ]
+ }
+].

src/ra.erl

@@ -6,7 +6,8 @@
          start_cluster/4,
          send/2,
          send_and_await_consensus/2,
-         send_and_notify/2
+         send_and_notify/2,
+         dirty_query/2
         ]).
 
 start_cluster(Num, Name, ApplyFun, InitialState) ->
@@ -37,3 +38,8 @@ send_and_await_consensus(Ref, Data) ->
     {ok, IdxTerm::{ra_index(), ra_term()}, Leader::ra_node_proc:server_ref()}.
 send_and_notify(Ref, Data) ->
     ra_node_proc:command(Ref, Data, notify_on_consensus).
+
+-spec dirty_query(Node::ra_node_proc:server_ref(), QueryFun::fun((term()) -> term())) ->
+    {ok, {ra_index(), ra_term()}, term()}.
+dirty_query(Node, QueryFun) ->
+    ra_node_proc:query(Node, QueryFun, dirty).

src/ra_node_proc.erl

@@ -6,7 +6,8 @@
 
 %% API
 -export([start_link/1,
-         command/3
+         command/3,
+         query/3
         ]).
 
 %% State functions
@@ -37,6 +38,8 @@
 -type pending_reply() :: {reply | notify, {ra_index(), ra_term()},
                           From::term()}.
 
+-type query_fun() :: fun((term()) -> term()).
+
 -export_type([server_ref/0]).
 
 -record(state, {node_state :: ra_node:ra_node_state(),
@@ -64,6 +67,12 @@ command(ServerRef, Data, ReplyMode) ->
         Reply -> {ok, Reply, ServerRef}
     end.
 
+-spec query(ra_node_proc:server_ref(), query_fun(), dirty | consensus) ->
+    {ok, IdxTerm::{ra_index(), ra_term()}, term()}.
+query(ServerRef, QueryFun, dirty) ->
+    gen_statem:call(ServerRef, {query, QueryFun, dirty}).
+
+
 %%%===================================================================
 %%% gen_statem callbacks
 %%%===================================================================
@@ -162,6 +171,9 @@ follower({call, From}, {command, _Data, _Flag},
 follower({call, From}, {command, _Data, _Flag} = Cmd,
          State = #state{pending_commands = Pending}) ->
     {keep_state, State#state{pending_commands = [{From, Cmd} | Pending]}};
+follower({call, From}, {query, QueryFun, dirty}, State = #state{node_state = NodeState}) ->
+    Reply = perform_query(QueryFun, dirty, NodeState),
+    {keep_state, State, [{reply, From, Reply}]};
 follower(EventType, Msg,
          State0 = #state{node_state = NodeState0 = #{id := Id}}) ->
     ?DBG("~p follower: ~p~n", [Id, Msg]),
@@ -179,7 +191,6 @@ follower(EventType, Msg,
              election_timeout_action(State)}
     end.
 
-
 handle_event(_EventType, EventContent, StateName, State) ->
     ?DBG("handle_event unknownn ~p~n", [EventContent]),
     {next_state, StateName, State}.
@@ -203,6 +214,11 @@ format_status(_Opt, [_PDict, _StateName, _State]) ->
 %%% Internal functions
 %%%===================================================================
 
+perform_query(QueryFun, dirty, #{machine_state := MacState,
+                                 last_applied := Last,
+                                 current_term := Term}) ->
+     {ok, {Last, Term}, QueryFun(MacState)}.
+
 make_caller_reply_actions(State = #state{pending_replies = []}) ->
     {State, [], []};
 make_caller_reply_actions(State = #state{pending_replies = Pending,

test/ra_SUITE.erl

@@ -8,7 +8,8 @@ all() ->
     [
      queue,
      send_and_await_consensus,
-     send_and_notify
+     send_and_notify,
+     dirty_query
     ].
 
 groups() ->
@@ -34,10 +35,7 @@ queue(_Config) ->
 send_and_await_consensus(_Config) ->
     [{APid, _A}, _B, _C]  = Cluster =
     ra:start_cluster(3, "test", fun erlang:'+'/2, 9),
-
     {ok, {1, 1}, _Leader} = ra:send_and_await_consensus(APid, 5),
-
-    % {{1, 1}, 14} = ra:query(APid, fun(S) -> S end, dirty),
     terminate_cluster(Cluster).
 
 send_and_notify(_Config) ->
@@ -51,6 +49,13 @@ send_and_notify(_Config) ->
     end,
     terminate_cluster(Cluster).
 
+dirty_query(_Config) ->
+    [{APid, _A}, _B, _C]  = Cluster =
+    ra:start_cluster(3, "test", fun erlang:'+'/2, 9),
+    {ok, {1, 1}, _Leader} = ra:send_and_await_consensus(APid, 5),
+    {ok, {1, 1}, 14} = ra:dirty_query(APid, fun(S) -> S end),
+    terminate_cluster(Cluster).
+
 % implements a simple queue machine
 queue_apply({enqueue, Msg}, State =#{queue := Q0, pending_dequeues := []}) ->
     Q = queue:in(Msg, Q0),

test/ra_node_SUITE.erl

@@ -17,6 +17,9 @@ all() ->
      node_steps_down
     ].
 
+groups() ->
+    [ {tests, [], all()} ].
+
 
 node_steps_down(_Config) ->
     State = base_state(3),