wip ra_kv

Author: kjnilsson

src/ra.erl

@@ -35,7 +35,6 @@
          consistent_query/3,
          ping/2,
          % cluster operations
-         start_cluster/1,
          start_cluster/2,
          start_cluster/3,
          start_cluster/4,
@@ -350,16 +349,6 @@ start_cluster(System, ClusterName, Machine, ServerIds, Timeout)
                end || Id <- ServerIds],
     start_cluster(System, Configs, Timeout).
 
-%% @doc Same as `start_cluster/2' but uses the default Ra system.
-%% @param ServerConfigs a list of initial server configurations
-%% DEPRECATED: use start_cluster/2
-%% @end
--spec start_cluster([ra_server:ra_server_config()]) ->
-    {ok, [ra_server_id()], [ra_server_id()]} |
-    {error, cluster_not_formed}.
-start_cluster(ServerConfigs) ->
-  start_cluster(default, ServerConfigs).
-
 %% @doc Starts a new distributed ra cluster.
 %% @param System the system name
 %% @param ServerConfigs a list of initial server configurations

src/ra.hrl

@@ -83,7 +83,7 @@
 %% represent a unique entry in the ra log
 -type log_entry() :: {ra_index(), ra_term(), term()}.
 
--type chunk_flag() :: next | last.
+-type chunk_flag() :: pre | next | last.
 
 -type consistent_query_ref() :: {From :: term(), Query :: ra:query_fun(), ConmmitIndex :: ra_index()}.
 
@@ -169,7 +169,7 @@
         {term :: ra_term(), % the leader's term
          leader_id :: ra_server_id(),
          meta :: snapshot_meta(),
-         chunk_state :: {pos_integer(), chunk_flag()} | undefined,
+         chunk_state = {0, pre} :: {pos_integer(), chunk_flag()},
          data :: term()
         }).
 

src/ra_kv.erl

@@ -0,0 +1,138 @@
+-module(ra_kv).
+-behaviour(ra_machine).
+-include("src/ra.hrl").
+
+-include_lib("eunit/include/eunit.hrl").
+
+-export([
+         init/1,
+         apply/3,
+         % state_enter/2,
+         % tick/2,
+         init_aux/1,
+         handle_aux/5,
+         live_indexes/1,
+         overview/1
+        ]).
+
+-export([
+         start_cluster/3,
+         add_member/3,
+
+         put/4
+        ]).
+
+
+-define(STATE, ?MODULE).
+
+-type key() :: binary().
+-type value() :: term().
+
+-record(?STATE, {keys = #{} ::
+                 #{key() := [ra:index() | Hash :: non_neg_integer()]}}).
+
+
+-record(put, {key :: key(),
+              value :: term(),
+              meta :: #{size := non_neg_integer(),
+                        hash := integer()}}).
+
+-type command() :: #put{}.
+-opaque state() :: #?STATE{}.
+
+-export_type([state/0,
+              command/0]).
+
+%% mgmt
+-spec start_cluster(atom(), atom(), map()) ->
+    {ok, [ra_server_id()], [ra_server_id()]} |
+    {error, cluster_not_formed}.
+start_cluster(System, Name, #{members := ServerIds})
+  when is_atom(Name) andalso
+       is_atom(System) ->
+    Machine = {module, ?MODULE, #{}},
+    Configs = [begin
+                   UId = ra:new_uid(ra_lib:to_binary(Name)),
+                   #{id => Id,
+                     uid => UId,
+                     cluster_name => Name,
+                     log_init_args => #{uid => UId},
+                     initial_members => ServerIds,
+                     machine => Machine}
+               end || Id <- ServerIds],
+    ra:start_cluster(System, Configs).
+
+add_member(System, {Name, _} = Id, LeaderId) ->
+    {ok, Members, _} = ra:members(LeaderId),
+    UId = ra:new_uid(ra_lib:to_binary(Name)),
+    Machine = {module, ?MODULE, #{}},
+    Config = #{id => Id,
+               uid => UId,
+               cluster_name => Name,
+               log_init_args => #{uid => UId},
+               initial_members => Members,
+               machine => Machine},
+    ok = ra:start_server(System, Config),
+    {ok, _, _} =  ra:add_member(LeaderId, Id),
+    ok.
+
+
+%% client
+-spec put(ra:server_id(), key(), value(), non_neg_integer()) ->
+    {ok, map()} | {error, term()} | {timeout, ra:server_id()}.
+put(ServerId, Key, Value, Timeout) ->
+    Hash = erlang:phash2(Value),
+    Put = #put{key = Key,
+               value = Value,
+               meta = #{size => erlang:external_size(Value),
+                        hash => Hash}},
+    case ra:process_command(ServerId, Put, Timeout) of
+        {ok, {ok, Meta}, LeaderId} ->
+            {ok, Meta#{leader => LeaderId}};
+        Err ->
+            Err
+    end.
+
+
+%% get performs a consistent query that returns the index, hash and member set
+%% then perform an aux query to actually get the data for a given index.
+%% if addressing a follower (say there is a local one) then the read may need
+%% to wait if the index isn't yet available locally (term also need to be checked)
+%% or check that the machien state has the right index for a given key before
+%% reading the value from the log
+
+
+%% state machine
+
+init(_) ->
+    #?MODULE{}.
+
+apply(#{index := Idx} = Meta,
+      #put{key = Key,
+           meta = #{hash := Hash}},
+      #?STATE{keys = Keys} = State0) ->
+    State = State0#?STATE{keys = maps:put(Key, [Idx | Hash], Keys)},
+    {State, {ok, Meta}, []}.
+
+live_indexes(#?STATE{keys = Keys}) ->
+    maps:fold(fun (_K, [Idx | _], Acc) ->
+                      [Idx | Acc]
+              end, [], Keys).
+
+-record(aux, {}).
+init_aux(_) ->
+    #aux{}.
+
+handle_aux(_RaState, {call, _From}, take_snapshot, Aux, Internal) ->
+    MacState = ra_aux:machine_state(Internal),
+    LastAppliedIdx = ra_aux:last_applied(Internal),
+    %% TODO: replace release cursor with simpler snapshot effect that is always
+    %% attempted?
+    {reply, ok, Aux, Internal,
+     [{release_cursor, LastAppliedIdx, MacState}]};
+handle_aux(_RaState, _, _, Aux, Internal) ->
+    {no_reply, Aux, Internal}.
+
+overview(#?STATE{keys = Keys} = State) ->
+    #{num_keys => maps:size(Keys),
+      live_indexes => live_indexes(State)}.