Packages
ra
0.9.4
3.1.9
3.1.8
3.1.7
3.1.6
3.1.5
3.1.4
3.1.3
3.1.2
3.1.1
3.1.0
3.0.2
3.0.1
3.0.0
3.0.0-beta.1
2.17.3
2.17.2
2.17.1
2.17.0
2.16.13
2.16.12
2.16.11
2.16.10
2.16.9
2.16.8
2.16.7
2.16.6
2.16.5
2.16.4
2.16.3
2.16.2
2.16.1
2.16.0
2.16.0-pre.12
2.16.0-pre.11
2.16.0-pre.10
2.16.0-pre.9
2.16.0-pre.8
2.16.0-pre.7
2.16.0-pre.6
2.16.0-pre.5
2.16.0-pre.4
2.16.0-pre.3
2.16.0-pre.2
2.16.0-pre.1
2.15.4
2.15.3
2.15.2
2.15.1
2.15.0
2.14.0
2.13.6
2.13.5
2.13.4
2.13.3
2.13.2
2.13.1
2.13.0
2.13.0-pre.1
2.12.0
2.11.0
2.11.0-pre.1
2.10.2-pre.2
2.10.2-pre.1
2.10.1
2.10.0
2.10.0-pre.3
2.10.0-pre.2
2.10.0-pre.1
2.9.10-pre.1
2.9.1
2.9.1-pre.2
2.9.1-pre.1
2.9.0
2.8.0
retired
2.7.3
2.7.2
2.7.1
2.7.0
2.7.0-pre.3
2.7.0-pre.2
2.7.0-pre.1
2.6.3
2.6.2
2.6.1
2.6.0-pre.1
2.5.1
2.5.1-pre.1
2.5.0
2.4.9
2.4.8
2.4.7
2.4.6
2.4.5
2.4.4
2.4.3
2.4.2
retired
2.4.1
2.4.0
2.3.0
2.2.0
2.1.0
2.0.13
2.0.12
2.0.11
2.0.10
2.0.9
2.0.8
2.0.7
2.0.6
2.0.5
2.0.4
2.0.3
2.0.2
2.0.1
2.0.0
1.1.9
1.1.8
1.1.7
1.1.6
1.1.5
1.1.4
1.1.3
1.1.2
1.1.1
1.1.0
1.0.8
1.0.7
1.0.6
1.0.5
1.0.4
1.0.3
1.0.2
1.0.1
1.0.0
0.9.6
0.9.5
0.9.4
0.9.2
0.3.3
retired
0.3.2
retired
0.3.1
retired
Raft library
Current section
Files
Jump to
Current section
Files
src/ra_log_meta.erl
%% @hidden
-module(ra_log_meta).
-behaviour(gen_batch_server).
-export([start_link/1,
init/1,
handle_batch/2,
terminate/2,
format_status/1,
store/3,
store_sync/3,
delete/1,
delete_sync/1,
fetch/2,
fetch/3
]).
-include("ra.hrl").
%% centralised meta data storage server for ra servers.
-type key() :: current_term | voted_for | last_applied.
-type value() :: non_neg_integer() | atom() | {atom(), atom()}.
-define(TBL_NAME, ?MODULE).
-define(TIMEOUT, 30000).
-record(?MODULE, {ref :: reference()}).
-opaque state() :: #?MODULE{}.
-export_type([state/0]).
-spec start_link(Config :: map()) ->
{ok, pid()} | {error, {already_started, pid()}}.
start_link(Config) ->
gen_batch_server:start_link({local, ?MODULE}, ?MODULE, Config, []).
-spec init(file:filename()) -> {ok, state()}.
init(Dir) ->
MetaFile = filename:join(Dir, "meta.dets"),
ok = filelib:ensure_dir(MetaFile),
{ok, Ref} = dets:open_file(?TBL_NAME, [{file, MetaFile},
%% fsync is done explicitly
{auto_save, infinity}]),
_ = ets:new(?TBL_NAME, [named_table, public, {read_concurrency, true}]),
?TBL_NAME = dets:to_ets(?TBL_NAME, ?TBL_NAME),
?INFO("ra: meta data store initialised. ~b record(s) recovered",
[ets:info(?TBL_NAME, size)]),
{ok, #?MODULE{ref = Ref}}.
handle_batch(Commands, #?MODULE{ref = Ref} = State) ->
DoInsert =
fun (Id, Key, Value, Inserts0) ->
case Inserts0 of
#{Id := Data} ->
Inserts0#{Id => update_key(Key, Value, Data)};
_ ->
case dets:lookup(Ref, Id) of
[{Id, T, V, A}] ->
Data = {Id, T, V, A},
Inserts0#{Id => update_key(Key, Value, Data)};
[] ->
Data = {Id, undefined, undefined, undefined},
Inserts0#{Id => update_key(Key, Value, Data)}
end
end
end,
{Inserts, Replies} =
lists:foldl(
fun ({cast, {store, Id, Key, Value}},
{Inserts0, Replies}) ->
{DoInsert(Id, Key, Value, Inserts0), Replies};
({call, From, {store, Id, Key, Value}},
{Inserts0, Replies}) ->
{DoInsert(Id, Key, Value, Inserts0),
[{reply, From, ok} | Replies]};
({cast, {delete, Id}}, {Inserts0, Replies}) ->
{handle_delete(Id, Ref, Inserts0), Replies};
({call, From, {delete, Id}},{Inserts0, Replies}) ->
{handle_delete(Id, Ref, Inserts0),
[{reply, From, ok} | Replies]}
end, {#{}, []}, Commands),
Objects = maps:values(Inserts),
ok = dets:insert(?MODULE, Objects),
true = ets:insert(?MODULE, Objects),
ok = dets:sync(?MODULE),
{ok, Replies, State}.
terminate(_, #?MODULE{ref = Ref}) ->
_ = dets:close(Ref),
ok.
format_status(State) ->
State.
%% send a message to the meta data store using cast
-spec store(ra_uid(), key(), value()) -> ok.
store(UId, Key, Value) ->
gen_batch_server:cast(?MODULE, {store, UId, Key, Value}).
%% waits until batch has been processed and synced.
%% when it returns the store request has been safely flushed to disk
-spec store_sync(ra_uid(), key(), value()) -> ok.
store_sync(UId, Key, Value) ->
gen_batch_server:call(?MODULE, {store, UId, Key, Value}, ?TIMEOUT).
-spec delete(ra_uid()) -> ok.
delete(UId) ->
gen_batch_server:cast(?MODULE, {delete, UId}).
-spec delete_sync(ra_uid()) -> ok.
delete_sync(UId) ->
gen_batch_server:call(?MODULE, {delete, UId}, ?TIMEOUT).
%% READER API
-spec fetch(ra_uid(), key()) -> value() | undefined.
fetch(Id, current_term) ->
maybe_fetch(Id, 2);
fetch(Id, voted_for) ->
maybe_fetch(Id, 3);
fetch(Id, last_applied) ->
maybe_fetch(Id, 4).
-spec fetch(ra_uid(), key(), term()) -> value().
fetch(Id, Key, Default) ->
case fetch(Id, Key) of
undefined -> Default;
Value -> Value
end.
%%% internal
maybe_fetch(Id, Pos) ->
try ets:lookup_element(?TBL_NAME, Id, Pos) of
E -> E
catch
_:badarg ->
undefined
end.
handle_delete(Id, Ref, Inserts) ->
_ = dets:delete(Ref, Id),
_ = ets:delete(?MODULE, Id),
maps:remove(Id, Inserts).
update_key(current_term, Value, Data) ->
setelement(2, Data, Value);
update_key(voted_for, Value, Data) ->
setelement(3, Data, Value);
update_key(last_applied, Value, Data) ->
setelement(4, Data, Value).