%%
%% %CopyrightBegin%
%%
%% Copyright Ericsson AB 2001-2013. All Rights Reserved.
%%
%% The contents of this file are subject to the Erlang Public License,
%% Version 1.1, (the "License"); you may not use this file except in
%% compliance with the License. You should have received a copy of the
%% Erlang Public License along with this software. If not, it can be
%% retrieved online at http://www.erlang.org/.
%%
%% Software distributed under the License is distributed on an "AS IS"
%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
%% the License for the specific language governing rights and limitations
%% under the License.
%%
%% %CopyrightEnd%
%%
-module(dets_v9).
%% Dets files, implementation part. This module handles version 9.
%% To be called from dets.erl only.
-export([mark_dirty/1, read_file_header/2,
check_file_header/2, do_perform_save/1, initiate_file/11,
prep_table_copy/9, init_freelist/2, fsck_input/4,
bulk_input/3, output_objs/4, bchunk_init/2,
try_bchunk_header/2, compact_init/3, read_bchunks/2,
write_cache/1, may_grow/3, find_object/2, slot_objs/2,
scan_objs/8, db_hash/2, no_slots/1, table_parameters/1]).
-export([file_info/1, v_segments/1]).
-export([cache_segps/3]).
-compile({inline, [{max_objsize,1},{maxobjsize,1}]}).
-compile({inline, [{write_segment_file,6}]}).
-compile({inline, [{sz2pos,1},{adjsz,1}]}).
-compile({inline, [{skip_bytes,6},{make_object,4}]}).
-compile({inline, [{segp_cache,2},{get_segp,1},{get_arrpart,1}]}).
-compile({inline, [{h,2}]}).
-include("dets.hrl").
%% The layout of the file is :
%%
%% bytes decsription
%% ---------------------- File header
%% 4 FreelistsPointer
%% 4 Cookie
%% 4 ClosedProperly (pos=8)
%% 4 Type (pos=12)
%% 4 Version (pos=16)
%% 4 M
%% 4 Next
%% 4 KeyPos
%% 4 NoObjects
%% 4 NoKeys
%% 4 MinNoSlots
%% 4 MaxNoSlots
%% 4 HashMethod
%% 4 N
%% ---
%% 256 Version 9(a): Reserved for future versions. Initially zeros.
%% Version 9(b) has instead:
%% 112 28 counters for the buddy system sizes 2^4 to 2^31.
%% 144 Reserved for future versions. Initially zeros.
%% Version 9(c) has instead:
%% 112 28 counters for the buddy system sizes (as for 9(b)).
%% 16 MD5-sum for the 44 plus 112 bytes before the MD5-sum.
%% (FreelistsPointer, Cookie and ClosedProperly are not digested.)
%% 128 Reserved for future versions. Initially zeros.
%% Version 9(d), introduced in R15A, has instead:
%% 112 28 counters for the buddy system sizes (as for 9(b)).
%% 16 MD5-sum for the 44 plus 112 bytes before the MD5-sum.
%% (FreelistsPointer, Cookie and ClosedProperly are not digested.)
%% 4 Base of the buddy system.
%% 0 (zero) if the base is equal to ?BASE. Compatible with R14B.
%% File size at the end of the file is RealFileSize - Base.
%% The reason for modifying file size is that when a file created
%% by R15 is read by R14 a repair takes place immediately, which
%% is acceptable when downgrading.
%% 124 Reserved for future versions. Initially zeros.
%% ---
%% ------------------ end of file header
%% 4*256 SegmentArray Pointers.
%% ------------------ This is BASE.
%% 4*512 SegmentArray Part 1
%% ... More SegmentArray Parts
%% 8*256 First segment
%% ??? Objects (free and alive)
%% 4*512 Further SegmentArray Part.
%% ??? Objects (free and alive)
%% 8*256 Further segment.
%% ??? Objects (free and alive)
%% ... more objects, segment array parts, and segments ...
%% -----------------------------
%% ??? Free lists
%% -----------------------------
%% 4 File size, in bytes. See 9(d) obove.
%% Before we can find an object we must find the slot where the
%% object resides. Each slot is a (possibly empty) list (or chain) of
%% objects that hash to the same slot. If the value stored in the
%% slot is zero, the slot chain is empty. If the slot value is
%% non-zero, the value points to a position in the file where the
%% collection of objects resides. Each collection has the following
%% layout:
%%
%% bytes decsription
%% --------------------
%% 4 Size of the area allocated for the collection (8+Sz)
%% 4 Status (FREE or ACTIVE). These two are the Object Header.
%% Sz A binary containing the objects per key, sorted on key.
%%
%% When repairing or converting a file, the status field is used.
%%
%% The binary containing the objects per key of a table of type 'set'
%% has the following layout:
%%
%% bytes decsription
%% --------------------
%% 4 Size of the object of the first key (4+OSz1)
%% OSz1 The object of the first key
%% ...
%% 4 Size of the object of the ith key (4+OSzi)
%% OSzi The object of the ith key
%%
%% The binary containing the objects per key of a table of type 'bag'
%% or 'duplicate_bag' has the following layout:
%%
%% bytes decsription
%% ----------------------
%% 4 Size of the objects of the first key (4 + OSz1_1+...+OSz1_j+...)
%% 4 Size of the first object of the first key (4+OSz1_1)
%% OSz1_1 The first object of the first key
%% ...
%% 4 Size of the jth object of the first key (4+OSz1_j)
%% OSz1_j The jth object of the first key
%% ...
%% 4 Size of the objects of the ith key (4 + OSzi_1+...+OSzi_k+...)
%% 4 Size of the first object of the ith key (4+OSzi_1)
%% OSzi_1 The first object of the ith key
%% ...
%% 4 Size of the kth object of the ith key (4+OSzi_k)
%% OSzi_k The kth object of the ith key
%% ...
%%
%% The objects of a key are placed in time order, that is, the older
%% objects come first. If a new object is inserted, it is inserted
%% last.
%%
%%
%%
%%|---------------|
%%| head |
%%| |
%%| |
%%|_______________|
%%| |--|
%%|___part ptr 1__| |
%%| | | segarr part 1
%%|___part ptr 2__| V______________|
%%| | | p1 |
%%| | |______________|--|
%%| .... | | p2 | |
%% (256) |______________| |
%% | | |
%% | .... | | segment 1
%% | (512) | V __slot 0 ____|
%% | size |
%% | pointer |--|
%% |___slot 1 ____| |
%% | | |
%% | .... | | objects in slot 0
%% (256) V segment 1
%% |___________|
%% | size |
%% |___________|
%% | status |
%% |___________|
%% | |
%% | object |
%% | collec. |
%% |___________|
%%%
%%% File header
%%%
-define(RESERVED, 124). % Reserved for future use.
-define(COLL_CNTRS, (28*4)). % Counters for the buddy system.
-define(MD5SZ, 16).
-define(FL_BASE, 4).
-define(HEADSZ, 56+?COLL_CNTRS % The size of the file header, in bytes,
+?MD5SZ+?FL_BASE). % not including the reserved part.
-define(HEADEND, (?HEADSZ+?RESERVED)).
% End of header and reserved area.
-define(SEGSZ, 512). % Size of a segment, in words. SZOBJP*SEGSZP.
-define(SEGSZP, 256). % Size of a segment, in number of pointers.
-define(SEGSZP_LOG2, 8).
-define(SEGOBJSZ, (4 * ?SZOBJP)).
-define(SEGPARTSZ, 512). % Size of segment array part, in words.
-define(SEGPARTSZ_LOG2, 9).
-define(SEGARRSZ, 256). % Maximal number of segment array parts..
-define(SEGARRADDR(PartN), (?HEADEND + (4 * (PartN)))).
-define(SEGPARTADDR(P,SegN), ((P) + (4 * ?REM2(SegN, ?SEGPARTSZ)))).
-define(BASE, ?SEGARRADDR(?SEGARRSZ)).
-define(MAXSLOTS, (?SEGARRSZ * ?SEGPARTSZ * ?SEGSZP)).
-define(SLOT2SEG(S), ((S) bsr ?SEGSZP_LOG2)).
-define(SEG2SEGARRPART(S), ((S) bsr ?SEGPARTSZ_LOG2)).
-define(PHASH, 0).
-define(PHASH2, 1).
%% BIG is used for hashing. BIG must be greater than the maximum
%% number of slots, currently 32 M (MAXSLOTS).
-define(BIG, 16#3ffffff). % 64 M
%% Hard coded positions into the file header:
-define(FREELIST_POS, 0).
-define(CLOSED_PROPERLY_POS, 8).
-define(D_POS, 20).
%%% Dets file versions up to 8 are handled in dets_v8. This module
%%% handles version 9, introduced in R8.
%%%
%%% Version 9(a) tables have 256 reserved bytes in the file header,
%%% all initialized to zero.
%%% Version 9(b) tables use the first 112 of these bytes for storing
%%% number of objects for each size of the buddy system. An empty 9(b)
%%% table cannot be distinguished from an empty 9(a) table.
%%% 9(c) has an MD5-sum for the file header.
-define(FILE_FORMAT_VERSION, 9).
-define(NOT_PROPERLY_CLOSED,0).
-define(CLOSED_PROPERLY,1).
%% Size of object pointer, in words. SEGSZ = SZOBJP * SEGSZP.
-define(SZOBJP, 2).
-define(OHDSZ, 8). % The size of the object header, in bytes.
-define(STATUS_POS, 4). % Position of the status field.
-define(OHDSZ_v8, 12). % The size of the version 8 object header.
%% The size of each object is a multiple of 16.
%% BUMP is used when repairing files.
-define(BUMP, 16).
%%% '$hash' is the value of HASH_PARMS in R8, '$hash2' is the value in R9.
%%%
%%% The fields of the ?HASH_PARMS records are the same, but having
%%% different tags makes bchunk_init on R8 nodes reject data from R9
%%% nodes, and vice versa. This is overkill, and due to an oversight.
%%% What should have been done in R8 was to check the hash method, not
%%% only the type of the table and the key position. R8 nodes cannot
%%% handle the phash2 method.
-define(HASH_PARMS, '$hash2').
-define(BCHUNK_FORMAT_VERSION, 1).
-record(?HASH_PARMS, {
file_format_version,
bchunk_format_version,
file, type, keypos, hash_method,
n,m,next,
min,max,
no_objects,no_keys,
no_colls % [{LogSz,NoColls}], NoColls >= 0
}).
-define(ACTUAL_SEG_SIZE, (?SEGSZ*4)).
-define(MAXBUD, 32).
%%-define(DEBUGF(X,Y), io:format(X, Y)).
-define(DEBUGF(X,Y), void).
%% -> ok | throw({NewHead,Error})
mark_dirty(Head) ->
Dirty = [{?CLOSED_PROPERLY_POS, <<?NOT_PROPERLY_CLOSED:32>>}],
{_H, ok} = dets_utils:pwrite(Head, Dirty),
ok = dets_utils:sync(Head),
{ok, _Pos} = dets_utils:position(Head, Head#head.freelists_p),
dets_utils:truncate(Head, cur).
%% -> {ok, head()} | throw(Error) | throw(badarg)
prep_table_copy(Fd, Tab, Fname, Type, Kp, Ram, CacheSz, Auto, Parms) ->
case Parms of
#?HASH_PARMS{file_format_version = ?FILE_FORMAT_VERSION,
bchunk_format_version = ?BCHUNK_FORMAT_VERSION,
n = N, m = M, next = Next,
min = Min, max = Max,
hash_method = HashMethodCode,
no_objects = NoObjects, no_keys = NoKeys,
no_colls = _NoColls}
when is_integer(N), is_integer(M), is_integer(Next),
is_integer(Min), is_integer(Max),
is_integer(NoObjects), is_integer(NoKeys),
NoObjects >= NoKeys ->
HashMethod = code_to_hash_method(HashMethodCode),
case hash_invars(N, M, Next, Min, Max) of
false ->
throw(badarg);
true ->
init_file(Fd, Tab, Fname, Type, Kp, Min, Max, Ram,
CacheSz, Auto, false, M, N, Next, HashMethod,
NoObjects, NoKeys)
end;
_ ->
throw(badarg)
end.
%% -> {ok, head()} | throw(Error)
%% The File header and the SegmentArray Pointers are written here.
%% SegmentArray Parts are also written, but the segments are are not
%% initialized on file unless DoInitSegments is 'true'. (When
%% initializing a file by calling init_table, some time is saved by
%% not writing the segments twice.)
initiate_file(Fd, Tab, Fname, Type, Kp, MinSlots0, MaxSlots0,
Ram, CacheSz, Auto, DoInitSegments) ->
MaxSlots1 = erlang:min(MaxSlots0, ?MAXSLOTS),
MinSlots1 = erlang:min(MinSlots0, MaxSlots1),
MinSlots = slots2(MinSlots1),
MaxSlots = slots2(MaxSlots1),
M = Next = MinSlots,
N = 0,
init_file(Fd, Tab, Fname, Type, Kp, MinSlots, MaxSlots, Ram, CacheSz,
Auto, DoInitSegments, M, N, Next, phash2, 0, 0).
init_file(Fd, Tab, Fname, Type, Kp, MinSlots, MaxSlots, Ram, CacheSz,
Auto, DoInitSegments, M, N, Next, HashMethod, NoObjects, NoKeys) ->
Ftab = dets_utils:init_alloc(?BASE),
Head0 = #head{
m = M,
m2 = M * 2,
next = Next,
fptr = Fd,
no_objects = NoObjects,
no_keys = NoKeys,
maxobjsize = 0,
n = N,
type = Type,
update_mode = dirty,
freelists = Ftab,
no_collections = orddict:new(),
auto_save = Auto,
hash_bif = HashMethod,
has_md5 = true,
keypos = Kp,
min_no_slots = MinSlots,
max_no_slots = MaxSlots,
ram_file = Ram,
filename = Fname,
name = Tab,
cache = dets_utils:new_cache(CacheSz),
version = ?FILE_FORMAT_VERSION,
bump = ?BUMP,
base = ?BASE, % to be overwritten
mod = ?MODULE
},
FreeListsPointer = 0,
NoColls = <<0:?COLL_CNTRS/unit:8>>, %% Buddy system counters.
FileHeader = file_header(Head0, FreeListsPointer,
?NOT_PROPERLY_CLOSED, NoColls),
W0 = {0, [FileHeader |
<<0:(4*?SEGARRSZ)/unit:8>>]}, %% SegmentArray Pointers
%% Remove cached pointers to segment array parts and segments:
lists:foreach(fun({I1,I2}) when is_integer(I1), is_integer(I2) -> ok;
({K,V}) -> put(K, V)
end, erase()),
%% Initialize array parts.
%% All parts before segments, for the sake of repair and initialization.
Zero = seg_zero(),
{Head1, Ws1} = init_parts(Head0, 0, no_parts(Next), Zero, []),
NoSegs = no_segs(Next),
{Head2, WsI, WsP} = init_segments(Head1, 0, NoSegs, Zero, [], []),
Ws2 = if
DoInitSegments -> WsP ++ WsI;
true -> WsP
end,
dets_utils:pwrite(Fd, Fname, [W0 | lists:append(Ws1) ++ Ws2]),
true = hash_invars(Head2),
%% The allocations that have been made so far (parts, segments)
%% are permanent; the table will never shrink. Therefore the base
%% of the Buddy system can be set to the first free object.
%% This is used in allocate_all(), see below.
{_, Where, _} = dets_utils:alloc(Head2, ?BUMP),
NewFtab = dets_utils:init_alloc(Where),
Head = Head2#head{freelists = NewFtab, base = Where},
{ok, Head}.
%% Returns a power of two not less than 256.
slots2(NoSlots) when NoSlots >= 256 ->
?POW(dets_utils:log2(NoSlots)).
init_parts(Head, PartNo, NoParts, Zero, Ws) when PartNo < NoParts ->
PartPos = ?SEGARRADDR(PartNo),
{NewHead, W, _Part} = alloc_part(Head, Zero, PartPos),
init_parts(NewHead, PartNo+1, NoParts, Zero, [W | Ws]);
init_parts(Head, _PartNo, _NoParts, _Zero, Ws) ->
{Head, Ws}.
%% -> {Head, SegInitList, OtherList};
%% SegPtrList = SegInitList = pwrite_list().
init_segments(Head, SegNo, NoSegs, SegZero, WsP, WsI) when SegNo < NoSegs ->
{NewHead, WI, Ws} = allocate_segment(Head, SegZero, SegNo),
init_segments(NewHead, SegNo+1, NoSegs, SegZero, Ws ++ WsP, [WI | WsI]);
init_segments(Head, _SegNo, _NoSegs, _SegZero, WsP, WsI) ->
{Head, WsI, WsP}.
%% -> {NewHead, SegInit, [SegPtr | PartStuff]}
allocate_segment(Head, SegZero, SegNo) ->
PartPos = ?SEGARRADDR(SegNo div ?SEGPARTSZ),
case get_arrpart(PartPos) of
undefined ->
%% may throw error:
{Head1, [InitArrPart, ArrPartPointer], Part} =
alloc_part(Head, SegZero, PartPos),
{NewHead, InitSegment, [SegPointer]} =
alloc_seg(Head1, SegZero, SegNo, Part),
{NewHead, InitSegment, [InitArrPart, SegPointer, ArrPartPointer]};
Part ->
alloc_seg(Head, SegZero, SegNo, Part)
end.
alloc_part(Head, PartZero, PartPos) ->
%% may throw error:
{NewHead, Part, _} = dets_utils:alloc(Head, adjsz(4 * ?SEGPARTSZ)),
arrpart_cache(PartPos, Part),
InitArrPart = {Part, PartZero}, % same size as segment
ArrPartPointer = {PartPos, <<Part:32>>},
{NewHead, [InitArrPart, ArrPartPointer], Part}.
alloc_seg(Head, SegZero, SegNo, Part) ->
%% may throw error:
{NewHead, Segment, _} = dets_utils:alloc(Head, adjsz(4 * ?SEGSZ)),
InitSegment = {Segment, SegZero},
Pos = ?SEGPARTADDR(Part, SegNo),
segp_cache(Pos, Segment),
dets_utils:disk_map_segment(Segment, SegZero),
SegPointer = {Pos, <<Segment:32>>},
{NewHead, InitSegment, [SegPointer]}.
%% Read free lists (using a Buddy System) from file.
init_freelist(Head, true) ->
Pos = Head#head.freelists_p,
free_lists_from_file(Head, Pos).
%% -> {ok, Fd, fileheader()} | throw(Error)
read_file_header(Fd, FileName) ->
{ok, Bin} = dets_utils:pread_close(Fd, FileName, 0, ?HEADSZ),
<<FreeList:32, Cookie:32, CP:32, Type2:32,
Version:32, M:32, Next:32, Kp:32,
NoObjects:32, NoKeys:32, MinNoSlots:32, MaxNoSlots:32,
HashMethod:32, N:32, NoCollsB:?COLL_CNTRS/binary,
MD5:?MD5SZ/binary, FlBase:32>> = Bin,
<<_:12/binary,MD5DigestedPart:(?HEADSZ-?MD5SZ-?FL_BASE-12)/binary,
_/binary>> = Bin,
{ok, EOF} = dets_utils:position_close(Fd, FileName, eof),
{ok, <<FileSize:32>>} = dets_utils:pread_close(Fd, FileName, EOF-4, 4),
{CL, <<>>} = lists:foldl(fun(LSz, {Acc,<<NN:32,R/binary>>}) ->
if
NN =:= 0 -> {Acc, R};
true -> {[{LSz,NN} | Acc], R}
end
end, {[], NoCollsB}, lists:seq(4, ?MAXBUD-1)),
NoColls =
if
CL =:= [], NoObjects > 0 -> % Version 9(a)
undefined;
true ->
lists:reverse(CL)
end,
Base = case FlBase of
0 -> ?BASE;
_ -> FlBase
end,
FH = #fileheader{freelist = FreeList,
fl_base = Base,
cookie = Cookie,
closed_properly = CP,
type = dets_utils:code_to_type(Type2),
version = Version,
m = M,
next = Next,
keypos = Kp,
no_objects = NoObjects,
no_keys = NoKeys,
min_no_slots = MinNoSlots,
max_no_slots = MaxNoSlots,
no_colls = NoColls,
hash_method = HashMethod,
read_md5 = MD5,
has_md5 = <<0:?MD5SZ/unit:8>> =/= MD5,
md5 = erlang:md5(MD5DigestedPart),
trailer = FileSize + FlBase,
eof = EOF,
n = N,
mod = ?MODULE},
{ok, Fd, FH}.
%% -> {ok, head(), ExtraInfo} | {error, Reason} (Reason lacking file name)
%% ExtraInfo = true
check_file_header(FH, Fd) ->
HashBif = code_to_hash_method(FH#fileheader.hash_method),
Test =
if
FH#fileheader.cookie =/= ?MAGIC ->
{error, not_a_dets_file};
FH#fileheader.type =:= badtype ->
{error, invalid_type_code};
FH#fileheader.version =/= ?FILE_FORMAT_VERSION ->
{error, bad_version};
FH#fileheader.has_md5,
FH#fileheader.read_md5 =/= FH#fileheader.md5 ->
{error, not_a_dets_file}; % harsh but fair
FH#fileheader.trailer =/= FH#fileheader.eof ->
{error, not_closed};
HashBif =:= undefined ->
{error, bad_hash_bif};
FH#fileheader.closed_properly =:= ?CLOSED_PROPERLY ->
{ok, true};
FH#fileheader.closed_properly =:= ?NOT_PROPERLY_CLOSED ->
{error, not_closed};
true ->
{error, not_a_dets_file}
end,
case Test of
{ok, ExtraInfo} ->
MaxObjSize = max_objsize(FH#fileheader.no_colls),
H = #head{
m = FH#fileheader.m,
m2 = FH#fileheader.m * 2,
next = FH#fileheader.next,
fptr = Fd,
no_objects = FH#fileheader.no_objects,
no_keys = FH#fileheader.no_keys,
maxobjsize = MaxObjSize,
n = FH#fileheader.n,
type = FH#fileheader.type,
update_mode = saved,
auto_save = infinity, % not saved on file
fixed = false, % not saved on file
freelists_p = FH#fileheader.freelist,
hash_bif = HashBif,
has_md5 = FH#fileheader.has_md5,
keypos = FH#fileheader.keypos,
min_no_slots = FH#fileheader.min_no_slots,
max_no_slots = FH#fileheader.max_no_slots,
no_collections = FH#fileheader.no_colls,
version = ?FILE_FORMAT_VERSION,
mod = ?MODULE,
bump = ?BUMP,
base = FH#fileheader.fl_base},
{ok, H, ExtraInfo};
Error ->
Error
end.
%% Inlined.
max_objsize(NoColls = undefined) ->
NoColls;
max_objsize(NoColls) ->
max_objsize(NoColls, 0).
max_objsize([], Max) ->
Max;
max_objsize([{_,0} | L], Max) ->
max_objsize(L, Max);
max_objsize([{I,_} | L], _Max) ->
max_objsize(L, I).
cache_segps(Fd, FileName, M) ->
NoParts = no_parts(M),
ArrStart = ?SEGARRADDR(0),
{ok, Bin} = dets_utils:pread_close(Fd, FileName, ArrStart, 4 * NoParts),
cache_arrparts(Bin, ?HEADEND, Fd, FileName).
cache_arrparts(<<ArrPartPos:32, B/binary>>, Pos, Fd, FileName) ->
arrpart_cache(Pos, ArrPartPos),
{ok, ArrPartBin} = dets_utils:pread_close(Fd, FileName,
ArrPartPos,
?SEGPARTSZ*4),
cache_segps1(Fd, ArrPartBin, ArrPartPos),
cache_arrparts(B, Pos+4, Fd, FileName);
cache_arrparts(<<>>, _Pos, _Fd, _FileName) ->
ok.
cache_segps1(_Fd, <<0:32,_/binary>>, _P) ->
ok;
cache_segps1(Fd, <<S:32,B/binary>>, P) ->
dets_utils:disk_map_segment_p(Fd, S),
segp_cache(P, S),
cache_segps1(Fd, B, P+4);
cache_segps1(_Fd, <<>>, _P) ->
ok.
no_parts(NoSlots) ->
((NoSlots - 1) div (?SEGSZP * ?SEGPARTSZ)) + 1.
no_segs(NoSlots) ->
((NoSlots - 1) div ?SEGSZP) + 1.
%%%
%%% Repair, conversion and initialization of a dets file.
%%%
%%% bulk_input/3. Initialization, the general case (any stream of objects).
%%% output_objs/4. Initialization (general case) and repair.
%%% bchunk_init/2. Initialization using bchunk.
bulk_input(Head, InitFun, _Cntrs) ->
bulk_input(Head, InitFun, make_ref(), 0).
bulk_input(Head, InitFun, Ref, Seq) ->
fun(close) ->
_ = (catch InitFun(close));
(read) ->
case catch {Ref, InitFun(read)} of
{Ref, end_of_input} ->
end_of_input;
{Ref, {L0, NewInitFun}} when is_list(L0),
is_function(NewInitFun) ->
Kp = Head#head.keypos,
case catch bulk_objects(L0, Head, Kp, Seq, []) of
{'EXIT', _Error} ->
_ = (catch NewInitFun(close)),
{error, invalid_objects_list};
{L, NSeq} ->
{L, bulk_input(Head, NewInitFun, Ref, NSeq)}
end;
{Ref, Value} ->
{error, {init_fun, Value}};
Error ->
throw({thrown, Error})
end
end.
bulk_objects([T | Ts], Head, Kp, Seq, L) ->
BT = term_to_binary(T),
Key = element(Kp, T),
bulk_objects(Ts, Head, Kp, Seq+1, [make_object(Head, Key, Seq, BT) | L]);
bulk_objects([], _Head, Kp, Seq, L) when is_integer(Kp), is_integer(Seq) ->
{L, Seq}.
-define(FSCK_SEGMENT, 1).
-define(FSCK_SEGMENT2, 10000).
-define(VEMPTY, {}).
-define(VSET(I, V, E), setelement(I, V, E)).
-define(VGET(I, V), element(I, V)).
-define(VEXT(S, V, T),
list_to_tuple(tuple_to_list(V) ++ lists:duplicate(S-tuple_size(V), T))).
%% Number of bytes that will be handled before the cache is written to
%% file. Used when compacting or writing chunks.
-define(CACHE_SIZE, (60*?CHUNK_SIZE)).
%% {LogSize,NoObjects} in Cntrs is replaced by
%% {LogSize,Position,{FileName,FileDescriptor},NoCollections}.
%% There is also an object {no, NoObjects, NoKeys}.
-define(COUNTERS, no).
-define(OBJ_COUNTER, 2).
-define(KEY_COUNTER, 3).
output_objs(OldV, Head, SlotNums, Cntrs) when OldV =< 9 ->
fun(close) ->
%% Make sure that the segments are initialized in case
%% init_table has been called.
Cache = ?VEMPTY,
Acc = [], % This is the only way Acc can be empty.
true = ets:insert(Cntrs, {?FSCK_SEGMENT,0,[],0}),
true = ets:insert(Cntrs, {?COUNTERS, 0, 0}),
Fun = output_objs2(foo, Acc, OldV, Head, Cache, Cntrs,
SlotNums, bar),
Fun(close);
([]) ->
output_objs(OldV, Head, SlotNums, Cntrs);
(L) ->
%% Information about number of objects per size is not
%% relevant for version 9. It is the number of collections
%% that matters.
true = ets:delete_all_objects(Cntrs),
true = ets:insert(Cntrs, {?COUNTERS, 0, 0}),
Es = bin2term(L, OldV, Head#head.keypos),
%% The cache is a tuple indexed by the (log) size. An element
%% is [BinaryObject].
Cache = ?VEMPTY,
{NE, NAcc, NCache} = output_slots(Es, Head, Cache, Cntrs, 0, 0),
output_objs2(NE, NAcc, OldV, Head, NCache, Cntrs, SlotNums, 1)
end.
output_objs2(E, Acc, OldV, Head, Cache, SizeT, SlotNums, 0) ->
NCache = write_all_sizes(Cache, SizeT, Head, more),
%% Number of handled file_sorter chunks before writing:
Max = erlang:max(1, erlang:min(tuple_size(NCache), 10)),
output_objs2(E, Acc, OldV, Head, NCache, SizeT, SlotNums, Max);
output_objs2(E, Acc, OldV, Head, Cache, SizeT, SlotNums, ChunkI) ->
fun(close) ->
{_, [], Cache1} =
if
Acc =:= [] -> {foo, [], Cache};
true -> output_slot(Acc, Head, Cache, [], SizeT, 0, 0)
end,
_NCache = write_all_sizes(Cache1, SizeT, Head, no_more),
SegSz = ?ACTUAL_SEG_SIZE,
{_, SegEnd, _} = dets_utils:alloc(Head, adjsz(SegSz)),
[{?COUNTERS,NoObjects,NoKeys}] = ets:lookup(SizeT, ?COUNTERS),
Head1 = Head#head{no_objects = NoObjects, no_keys = NoKeys},
true = ets:delete(SizeT, ?COUNTERS),
{NewHead, NL, _MaxSz, _End} = allocate_all_objects(Head1, SizeT),
%% It is not known until all objects have been collected
%% how many object collections there are per size. Now
%% that is known and the absolute positions of the object
%% collections can be calculated.
segment_file(SizeT, NewHead, NL, SegEnd),
{MinSlots, EstNoSlots, MaxSlots} = SlotNums,
if
EstNoSlots =:= bulk_init ->
{ok, 0, NewHead};
true ->
EstNoSegs = no_segs(EstNoSlots),
MinNoSegs = no_segs(MinSlots),
MaxNoSegs = no_segs(MaxSlots),
NoSegs = no_segs(NoKeys),
Diff = abs(NoSegs - EstNoSegs),
if
Diff > 5, NoSegs =< MaxNoSegs, NoSegs >= MinNoSegs ->
{try_again, NoKeys};
true ->
{ok, 0, NewHead}
end
end;
(L) ->
Es = bin2term(L, OldV, Head#head.keypos),
{NE, NAcc, NCache} =
output_slots(E, Es, Acc, Head, Cache, SizeT, 0, 0),
output_objs2(NE, NAcc, OldV, Head, NCache, SizeT, SlotNums,
ChunkI-1)
end.
%%% Compaction.
compact_init(ReadHead, WriteHead, TableParameters) ->
SizeT = ets:new(dets_compact, []),
#head{no_keys = NoKeys, no_objects = NoObjects} = ReadHead,
NoObjsPerSize = TableParameters#?HASH_PARMS.no_colls,
{NewWriteHead, Bases, SegAddr, SegEnd} =
prepare_file_init(NoObjects, NoKeys, NoObjsPerSize, SizeT, WriteHead),
Input = compact_input(ReadHead, NewWriteHead, SizeT, tuple_size(Bases)),
Output = fast_output(NewWriteHead, SizeT, Bases, SegAddr, SegEnd),
TmpDir = filename:dirname(NewWriteHead#head.filename),
Reply = (catch file_sorter:sort(Input, Output,
[{format, binary},{tmpdir, TmpDir},
{header, 1}])), % compact_objs/9: 13 bytes
ets:delete(SizeT),
Reply.
compact_input(Head, WHead, SizeT, NoSizes) ->
L = dets_utils:all_allocated_as_list(Head),
Cache = ?VEXT(NoSizes, ?VEMPTY, [0 | []]),
compact_input(Head, WHead, SizeT, Cache, L).
compact_input(Head, WHead, SizeT, Cache, L) ->
fun(close) ->
ok;
(read) ->
compact_read(Head, WHead, SizeT, Cache, L, 0, [], 0)
end.
compact_read(_Head, WHead, SizeT, Cache, [], _Min, [], _ASz) ->
_ = fast_write_all_sizes(Cache, SizeT, WHead),
end_of_input;
compact_read(Head, WHead, SizeT, Cache, L, Min, SegBs, ASz)
when ASz + Min >= ?CACHE_SIZE, ASz > 0 ->
NCache = fast_write_all_sizes(Cache, SizeT, WHead),
{SegBs, compact_input(Head, WHead, SizeT, NCache, L)};
compact_read(Head, WHead, SizeT, Cache, [[From | To] | L], Min, SegBs, ASz) ->
Max = erlang:max(?CHUNK_SIZE*3, Min),
case check_pread_arg(Max, Head) of
true ->
case dets_utils:pread_n(Head#head.fptr, From, Max) of
eof ->
%% Should never happen since compaction will not
%% be tried unless the file trailer is valid.
not_ok; % try a proper repair
Bin1 when byte_size(Bin1) < Min ->
%% The last object may not be padded.
Pad = Min - byte_size(Bin1),
NewBin = <<Bin1/binary, 0:Pad/unit:8>>,
compact_objs(Head, WHead, SizeT, NewBin, L,
From, To, SegBs, Cache, ASz);
NewBin ->
compact_objs(Head, WHead, SizeT, NewBin, L,
From, To, SegBs, Cache, ASz)
end;
false ->
not_ok % try a proper repair
end.
compact_objs(Head, WHead, SizeT, Bin, L, From, To, SegBs, Cache, ASz)
when From =:= To ->
case L of
[] ->
{SegBs, compact_input(Head, WHead, SizeT, Cache, L)};
[[From1 | To1] | L1] ->
Skip1 = From1 - From,
case Bin of
<<_:Skip1/binary,NewBin/binary>> ->
compact_objs(Head, WHead, SizeT, NewBin, L1, From1, To1,
SegBs, Cache, ASz);
_ when byte_size(Bin) < Skip1 ->
compact_read(Head, WHead, SizeT, Cache, L, 0, SegBs, ASz)
end
end;
compact_objs(Head, WHead, SizeT, <<Size:32, St:32, _Sz:32, KO/binary>> = Bin,
L, From, To, SegBs, Cache, ASz) when St =:= ?ACTIVE ->
LSize = sz2pos(Size),
Size2 = ?POW(LSize-1),
if
byte_size(Bin) >= Size2 ->
NASz = ASz + Size2,
<<SlotObjs:Size2/binary, NewBin/binary>> = Bin,
Term = if
Head#head.type =:= set ->
binary_to_term(KO);
true ->
<<_KSz:32,B2/binary>> = KO,
binary_to_term(B2)
end,
Key = element(Head#head.keypos, Term),
Slot = db_hash(Key, Head),
From1 = From + Size2,
[Addr | AL] = ?VGET(LSize, Cache),
NCache = ?VSET(LSize, Cache, [Addr + Size2 | [SlotObjs | AL]]),
NSegBs = [<<Slot:32,Size:32,Addr:32,LSize:8>> | SegBs],
compact_objs(Head, WHead, SizeT, NewBin, L, From1,
To, NSegBs, NCache, NASz);
true ->
compact_read(Head, WHead, SizeT, Cache, [[From|To] | L],
Size2, SegBs, ASz)
end;
compact_objs(Head, WHead, SizeT, <<_:32, _St:32, _:32, _/binary>> = Bin,
L, From, To, SegBs, Cache, ASz)
when byte_size(Bin) >= ?ACTUAL_SEG_SIZE -> % , _St =/= ?ACTIVE
<<_:?ACTUAL_SEG_SIZE/binary, NewBin/binary>> = Bin,
compact_objs(Head, WHead, SizeT, NewBin, L, From + ?ACTUAL_SEG_SIZE,
To, SegBs, Cache, ASz);
compact_objs(Head, WHead, SizeT, <<_:32, _St:32, _:32, _/binary>> = Bin,
L, From, To, SegBs, Cache, ASz)
when byte_size(Bin) < ?ACTUAL_SEG_SIZE -> % , _St =/= ?ACTIVE
compact_read(Head, WHead, SizeT, Cache, [[From|To] | L],
?ACTUAL_SEG_SIZE, SegBs, ASz);
compact_objs(Head, WHead, SizeT, _Bin, L, From, To, SegBs, Cache, ASz) ->
compact_read(Head, WHead, SizeT, Cache, [[From|To] | L], 0, SegBs, ASz).
%%% End compaction.
%%% Bchunk.
read_bchunks(Head, L) ->
read_bchunks(Head, L, 0, [], 0).
read_bchunks(_Head, L, Min, Bs, ASz) when ASz + Min >= 4*?CHUNK_SIZE,
Bs =/= [] ->
{lists:reverse(Bs), L};
read_bchunks(Head, {From, To, L}, Min, Bs, ASz) ->
Max = erlang:max(?CHUNK_SIZE*2, Min),
case check_pread_arg(Max, Head) of
true ->
case dets_utils:pread_n(Head#head.fptr, From, Max) of
eof ->
%% Should never happen.
{error, premature_eof};
NewBin when byte_size(NewBin) >= Min ->
bchunks(Head, L, NewBin, Bs, ASz, From, To);
Bin1 when To - From =:= Min, L =:= <<>> ->
%% when byte_size(Bin1) < Min.
%% The last object may not be padded.
Pad = Min - byte_size(Bin1),
NewBin = <<Bin1/binary, 0:Pad/unit:8>>,
bchunks(Head, L, NewBin, Bs, ASz, From, To);
_ ->
{error, premature_eof}
end;
false ->
{error, dets_utils:bad_object(bad_object, {read_bchunks, Max})}
end.
bchunks(Head, L, Bin, Bs, ASz, From, To) when From =:= To ->
if
L =:= <<>> ->
{finished, lists:reverse(Bs)};
true ->
<<From1:32, To1:32, L1/binary>> = L,
Skip1 = From1 - From,
case Bin of
<<_:Skip1/binary,NewBin/binary>> ->
bchunks(Head, L1, NewBin, Bs, ASz, From1, To1);
_ when byte_size(Bin) < Skip1 ->
read_bchunks(Head, {From1,To1,L1}, 0, Bs, ASz)
end
end;
bchunks(Head, L, <<Size:32, St:32, _Sz:32, KO/binary>> = Bin, Bs, ASz,
From, To) when St =:= ?ACTIVE; St =:= ?FREE ->
LSize = sz2pos(Size),
Size2 = ?POW(LSize-1),
if
byte_size(Bin) >= Size2 ->
<<B0:Size2/binary, NewBin/binary>> = Bin,
%% LSize and Slot are used in make_slots/6. The reason to
%% calculate Slot here is to reduce the CPU load in
%% make_slots/6.
Term = if
Head#head.type =:= set ->
binary_to_term(KO);
true ->
<<_KSz:32,B2/binary>> = KO,
binary_to_term(B2)
end,
Key = element(Head#head.keypos, Term),
Slot = db_hash(Key, Head),
B = {LSize,Slot,B0},
bchunks(Head, L, NewBin, [B | Bs], ASz + Size2, From+Size2, To);
true ->
read_bchunks(Head, {From, To, L}, Size2, Bs, ASz)
end;
bchunks(Head, L, <<_:32, _St:32, _:32, _/binary>> = Bin, Bs, ASz, From, To)
when byte_size(Bin) >= ?ACTUAL_SEG_SIZE ->
<<_:?ACTUAL_SEG_SIZE/binary, NewBin/binary>> = Bin,
bchunks(Head, L, NewBin, Bs, ASz, From + ?ACTUAL_SEG_SIZE, To);
bchunks(Head, L, <<_:32, _St:32, _:32, _/binary>> = Bin, Bs, ASz, From, To)
when byte_size(Bin) < ?ACTUAL_SEG_SIZE ->
read_bchunks(Head, {From, To, L}, ?ACTUAL_SEG_SIZE, Bs, ASz);
bchunks(Head, L, _Bin, Bs, ASz, From, To) ->
read_bchunks(Head, {From, To, L}, 0, Bs, ASz).
%%% End bchunk.
%% -> {ok, NewHead} | throw(Error) | Error
bchunk_init(Head, InitFun) ->
Ref = make_ref(),
%% The non-empty list of data begins with the table parameters.
case catch {Ref, InitFun(read)} of
{Ref, end_of_input} ->
{error, {init_fun, end_of_input}};
{Ref, {[], NInitFun}} when is_function(NInitFun) ->
bchunk_init(Head, NInitFun);
{Ref, {[ParmsBin | L], NInitFun}}
when is_list(L), is_function(NInitFun) ->
#head{fptr = Fd, type = Type, keypos = Kp,
auto_save = Auto, cache = Cache,
filename = Fname, ram_file = Ram,
name = Tab} = Head,
case try_bchunk_header(ParmsBin, Head) of
{ok, Parms} ->
#?HASH_PARMS{no_objects = NoObjects,
no_keys = NoKeys,
no_colls = NoObjsPerSize} = Parms,
CacheSz = dets_utils:cache_size(Cache),
{ok, Head1} =
prep_table_copy(Fd, Tab, Fname, Type,
Kp, Ram, CacheSz,
Auto, Parms),
SizeT = ets:new(dets_init, []),
{NewHead, Bases, SegAddr, SegEnd} =
prepare_file_init(NoObjects, NoKeys,
NoObjsPerSize, SizeT, Head1),
ECache = ?VEXT(tuple_size(Bases), ?VEMPTY, [0 | []]),
Input =
fun(close) ->
_ = (catch NInitFun(close));
(read) ->
do_make_slots(L, ECache, SizeT, NewHead, Ref,
0, NInitFun)
end,
Output = fast_output(NewHead, SizeT, Bases, SegAddr,SegEnd),
TmpDir = filename:dirname(Head#head.filename),
Reply = (catch file_sorter:sort(Input, Output,
[{format, binary},
{tmpdir, TmpDir},
{header, 1}])),
ets:delete(SizeT),
Reply;
not_ok ->
{error, {init_fun, ParmsBin}}
end;
{Ref, Value} ->
{error, {init_fun, Value}};
Error ->
{thrown, Error}
end.
try_bchunk_header(ParmsBin, Head) ->
#head{type = Type, keypos = Kp, hash_bif = HashBif} = Head,
HashMethod = hash_method_to_code(HashBif),
case catch binary_to_term(ParmsBin) of
Parms when is_record(Parms, ?HASH_PARMS),
Parms#?HASH_PARMS.type =:= Type,
Parms#?HASH_PARMS.keypos =:= Kp,
Parms#?HASH_PARMS.hash_method =:= HashMethod,
Parms#?HASH_PARMS.bchunk_format_version =:=
?BCHUNK_FORMAT_VERSION ->
{ok, Parms};
_ ->
not_ok
end.
bchunk_input(InitFun, SizeT, Head, Ref, Cache, ASz) ->
fun(close) ->
_ = (catch InitFun(close));
(read) ->
case catch {Ref, InitFun(read)} of
{Ref, end_of_input} ->
_ = fast_write_all_sizes(Cache, SizeT, Head),
end_of_input;
{Ref, {L, NInitFun}} when is_list(L), is_function(NInitFun) ->
do_make_slots(L, Cache, SizeT, Head, Ref, ASz,
NInitFun);
{Ref, Value} ->
{error, {init_fun, Value}};
Error ->
throw({thrown, Error})
end
end.
do_make_slots(L, Cache, SizeT, Head, Ref, ASz, InitFun) ->
case catch make_slots(L, Cache, [], ASz) of
{'EXIT', _} ->
_ = (catch InitFun(close)),
{error, invalid_objects_list};
{Cache1, SegBs, NASz} when NASz > ?CACHE_SIZE ->
NCache = fast_write_all_sizes(Cache1, SizeT, Head),
F = bchunk_input(InitFun, SizeT, Head, Ref, NCache, 0),
{SegBs, F};
{NCache, SegBs, NASz} ->
F = bchunk_input(InitFun, SizeT, Head, Ref, NCache, NASz),
{SegBs, F}
end.
make_slots([{LSize,Slot,<<Size:32, St:32, Sz:32, KO/binary>> = Bin0} | Bins],
Cache, SegBs, ASz) ->
Bin = if
St =:= ?ACTIVE ->
Bin0;
St =:= ?FREE ->
<<Size:32,?ACTIVE:32,Sz:32,KO/binary>>
end,
BSz = byte_size(Bin0),
true = (BSz =:= ?POW(LSize-1)),
NASz = ASz + BSz,
[Addr | L] = ?VGET(LSize, Cache),
NSegBs = [<<Slot:32,Size:32,Addr:32,LSize:8>> | SegBs],
NCache = ?VSET(LSize, Cache, [Addr + BSz | [Bin | L]]),
make_slots(Bins, NCache, NSegBs, NASz);
make_slots([], Cache, SegBs, ASz) ->
{Cache, SegBs, ASz}.
fast_output(Head, SizeT, Bases, SegAddr, SegEnd) ->
fun(close) ->
fast_output_end(Head, SizeT);
(L) ->
case file:position(Head#head.fptr, SegAddr) of
{ok, SegAddr} ->
NewSegAddr = write_segment_file(L, Bases, Head, [],
SegAddr, SegAddr),
fast_output2(Head, SizeT, Bases, NewSegAddr,
SegAddr, SegEnd);
Error ->
catch dets_utils:file_error(Error, Head#head.filename)
end
end.
fast_output2(Head, SizeT, Bases, SegAddr, SS, SegEnd) ->
fun(close) ->
FinalZ = SegEnd - SegAddr,
dets_utils:write(Head, dets_utils:make_zeros(FinalZ)),
fast_output_end(Head, SizeT);
(L) ->
NewSegAddr = write_segment_file(L, Bases, Head, [], SegAddr, SS),
fast_output2(Head, SizeT, Bases, NewSegAddr, SS, SegEnd)
end.
fast_output_end(Head, SizeT) ->
case ets:foldl(fun({_Sz,_Pos,Cnt,NoC}, Acc) -> (Cnt =:= NoC) and Acc end,
true, SizeT) of
true -> {ok, Head};
false -> {error, invalid_objects_list}
end.
%% Inlined.
write_segment_file([<<Slot:32,BSize:32,AddrToBe:32,LSize:8>> | Bins],
Bases, Head, Ws, SegAddr, SS) ->
%% Should call slot_position/1, but since all segments are
%% allocated in a sequence, the position of a slot can be
%% calculated faster.
Pos = SS + ?SZOBJP*4 * Slot, % Same as Pos = slot_position(Slot).
write_segment_file(Bins, Bases, Head, Ws, SegAddr, SS, Pos,
BSize, AddrToBe, LSize);
write_segment_file([], _Bases, Head, Ws, SegAddr, _SS) ->
dets_utils:write(Head, Ws),
SegAddr.
write_segment_file(Bins, Bases, Head, Ws, SegAddr, SS, Pos, BSize,
AddrToBe, LSize) when Pos =:= SegAddr ->
Addr = AddrToBe + element(LSize, Bases),
NWs = [Ws | <<BSize:32,Addr:32>>],
write_segment_file(Bins, Bases, Head, NWs, SegAddr + ?SZOBJP*4, SS);
write_segment_file(Bins, Bases, Head, Ws, SegAddr, SS, Pos, BSize,
AddrToBe, LSize) when Pos - SegAddr < 100 ->
Addr = AddrToBe + element(LSize, Bases),
NoZeros = Pos - SegAddr,
NWs = [Ws | <<0:NoZeros/unit:8,BSize:32,Addr:32>>],
NSegAddr = SegAddr + NoZeros + ?SZOBJP*4,
write_segment_file(Bins, Bases, Head, NWs, NSegAddr, SS);
write_segment_file(Bins, Bases, Head, Ws, SegAddr, SS, Pos, BSize,
AddrToBe, LSize) ->
Addr = AddrToBe + element(LSize, Bases),
NoZeros = Pos - SegAddr,
NWs = [Ws, dets_utils:make_zeros(NoZeros) | <<BSize:32,Addr:32>>],
NSegAddr = SegAddr + NoZeros + ?SZOBJP*4,
write_segment_file(Bins, Bases, Head, NWs, NSegAddr, SS).
fast_write_all_sizes(Cache, SizeT, Head) ->
CacheL = lists:reverse(tuple_to_list(Cache)),
fast_write_sizes(CacheL, tuple_size(Cache), SizeT, Head, [], []).
fast_write_sizes([], _Sz, _SizeT, Head, NCL, PwriteList) ->
#head{filename = FileName, fptr = Fd} = Head,
ok = dets_utils:pwrite(Fd, FileName, PwriteList),
list_to_tuple(NCL);
fast_write_sizes([[_Addr] = C | CL], Sz, SizeT, Head, NCL, PwriteList) ->
fast_write_sizes(CL, Sz-1, SizeT, Head, [C | NCL], PwriteList);
fast_write_sizes([[Addr | C] | CL], Sz, SizeT, Head, NCL, PwriteList) ->
case ets:lookup(SizeT, Sz) of
[] ->
throw({error, invalid_objects_list});
[{Sz,Position,_ObjCounter,_NoCollections}] ->
%% Update ObjCounter:
NoColls = length(C),
_ = ets:update_counter(SizeT, Sz, {3, NoColls}),
Pos = Position + Addr - NoColls*?POW(Sz-1),
fast_write_sizes(CL, Sz-1, SizeT, Head, [[Addr] | NCL],
[{Pos,lists:reverse(C)} | PwriteList])
end.
prepare_file_init(NoObjects, NoKeys, NoObjsPerSize, SizeT, Head) ->
SegSz = ?ACTUAL_SEG_SIZE,
{_, SegEnd, _} = dets_utils:alloc(Head, adjsz(SegSz)),
Head1 = Head#head{no_objects = NoObjects, no_keys = NoKeys},
true = ets:insert(SizeT, {?FSCK_SEGMENT,0,[],0}),
lists:foreach(fun({LogSz,NoColls}) ->
true = ets:insert(SizeT, {LogSz+1,0,0,NoColls})
end, NoObjsPerSize),
{NewHead, NL0, MaxSz, EndOfFile} = allocate_all_objects(Head1, SizeT),
[{?FSCK_SEGMENT,SegAddr,[],0} | NL] = NL0,
true = ets:delete_all_objects(SizeT),
lists:foreach(fun(X) -> true = ets:insert(SizeT, X) end, NL),
Bases = lists:foldl(fun({LSz,P,_D,_N}, A) -> setelement(LSz,A,P) end,
erlang:make_tuple(MaxSz, 0), NL),
Est = lists:foldl(fun({LSz,_,_,N}, A) -> A + ?POW(LSz-1)*N end, 0, NL),
ok = write_bytes(NewHead, EndOfFile, Est),
{NewHead, Bases, SegAddr, SegEnd}.
%% Writes "zeros" to the file. This ensures that the file blocks are
%% allocated more or less contiguously, which reduces the seek times
%% to a minimum when the file is later read serially from beginning to
%% end (as is done when calling select and the like). A well-formed
%% file will be created also if nothing is written (as is the case for
%% small files, for efficiency).
write_bytes(_Head, _EndOfFile, Est) when Est < ?CACHE_SIZE ->
ok;
write_bytes(Head, EndOfFile, _Est) ->
Fd = Head#head.fptr,
{ok, Start} = file:position(Fd, eof),
BytesToWrite = EndOfFile - Start,
SizeInKB = 64,
Bin = list_to_binary(lists:duplicate(SizeInKB * 4, lists:seq(0, 255))),
write_loop(Head, BytesToWrite, Bin).
write_loop(Head, BytesToWrite, Bin) when BytesToWrite >= byte_size(Bin) ->
case file:write(Head#head.fptr, Bin) of
ok -> write_loop(Head, BytesToWrite - byte_size(Bin), Bin);
Error -> dets_utils:file_error(Error, Head#head.filename)
end;
write_loop(_Head, 0, _Bin) ->
ok;
write_loop(Head, BytesToWrite, Bin) ->
<<SmallBin:BytesToWrite/binary,_/binary>> = Bin,
write_loop(Head, BytesToWrite, SmallBin).
%% By allocating bigger objects before smaller ones, holes in the
%% buddy system memory map are avoided.
allocate_all_objects(Head, SizeT) ->
DTL = lists:reverse(lists:keysort(1, ets:tab2list(SizeT))),
MaxSz = element(1, hd(DTL)),
{Head1, NL} = allocate_all(Head, DTL, []),
%% Find the position that will be the end of the file by allocating
%% a minimal object.
{_Head, EndOfFile, _} = dets_utils:alloc(Head1, ?BUMP),
NewHead = Head1#head{maxobjsize = max_objsize(Head1#head.no_collections)},
{NewHead, NL, MaxSz, EndOfFile}.
%% One (temporary) file for each buddy size, write all objects of that
%% size to the file.
%%
%% Before R15 a "hole" was needed before the first bucket if the size
%% of the biggest bucket was greater than the size of a segment. The
%% hole proved to be a problem with almost full tables with huge
%% buckets. Since R15 the hole is no longer needed due to the fact
%% that the base of the Buddy system is flexible.
allocate_all(Head, [{?FSCK_SEGMENT,_,Data,_}], L) ->
%% And one file for the segments...
%% Note that space for the array parts and the segments has
%% already been allocated, but the segments have not been
%% initialized on disk.
NoParts = no_parts(Head#head.next),
%% All parts first, ensured by init_segments/6.
Addr = ?BASE + NoParts * 4 * ?SEGPARTSZ,
{Head, [{?FSCK_SEGMENT,Addr,Data,0} | L]};
allocate_all(Head, [{LSize,_,Data,NoCollections} | DTL], L) ->
Size = ?POW(LSize-1),
{_Head, Addr, _} = dets_utils:alloc(Head, adjsz(Size)),
Head1 = dets_utils:alloc_many(Head, Size, NoCollections, Addr),
NoColls = Head1#head.no_collections,
NewNoColls = orddict:update_counter(LSize-1, NoCollections, NoColls),
NewHead = Head1#head{no_collections = NewNoColls},
E = {LSize,Addr,Data,NoCollections},
allocate_all(NewHead, DTL, [E | L]).
bin2term(Bin, 9, Kp) ->
bin2term1(Bin, Kp, []);
bin2term(Bin, 8, Kp) ->
bin2term_v8(Bin, Kp, []).
bin2term1([<<Slot:32, Seq:32, BinTerm/binary>> | BTs], Kp, L) ->
Term = binary_to_term(BinTerm),
Key = element(Kp, Term),
bin2term1(BTs, Kp, [{Slot, Key, Seq, Term, BinTerm} | L]);
bin2term1([], _Kp, L) ->
lists:reverse(L).
bin2term_v8([<<Slot:32, BinTerm/binary>> | BTs], Kp, L) ->
Term = binary_to_term(BinTerm),
Key = element(Kp, Term),
bin2term_v8(BTs, Kp, [{Slot, Key, foo, Term, BinTerm} | L]);
bin2term_v8([], _Kp, L) ->
lists:reverse(L).
write_all_sizes({}=Cache, _SizeT, _Head, _More) ->
Cache;
write_all_sizes(Cache, SizeT, Head, More) ->
CacheL = lists:reverse(tuple_to_list(Cache)),
Sz = length(CacheL),
NCL = case ets:info(SizeT, size) of
1 when More =:= no_more -> % COUNTERS only...
all_sizes(CacheL, Sz, SizeT);
_ ->
write_sizes(CacheL, Sz, SizeT, Head)
end,
list_to_tuple(NCL).
all_sizes([]=CL, _Sz, _SizeT) ->
CL;
all_sizes([[]=C | CL], Sz, SizeT) ->
[C | all_sizes(CL, Sz-1, SizeT)];
all_sizes([C0 | CL], Sz, SizeT) ->
C = lists:reverse(C0),
NoCollections = length(C),
true = ets:insert(SizeT, {Sz,0,C,NoCollections}),
[[] | all_sizes(CL, Sz-1, SizeT)].
write_sizes([]=CL, _Sz, _SizeT, _Head) ->
CL;
write_sizes([[]=C | CL], Sz, SizeT, Head) ->
[C | write_sizes(CL, Sz-1, SizeT, Head)];
write_sizes([C | CL], Sz, SizeT, Head) ->
{FileName, Fd} =
case ets:lookup(SizeT, Sz) of
[] ->
temp_file(Head, SizeT, Sz);
[{_,_,{FN,F},_}] ->
{FN, F}
end,
NoCollections = length(C),
_ = ets:update_counter(SizeT, Sz, {4,NoCollections}),
case file:write(Fd, lists:reverse(C)) of
ok ->
[[] | write_sizes(CL, Sz-1, SizeT, Head)];
Error ->
dets_utils:file_error(FileName, Error)
end.
output_slots([E | Es], Head, Cache, SizeT, NoKeys, NoObjs) ->
output_slots(E, Es, [E], Head, Cache, SizeT, NoKeys, NoObjs);
output_slots([], _Head, Cache, SizeT, NoKeys, NoObjs) ->
_ = ets:update_counter(SizeT, ?COUNTERS, {?OBJ_COUNTER,NoObjs}),
_ = ets:update_counter(SizeT, ?COUNTERS, {?KEY_COUNTER,NoKeys}),
{not_a_tuple, [], Cache}.
output_slots(E, [E1 | Es], Acc, Head, Cache, SizeT, NoKeys, NoObjs)
when element(1, E) =:= element(1, E1) ->
output_slots(E1, Es, [E1 | Acc], Head, Cache, SizeT, NoKeys, NoObjs);
output_slots(E, [], Acc, _Head, Cache, SizeT, NoKeys, NoObjs) ->
_ = ets:update_counter(SizeT, ?COUNTERS, {?OBJ_COUNTER,NoObjs}),
_ = ets:update_counter(SizeT, ?COUNTERS, {?KEY_COUNTER,NoKeys}),
{E, Acc, Cache};
output_slots(_E, L, Acc, Head, Cache, SizeT, NoKeys, NoObjs) ->
output_slot(Acc, Head, Cache, L, SizeT, NoKeys, NoObjs).
output_slot(Es, Head, Cache, L, SizeT, NoKeys, NoObjs) ->
Slot = element(1, hd(Es)),
%% Plain lists:sort/1 will do.
{Bins, Size, No, KNo} = prep_slot(lists:sort(Es), Head),
NNoKeys = NoKeys + KNo,
NNoObjs = NoObjs + No,
%% First the object collection.
BSize = Size + ?OHDSZ,
LSize = sz2pos(BSize),
Size2 = ?POW(LSize-1),
Pad = <<0:(Size2-BSize)/unit:8>>,
BinObject = [<<BSize:32, ?ACTIVE:32>>, Bins | Pad],
Cache1 =
if
LSize > tuple_size(Cache) ->
C1 = ?VEXT(LSize, Cache, []),
?VSET(LSize, C1, [BinObject]);
true ->
CL = ?VGET(LSize, Cache),
?VSET(LSize, Cache, [BinObject | CL])
end,
%% Then the pointer to the object collection.
%% Cannot yet determine the absolute pointers; segment_file/4 does that.
PBin = <<Slot:32,BSize:32,LSize:8>>,
PL = ?VGET(?FSCK_SEGMENT, Cache1),
NCache = ?VSET(?FSCK_SEGMENT, Cache1, [PBin | PL]),
output_slots(L, Head, NCache, SizeT, NNoKeys, NNoObjs).
prep_slot(L, Head) when Head#head.type =/= set ->
prep_slot(L, Head, []);
prep_slot([{_Slot,Key,_Seq,_T,BT} | L], _Head) ->
prep_set_slot(L, Key, BT, 0, 0, 0, []).
prep_slot([{_Slot, Key, Seq, T, _BT} | L], Head, W) ->
prep_slot(L, Head, [{Key, {Seq, {insert,T}}} | W]);
prep_slot([], Head, W) ->
WLs = dets_utils:family(W),
{[], Bins, Size, No, KNo, _} =
eval_slot(WLs, [], Head#head.type, [], [], 0, 0, 0, false),
{Bins, Size, No, KNo}.
%% Optimization, prep_slot/3 would work for set tables as well.
prep_set_slot([{_,K,_Seq,_T1,BT1} | L], K, _BT, Sz, NoKeys, NoObjs, Ws) ->
prep_set_slot(L, K, BT1, Sz, NoKeys, NoObjs, Ws);
prep_set_slot([{_,K1,_Seq,_T1,BT1} | L], _K, BT, Sz, NoKeys, NoObjs, Ws) ->
BSize = byte_size(BT) + 4,
NWs = [Ws,<<BSize:32>>|BT],
prep_set_slot(L, K1, BT1, Sz+BSize, NoKeys+1, NoObjs+1, NWs);
prep_set_slot([], _K, BT, Sz, NoKeys, NoObjs, Ws) ->
BSize = byte_size(BT) + 4,
{[Ws, <<BSize:32>> | BT], Sz + BSize, NoKeys+1, NoObjs+1}.
segment_file(SizeT, Head, FileData, SegEnd) ->
I = 2,
true = ets:delete_all_objects(SizeT),
lists:foreach(fun(X) -> true = ets:insert(SizeT, X) end, FileData),
[{?FSCK_SEGMENT,SegAddr,Data,0} | FileData1] = FileData,
NewData =
case Data of
{InFile,In0} ->
{OutFile, Out} = temp_file(Head, SizeT, I),
_ = file:close(In0),
{ok, In} = dets_utils:open(InFile, [raw,binary,read]),
{ok, 0} = dets_utils:position(In, InFile, bof),
seg_file(SegAddr, SegAddr, In, InFile, Out, OutFile, SizeT,
SegEnd),
_ = file:close(In),
_ = file:delete(InFile),
{OutFile,Out};
Objects ->
{LastAddr, B} = seg_file(Objects, SegAddr, SegAddr, SizeT, []),
dets_utils:disk_map_segment(SegAddr, B),
FinalZ = SegEnd - LastAddr,
[B | dets_utils:make_zeros(FinalZ)]
end,
%% Restore the positions.
true = ets:delete_all_objects(SizeT),
%% To get the segments copied first by dets:fsck_copy/4, use a big
%% number here, FSCK_SEGMENT2.
lists:foreach(fun(X) -> true = ets:insert(SizeT, X) end,
[{?FSCK_SEGMENT2,SegAddr,NewData,0} | FileData1]),
ok.
seg_file(Addr, SS, In, InFile, Out, OutFile, SizeT, SegEnd) ->
case dets_utils:read_n(In, 4500) of
eof ->
FinalZ = SegEnd - Addr,
dets_utils:fwrite(Out, OutFile, dets_utils:make_zeros(FinalZ));
Bin ->
{NewAddr, L} = seg_file(Bin, Addr, SS, SizeT, []),
dets_utils:disk_map_segment(Addr, L),
ok = dets_utils:fwrite(Out, OutFile, L),
seg_file(NewAddr, SS, In, InFile, Out, OutFile, SizeT, SegEnd)
end.
seg_file(<<Slot:32,BSize:32,LSize:8,T/binary>>, Addr, SS, SizeT, L) ->
seg_file_item(T, Addr, SS, SizeT, L, Slot, BSize, LSize);
seg_file([<<Slot:32,BSize:32,LSize:8>> | T], Addr, SS, SizeT, L) ->
seg_file_item(T, Addr, SS, SizeT, L, Slot, BSize, LSize);
seg_file([], Addr, _SS, _SizeT, L) ->
{Addr, lists:reverse(L)};
seg_file(<<>>, Addr, _SS, _SizeT, L) ->
{Addr, lists:reverse(L)}.
seg_file_item(T, Addr, SS, SizeT, L, Slot, BSize, LSize) ->
%% Should call slot_position/1, but since all segments are
%% allocated in a sequence, the position of a slot can be
%% calculated faster.
SlotPos = SS + ?SZOBJP*4 * Slot, % SlotPos = slot_position(Slot)
NoZeros = SlotPos - Addr,
PSize = NoZeros+?SZOBJP*4,
Inc = ?POW(LSize-1),
CollP = ets:update_counter(SizeT, LSize, Inc) - Inc,
PointerBin = if
NoZeros =:= 0 ->
<<BSize:32, CollP:32>>;
NoZeros > 100 ->
[dets_utils:make_zeros(NoZeros) |
<<BSize:32, CollP:32>>];
true ->
<<0:NoZeros/unit:8, BSize:32, CollP:32>>
end,
seg_file(T, Addr + PSize, SS, SizeT, [PointerBin | L]).
temp_file(Head, SizeT, N) ->
TmpName = lists:concat([Head#head.filename, '.', N]),
{ok, Fd} = dets_utils:open(TmpName, [raw, binary, write]),
%% The file table is consulted when cleaning up.
true = ets:insert(SizeT, {N,0,{TmpName,Fd},0}),
{TmpName, Fd}.
%% Does not close Fd.
fsck_input(Head, Fd, Cntrs, FileHeader) ->
MaxSz0 = case FileHeader#fileheader.has_md5 of
true when is_integer(FileHeader#fileheader.no_colls) ->
?POW(max_objsize(FileHeader#fileheader.no_colls));
_ ->
%% The file is not compressed, so the bucket size
%% cannot exceed the filesize, for all buckets.
case file:position(Fd, eof) of
{ok, Pos} ->
Pos;
_ ->
1 bsl 32
end
end,
MaxSz = erlang:max(MaxSz0, ?CHUNK_SIZE),
State0 = fsck_read(?BASE, Fd, [], 0),
fsck_input(Head, State0, Fd, MaxSz, Cntrs).
fsck_input(Head, State, Fd, MaxSz, Cntrs) ->
fun(close) ->
ok;
(read) ->
case State of
done ->
end_of_input;
{done, L, _Seq} ->
R = count_input(Head, Cntrs, L),
{R, fsck_input(Head, done, Fd, MaxSz, Cntrs)};
{cont, L, Bin, Pos, Seq} ->
R = count_input(Head, Cntrs, L),
FR = fsck_objs(Bin, Head#head.keypos, Head, [], Seq),
NewState = fsck_read(FR, Pos, Fd, MaxSz, Head),
{R, fsck_input(Head, NewState, Fd, MaxSz, Cntrs)}
end
end.
%% The ets table Cntrs is used for counting objects per size.
count_input(Head, Cntrs, L) when Head#head.version =:= 8 ->
count_input1(Cntrs, L, []);
count_input(_Head, _Cntrs, L) ->
lists:reverse(L).
count_input1(Cntrs, [[LogSz | B] | Ts], L) ->
case catch ets:update_counter(Cntrs, LogSz, 1) of
N when is_integer(N) -> ok;
_Badarg -> true = ets:insert(Cntrs, {LogSz, 1})
end,
count_input1(Cntrs, Ts, [B | L]);
count_input1(_Cntrs, [], L) ->
L.
fsck_read(Pos, F, L, Seq) ->
case file:position(F, Pos) of
{ok, _} ->
read_more_bytes([], 0, Pos, F, L, Seq);
_Error ->
{done, L, Seq}
end.
fsck_read({more, Bin, Sz, L, Seq}, Pos, F, MaxSz, Head) when Sz > MaxSz ->
FR = skip_bytes(Bin, ?BUMP, Head#head.keypos, Head, L, Seq),
fsck_read(FR, Pos, F, MaxSz, Head);
fsck_read({more, Bin, Sz, L, Seq}, Pos, F, _MaxSz, _Head) ->
read_more_bytes(Bin, Sz, Pos, F, L, Seq);
fsck_read({new, Skip, L, Seq}, Pos, F, _MaxSz, _Head) ->
NewPos = Pos + Skip,
fsck_read(NewPos, F, L, Seq).
read_more_bytes(B, Min, Pos, F, L, Seq) ->
Max = if
Min < ?CHUNK_SIZE -> ?CHUNK_SIZE;
true -> Min
end,
case dets_utils:read_n(F, Max) of
eof ->
{done, L, Seq};
Bin ->
NewPos = Pos + byte_size(Bin),
{cont, L, list_to_binary([B, Bin]), NewPos, Seq}
end.
fsck_objs(Bin = <<Sz:32, Status:32, Tail/binary>>, Kp, Head, L, Seq) ->
if
Status =:= ?ACTIVE ->
Sz1 = Sz-?OHDSZ,
case Tail of
<<BinTerm:Sz1/binary, Tail2/binary>> ->
case catch bin2keybins(BinTerm, Head) of
{'EXIT', _Reason} ->
%% The whole collection of objects is skipped.
skip_bytes(Bin, ?BUMP, Kp, Head, L, Seq);
BOs ->
{NL, NSeq} = make_objects(BOs, Seq, Kp, Head, L),
Skip = ?POW(sz2pos(Sz)-1) - Sz,
skip_bytes(Tail2, Skip, Kp, Head, NL, NSeq)
end;
_ when byte_size(Tail) < Sz1 ->
{more, Bin, Sz, L, Seq}
end;
true ->
skip_bytes(Bin, ?BUMP, Kp, Head, L, Seq)
end;
fsck_objs(Bin, _Kp, _Head, L, Seq) ->
{more, Bin, 0, L, Seq}.
make_objects([{K,BT}|Os], Seq, Kp, Head, L) when Head#head.version =:= 8 ->
LogSz = dets_v8:sz2pos(byte_size(BT)+?OHDSZ_v8),
Slot = dets_v8:db_hash(K, Head),
Obj = [LogSz | <<Slot:32, LogSz:8, BT/binary>>],
make_objects(Os, Seq, Kp, Head, [Obj | L]);
make_objects([{K,BT} | Os], Seq, Kp, Head, L) ->
Obj = make_object(Head, K, Seq, BT),
make_objects(Os, Seq+1, Kp, Head, [Obj | L]);
make_objects([], Seq, _Kp, _Head, L) ->
{L, Seq}.
%% Inlined.
make_object(Head, Key, Seq, BT) ->
Slot = db_hash(Key, Head),
<<Slot:32, Seq:32, BT/binary>>.
%% Inlined.
skip_bytes(Bin, Skip, Kp, Head, L, Seq) ->
case Bin of
<<_:Skip/binary, Tail/binary>> ->
fsck_objs(Tail, Kp, Head, L, Seq);
_ when byte_size(Bin) < Skip ->
{new, Skip - byte_size(Bin), L, Seq}
end.
%%%
%%% End of repair, conversion and initialization of a dets file.
%%%
%% -> {NewHead, ok} | throw({Head, Error})
do_perform_save(H) ->
{ok, FreeListsPointer} = dets_utils:position(H, eof),
H1 = H#head{freelists_p = FreeListsPointer},
{FLW, FLSize} = free_lists_to_file(H1),
FileSize = FreeListsPointer + FLSize + 4,
AdjustedFileSize = case H#head.base of
?BASE -> FileSize;
Base -> FileSize - Base
end,
ok = dets_utils:write(H1, [FLW | <<AdjustedFileSize:32>>]),
FileHeader = file_header(H1, FreeListsPointer, ?CLOSED_PROPERLY),
case dets_utils:debug_mode() of
true ->
TmpHead0 = init_freelist(H1#head{fixed = false}, true),
TmpHead = TmpHead0#head{base = H1#head.base},
case
catch dets_utils:all_allocated_as_list(TmpHead)
=:= dets_utils:all_allocated_as_list(H1)
of
true ->
dets_utils:pwrite(H1, [{0, FileHeader}]);
_ ->
throw(
dets_utils:corrupt_reason(H1, {failed_to_save_free_lists,
FreeListsPointer,
TmpHead#head.freelists,
H1#head.freelists}))
end;
false ->
dets_utils:pwrite(H1, [{0, FileHeader}])
end.
file_header(Head, FreeListsPointer, ClosedProperly) ->
NoColls = case Head#head.no_collections of
undefined -> [];
NC -> NC
end,
L = orddict:merge(fun(_K, V1, V2) -> V1 + V2 end,
NoColls,
lists:map(fun(X) -> {X,0} end, lists:seq(4,?MAXBUD-1))),
CW = lists:map(fun({_LSz,N}) -> <<N:32>> end, L),
file_header(Head, FreeListsPointer, ClosedProperly, CW).
file_header(Head, FreeListsPointer, ClosedProperly, NoColls) ->
Cookie = ?MAGIC,
TypeCode = dets_utils:type_to_code(Head#head.type),
Version = ?FILE_FORMAT_VERSION,
HashMethod = hash_method_to_code(Head#head.hash_bif),
H1 = <<FreeListsPointer:32, Cookie:32, ClosedProperly:32>>,
H2 = <<TypeCode:32,
Version:32,
(Head#head.m):32,
(Head#head.next):32,
(Head#head.keypos):32,
(Head#head.no_objects):32,
(Head#head.no_keys):32,
(Head#head.min_no_slots):32,
(Head#head.max_no_slots):32,
HashMethod:32,
(Head#head.n):32>>,
DigH = [H2 | NoColls],
MD5 = case Head#head.has_md5 of
true -> erlang:md5(DigH);
false -> <<0:?MD5SZ/unit:8>>
end,
Base = case Head#head.base of
?BASE -> <<0:32>>;
FlBase -> <<FlBase:32>>
end,
[H1, DigH, MD5, Base | <<0:?RESERVED/unit:8>>].
%% Going through some trouble to avoid creating one single binary for
%% the free lists. If the free lists are huge, binary_to_term and
%% term_to_binary could otherwise stop the emulator for quite some time.
-define(MAXFREEOBJ, 4096).
-define(ENDFREE, 12345).
free_lists_to_file(H) ->
FL = dets_utils:get_freelists(H),
free_list_to_file(FL, H, 1, tuple_size(FL), [], 0).
free_list_to_file(_Ftab, _H, Pos, Sz, Ws, WsSz) when Pos > Sz ->
{[Ws | <<(4+?OHDSZ):32, ?FREE:32, ?ENDFREE:32>>], WsSz+4+?OHDSZ};
free_list_to_file(Ftab, H, Pos, Sz, Ws, WsSz) ->
Max = (?MAXFREEOBJ - 4 - ?OHDSZ) div 4,
F = fun(N, L, W, S) when N =:= 0 -> {N, L, W, S};
(N, L, W, S) ->
{L1, N1, More} =
if
N > Max ->
{lists:sublist(L, Max), Max,
{N-Max, lists:nthtail(Max, L)}};
true ->
{L, N, no_more}
end,
Size = N1*4 + 4 + ?OHDSZ,
Header = <<Size:32, ?FREE:32, Pos:32>>,
NW = [W, Header | L1],
case More of
no_more ->
{0, [], NW, S+Size};
{NN, NL} ->
ok = dets_utils:write(H, NW),
{NN, NL, [], S+Size}
end
end,
{NWs,NWsSz} = dets_utils:tree_to_bin(element(Pos, Ftab), F, Max, Ws, WsSz),
free_list_to_file(Ftab, H, Pos+1, Sz, NWs, NWsSz).
free_lists_from_file(H, Pos) ->
{ok, Pos} = dets_utils:position(H#head.fptr, H#head.filename, Pos),
FL = dets_utils:empty_free_lists(),
case catch bin_to_tree([], H, start, FL, -1, []) of
{'EXIT', _} ->
throw({error, {bad_freelists, H#head.filename}});
Ftab ->
H#head{freelists = Ftab, base = ?BASE}
end.
bin_to_tree(Bin, H, LastPos, Ftab, A0, L) ->
case Bin of
<<_Size:32,?FREE:32,?ENDFREE:32,_/binary>> when L =:= [] ->
Ftab;
<<_Size:32,?FREE:32,?ENDFREE:32,_/binary>> ->
setelement(LastPos, Ftab, dets_utils:list_to_tree(L));
<<Size:32,?FREE:32,Pos:32,T/binary>>
when byte_size(T) >= Size-4-?OHDSZ ->
{NFtab, L1, A1} =
if
Pos =/= LastPos, LastPos =/= start ->
Tree = dets_utils:list_to_tree(L),
{setelement(LastPos, Ftab, Tree), [], -1};
true ->
{Ftab, L, A0}
end,
{NL, B2, A2} = bin_to_tree1(T, Size-?OHDSZ-4, A1, L1),
bin_to_tree(B2, H, Pos, NFtab, A2, NL);
_ ->
Bin2 = dets_utils:read_n(H#head.fptr, ?MAXFREEOBJ),
bin_to_tree(list_to_binary([Bin | Bin2]), H, LastPos, Ftab, A0, L)
end.
bin_to_tree1(<<A1:32,A2:32,A3:32,A4:32,T/binary>>, Size, A, L)
when Size >= 16, A < A1, A1 < A2, A2 < A3, A3 < A4 ->
bin_to_tree1(T, Size-16, A4, [A4, A3, A2, A1 | L]);
bin_to_tree1(<<A1:32,T/binary>>, Size, A, L) when Size >= 4, A < A1 ->
bin_to_tree1(T, Size - 4, A1, [A1 | L]);
bin_to_tree1(B, 0, A, L) ->
{L, B, A}.
%% -> [term()] | throw({Head, Error})
slot_objs(H, Slot) when Slot >= H#head.next ->
'$end_of_table';
slot_objs(H, Slot) ->
{ok, _Pointer, Objects} = slot_objects(H, Slot),
Objects.
%% Inlined.
h(I, phash2) -> erlang:phash2(I); % -> [0..2^27-1]
h(I, phash) -> erlang:phash(I, ?BIG) - 1.
db_hash(Key, Head) when Head#head.hash_bif =:= phash2 ->
H = erlang:phash2(Key),
Hash = ?REM2(H, Head#head.m),
if
Hash < Head#head.n ->
?REM2(H, Head#head.m2); % H rem (2 * m)
true ->
Hash
end;
db_hash(Key, Head) ->
H = h(Key, Head#head.hash_bif),
Hash = H rem Head#head.m,
if
Hash < Head#head.n ->
H rem (Head#head.m2); % H rem (2 * m)
true ->
Hash
end.
hash_method_to_code(phash2) -> ?PHASH2;
hash_method_to_code(phash) -> ?PHASH.
code_to_hash_method(?PHASH2) -> phash2;
code_to_hash_method(?PHASH) -> phash;
code_to_hash_method(_) -> undefined.
no_slots(Head) ->
{Head#head.min_no_slots, Head#head.next, Head#head.max_no_slots}.
table_parameters(Head) ->
case Head#head.no_collections of
undefined ->
undefined; % Version 9(a)
CL ->
NoColls0 = lists:foldl(fun({_,0}, A) -> A;
(E, A) -> [E | A]
end, [], CL),
NoColls = lists:reverse(NoColls0),
#?HASH_PARMS{file_format_version = Head#head.version,
bchunk_format_version = ?BCHUNK_FORMAT_VERSION,
file = filename:basename(Head#head.filename),
type = Head#head.type,
keypos = Head#head.keypos,
hash_method = hash_method_to_code(Head#head.hash_bif),
n = Head#head.n, m = Head#head.m,
next = Head#head.next,
min = Head#head.min_no_slots,
max = Head#head.max_no_slots,
no_objects = Head#head.no_objects,
no_keys = Head#head.no_keys, no_colls = NoColls}
end.
%% Allow quite a lot when reading object collections.
-define(MAXCOLL, (10 * ?CHUNK_SIZE)).
%% Re-hashing a segment, starting with SlotStart.
%%
%% On the average, half of the keys of the slot are put in a new slot.
%% If the old slot is i, then the new slot is i+m. The new slots
%% reside in a newly allocated segment.
%%
%% -> {NewHead, ok} | throw({Head, Error})
re_hash(Head, SlotStart) ->
FromSlotPos = slot_position(SlotStart),
ToSlotPos = slot_position(SlotStart + Head#head.m),
RSpec = [{FromSlotPos, 4 * ?SEGSZ}],
{ok, [FromBin]} = dets_utils:pread(RSpec, Head),
split_bins(FromBin, Head, FromSlotPos, ToSlotPos, [], [], 0).
split_bins(<<>>, Head, _Pos1, _Pos2, _ToRead, _L, 0) ->
{Head, ok};
split_bins(<<>>, Head, Pos1, Pos2, ToRead, L, _SoFar) ->
re_hash_write(Head, ToRead, L, Pos1, Pos2);
split_bins(FB, Head, Pos1, Pos2, ToRead, L, SoFar) ->
<<Sz1:32, P1:32, FT/binary>> = FB,
<<B1:?OHDSZ/binary, _/binary>> = FB,
NSoFar = SoFar + Sz1,
NPos1 = Pos1 + ?SZOBJP*4,
NPos2 = Pos2 + ?SZOBJP*4,
if
NSoFar > ?MAXCOLL, ToRead =/= [] ->
{NewHead, ok} = re_hash_write(Head, ToRead, L, Pos1, Pos2),
split_bins(FB, NewHead, Pos1, Pos2, [], [], 0);
Sz1 =:= 0 ->
E = {skip,B1},
split_bins(FT, Head, NPos1, NPos2, ToRead, [E | L], NSoFar);
true ->
E = {Sz1,P1,B1,Pos1,Pos2},
NewToRead = [{P1,Sz1} | ToRead],
split_bins(FT, Head, NPos1, NPos2, NewToRead, [E | L], NSoFar)
end.
re_hash_write(Head, ToRead, L, Pos1, Pos2) ->
check_pread2_arg(ToRead, Head),
{ok, Bins} = dets_utils:pread(ToRead, Head),
Z = <<0:32, 0:32>>,
{Head1, BinFS, BinTS, WsB} = re_hash_slots(Bins, L, Head, Z, [],[],[]),
WPos1 = Pos1 - ?SZOBJP*4*length(L),
WPos2 = Pos2 - ?SZOBJP*4*length(L),
ToWrite = [{WPos1,BinFS}, {WPos2, BinTS} | WsB],
dets_utils:pwrite(Head1, ToWrite).
re_hash_slots(Bins, [{skip,B1} | L], Head, Z, BinFS, BinTS, WsB) ->
re_hash_slots(Bins, L, Head, Z, [B1 | BinFS], [Z | BinTS], WsB);
re_hash_slots([FB | Bins], [E | L], Head, Z, BinFS, BinTS, WsB) ->
{Sz1,P1,B1,Pos1,Pos2} = E,
KeyObjs = case catch per_key(Head, FB) of
{'EXIT', _Error} ->
Bad = dets_utils:bad_object(re_hash_slots, {FB, E}),
throw(dets_utils:corrupt_reason(Head, Bad));
Else ->
Else
end,
case re_hash_split(KeyObjs, Head, [], 0, [], 0) of
{_KL, _KSz, [], 0} ->
Sz1 = _KSz + ?OHDSZ,
re_hash_slots(Bins, L, Head, Z, [B1 | BinFS], [Z | BinTS], WsB);
{[], 0, _ML, _MSz} -> %% Optimization.
Sz1 = _MSz + ?OHDSZ,
re_hash_slots(Bins, L, Head, Z, [Z | BinFS], [B1 | BinTS], WsB);
{KL, KSz, ML, MSz} when KL =/= [], KSz > 0, ML =/= [], MSz > 0 ->
{Head1, FS1, Ws1} =
updated(Head, P1, Sz1, KSz, Pos1, KL, true, foo, bar),
{NewHead, [{Pos2,Bin2}], Ws2} =
updated(Head1, 0, 0, MSz, Pos2, ML, true, foo, bar),
NewBinFS = case FS1 of
[{Pos1,Bin1}] -> [Bin1 | BinFS];
[] -> [B1 | BinFS] % cannot happen
end,
NewBinTS = [Bin2 | BinTS],
NewWsB = Ws2 ++ Ws1 ++ WsB,
re_hash_slots(Bins, L, NewHead, Z, NewBinFS, NewBinTS, NewWsB)
end;
re_hash_slots([], [], Head, _Z, BinFS, BinTS, WsB) ->
{Head, BinFS, BinTS, lists:reverse(WsB)}.
re_hash_split([E | KeyObjs], Head, KL, KSz, ML, MSz) ->
{Key,Sz,Bin,_Item,_Objs} = E,
New = h(Key, Head#head.hash_bif) rem Head#head.m2, % h(key) rem (m * 2)
if
New >= Head#head.m ->
re_hash_split(KeyObjs, Head, KL, KSz, [Bin | ML], MSz + Sz);
true ->
re_hash_split(KeyObjs, Head, [Bin | KL], KSz + Sz, ML, MSz)
end;
re_hash_split([], _Head, KL, KSz, ML, MSz) ->
{lists:reverse(KL), KSz, lists:reverse(ML), MSz}.
%% -> {NewHead, [LookedUpObject], pwrite_list()} | throw({NewHead, Error})
write_cache(Head) ->
C = Head#head.cache,
case dets_utils:is_empty_cache(C) of
true -> {Head, [], []};
false ->
{NewC, MaxInserts, PerKey} = dets_utils:reset_cache(C),
%% MaxNoInsertedKeys is an upper limit on the number of new keys.
MaxNoInsertedKeys = erlang:min(MaxInserts, length(PerKey)),
Head1 = Head#head{cache = NewC},
case may_grow(Head1, MaxNoInsertedKeys, once) of
{Head2, ok} ->
eval_work_list(Head2, PerKey);
HeadError ->
throw(HeadError)
end
end.
%% -> {NewHead, ok} | {NewHead, Error}
may_grow(Head, 0, once) ->
%% Do not re-hash if there is a chance that the file is not dirty.
{Head, ok};
may_grow(Head, _N, _How) when Head#head.fixed =/= false ->
{Head, ok};
may_grow(#head{access = read}=Head, _N, _How) ->
{Head, ok};
may_grow(Head, _N, _How) when Head#head.next >= Head#head.max_no_slots ->
{Head, ok};
may_grow(Head, N, How) ->
Extra = erlang:min(2*?SEGSZP, Head#head.no_keys + N - Head#head.next),
case catch may_grow1(Head, Extra, How) of
{error, _Reason} = Error -> % alloc may throw error
dets_utils:corrupt(Head, Error);
{NewHead, Reply} when is_record(Head, head) ->
{NewHead, Reply}
end.
may_grow1(Head, Extra, many_times) when Extra > ?SEGSZP ->
Reply = grow(Head, 1, undefined),
self() ! ?DETS_CALL(self(), may_grow),
Reply;
may_grow1(Head, Extra, _How) ->
grow(Head, Extra, undefined).
%% -> {Head, ok} | throw({Head, Error})
grow(Head, Extra, _SegZero) when Extra =< 0 ->
{Head, ok};
grow(Head, Extra, undefined) ->
grow(Head, Extra, seg_zero());
grow(Head, _Extra, _SegZero) when Head#head.next >= Head#head.max_no_slots ->
{Head, ok};
grow(Head, Extra, SegZero) ->
#head{n = N, next = Next, m = M} = Head,
SegNum = Next div ?SEGSZP,
{Head0, W, Ws1} = allocate_segment(Head, SegZero, SegNum),
%% re_hash/2 will overwrite the segment, but initialize it anyway...
{Head1, ok} = dets_utils:pwrite(Head0, [W | Ws1]),
%% If re_hash fails, segp_cache has been called, but it does not matter.
{Head2, ok} = re_hash(Head1, N),
NewHead =
if
N + ?SEGSZP =:= M ->
Head2#head{n = 0, next = Next + ?SEGSZP, m = 2 * M, m2 = 4 * M};
true ->
Head2#head{n = N + ?SEGSZP, next = Next + ?SEGSZP}
end,
true = hash_invars(NewHead),
grow(NewHead, Extra - ?SEGSZP, SegZero).
hash_invars(H) ->
hash_invars(H#head.n, H#head.m, H#head.next, H#head.min_no_slots,
H#head.max_no_slots).
-define(M8(X), (((X) band (?SEGSZP - 1)) =:= 0)).
hash_invars(N, M, Next, Min, Max) ->
?M8(N) and ?M8(M) and ?M8(Next) and ?M8(Min) and ?M8(Max)
and (0 =< N) and (N =< M) and (N =< 2*Next) and (M =< Next)
and (Next =< 2*M) and (0 =< Min) and (Min =< Next) and (Next =< Max)
and (Min =< M).
seg_zero() ->
<<0:(4*?SEGSZ)/unit:8>>.
find_object(Head, Object) ->
Key = element(Head#head.keypos, Object),
Slot = db_hash(Key, Head),
find_object(Head, Object, Slot).
find_object(H, _Obj, Slot) when Slot >= H#head.next ->
false;
find_object(H, Obj, Slot) ->
case catch slot_objects(H, Slot) of
{ok, Pointer, Objects} ->
case lists:member(Obj, Objects) of
true -> {ok, Pointer};
false -> false
end;
_ -> false
end.
%% -> {ok, BucketP, Objects} | throw({Head, Error})
slot_objects(Head, Slot) ->
SlotPos = slot_position(Slot),
MaxSize = maxobjsize(Head),
case dets_utils:ipread(Head, SlotPos, MaxSize) of
{ok, {BucketSz, Pointer, <<BucketSz:32, _St:32, KeysObjs/binary>>}} ->
case catch bin2objs(KeysObjs, Head#head.type, []) of
{'EXIT', _Error} ->
Bad = dets_utils:bad_object(slot_objects,
{SlotPos, KeysObjs}),
throw(dets_utils:corrupt_reason(Head, Bad));
Objs when is_list(Objs) ->
{ok, Pointer, lists:reverse(Objs)}
end;
[] ->
{ok, 0, []};
BadRead -> % eof or bad badly formed binary
Bad = dets_utils:bad_object(slot_objects, {SlotPos, BadRead}),
throw(dets_utils:corrupt_reason(Head, Bad))
end.
%%%
%%% Cache routines depending on the dets file format.
%%%
%% -> {Head, [LookedUpObject], pwrite_list()} | throw({Head, Error})
eval_work_list(Head, [{Key,[{_Seq,{lookup,Pid}}]}]) ->
SlotPos = slot_position(db_hash(Key, Head)),
MaxSize = maxobjsize(Head),
Objs = case dets_utils:ipread(Head, SlotPos, MaxSize) of
{ok, {_BucketSz, _Pointer, Bin}} ->
case catch per_key(Head, Bin) of
{'EXIT', _Error} ->
Bad = dets_utils:bad_object(eval_work_list,
{SlotPos, Bin}),
throw(dets_utils:corrupt_reason(Head, Bad));
KeyObjs when is_list(KeyObjs) ->
case dets_utils:mkeysearch(Key, 1, KeyObjs) of
false ->
[];
{value, {Key,_KS,_KB,O,Os}} ->
case catch binobjs2terms(Os) of
{'EXIT', _Error} ->
Bad = dets_utils:bad_object
(eval_work_list,
{SlotPos, Bin, KeyObjs}),
throw(dets_utils:corrupt_reason
(Head, Bad));
Terms when is_list(Terms) ->
get_objects([O | Terms])
end
end
end;
[] ->
[];
BadRead -> % eof or bad badly formed binary
Bad = dets_utils:bad_object(eval_work_list,
{SlotPos, BadRead}),
throw(dets_utils:corrupt_reason(Head, Bad))
end,
{Head, [{Pid,Objs}], []};
eval_work_list(Head, PerKey) ->
SWLs = tag_with_slot(PerKey, Head, []),
P1 = dets_utils:family(SWLs),
{PerSlot, SlotPositions} = remove_slot_tag(P1, [], []),
{ok, Bins} = dets_utils:pread(SlotPositions, Head),
read_buckets(PerSlot, SlotPositions, Bins, Head, [], [], [], [], 0, 0, 0).
tag_with_slot([{K,_} = WL | WLs], Head, L) ->
tag_with_slot(WLs, Head, [{db_hash(K, Head), WL} | L]);
tag_with_slot([], _Head, L) ->
L.
remove_slot_tag([{S,SWLs} | SSWLs], Ls, SPs) ->
remove_slot_tag(SSWLs, [SWLs | Ls], [{slot_position(S), ?SEGOBJSZ} | SPs]);
remove_slot_tag([], Ls, SPs) ->
{Ls, SPs}.
read_buckets([WLs | SPs], [{P1,_8} | Ss], [<<_Zero:32,P2:32>> | Bs], Head,
PWLs, ToRead, LU, Ws, NoObjs, NoKeys, SoFar) when P2 =:= 0 ->
{NewHead, NLU, NWs, No, KNo} =
eval_bucket_keys(WLs, P1, 0, 0, [], Head, Ws, LU),
NewNoObjs = No + NoObjs,
NewNoKeys = KNo + NoKeys,
read_buckets(SPs, Ss, Bs, NewHead, PWLs, ToRead, NLU, NWs,
NewNoObjs, NewNoKeys, SoFar);
read_buckets([WorkLists| SPs], [{P1,_8} | Ss], [<<Size:32,P2:32>> | Bs], Head,
PWLs, ToRead, LU, Ws, NoObjs, NoKeys, SoFar)
when SoFar + Size < ?MAXCOLL; ToRead =:= [] ->
NewToRead = [{P2, Size} | ToRead],
NewPWLs = [{P2,P1,WorkLists} | PWLs],
NewSoFar = SoFar + Size,
read_buckets(SPs, Ss, Bs, Head, NewPWLs, NewToRead, LU, Ws,
NoObjs, NoKeys, NewSoFar);
read_buckets(SPs, Ss, Bs, Head, PWLs0, ToRead0, LU, Ws, NoObjs, NoKeys, SoFar)
when SoFar > 0 ->
%% It pays off to sort the positions. The seek times are reduced,
%% at least if the file blocks are reasonably contiguous, as is
%% often the case.
PWLs = lists:keysort(1, PWLs0),
ToRead = lists:keysort(1, ToRead0),
check_pread2_arg(ToRead, Head),
{ok, Bins} = dets_utils:pread(ToRead, Head),
case catch eval_buckets(Bins, PWLs, Head, LU, Ws, 0, 0) of
{ok, NewHead, NLU, [], 0, 0} ->
read_buckets(SPs, Ss, Bs, NewHead, [], [], NLU, [],
NoObjs, NoKeys, 0);
{ok, Head1, NLU, NWs, No, KNo} ->
NewNoObjs = NoObjs + No,
NewNoKeys = NoKeys + KNo,
%% It does not seem to reduce seek times to sort positions
%% when writing (maybe because it takes several calls to
%% write_cache/1 to fill the file system's buffer cache).
{NewHead, ok} = dets_utils:pwrite(Head1, lists:reverse(NWs)),
read_buckets(SPs, Ss, Bs, NewHead, [], [], NLU, [],
NewNoObjs, NewNoKeys, 0);
Error ->
Bad = dets_utils:bad_object(read_buckets, {Bins, Error}),
throw(dets_utils:corrupt_reason(Head, Bad))
end;
read_buckets([], [], [], Head, [], [], LU, Ws, NoObjs, NoKeys, 0) ->
{NewHead, NWs} = update_no_keys(Head, Ws, NoObjs, NoKeys),
{NewHead, LU, lists:reverse(NWs)}.
eval_buckets([Bin | Bins], [SP | SPs], Head, LU, Ws, NoObjs, NoKeys) ->
{Pos, P1, WLs} = SP,
KeyObjs = per_key(Head, Bin),
{NewHead, NLU, NWs, No, KNo} =
eval_bucket_keys(WLs, P1, Pos, byte_size(Bin), KeyObjs, Head,Ws,LU),
eval_buckets(Bins, SPs, NewHead, NLU, NWs, NoObjs + No, NoKeys + KNo);
eval_buckets([], [], Head, LU, Ws, NoObjs, NoKeys) ->
{ok, Head, LU, Ws, NoObjs, NoKeys}.
eval_bucket_keys(WLs, SlotPos, Pos, OldSize, KeyObjs, Head, Ws, LU) ->
{NLU, Bins, BSize, No, KNo, Ch} =
eval_slot(WLs, KeyObjs, Head#head.type, LU, [], 0, 0, 0, false),
{NewHead, W1, W2} =
updated(Head, Pos, OldSize, BSize, SlotPos, Bins, Ch, No, KNo),
{NewHead, NLU, W2++W1++Ws, No, KNo}.
updated(Head, Pos, OldSize, BSize, SlotPos, Bins, Ch, DeltaNoOs, DeltaNoKs) ->
BinsSize = BSize + ?OHDSZ,
if
Pos =:= 0, BSize =:= 0 ->
{Head, [], []};
Pos =:= 0, BSize > 0 ->
{Head1, NewPos, FPos} = dets_utils:alloc(Head, adjsz(BinsSize)),
NewHead = one_bucket_added(Head1, FPos-1),
W1 = {NewPos, [<<BinsSize:32, ?ACTIVE:32>> | Bins]},
W2 = {SlotPos, <<BinsSize:32, NewPos:32>>},
{NewHead, [W2], [W1]};
Pos =/= 0, BSize =:= 0 ->
{Head1, FPos} = dets_utils:free(Head, Pos, adjsz(OldSize)),
NewHead = one_bucket_removed(Head1, FPos-1),
W1 = {Pos+?STATUS_POS, <<?FREE:32>>},
W2 = {SlotPos, <<0:32, 0:32>>},
{NewHead, [W2], [W1]};
Pos =/= 0, BSize > 0, Ch =:= false ->
{Head, [], []};
Pos =/= 0, BSize > 0 ->
%% Doubtful. The scan function has to be careful since
%% partly scanned objects may be overwritten.
Overwrite0 = if
OldSize =:= BinsSize -> same;
true -> sz2pos(OldSize) =:= sz2pos(BinsSize)
end,
Overwrite = if
Head#head.fixed =/= false ->
%% Make sure that if the table is
%% fixed, nothing is overwritten,
%% unless the number of objects and
%% the number of keys remain the same.
%% This is used by bchunk, which
%% assumes that it traverses exactly
%% the same number of objects and keys
%% (and collections) as were present
%% when chunking started (the table
%% must have been fixed).
(Overwrite0 =/= false) and
(DeltaNoOs =:= 0) and (DeltaNoKs =:= 0);
true ->
Overwrite0
end,
if
Overwrite =:= same ->
W1 = {Pos+?OHDSZ, Bins},
{Head, [], [W1]};
Overwrite ->
W1 = {Pos, [<<BinsSize:32, ?ACTIVE:32>> | Bins]},
%% Pos is already there, but return {SlotPos, <8 bytes>}.
W2 = {SlotPos, <<BinsSize:32, Pos:32>>},
{Head, [W2], [W1]};
true ->
{Head1, FPosF} = dets_utils:free(Head, Pos, adjsz(OldSize)),
{Head2, NewPos, FPosA} =
dets_utils:alloc(Head1, adjsz(BinsSize)),
Head3 = one_bucket_added(Head2, FPosA-1),
NewHead = one_bucket_removed(Head3, FPosF-1),
W0 = {NewPos, [<<BinsSize:32, ?ACTIVE:32>> | Bins]},
W2 = {SlotPos, <<BinsSize:32, NewPos:32>>},
W1 = if
Pos =/= NewPos ->
%% W0 first.
[W0, {Pos+?STATUS_POS, <<?FREE:32>>}];
true ->
[W0]
end,
{NewHead, [W2], W1}
end
end.
one_bucket_added(H, _Log2) when H#head.no_collections =:= undefined ->
H;
one_bucket_added(H, Log2) when H#head.maxobjsize >= Log2 ->
NewNoColls = orddict:update_counter(Log2, 1, H#head.no_collections),
H#head{no_collections = NewNoColls};
one_bucket_added(H, Log2) ->
NewNoColls = orddict:update_counter(Log2, 1, H#head.no_collections),
H#head{no_collections = NewNoColls, maxobjsize = Log2}.
one_bucket_removed(H, _FPos) when H#head.no_collections =:= undefined ->
H;
one_bucket_removed(H, Log2) when H#head.maxobjsize > Log2 ->
NewNoColls = orddict:update_counter(Log2, -1, H#head.no_collections),
H#head{no_collections = NewNoColls};
one_bucket_removed(H, Log2) when H#head.maxobjsize =:= Log2 ->
NewNoColls = orddict:update_counter(Log2, -1, H#head.no_collections),
MaxObjSize = max_objsize(NewNoColls),
H#head{no_collections = NewNoColls, maxobjsize = MaxObjSize}.
eval_slot([{Key,Commands} | WLs] = WLs0, [{K,KS,KB,O,Os} | KOs1]=KOs,
Type, LU, Ws, No, KNo,BSz, Ch) ->
case dets_utils:cmp(K, Key) of
0 ->
Old = [O | binobjs2terms(Os)],
{NLU, NWs, Sz, No1, KNo1, NCh} =
eval_key(Key, Commands, Old, Type, KB, KS, LU, Ws, Ch),
eval_slot(WLs, KOs1, Type, NLU, NWs, No1 + No,
KNo1 + KNo, Sz + BSz, NCh);
-1 ->
eval_slot(WLs0, KOs1, Type, LU, [Ws | KB], No,
KNo, KS + BSz, Ch);
1 ->
{NLU, NWs, Sz, No1, KNo1, NCh} =
eval_key(Key, Commands, [], Type, [], 0, LU, Ws, Ch),
eval_slot(WLs, KOs, Type, NLU, NWs, No1 + No,
KNo1 + KNo, Sz + BSz, NCh)
end;
eval_slot([{Key,Commands} | WLs], [], Type, LU, Ws, No, KNo,BSz, Ch) ->
{NLU, NWs, Sz, No1, KNo1, NCh} =
eval_key(Key, Commands, [], Type, [], 0, LU, Ws, Ch),
eval_slot(WLs, [], Type, NLU, NWs, No1 + No, KNo1 + KNo, Sz + BSz, NCh);
eval_slot([], [{_Key,Size,KeyBin,_,_} | KOs], Type, LU, Ws, No, KNo,BSz, Ch) ->
eval_slot([], KOs, Type, LU, [Ws | KeyBin], No, KNo, Size + BSz, Ch);
eval_slot([], [], _Type, LU, Ws, No, KNo, BSz, Ch) ->
{LU, Ws, BSz, No, KNo, Ch}.
eval_key(_K, [{_Seq,{lookup,Pid}}], [], _Type, _KeyBin, _KeySz, LU, Ws, Ch) ->
NLU = [{Pid, []} | LU],
{NLU, Ws, 0, 0, 0, Ch};
eval_key(_K, [{_Seq,{lookup,Pid}}], Old0, _Type, KeyBin, KeySz, LU, Ws, Ch) ->
Old = lists:keysort(2, Old0), % sort on sequence number
Objs = get_objects(Old),
NLU = [{Pid, Objs} | LU],
{NLU, [Ws | KeyBin], KeySz, 0, 0, Ch};
eval_key(K, Comms, Orig, Type, KeyBin, KeySz, LU, Ws, Ch) ->
Old = dets_utils:msort(Orig),
case eval_key1(Comms, [], Old, Type, K, LU, Ws, 0, Orig) of
{ok, NLU} when Old =:= [] ->
{NLU, Ws, 0, 0, 0, Ch};
{ok, NLU} ->
{NLU, [Ws | KeyBin], KeySz, 0, 0, Ch};
{NLU, NWs, NSz, No} when Old =:= [], NSz > 0 ->
{NLU, NWs, NSz, No, 1, true};
{NLU, NWs, NSz, No} when Old =/= [], NSz =:= 0 ->
{NLU, NWs, NSz, No, -1, true};
{NLU, NWs, NSz, No} ->
{NLU, NWs, NSz, No, 0, true}
end.
%% First find 'delete_key' and 'lookup' commands, and handle the 'set' type.
eval_key1([{_Seq,{insert,Term}} | L], Cs, [{Term,_,_}] = Old, Type=set, K,
LU, Ws, No, Orig) ->
eval_key1(L, Cs, Old, Type, K, LU, Ws, No, Orig);
eval_key1([{Seq,{insert,Term}} | L], Cs, Old, Type=set, K, LU, Ws, No, Orig)
->
NNo = No + 1 - length(Old),
eval_key1(L, Cs, [{Term,Seq,insert}], Type, K, LU, Ws, NNo, Orig);
eval_key1([{_Seq,{lookup,Pid}} | L], Cs, Old, Type, Key, LU, Ws, No, Orig) ->
{ok, New0, NewNo} = eval_comms(Cs, Old, Type, No),
New = lists:keysort(2, New0), % sort on sequence number
Objs = get_objects(New),
NLU = [{Pid, Objs} | LU],
if
L =:= [] ->
eval_end(New, NLU, Type, Ws, NewNo, Orig);
true ->
NewOld = dets_utils:msort(New),
eval_key1(L, [], NewOld, Type, Key, NLU, Ws, NewNo, Orig)
end;
eval_key1([{_Seq,delete_key} | L], _Cs, Old, Type, K, LU, Ws, No, Orig) ->
NewNo = No - length(Old),
eval_key1(L, [], [], Type, K, LU, Ws, NewNo, Orig);
eval_key1([{_Seq,{delete_object,Term}} | L], Cs, [{Term,_,_}], Type=set, K,
LU, Ws, No, Orig) ->
eval_key1(L, Cs, [], Type, K, LU, Ws, No-1, Orig);
eval_key1([{_Seq,{delete_object,_T}}| L], Cs, Old1, Type=set, K, LU,
Ws, No, Orig) ->
eval_key1(L, Cs, Old1, Type, K, LU, Ws, No, Orig);
eval_key1([{Seq,{Comm,Term}} | L], Cs, Old, Type, K, LU, Ws, No, Orig)
when Type =/= set ->
eval_key1(L, [{Term,Seq,Comm} | Cs], Old, Type, K, LU, Ws, No, Orig);
eval_key1([], Cs, Old, Type=set, _Key, LU, Ws, No, Orig) ->
[] = Cs,
eval_end(Old, LU, Type, Ws, No, Orig);
eval_key1([], Cs, Old, Type, _Key, LU, Ws, No, Orig) ->
{ok, New, NewNo} = eval_comms(Cs, Old, Type, No),
eval_end(New, LU, Type, Ws, NewNo, Orig).
eval_comms([], L, _Type=set, No) ->
{ok, L, No};
eval_comms(Cs, Old, Type, No) ->
Commands = dets_utils:msort(Cs),
case Type of
bag -> eval_bag(Commands, Old, [], No);
duplicate_bag -> eval_dupbag(Commands, Old, [], No)
end.
eval_end(New0, LU, Type, Ws, NewNo, Orig) ->
New = lists:keysort(2, New0), % sort on sequence number
NoChange = if
length(New) =/= length(Orig) -> false;
true ->
same_terms(Orig, New)
end,
if
NoChange ->
%% The key's objects have not changed.
{ok, LU};
New =:= [] ->
{LU, Ws, 0, NewNo};
true ->
{Ws1, Sz} = make_bins(New, [], 0),
if
Type =:= set ->
{LU, [Ws | Ws1], Sz, NewNo};
true ->
NSz = Sz + 4,
{LU, [Ws, <<NSz:32>> | Ws1], NSz, NewNo}
end
end.
same_terms([E1 | L1], [E2 | L2]) when element(1, E1) =:= element(1, E2) ->
same_terms(L1, L2);
same_terms([], []) ->
true;
same_terms(_L1, _L2) ->
false.
make_bins([{_Term,_Seq,B} | L], W, Sz) when is_binary(B) ->
make_bins(L, [W | B], Sz + byte_size(B));
make_bins([{Term,_Seq,insert} | L], W, Sz) ->
B = term_to_binary(Term),
BSize = byte_size(B) + 4,
make_bins(L, [W, [<<BSize:32>> | B]], Sz + BSize);
make_bins([], W, Sz) ->
{W, Sz}.
get_objects([{T,_S,_BT} | L]) ->
[T | get_objects(L)];
get_objects([]) ->
[].
eval_bag([{Term1,_S1,Op}=N | L]=L0, [{Term2,_,_}=O | Old]=Old0, New, No) ->
case {Op, dets_utils:cmp(Term1, Term2)} of
{delete_object, -1} ->
eval_bag(L, Old0, New, No);
{insert, -1} ->
bag_object(L, Old0, New, No, [N], Term1);
{delete_object, 0} ->
bag_object(L, Old, New, No-1, [], Term1);
{insert, 0} ->
bag_object(L, Old, New, No-1, [N], Term1);
{_, 1} ->
eval_bag(L0, Old, [O | New], No)
end;
eval_bag([{_Term1,_Seq1,delete_object} | L], []=Old, New, No) ->
eval_bag(L, Old, New, No);
eval_bag([{Term,_Seq1,insert} = N | L], []=Old, New, No) ->
bag_object(L, Old, New, No, [N], Term);
eval_bag([]=L, [O | Old], New, No) ->
eval_bag(L, Old, [O | New], No);
eval_bag([], [], New, No) ->
{ok, New, No}.
bag_object([{Term,_,insert} = N | L], Old, New, No, _N, Term) ->
bag_object(L, Old, New, No, [N], Term);
bag_object([{Term,_,delete_object} | L], Old, New, No, _N, Term) ->
bag_object(L, Old, New, No, [], Term);
bag_object(L, Old, New, No, [], _Term) ->
eval_bag(L, Old, New, No);
bag_object(L, Old, New, No, [N], _Term) ->
eval_bag(L, Old, [N | New], No+1).
eval_dupbag([{Term1,_S1,Op}=N | L]=L0, [{Term2,_,_}=O | Old]=Old0, New, No) ->
case {Op, dets_utils:cmp(Term1, Term2)} of
{delete_object, -1} ->
eval_dupbag(L, Old0, New, No);
{insert, -1} ->
dup_object(L, Old0, New, No+1, Term1, [N]);
{_, 0} ->
old_dup_object(L0, Old, New, No, Term1, [O]);
{_, 1} ->
eval_dupbag(L0, Old, [O | New], No)
end;
eval_dupbag([{_Term1,_Seq1,delete_object} | L], []=Old, New, No) ->
eval_dupbag(L, Old, New, No);
eval_dupbag([{Term,_Seq1,insert} = N | L], []=Old, New, No) ->
dup_object(L, Old, New, No+1, Term, [N]);
eval_dupbag([]=L, [O | Old], New, No) ->
eval_dupbag(L, Old, [O | New], No);
eval_dupbag([], [], New, No) ->
{ok, New, No}.
old_dup_object(L, [{Term,_,_} = Obj | Old], New, No, Term, N) ->
old_dup_object(L, Old, New, No, Term, [Obj | N]);
old_dup_object(L, Old, New, No, Term, N) ->
dup_object(L, Old, New, No, Term, N).
dup_object([{Term,_,insert} = Obj | L], Old, New, No, Term, Q) ->
dup_object(L, Old, New, No+1, Term, [Obj | Q]);
dup_object([{Term,_Seq,delete_object} | L], Old, New, No, Term, Q) ->
%% All objects are deleted.
NewNo = No - length(Q),
dup_object(L, Old, New, NewNo, Term, []);
dup_object(L, Old, New, No, _Term, Q) ->
eval_dupbag(L, Old, Q ++ New, No).
%% Update no_keys on the file too, if the number of segments that
%% dets:fsck/6 uses for estimate has changed.
update_no_keys(Head, Ws, 0, 0) -> {Head, Ws};
update_no_keys(Head, Ws, DeltaObjects, DeltaKeys) ->
NoKeys = Head#head.no_keys,
NewNoKeys = NoKeys + DeltaKeys,
NewNoObject = Head#head.no_objects + DeltaObjects,
NewHead = Head#head{no_objects = NewNoObject, no_keys = NewNoKeys},
NWs =
if
NewNoKeys > NewHead#head.max_no_slots ->
Ws;
NoKeys div ?SEGSZP =:= NewNoKeys div ?SEGSZP ->
Ws;
true ->
[{0, file_header(NewHead, 0, ?NOT_PROPERLY_CLOSED)} | Ws]
end,
{NewHead, NWs}.
slot_position(S) ->
SegNo = ?SLOT2SEG(S), % S div ?SEGSZP
PartPos = ?SEGARRADDR(?SEG2SEGARRPART(SegNo)), % SegNo div ?SEGPARTSZ
Part = get_arrpart(PartPos),
Pos = ?SEGPARTADDR(Part, SegNo),
get_segp(Pos) + (?SEGOBJSZ * ?REM2(S, ?SEGSZP)).
check_pread2_arg([{_Pos,Sz}], Head) when Sz > ?MAXCOLL ->
case check_pread_arg(Sz, Head) of
true ->
ok;
false ->
Bad = dets_utils:bad_object(check_pread2_arg, Sz),
throw(dets_utils:corrupt_reason(Head, Bad))
end;
check_pread2_arg(_ToRead, _Head) ->
ok.
check_pread_arg(Sz, Head) when Sz > ?MAXCOLL ->
maxobjsize(Head) >= Sz;
check_pread_arg(_Sz, _Head) ->
true.
%% Inlined.
segp_cache(Pos, Segment) ->
put(Pos, Segment).
%% Inlined.
get_segp(Pos) ->
get(Pos).
arrpart_cache(Pos, ArrPart) ->
put(Pos, ArrPart).
%% Inlined.
get_arrpart(Pos) ->
get(Pos).
sz2pos(N) ->
1 + dets_utils:log2(N).
%% Inlined. Compensates for the bug in dets_utils:sz2pos/1.
adjsz(N) ->
N-1.
%% Inlined.
maxobjsize(Head) when Head#head.maxobjsize =:= undefined ->
?POW(32);
maxobjsize(Head) ->
?POW(Head#head.maxobjsize).
scan_objs(Head, Bin, From, To, L, Ts, R, Type) ->
case catch scan_skip(Bin, From, To, L, Ts, R, Type, 0) of
{'EXIT', _Reason} ->
bad_object;
Reply = {more, _From1, _To, _L, _Ts, _R, Size} when Size > ?MAXCOLL ->
case check_pread_arg(Size, Head) of
true -> Reply;
false -> bad_object
end;
Reply ->
Reply
end.
scan_skip(Bin, From, To, L, Ts, R, Type, Skip) ->
From1 = From + Skip,
case Bin of
_ when From1 >= To ->
if
From1 > To; L =:= <<>> ->
{more, From1, To, L, Ts, R, 0};
true ->
<<From2:32, To1:32, L1/binary>> = L,
Skip1 = From2 - From,
scan_skip(Bin, From, To1, L1, Ts, R, Type, Skip1)
end;
<<_:Skip/binary, _Size:32, St:32, _Sz:32, KO/binary>>
when St =/= ?ACTIVE, St =/= ?FREE ->
%% Neither ?ACTIVE nor ?FREE is a multiple of ?BUMP and
%% thus cannot be found in segments or segment array
%% parts.
scan_skip(KO, From1+12, To, L, Ts, R, Type, ?ACTUAL_SEG_SIZE-12);
<<_:Skip/binary, Size:32, _St:32, Sz:32, KO/binary>>
when Size-12 =< byte_size(KO) ->
%% St = ?FREE means that the object was deleted after
%% scanning started
bin2bins(KO, From1+12, To, L, Ts, R, Type, Size, Sz);
<<_:Skip/binary, Size:32, _St:32, _Sz:32, _KO/binary>> ->
{more, From1, To, L, Ts, R, Size};
_ when Skip >= 0 ->
{more, From1, To, L, Ts, R, 0}
end.
%% Appends objects in reversed order. All objects of the slot are
%% extracted. Note that binary_to_term/1 ignores garbage at the end.
bin2bins(Bin, From, To, L, Ts, R, Type=set, Size, ObjSz0) ->
ObjsSz1 = Size - ObjSz0,
if
ObjsSz1 =:= ?OHDSZ ->
slot_end(Bin, From, To, L, [Bin | Ts], R, Type, Size, 1);
true ->
ObjSz = ObjSz0-4,
<<_:ObjSz/binary, NObjSz:32, T/binary>> = Bin,
bins_set(T, From, To, L, [Bin | Ts], R, Type, Size, 2,
NObjSz, ObjsSz1-NObjSz, Bin)
end;
bin2bins(<<ObjSz:32, Bin/binary>> = KO, From, To, L, Ts, R, Type, Size, Sz) ->
bins_bag(Bin, From, To, L, Ts, R, Type, Size, 1,
Sz-ObjSz-4, ObjSz-4, Size-Sz, KO).
bins_set(Bin, From, To, L, Ts, R, Type, Size, NoObjs, _ObjSz0, ?OHDSZ, KO) ->
slot_end(KO, From, To, L, [Bin | Ts], R, Type, Size, NoObjs);
bins_set(Bin, From, To, L, Ts, R, Type, Size, NoObjs, ObjSz0, ObjsSz, KO) ->
ObjSz = ObjSz0 - 4,
<<_:ObjSz/binary, NObjSz:32, T/binary>> = Bin,
bins_set(T, From, To, L, [Bin | Ts], R, Type, Size, NoObjs + 1,
NObjSz, ObjsSz-NObjSz, KO).
bins_bag(Bin, From, To, L, Ts, R, Type, Size, NoObjs, Sz, ObjSz, ObjsSz, KO)
when Sz > 0 ->
<<_:ObjSz/binary, NObjSz:32, T/binary>> = Bin,
bins_bag(T, From, To, L, [Bin | Ts], R, Type, Size, NoObjs + 1,
Sz-NObjSz, NObjSz-4, ObjsSz, KO);
bins_bag(Bin, From, To, L, Ts, R, Type, Size, NoObjs, _Z, _ObjSz, ?OHDSZ, KO) ->
slot_end(KO, From, To, L, [Bin | Ts], R, Type, Size, NoObjs);
bins_bag(Bin, From, To, L, Ts, R, Type, Size, NoObjs, _Z, ObjSz, ObjsSz, KO) ->
<<_:ObjSz/binary, Sz:32, NObjSz:32, T/binary>> = Bin,
bins_bag(T, From, To, L, [Bin | Ts], R, Type, Size, NoObjs + 1,
Sz-NObjSz-4, NObjSz-4, ObjsSz-Sz, KO).
slot_end(KO, From, To, L, Ts, R, Type, Size, NoObjs) ->
Skip = ?POW(dets_utils:log2(Size)) - 12, % expensive...
if
R >= 0 ->
scan_skip(KO, From, To, L, Ts, R+Size, Type, Skip);
true ->
%% Should check this at the end of every key.
case R + NoObjs of
R1 when R1 >= -1 ->
From1 = From + Skip,
Bin1 = case KO of
<<_:Skip/binary, B/binary>> -> B;
_ -> <<>>
end,
{stop, Bin1, From1, To, L, Ts};
R1 ->
scan_skip(KO, From, To, L, Ts, R1, Type, Skip)
end
end.
%%%%%%%%%%%%%%%%% DEBUG functions %%%%%%%%%%%%%%%%
file_info(FH) ->
#fileheader{closed_properly = CP, keypos = Kp,
m = M, next = Next, n = N, version = Version,
type = Type, no_objects = NoObjects, no_keys = NoKeys}
= FH,
if
CP =:= 0 ->
{error, not_closed};
FH#fileheader.cookie =/= ?MAGIC ->
{error, not_a_dets_file};
FH#fileheader.version =/= ?FILE_FORMAT_VERSION ->
{error, bad_version};
true ->
{ok, [{closed_properly,CP},{keypos,Kp},{m, M},{n,N},
{next,Next},{no_objects,NoObjects},{no_keys,NoKeys},
{type,Type},{version,Version}]}
end.
v_segments(#head{}=H) ->
v_parts(H, 0, 0).
v_parts(_H, ?SEGARRSZ, _SegNo) ->
done;
v_parts(H, PartNo, SegNo) ->
Fd = H#head.fptr,
PartPos = dets_utils:read_4(Fd, ?SEGARRADDR(PartNo)),
if
PartPos =:= 0 ->
done;
true ->
PartBin = dets_utils:pread_n(Fd, PartPos, ?SEGPARTSZ*4),
v_segments(H, PartBin, PartNo+1, SegNo)
end.
v_segments(H, <<>>, PartNo, SegNo) ->
v_parts(H, PartNo, SegNo);
v_segments(_H, <<0:32,_/binary>>, _PartNo, _SegNo) ->
done;
v_segments(H, <<Seg:32,T/binary>>, PartNo, SegNo) ->
io:format("<~w>SEGMENT ~w~n", [Seg, SegNo]),
v_segment(H, SegNo, Seg, 0),
v_segments(H, T, PartNo, SegNo+1).
v_segment(_H, _, _SegPos, ?SEGSZP) ->
done;
v_segment(H, SegNo, SegPos, SegSlot) ->
Slot = SegSlot + (SegNo * ?SEGSZP),
BucketP = SegPos + (4 * ?SZOBJP * SegSlot),
case catch read_bucket(H, BucketP, H#head.type) of
{'EXIT', Reason} ->
dets_utils:vformat("** dets: Corrupt or truncated dets file~n",
[]),
io:format("~nERROR ~tp~n", [Reason]);
[] -> %% don't print empty buckets
true;
{Size, CollP, Objects} ->
io:format(" <~w>~w: <~w:~p>~w~n",
[BucketP, Slot, CollP, Size, Objects])
end,
v_segment(H, SegNo, SegPos, SegSlot+1).
%% -> [] | {Pointer, [object()]} | throw(EXIT)
read_bucket(Head, Position, Type) ->
MaxSize = maxobjsize(Head),
case dets_utils:ipread(Head, Position, MaxSize) of
{ok, {Size, Pointer, <<Size:32, _Status:32, KeysObjs/binary>>}} ->
Objs = bin2objs(KeysObjs, Type, []),
{Size, Pointer, lists:reverse(Objs)};
[] ->
[]
end.
-define(SEQSTART, -(1 bsl 26)).
%% -> [{Key,SizeOfWholeKey,WholeKeyBin,FirstObject,OtherObjects}] |throw(EXIT)
%% FirstObject = {Term, Seq, Binary}
%% Seq < 0 (and ascending).
per_key(Head, <<BinSize:32, ?ACTIVE:32, Bin/binary>> = B) ->
true = (byte_size(B) =:= BinSize),
if
Head#head.type =:= set ->
per_set_key(Bin, Head#head.keypos, []);
true ->
per_bag_key(Bin, Head#head.keypos, [])
end.
per_set_key(<<Size:32, T/binary>> = B, KeyPos, L) ->
<<KeyBin:Size/binary, R/binary>> = B,
Term = binary_to_term(T),
Key = element(KeyPos, Term),
Item = {Term, ?SEQSTART, KeyBin},
per_set_key(R, KeyPos, [{Key,Size,KeyBin,Item,[]} | L]);
per_set_key(<<>>, KeyPos, L) when is_integer(KeyPos) ->
lists:reverse(L).
per_bag_key(<<Size:32, ObjSz:32, T/binary>> = B, KeyPos, L) ->
<<KeyBin:Size/binary, R/binary>> = B,
ObjSz1 = ObjSz - 4,
Size1 = Size - ObjSz - 4,
<<_:ObjSz1/binary, KeyObjs:Size1/binary, _/binary>> = T,
<<_Size:32, Bin:ObjSz/binary, _/binary>> = B,
Term = binary_to_term(T),
Key = element(KeyPos, Term),
Item = {Term, ?SEQSTART, Bin},
per_bag_key(R, KeyPos, [{Key,Size,KeyBin,Item,KeyObjs} | L]);
per_bag_key(<<>>, KeyPos, L) when is_integer(KeyPos) ->
lists:reverse(L).
binobjs2terms(<<ObjSz:32, T/binary>> = B) ->
binobjs2terms(B, T, ObjSz, byte_size(B)-ObjSz, ?SEQSTART+1, []);
binobjs2terms([] = B) ->
B;
binobjs2terms(<<>>) ->
[].
binobjs2terms(Bin, Obj, _ObjSz, _Size=0, N, L) ->
lists:reverse(L, [{binary_to_term(Obj), N, Bin}]);
binobjs2terms(Bin, Bin1, ObjSz, Size, N, L) ->
<<B:ObjSz/binary, T/binary>> = Bin,
<<NObjSz:32, T1/binary>> = T,
Item = {binary_to_term(Bin1), N, B},
binobjs2terms(T, T1, NObjSz, Size-NObjSz, N+1, [Item | L]).
%% Appends objects in reversed order.
bin2objs(KeysObjs, set, Ts) ->
<<ObjSz:32, T/binary>> = KeysObjs,
bin2objs(T, ObjSz-4, byte_size(KeysObjs)-ObjSz, Ts);
bin2objs(KeysObjs, _Type, Ts) ->
bin2objs2(KeysObjs, Ts).
bin2objs2(<<Size:32, ObjSz:32, T/binary>>, Ts) ->
bin2objs(T, ObjSz-4, Size-ObjSz-4, Ts);
bin2objs2(<<>>, Ts) ->
Ts.
bin2objs(Bin, ObjSz, _Size=0, Ts) ->
<<_:ObjSz/binary, T/binary>> = Bin,
bin2objs2(T, [binary_to_term(Bin) | Ts]);
bin2objs(Bin, ObjSz, Size, Ts) ->
<<_:ObjSz/binary, NObjSz:32, T/binary>> = Bin,
bin2objs(T, NObjSz-4, Size-NObjSz, [binary_to_term(Bin) | Ts]).
bin2keybins(KeysObjs, Head) when Head#head.type =:= set ->
<<ObjSz:32, T/binary>> = KeysObjs,
bin2keybins(T, Head#head.keypos, ObjSz-4, byte_size(KeysObjs)-ObjSz,[]);
bin2keybins(KeysObjs, Head) ->
bin2keybins2(KeysObjs, Head#head.keypos, []).
bin2keybins2(<<Size:32, ObjSz:32, T/binary>>, Kp, L) ->
bin2keybins(T, Kp, ObjSz-4, Size-ObjSz-4, L);
bin2keybins2(<<>>, Kp, L) when is_integer(Kp) ->
lists:reverse(L).
bin2keybins(Bin, Kp, ObjSz, _Size=0, L) ->
<<Obj:ObjSz/binary, T/binary>> = Bin,
Term = binary_to_term(Obj),
bin2keybins2(T, Kp, [{element(Kp, Term),Obj} | L]);
bin2keybins(Bin, Kp, ObjSz, Size, L) ->
<<Obj:ObjSz/binary, NObjSz:32, T/binary>> = Bin,
Term = binary_to_term(Obj),
bin2keybins(T, Kp, NObjSz-4, Size-NObjSz, [{element(Kp,Term),Obj} | L]).
