aaeac7ba36
* Add eqwalizer and clear for codec & sst The eqwalizer errors highlighted the need in several places for type clarification. Within tests there are some issue where a type is assumed, and so ignore has been used to handle this rather than write more complex code to be explicit about the assumption. The handling of arrays isn't great by eqwalizer - to be specific about the content of array causes issues when initialising an array. Perhaps a type (map maybe) where one can be more explicit about types might be a better option (even if there is a minimal performance impact). The use of a ?TOMB_COUNT defined option complicated the code much more with eqwalizer. So for now, there is no developer option to disable ?TOMB_COUNT. Test fixes required where strings have been used for buckets/keys not binaries. The leveled_sst statem needs a different state record for starting when compared to other modes. The state record has been divided up to reflect this, to make type management easier. The impact on performance needs to be tested. * Update ct tests to support binary keys/buckets only * Eqwalizer for leveled_cdb and leveled_tictac As array is used in leveled_tictac - there is the same issue as with leveled_sst * Remove redundant indirection of leveled_rand A legacy of pre-20 OTP * Morde modules eqwalized ebloom/log/util/monitor * Eqwalize further modules elp eqwalize leveled_codec; elp eqwalize leveled_sst; elp eqwalize leveled_cdb; elp eqwalize leveled_tictac; elp eqwalize leveled_log; elp eqwalize leveled_monitor; elp eqwalize leveled_head; elp eqwalize leveled_ebloom; elp eqwalize leveled_iclerk All concurrently OK * Refactor unit tests to use binary() no string() in key Previously string() was allowed just to avoid having to change all these tests. Go through the pain now, as part of eqwalizing. * Add fixes for penciller, inker Add a new ?IS_DEF macro to replace =/= undefined. Now more explicit about primary, object and query keys * Further fixes Need to clarify functions used by runner - where keys , query keys and object keys are used * Further eqwalisation * Eqwalize leveled_pmanifest Also make implementation independent of choice of dict - i.e. one can save a manifest using dict for blooms/pending_deletions and then open a manifest with code that uses a different type. Allow for slow dict to be replaced with map. Would not be backwards compatible though, without further thought - i.e. if you upgrade then downgrade. Redundant code created by leveled_sst refactoring removed. * Fix backwards compatibility issues * Manifest Entry to belong to leveled_pmanifest There are two manifests - leveled_pmanifest and leveled_imanifest. Both have manifest_entry() type objects, but these types are different. To avoid confusion don't include the pmanifest manifest_entry() within the global include file - be specific that it belongs to the leveled_pmanifest module * Ignore elp file - large binary * Update src/leveled_pmem.erl Remove unnecessary empty list from type definition Co-authored-by: Thomas Arts <thomas.arts@quviq.com> --------- Co-authored-by: Thomas Arts <thomas.arts@quviq.com>
746 lines
28 KiB
Erlang
746 lines
28 KiB
Erlang
%% -------- RUNNER ---------
|
|
%%
|
|
%% A bookie's runner would traditionally allow remote actors to place bets
|
|
%% via the runner. In this case the runner will allow a remote actor to
|
|
%% have query access to the ledger or journal. Runners provide a snapshot of
|
|
%% the book for querying the backend.
|
|
%%
|
|
%% Runners implement the {async, Folder} within Riak backends - returning an
|
|
%% {async, Runner}. Runner is just a function that provides access to a
|
|
%% snapshot of the database to allow for a particular query. The
|
|
%% Runner may make the snapshot at the point it is called, or the snapshot can
|
|
%% be generated and encapsulated within the function (known as snap_prefold).
|
|
%%
|
|
%% Runners which view only the Ledger (the Penciller view of the state) may
|
|
%% have a CheckPresence boolean - which causes the function to perform a basic
|
|
%% check that the item is available in the Journal via the Inker as part of
|
|
%% the fold. This may be useful for anti-entropy folds
|
|
|
|
-module(leveled_runner).
|
|
|
|
-include("leveled.hrl").
|
|
|
|
-export([
|
|
bucket_sizestats/3,
|
|
bucket_list/4,
|
|
bucket_list/5,
|
|
index_query/3,
|
|
bucketkey_query/4,
|
|
bucketkey_query/6,
|
|
hashlist_query/3,
|
|
tictactree/5,
|
|
foldheads_allkeys/7,
|
|
foldobjects_allkeys/4,
|
|
foldheads_bybucket/8,
|
|
foldobjects_bybucket/4,
|
|
foldobjects_byindex/3
|
|
]).
|
|
|
|
-define(CHECKJOURNAL_PROB, 0.2).
|
|
|
|
-type key_range()
|
|
:: {leveled_codec:query_key(), leveled_codec:query_key()}.
|
|
-type foldacc() :: any().
|
|
% Can't currently be specific about what an acc might be
|
|
|
|
-type fold_objects_fun()
|
|
:: fun((leveled_codec:key(), leveled_codec:key(), any(), foldacc())
|
|
-> foldacc()).
|
|
-type fold_keys_fun()
|
|
:: fun((leveled_codec:key(), leveled_codec:key(), foldacc())
|
|
-> foldacc()).
|
|
-type fold_buckets_fun()
|
|
:: fun((leveled_codec:key(), foldacc()) -> foldacc()).
|
|
-type fold_filter_fun()
|
|
:: fun((leveled_codec:key(), leveled_codec:key()) -> accumulate|pass).
|
|
|
|
-type snap_fun()
|
|
:: fun(() -> {ok, pid(), pid()|null, fun(() -> ok)}).
|
|
-type runner_fun()
|
|
:: fun(() -> foldacc()).
|
|
-type objectacc_fun()
|
|
:: fun((leveled_codec:object_key(), any(), foldacc()) -> foldacc()).
|
|
-type mp()
|
|
:: {re_pattern, term(), term(), term(), term()}.
|
|
|
|
-export_type([fold_keys_fun/0, mp/0]).
|
|
|
|
%%%============================================================================
|
|
%%% External functions
|
|
%%%============================================================================
|
|
|
|
|
|
-spec bucket_sizestats(snap_fun(),leveled_codec:key(), leveled_codec:tag())
|
|
-> {async, runner_fun()}.
|
|
%% @doc
|
|
%% Fold over a bucket accumulating the count of objects and their total sizes
|
|
bucket_sizestats(SnapFun, Bucket, Tag) ->
|
|
StartKey = leveled_codec:to_querykey(Bucket, null, Tag),
|
|
EndKey = leveled_codec:to_querykey(Bucket, null, Tag),
|
|
AccFun = accumulate_size(),
|
|
Runner =
|
|
fun() ->
|
|
{ok, LedgerSnap, _JournalSnap, AfterFun} = SnapFun(),
|
|
Acc =
|
|
leveled_penciller:pcl_fetchkeys(
|
|
LedgerSnap, StartKey, EndKey, AccFun, {0, 0}, as_pcl),
|
|
AfterFun(),
|
|
Acc
|
|
end,
|
|
{async, Runner}.
|
|
|
|
-spec bucket_list(snap_fun(),
|
|
leveled_codec:tag(),
|
|
fold_buckets_fun(), foldacc()) -> {async, runner_fun()}.
|
|
%% @doc
|
|
%% List buckets for tag, assuming bucket names are all either binary, ascii
|
|
%% strings or integers
|
|
bucket_list(SnapFun, Tag, FoldBucketsFun, InitAcc) ->
|
|
bucket_list(SnapFun, Tag, FoldBucketsFun, InitAcc, -1).
|
|
|
|
-spec bucket_list(snap_fun(),
|
|
leveled_codec:tag(),
|
|
fold_buckets_fun(), foldacc(),
|
|
integer()) -> {async, runner_fun()}.
|
|
%% @doc
|
|
%% set Max Buckets to -1 to list all buckets, otherwise will only return
|
|
%% MaxBuckets (use 1 to confirm that there exists any bucket for a given Tag)
|
|
bucket_list(SnapFun, Tag, FoldBucketsFun, InitAcc, MaxBuckets) ->
|
|
Runner =
|
|
fun() ->
|
|
{ok, LedgerSnapshot, _JournalSnapshot, AfterFun} = SnapFun(),
|
|
BucketAcc =
|
|
get_nextbucket(
|
|
null, null, Tag, LedgerSnapshot, [], {0, MaxBuckets}),
|
|
FoldRunner =
|
|
fun() ->
|
|
lists:foldr(
|
|
fun({B, _K}, Acc) -> FoldBucketsFun(B, Acc) end,
|
|
InitAcc,
|
|
BucketAcc)
|
|
% Buckets in reverse alphabetical order so foldr
|
|
end,
|
|
% For this fold, the fold over the store is actually completed
|
|
% before results are passed to the FoldBucketsFun to be
|
|
% accumulated. Using a throw to exit the fold early will not
|
|
% in this case save significant time.
|
|
wrap_runner(FoldRunner, AfterFun)
|
|
end,
|
|
{async, Runner}.
|
|
|
|
-spec index_query(snap_fun(),
|
|
{leveled_codec:ledger_key(),
|
|
leveled_codec:ledger_key(),
|
|
{boolean(), undefined|mp()}},
|
|
{fold_keys_fun(), foldacc()})
|
|
-> {async, runner_fun()}.
|
|
%% @doc
|
|
%% Secondary index query
|
|
%% This has the special capability that it will expect a message to be thrown
|
|
%% during the query - and handle this without crashing the penciller snapshot
|
|
%% This allows for this query to be used with a max_results check in the
|
|
%% applictaion - and to throw a stop message to be caught by the worker
|
|
%% handling the runner. This behaviour will not prevent the snapshot from
|
|
%% closing neatly, allowing delete_pending files to be cleared without waiting
|
|
%% for a timeout
|
|
index_query(SnapFun, {StartKey, EndKey, TermHandling}, FoldAccT) ->
|
|
{FoldKeysFun, InitAcc} = FoldAccT,
|
|
|
|
Runner =
|
|
fun() ->
|
|
{ok, LedgerSnapshot, _JournalSnapshot, AfterFun} = SnapFun(),
|
|
Folder =
|
|
leveled_penciller:pcl_fetchkeys(
|
|
LedgerSnapshot,
|
|
StartKey,
|
|
EndKey,
|
|
leveled_codec:accumulate_index(TermHandling, FoldKeysFun),
|
|
InitAcc,
|
|
by_runner),
|
|
wrap_runner(Folder, AfterFun)
|
|
end,
|
|
{async, Runner}.
|
|
|
|
-spec bucketkey_query(snap_fun(),
|
|
leveled_codec:tag(),
|
|
leveled_codec:key()|null,
|
|
{leveled_codec:single_key()|null, leveled_codec:single_key()|null},
|
|
{fold_keys_fun(), foldacc()},
|
|
leveled_codec:regular_expression())
|
|
-> {async, runner_fun()}.
|
|
%% @doc
|
|
%% Fold over all keys in `KeyRange' under tag (restricted to a given bucket)
|
|
bucketkey_query(SnapFun, Tag, Bucket,
|
|
{StartKey, EndKey},
|
|
{FoldKeysFun, InitAcc},
|
|
TermRegex) ->
|
|
SK = leveled_codec:to_querykey(Bucket, StartKey, Tag),
|
|
EK = leveled_codec:to_querykey(Bucket, EndKey, Tag),
|
|
AccFun = accumulate_keys(FoldKeysFun, TermRegex),
|
|
Runner =
|
|
fun() ->
|
|
{ok, LedgerSnapshot, _JournalSnapshot, AfterFun} = SnapFun(),
|
|
Folder =
|
|
leveled_penciller:pcl_fetchkeys(
|
|
LedgerSnapshot, SK, EK, AccFun, InitAcc, by_runner),
|
|
wrap_runner(Folder, AfterFun)
|
|
end,
|
|
{async, Runner}.
|
|
|
|
-spec bucketkey_query(snap_fun(),
|
|
leveled_codec:tag(),
|
|
leveled_codec:key()|null,
|
|
{fold_keys_fun(), foldacc()}) -> {async, runner_fun()}.
|
|
%% @doc
|
|
%% Fold over all keys under tag (potentially restricted to a given bucket)
|
|
bucketkey_query(SnapFun, Tag, Bucket, FunAcc) ->
|
|
bucketkey_query(SnapFun, Tag, Bucket, {null, null}, FunAcc, undefined).
|
|
|
|
-spec hashlist_query(snap_fun(),
|
|
leveled_codec:tag(),
|
|
boolean()) -> {async, runner_fun()}.
|
|
%% @doc
|
|
%% Fold over the keys under a given Tag accumulating the hashes
|
|
hashlist_query(SnapFun, Tag, JournalCheck) ->
|
|
StartKey = leveled_codec:to_querykey(null, null, Tag),
|
|
EndKey = leveled_codec:to_querykey(null, null, Tag),
|
|
Runner =
|
|
fun() ->
|
|
{ok, LedgerSnapshot, JournalSnapshot, AfterFun} = SnapFun(),
|
|
AccFun = accumulate_hashes(JournalCheck, JournalSnapshot),
|
|
Acc =
|
|
leveled_penciller:pcl_fetchkeys(
|
|
LedgerSnapshot, StartKey, EndKey, AccFun, []),
|
|
AfterFun(),
|
|
Acc
|
|
end,
|
|
{async, Runner}.
|
|
|
|
-spec tictactree(
|
|
snap_fun(),
|
|
{leveled_codec:tag(), leveled_codec:key(), tuple()},
|
|
boolean(), leveled_tictac:tree_size(), fold_filter_fun())
|
|
-> {async, runner_fun()}.
|
|
%% @doc
|
|
%% Return a merkle tree from the fold, directly accessing hashes cached in the
|
|
%% metadata
|
|
tictactree(SnapFun, {Tag, Bucket, Query}, JournalCheck, TreeSize, Filter) ->
|
|
% Journal check can be used for object key folds to confirm that the
|
|
% object is still indexed within the journal
|
|
Tree = leveled_tictac:new_tree(temp, TreeSize),
|
|
Runner =
|
|
fun() ->
|
|
{ok, LedgerSnap, JournalSnap, AfterFun} = SnapFun(),
|
|
% The start key and end key will vary depending on whether the
|
|
% fold is to fold over an index or a key range
|
|
EnsureKeyBinaryFun =
|
|
fun(K, T) ->
|
|
case is_binary(K) of
|
|
true ->
|
|
{K, T};
|
|
false ->
|
|
{leveled_util:t2b(K), T}
|
|
end
|
|
end,
|
|
{StartKey, EndKey, ExtractFun} =
|
|
case Tag of
|
|
?IDX_TAG ->
|
|
{IdxFld, StartIdx, EndIdx} = Query,
|
|
KeyDefFun = fun leveled_codec:to_querykey/5,
|
|
{KeyDefFun(Bucket, null, ?IDX_TAG, IdxFld, StartIdx),
|
|
KeyDefFun(Bucket, null, ?IDX_TAG, IdxFld, EndIdx),
|
|
EnsureKeyBinaryFun};
|
|
_ ->
|
|
{StartOKey, EndOKey} = Query,
|
|
{leveled_codec:to_querykey(Bucket, StartOKey, Tag),
|
|
leveled_codec:to_querykey(Bucket, EndOKey, Tag),
|
|
fun(K, H) ->
|
|
V = {is_hash, H},
|
|
EnsureKeyBinaryFun(K, V)
|
|
end}
|
|
end,
|
|
AccFun =
|
|
accumulate_tree(Filter, JournalCheck, JournalSnap, ExtractFun),
|
|
Acc =
|
|
leveled_penciller:pcl_fetchkeys(
|
|
LedgerSnap, StartKey, EndKey, AccFun, Tree),
|
|
AfterFun(),
|
|
Acc
|
|
end,
|
|
{async, Runner}.
|
|
|
|
-spec foldheads_allkeys(snap_fun(), leveled_codec:tag(),
|
|
fold_objects_fun()|{fold_objects_fun(), foldacc()},
|
|
boolean(), false|list(integer()),
|
|
false|leveled_codec:lastmod_range(),
|
|
false|pos_integer()) -> {async, runner_fun()}.
|
|
%% @doc
|
|
%% Fold over all heads in the store for a given tag - applying the passed
|
|
%% function to each proxy object
|
|
foldheads_allkeys(SnapFun, Tag, FoldFun, JournalCheck,
|
|
SegmentList, LastModRange, MaxObjectCount) ->
|
|
StartKey = leveled_codec:to_querykey(null, null, Tag),
|
|
EndKey = leveled_codec:to_querykey(null, null, Tag),
|
|
foldobjects(SnapFun,
|
|
Tag,
|
|
[{StartKey, EndKey}],
|
|
FoldFun,
|
|
{true, JournalCheck},
|
|
SegmentList,
|
|
LastModRange,
|
|
MaxObjectCount).
|
|
|
|
-spec foldobjects_allkeys(snap_fun(),
|
|
leveled_codec:tag(),
|
|
fold_objects_fun()|{fold_objects_fun(), foldacc()},
|
|
key_order|sqn_order)
|
|
-> {async, runner_fun()}.
|
|
%% @doc
|
|
%% Fold over all objects for a given tag
|
|
foldobjects_allkeys(SnapFun, Tag, FoldFun, key_order) ->
|
|
StartKey = leveled_codec:to_querykey(null, null, Tag),
|
|
EndKey = leveled_codec:to_querykey(null, null, Tag),
|
|
foldobjects(SnapFun,
|
|
Tag,
|
|
[{StartKey, EndKey}],
|
|
FoldFun,
|
|
false,
|
|
false);
|
|
foldobjects_allkeys(SnapFun, Tag, FoldObjectsFun, sqn_order) ->
|
|
% Fold over the journal in order of receipt
|
|
{FoldFun, InitAcc} =
|
|
case is_tuple(FoldObjectsFun) of
|
|
true ->
|
|
% FoldObjectsFun is already a tuple with a Fold function and an
|
|
% initial accumulator
|
|
FoldObjectsFun;
|
|
false ->
|
|
% no initial accumulator passed, and so should be just a list
|
|
{FoldObjectsFun, []}
|
|
end,
|
|
|
|
FilterFun =
|
|
fun(JKey, JVal, _Pos, Acc, ExtractFun) ->
|
|
{SQN, InkTag, LedgerKey} = JKey,
|
|
case {InkTag, leveled_codec:from_ledgerkey(Tag, LedgerKey)} of
|
|
{?INKT_STND, {B, K}} ->
|
|
% Ignore tombstones and non-matching Tags and Key changes
|
|
% objects.
|
|
{MinSQN, MaxSQN, BatchAcc} = Acc,
|
|
case SQN of
|
|
SQN when SQN < MinSQN ->
|
|
{loop, Acc};
|
|
SQN when SQN > MaxSQN ->
|
|
{stop, Acc};
|
|
_ ->
|
|
{VBin, _VSize} = ExtractFun(JVal),
|
|
{Obj, _IdxSpecs} =
|
|
leveled_codec:revert_value_from_journal(VBin),
|
|
ToLoop =
|
|
case SQN of
|
|
MaxSQN -> stop;
|
|
_ -> loop
|
|
end,
|
|
{ToLoop,
|
|
{MinSQN, MaxSQN, [{B, K, SQN, Obj}|BatchAcc]}}
|
|
end;
|
|
_ ->
|
|
{loop, Acc}
|
|
end
|
|
end,
|
|
|
|
InitAccFun = fun(_FN, _SQN) -> [] end,
|
|
|
|
Folder =
|
|
fun() ->
|
|
{ok, LedgerSnapshot, JournalSnapshot, AfterFun} =
|
|
case SnapFun() of
|
|
{ok, LS, JS, AF} when is_pid(JS) ->
|
|
{ok, LS, JS, AF}
|
|
end,
|
|
{ok, JournalSQN} =
|
|
leveled_inker:ink_getjournalsqn(JournalSnapshot),
|
|
IsValidFun =
|
|
fun(Bucket, Key, SQN) ->
|
|
LedgerKey = leveled_codec:to_objectkey(Bucket, Key, Tag),
|
|
CheckSQN =
|
|
leveled_penciller:pcl_checksequencenumber(
|
|
LedgerSnapshot, LedgerKey, SQN),
|
|
% Need to check that we have not folded past the point
|
|
% at which the snapshot was taken
|
|
(JournalSQN >= SQN) and (CheckSQN == current)
|
|
end,
|
|
|
|
BatchFoldFun =
|
|
fun(BatchAcc, ObjAcc) ->
|
|
ObjFun =
|
|
fun({B, K, SQN, Obj}, Acc) ->
|
|
case IsValidFun(B, K, SQN) of
|
|
true ->
|
|
FoldFun(B, K, Obj, Acc);
|
|
false ->
|
|
Acc
|
|
end
|
|
end,
|
|
leveled_log:log(r0001, [length(BatchAcc)]),
|
|
lists:foldr(ObjFun, ObjAcc, BatchAcc)
|
|
end,
|
|
|
|
InkFolder =
|
|
leveled_inker:ink_fold(
|
|
JournalSnapshot,
|
|
0,
|
|
{FilterFun, InitAccFun, BatchFoldFun},
|
|
InitAcc),
|
|
wrap_runner(InkFolder, AfterFun)
|
|
end,
|
|
{async, Folder}.
|
|
|
|
|
|
-spec foldobjects_bybucket(snap_fun(),
|
|
leveled_codec:tag(),
|
|
list(key_range()),
|
|
fold_objects_fun()|{fold_objects_fun(), foldacc()})
|
|
-> {async, runner_fun()}.
|
|
%% @doc
|
|
%% Fold over all objects within a given key range in a bucket
|
|
foldobjects_bybucket(SnapFun, Tag, KeyRanges, FoldFun) ->
|
|
foldobjects(
|
|
SnapFun, Tag, KeyRanges, FoldFun, false, false).
|
|
|
|
-spec foldheads_bybucket(snap_fun(),
|
|
leveled_codec:tag(),
|
|
list(key_range()),
|
|
fold_objects_fun()|{fold_objects_fun(), foldacc()},
|
|
boolean(),
|
|
false|list(integer()),
|
|
false|leveled_codec:lastmod_range(),
|
|
false|pos_integer())
|
|
-> {async, runner_fun()}.
|
|
%% @doc
|
|
%% Fold over all object metadata within a given key range in a bucket
|
|
foldheads_bybucket(SnapFun,
|
|
Tag,
|
|
KeyRanges,
|
|
FoldFun,
|
|
JournalCheck,
|
|
SegmentList, LastModRange, MaxObjectCount) ->
|
|
foldobjects(SnapFun,
|
|
Tag,
|
|
KeyRanges,
|
|
FoldFun,
|
|
{true, JournalCheck},
|
|
SegmentList,
|
|
LastModRange,
|
|
MaxObjectCount).
|
|
|
|
-spec foldobjects_byindex(snap_fun(),
|
|
tuple(),
|
|
fold_objects_fun()|{fold_objects_fun(), foldacc()})
|
|
-> {async, runner_fun()}.
|
|
%% @doc
|
|
%% Folds over an index, fetching the objects associated with the keys returned
|
|
%% and passing those objects into the fold function
|
|
foldobjects_byindex(SnapFun, {Tag, Bucket, Field, FromTerm, ToTerm}, FoldFun) ->
|
|
StartKey =
|
|
leveled_codec:to_querykey(Bucket, null, ?IDX_TAG, Field, FromTerm),
|
|
EndKey =
|
|
leveled_codec:to_querykey(Bucket, null, ?IDX_TAG, Field, ToTerm),
|
|
foldobjects(SnapFun,
|
|
Tag,
|
|
[{StartKey, EndKey}],
|
|
FoldFun,
|
|
false,
|
|
false).
|
|
|
|
|
|
|
|
%%%============================================================================
|
|
%%% Internal functions
|
|
%%%============================================================================
|
|
|
|
get_nextbucket(_NextB, _NextK, _Tag, _LS, BKList, {Limit, Limit}) ->
|
|
lists:reverse(BKList);
|
|
get_nextbucket(NextBucket, NextKey, Tag, LedgerSnapshot, BKList, {C, L}) ->
|
|
StartKey = leveled_codec:to_querykey(NextBucket, NextKey, Tag),
|
|
EndKey = leveled_codec:to_querykey(null, null, Tag),
|
|
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
|
|
{1, null} ->
|
|
leveled_log:log(b0008,[]),
|
|
BKList;
|
|
{0, {{B, K}, _V}} when is_binary(B); is_tuple(B) ->
|
|
leveled_log:log(b0009,[B]),
|
|
get_nextbucket(
|
|
leveled_codec:next_key(B),
|
|
null,
|
|
Tag,
|
|
LedgerSnapshot,
|
|
[{B, K}|BKList],
|
|
{C + 1, L}
|
|
)
|
|
end.
|
|
|
|
|
|
-spec foldobjects(
|
|
snap_fun(),
|
|
atom(),
|
|
list(),
|
|
fold_objects_fun()|{fold_objects_fun(), foldacc()},
|
|
false|{true, boolean()}, false|list(integer()))
|
|
-> {async, runner_fun()}.
|
|
foldobjects(SnapFun, Tag, KeyRanges, FoldObjFun, DeferredFetch, SegmentList) ->
|
|
foldobjects(SnapFun, Tag, KeyRanges,
|
|
FoldObjFun, DeferredFetch, SegmentList, false, false).
|
|
|
|
-spec foldobjects(snap_fun(), atom(), list(),
|
|
fold_objects_fun()|{fold_objects_fun(), foldacc()},
|
|
false|{true, boolean()},
|
|
false|list(integer()),
|
|
false|leveled_codec:lastmod_range(),
|
|
false|pos_integer()) -> {async, runner_fun()}.
|
|
%% @doc
|
|
%% The object folder should be passed DeferredFetch.
|
|
%% DeferredFetch can either be false (which will return to the fold function
|
|
%% the full object), or {true, CheckPresence} - in which case a proxy object
|
|
%% will be created that if understood by the fold function will allow the fold
|
|
%% function to work on the head of the object, and defer fetching the body in
|
|
%% case such a fetch is unecessary.
|
|
foldobjects(SnapFun, Tag, KeyRanges, FoldObjFun, DeferredFetch,
|
|
SegmentList, LastModRange, MaxObjectCount) ->
|
|
{FoldFun, InitAcc} =
|
|
case is_tuple(FoldObjFun) of
|
|
true ->
|
|
% FoldObjectsFun is already a tuple with a Fold function and an
|
|
% initial accumulator
|
|
FoldObjFun;
|
|
false ->
|
|
% no initial accumulator passed, and so should be just a list
|
|
{FoldObjFun, []}
|
|
end,
|
|
{LimitByCount, InitAcc0} =
|
|
case MaxObjectCount of
|
|
false ->
|
|
{false, InitAcc};
|
|
MOC when is_integer(MOC) ->
|
|
{true, {MOC, InitAcc}}
|
|
end,
|
|
|
|
Folder =
|
|
fun() ->
|
|
{ok, LedgerSnapshot, JournalSnapshot, AfterFun} = SnapFun(),
|
|
AccFun =
|
|
accumulate_objects(
|
|
FoldFun, JournalSnapshot, Tag, DeferredFetch),
|
|
FoldFunGen =
|
|
fun({StartKey, EndKey}, FoldAcc) ->
|
|
leveled_penciller:pcl_fetchkeysbysegment(
|
|
LedgerSnapshot,
|
|
StartKey,
|
|
EndKey,
|
|
AccFun,
|
|
FoldAcc,
|
|
SegmentList,
|
|
LastModRange,
|
|
LimitByCount
|
|
)
|
|
end,
|
|
ListFoldFun =
|
|
fun(KeyRange, Acc) ->
|
|
Folder = FoldFunGen(KeyRange, Acc),
|
|
Folder()
|
|
end,
|
|
FolderToWrap =
|
|
fun() -> lists:foldl(ListFoldFun, InitAcc0, KeyRanges) end,
|
|
wrap_runner(FolderToWrap, AfterFun)
|
|
end,
|
|
{async, Folder}.
|
|
|
|
|
|
accumulate_size() ->
|
|
AccFun =
|
|
fun(Key, Value, {Size, Count}) ->
|
|
{Size + leveled_codec:get_size(Key, Value), Count + 1}
|
|
end,
|
|
AccFun.
|
|
|
|
accumulate_hashes(JournalCheck, InkerClone) ->
|
|
AddKeyFun =
|
|
fun(B, K, H, Acc) ->
|
|
[{B, K, H}|Acc]
|
|
end,
|
|
get_hashaccumulator(JournalCheck, InkerClone, AddKeyFun).
|
|
|
|
accumulate_tree(FilterFun, JournalCheck, InkerClone, HashFun) ->
|
|
AddKeyFun =
|
|
fun(B, K, H, Tree) ->
|
|
case FilterFun(B, K) of
|
|
accumulate ->
|
|
leveled_tictac:add_kv(Tree, K, H, HashFun);
|
|
pass ->
|
|
Tree
|
|
end
|
|
end,
|
|
get_hashaccumulator(JournalCheck, InkerClone, AddKeyFun).
|
|
|
|
get_hashaccumulator(JournalCheck, InkerClone, AddKeyFun) ->
|
|
AccFun =
|
|
fun(LK, V, Acc) ->
|
|
{B, K, H} = leveled_codec:get_keyandobjhash(LK, V),
|
|
Check = rand:uniform() < ?CHECKJOURNAL_PROB,
|
|
case JournalCheck and Check of
|
|
true ->
|
|
case check_presence(LK, V, InkerClone) of
|
|
true ->
|
|
AddKeyFun(B, K, H, Acc);
|
|
false ->
|
|
Acc
|
|
end;
|
|
_ ->
|
|
AddKeyFun(B, K, H, Acc)
|
|
end
|
|
end,
|
|
AccFun.
|
|
|
|
-spec accumulate_objects
|
|
(fold_objects_fun(), pid(), leveled_head:object_tag(), false|{true, boolean()})
|
|
-> objectacc_fun();
|
|
(fold_objects_fun(), null, leveled_head:headonly_tag(), {true, false})
|
|
-> objectacc_fun().
|
|
accumulate_objects(FoldObjectsFun, InkerClone, Tag, DeferredFetch) ->
|
|
AccFun =
|
|
fun(LK, V, Acc) ->
|
|
% The function takes the Ledger Key and the value from the
|
|
% ledger (with the value being the object metadata)
|
|
%
|
|
% Need to check if this is an active object (so TTL has not
|
|
% expired).
|
|
% If this is a deferred_fetch (i.e. the fold is a fold_heads not
|
|
% a fold_objects), then a metadata object needs to be built to be
|
|
% returned - but a quick check that Key is present in the Journal
|
|
% is made first
|
|
{SQN, _St, _MH, MD} =
|
|
leveled_codec:striphead_to_v1details(V),
|
|
{B, K} =
|
|
case leveled_codec:from_ledgerkey(LK) of
|
|
{B0, K0} ->
|
|
{B0, K0};
|
|
{B0, K0, _T0} ->
|
|
{B0, K0}
|
|
end,
|
|
JK = {leveled_codec:to_objectkey(B, K, Tag), SQN},
|
|
case DeferredFetch of
|
|
{true, JournalCheck} when MD =/= null ->
|
|
ProxyObj =
|
|
leveled_codec:return_proxy(Tag, MD, InkerClone, JK),
|
|
case {JournalCheck, InkerClone} of
|
|
{true, InkerClone} when is_pid(InkerClone) ->
|
|
InJournal =
|
|
leveled_inker:ink_keycheck(
|
|
InkerClone, LK, SQN),
|
|
case InJournal of
|
|
probably ->
|
|
FoldObjectsFun(B, K, ProxyObj, Acc);
|
|
missing ->
|
|
Acc
|
|
end;
|
|
{false, _} ->
|
|
FoldObjectsFun(B, K, ProxyObj, Acc)
|
|
end;
|
|
false ->
|
|
R = leveled_bookie:fetch_value(InkerClone, JK),
|
|
case R of
|
|
not_present ->
|
|
Acc;
|
|
Value ->
|
|
FoldObjectsFun(B, K, Value, Acc)
|
|
end
|
|
end
|
|
end,
|
|
AccFun.
|
|
|
|
|
|
check_presence(Key, Value, InkerClone) ->
|
|
{LedgerKey, SQN} = leveled_codec:strip_to_keyseqonly({Key, Value}),
|
|
case leveled_inker:ink_keycheck(InkerClone, LedgerKey, SQN) of
|
|
probably ->
|
|
true;
|
|
missing ->
|
|
false
|
|
end.
|
|
|
|
accumulate_keys(FoldKeysFun, undefined) ->
|
|
fun(Key, _Value, Acc) ->
|
|
{B, K} = leveled_codec:from_ledgerkey(Key),
|
|
FoldKeysFun(B, K, Acc)
|
|
end;
|
|
accumulate_keys(FoldKeysFun, TermRegex) ->
|
|
fun(Key, _Value, Acc) ->
|
|
{B, K} = leveled_codec:from_ledgerkey(Key),
|
|
case re:run(K, TermRegex) of
|
|
nomatch ->
|
|
Acc;
|
|
_ ->
|
|
FoldKeysFun(B, K, Acc)
|
|
end
|
|
end.
|
|
|
|
-spec wrap_runner(fun(), fun()) -> any().
|
|
%% @doc
|
|
%% Allow things to be thrown in folds, and ensure clean-up action is still
|
|
%% undertaken if they are.
|
|
%%
|
|
%% It is assumed this is only used at present by index queries and key folds,
|
|
%% but the wrap could be applied more generally with further work
|
|
wrap_runner(FoldAction, AfterAction) ->
|
|
try FoldAction()
|
|
catch throw:Throw ->
|
|
throw(Throw)
|
|
after AfterAction()
|
|
end.
|
|
|
|
|
|
%%%============================================================================
|
|
%%% Test
|
|
%%%============================================================================
|
|
|
|
-ifdef(TEST).
|
|
|
|
-include_lib("eunit/include/eunit.hrl").
|
|
|
|
%% Note in OTP 22 we see a compile error with the assertException if using the
|
|
%% eqc_cover parse_transform. This test is excluded in the eqc profle, due to
|
|
%% this error
|
|
|
|
-ifdef(EQC).
|
|
|
|
throw_test() ->
|
|
StoppedFolder =
|
|
fun() ->
|
|
throw(stop_fold)
|
|
end,
|
|
CompletedFolder =
|
|
fun() ->
|
|
{ok, ['1']}
|
|
end,
|
|
AfterAction =
|
|
fun() ->
|
|
error
|
|
end,
|
|
?assertMatch({ok, ['1']}, wrap_runner(CompletedFolder, AfterAction)),
|
|
?assertException(
|
|
throw, stop_fold, wrap_runner(StoppedFolder, AfterAction)
|
|
).
|
|
|
|
-endif.
|
|
|
|
-endif.
|
|
|
|
|
|
|