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

Removed SFT

Browse source 
Now moved over to SST on this branch
This commit is contained in:
martinsumner 2016-12-29 02:07:14 +00:00
parent c664483f03
commit dc28388c76
7 changed files with 481 additions and 2246 deletions
Download patch file Download diff file Expand all files Collapse all files

5
src/leveled_bookie.erl
View file

@ -1186,7 +1186,10 @@ hashtree_query_test() ->
{hashtree_query,
?STD_TAG,
false}),
?assertMatch(KeyHashList, HTFolder2()),
L0 = length(KeyHashList),
HTR2 = HTFolder2(),
?assertMatch(L0, length(HTR2)),
?assertMatch(KeyHashList, HTR2),
ok = book_close(Bookie2),
reset_filestructure().

13
src/leveled_log.erl
View file

@ -19,7 +19,7 @@
-define(GET_LOGPOINT, 160000).
-define(SST_LOGPOINT, 200000).
-define(LOG_LEVEL, [info, warn, error, critical]).
-define(SAMPLE_RATE, 15).
-define(SAMPLE_RATE, 16).
-define(LOGBASE, dict:from_list([
@ -96,7 +96,7 @@
{info, "Response to push_mem of ~w with "
++ "L0 pending ~w and merge backlog ~w"}},
{"P0019",
{info, "Rolling level zero to filename ~s"}},
{info, "Rolling level zero to filename ~s at ledger sqn ~w"}},
{"P0020",
{info, "Work at Level ~w to be scheduled for ~w with ~w "
++ "queue items outstanding at all levels"}},
@ -238,11 +238,18 @@
{error, "False result returned from SST with filename ~s as "
++ "slot ~w has failed crc check"}},
{"SST03",
{info, "Opening SST file with filename ~s keys ~w and slots ~w"}},
{info, "Opening SST file with filename ~s keys ~w slots ~w and"
++ " max sqn ~w"}},
{"SST04",
{info, "Exit called for reason ~w on filename ~s"}},
{"SST05",
{warn, "Rename rogue filename ~s to ~s"}},
{"SST06",
{info, "File ~s has been set for delete"}},
{"SST07",
{info, "Exit called and now clearing ~s"}},
{"SST08",
{info, "Completed creation of ~s at level ~w with max sqn ~w"}},
{"SFT01",
{info, "Opened filename with name ~s"}},

75
src/leveled_pclerk.erl
View file

@ -9,7 +9,7 @@
%%
%% -------- COMMITTING MANIFEST CHANGES ---------
%%
%% Once the Penciller has taken a manifest change, the SFT file owners which no
%% Once the Penciller has taken a manifest change, the SST file owners which no
%% longer form part of the manifest will be marked for delete. By marking for
%% deletion, the owners will poll to confirm when it is safe for them to be
%% deleted.
@ -225,7 +225,7 @@ merge(WI) ->
mark_for_delete([], _Penciller) ->
ok;
mark_for_delete([Head|Tail], Penciller) ->
ok = leveled_sft:sft_setfordelete(Head#manifest_entry.owner, Penciller),
ok = leveled_sst:sst_setfordelete(Head#manifest_entry.owner, Penciller),
mark_for_delete(Tail, Penciller).
@ -268,13 +268,13 @@ select_filetomerge(SrcLevel, Manifest) ->
%% Assumption is that there is a single SFT from a higher level that needs
%% to be merged into multiple SFTs at a lower level. This should create an
%% entirely new set of SFTs, and the calling process can then update the
%% Assumption is that there is a single SST from a higher level that needs
%% to be merged into multiple SSTs at a lower level. This should create an
%% entirely new set of SSTs, and the calling process can then update the
%% manifest.
%%
%% Once the FileToMerge has been emptied, the remainder of the candidate list
%% needs to be placed in a remainder SFT that may be of a sub-optimal (small)
%% needs to be placed in a remainder SST that may be of a sub-optimal (small)
%% size. This stops the need to perpetually roll over the whole level if the
%% level consists of already full files. Some smartness may be required when
%% selecting the candidate list so that small files just outside the candidate
@ -293,18 +293,22 @@ perform_merge({SrcPid, SrcFN}, CandidateList, LevelInfo, {Filepath, MSN}) ->
PointerList = lists:map(fun(P) ->
{next, P#manifest_entry.owner, all} end,
CandidateList),
MaxSQN = leveled_sst:sst_getmaxsequencenumber(SrcPid),
do_merge([{next, SrcPid, all}],
PointerList,
LevelInfo,
{Filepath, MSN},
MaxSQN,
0,
[]).
do_merge([], [], {SrcLevel, _IsB}, {_Filepath, MSN}, FileCounter, OutList) ->
do_merge([], [], {SrcLevel, _IsB}, {_Filepath, MSN}, _MaxSQN,
FileCounter, OutList) ->
leveled_log:log("PC011", [MSN, SrcLevel, FileCounter]),
OutList;
do_merge(KL1, KL2, {SrcLevel, IsB}, {Filepath, MSN}, FileCounter, OutList) ->
FileName = lists:flatten(io_lib:format(Filepath ++ "_~w_~w.sft",
do_merge(KL1, KL2, {SrcLevel, IsB}, {Filepath, MSN}, MaxSQN,
FileCounter, OutList) ->
FileName = lists:flatten(io_lib:format(Filepath ++ "_~w_~w.sst",
[SrcLevel + 1, FileCounter])),
leveled_log:log("PC012", [MSN, FileName]),
TS1 = os:timestamp(),
@ -312,12 +316,13 @@ do_merge(KL1, KL2, {SrcLevel, IsB}, {Filepath, MSN}, FileCounter, OutList) ->
KL1,
KL2,
IsB,
SrcLevel + 1),
SrcLevel + 1,
MaxSQN),
case Reply of
{{[], []}, null, _} ->
leveled_log:log("PC013", [FileName]),
leveled_log:log("PC014", [FileName]),
ok = leveled_sft:sft_clear(Pid),
ok = leveled_sst:sst_clear(Pid),
OutList;
{{KL1Rem, KL2Rem}, SmallestKey, HighestKey} ->
ExtMan = lists:append(OutList,
@ -327,7 +332,7 @@ do_merge(KL1, KL2, {SrcLevel, IsB}, {Filepath, MSN}, FileCounter, OutList) ->
filename=FileName}]),
leveled_log:log_timer("PC015", [], TS1),
do_merge(KL1Rem, KL2Rem,
{SrcLevel, IsB}, {Filepath, MSN},
{SrcLevel, IsB}, {Filepath, MSN}, MaxSQN,
FileCounter + 1, ExtMan)
end.
@ -377,7 +382,7 @@ find_randomkeys(FList, Count, Source) ->
KV1 = lists:nth(random:uniform(length(Source)), Source),
K1 = leveled_codec:strip_to_keyonly(KV1),
P1 = choose_pid_toquery(FList, K1),
FoundKV = leveled_sft:sft_get(P1, K1),
FoundKV = leveled_sst:sst_get(P1, K1),
Found = leveled_codec:strip_to_keyonly(FoundKV),
io:format("success finding ~w in ~w~n", [K1, P1]),
?assertMatch(K1, Found),
@ -386,21 +391,31 @@ find_randomkeys(FList, Count, Source) ->
merge_file_test() ->
KL1_L1 = lists:sort(generate_randomkeys(8000, 0, 1000)),
{ok, PidL1_1, _} = leveled_sft:sft_new("../test/KL1_L1.sft",
KL1_L1, [], 1),
{ok, PidL1_1, _} = leveled_sst:sst_new("../test/KL1_L1.sst",
1,
KL1_L1,
undefined),
KL1_L2 = lists:sort(generate_randomkeys(8000, 0, 250)),
{ok, PidL2_1, _} = leveled_sft:sft_new("../test/KL1_L2.sft",
KL1_L2, [], 2),
{ok, PidL2_1, _} = leveled_sst:sst_new("../test/KL1_L2.sst",
2,
KL1_L2,
undefined),
KL2_L2 = lists:sort(generate_randomkeys(8000, 250, 250)),
{ok, PidL2_2, _} = leveled_sft:sft_new("../test/KL2_L2.sft",
KL2_L2, [], 2),
{ok, PidL2_2, _} = leveled_sst:sst_new("../test/KL2_L2.sst",
2,
KL2_L2,
undefined),
KL3_L2 = lists:sort(generate_randomkeys(8000, 500, 250)),
{ok, PidL2_3, _} = leveled_sft:sft_new("../test/KL3_L2.sft",
KL3_L2, [], 2),
{ok, PidL2_3, _} = leveled_sst:sst_new("../test/KL3_L2.sst",
2,
KL3_L2,
undefined),
KL4_L2 = lists:sort(generate_randomkeys(8000, 750, 250)),
{ok, PidL2_4, _} = leveled_sft:sft_new("../test/KL4_L2.sft",
KL4_L2, [], 2),
Result = perform_merge({PidL1_1, "../test/KL1_L1.sft"},
{ok, PidL2_4, _} = leveled_sst:sst_new("../test/KL4_L2.sst",
2,
KL4_L2,
undefined),
Result = perform_merge({PidL1_1, "../test/KL1_L1.sst"},
[#manifest_entry{owner=PidL2_1},
#manifest_entry{owner=PidL2_2},
#manifest_entry{owner=PidL2_3},
@ -422,13 +437,13 @@ merge_file_test() ->
ok = find_randomkeys(Result, 50, KL3_L2),
io:format("Finding keys in KL4_L2~n"),
ok = find_randomkeys(Result, 50, KL4_L2),
leveled_sft:sft_clear(PidL1_1),
leveled_sft:sft_clear(PidL2_1),
leveled_sft:sft_clear(PidL2_2),
leveled_sft:sft_clear(PidL2_3),
leveled_sft:sft_clear(PidL2_4),
leveled_sst:sst_clear(PidL1_1),
leveled_sst:sst_clear(PidL2_1),
leveled_sst:sst_clear(PidL2_2),
leveled_sst:sst_clear(PidL2_3),
leveled_sst:sst_clear(PidL2_4),
lists:foreach(fun(ManEntry) ->
leveled_sft:sft_clear(ManEntry#manifest_entry.owner) end,
leveled_sst:sst_clear(ManEntry#manifest_entry.owner) end,
Result).
select_merge_candidates_test() ->

258
src/leveled_penciller.erl
View file

@ -22,17 +22,17 @@
%%
%% The Ledger is divided into many levels
%% - L0: New keys are received from the Bookie and merged into a single
%% gb_tree, until that tree is the size of a SFT file, and it is then persisted
%% as a SFT file at this level. L0 SFT files can be larger than the normal
%% gb_tree, until that tree is the size of a SST file, and it is then persisted
%% as a SST file at this level. L0 SST files can be larger than the normal
%% maximum size - so we don't have to consider problems of either having more
%% than one L0 file (and handling what happens on a crash between writing the
%% files when the second may have overlapping sequence numbers), or having a
%% remainder with overlapping in sequence numbers in memory after the file is
%% written. Once the persistence is completed, the L0 tree can be erased.
%% There can be only one SFT file at Level 0, so the work to merge that file
%% There can be only one SST file at Level 0, so the work to merge that file
%% to the lower level must be the highest priority, as otherwise writes to the
%% ledger will stall, when there is next a need to persist.
%% - L1 TO L7: May contain multiple processes managing non-overlapping sft
%% - L1 TO L7: May contain multiple processes managing non-overlapping SST
%% files. Compaction work should be sheduled if the number of files exceeds
%% the target size of the level, where the target size is 8 ^ n.
%%
@ -67,14 +67,14 @@
%% completed to merge the tree into the L0 tree.
%%
%% The Penciller MUST NOT accept a new PUSH if the Clerk has commenced the
%% conversion of the current L0 tree into a SFT file, but not completed this
%% conversion of the current L0 tree into a SST file, but not completed this
%% change. The Penciller in this case returns the push, and the Bookie should
%% continue to grow the cache before trying again.
%%
%% ---------- FETCH ----------
%%
%% On request to fetch a key the Penciller should look first in the in-memory
%% L0 tree, then look in the SFT files Level by Level (including level 0),
%% L0 tree, then look in the SST files Level by Level (including level 0),
%% consulting the Manifest to determine which file should be checked at each
%% level.
%%
@ -82,16 +82,16 @@
%%
%% Iterators may request a snapshot of the database. A snapshot is a cloned
%% Penciller seeded not from disk, but by the in-memory L0 gb_tree and the
%% in-memory manifest, allowing for direct reference for the SFT file processes.
%% in-memory manifest, allowing for direct reference for the SST file processes.
%%
%% Clones formed to support snapshots are registered by the Penciller, so that
%% SFT files valid at the point of the snapshot until either the iterator is
%% SST files valid at the point of the snapshot until either the iterator is
%% completed or has timed out.
%%
%% ---------- ON STARTUP ----------
%%
%% On Startup the Bookie with ask the Penciller to initiate the Ledger first.
%% To initiate the Ledger the must consult the manifest, and then start a SFT
%% To initiate the Ledger the must consult the manifest, and then start a SST
%% management process for each file in the manifest.
%%
%% The penciller should then try and read any Level 0 file which has the
@ -103,14 +103,14 @@
%% ---------- ON SHUTDOWN ----------
%%
%% On a controlled shutdown the Penciller should attempt to write any in-memory
%% ETS table to a L0 SFT file, assuming one is nto already pending. If one is
%% ETS table to a L0 SST file, assuming one is nto already pending. If one is
%% already pending then the Penciller will not persist this part of the Ledger.
%%
%% ---------- FOLDER STRUCTURE ----------
%%
%% The following folders are used by the Penciller
%% $ROOT/ledger/ledger_manifest/ - used for keeping manifest files
%% $ROOT/ledger/ledger_files/ - containing individual SFT files
%% $ROOT/ledger/ledger_files/ - containing individual SST files
%%
%% In larger stores there could be a large number of files in the ledger_file
%% folder - perhaps o(1000). It is assumed that modern file systems should
@ -120,7 +120,7 @@
%%
%% The Penciller can have one and only one Clerk for performing compaction
%% work. When the Clerk has requested and taken work, it should perform the
%5 compaction work starting the new SFT process to manage the new Ledger state
%5 compaction work starting the new SST process to manage the new Ledger state
%% and then write a new manifest file that represents that state with using
%% the next Manifest sequence number as the filename:
%% - nonzero_<ManifestSQN#>.pnd
@ -130,14 +130,14 @@
%%
%% On startup, the Penciller should look for the nonzero_*.crr file with the
%% highest such manifest sequence number. This will be started as the
%% manifest, together with any _0_0.sft file found at that Manifest SQN.
%% manifest, together with any _0_0.sst file found at that Manifest SQN.
%% Level zero files are not kept in the persisted manifest, and adding a L0
%% file does not advanced the Manifest SQN.
%%
%% The pace at which the store can accept updates will be dependent on the
%% speed at which the Penciller's Clerk can merge files at lower levels plus
%% the time it takes to merge from Level 0. As if a clerk has commenced
%% compaction work at a lower level and then immediately a L0 SFT file is
%% compaction work at a lower level and then immediately a L0 SST file is
%% written the Penciller will need to wait for this compaction work to
%% complete and the L0 file to be compacted before the ETS table can be
%% allowed to again reach capacity
@ -145,7 +145,7 @@
%% The writing of L0 files do not require the involvement of the clerk.
%% The L0 files are prompted directly by the penciller when the in-memory tree
%% has reached capacity. This places the penciller in a levelzero_pending
%% state, and in this state it must return new pushes. Once the SFT file has
%% state, and in this state it must return new pushes. Once the SST file has
%% been completed it will confirm completion to the penciller which can then
%% revert the levelzero_pending state, add the file to the manifest and clear
%% the current level zero in-memory view.
@ -399,10 +399,11 @@ handle_call({fetch_keys, StartKey, EndKey, AccFun, InitAcc, MaxKeys},
List ->
List
end,
SFTiter = initiate_rangequery_frommanifest(StartKey,
SSTiter = initiate_rangequery_frommanifest(StartKey,
EndKey,
State#state.manifest),
Acc = keyfolder({L0AsList, SFTiter},
io:format("SSTiter on query ~w~n", [SSTiter]),
Acc = keyfolder({L0AsList, SSTiter},
{StartKey, EndKey},
{AccFun, InitAcc},
MaxKeys),
@ -456,7 +457,7 @@ handle_cast({confirm_delete, FileName}, State=#state{is_snapshot=Snap})
{true, Pid} ->
UF1 = lists:keydelete(FileName, 1, State#state.unreferenced_files),
leveled_log:log("P0005", [FileName]),
ok = leveled_sft:sft_deleteconfirmed(Pid),
ok = leveled_sst:sst_deleteconfirmed(Pid),
{noreply, State#state{unreferenced_files=UF1}};
_ ->
{noreply, State}
@ -525,7 +526,7 @@ terminate(Reason, State) ->
leveled_log:log("P0009", []);
{false, [], _N} ->
L0Pid = roll_memory(UpdState, true),
ok = leveled_sft:sft_close(L0Pid);
ok = leveled_sst:sst_close(L0Pid);
StatusTuple ->
leveled_log:log("P0010", [StatusTuple])
end,
@ -533,7 +534,7 @@ terminate(Reason, State) ->
% Tidy shutdown of individual files
ok = close_files(0, UpdState#state.manifest),
lists:foreach(fun({_FN, Pid, _SN}) ->
ok = leveled_sft:sft_close(Pid) end,
ok = leveled_sst:sst_close(Pid) end,
UpdState#state.unreferenced_files),
leveled_log:log("P0011", []),
ok.
@ -608,14 +609,14 @@ start_from_file(PCLopts) ->
leveled_log:log("P0014", [MaxSQN]),
%% Find any L0 files
L0FN = filepath(RootPath, TopManSQN, new_merge_files) ++ "_0_0.sft",
L0FN = filepath(RootPath, TopManSQN, new_merge_files) ++ "_0_0.sst",
case filelib:is_file(L0FN) of
true ->
leveled_log:log("P0015", [L0FN]),
{ok,
L0Pid,
{L0StartKey, L0EndKey}} = leveled_sft:sft_open(L0FN),
L0SQN = leveled_sft:sft_getmaxsequencenumber(L0Pid),
{L0StartKey, L0EndKey}} = leveled_sst:sst_open(L0FN),
L0SQN = leveled_sst:sst_getmaxsequencenumber(L0Pid),
ManifestEntry = #manifest_entry{start_key=L0StartKey,
end_key=L0EndKey,
owner=L0Pid,
@ -696,7 +697,7 @@ update_levelzero(L0Size, {PushedTree, MinSQN, MaxSQN},
%% to an immediate return as expected. With 32K keys in the TreeList it could
%% take around 35-40ms.
%%
%% To avoid blocking this gen_server, the SFT file can request each item of the
%% To avoid blocking this gen_server, the SST file can request each item of the
%% cache one at a time.
%%
%% The Wait is set to false to use a cast when calling this in normal operation
@ -704,25 +705,22 @@ update_levelzero(L0Size, {PushedTree, MinSQN, MaxSQN},
roll_memory(State, false) ->
FileName = levelzero_filename(State),
leveled_log:log("P0019", [FileName]),
Opts = #sft_options{wait=false, penciller=self()},
leveled_log:log("P0019", [FileName, State#state.ledger_sqn]),
PCL = self(),
FetchFun = fun(Slot) -> pcl_fetchlevelzero(PCL, Slot) end,
% FetchFun = fun(Slot) -> lists:nth(Slot, State#state.levelzero_cache) end,
R = leveled_sft:sft_newfroml0cache(FileName,
R = leveled_sst:sst_newlevelzero(FileName,
length(State#state.levelzero_cache),
FetchFun,
Opts),
PCL,
State#state.ledger_sqn),
{ok, Constructor, _} = R,
Constructor;
roll_memory(State, true) ->
FileName = levelzero_filename(State),
Opts = #sft_options{wait=true},
FetchFun = fun(Slot) -> lists:nth(Slot, State#state.levelzero_cache) end,
R = leveled_sft:sft_newfroml0cache(FileName,
length(State#state.levelzero_cache),
FetchFun,
Opts),
KVList = leveled_pmem:to_list(length(State#state.levelzero_cache),
FetchFun),
R = leveled_sst:sst_new(FileName, 0, KVList, State#state.ledger_sqn),
{ok, Constructor, _} = R,
Constructor.
@ -753,7 +751,7 @@ fetch_mem(Key, Hash, Manifest, L0Cache, none) ->
L0Check = leveled_pmem:check_levelzero(Key, Hash, L0Cache),
case L0Check of
{false, not_found} ->
fetch(Key, Hash, Manifest, 0, fun timed_sft_get/3);
fetch(Key, Hash, Manifest, 0, fun timed_sst_get/3);
{true, KV} ->
{KV, 0}
end;
@ -762,7 +760,7 @@ fetch_mem(Key, Hash, Manifest, L0Cache, L0Index) ->
true ->
fetch_mem(Key, Hash, Manifest, L0Cache, none);
false ->
fetch(Key, Hash, Manifest, 0, fun timed_sft_get/3)
fetch(Key, Hash, Manifest, 0, fun timed_sst_get/3)
end.
fetch(_Key, _Hash, _Manifest, ?MAX_LEVELS + 1, _FetchFun) ->
@ -791,9 +789,9 @@ fetch(Key, Hash, Manifest, Level, FetchFun) ->
end
end.
timed_sft_get(PID, Key, Hash) ->
timed_sst_get(PID, Key, Hash) ->
SW = os:timestamp(),
R = leveled_sft:sft_get(PID, Key, Hash),
R = leveled_sst:sst_get(PID, Key, Hash),
T0 = timer:now_diff(os:timestamp(), SW),
case {T0, R} of
{T, R} when T < ?SLOW_FETCH ->
@ -880,7 +878,7 @@ close_files(?MAX_LEVELS - 1, _Manifest) ->
close_files(Level, Manifest) ->
LevelList = get_item(Level, Manifest, []),
lists:foreach(fun(F) ->
ok = leveled_sft:sft_close(F#manifest_entry.owner) end,
ok = leveled_sst:sst_close(F#manifest_entry.owner) end,
LevelList),
close_files(Level + 1, Manifest).
@ -897,8 +895,8 @@ open_all_filesinmanifest({Manifest, TopSQN}, Level) ->
%5 replace them
LvlR = lists:foldl(fun(F, {FL, FL_SQN}) ->
FN = F#manifest_entry.filename,
{ok, P, _Keys} = leveled_sft:sft_open(FN),
F_SQN = leveled_sft:sft_getmaxsequencenumber(P),
{ok, P, _Keys} = leveled_sst:sst_open(FN),
F_SQN = leveled_sst:sst_getmaxsequencenumber(P),
{lists:append(FL,
[F#manifest_entry{owner = P}]),
max(FL_SQN, F_SQN)}
@ -932,12 +930,13 @@ initiate_rangequery_frommanifest(StartKey, EndKey, Manifest) ->
C2 = leveled_codec:endkey_passed(EndKey,
M#manifest_entry.start_key),
not (C1 or C2) end,
lists:foldl(fun(L, AccL) ->
FoldFun =
fun(L, AccL) ->
Level = get_item(L, Manifest, []),
FL = lists:foldl(fun(M, Acc) ->
case CompareFun(M) of
true ->
Acc ++ [{next_file, M}];
Acc ++ [{next, M, StartKey}];
false ->
Acc
end end,
@ -948,8 +947,7 @@ initiate_rangequery_frommanifest(StartKey, EndKey, Manifest) ->
FL -> AccL ++ [{L, FL}]
end
end,
[],
lists:seq(0, ?MAX_LEVELS - 1)).
lists:foldl(FoldFun, [], lists:seq(0, ?MAX_LEVELS - 1)).
%% Looks to find the best choice for the next key across the levels (other
%% than in-memory table)
@ -960,22 +958,25 @@ find_nextkey(QueryArray, StartKey, EndKey) ->
find_nextkey(QueryArray,
0,
{null, null},
{fun leveled_sft:sft_getkvrange/4, StartKey, EndKey, 1}).
StartKey,
EndKey,
1).
find_nextkey(_QueryArray, LCnt, {null, null}, _QueryFunT)
find_nextkey(_QueryArray, LCnt, {null, null}, _StartKey, _EndKey, _Width)
when LCnt > ?MAX_LEVELS ->
% The array has been scanned wihtout finding a best key - must be
% exhausted - respond to indicate no more keys to be found by the
% iterator
no_more_keys;
find_nextkey(QueryArray, LCnt, {BKL, BestKV}, _QueryFunT)
find_nextkey(QueryArray, LCnt, {BKL, BestKV}, _StartKey, _EndKey, _Width)
when LCnt > ?MAX_LEVELS ->
% All levels have been scanned, so need to remove the best result from
% the array, and return that array along with the best key/sqn/status
% combination
{BKL, [BestKV|Tail]} = lists:keyfind(BKL, 1, QueryArray),
{lists:keyreplace(BKL, 1, QueryArray, {BKL, Tail}), BestKV};
find_nextkey(QueryArray, LCnt, {BestKeyLevel, BestKV}, QueryFunT) ->
find_nextkey(QueryArray, LCnt, {BestKeyLevel, BestKV},
StartKey, EndKey, Width) ->
% Get the next key at this level
{NextKey, RestOfKeys} = case lists:keyfind(LCnt, 1, QueryArray) of
false ->
@ -989,39 +990,46 @@ find_nextkey(QueryArray, LCnt, {BestKeyLevel, BestKV}, QueryFunT) ->
case {NextKey, BestKeyLevel, BestKV} of
{null, BKL, BKV} ->
% There is no key at this level - go to the next level
find_nextkey(QueryArray, LCnt + 1, {BKL, BKV}, QueryFunT);
{{next_file, ManifestEntry}, BKL, BKV} ->
find_nextkey(QueryArray,
LCnt + 1,
{BKL, BKV},
StartKey, EndKey, Width);
{{next, ManifestEntry, _SK}, BKL, BKV} ->
% The first key at this level is pointer to a file - need to query
% the file to expand this level out before proceeding
Owner = ManifestEntry#manifest_entry.owner,
{QueryFun, StartKey, EndKey, ScanSize} = QueryFunT,
QueryResult = QueryFun(Owner, StartKey, EndKey, ScanSize),
NewEntry = {LCnt, QueryResult ++ RestOfKeys},
Pointer = {next, Owner, StartKey, EndKey},
UpdList = leveled_sst:expand_list_by_pointer(Pointer,
RestOfKeys,
Width),
NewEntry = {LCnt, UpdList},
% Need to loop around at this level (LCnt) as we have not yet
% examined a real key at this level
find_nextkey(lists:keyreplace(LCnt, 1, QueryArray, NewEntry),
LCnt,
{BKL, BKV},
QueryFunT);
{{next, SFTpid, NewStartKey}, BKL, BKV} ->
StartKey, EndKey, Width);
{{pointer, SSTPid, Slot, PSK, PEK}, BKL, BKV} ->
% The first key at this level is pointer within a file - need to
% query the file to expand this level out before proceeding
{QueryFun, _StartKey, EndKey, ScanSize} = QueryFunT,
QueryResult = QueryFun(SFTpid, NewStartKey, EndKey, ScanSize),
NewEntry = {LCnt, QueryResult ++ RestOfKeys},
Pointer = {pointer, SSTPid, Slot, PSK, PEK},
UpdList = leveled_sst:expand_list_by_pointer(Pointer,
RestOfKeys,
Width),
NewEntry = {LCnt, UpdList},
% Need to loop around at this level (LCnt) as we have not yet
% examined a real key at this level
find_nextkey(lists:keyreplace(LCnt, 1, QueryArray, NewEntry),
LCnt,
{BKL, BKV},
QueryFunT);
StartKey, EndKey, Width);
{{Key, Val}, null, null} ->
% No best key set - so can assume that this key is the best key,
% and check the lower levels
find_nextkey(QueryArray,
LCnt + 1,
{LCnt, {Key, Val}},
QueryFunT);
StartKey, EndKey, Width);
{{Key, Val}, _BKL, {BestKey, _BestVal}} when Key < BestKey ->
% There is a real key and a best key to compare, and the real key
% at this level is before the best key, and so is now the new best
@ -1030,7 +1038,7 @@ find_nextkey(QueryArray, LCnt, {BestKeyLevel, BestKV}, QueryFunT) ->
find_nextkey(QueryArray,
LCnt + 1,
{LCnt, {Key, Val}},
QueryFunT);
StartKey, EndKey, Width);
{{Key, Val}, BKL, {BestKey, BestVal}} when Key == BestKey ->
SQN = leveled_codec:strip_to_seqonly({Key, Val}),
BestSQN = leveled_codec:strip_to_seqonly({BestKey, BestVal}),
@ -1041,7 +1049,7 @@ find_nextkey(QueryArray, LCnt, {BestKeyLevel, BestKV}, QueryFunT) ->
find_nextkey(lists:keyreplace(LCnt, 1, QueryArray, NewEntry),
LCnt + 1,
{BKL, {BestKey, BestVal}},
QueryFunT);
StartKey, EndKey, Width);
SQN > BestSQN ->
% There is a real key at the front of this level and it has
% a higher SQN than the best key, so we should use this as
@ -1056,29 +1064,32 @@ find_nextkey(QueryArray, LCnt, {BestKeyLevel, BestKV}, QueryFunT) ->
{BKL, BestTail}),
LCnt + 1,
{LCnt, {Key, Val}},
QueryFunT)
StartKey, EndKey, Width)
end;
{_, BKL, BKV} ->
% This is not the best key
find_nextkey(QueryArray, LCnt + 1, {BKL, BKV}, QueryFunT)
find_nextkey(QueryArray,
LCnt + 1,
{BKL, BKV},
StartKey, EndKey, Width)
end.
keyfolder(IMMiter, SFTiter, StartKey, EndKey, {AccFun, Acc}) ->
keyfolder({IMMiter, SFTiter}, {StartKey, EndKey}, {AccFun, Acc}, -1).
keyfolder(IMMiter, SSTiter, StartKey, EndKey, {AccFun, Acc}) ->
keyfolder({IMMiter, SSTiter}, {StartKey, EndKey}, {AccFun, Acc}, -1).
keyfolder(_Iterators, _KeyRange, {_AccFun, Acc}, MaxKeys) when MaxKeys == 0 ->
Acc;
keyfolder({[], SFTiter}, KeyRange, {AccFun, Acc}, MaxKeys) ->
keyfolder({[], SSTiter}, KeyRange, {AccFun, Acc}, MaxKeys) ->
{StartKey, EndKey} = KeyRange,
case find_nextkey(SFTiter, StartKey, EndKey) of
case find_nextkey(SSTiter, StartKey, EndKey) of
no_more_keys ->
Acc;
{NxSFTiter, {SFTKey, SFTVal}} ->
Acc1 = AccFun(SFTKey, SFTVal, Acc),
keyfolder({[], NxSFTiter}, KeyRange, {AccFun, Acc1}, MaxKeys - 1)
{NxSSTiter, {SSTKey, SSTVal}} ->
Acc1 = AccFun(SSTKey, SSTVal, Acc),
keyfolder({[], NxSSTiter}, KeyRange, {AccFun, Acc1}, MaxKeys - 1)
end;
keyfolder({[{IMMKey, IMMVal}|NxIMMiterator], SFTiterator}, KeyRange,
keyfolder({[{IMMKey, IMMVal}|NxIMMiterator], SSTiterator}, KeyRange,
{AccFun, Acc}, MaxKeys) ->
{StartKey, EndKey} = KeyRange,
case {IMMKey < StartKey, leveled_codec:endkey_passed(EndKey, IMMKey)} of
@ -1087,7 +1098,7 @@ keyfolder({[{IMMKey, IMMVal}|NxIMMiterator], SFTiterator}, KeyRange,
% Normally everything is pre-filterd, but the IMM iterator can
% be re-used and so may be behind the StartKey if the StartKey has
% advanced from the previous use
keyfolder({NxIMMiterator, SFTiterator},
keyfolder({NxIMMiterator, SSTiterator},
KeyRange,
{AccFun, Acc},
MaxKeys);
@ -1095,44 +1106,44 @@ keyfolder({[{IMMKey, IMMVal}|NxIMMiterator], SFTiterator}, KeyRange,
% There are no more keys in-range in the in-memory
% iterator, so take action as if this iterator is empty
% (see above)
keyfolder({[], SFTiterator},
keyfolder({[], SSTiterator},
KeyRange,
{AccFun, Acc},
MaxKeys);
{false, false} ->
case find_nextkey(SFTiterator, StartKey, EndKey) of
case find_nextkey(SSTiterator, StartKey, EndKey) of
no_more_keys ->
% No more keys in range in the persisted store, so use the
% in-memory KV as the next
Acc1 = AccFun(IMMKey, IMMVal, Acc),
keyfolder({NxIMMiterator, SFTiterator},
keyfolder({NxIMMiterator, SSTiterator},
KeyRange,
{AccFun, Acc1},
MaxKeys - 1);
{NxSFTiterator, {SFTKey, SFTVal}} ->
{NxSSTiterator, {SSTKey, SSTVal}} ->
% There is a next key, so need to know which is the
% next key between the two (and handle two keys
% with different sequence numbers).
case leveled_codec:key_dominates({IMMKey,
IMMVal},
{SFTKey,
SFTVal}) of
{SSTKey,
SSTVal}) of
left_hand_first ->
Acc1 = AccFun(IMMKey, IMMVal, Acc),
keyfolder({NxIMMiterator, SFTiterator},
keyfolder({NxIMMiterator, SSTiterator},
KeyRange,
{AccFun, Acc1},
MaxKeys - 1);
right_hand_first ->
Acc1 = AccFun(SFTKey, SFTVal, Acc),
Acc1 = AccFun(SSTKey, SSTVal, Acc),
keyfolder({[{IMMKey, IMMVal}|NxIMMiterator],
NxSFTiterator},
NxSSTiterator},
KeyRange,
{AccFun, Acc1},
MaxKeys - 1);
left_hand_dominant ->
Acc1 = AccFun(IMMKey, IMMVal, Acc),
keyfolder({NxIMMiterator, NxSFTiterator},
keyfolder({NxIMMiterator, NxSSTiterator},
KeyRange,
{AccFun, Acc1},
MaxKeys - 1)
@ -1286,6 +1297,27 @@ confirm_delete(Filename, UnreferencedFiles, RegisteredSnapshots) ->
-ifdef(TEST).
generate_randomkeys({Count, StartSQN}) ->
generate_randomkeys(Count, StartSQN, []);
generate_randomkeys(Count) ->
generate_randomkeys(Count, 0, []).
generate_randomkeys(0, _SQN, Acc) ->
lists:reverse(Acc);
generate_randomkeys(Count, SQN, Acc) ->
K = {o,
lists:concat(["Bucket", random:uniform(1024)]),
lists:concat(["Key", random:uniform(1024)]),
null},
RandKey = {K,
{SQN,
{active, infinity},
leveled_codec:magic_hash(K),
null}},
generate_randomkeys(Count - 1, SQN + 1, [RandKey|Acc]).
clean_testdir(RootPath) ->
clean_subdir(filepath(RootPath, manifest)),
clean_subdir(filepath(RootPath, files)).
@ -1332,8 +1364,8 @@ compaction_work_assessment_test() ->
?assertMatch([{1, Manifest3, 1}], WorkQ3).
confirm_delete_test() ->
Filename = 'test.sft',
UnreferencedFiles = [{'other.sft', dummy_owner, 15},
Filename = 'test.sst',
UnreferencedFiles = [{'other.sst', dummy_owner, 15},
{Filename, dummy_owner, 10}],
RegisteredIterators1 = [{dummy_pid, 16}, {dummy_pid, 12}],
R1 = confirm_delete(Filename, UnreferencedFiles, RegisteredIterators1),
@ -1376,20 +1408,20 @@ simple_server_test() ->
Key1_Pre = {{o,"Bucket0001", "Key0001", null},
{1, {active, infinity}, null}},
Key1 = add_missing_hash(Key1_Pre),
KL1 = leveled_sft:generate_randomkeys({1000, 2}),
KL1 = generate_randomkeys({1000, 2}),
Key2_Pre = {{o,"Bucket0002", "Key0002", null},
{1002, {active, infinity}, null}},
Key2 = add_missing_hash(Key2_Pre),
KL2 = leveled_sft:generate_randomkeys({900, 1003}),
KL2 = generate_randomkeys({900, 1003}),
% Keep below the max table size by having 900 not 1000
Key3_Pre = {{o,"Bucket0003", "Key0003", null},
{2003, {active, infinity}, null}},
Key3 = add_missing_hash(Key3_Pre),
KL3 = leveled_sft:generate_randomkeys({1000, 2004}),
KL3 = generate_randomkeys({1000, 2004}),
Key4_Pre = {{o,"Bucket0004", "Key0004", null},
{3004, {active, infinity}, null}},
Key4 = add_missing_hash(Key4_Pre),
KL4 = leveled_sft:generate_randomkeys({1000, 3005}),
KL4 = generate_randomkeys({1000, 3005}),
ok = maybe_pause_push(PCL, [Key1]),
?assertMatch(Key1, pcl_fetch(PCL, {o,"Bucket0001", "Key0001", null})),
ok = maybe_pause_push(PCL, KL1),
@ -1464,7 +1496,7 @@ simple_server_test() ->
Key1A_Pre = {{o,"Bucket0001", "Key0001", null},
{4005, {active, infinity}, null}},
Key1A = add_missing_hash(Key1A_Pre),
KL1A = leveled_sft:generate_randomkeys({2000, 4006}),
KL1A = generate_randomkeys({2000, 4006}),
ok = maybe_pause_push(PCLr, [Key1A]),
ok = maybe_pause_push(PCLr, KL1A),
?assertMatch(true, pcl_checksequencenumber(PclSnap,
@ -1528,17 +1560,16 @@ rangequery_manifest_test() ->
end_key={o, "Bucket1", "K996", null},
filename="Z6"}},
Man = [{1, [E1, E2, E3]}, {2, [E4, E5, E6]}],
R1 = initiate_rangequery_frommanifest({o, "Bucket1", "K711", null},
{o, "Bucket1", "K999", null},
Man),
?assertMatch([{1, [{next_file, E3}]},
{2, [{next_file, E5}, {next_file, E6}]}],
SK1 = {o, "Bucket1", "K711", null},
EK1 = {o, "Bucket1", "K999", null},
R1 = initiate_rangequery_frommanifest(SK1, EK1, Man),
?assertMatch([{1, [{next, E3, SK1}]},
{2, [{next, E5, SK1}, {next, E6, SK1}]}],
R1),
R2 = initiate_rangequery_frommanifest({i, "Bucket1", {"Idx1", "Fld8"}, null},
{i, "Bucket1", {"Idx1", "Fld8"}, null},
Man),
?assertMatch([{1, [{next_file, E1}]}, {2, [{next_file, E5}]}],
R2),
SK2 = {i, "Bucket1", {"Idx1", "Fld8"}, null},
EK2 = {i, "Bucket1", {"Idx1", "Fld8"}, null},
R2 = initiate_rangequery_frommanifest(SK2, EK2, Man),
?assertMatch([{1, [{next, E1, SK2}]}, {2, [{next, E5, SK2}]}], R2),
R3 = initiate_rangequery_frommanifest({i, "Bucket1", {"Idx0", "Fld8"}, null},
{i, "Bucket1", {"Idx0", "Fld9"}, null},
Man),
@ -1693,17 +1724,18 @@ foldwithimm_simple_test() ->
{{o, "Bucket1", "Key6"}, 7}], AccB).
create_file_test() ->
Filename = "../test/new_file.sft",
Filename = "../test/new_file.sst",
ok = file:write_file(Filename, term_to_binary("hello")),
KVL = lists:usort(leveled_sft:generate_randomkeys(10000)),
KVL = lists:usort(generate_randomkeys(10000)),
Tree = leveled_skiplist:from_list(KVL),
FetchFun = fun(Slot) -> lists:nth(Slot, [Tree]) end,
{ok,
SP,
noreply} = leveled_sft:sft_newfroml0cache(Filename,
noreply} = leveled_sst:sst_newlevelzero(Filename,
1,
FetchFun,
#sft_options{wait=false}),
undefined,
10000),
lists:foreach(fun(X) ->
case checkready(SP) of
timeout ->
@ -1716,9 +1748,9 @@ create_file_test() ->
io:format("StartKey ~w EndKey ~w~n", [StartKey, EndKey]),
?assertMatch({o, _, _, _}, StartKey),
?assertMatch({o, _, _, _}, EndKey),
?assertMatch("../test/new_file.sft", SrcFN),
ok = leveled_sft:sft_clear(SP),
{ok, Bin} = file:read_file("../test/new_file.sft.discarded"),
?assertMatch("../test/new_file.sst", SrcFN),
ok = leveled_sst:sst_clear(SP),
{ok, Bin} = file:read_file("../test/new_file.sst.discarded"),
?assertMatch("hello", binary_to_term(Bin)).
commit_manifest_test() ->
@ -1735,14 +1767,14 @@ commit_manifest_test() ->
ok = file:write_file(ManifestFP ++ "nonzero_1.pnd",
term_to_binary("dummy data")),
L1_0 = [{1, [#manifest_entry{filename="1.sft"}]}],
L1_0 = [{1, [#manifest_entry{filename="1.sst"}]}],
Resp_WI0 = Resp_WI#penciller_work{new_manifest=L1_0,
unreferenced_files=[]},
{ok, State0} = commit_manifest_change(Resp_WI0, State),
?assertMatch(1, State0#state.manifest_sqn),
?assertMatch([], get_item(0, State0#state.manifest, [])),
L0Entry = [#manifest_entry{filename="0.sft"}],
L0Entry = [#manifest_entry{filename="0.sst"}],
ManifestPlus = [{0, L0Entry}|State0#state.manifest],
NxtSent_WI = #penciller_work{next_sqn=2,
@ -1756,7 +1788,7 @@ commit_manifest_test() ->
ok = file:write_file(ManifestFP ++ "nonzero_2.pnd",
term_to_binary("dummy data")),
L2_0 = [#manifest_entry{filename="2.sft"}],
L2_0 = [#manifest_entry{filename="2.sst"}],
NxtResp_WI0 = NxtResp_WI#penciller_work{new_manifest=[{2, L2_0}],
unreferenced_files=[]},
{ok, State2} = commit_manifest_change(NxtResp_WI0, State1),
@ -1777,7 +1809,7 @@ badmanifest_test() ->
Key1_pre = {{o,"Bucket0001", "Key0001", null},
{1001, {active, infinity}, null}},
Key1 = add_missing_hash(Key1_pre),
KL1 = leveled_sft:generate_randomkeys({1000, 1}),
KL1 = generate_randomkeys({1000, 1}),
ok = maybe_pause_push(PCL, KL1 ++ [Key1]),
%% Added together, as split apart there will be a race between the close
@ -1798,7 +1830,7 @@ badmanifest_test() ->
checkready(Pid) ->
try
leveled_sft:sft_checkready(Pid)
leveled_sst:sst_checkready(Pid)
catch
exit:{timeout, _} ->
timeout

2024
src/leveled_sft.erl
View file

File diff suppressed because it is too large Load diff

20
src/leveled_skiplist.erl
View file

@ -29,6 +29,7 @@
lookup/2,
lookup/3,
key_above/2,
key_above_notequals/2,
empty/0,
empty/1,
size/1
@ -123,8 +124,15 @@ to_range(SkipList, Start, End) ->
to_list(SkipList) ->
to_list(element(2, SkipList), ?LIST_HEIGHT).
%% If a mark is found that matches the key, will return that mark
key_above(SkipList, Key) ->
key_above(element(2, SkipList), Key, ?LIST_HEIGHT).
TestFun = fun(Mark, K) -> Mark >= K end,
key_above(element(2, SkipList), Key, ?LIST_HEIGHT, TestFun).
%% If a mark is found that matches the key, will return the next mark
key_above_notequals(SkipList, Key) ->
TestFun = fun(Mark, K) -> Mark > K end,
key_above(element(2, SkipList), Key, ?LIST_HEIGHT, TestFun).
empty() ->
empty(false).
@ -321,11 +329,11 @@ sublist_above(SkipList, StartKey, Level, StartIncl) ->
sublist_above(SL, StartKey, Level - 1, StartIncl)
end.
key_above(SkipList, Key, 0) ->
key_above(SkipList, Key, 0, TestFun) ->
FindFun = fun({Mark, V}, Found) ->
case Found of
false ->
case Key =< Mark of
case TestFun(Mark, Key) of
true ->
{Mark, V};
false ->
@ -336,13 +344,13 @@ key_above(SkipList, Key, 0) ->
end
end,
lists:foldl(FindFun, false, SkipList);
key_above(SkipList, Key, Level) ->
key_above(SkipList, Key, Level, TestFun) ->
FindFun = fun({Mark, SL}, Found) ->
case Found of
false ->
case Key =< Mark of
case TestFun(Mark, Key) of
true ->
key_above(SL, Key, Level - 1);
key_above(SL, Key, Level - 1, TestFun);
false ->
false
end;

294
src/leveled_sst.erl
View file

@ -65,6 +65,7 @@
-define(LEVEL_BLOOM_SLOTS, [{0, 64}, {1, 48}, {default, 32}]).
-define(MERGE_SCANWIDTH, 16).
-define(DISCARD_EXT, ".discarded").
-define(DELETE_TIMEOUT, 10000).
-include_lib("eunit/include/eunit.hrl").
@ -74,21 +75,28 @@
handle_info/3,
terminate/3,
code_change/4,
starting/2,
starting/3,
reader/3]).
reader/3,
delete_pending/2,
delete_pending/3]).
-export([sst_new/3,
sst_new/5,
sst_newlevelzero/4,
-export([sst_new/4,
sst_new/6,
sst_newlevelzero/5,
sst_open/1,
sst_get/2,
sst_get/3,
sst_getkvrange/4,
sst_getslots/2,
sst_getmaxsequencenumber/1,
sst_setfordelete/2,
sst_clear/1,
sst_checkready/1,
sst_deleteconfirmed/1,
sst_close/1]).
-export([generate_randomkeys/1]).
-export([expand_list_by_pointer/3]).
-record(slot_index_value, {slot_id :: integer(),
@ -100,12 +108,14 @@
last_key :: tuple(),
index :: list(), % leveled_skiplist
bloom :: tuple(), % leveled_tinybloom
size :: integer()}).
size :: integer(),
max_sqn :: integer()}).
-record(state, {summary,
handle :: file:fd(),
sst_timings :: tuple(),
slot_lengths :: list(),
penciller :: pid(),
filename,
cache}).
@ -121,33 +131,42 @@ sst_open(Filename) ->
{ok, Pid, {SK, EK}}
end.
sst_new(Filename, Level, KVList) ->
sst_new(Filename, Level, KVList, MaxSQN) ->
{ok, Pid} = gen_fsm:start(?MODULE, [], []),
case gen_fsm:sync_send_event(Pid,
{sst_new, Filename, Level, KVList},
{sst_new,
Filename,
Level,
KVList,
MaxSQN},
infinity) of
{ok, {SK, EK}} ->
{ok, Pid, {SK, EK}}
end.
sst_new(Filename, KL1, KL2, IsBasement, Level) ->
sst_new(Filename, KL1, KL2, IsBasement, Level, MaxSQN) ->
{{Rem1, Rem2}, MergedList} = merge_lists(KL1, KL2, {IsBasement, Level}),
{ok, Pid} = gen_fsm:start(?MODULE, [], []),
case gen_fsm:sync_send_event(Pid,
{sst_new, Filename, Level, MergedList},
{sst_new,
Filename,
Level,
MergedList,
MaxSQN},
infinity) of
{ok, {SK, EK}} ->
{ok, Pid, {{Rem1, Rem2}, SK, EK}}
end.
sst_newlevelzero(Filename, Slots, FetchFun, Penciller) ->
sst_newlevelzero(Filename, Slots, FetchFun, Penciller, MaxSQN) ->
{ok, Pid} = gen_fsm:start(?MODULE, [], []),
gen_fsm:send_event(Pid,
{sst_newlevelzero,
Filename,
Slots,
FetchFun,
Penciller}),
Penciller,
MaxSQN}),
{ok, Pid, noreply}.
sst_get(Pid, LedgerKey) ->
@ -164,6 +183,24 @@ sst_getkvrange(Pid, StartKey, EndKey, ScanWidth) ->
sst_getslots(Pid, SlotList) ->
gen_fsm:sync_send_event(Pid, {get_slots, SlotList}, infinity).
sst_getmaxsequencenumber(Pid) ->
gen_fsm:sync_send_event(Pid, get_maxsequencenumber, infinity).
sst_setfordelete(Pid, Penciller) ->
gen_fsm:sync_send_event(Pid, {set_for_delete, Penciller}, infinity).
sst_clear(Pid) ->
gen_fsm:sync_send_event(Pid, {set_for_delete, false}, infinity),
gen_fsm:sync_send_event(Pid, close, 1000).
sst_deleteconfirmed(Pid) ->
gen_fsm:send_event(Pid, close).
sst_checkready(Pid) ->
%% Only used in test
gen_fsm:sync_send_event(Pid, background_complete, 100).
sst_close(Pid) ->
gen_fsm:sync_send_event(Pid, close, 2000).
@ -186,31 +223,48 @@ starting({sst_open, Filename}, _From, State) ->
{ok, {Summary#summary.first_key, Summary#summary.last_key}},
reader,
UpdState};
starting({sst_new, Filename, Level, KVList}, _From, State) ->
starting({sst_new, Filename, Level, KVList, MaxSQN}, _From, State) ->
{FirstKey, L, SlotIndex, AllHashes, SlotsBin} = build_all_slots(KVList),
SummaryBin = build_table_summary(SlotIndex, AllHashes, Level, FirstKey, L),
SummaryBin = build_table_summary(SlotIndex,
AllHashes,
Level,
FirstKey,
L,
MaxSQN),
ActualFilename = write_file(Filename, SummaryBin, SlotsBin),
UpdState = read_file(ActualFilename,
State#state{filename=ActualFilename}),
UpdState = read_file(ActualFilename, State),
Summary = UpdState#state.summary,
leveled_log:log("SST08", [ActualFilename, Level, Summary#summary.max_sqn]),
{reply,
{ok, {Summary#summary.first_key, Summary#summary.last_key}},
reader,
UpdState}.
starting({sst_newlevelzero, Filename, Slots, FetchFun, Penciller}, State) ->
starting({sst_newlevelzero, Filename, Slots, FetchFun, Penciller, MaxSQN},
State) ->
KVList = leveled_pmem:to_list(Slots, FetchFun),
{FirstKey, L, SlotIndex, AllHashes, SlotsBin} = build_all_slots(KVList),
SummaryBin = build_table_summary(SlotIndex, AllHashes, 0, FirstKey, L),
SummaryBin = build_table_summary(SlotIndex,
AllHashes,
0,
FirstKey,
L,
MaxSQN),
ActualFilename = write_file(Filename, SummaryBin, SlotsBin),
UpdState = read_file(ActualFilename,
State#state{filename=ActualFilename}),
UpdState = read_file(ActualFilename, State),
Summary = UpdState#state.summary,
leveled_log:log("SST08", [ActualFilename, 0, Summary#summary.max_sqn]),
case Penciller of
undefined ->
{next_state, reader, UpdState};
_ ->
leveled_penciller:pcl_confirml0complete(Penciller,
UpdState#state.filename,
Summary#summary.first_key,
Summary#summary.last_key),
{next_state, reader, UpdState}.
{next_state, reader, UpdState}
end.
reader({get_kv, LedgerKey, Hash}, _From, State) ->
SW = os:timestamp(),
@ -240,13 +294,70 @@ reader({get_slots, SlotList}, _From, State) ->
fun({SlotBin, SK, EK}, Acc) ->
Acc ++ trim_slot(SlotBin, SK, EK) end,
{reply, lists:foldl(FoldFun, [], SlotBins), reader, State};
reader(get_maxsequencenumber, _From, State) ->
Summary = State#state.summary,
{reply, Summary#summary.max_sqn, reader, State};
reader(print_timings, _From, State) ->
io:format(user, "Timings of ~w~n", [State#state.sst_timings]),
{reply, ok, reader, State#state{sst_timings = undefined}};
reader({set_for_delete, Penciller}, _From, State) ->
leveled_log:log("SST06", [State#state.filename]),
{reply,
ok,
delete_pending,
State#state{penciller=Penciller},
?DELETE_TIMEOUT};
reader(background_complete, _From, State) ->
Summary = State#state.summary,
{reply,
{ok,
State#state.filename,
Summary#summary.first_key,
Summary#summary.last_key},
reader,
State};
reader(close, _From, State) ->
ok = file:close(State#state.handle),
{stop, normal, ok, State}.
delete_pending({get_kv, LedgerKey, Hash}, _From, State) ->
{Result, Stage, SlotID} = fetch(LedgerKey, Hash, State),
case {Result, Stage} of
{not_present, slot_crc_wonky} ->
leveled_log:log("SST02", [State#state.filename, SlotID]),
{reply, Result, reader, State, ?DELETE_TIMEOUT};
{not_present, _} ->
{reply, Result, reader, State, ?DELETE_TIMEOUT};
{KV, slot_lookup_hit} ->
UpdCache = array:set(SlotID, KV, State#state.cache),
UpdState = State#state{cache = UpdCache},
{reply, Result, reader, UpdState, ?DELETE_TIMEOUT};
_ ->
{reply, Result, reader, State, ?DELETE_TIMEOUT}
end;
delete_pending({get_kvrange, StartKey, EndKey, ScanWidth}, _From, State) ->
{reply,
fetch_range(StartKey, EndKey, ScanWidth, State),
reader,
State,
?DELETE_TIMEOUT};
delete_pending(close, _From, State) ->
leveled_log:log("SST07", [State#state.filename]),
ok = file:close(State#state.handle),
ok = file:delete(State#state.filename),
{stop, normal, ok, State}.
delete_pending(timeout, State) ->
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("SST07", [State#state.filename]),
ok = file:close(State#state.handle),
ok = file:delete(State#state.filename),
{stop, normal, State}.
handle_sync_event(_Msg, _From, StateName, State) ->
{reply, undefined, StateName, State}.
@ -413,10 +524,14 @@ read_file(Filename, State) ->
SlotCount = length(SlotLengths),
SkipL = leveled_skiplist:from_sortedlist(Summary#summary.index),
UpdSummary = Summary#summary{index = SkipL},
leveled_log:log("SST03", [Filename, Summary#summary.size, SlotCount]),
leveled_log:log("SST03", [Filename,
Summary#summary.size,
SlotCount,
Summary#summary.max_sqn]),
State#state{summary = UpdSummary,
slot_lengths = SlotLengths,
handle = Handle,
filename = Filename,
cache = array:new({size, SlotCount + 1})}.
open_reader(Filename) ->
@ -426,7 +541,7 @@ open_reader(Filename) ->
{ok, SummaryBin} = file:pread(Handle, SlotsLength + 8, SummaryLength),
{Handle, SummaryBin}.
build_table_summary(SlotIndex, AllHashes, Level, FirstKey, L) ->
build_table_summary(SlotIndex, AllHashes, Level, FirstKey, L, MaxSQN) ->
BloomSlots =
case lists:keyfind(Level, 1, ?LEVEL_BLOOM_SLOTS) of
{Level, N} ->
@ -442,7 +557,8 @@ build_table_summary(SlotIndex, AllHashes, Level, FirstKey, L) ->
last_key = LastKey,
size = L,
index = lists:reverse(SlotIndex),
bloom = Bloom},
bloom = Bloom,
max_sqn = MaxSQN},
SummBin = term_to_binary(Summary, [{compressed, ?COMPRESSION_LEVEL}]),
SummCRC = erlang:crc32(SummBin),
<<SummCRC:32/integer, SummBin/binary>>.
@ -546,8 +662,13 @@ lookup_slots_int(StartKey, EndKey, SkipList) ->
EndKey ->
{L0, true, false};
_ ->
LTail = leveled_skiplist:key_above(SkipList, EndKey),
LTail = leveled_skiplist:key_above_notequals(SkipList, LastKey),
case LTail of
false ->
{L0, true, false};
_ ->
{L0 ++ [LTail], true, true}
end
end.
@ -751,13 +872,29 @@ key_dominates_expanded([H1|T1], [H2|T2], Level) ->
maybe_expand_pointer([]) ->
[];
maybe_expand_pointer([{pointer, SFTPid, Slot, StartKey, all}|Tail]) ->
maybe_expand_pointer([{pointer, SSTPid, Slot, StartKey, all}|Tail]) ->
expand_list_by_pointer({pointer, SSTPid, Slot, StartKey, all},
Tail,
?MERGE_SCANWIDTH);
maybe_expand_pointer([{next, SSTPid, StartKey}|Tail]) ->
expand_list_by_pointer({next, SSTPid, StartKey, all},
Tail,
?MERGE_SCANWIDTH);
maybe_expand_pointer(List) ->
List.
expand_list_by_pointer({pointer, SSTPid, Slot, StartKey, EndKey}, Tail, 1) ->
AccPointers = [{pointer, Slot, StartKey, EndKey}],
ExpPointers = leveled_sst:sst_getslots(SSTPid, AccPointers),
lists:append(ExpPointers, Tail);
expand_list_by_pointer({pointer, SSTPid, Slot, StartKey, all}, Tail, Width) ->
FoldFun =
fun(X, {Pointers, Remainder}) ->
case length(Pointers) of
L when L < ?MERGE_SCANWIDTH ->
L when L < Width ->
case X of
{pointer, SFTPid, S, SK, EK} ->
{pointer, SSTPid, S, SK, EK} ->
{Pointers ++ [{pointer, S, SK, EK}], Remainder};
_ ->
{Pointers, Remainder ++ [X]}
@ -768,16 +905,11 @@ maybe_expand_pointer([{pointer, SFTPid, Slot, StartKey, all}|Tail]) ->
end,
InitAcc = {[{pointer, Slot, StartKey, all}], []},
{AccPointers, AccTail} = lists:foldl(FoldFun, InitAcc, Tail),
SW = os:timestamp(),
ExpPointers = sst_getslots(SFTPid, AccPointers),
leveled_log:log_timer("SFT14", [SFTPid], SW),
ExpPointers = leveled_sst:sst_getslots(SSTPid, AccPointers),
lists:append(ExpPointers, AccTail);
maybe_expand_pointer([{next, SFTPid, StartKey}|Tail]) ->
ExpPointer = sst_getkvrange(SFTPid, StartKey, all, ?MERGE_SCANWIDTH),
maybe_expand_pointer(ExpPointer ++ Tail);
maybe_expand_pointer(List) ->
List.
expand_list_by_pointer({next, SSTPid, StartKey, EndKey}, Tail, Width) ->
ExpPointer = leveled_sst:sst_getkvrange(SSTPid, StartKey, EndKey, Width),
ExpPointer ++ Tail.
@ -787,13 +919,6 @@ maybe_expand_pointer(List) ->
-ifdef(TEST).
generate_randomkeys({Count, StartSQN}) ->
BucketNumber = random:uniform(1024),
generate_randomkeys(Count, StartSQN, [], BucketNumber, BucketNumber);
generate_randomkeys(Count) ->
BucketNumber = random:uniform(1024),
generate_randomkeys(Count, 0, [], BucketNumber, BucketNumber).
generate_randomkeys(Seqn, Count, BucketRangeLow, BucketRangeHigh) ->
generate_randomkeys(Seqn,
Count,
@ -834,8 +959,8 @@ merge_test() ->
KVL2 = lists:ukeysort(1, generate_randomkeys(1, N, 1, 20)),
KVL3 = lists:ukeymerge(1, KVL1, KVL2),
SW0 = os:timestamp(),
{ok, P1, {FK1, LK1}} = sst_new("../test/level1_src", 1, KVL1),
{ok, P2, {FK2, LK2}} = sst_new("../test/level2_src", 2, KVL2),
{ok, P1, {FK1, LK1}} = sst_new("../test/level1_src", 1, KVL1, 6000),
{ok, P2, {FK2, LK2}} = sst_new("../test/level2_src", 2, KVL2, 3000),
ExpFK1 = element(1, lists:nth(1, KVL1)),
ExpLK1 = element(1, lists:last(KVL1)),
ExpFK2 = element(1, lists:nth(1, KVL2)),
@ -850,7 +975,8 @@ merge_test() ->
ML1,
ML2,
false,
2),
2,
N * 2),
?assertMatch([], Rem1),
?assertMatch([], Rem2),
?assertMatch(true, FK3 == min(FK1, FK2)),
@ -915,7 +1041,8 @@ simple_slotbinsummary_test() ->
AllHashes,
2,
FirstKey,
length(KVList1)),
length(KVList1),
undefined),
Summary = read_table_summary(SummaryBin),
SummaryIndex = leveled_skiplist:from_sortedlist(Summary#summary.index),
FetchFun =
@ -945,7 +1072,10 @@ simple_persisted_test() ->
KVList1 = lists:ukeysort(1, KVList0),
[{FirstKey, _FV}|_Rest] = KVList1,
{LastKey, _LV} = lists:last(KVList1),
{ok, Pid, {FirstKey, LastKey}} = sst_new(Filename, 1, KVList1),
{ok, Pid, {FirstKey, LastKey}} = sst_new(Filename,
1,
KVList1,
length(KVList1)),
SW1 = os:timestamp(),
lists:foreach(fun({K, V}) ->
?assertMatch({K, V}, sst_get(Pid, K)),
@ -1014,7 +1144,71 @@ simple_persisted_test() ->
?assertMatch(SubKVListA1L, length(FetchedListB2)),
?assertMatch(SubKVListA1, FetchedListB2),
FetchListB3 = sst_getkvrange(Pid,
Eight000Key,
{o, null, null, null},
4),
FetchedListB3 = lists:foldl(FoldFun, [], FetchListB3),
SubKVListA3 = lists:nthtail(800 - 1, KVList1),
SubKVListA3L = length(SubKVListA3),
io:format("Length expected ~w~n", [SubKVListA3L]),
?assertMatch(SubKVListA3L, length(FetchedListB3)),
?assertMatch(SubKVListA3, FetchedListB3),
ok = sst_close(Pid),
ok = file:delete(Filename ++ ".sst").
key_dominates_test() ->
KV1 = {{o, "Bucket", "Key1", null}, {5, {active, infinity}, 0, []}},
KV2 = {{o, "Bucket", "Key3", null}, {6, {active, infinity}, 0, []}},
KV3 = {{o, "Bucket", "Key2", null}, {3, {active, infinity}, 0, []}},
KV4 = {{o, "Bucket", "Key4", null}, {7, {active, infinity}, 0, []}},
KV5 = {{o, "Bucket", "Key1", null}, {4, {active, infinity}, 0, []}},
KV6 = {{o, "Bucket", "Key1", null}, {99, {tomb, 999}, 0, []}},
KV7 = {{o, "Bucket", "Key1", null}, {99, tomb, 0, []}},
KL1 = [KV1, KV2],
KL2 = [KV3, KV4],
?assertMatch({{next_key, KV1}, [KV2], KL2},
key_dominates(KL1, KL2, {undefined, 1})),
?assertMatch({{next_key, KV1}, KL2, [KV2]},
key_dominates(KL2, KL1, {undefined, 1})),
?assertMatch({skipped_key, KL2, KL1},
key_dominates([KV5|KL2], KL1, {undefined, 1})),
?assertMatch({{next_key, KV1}, [KV2], []},
key_dominates(KL1, [], {undefined, 1})),
?assertMatch({skipped_key, [KV6|KL2], [KV2]},
key_dominates([KV6|KL2], KL1, {undefined, 1})),
?assertMatch({{next_key, KV6}, KL2, [KV2]},
key_dominates([KV6|KL2], [KV2], {undefined, 1})),
?assertMatch({skipped_key, [KV6|KL2], [KV2]},
key_dominates([KV6|KL2], KL1, {true, 1})),
?assertMatch({skipped_key, [KV6|KL2], [KV2]},
key_dominates([KV6|KL2], KL1, {true, 1000})),
?assertMatch({{next_key, KV6}, KL2, [KV2]},
key_dominates([KV6|KL2], [KV2], {true, 1})),
?assertMatch({skipped_key, KL2, [KV2]},
key_dominates([KV6|KL2], [KV2], {true, 1000})),
?assertMatch({skipped_key, [], []},
key_dominates([KV6], [], {true, 1000})),
?assertMatch({skipped_key, [], []},
key_dominates([], [KV6], {true, 1000})),
?assertMatch({{next_key, KV6}, [], []},
key_dominates([KV6], [], {true, 1})),
?assertMatch({{next_key, KV6}, [], []},
key_dominates([], [KV6], {true, 1})),
?assertMatch({skipped_key, [], []},
key_dominates([KV7], [], {true, 1})),
?assertMatch({skipped_key, [], []},
key_dominates([], [KV7], {true, 1})),
?assertMatch({skipped_key, [KV7|KL2], [KV2]},
key_dominates([KV7|KL2], KL1, {undefined, 1})),
?assertMatch({{next_key, KV7}, KL2, [KV2]},
key_dominates([KV7|KL2], [KV2], {undefined, 1})),
?assertMatch({skipped_key, [KV7|KL2], [KV2]},
key_dominates([KV7|KL2], KL1, {true, 1})),
?assertMatch({skipped_key, KL2, [KV2]},
key_dominates([KV7|KL2], [KV2], {true, 1})).
-endif.