dets_v8.erl

来自「OTP是开放电信平台的简称」· ERL 代码 · 共 1,591 行 · 第 1/4 页

	      ObjsToRead, ToRead, Ls, LU) ->
    E = {P1,P2,WorkLists},
    first_object(SPs, Ss, Bs, Head, 
		 [E | ObjsToRead], [{P2, ?ReadAhead} | ToRead], Ls, LU);
first_object([], [], [], Head, ObjsToRead, ToRead, Ls, LU) ->
    {ok, Bins} = dets_utils:pread(ToRead, Head),
    case catch eval_first(Bins, ObjsToRead, Head, Ls, LU) of
	{ok, NLs, NLU} -> 
	    case create_writes(NLs, Head, [], 0) of
		{Head1, [], 0} ->
		    {Head1, NLU, []};
		{Head1, Ws, No} ->
		    {NewHead, Ws2} = update_no_objects(Head1, Ws, No),
		    {NewHead, NLU, Ws2}
	    end;
	_Error -> 
	    throw(dets_utils:corrupt_reason(Head, bad_object))
    end.

%% Update no_objects on the file too, if the number of segments that
%% dets:fsck/6 use for estimate has changed.
update_no_objects(Head, Ws, 0) -> {Head, Ws};
update_no_objects(Head, Ws, Delta) ->
    No = Head#head.no_objects,
    NewNo = No + Delta,
    NWs = 
	if 
	    NewNo > ?MAXOBJS ->
		Ws;
	    ?SLOT2SEG(No) =:= ?SLOT2SEG(NewNo) ->
		Ws;
	    true ->
		[{?NO_OBJECTS_POS, <<NewNo:32>>} | Ws]
	end,
    {Head#head{no_objects = NewNo}, NWs}.

eval_first([<<Next:32, Sz:32, _Status:32, Bin/binary>> | Bins], 
	   [SP | SPs], Head, Ls, LU) ->
    {P1, P2, WLs} = SP,
    L0 = [{old,P1}],
    case size(Bin) of
	BinSz when BinSz >= Sz ->
	    Term = binary_to_term(Bin),
	    Key = element(Head#head.keypos, Term),
	    {L, NLU} = find_key(Head, P2, Next, Sz, Term, Key, WLs, L0, LU),
	    eval_first(Bins, SPs, Head, [L | Ls], NLU);
	_BinSz ->
	    {L, NLU} = eval_slot(Head, Sz+?OHDSZ, P2, WLs, L0, LU),
	    eval_first(Bins, SPs, Head, [L | Ls], NLU)
    end;
eval_first([], [], _Head, Ls, LU) ->
    {ok, Ls, LU}.

eval_slot(_Head, _TrySize, _Pos=0, [], L, LU) ->
    {L, LU};
eval_slot(Head, _TrySize, Pos=0, [WL | WLs], L, LU) ->
    {_Key, {_Delete, LookUp, Objects}} = WL,
    {NL, NLU} = end_of_key(Objects, LookUp, L, []),
    eval_slot(Head, ?ReadAhead, Pos, WLs, NL, NLU++LU);
eval_slot(Head, TrySize, Pos, WLs, L, LU) ->
    {NextPos, Size, Term} = prterm(Head, Pos, TrySize),
    Key = element(Head#head.keypos, Term),
    find_key(Head, Pos, NextPos, Size, Term, Key, WLs, L, LU).

find_key(Head, Pos, NextPos, Size, Term, Key, WLs, L, LU) ->
    case lists:keysearch(Key, 1, WLs) of
	{value, {_, {Delete, LookUp, Objects}} = WL} ->
	    NWLs = lists:delete(WL, WLs),
	    {NewObjects, NL, LUK} = eval_object(Size, Term, Delete, LookUp, 
						Objects, Head, Pos, L, []),
	    eval_key(Key, Delete, LookUp, NewObjects, Head, NextPos, 
		     NWLs, NL, LU, LUK);
	false ->
	    L0 = [{old,Pos} | L],
	    eval_slot(Head, ?ReadAhead, NextPos, WLs, L0, LU)
    end.

eval_key(_Key, _Delete, Lookup, _Objects, Head, Pos, WLs, L, LU, LUK) 
                            when Head#head.type =:= set ->
    NLU = case Lookup of
	      {lookup, Pid} -> [{Pid,LUK} | LU];
	      skip -> LU
	  end,
    eval_slot(Head, ?ReadAhead, Pos, WLs, L, NLU);
eval_key(_Key, _Delete, LookUp, Objects, Head, Pos, WLs, L, LU, LUK) 
                                                          when Pos =:= 0 ->
    {NL, NLU} = end_of_key(Objects, LookUp, L, LUK),
    eval_slot(Head, ?ReadAhead, Pos, WLs, NL, NLU++LU);
eval_key(Key, Delete, LookUp, Objects, Head, Pos, WLs, L, LU, LUK) ->
    {NextPos, Size, Term} = prterm(Head, Pos, ?ReadAhead),
    case element(Head#head.keypos, Term) of
	Key ->
	    {NewObjects, NL, LUK1} = 
		eval_object(Size, Term, Delete, LookUp,Objects,Head,Pos,L,LUK),
	    eval_key(Key, Delete, LookUp, NewObjects, Head, NextPos, WLs, 
		     NL, LU, LUK1);
	Key2 ->
	    {L1, NLU} = end_of_key(Objects, LookUp, L, LUK),
	    find_key(Head, Pos, NextPos, Size, Term, Key2, WLs, L1, NLU++LU)
    end.

%% All objects in Objects have the key Key.
eval_object(Size, Term, Delete, LookUp, Objects, Head, Pos, L, LU) ->
    Type = Head#head.type,
    case lists:keysearch(Term, 1, Objects) of
	{value, {_Object, N}} when N =:= 0 ->
	    L1 = [{delete,Pos,Size} | L],
	    {Objects, L1, LU};
	{value, {_Object, N}} when N < 0, Type =:= set ->
	    L1 = [{old,Pos} | L],
	    wl_lookup(LookUp, Objects, Term, L1, LU);
	{value, {Object, _N}} when Type =:= bag -> % when N =:= 1; N =:= -1
	    L1 = [{old,Pos} | L],
	    Objects1 = lists:keydelete(Object, 1, Objects),
	    wl_lookup(LookUp, Objects1, Term, L1, LU);
	{value, {Object, N}} when N < 0, Type =:= duplicate_bag ->
	    L1 = [{old,Pos} | L],
	    Objects1 = lists:keyreplace(Object, 1, Objects, {Object,N+1}),
	    wl_lookup(LookUp, Objects1, Term, L1, LU);
	{value, {_Object, N}} when N > 0, Type =:= duplicate_bag ->
	    L1 = [{old,Pos} | L],
	    wl_lookup(LookUp, Objects, Term, L1, LU);
	false when Type =:= set, Delete =:= delete ->
	    case lists:keysearch(-1, 2, Objects) of
		false -> % no inserted object, perhaps deleted objects
		    L1 = [{delete,Pos,Size} | L],
		    {[], L1, LU};
		{value, {Term2,-1}} ->
		    Bin2 = term_to_binary(Term2),
		    NSize = size(Bin2),
		    Overwrite = 
			if
			    NSize =:= Size ->
				true;
			    true ->
				SizePos = sz2pos(Size+?OHDSZ),
				NSizePos = sz2pos(NSize+?OHDSZ),
				SizePos =:= NSizePos
			end,
		    E = if 
			    Overwrite ->
				{overwrite,Bin2,Pos};
			    true ->
				{replace,Bin2,Pos,Size}
			end,
		    wl_lookup(LookUp, [], Term2, [E | L], LU)
	    end;
	false when Delete =:= delete ->
	    L1 = [{delete,Pos,Size} | L],
	    {Objects, L1, LU};
	false ->
	    L1 = [{old,Pos} | L],
	    wl_lookup(LookUp, Objects, Term, L1, LU)
    end.

%% Inlined.
wl_lookup({lookup,_}, Objects, Term, L, LU) ->
    {Objects, L, [Term | LU]};
wl_lookup(skip, Objects, _Term, L, LU) ->
    {Objects, L, LU}.

end_of_key([{Object,N0} | Objs], LookUp, L, LU) when N0 =/= 0 ->
    N = abs(N0),
    NL = [{insert,N,term_to_binary(Object)} | L],
    NLU = case LookUp of 
	      {lookup, _} ->
		  lists:duplicate(N, Object) ++ LU;
	      skip ->
		  LU
	  end,
    end_of_key(Objs, LookUp, NL, NLU);
end_of_key([_ | Objects], LookUp, L, LU) ->
    end_of_key(Objects, LookUp, L, LU);
end_of_key([], {lookup,Pid}, L, LU) ->
    {L, [{Pid,LU}]};
end_of_key([], skip, L, LU) ->
    {L, LU}.

create_writes([L | Ls], H, Ws, No) ->
    {NH, NWs, NNo} = create_writes(L, H, Ws, No, 0, true),
    create_writes(Ls, NH, NWs, NNo);
create_writes([], H, Ws, No) ->
    {H, lists:reverse(Ws), No}.

create_writes([{old,Pos} | L], H, Ws, No, _Next, true) ->
    create_writes(L, H, Ws, No, Pos, true);
create_writes([{old,Pos} | L], H, Ws, No, Next, false) ->
    W = {Pos, <<Next:32>>},
    create_writes(L, H, [W | Ws], No, Pos, true);
create_writes([{insert,N,Bin} | L], H, Ws, No, Next, _NextIsOld) ->
    {NH, NWs, Pos} = create_inserts(N, H, Ws, Next, size(Bin), Bin),
    create_writes(L, NH, NWs, No+N, Pos, false);
create_writes([{overwrite,Bin,Pos} | L], H, Ws, No, Next, _) ->
    Size = size(Bin),
    W = {Pos, [<<Next:32, Size:32, ?ACTIVE:32>>, Bin]},
    create_writes(L, H, [W | Ws], No, Pos, true);
create_writes([{replace,Bin,Pos,OSize} | L], H, Ws, No, Next, _) ->
    Size = size(Bin),
    {H1, _} = dets_utils:free(H, Pos, OSize+?OHDSZ),
    {NH, NewPos, _} = dets_utils:alloc(H1, ?OHDSZ + Size),
    W1 = {NewPos, [<<Next:32, Size:32, ?ACTIVE:32>>, Bin]},
    NWs = if 
	      Pos =:= NewPos -> 
		  [W1 | Ws];
	      true -> 
		  W2 = {Pos+?STATUS_POS, <<?FREE:32>>},
		  [W1,W2 | Ws]
	  end,
    create_writes(L, NH, NWs, No, NewPos, false);
create_writes([{delete,Pos,Size} | L], H, Ws, No, Next, _) ->
    {NH, _} = dets_utils:free(H, Pos, Size+?OHDSZ),
    NWs = [{Pos+?STATUS_POS,<<?FREE:32>>} | Ws],
    create_writes(L, NH, NWs, No-1, Next, false);
create_writes([], H, Ws, No, _Next, _NextIsOld) ->
    {H, Ws, No}.

create_inserts(0, H, Ws, Next, _Size, _Bin) ->
    {H, Ws, Next};
create_inserts(N, H, Ws, Next, Size, Bin) ->
    {NH, Pos, _} = dets_utils:alloc(H, ?OHDSZ + Size),
    W = {Pos, [<<Next:32, Size:32, ?ACTIVE:32>>, Bin]},
    create_inserts(N-1, NH, [W | Ws], Pos, Size, Bin).

slot_position(S) ->
    Pos = ?SEGADDR(?SLOT2SEG(S)),
    Segment = get_segp(Pos),
    FinalPos = Segment + (4 * ?REM2(S, ?SEGSZ)),
    {FinalPos, 4}.

%% Twice the size of a segment due to the bug in sz2pos/1. Inlined.
actual_seg_size() ->
    ?POW(sz2pos(?SEGSZ*4)-1).

segp_cache(Pos, Segment) ->
    put(Pos, Segment).

%% Inlined.
get_segp(Pos) ->
    get(Pos).

%% Bug: If Sz0 is equal to 2**k for some k, then 2**(k+1) bytes are
%% allocated (wasting 2**k bytes).
sz2pos(N) ->
    1 + dets_utils:log2(N+1).

scan_objs(_Head, Bin, From, To, L, Ts, R, _Type) ->
    scan_objs(Bin, From, To, L, Ts, R).

scan_objs(Bin, From, To, L, Ts, -1) ->
    {stop, Bin, From, To, L, Ts};
scan_objs(B = <<_N:32, Sz:32, St:32, T/binary>>, From, To, L, Ts, R) ->
    if 
	St =:= ?ACTIVE;
	St =:= ?FREE -> % deleted after scanning started
	    case T of
		<<BinTerm:Sz/binary, T2/binary>> ->
		    NTs = [BinTerm | Ts],
		    OSz = Sz + ?OHDSZ,
		    Skip = ?POW(sz2pos(OSz)-1) - OSz,
		    F2 = From + OSz,
		    NR = if 
			     R < 0 ->
				 R + 1;
			     true ->
				 R + OSz + Skip
			 end,
		    scan_skip(T2, F2, To, Skip, L, NTs, NR);
		_ ->
                    {more, From, To, L, Ts, R, Sz+?OHDSZ}
	    end;
	true -> % a segment
	    scan_skip(B, From, To, actual_seg_size(), L, Ts, R)
    end;
scan_objs(_B, From, To, L, Ts, R) ->
    {more, From, To, L, Ts, R, 0}.

scan_skip(Bin, From, To, Skip, L, Ts, R) when From + Skip < To ->
    SkipPos = From + Skip,
    case Bin of
	<<_:Skip/binary, Tail/binary>> ->
	    scan_objs(Tail, SkipPos, To, L, Ts, R);
	_ ->
            {more, SkipPos, To, L, Ts, R, 0}
    end;
scan_skip(Bin, From, To, Skip, L, Ts, R) when From + Skip =:= To ->
    scan_next_allocated(Bin, From, To, L, Ts, R);
scan_skip(_Bin, From, _To, Skip, L, Ts, R) -> % when From + Skip > _To 
    From1 = From + Skip,
    {more, From1, From1, L, Ts, R, 0}.

scan_next_allocated(_Bin, _From, To, <<>>=L, Ts, R) ->
    {more, To, To, L, Ts, R, 0};
scan_next_allocated(Bin, From0, _To, <<From:32, To:32, L/binary>>, Ts, R) ->
    Skip = From - From0,
    scan_skip(Bin, From0, To, Skip, L, Ts, R).

%% Read term from file at position Pos
prterm(Head, Pos, ReadAhead) ->
    Res = dets_utils:pread(Head, Pos, ?OHDSZ, ReadAhead),
    ?DEBUGF("file:pread(~p, ~p, ?) -> ~p~n", [Head#head.filename, Pos, Res]),
    {ok, <<Next:32, Sz:32, _Status:32, Bin0/binary>>} = Res,
    ?DEBUGF("{Next, Sz} = ~p~n", [{Next, Sz}]),
    Bin = case size(Bin0) of
	      Actual when Actual >= Sz ->
		  Bin0;
	      _ ->
		  {ok, Bin1} = dets_utils:pread(Head, Pos +  ?OHDSZ, Sz, 0),
		  Bin1
	  end,
    Term = binary_to_term(Bin),
    {Next, Sz, Term}.

%%%%%%%%%%%%%%%%%  DEBUG functions %%%%%%%%%%%%%%%%

file_info(FH) ->
    #fileheader{closed_properly = CP, keypos = Kp,
                m = M, next = Next, n = N, version = Version,
                type = Type, no_objects = NoObjects} 
        = 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},
                  {type,Type},{version,Version}]}
    end.

v_segments(H) ->
    v_segments(H, 0).

v_segments(_H, ?SEGARRSZ) ->
    done;
v_segments(H, SegNo) ->
    Seg = dets_utils:read_4(H#head.fptr, ?SEGADDR(SegNo)),
    if
	Seg =:= 0 ->
	    done;
	true ->
	    io:format("SEGMENT ~w ", [SegNo]),
	    io:format("At position ~w~n", [Seg]),
	    v_segment(H, SegNo, Seg, 0),
	    v_segments(H, SegNo+1)
    end.

v_segment(_H, _, _SegPos, ?SEGSZ) ->
    done;
v_segment(H, SegNo, SegPos, SegSlot) ->
    Slot = SegSlot + (SegNo * ?SEGSZ),
    Chain = dets_utils:read_4(H#head.fptr, SegPos + (4 * SegSlot)),
    if 
	Chain =:= 0 ->  %% don't print empty chains
	    true;
	true ->
	    io:format("   <~p>~p: [",[SegPos + (4 * SegSlot), Slot]),
	    print_chain(H, Chain)
    end,
    v_segment(H, SegNo, SegPos, SegSlot+1).

print_chain(_H, 0) ->
    io:format("] \n", []);
print_chain(H, Pos) ->
    {ok, _} = file:position(H#head.fptr, Pos),
    case rterm(H#head.fptr) of
	{ok, 0, _Sz, Term} ->
	    io:format("<~p>~p] \n",[Pos, Term]);
	{ok, Next, _Sz, Term} ->
	    io:format("<~p>~p, ", [Pos, Term]),
	    print_chain(H, Next);
	Other ->
	    io:format("~nERROR ~p~n", [Other])
    end.

%% Can't be used at the bucket level!!!!
%% Only when we go down a chain
rterm(F) ->
    case catch rterm2(F) of
	{'EXIT', Reason} -> %% truncated DAT file 
	    dets_utils:vformat("** dets: Corrupt or truncated dets file~n", 
                               []), 
	    {error, Reason};
	Other -> 
	    Other
    end.

rterm2(F) ->
    {ok, <<Next:32, Sz:32, _:32>>} = file:read(F, ?OHDSZ),
    {ok, Bin} = file:read(F, Sz),
    Term = binary_to_term(Bin),
    {ok, Next, Sz, Term}.