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

Mas i389 rebuildledger (#390)

Browse source 
* Protect penciller from empty ledger cache updates

which may occur when loading the ledger from the journal, after the ledger has been cleared.

* Score caching and randomisation

The test allkeydelta_journal_multicompact can occasionally fail when a compaction doesn't happen, but then does the next loop.  Suspect this is as a result of score caching, randomisation of key grabs for scoring, plus jitter on size boundaries.

Modified test for predictability.

Plus formatting changes

* Avoid small batches

Avoid small batches due to large SQN gaps

* Rationalise tests

Two tests overlaps with the new, much broader, replace_everything/1 test.  Ported over any remaining checks of interest and dropped two tests.
This commit is contained in:
Martin Sumner 2023-01-18 11:44:02 +00:00
parent a033e280e6
commit a01c74f268
9 changed files with 358 additions and 320 deletions
Download patch file Download diff file Expand all files Collapse all files

102
src/leveled_bookie.erl
View file

@ -373,16 +373,21 @@
{monitor_loglist, list(leveled_monitor:log_type())}
].
-type load_item() ::
{leveled_codec:journal_key_tag()|null,
leveled_codec:ledger_key()|?DUMMY,
non_neg_integer(), any(), integer(),
leveled_codec:journal_keychanges()}.
-type initial_loadfun() ::
fun((leveled_codec:journal_key(),
any(),
non_neg_integer(),
{non_neg_integer(), non_neg_integer(), ledger_cache()},
fun((any()) -> {binary(), non_neg_integer()})) ->
{loop|stop,
{non_neg_integer(), non_neg_integer(), ledger_cache()}}).
{loop|stop, list(load_item())}).
-export_type([initial_loadfun/0]).
-export_type([initial_loadfun/0, ledger_cache/0]).
%%%============================================================================
%%% API
@ -1562,13 +1567,46 @@ empty_ledgercache() ->
#ledger_cache{mem = ets:new(empty, [ordered_set])}.
-spec push_to_penciller(pid(), ledger_cache()) -> ok.
-spec push_to_penciller(
pid(),
list(load_item()),
ledger_cache(),
leveled_codec:compaction_strategy())
-> ledger_cache().
%% @doc
%% The push to penciller must start as a tree to correctly de-duplicate
%% the list by order before becoming a de-duplicated list for loading
push_to_penciller(Penciller, LedgerCache) ->
push_to_penciller_loop(Penciller, loadqueue_ledgercache(LedgerCache)).
push_to_penciller(Penciller, LoadItemList, LedgerCache, ReloadStrategy) ->
UpdLedgerCache =
lists:foldl(
fun({InkTag, PK, SQN, Obj, IndexSpecs, ValSize}, AccLC) ->
Chngs =
case leveled_codec:get_tagstrategy(PK, ReloadStrategy) of
recalc ->
recalcfor_ledgercache(
InkTag, PK, SQN, Obj, ValSize, IndexSpecs,
AccLC, Penciller);
_ ->
preparefor_ledgercache(
InkTag, PK, SQN, Obj, ValSize, IndexSpecs)
end,
addto_ledgercache(Chngs, AccLC, loader)
end,
LedgerCache,
lists:reverse(LoadItemList)
),
case length(UpdLedgerCache#ledger_cache.load_queue) of
N when N > ?LOADING_BATCH ->
leveled_log:log(b0006, [UpdLedgerCache#ledger_cache.max_sqn]),
ok =
push_to_penciller_loop(
Penciller, loadqueue_ledgercache(UpdLedgerCache)),
empty_ledgercache();
_ ->
UpdLedgerCache
end.
-spec push_to_penciller_loop(pid(), ledger_cache()) -> ok.
push_to_penciller_loop(Penciller, LedgerCache) ->
case push_ledgercache(Penciller, LedgerCache) of
returned ->
@ -1686,21 +1724,21 @@ startup(InkerOpts, PencillerOpts) ->
LedgerSQN = leveled_penciller:pcl_getstartupsequencenumber(Penciller),
leveled_log:log(b0005, [LedgerSQN]),
ReloadStrategy = InkerOpts#inker_options.reload_strategy,
LoadFun = get_loadfun(ReloadStrategy, Penciller),
LoadFun = get_loadfun(),
BatchFun =
fun(BatchAcc, _Acc) ->
push_to_penciller(Penciller, BatchAcc)
fun(BatchAcc, Acc) ->
push_to_penciller(
Penciller, BatchAcc, Acc, ReloadStrategy)
end,
InitAccFun =
fun(FN, CurrentMinSQN) ->
leveled_log:log(i0014, [FN, CurrentMinSQN]),
empty_ledgercache()
[]
end,
ok = leveled_inker:ink_loadpcl(Inker,
LedgerSQN + 1,
LoadFun,
InitAccFun,
BatchFun),
FinalAcc =
leveled_inker:ink_loadpcl(
Inker, LedgerSQN + 1, LoadFun, InitAccFun, BatchFun),
ok = push_to_penciller_loop(Penciller, loadqueue_ledgercache(FinalAcc)),
ok = leveled_inker:ink_checksqn(Inker, LedgerSQN),
{Inker, Penciller}.
@ -2414,44 +2452,36 @@ maybe_withjitter(_CacheSize, _MaxCacheSize, _MaxCacheMult) ->
false.
-spec get_loadfun(
leveled_codec:compaction_strategy(), pid()) -> initial_loadfun().
-spec get_loadfun() -> initial_loadfun().
%% @doc
%% The LoadFun will be used by the Inker when walking across the Journal to
%% load the Penciller at startup.
get_loadfun(ReloadStrat, Penciller) ->
get_loadfun() ->
fun(KeyInJournal, ValueInJournal, _Pos, Acc0, ExtractFun) ->
{MinSQN, MaxSQN, LedgerCache} = Acc0,
{MinSQN, MaxSQN, LoadItems} = Acc0,
{SQN, InkTag, PK} = KeyInJournal,
case SQN of
SQN when SQN < MinSQN ->
{loop, Acc0};
SQN when SQN > MaxSQN ->
leveled_log:log(b0007, [MaxSQN, SQN]),
{stop, Acc0};
_ ->
{VBin, ValSize} = ExtractFun(ValueInJournal),
% VBin may already be a term
{Obj, IdxSpecs} = leveled_codec:split_inkvalue(VBin),
Chngs =
case leveled_codec:get_tagstrategy(PK, ReloadStrat) of
recalc ->
recalcfor_ledgercache(InkTag, PK, SQN,
Obj, ValSize, IdxSpecs,
LedgerCache,
Penciller);
_ ->
preparefor_ledgercache(InkTag, PK, SQN,
Obj, ValSize, IdxSpecs)
end,
case SQN of
MaxSQN ->
leveled_log:log(b0006, [SQN]),
LC0 = addto_ledgercache(Chngs, LedgerCache, loader),
{stop, {MinSQN, MaxSQN, LC0}};
{stop,
{MinSQN,
MaxSQN,
[{InkTag, PK, SQN, Obj, IdxSpecs, ValSize}
|LoadItems]}};
_ ->
LC0 = addto_ledgercache(Chngs, LedgerCache, loader),
{loop, {MinSQN, MaxSQN, LC0}}
{loop,
{MinSQN,
MaxSQN,
[{InkTag, PK, SQN, Obj, IdxSpecs, ValSize}
|LoadItems]}}
end
end
end.

5
src/leveled_cdb.erl
View file

@ -375,8 +375,9 @@ cdb_deletepending(Pid) ->
cdb_deletepending(Pid, ManSQN, Inker) ->
gen_fsm:send_event(Pid, {delete_pending, ManSQN, Inker}).
-spec cdb_scan(pid(), filter_fun(), any(), integer()|undefined) ->
{integer()|eof, any()}.
-spec cdb_scan(
pid(), filter_fun(), any(), integer()|undefined)
-> {integer()|eof, any()}.
%% @doc
%% cdb_scan returns {LastPosition, Acc}. Use LastPosition as StartPosiiton to
%% continue from that point (calling function has to protect against) double

16
src/leveled_inker.erl
View file

@ -133,7 +133,6 @@
-define(WASTE_FP, "waste").
-define(JOURNAL_FILEX, "cdb").
-define(PENDING_FILEX, "pnd").
-define(LOADING_BATCH, 1000).
-define(TEST_KC, {[], infinity}).
-record(state, {manifest = [] :: list(),
@ -337,11 +336,14 @@ ink_fold(Pid, MinSQN, FoldFuns, Acc) ->
{fold, MinSQN, FoldFuns, Acc, by_runner},
infinity).
-spec ink_loadpcl(pid(),
integer(),
leveled_bookie:initial_loadfun(),
fun((string(), non_neg_integer()) -> any()),
fun((any(), any()) -> ok)) -> ok.
-spec ink_loadpcl(
pid(),
integer(),
leveled_bookie:initial_loadfun(),
fun((string(), non_neg_integer()) -> any()),
fun((any(), leveled_bookie:ledger_cache())
-> leveled_bookie:ledger_cache()))
-> leveled_bookie:ledger_cache().
%%
%% Function to prompt load of the Ledger at startup. The Penciller should
%% have determined the lowest SQN not present in the Ledger, and the inker
@ -355,7 +357,7 @@ ink_loadpcl(Pid, MinSQN, LoadFun, InitAccFun, BatchFun) ->
{fold,
MinSQN,
{LoadFun, InitAccFun, BatchFun},
ok,
leveled_bookie:empty_ledgercache(),
as_ink},
infinity).

2
src/leveled_log.erl
View file

@ -54,8 +54,6 @@
{info, <<"LedgerSQN=~w at startup">>},
b0006 =>
{info, <<"Reached end of load batch with SQN ~w">>},
b0007 =>
{info, <<"Skipping as exceeded MaxSQN ~w with SQN ~w">>},
b0008 =>
{info, <<"Bucket list finds no more results">>},
b0009 =>

59
src/leveled_penciller.erl
View file

@ -686,26 +686,31 @@ handle_call({push_mem, {LedgerTable, PushedIdx, MinSQN, MaxSQN}},
false ->
leveled_tree:from_orderedset(LedgerTable, ?CACHE_TYPE)
end,
{UpdMaxSQN, NewL0Size, UpdL0Cache} =
leveled_pmem:add_to_cache(
case leveled_pmem:add_to_cache(
L0Size,
{PushedTree, MinSQN, MaxSQN},
State#state.ledger_sqn,
State#state.levelzero_cache),
UpdL0Index =
leveled_pmem:add_to_index(
PushedIdx,
State#state.levelzero_index,
length(State#state.levelzero_cache) + 1),
leveled_log:log_randomtimer(
p0031, [NewL0Size, true, true, MinSQN, MaxSQN], SW, 0.1),
{reply,
ok,
State#state{
levelzero_cache = UpdL0Cache,
levelzero_size = NewL0Size,
levelzero_index = UpdL0Index,
ledger_sqn = UpdMaxSQN}}
State#state.levelzero_cache,
true) of
empty_push ->
{reply, ok, State};
{UpdMaxSQN, NewL0Size, UpdL0Cache} ->
UpdL0Index =
leveled_pmem:add_to_index(
PushedIdx,
State#state.levelzero_index,
length(State#state.levelzero_cache) + 1),
leveled_log:log_randomtimer(
p0031,
[NewL0Size, true, true, MinSQN, MaxSQN], SW, 0.1),
{reply,
ok,
State#state{
levelzero_cache = UpdL0Cache,
levelzero_size = NewL0Size,
levelzero_index = UpdL0Index,
ledger_sqn = UpdMaxSQN}}
end
end;
handle_call({fetch, Key, Hash, UseL0Index}, _From, State) ->
L0Idx =
@ -836,10 +841,12 @@ handle_call({register_snapshot, Snapshot, Query, BookiesMem, LongRunning},
case Query of
no_lookup ->
{UpdMaxSQN, UpdSize, L0Cache} =
leveled_pmem:add_to_cache(State#state.levelzero_size,
{LM1Cache, MinSQN, MaxSQN},
State#state.ledger_sqn,
State#state.levelzero_cache),
leveled_pmem:add_to_cache(
State#state.levelzero_size,
{LM1Cache, MinSQN, MaxSQN},
State#state.ledger_sqn,
State#state.levelzero_cache,
false),
{#state{levelzero_cache = L0Cache,
ledger_sqn = UpdMaxSQN,
levelzero_size = UpdSize,
@ -865,10 +872,12 @@ handle_call({register_snapshot, Snapshot, Query, BookiesMem, LongRunning},
EndKey}};
undefined ->
{UpdMaxSQN, UpdSize, L0Cache} =
leveled_pmem:add_to_cache(State#state.levelzero_size,
{LM1Cache, MinSQN, MaxSQN},
State#state.ledger_sqn,
State#state.levelzero_cache),
leveled_pmem:add_to_cache(
State#state.levelzero_size,
{LM1Cache, MinSQN, MaxSQN},
State#state.ledger_sqn,
State#state.levelzero_cache,
false),
LM1Idx =
case BookieIdx of
empty_index ->

56
src/leveled_pmem.erl
View file

@ -30,7 +30,7 @@
-export([
prepare_for_index/2,
add_to_cache/4,
add_to_cache/5,
to_list/2,
check_levelzero/3,
check_levelzero/4,
@ -109,22 +109,30 @@ check_index(Hash, L0Index) ->
L0Index),
lists:reverse(Positions).
-spec add_to_cache(integer(),
{tuple(), integer(), integer()},
integer(),
list()) ->
{integer(), integer(), list()}.
-spec add_to_cache(
integer(),
{tuple(), integer(), integer()},
integer(),
list(),
boolean()) -> {integer(), integer(), list()}|empty_push.
%% @doc
%% The penciller's cache is a list of leveled_trees, this adds a new tree to
%% that cache, providing an update to the approximate size of the cache and
%% the Ledger's SQN.
add_to_cache(L0Size, {LevelMinus1, MinSQN, MaxSQN}, LedgerSQN, TreeList) ->
LM1Size = leveled_tree:tsize(LevelMinus1),
if
MinSQN >= LedgerSQN ->
{MaxSQN,
L0Size + LM1Size,
[LevelMinus1|TreeList]}
%% Updates to cache must set Writable to true if the update could generate a
%% Level 0 file - as this must guard against empty entries (which may lead to
%% an attempt to write an empty L0 file)
add_to_cache(L0Size, {LM1, MinSQN, MaxSQN}, LedgerSQN, TreeList, Writeable) ->
case {Writeable, leveled_tree:tsize(LM1)} of
{true, 0} ->
empty_push;
{_, LM1Size} ->
if
MinSQN >= LedgerSQN ->
{MaxSQN,
L0Size + LM1Size,
[LM1|TreeList]}
end
end.
-spec to_list(
@ -268,12 +276,12 @@ compare_method_test() ->
R = lists:foldl(fun(_X, {LedgerSQN, L0Size, L0TreeList}) ->
LM1 = generate_randomkeys(LedgerSQN + 1,
2000, 1, 500),
add_to_cache(L0Size,
{LM1,
LedgerSQN + 1,
LedgerSQN + 2000},
LedgerSQN,
L0TreeList)
add_to_cache(
L0Size,
{LM1, LedgerSQN + 1, LedgerSQN + 2000},
LedgerSQN,
L0TreeList,
true)
end,
{0, 0, []},
lists:seq(1, 16)),
@ -365,10 +373,12 @@ with_index_test2() ->
LM1Array = lists:foldl(IndexPrepareFun, new_index(), LM1),
LM1SL = leveled_tree:from_orderedlist(lists:ukeysort(1, LM1), ?CACHE_TYPE),
UpdL0Index = add_to_index(LM1Array, L0Idx, length(L0TreeList) + 1),
R = add_to_cache(L0Size,
{LM1SL, LedgerSQN + 1, LedgerSQN + 2000},
LedgerSQN,
L0TreeList),
R = add_to_cache(
L0Size,
{LM1SL, LedgerSQN + 1, LedgerSQN + 2000},
LedgerSQN,
L0TreeList,
true),
{R, UpdL0Index, lists:ukeymerge(1, LM1, SrcList)}
end,