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

Merge remote-tracking branch 'refs/remotes/origin/master' into mas-specs-i61b

Browse source
This commit is contained in:
martinsumner 2017-05-26 13:45:20 +01:00
commit f23072fb54
5 changed files with 112 additions and 32 deletions
Download patch file Download diff file Expand all files Collapse all files

47
src/leveled_bookie.erl
View file

@ -729,6 +729,7 @@ binary_bucketlist(State, Tag, {FoldBucketsFun, InitAcc}) ->
no_lookup),
Folder = fun() ->
BucketAcc = get_nextbucket(null,
null,
Tag,
LedgerSnapshot,
[]),
@ -739,26 +740,40 @@ binary_bucketlist(State, Tag, {FoldBucketsFun, InitAcc}) ->
end,
{async, Folder}.
get_nextbucket(NextBucket, Tag, LedgerSnapshot, BKList) ->
StartKey = leveled_codec:to_ledgerkey(NextBucket, null, Tag),
get_nextbucket(NextBucket, NextKey, Tag, LedgerSnapshot, BKList) ->
Now = leveled_codec:integer_now(),
StartKey = leveled_codec:to_ledgerkey(NextBucket, NextKey, Tag),
EndKey = leveled_codec:to_ledgerkey(null, null, Tag),
ExtractFun = fun(LK, _V, _Acc) -> leveled_codec:from_ledgerkey(LK) end,
BK = leveled_penciller:pcl_fetchnextkey(LedgerSnapshot,
StartKey,
EndKey,
ExtractFun,
null),
case BK of
ExtractFun =
fun(LK, V, _Acc) ->
{leveled_codec:from_ledgerkey(LK), V}
end,
R = leveled_penciller:pcl_fetchnextkey(LedgerSnapshot,
StartKey,
EndKey,
ExtractFun,
null),
case R of
null ->
leveled_log:log("B0008",[]),
BKList;
{B, K} when is_binary(B) ->
leveled_log:log("B0009",[B]),
get_nextbucket(<<B/binary, 0>>,
Tag,
LedgerSnapshot,
[{B, K}|BKList]);
NB ->
{{B, K}, V} when is_binary(B), is_binary(K) ->
case leveled_codec:is_active({B, K}, V, Now) of
true ->
leveled_log:log("B0009",[B]),
get_nextbucket(<<B/binary, 0>>,
null,
Tag,
LedgerSnapshot,
[{B, K}|BKList]);
false ->
get_nextbucket(B,
<<K/binary, 0>>,
Tag,
LedgerSnapshot,
BKList)
end;
{NB, _V} ->
leveled_log:log("B0010",[NB]),
[]
end.

11
src/leveled_iclerk.erl
View file

@ -80,7 +80,7 @@
handle_info/2,
terminate/2,
clerk_new/1,
clerk_compact/6,
clerk_compact/7,
clerk_hashtablecalc/3,
clerk_stop/1,
code_change/3]).
@ -124,14 +124,15 @@
clerk_new(InkerClerkOpts) ->
gen_server:start(?MODULE, [InkerClerkOpts], []).
clerk_compact(Pid, Checker, InitiateFun, FilterFun, Inker, Timeout) ->
clerk_compact(Pid, Checker, InitiateFun, CloseFun, FilterFun, Inker, TimeO) ->
gen_server:cast(Pid,
{compact,
Checker,
InitiateFun,
CloseFun,
FilterFun,
Inker,
Timeout}).
TimeO}).
clerk_hashtablecalc(HashTree, StartPos, CDBpid) ->
{ok, Clerk} = gen_server:start(?MODULE, [#iclerk_options{}], []),
@ -171,7 +172,7 @@ init([IClerkOpts]) ->
handle_call(_Msg, _From, State) ->
{reply, not_supported, State}.
handle_cast({compact, Checker, InitiateFun, FilterFun, Inker, _Timeout},
handle_cast({compact, Checker, InitiateFun, CloseFun, FilterFun, Inker, _TO},
State) ->
% Empty the waste folder
clear_waste(State),
@ -207,11 +208,13 @@ handle_cast({compact, Checker, InitiateFun, FilterFun, Inker, _Timeout},
update_inker(Inker,
ManifestSlice,
FilesToDelete),
ok = CloseFun(FilterServer),
{noreply, State}
end;
Score ->
leveled_log:log("IC003", [Score]),
ok = leveled_inker:ink_compactioncomplete(Inker),
ok = CloseFun(FilterServer),
{noreply, State}
end;
handle_cast({hashtable_calc, HashTree, StartPos, CDBpid}, State) ->

11
src/leveled_inker.erl
View file

@ -275,22 +275,25 @@ ink_loadpcl(Pid, MinSQN, FilterFun, Penciller) ->
%% required to reload the Ledger on startup).
ink_compactjournal(Pid, Bookie, Timeout) ->
CheckerInitiateFun = fun initiate_penciller_snapshot/1,
CheckerCloseFun = fun leveled_penciller:pcl_close/1,
CheckerFilterFun = fun leveled_penciller:pcl_checksequencenumber/3,
gen_server:call(Pid,
{compact,
Bookie,
CheckerInitiateFun,
CheckerCloseFun,
CheckerFilterFun,
Timeout},
infinity).
%% Allows the Checker to be overriden in test, use something other than a
%% penciller
ink_compactjournal(Pid, Checker, InitiateFun, FilterFun, Timeout) ->
ink_compactjournal(Pid, Checker, InitiateFun, CloseFun, FilterFun, Timeout) ->
gen_server:call(Pid,
{compact,
Checker,
InitiateFun,
CloseFun,
FilterFun,
Timeout},
infinity).
@ -427,12 +430,14 @@ handle_call(print_manifest, _From, State) ->
handle_call({compact,
Checker,
InitiateFun,
CloseFun,
FilterFun,
Timeout},
_From, State) ->
leveled_iclerk:clerk_compact(State#state.clerk,
Checker,
InitiateFun,
CloseFun,
FilterFun,
self(),
Timeout),
@ -839,7 +844,6 @@ initiate_penciller_snapshot(Bookie) ->
MaxSQN = leveled_penciller:pcl_getstartupsequencenumber(LedgerSnap),
{LedgerSnap, MaxSQN}.
%%%============================================================================
%%% Test
%%%============================================================================
@ -1001,6 +1005,7 @@ compact_journal_test() ->
ok = ink_compactjournal(Ink1,
Checker,
fun(X) -> {X, 55} end,
fun(_F) -> ok end,
fun(L, K, SQN) -> lists:member({SQN, K}, L) end,
5000),
timer:sleep(1000),
@ -1010,6 +1015,7 @@ compact_journal_test() ->
ok = ink_compactjournal(Ink1,
Checker2,
fun(X) -> {X, 55} end,
fun(_F) -> ok end,
fun(L, K, SQN) -> lists:member({SQN, K}, L) end,
5000),
timer:sleep(1000),
@ -1041,6 +1047,7 @@ empty_manifest_test() ->
ok = ink_compactjournal(Ink1,
[],
fun(X) -> {X, 55} end,
fun(_F) -> ok end,
CheckFun,
5000),
timer:sleep(1000),