add an ets backend and rafter_backend behaviour

This commit incorporates the following changes:

Improvements
 * Separate Read and Write Path
   * Read path is linearizable, but doesn't require logging operations
   * Uses successful next tick of append_entries as an implicit quorum
 * ETS Backend
 * Backends can maintain their internal state now
 * Statem test for ETS Backend
 * script to start a node for development purposes

Bugfixes
  * uses noops to advance commitIndex
  * explicitly handle timeouts in rafter_consensus_fsm

Libraries
  * Removed eper from rebar.config

Author: andrewjstone

README.md

@@ -0,0 +1 @@
+erl -pa deps/*/ebin ebin -setcookie rafter_localhost_test -name $1@127.0.0.1 -eval "rafter:start_test_node($1)."

bin/start-node

@@ -18,7 +18,10 @@
             prev_log_index :: non_neg_integer(),
             prev_log_term :: non_neg_integer(),
             entries :: term(),
-            commit_index :: non_neg_integer()}).
+            commit_index :: non_neg_integer(),
+
+            %% This is used during read-only operations
+            send_clock :: non_neg_integer()}).
 
 -record(append_entries_rpy, {
             from :: atom(),
@@ -28,13 +31,17 @@
             %% it prevents duplicate responses from causing recommits and helps
             %% maintain safety. In the raft reference implementation (logcabin)
             %% they cancel the in flight RPC's instead. That's difficult
-            %% to do correctly(without races) in erlang with asynchronous 
+            %% to do correctly(without races) in erlang with asynchronous
             %% messaging and mailboxes.
             index :: non_neg_integer(),
+
+            %% This is used during read-only operations
+            send_clock :: non_neg_integer(),
+
             success :: boolean()}).
 
 -record(rafter_entry, {
-        type :: config | op,
+        type :: noop | config | op,
         term :: non_neg_integer(),
         index :: non_neg_integer(),
         cmd :: term()}).
@@ -44,16 +51,16 @@
     term = 0 :: non_neg_integer()}).
 
 -record(config, {
-    state = blank :: 
+    state = blank ::
         %% The configuration specifies no servers. Servers that are new to the
         %% cluster and have empty logs start in this state.
-        blank   | 
+        blank   |
         %% The configuration specifies a single list of servers: a quorum
         %% requires any majority of oldservers.
-        stable  | 
+        stable  |
         %% The configuration specifies two lists of servers: a quorum requires
         %% any majority of oldservers, but the newservers also receive log entries.
-        staging | 
+        staging |
         %% The configuration specifies two lists of servers: a quorum requires
         %% any majority of oldservers and any majority of the newservers.
         transitional,

include/rafter.hrl

@@ -3,7 +3,10 @@
     timer :: timer:tref(),
     from :: term(),
     index :: non_neg_integer(),
-    term :: non_neg_integer()}).
+    term :: non_neg_integer(),
+
+    %% only used during read_only commands
+    cmd :: term()}).
 
 -record(state, {
     leader :: term(),
@@ -12,25 +15,37 @@
     commit_index = 0 :: non_neg_integer(),
     init_config :: undefined | list() | complete,
 
-    %% The last time a timer was created
-    timer_start :: non_neg_integer(),
-
-    %% The duration of the timer
-    timer_duration :: non_neg_integer(),
+    %% Used for Election and Heartbeat timeouts
+    timer :: timer:tref(),
 
-    %% leader state
+    %% leader state: contains nextIndex for each peer
     followers = dict:new() :: dict(),
 
-    %% Responses from RPCs to other servers
+    %% Dict keyed by peer id.
+    %% contains true as val when candidate
+    %% contains match_indexes as val when leader
     responses = dict:new() :: dict(),
 
-    %% Outstanding Client Requests
+    %% Logical clock to allow read linearizability
+    %% Reset to 0 on leader election.
+    send_clock = 0 :: non_neg_integer(),
+
+    %% Keep track of the highest send_clock received from each peer
+    %% Reset on leader election
+    send_clock_responses = dict:new() :: dict(),
+
+    %% Outstanding Client Write Requests
     client_reqs = [] :: [#client_req{}],
 
+    %% Outstanding Client Read Requests
+    %% Keyed on send_clock, Val = [#client_req{}]
+    read_reqs = orddict:new() :: orddict:orddict(),
+
     %% All servers making up the ensemble
     me :: string(),
 
     config :: term(),
-    
-    %% We allow pluggable state machine modules.
-    state_machine :: atom()}).
+
+    %% We allow pluggable backend state machine modules.
+    state_machine :: atom(),
+    backend_state :: term()}).

include/rafter_consensus_fsm.hrl

@@ -1,2 +1,2 @@
--record(rafter_opts, {state_machine = rafter_sm_echo :: atom(),
+-record(rafter_opts, {state_machine = rafter_backend_echo :: atom(),
                       logdir :: string()}).

include/rafter_opts.hrl

@@ -5,6 +5,5 @@
 {erl_opts, [debug_info, warnings_as_errors, {parse_transform, lager_transform}]}.
 
 {deps, [{lager, ".*", {git, "git://github.com/basho/lager", {tag, "2.0.0rc2"}}},
-        {eper, ".*", {git, "git://github.com/basho/eper.git", {tag, "3280b736"}}},
         {druuid, ".*", {git, "git://github.com/kellymclaughlin/druuid.git", {tag, "0.2"}}} 
 ]}.

rebar.config

@@ -5,7 +5,7 @@
 -include("rafter_opts.hrl").
 
 %% API
--export([start_node/2, stop_node/1, op/2, set_config/2, 
+-export([start_node/2, stop_node/1, op/2, read_op/2, set_config/2,
          get_state/1, get_leader/1, get_entry/2, get_last_entry/1]).
 
 %% Test API
@@ -17,12 +17,16 @@ start_node(Peer, Opts) ->
 stop_node(Peer) ->
     rafter_sup:stop_peer(Peer).
 
-%% @doc Run an operation on the backend state machine. 
+%% @doc Run an operation on the backend state machine.
 %% Note: Peer is just the local node in production.
 op(Peer, Command) ->
     Id = druuid:v4(),
     rafter_consensus_fsm:op(Peer, {Id, Command}).
 
+read_op(Peer, Command) ->
+    Id = druuid:v4(),
+    rafter_consensus_fsm:read_op(Peer, {Id, Command}).
+
 set_config(Peer, NewServers) ->
     Id = druuid:v4(),
     rafter_consensus_fsm:set_config(Peer, {Id, NewServers}).
@@ -51,13 +55,13 @@ get_last_entry(Peer) ->
 start_cluster() ->
     application:start(lager),
     application:start(rafter),
-    Opts = #rafter_opts{state_machine=rafter_sm_echo, logdir="./log"},
+    Opts = #rafter_opts{state_machine=rafter_backend_echo, logdir="./log"},
     Peers = [peer1, peer2, peer3, peer4, peer5],
     [rafter_sup:start_peer(Me, Opts) || Me <- Peers].
 
 start_test_node(Name) ->
     application:start(lager),
     application:start(rafter),
     Me = {Name, node()},
-    Opts = #rafter_opts{state_machine=rafter_sm_echo, logdir="./"},
+    Opts = #rafter_opts{state_machine=rafter_backend_ets, logdir="./data"},
     start_node(Me, Opts).

src/rafter.erl

@@ -0,0 +1,9 @@
+-module(rafter_backend).
+
+-export([behaviour_info/1]).
+
+behaviour_info(callbacks) ->
+    [{init, 1}, {read, 2}, {write, 2}];
+behaviour_info(_) ->
+    undefined.
+

src/rafter_backend.erl

@@ -0,0 +1,15 @@
+-module(rafter_backend_echo).
+
+-behaviour(rafter_backend).
+
+%% Rafter backend callbacks
+-export([init/1, read/2, write/2]).
+
+init(_) ->
+    ok.
+
+read(Command, State) ->
+    {{ok, Command}, State}.
+
+write(Command, State) ->
+    {{ok, Command}, State}.

src/rafter_backend_echo.erl

@@ -0,0 +1,92 @@
+-module(rafter_backend_ets).
+
+-behaviour(rafter_backend).
+
+%% rafter_backend callbacks
+-export([init/1, stop/1, read/2, write/2]).
+
+-record(state, {peer :: atom() | {atom(), atom()}}).
+
+init(Peer) ->
+    State = #state{peer=Peer},
+    NewState = stop(State),
+    ets:new(rafter_backend_ets, [set, named_table, public]),
+    ets:new(rafter_backend_ets_tables, [set, named_table, public]),
+    NewState.
+
+stop(State) ->
+    catch ets:delete(rafter_backend_ets),
+    catch ets:delete(rafter_backend_ets_tables),
+    State.
+
+read({get, Table, Key}, State) ->
+    Val = try
+              case ets:lookup(Table, Key) of
+                  [{Key, Value}] ->
+                      {ok, Value};
+                  [] ->
+                      {ok, not_found}
+              end
+          catch _:E ->
+              {error, E}
+          end,
+     {Val, State};
+
+read(list_tables, State) ->
+    {{ok, [Table || {Table} <- ets:tab2list(rafter_backend_ets_tables)]},
+        State};
+
+read({list_keys, Table}, State) ->
+    Val = try
+              list_keys(Table)
+          catch _:E ->
+              {error, E}
+          end,
+    {Val, State}.
+
+write({new, Name}, State) ->
+    Val = try
+              ets:new((Name), [ordered_set, named_table, public]),
+              ets:insert(rafter_backend_ets_tables, {Name}),
+              {ok, Name}
+          catch _:E ->
+              {error, E}
+          end,
+    {Val, State};
+
+write({put, Table, Key, Value}, State) ->
+    Val = try
+              ets:insert(Table, {Key, Value}),
+              {ok, Value}
+          catch _:E ->
+              {error, E}
+          end,
+    {Val, State};
+
+write({delete, Table}, State) ->
+    Val =
+        try
+            ets:delete(Table),
+            ets:delete(rafter_backend_ets_tables, Table),
+            {ok, true}
+        catch _:E ->
+            {error, E}
+        end,
+    {Val, State};
+
+
+write({delete, Table, Key}, State) ->
+    Val = try
+              {ok, ets:delete(Table, Key)}
+          catch _:E ->
+              {error, E}
+          end,
+    {Val, State}.
+
+list_keys(Table) ->
+    list_keys(ets:first(Table), Table, []).
+
+list_keys('$end_of_table', _Table, Keys) ->
+    {ok, Keys};
+list_keys(Key, Table, Keys) ->
+    list_keys(ets:next(Table, Key), Table, [Key | Keys]).