dets.erl

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

                    einval(Reply, [Tab, Args])
            end;
	_ ->
	    erlang:error(badarg, [Tab, Args])
    end;
open_file(Tab, Arg) ->
    open_file(Tab, [Arg]).

pid2name(Pid) ->
    dets_server:pid2name(Pid).

remove_user(Pid, From) ->
    req(Pid, {close, From}).

repair_continuation(#dets_cont{match_program = B}=Cont, MS) 
    when is_binary(B) ->
    case ets:is_compiled_ms(B) of
	true ->
	    Cont;
	false ->
            Cont#dets_cont{match_program = ets:match_spec_compile(MS)}
    end;
repair_continuation(#dets_cont{}=Cont, _MS) ->
    Cont;
repair_continuation(T, MS) ->
    erlang:error(badarg, [T, MS]).

safe_fixtable(Tab, Bool) when Bool; not Bool ->
    badarg(treq(Tab, {safe_fixtable, Bool}), [Tab, Bool]);
safe_fixtable(Tab, Term) ->
    erlang:error(badarg, [Tab, Term]).

select(Tab, Pat) ->
    badarg(safe_match(Tab, Pat, select), [Tab, Pat]).

select(Tab, Pat, N) ->
    badarg(init_chunk_match(Tab, Pat, select, N), [Tab, Pat, N]).
    
select(State) when State#dets_cont.what =:= select ->
    badarg(chunk_match(State), [State]);
select(Term) ->
    erlang:error(badarg, [Term]).

select_delete(Tab, Pat) ->
    badarg(match_delete(Tab, Pat, select), [Tab, Pat]).

slot(Tab, Slot) when is_integer(Slot), Slot >= 0 ->
    badarg(treq(Tab, {slot, Slot}), [Tab, Slot]);
slot(Tab, Term) ->
    erlang:error(badarg, [Tab, Term]).

start() ->
    dets_server:start().

stop() ->
    dets_server:stop().

istart_link(Server) ->
    {ok, proc_lib:spawn_link(dets, init, [self(), Server])}.

sync(Tab) ->
    badarg(treq(Tab, sync), [Tab]).

table(Tab) ->
    table(Tab, []).

table(Tab, Opts) ->
    case options(Opts, [traverse, n_objects]) of
        {badarg,_} ->
            erlang:error(badarg, [Tab, Opts]);
        [Traverse, NObjs] ->
            TF = case Traverse of
                     first_next -> 
                         fun() -> qlc_next(Tab, first(Tab)) end;
                     select -> 
                         fun(MS) -> qlc_select(select(Tab, MS, NObjs)) end;
                     {select, MS} ->
                         fun() -> qlc_select(select(Tab, MS, NObjs)) end
                 end,
            PreFun = fun(_) -> safe_fixtable(Tab, true) end,
            PostFun = fun() -> safe_fixtable(Tab, false) end,
            InfoFun = fun(Tag) -> table_info(Tab, Tag) end,
            %% lookup_keys is not public, but convenient
            LookupFun = 
                case Traverse of
                    {select, _MS} -> 
                        undefined;
                    _ -> 
                        fun(_KeyPos, [K]) -> lookup(Tab, K);
                           (_KeyPos, Ks) -> lookup_keys(Tab, Ks) 
                        end
                end,
            FormatFun = 
                fun(all) ->
                        As = case Opts of
                                 [] -> [Tab];
                                 _ -> [Tab, Opts]
                             end,
                        {?MODULE, table, As};
                   ({match_spec, MS}) ->
                        {?MODULE, table, [Tab, [{traverse, {select, MS}} | 
                                                listify(Opts)]]};
                   ({lookup, _KeyPos, [Value]}) ->
                        io_lib:format("~w:lookup(~w, ~w)", 
                                      [?MODULE, Tab, Value]);
                   ({lookup, _KeyPos, Values}) ->
                        io_lib:format("lists:flatmap(fun(V) -> "
                                      "~w:lookup(~w, V) end, ~w)", 
                                      [?MODULE, Tab, Values])
                end,
            qlc:table(TF, [{pre_fun, PreFun}, {post_fun, PostFun}, 
                           {info_fun, InfoFun}, {format_fun, FormatFun},
                           {lookup_fun, LookupFun}])
    end.
         
qlc_next(_Tab, '$end_of_table') ->
    [];
qlc_next(Tab, Key) ->
    case lookup(Tab, Key) of
        Objects when is_list(Objects) ->
            Objects ++ fun() -> qlc_next(Tab, next(Tab, Key)) end;
        Error ->
            %% Do what first and next do.
            exit(Error)
    end.

qlc_select('$end_of_table') -> 
    [];
qlc_select({Objects, Cont}) when is_list(Objects) -> 
    Objects ++ fun() -> qlc_select(select(Cont)) end;
qlc_select(Error) ->
    Error.

table_info(Tab, num_of_objects) ->
    info(Tab, size);
table_info(Tab, keypos) ->
    info(Tab, keypos);
table_info(Tab, is_unique_objects) ->
    info(Tab, type) =/= duplicate_bag;
table_info(_Tab, _) ->
    undefined.

%% End of table/2.

to_ets(DTab, ETab) ->
    case ets:info(ETab, protection) of
	undefined ->
	    erlang:error(badarg, [DTab, ETab]);
        _ ->
	    Fun = fun(X, T) -> true = ets:insert(T, X), T end,
	    foldl(Fun, ETab, DTab)
    end.

traverse(Tab, Fun) ->
    Ref = make_ref(),
    TFun = 
	fun(O, Acc) ->
		case Fun(O) of
		    continue  ->
			Acc;
		    {continue, Val} ->
			[Val | Acc];
		    {done, Value} ->
			throw({Ref, [Value | Acc]});
		    Other ->
			throw({Ref, Other})
		end
	end,
    do_traverse(TFun, [], Tab, Ref).

update_counter(Tab, Key, C) ->
    badarg(treq(Tab, {update_counter, Key, C}), [Tab, Key, C]).

verbose() ->           
    verbose(true).

verbose(What) ->
    ok = dets_server:verbose(What),
    All = dets_server:all(),
    Fun = fun(Tab) -> treq(Tab, {set_verbose, What}) end,
    lists:foreach(Fun, All),
    All.

%% Where in the (open) table is Object located? 
%% The address of the first matching object is returned.
%% Format 9 returns the address of the object collection.
%% -> {ok, Address} | false
where(Tab, Object) ->
    badarg(treq(Tab, {where, Object}), [Tab, Object]).

do_traverse(Fun, Acc, Tab, Ref) ->
    safe_fixtable(Tab, true),    
    Proc = dets_server:get_pid(Tab),
    try
        do_trav(Proc, Acc, Fun)
    catch {Ref, Result} ->
        Result
    after 
        safe_fixtable(Tab, false)
    end.

do_trav(Proc, Acc, Fun) ->
    {Spec, MP} = compile_match_spec(object, '_'),
    %% MP not used
    case req(Proc, {match, MP, Spec, default}) of
	{cont, State} ->
	    do_trav(State, Proc, Acc, Fun);
	Error ->
	    Error
    end.
    
do_trav(#dets_cont{bin = eof}, _Proc, Acc, _Fun) ->
    Acc;
do_trav(State, Proc, Acc, Fun) ->
    case req(Proc, {match_init, State}) of
	{cont, {Bins, NewState}} ->
	    do_trav_bins(NewState, Proc, Acc, Fun, lists:reverse(Bins));
	Error ->
	    Error
    end.

do_trav_bins(State, Proc, Acc, Fun, []) ->
    do_trav(State, Proc, Acc, Fun);
do_trav_bins(State, Proc, Acc, Fun, [Bin | Bins]) ->
    %% Unpack one binary at a time, using the client's heap.
    case catch binary_to_term(Bin) of 
	{'EXIT', _} ->
	    req(Proc, {corrupt, bad_object});
	Term ->
	    NewAcc = Fun(Term, Acc),
	    do_trav_bins(State, Proc, NewAcc, Fun, Bins)
    end.

safe_match(Tab, Pat, What) ->
    safe_fixtable(Tab, true),
    R = do_safe_match(init_chunk_match(Tab, Pat, What, default), []),
    safe_fixtable(Tab, false),
    R.
    
do_safe_match({error, Error}, _L) ->
    {error, Error};
do_safe_match({L, C}, LL) ->
    do_safe_match(chunk_match(C), L++LL);
do_safe_match('$end_of_table', L) ->
    L;
do_safe_match(badarg, _L) ->
    badarg.

%% What = object | bindings | select
init_chunk_match(Tab, Pat, What, N) when is_integer(N), N >= 0; 
                                         N =:= default ->
    case compile_match_spec(What, Pat) of
	{Spec, MP} ->
	    case req(dets_server:get_pid(Tab), {match, MP, Spec, N}) of
		{done, L} ->
		    {L, #dets_cont{tab = Tab, what = What, bin = eof}};
		{cont, State} ->
		    chunk_match(State#dets_cont{what = What, tab = Tab});
		Error ->
		    Error
	    end;
	badarg ->
	    badarg
    end;
init_chunk_match(_Tab, _Pat, _What, _) ->
    badarg.

chunk_match(State) ->
    case catch dets_server:get_pid(State#dets_cont.tab) of
	{'EXIT', _Reason} ->
	    badarg;
	_Proc when State#dets_cont.bin =:= eof ->
	    '$end_of_table';
	Proc ->
	    case req(Proc, {match_init, State}) of
		{cont, {Bins, NewState}} ->
		    MP = NewState#dets_cont.match_program,
		    case catch do_foldl_bins(Bins, MP) of
			{'EXIT', _} ->
                            case ets:is_compiled_ms(MP) of
                                true ->
                                    req(Proc, {corrupt, bad_object});
                                false ->
                                    badarg
                            end;
                        [] ->
                            chunk_match(NewState);
			Terms ->
			    {Terms, NewState}
		    end;
		Error ->
		    Error
	    end
    end.

do_foldl_bins(Bins, true) ->
    foldl_bins(Bins, []);
do_foldl_bins(Bins, MP) ->
    foldl_bins(Bins, MP, []).

foldl_bins([], Terms) ->
    %% Preserve time order (version 9). 
    Terms;
foldl_bins([Bin | Bins], Terms) ->    
    foldl_bins(Bins, [binary_to_term(Bin) | Terms]).

foldl_bins([], _MP, Terms) ->
    %% Preserve time order (version 9).
    Terms;
foldl_bins([Bin | Bins], MP, Terms) ->
    Term = binary_to_term(Bin),
    case ets:match_spec_run([Term], MP) of
	[] ->
	    foldl_bins(Bins, MP, Terms);
	[Result] ->
	    foldl_bins(Bins, MP, [Result | Terms])
    end.

%% -> {Spec, binary()} | badarg
compile_match_spec(select, ?PATTERN_TO_OBJECT_MATCH_SPEC('_') = Spec) ->
    {Spec, true};
compile_match_spec(select, Spec) ->
    case catch ets:match_spec_compile(Spec) of
	X when is_binary(X) ->
	    {Spec, X};
	_ ->
	    badarg
    end;
compile_match_spec(object, Pat) ->
    compile_match_spec(select, ?PATTERN_TO_OBJECT_MATCH_SPEC(Pat));
compile_match_spec(bindings, Pat) ->
    compile_match_spec(select, ?PATTERN_TO_BINDINGS_MATCH_SPEC(Pat));
compile_match_spec(delete, Pat) ->
    compile_match_spec(select, ?PATTERN_TO_TRUE_MATCH_SPEC(Pat)).

%% Process the args list as provided to open_file/2.
defaults(Tab, Args) ->
    Defaults = #open_args{file = to_list(Tab),
                          type = set,
                          keypos = 1,
                          repair = true, 
                          min_no_slots = default,
			  max_no_slots = default,
                          ram_file = false,
                          delayed_write = ?DEFAULT_CACHE,
                          auto_save = timer:minutes(?DEFAULT_AUTOSAVE),
                          access = read_write,
                          version = default},
    Fun = fun repl/2,
    lists:foldl(Fun, Defaults, Args).

to_list(T) when is_atom(T) -> atom_to_list(T);
to_list(T) -> T.

repl({access, A}, Defs) ->
    mem(A, [read, read_write]),
    Defs#open_args{access = A};
repl({auto_save, Int}, Defs) when is_integer(Int), Int >= 0 ->
    Defs#open_args{auto_save = Int};
repl({auto_save, infinity}, Defs) ->
    Defs#open_args{auto_save =infinity};
repl({cache_size, Int}, Defs) when is_integer(Int), Int >= 0 ->
    %% Recognized, but ignored.
    Defs;
repl({cache_size, infinity}, Defs) ->
    Defs;
repl({delayed_write, default}, Defs) ->
    Defs#open_args{delayed_write = ?DEFAULT_CACHE};
repl({delayed_write, {Delay,Size} = C}, Defs) 
          when is_integer(Delay), Delay >= 0, is_integer(Size), Size >= 0 ->
    Defs#open_args{delayed_write = C};
repl({estimated_no_objects, I}, Defs)  ->
    repl({min_no_slots, I}, Defs);
repl({file, File}, Defs) ->
    Defs#open_args{file = to_list(File)};
repl({keypos, P}, Defs) when is_integer(P), P > 0 ->
    Defs#open_args{keypos =P};
repl({max_no_slots, I}, Defs)  ->
    %% Version 9 only.
    MaxSlots = is_max_no_slots(I),
    is_comp_min_max(Defs#open_args{max_no_slots = MaxSlots});
repl({min_no_slots, I}, Defs)  ->
    MinSlots = is_min_no_slots(I),
    is_comp_min_max(Defs#open_args{min_no_slots = MinSlots});
repl({ram_file, Bool}, Defs) ->
    mem(Bool, [true, false]),
    Defs#open_args{ram_file = Bool};
repl({repair, T}, Defs) ->
    mem(T, [true, false, force]),
    Defs#open_args{repair = T};
repl({type, T}, Defs) ->
    mem(T, [set, bag, duplicate_bag]),
    Defs#open_args{type =T};
repl({version, Version}, Defs) ->
    V = is_version(Version),
    is_comp_min_max(Defs#open_args{version = V});
repl({_, _}, _) ->
    exit(badarg).

is_min_no_slots(default) -> default;
is_min_no_slots(I) when is_integer(I), I >= ?DEFAULT_MIN_NO_SLOTS -> I;
is_min_no_slots(I) when is_integer(I), I >= 0 -> ?DEFAULT_MIN_NO_SLOTS.

is_max_no_slots(default) -> default;
is_max_no_slots(I) when is_integer(I), I > 0, I < 1 bsl 31 -> I.

is_comp_min_max(Defs) ->
    %% Bug: accepts version = default, when default means version 8.
    #open_args{max_no_slots = Max, min_no_slots = Min, version = V} = Defs,
    case V of
	8 when Max =:= default, Min =:= default -> Defs;
	_ when V =/= 8, Min =:= default -> Defs;
	_ when V =/= 8, Max =:= default -> Defs;
	_ -> V =/= 8, true = Min =< Max, Defs
    end.

is_version(default) -> default;
is_version(8) -> 8;
is_version(9) -> 9.

mem(X, L) ->
    case lists:member(X, L) of
	true -> true;
	false -> exit(badarg)
    end.

options(Options, Keys) when is_list(Options) ->
    options(Options, Keys, []);
options(Option, Keys) ->
    options([Option], Keys, []).

options(Options, [Key | Keys], L) when is_list(Options) ->
    V = case lists:keysearch(Key, 1, Options) of
	    {value, {format, Format}} when Format =:= term; 
                                           Format =:= bchunk ->
		{ok, Format};