dets.erl

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

    case catch write_cache(H) of
	{H2, []} ->
	    {H2, catch file_size(H#head.fptr, H#head.filename)};
	{H2, _} = HeadError when is_record(H2, head) ->
	    HeadError
    end;
finfo(H, fixed) -> 
    %% true if fixtable/2 has been called
    {H, not (H#head.fixed =:= false)}; 
finfo(H, hash) -> {H, H#head.hash_bif};
finfo(H, keypos) -> {H, H#head.keypos};
finfo(H, memory) -> finfo(H, file_size);
finfo(H, no_objects) -> finfo(H, size);
finfo(H, no_keys) ->
    case catch write_cache(H) of
	{H2, []} ->
	    {H2, H2#head.no_keys};
	{H2, _} = HeadError when is_record(H2, head) ->
	    HeadError
    end;
finfo(H, no_slots) -> {H, (H#head.mod):no_slots(H)};
finfo(H, pid) -> {H, self()};
finfo(H, ram_file) -> {H, H#head.ram_file};
finfo(H, safe_fixed) -> {H, H#head.fixed};
finfo(H, size) -> 
    case catch write_cache(H) of
	{H2, []} ->
	    {H2, H2#head.no_objects};
	{H2, _} = HeadError when is_record(H2, head) ->
	    HeadError
    end;
finfo(H, type) -> {H, H#head.type};
finfo(H, version) -> {H, H#head.version};
finfo(H, _) -> {H, undefined}.

file_size(Fd, FileName) -> 
    {ok, Pos} = dets_utils:position(Fd, FileName, eof),
    Pos.

test_bchunk_format(_Head, undefined) ->
    false;
test_bchunk_format(Head, _Term) when Head#head.version =:= 8 ->
    false;
test_bchunk_format(Head, Term) ->
    dets_v9:try_bchunk_header(Term, Head) =/= not_ok.

do_open_file([Fname, Verbose], Parent, Server, Ref) ->
    case catch fopen2(Fname, Ref) of
	{error, _Reason} = Error ->
	    err(Error);
	{ok, Head} ->
	    maybe_put(verbose, Verbose),
	    {ok, Head#head{parent = Parent, server = Server}};
	{'EXIT', _Reason} = Error ->
	    Error;
	Bad ->
	    error_logger:format
	      ("** dets: Bug was found in open_file/1, reply was ~w.~n", 
	       [Bad]),
	    {error, {dets_bug, Fname, Bad}}
    end;
do_open_file([Tab, OpenArgs, Verb], Parent, Server, Ref) ->
    case catch fopen3(Tab, OpenArgs) of
	{error, {tooshort, _}} ->
	    file:delete(OpenArgs#open_args.file),
	    do_open_file([Tab, OpenArgs, Verb], Parent, Server, Ref);
	{error, _Reason} = Error ->
	    err(Error);
	{ok, Head} ->
	    maybe_put(verbose, Verb),
	    {ok, Head#head{parent = Parent, server = Server}};
	{'EXIT', _Reason} = Error ->
	    Error;
	Bad ->
	    error_logger:format
	      ("** dets: Bug was found in open_file/2, arguments were~n"
	       "** dets: ~w and reply was ~w.~n", 
	       [OpenArgs, Bad]),
	    {error, {dets_bug, Tab, {open_file, OpenArgs}, Bad}}
    end.

maybe_put(_, undefined) ->
    ignore;
maybe_put(K, V) ->
    put(K, V).

%% -> {Head, Result}, Result = ok | Error | {thrown, Error} | badarg
finit(Head, InitFun, _Format, _NoSlots) when Head#head.access =:= read ->
    _ = (catch InitFun(close)),
    {Head, {error, {access_mode, Head#head.filename}}};
finit(Head, InitFun, _Format, _NoSlots) when Head#head.fixed =/= false ->
    _ = (catch InitFun(close)),
    {Head, {error, {fixed_table, Head#head.name}}};
finit(Head, InitFun, Format, NoSlots) ->
    case catch do_finit(Head, InitFun, Format, NoSlots) of
	{ok, NewHead} ->
	    check_growth(NewHead),
	    start_auto_save_timer(NewHead),
	    {NewHead, ok};
	badarg ->
	    {Head, badarg};
	Error ->
	    dets_utils:corrupt(Head, Error)
    end.

%% -> {ok, NewHead} | throw(badarg) | throw(Error)
do_finit(Head, Init, Format, NoSlots) ->
    #head{fptr = Fd, type = Type, keypos = Kp, auto_save = Auto,
          cache = Cache, filename = Fname, ram_file = Ram,
	  min_no_slots = MinSlots0, max_no_slots = MaxSlots,
          name = Tab, update_mode = UpdateMode, mod = HMod} = Head,
    CacheSz = dets_utils:cache_size(Cache),
    {How, Head1} =
	case Format of
	    term when is_integer(NoSlots), NoSlots > MaxSlots ->
		throw(badarg);
	    term ->
		MinSlots = choose_no_slots(NoSlots, MinSlots0),
		if 
		    UpdateMode =:= new_dirty, MinSlots =:= MinSlots0 ->
			{general_init, Head};
		    true ->
			ok = dets_utils:truncate(Fd, Fname, bof),
			{ok, H} = HMod:initiate_file(Fd, Tab, Fname, Type, Kp,
						     MinSlots, MaxSlots, Ram,
						     CacheSz, Auto, false),
			{general_init, H}
		end;
	    bchunk ->
		ok = dets_utils:truncate(Fd, Fname, bof),
		{bchunk_init, Head}
	end,
    case How of
	bchunk_init -> 
	    case HMod:bchunk_init(Head1, Init) of
		{ok, NewHead} ->
		    {ok, NewHead#head{update_mode = dirty}};
		Error ->
		    Error
	    end;
	general_init ->
	    Cntrs = ets:new(dets_init, []),
	    Input = HMod:bulk_input(Head1, Init, Cntrs),
	    SlotNumbers = {Head1#head.min_no_slots, bulk_init, MaxSlots},
	    {Reply, SizeData} = 
		do_sort(Head1, SlotNumbers, Input, Cntrs, Fname, not_used),
	    Bulk = true,
	    case Reply of 
		{ok, NoDups, H1} ->
		    fsck_copy(SizeData, H1, Bulk, NoDups);
		Else ->
		    close_files(Bulk, SizeData, Head1),
		    Else
	    end
    end.

%% -> {NewHead, [LookedUpObject]} | {NewHead, Error}
flookup_keys(Head, Keys) ->
    case catch update_cache(Head, Keys, {lookup, nopid}) of
	{NewHead, [{_NoPid,Objs}]} -> 
	    {NewHead, Objs};
	{NewHead, L} when is_list(L) ->
	    {NewHead, lists:flatmap(fun({_Pid,OL}) -> OL end, L)};
	{NewHead, _} = HeadError when is_record(NewHead, head) ->
	    HeadError
    end.

%% -> {NewHead, Result}
fmatch_init(Head, C) ->
    case scan(Head, C) of
	{scan_error, Reason} ->
	    dets_utils:corrupt_reason(Head, Reason);
	{Ts, NC} ->
	    {Head, {cont, {Ts, NC}}}
    end.

%% -> {NewHead, Result}
fmatch(Head, MP, Spec, N) ->
    KeyPos = Head#head.keypos,
    case find_all_keys(Spec, KeyPos, []) of
	[] -> 
	    %% Complete match
	    case catch write_cache(Head) of
		{NewHead, []} ->
		    C0 = init_scan(NewHead, N),
		    {NewHead, {cont, C0#dets_cont{match_program = MP}}};
		{NewHead, _} = HeadError when is_record(NewHead, head) ->
		    HeadError
	    end;
	List ->
	    Keys = lists:usort(List),
	    {NewHead, Reply} = flookup_keys(Head, Keys),
	    case Reply of
		Objs when is_list(Objs) ->
		    MatchingObjs = ets:match_spec_run(Objs, MP),
		    {NewHead, {done, MatchingObjs}};
		Error ->
		    {NewHead, Error}
	    end
    end.

find_all_keys([], _, Ks) ->
    Ks;
find_all_keys([{H,_,_} | T], KeyPos, Ks) when is_tuple(H) ->
    case size(H) of
	Enough when Enough >= KeyPos ->
	    Key = element(KeyPos, H),
	    case contains_variable(Key) of
		true -> 
		    [];
		false ->
		    find_all_keys(T, KeyPos, [Key | Ks])
	    end;
	_ ->
	    find_all_keys(T, KeyPos, Ks)
    end;
find_all_keys(_, _, _) ->
    [].

contains_variable('_') ->
    true;
contains_variable(A) when is_atom(A) ->
    case atom_to_list(A) of
	[$$ | T] ->
	    case (catch list_to_integer(T)) of
		{'EXIT', _} ->
		    false;
		_ ->
		    true
	    end;
	_ ->
	    false
    end;
contains_variable(T) when is_tuple(T) ->
    contains_variable(tuple_to_list(T));
contains_variable([]) ->
    false;
contains_variable([H|T]) ->
    case contains_variable(H) of
	true ->
	    true;
	false ->
	    contains_variable(T)
    end;
contains_variable(_) ->
    false.

%% -> {NewHead, Res}
fmatch_delete_init(Head, MP, Spec)  ->
    KeyPos = Head#head.keypos,
    case catch 
        case find_all_keys(Spec, KeyPos, []) of
            [] -> 
                do_fmatch_delete_var_keys(Head, MP, Spec);
            List ->
                Keys = lists:usort(List),
                do_fmatch_constant_keys(Head, Keys, MP)
        end of
        {NewHead, _} = Reply when is_record(NewHead, head) ->
            Reply            
    end.

%% A note: If deleted objects reside in a bucket with other objects
%% that are not deleted, the bucket is moved. If the address of the
%% moved bucket is greater than original bucket address the kept
%% objects will be read once again later on. 
%% -> {NewHead, Res}
fmatch_delete(Head, C) ->
    case scan(Head, C) of
	{scan_error, Reason} ->
	    dets_utils:corrupt_reason(Head, Reason);
	{[], _} ->
	    {Head, {done, 0}};
	{RTs, NC} ->
	    MP = C#dets_cont.match_program,
	    case catch filter_binary_terms(RTs, MP, []) of
		{'EXIT', _} ->
		    dets_utils:corrupt_reason(Head, bad_object);
		Terms -> 
		    do_fmatch_delete(Head, Terms, NC)
	    end
    end.

do_fmatch_delete_var_keys(Head, _MP, ?PATTERN_TO_TRUE_MATCH_SPEC('_')) 
            when Head#head.fixed =:= false ->
    %% Handle the case where the file is emptied efficiently.
    %% Empty the cache just to get the number of objects right.
    {Head1, []} = write_cache(Head),
    N = Head1#head.no_objects,
    case fdelete_all_objects(Head1) of
	{NewHead, ok} ->
	    {NewHead, {done, N}};
	Reply ->
	    Reply
    end;
do_fmatch_delete_var_keys(Head, MP, _Spec) ->
    {NewHead, []} = write_cache(Head),
    C0 = init_scan(NewHead, default),
    {NewHead, {cont, C0#dets_cont{match_program = MP}, 0}}.

do_fmatch_constant_keys(Head, Keys, MP) ->
    case flookup_keys(Head, Keys) of
	{NewHead, ReadTerms} when is_list(ReadTerms) ->
	    Terms = filter_terms(ReadTerms, MP, []),
	    do_fmatch_delete(NewHead, Terms, fixed);
	Reply ->
	    Reply
    end.

filter_binary_terms([Bin | Bins], MP, L) ->
    Term = binary_to_term(Bin),
    case ets:match_spec_run([Term], MP) of
	[true] -> 
	    filter_binary_terms(Bins, MP, [Term | L]);
	_ ->
	    filter_binary_terms(Bins, MP, L)
    end;
filter_binary_terms([], _MP, L) ->
    L.

filter_terms([Term | Terms], MP, L) ->
    case ets:match_spec_run([Term], MP) of
	[true] -> 
	    filter_terms(Terms, MP, [Term | L]);
	_ ->
	    filter_terms(Terms, MP, L)
    end;
filter_terms([], _MP, L) ->
    L.

do_fmatch_delete(Head, Terms, What) ->
    N = length(Terms),
    case do_delete(Head, Terms, delete_object) of
	{NewHead, ok} when What =:= fixed ->
	    {NewHead, {done, N}};
	{NewHead, ok} ->
	    {NewHead, {cont, What, N}};
	Reply ->
	    Reply
    end.

do_delete(Head, Things, What) ->
    case catch update_cache(Head, Things, What) of
	{NewHead, []} ->
	    {NewHead, ok};
	{NewHead, _} = HeadError when is_record(NewHead, head) ->
	    HeadError
    end.

fmember(Head, Key) ->
    case catch begin
                   {Head2, [{_NoPid,Objs}]} = 
                       update_cache(Head, [Key], {lookup, nopid}),
                   {Head2, Objs =/= []}
               end of
        {NewHead, _} = Reply when is_record(NewHead, head) ->
            Reply
    end.

fnext(Head, Key) ->
    Slot = (Head#head.mod):db_hash(Key, Head),
    Ref = make_ref(),
    case catch {Ref, fnext(Head, Key, Slot)} of
	{Ref, {H, R}} -> 
	    {H, {ok, R}};
	{NewHead, _} = HeadError when is_record(NewHead, head) ->
	    HeadError
    end.

fnext(H, Key, Slot) ->
    {NH, []} = write_cache(H),
    case (H#head.mod):slot_objs(NH, Slot) of
	'$end_of_table' -> {NH, '$end_of_table'};
	L -> fnext_search(NH, Key, Slot, L)
    end.

fnext_search(H, K, Slot, L) ->
    Kp = H#head.keypos,
    case beyond_key(K, Kp, L) of
	[] -> fnext_slot(H, K, Slot+1);
	L2 -> {H, element(H#head.keypos, hd(L2))}
    end.

%% We've got to continue to search for the next key in the next slot
fnext_slot(H, K, Slot) ->
    case (H#head.mod):slot_objs(H, Slot) of
	'$end_of_table' -> {H, '$end_of_table'};
	[] -> fnext_slot(H, K, Slot+1);
	L -> {H, element(H#head.keypos, hd(L))}
    end.

beyond_key(_K, _Kp, []) -> [];
beyond_key(K, Kp, [H|T]) ->
    case dets_utils:cmp(element(Kp, H), K) of
        0 -> beyond_key2(K, Kp, T);
        _ -> beyond_key(K, Kp, T)
    end.

beyond_key2(_K, _Kp, []) -> [];
beyond_key2(K, Kp, [H|T]=L) ->
    case dets_utils:cmp(element(Kp, H), K) of
        0 -> beyond_key2(K, Kp, T);
        _ -> L
    end.

%% Open an already existing file, no arguments
%% -> {ok, head()} | throw(Error)
fopen2(Fname, Tab) ->
    case file:read_file_info(Fname) of
	{ok, _} ->
	    Acc = read_write,
	    Ram = false, 
	    %% Fd is not always closed upon error, but exit is soon called.
	    {ok, Fd, FH} = read_file_header(Fname, Acc, Ram),
            Mod = FH#fileheader.mod,
	    case Mod:check_file_header(FH, Fd) of
		{error, not_closed} ->
		    io:format(user,"dets: file ~p not properly closed, "
			      "repairing ...~n", [Fname]),
                    Version = default,
                    case fsck(Fd, Tab, Fname, FH, default, default, Version) of
                        ok ->
                            fopen2(Fname, Tab);
                        Error ->
                            throw(Error)
                    end;
		{ok, Head, ExtraInfo} ->
		    open_final(Head, Fname, Acc, Ram, ?DEFAULT_CACHE, 
			       Tab, ExtraInfo);
		{error, Reason} ->
		    throw({error, {Reason, Fname}})
	    end;
	Error ->