umgeher/leveled
0
Fork 0
Code Issues Pull requests Releases Activity Actions

Switch SFT to fsm

Browse source 
To amke it easier to context switch between the two file formats, make
the SFT file clerk a FSM like the CDB one.
This commit is contained in:
martinsumner 2016-11-07 17:08:50 +00:00
parent aacf377f0a
commit 05338a1e85
2 changed files with 147 additions and 124 deletions
Download patch file Download diff file Expand all files Collapse all files

16
src/leveled_cdb.erl
View file

@ -1,21 +1,19 @@
%% -------- CDB File Clerk ---------
%% %%
%% This is a modified version of the cdb module provided by Tom Whitcomb. %% This is a modified version of the cdb module provided by Tom Whitcomb.
%% %%
%% - https://github.com/thomaswhitcomb/erlang-cdb %% - https://github.com/thomaswhitcomb/erlang-cdb
%% %%
%% The CDB module is an implementation of the constant database format
%% described by DJ Bernstein
%%
%% - https://cr.yp.to/cdb.html
%%
%% The primary differences are: %% The primary differences are:
%% - Support for incrementally writing a CDB file while keeping the hash table %% - Support for incrementally writing a CDB file while keeping the hash table
%% in memory %% in memory
%% - The ability to scan a database and accumulate all the Key, Values to
%% rebuild in-memory tables on startup
%% - The ability to scan a database in blocks of sequence numbers %% - The ability to scan a database in blocks of sequence numbers
%% %% - The applictaion of a CRC chekc by default to all values
%% This is to be used in eleveledb, and in this context:
%% - Keys will be a combinatio of the PrimaryKey and the Sequence Number
%% - Values will be a serialised version on the whole object, and the
%% IndexChanges associated with the transaction
%% Where the IndexChanges are all the Key changes required to be added to the
%% ledger to complete the changes (the addition of postings and tombstones).
%% %%
%% This module provides functions to create and query a CDB (constant database). %% This module provides functions to create and query a CDB (constant database).
%% A CDB implements a two-level hashtable which provides fast {key,value} %% A CDB implements a two-level hashtable which provides fast {key,value}

255
src/leveled_sft.erl
View file

@ -142,16 +142,22 @@
-module(leveled_sft). -module(leveled_sft).
-behaviour(gen_server). -behaviour(gen_fsm).
-include("include/leveled.hrl"). -include("include/leveled.hrl").
-export([init/1, -export([init/1,
handle_call/3, handle_sync_event/4,
handle_cast/2, handle_event/3,
handle_info/2, handle_info/3,
terminate/2, terminate/3,
code_change/3, code_change/4,
sft_new/4, starting/2,
starting/3,
reader/3,
delete_pending/3,
delete_pending/2]).
-export([sft_new/4,
sft_newfroml0cache/4, sft_newfroml0cache/4,
sft_open/1, sft_open/1,
sft_get/2, sft_get/2,
@ -161,8 +167,9 @@
sft_checkready/1, sft_checkready/1,
sft_setfordelete/2, sft_setfordelete/2,
sft_deleteconfirmed/1, sft_deleteconfirmed/1,
sft_getmaxsequencenumber/1, sft_getmaxsequencenumber/1]).
generate_randomkeys/1]).
-export([generate_randomkeys/1]).
-include_lib("eunit/include/eunit.hrl"). -include_lib("eunit/include/eunit.hrl").
@ -202,7 +209,6 @@
handle :: file:fd(), handle :: file:fd(),
background_complete = false :: boolean(), background_complete = false :: boolean(),
oversized_file = false :: boolean(), oversized_file = false :: boolean(),
ready_for_delete = false ::boolean(),
penciller :: pid()}). penciller :: pid()}).
@ -221,65 +227,68 @@ sft_new(Filename, KL1, KL2, LevelInfo) ->
LevelInfo LevelInfo
end end
end, end,
{ok, Pid} = gen_server:start(?MODULE, [], []), {ok, Pid} = gen_fsm:start(?MODULE, [], []),
Reply = gen_server:call(Pid, Reply = gen_fsm:sync_send_event(Pid,
{sft_new, Filename, KL1, KL2, LevelR}, {sft_new, Filename, KL1, KL2, LevelR},
infinity), infinity),
{ok, Pid, Reply}. {ok, Pid, Reply}.
sft_newfroml0cache(Filename, Slots, FetchFun, Options) -> sft_newfroml0cache(Filename, Slots, FetchFun, Options) ->
{ok, Pid} = gen_server:start(?MODULE, [], []), {ok, Pid} = gen_fsm:start(?MODULE, [], []),
case Options#sft_options.wait of case Options#sft_options.wait of
true -> true ->
KL1 = leveled_pmem:to_list(Slots, FetchFun), KL1 = leveled_pmem:to_list(Slots, FetchFun),
Reply = gen_server:call(Pid, Reply = gen_fsm:sync_send_event(Pid,
{sft_new, {sft_new,
Filename, Filename,
KL1, KL1,
[], [],
#level{level=0}}, #level{level=0}},
infinity), infinity),
{ok, Pid, Reply}; {ok, Pid, Reply};
false -> false ->
gen_server:cast(Pid, gen_fsm:send_event(Pid,
{sft_newfroml0cache, {sft_newfroml0cache,
Filename, Filename,
Slots, Slots,
FetchFun, FetchFun,
Options#sft_options.penciller}), Options#sft_options.penciller}),
{ok, Pid, noreply} {ok, Pid, noreply}
end. end.
sft_open(Filename) -> sft_open(Filename) ->
{ok, Pid} = gen_server:start(?MODULE, [], []), {ok, Pid} = gen_fsm:start(?MODULE, [], []),
case gen_server:call(Pid, {sft_open, Filename}, infinity) of case gen_fsm:sync_send_event(Pid, {sft_open, Filename}, infinity) of
{ok, {SK, EK}} -> {ok, {SK, EK}} ->
{ok, Pid, {SK, EK}} {ok, Pid, {SK, EK}}
end. end.
sft_setfordelete(Pid, Penciller) -> sft_setfordelete(Pid, Penciller) ->
gen_server:call(Pid, {set_for_delete, Penciller}, infinity). gen_fsm:sync_send_event(Pid, {set_for_delete, Penciller}, infinity).
sft_get(Pid, Key) -> sft_get(Pid, Key) ->
gen_server:call(Pid, {get_kv, Key}, infinity). gen_fsm:sync_send_event(Pid, {get_kv, Key}, infinity).
sft_getkvrange(Pid, StartKey, EndKey, ScanWidth) -> sft_getkvrange(Pid, StartKey, EndKey, ScanWidth) ->
gen_server:call(Pid, {get_kvrange, StartKey, EndKey, ScanWidth}, infinity). gen_fsm:sync_send_event(Pid,
{get_kvrange, StartKey, EndKey, ScanWidth},
infinity).
sft_clear(Pid) -> sft_clear(Pid) ->
gen_server:call(Pid, clear, infinity). gen_fsm:sync_send_event(Pid, {set_for_delete, false}, infinity),
gen_fsm:sync_send_event(Pid, close, 1000).
sft_close(Pid) -> sft_close(Pid) ->
gen_server:call(Pid, close, 1000). gen_fsm:sync_send_event(Pid, close, 1000).
sft_deleteconfirmed(Pid) -> sft_deleteconfirmed(Pid) ->
gen_server:cast(Pid, close). gen_fsm:send_event(Pid, close).
sft_checkready(Pid) -> sft_checkready(Pid) ->
gen_server:call(Pid, background_complete, 20). gen_fsm:sync_send_event(Pid, background_complete, 20).
sft_getmaxsequencenumber(Pid) -> sft_getmaxsequencenumber(Pid) ->
gen_server:call(Pid, get_maxsqn, infinity). gen_fsm:sync_send_event(Pid, get_maxsqn, infinity).
@ -288,52 +297,75 @@ sft_getmaxsequencenumber(Pid) ->
%%%============================================================================ %%%============================================================================
init([]) -> init([]) ->
{ok, #state{}}. {ok, starting, #state{}}.
handle_call({sft_new, Filename, KL1, [], _LevelR=#level{level=L}}, starting({sft_new, Filename, KL1, [], _LevelR=#level{level=L}}, _From, _State)
_From, when L == 0 ->
_State) when L == 0 ->
{ok, State} = create_levelzero(KL1, Filename), {ok, State} = create_levelzero(KL1, Filename),
{reply, {reply,
{{[], []}, {{[], []}, State#state.smallest_key, State#state.highest_key},
State#state.smallest_key, reader,
State#state.highest_key},
State}; State};
handle_call({sft_new, Filename, KL1, KL2, LevelR}, _From, _State) -> starting({sft_new, Filename, KL1, KL2, LevelR}, _From, _State) ->
case create_file(Filename) of case create_file(Filename) of
{Handle, FileMD} -> {Handle, FileMD} ->
{ReadHandle, UpdFileMD, KeyRemainders} = complete_file(Handle, {ReadHandle, UpdFileMD, KeyRemainders} = complete_file(Handle,
FileMD, FileMD,
KL1, KL2, KL1, KL2,
LevelR), LevelR),
{reply, {KeyRemainders, {reply,
UpdFileMD#state.smallest_key, {KeyRemainders,
UpdFileMD#state.highest_key}, UpdFileMD#state.smallest_key,
UpdFileMD#state{handle=ReadHandle, filename=Filename}} UpdFileMD#state.highest_key},
reader,
UpdFileMD#state{handle=ReadHandle, filename=Filename}}
end; end;
handle_call({sft_open, Filename}, _From, _State) -> starting({sft_open, Filename}, _From, _State) ->
{_Handle, FileMD} = open_file(#state{filename=Filename}), {_Handle, FileMD} = open_file(#state{filename=Filename}),
leveled_log:log("SFT01", [Filename]), leveled_log:log("SFT01", [Filename]),
{reply, {reply,
{ok, {ok, {FileMD#state.smallest_key, FileMD#state.highest_key}},
{FileMD#state.smallest_key, FileMD#state.highest_key}}, reader,
FileMD}; FileMD}.
handle_call({get_kv, Key}, _From, State) ->
starting({sft_newfroml0cache, Filename, Slots, FetchFun, PCL}, _State) ->
SW = os:timestamp(),
Inp1 = leveled_pmem:to_list(Slots, FetchFun),
{ok, State} = create_levelzero(Inp1, Filename),
leveled_log:log_timer("SFT03", [Filename], SW),
case PCL of
undefined ->
{next_state, reader, State};
_ ->
leveled_penciller:pcl_confirml0complete(PCL,
State#state.filename,
State#state.smallest_key,
State#state.highest_key),
{next_state, reader, State}
end.
reader({get_kv, Key}, _From, State) ->
Reply = fetch_keyvalue(State#state.handle, State, Key), Reply = fetch_keyvalue(State#state.handle, State, Key),
statecheck_onreply(Reply, State); {reply, Reply, reader, State};
handle_call({get_kvrange, StartKey, EndKey, ScanWidth}, _From, State) -> reader({get_kvrange, StartKey, EndKey, ScanWidth}, _From, State) ->
Reply = pointer_append_queryresults(fetch_range_kv(State#state.handle, Reply = pointer_append_queryresults(fetch_range_kv(State#state.handle,
State, State,
StartKey, StartKey,
EndKey, EndKey,
ScanWidth), ScanWidth),
self()), self()),
statecheck_onreply(Reply, State); {reply, Reply, reader, State};
handle_call(close, _From, State) -> reader(get_maxsqn, _From, State) ->
{stop, normal, ok, State}; {reply, State#state.highest_sqn, reader, State};
handle_call(clear, _From, State) -> reader({set_for_delete, Penciller}, _From, State) ->
{stop, normal, ok, State#state{ready_for_delete=true}}; leveled_log:log("SFT02", [State#state.filename]),
handle_call(background_complete, _From, State) -> {reply,
ok,
delete_pending,
State#state{penciller=Penciller},
?DELETE_TIMEOUT};
reader(background_complete, _From, State) ->
if if
State#state.background_complete == true -> State#state.background_complete == true ->
{reply, {reply,
@ -341,67 +373,60 @@ handle_call(background_complete, _From, State) ->
State#state.filename, State#state.filename,
State#state.smallest_key, State#state.smallest_key,
State#state.highest_key}, State#state.highest_key},
reader,
State} State}
end; end;
handle_call({set_for_delete, Penciller}, _From, State) -> reader(close, _From, State) ->
leveled_log:log("SFT02", [State#state.filename]), ok = file:close(State#state.handle),
{reply, {stop, normal, ok, State}.
ok,
State#state{ready_for_delete=true,
penciller=Penciller},
?DELETE_TIMEOUT};
handle_call(get_maxsqn, _From, State) ->
statecheck_onreply(State#state.highest_sqn, State).
handle_cast({sft_newfroml0cache, Filename, Slots, FetchFun, PCL}, _State) -> delete_pending({get_kv, Key}, _From, State) ->
SW = os:timestamp(), Reply = fetch_keyvalue(State#state.handle, State, Key),
Inp1 = leveled_pmem:to_list(Slots, FetchFun), {reply, Reply, delete_pending, State, ?DELETE_TIMEOUT};
{ok, State} = create_levelzero(Inp1, Filename), delete_pending({get_kvrange, StartKey, EndKey, ScanWidth}, _From, State) ->
leveled_log:log_timer("SFT03", [Filename], SW), Reply = pointer_append_queryresults(fetch_range_kv(State#state.handle,
case PCL of State,
undefined -> StartKey,
{noreply, State}; EndKey,
_ -> ScanWidth),
leveled_penciller:pcl_confirml0complete(PCL, self()),
State#state.filename, {reply, Reply, delete_pending, State, ?DELETE_TIMEOUT};
State#state.smallest_key, delete_pending(get_maxsqn, _From, State) ->
State#state.highest_key), {reply, State#state.highest_sqn, delete_pending, State, ?DELETE_TIMEOUT};
{noreply, State} delete_pending(close, _From, State) ->
end; leveled_log:log("SFT06", [State#state.filename]),
handle_cast(close, State) -> ok = file:close(State#state.handle),
ok = file:delete(State#state.filename),
{stop, normal, ok, State}.
delete_pending(timeout, State) ->
leveled_log:log("SFT05", [timeout, State#state.filename]),
ok = leveled_penciller:pcl_confirmdelete(State#state.penciller,
State#state.filename),
{next_state, delete_pending, State, ?DELETE_TIMEOUT};
delete_pending(close, State) ->
leveled_log:log("SFT06", [State#state.filename]),
ok = file:close(State#state.handle),
ok = file:delete(State#state.filename),
{stop, normal, State}. {stop, normal, State}.
handle_info(timeout, State) -> handle_sync_event(_Msg, _From, StateName, State) ->
if {reply, undefined, StateName, State}.
State#state.ready_for_delete == true ->
leveled_log:log("SFT05", [timeout, State#state.filename]),
ok = leveled_penciller:pcl_confirmdelete(State#state.penciller,
State#state.filename),
{noreply, State, ?DELETE_TIMEOUT}
end.
terminate(Reason, State) -> handle_event(_Msg, StateName, State) ->
leveled_log:log("SFT05", [Reason, State#state.filename]), {next_state, StateName, State}.
case State#state.ready_for_delete of
true ->
leveled_log:log("SFT06", [State#state.filename]),
ok = file:close(State#state.handle),
ok = file:delete(State#state.filename);
_ ->
ok = file:close(State#state.handle)
end.
code_change(_OldVsn, State, _Extra) -> handle_info(_Msg, StateName, State) ->
{ok, State}. {next_state, StateName, State}.
terminate(Reason, _StateName, State) ->
leveled_log:log("SFT05", [Reason, State#state.filename]).
code_change(_OldVsn, StateName, State, _Extra) ->
{ok, StateName, State}.
statecheck_onreply(Reply, State) ->
case State#state.ready_for_delete of
true ->
{reply, Reply, State, ?DELETE_TIMEOUT};
false ->
{reply, Reply, State}
end.
%%%============================================================================ %%%============================================================================
%%% Internal functions %%% Internal functions