mnesia_schema.erl

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

	   {bad_type, Tab, {access_mode, Access}}),

    Snmp = Cs#cstruct.snmp,
    verify(true, mnesia_snmp_hook:check_ustruct(Snmp),
	   {badarg, Tab, {snmp, Snmp}}),

    CheckProp = fun(Prop) when tuple(Prop), size(Prop) >= 1 -> ok;
		   (Prop) -> mnesia:abort({bad_type, Tab, {user_properties, [Prop]}})
		end,
    lists:foreach(CheckProp, Cs#cstruct.user_properties),

    case Cs#cstruct.cookie of
	{{MegaSecs, Secs, MicroSecs}, _Node} 
	when integer(MegaSecs), integer(Secs),
	     integer(MicroSecs), atom(node) ->
            ok;
        Cookie ->
            mnesia:abort({bad_type, Tab, {cookie, Cookie}})
    end,
    case Cs#cstruct.version of
        {{Major, Minor}, _Detail}
                when integer(Major), integer(Minor) ->
            ok;
        Version ->
            mnesia:abort({bad_type, Tab, {version, Version}})
    end.

verify_nodes(Cs) ->
    Tab = Cs#cstruct.name,
    Ram = Cs#cstruct.ram_copies,
    Disc = Cs#cstruct.disc_copies,
    DiscOnly = Cs#cstruct.disc_only_copies,
    LoadOrder = Cs#cstruct.load_order,

    verify({alt, [nil, list]}, mnesia_lib:etype(Ram),
	   {bad_type, Tab, {ram_copies, Ram}}),
    verify({alt, [nil, list]}, mnesia_lib:etype(Disc),
	   {bad_type, Tab, {disc_copies, Disc}}),
    case Tab of
	schema ->
	    verify([], DiscOnly, {bad_type, Tab, {disc_only_copies, DiscOnly}});
	_ ->
	    verify({alt, [nil, list]},
		   mnesia_lib:etype(DiscOnly),
		   {bad_type, Tab, {disc_only_copies, DiscOnly}})
    end,
    verify(integer, mnesia_lib:etype(LoadOrder),
	   {bad_type, Tab, {load_order, LoadOrder}}),
    
    Nodes = Ram ++ Disc ++ DiscOnly,
    verify(list, mnesia_lib:etype(Nodes),
	   {combine_error, Tab,
	    [{ram_copies, []}, {disc_copies, []}, {disc_only_copies, []}]}),
    verify(false, has_duplicates(Nodes), {combine_error, Tab, Nodes}),
    AtomCheck = fun(N) -> verify(atom, mnesia_lib:etype(N), {bad_type, Tab, N}) end, 
    lists:foreach(AtomCheck, Nodes).
    
verify(Expected, Fun, Error) when function(Fun) ->
    do_verify(Expected, catch Fun(), Error);
verify(Expected, Actual, Error) ->
    do_verify(Expected, Actual, Error).

do_verify({alt, Values}, Value, Error) ->
    case lists:member(Value, Values) of
        true -> ok;
        false -> mnesia:abort(Error)
    end;
do_verify(Value, Value, _) ->
    ok;
do_verify(_Value, _, Error) ->
     mnesia:abort(Error).

ensure_writable(Tab) ->
    case val({Tab, where_to_write}) of
        [] -> mnesia:abort({read_only, Tab});
        _ -> ok
    end.

%% Ensure that all replicas on disk full nodes are active
ensure_active(Cs) ->
    ensure_active(Cs, active_replicas).

ensure_active(Cs, What) ->
    Tab = Cs#cstruct.name,
    case val({Tab, What}) of
        [] -> mnesia:abort({no_exists, Tab});
        _ -> ok
    end,
    Nodes = mnesia_lib:intersect(val({schema, disc_copies}),
                                 mnesia_lib:cs_to_nodes(Cs)), 
    W = {Tab, What},
    case Nodes -- val(W) of
        [] ->
            ok;
        Ns ->
            Expl = "All replicas on diskfull nodes are not active yet",
            case val({Tab, local_content}) of
                true ->
		    case rpc:multicall(Ns, ?MODULE, is_remote_member, [W]) of
			{Replies, []} ->
			    check_active(Replies, Expl, Tab);
			{_Replies, BadNs} ->
			    mnesia:abort({not_active, Expl, Tab, BadNs})
                    end;
                false ->
                    mnesia:abort({not_active, Expl, Tab, Ns})
            end
    end.

ensure_not_active(schema, Node) ->
    case lists:member(Node, val({schema, active_replicas})) of 
	false ->
	    ok;
	true ->
	    Expl = "Mnesia is running",
	    mnesia:abort({active, Expl, Node})
    end.

is_remote_member(Key) ->
    IsActive = lists:member(node(), val(Key)),
    {IsActive, node()}.

check_active([{true, _Node} | Replies], Expl, Tab) ->
    check_active(Replies, Expl, Tab);
check_active([{false, Node} | _Replies], Expl, Tab) ->
    mnesia:abort({not_active, Expl, Tab, [Node]});
check_active([{badrpc, Reason} | _Replies], Expl, Tab) ->
    mnesia:abort({not_active, Expl, Tab, Reason});
check_active([], _Expl, _Tab) ->
    ok.

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% Here's the real interface function to create a table

create_table(TabDef) ->
    schema_transaction(fun() -> do_multi_create_table(TabDef) end).

%% And the corresponding do routines ....

do_multi_create_table(TabDef) ->
    get_tid_ts_and_lock(schema, write),
    ensure_writable(schema),
    Cs = list2cs(TabDef),
    case Cs#cstruct.frag_properties of
	[] ->
	    do_create_table(Cs);
	_Props ->
	    CsList = mnesia_frag:expand_cstruct(Cs),
	    lists:foreach(fun do_create_table/1, CsList)
    end,
    ok.

do_create_table(Cs) ->
    {_Mod, _Tid, Ts} =  get_tid_ts_and_lock(schema, none),
    Store = Ts#tidstore.store,
    do_insert_schema_ops(Store, make_create_table(Cs)).

make_create_table(Cs) ->
    Tab = Cs#cstruct.name,
    verify(false, check_if_exists(Tab), {already_exists, Tab}),
    unsafe_make_create_table(Cs).

% unsafe_do_create_table(Cs) ->
%     {_Mod, Tid, Ts} =  get_tid_ts_and_lock(schema, none),
%     Store = Ts#tidstore.store,
%     do_insert_schema_ops(Store, unsafe_make_create_table(Cs)).

unsafe_make_create_table(Cs) ->
    {_Mod, Tid, Ts} =  get_tid_ts_and_lock(schema, none),
    verify_cstruct(Cs),
    Tab = Cs#cstruct.name,
        
    %% Check that we have all disc replica nodes running
    DiscNodes = Cs#cstruct.disc_copies ++ Cs#cstruct.disc_only_copies,
    RunningNodes = val({current, db_nodes}),
    CheckDisc = fun(N) ->
			verify(true, lists:member(N, RunningNodes),
			       {not_active, Tab, N})
		end,
    lists:foreach(CheckDisc, DiscNodes),

    Nodes = mnesia_lib:intersect(mnesia_lib:cs_to_nodes(Cs), RunningNodes),
    Store = Ts#tidstore.store,
    mnesia_locker:wlock_no_exist(Tid, Store, Tab, Nodes),
    [{op, create_table, cs2list(Cs)}].

check_if_exists(Tab) ->
    TidTs = get_tid_ts_and_lock(schema, write),
    {_, _, Ts} = TidTs,
    Store = Ts#tidstore.store, 
    ets:foldl(
      fun({op, create_table, [{name, T}|_]}, _Acc) when T==Tab ->
	      true;
	 ({op, delete_table, [{name,T}|_]}, _Acc) when T==Tab ->
	      false;
	 (_Other, Acc) ->
	      Acc
      end, existed_before(Tab), Store).

existed_before(Tab) ->
    ('EXIT' =/= element(1, ?catch_val({Tab,cstruct}))).


%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% Delete a table entirely on all nodes.

delete_table(Tab) ->
    schema_transaction(fun() -> do_delete_table(Tab) end).

do_delete_table(schema) ->
    mnesia:abort({bad_type, schema});
do_delete_table(Tab) ->
    TidTs = get_tid_ts_and_lock(schema, write),
    ensure_writable(schema),
    insert_schema_ops(TidTs, make_delete_table(Tab, whole_table)).

make_delete_table(Tab, Mode) ->
    case existed_before(Tab) of
	false ->
	    %% Deleting a table that was created in this very
	    %% schema transaction. Delete all ops in the Store
	    %% that operate on this table. We cannot run a normal
	    %% delete operation, since that involves checking live
	    %% nodes etc.
	    TidTs = get_tid_ts_and_lock(schema, write),
	    {_, _, Ts} = TidTs,
	    Store = Ts#tidstore.store, 
	    Deleted = ets:select_delete(
			Store, [{{op,'$1',[{name,Tab}|'_']},
				 [{'or',
				   {'==','$1',create_table},
				   {'==','$1',delete_table}}], [true]}]),
	    ets:select_delete(
	      Store, [{{op,'$1',[{name,Tab}|'_'],'_'},
		       [{'or',
			 {'==','$1',write_table_property},
			 {'==','$1',delete_table_property}}],
		       [true]}]),
	    case Deleted of
		0 -> mnesia:abort({no_exists, Tab});
		_ -> []
	    end;
	true ->
	    case Mode of
		whole_table ->
		    case val({Tab, frag_properties}) of
			[] ->
			    [make_delete_table2(Tab)];
			_Props ->
			    %% Check if it is a base table 
			    mnesia_frag:lookup_frag_hash(Tab),		    
			    
			    %% Check for foreigners
			    F = mnesia_frag:lookup_foreigners(Tab),
			    verify([], F, {combine_error,
					   Tab, "Too many foreigners", F}),
			    [make_delete_table2(T) ||
				T <- mnesia_frag:frag_names(Tab)]
		    end;
		single_frag ->
		    [make_delete_table2(Tab)]
	    end
    end.

make_delete_table2(Tab) ->
    get_tid_ts_and_lock(Tab, write),
    Cs = val({Tab, cstruct}),
    ensure_active(Cs),
    ensure_writable(Tab),
    {op, delete_table, cs2list(Cs)}.

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% Change fragmentation of a table
    
change_table_frag(Tab, Change) ->
    schema_transaction(fun() -> do_change_table_frag(Tab, Change) end).

do_change_table_frag(Tab, Change) when atom(Tab), Tab /= schema ->
    TidTs = get_tid_ts_and_lock(schema, write),
    Ops = mnesia_frag:change_table_frag(Tab, Change),
    [insert_schema_ops(TidTs, Op) || Op <- Ops],
    ok;
do_change_table_frag(Tab, _Change) ->
    mnesia:abort({bad_type, Tab}).
    
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% Clear a table

clear_table(Tab) ->
    schema_transaction(fun() -> do_clear_table(Tab) end).

do_clear_table(schema) ->
    mnesia:abort({bad_type, schema});
do_clear_table(Tab) ->
    TidTs = get_tid_ts_and_lock(schema, write),
    get_tid_ts_and_lock(Tab, write),
    insert_schema_ops(TidTs, make_clear_table(Tab)).

make_clear_table(Tab) ->
    ensure_writable(schema),
    Cs = val({Tab, cstruct}),
    ensure_active(Cs),
    ensure_writable(Tab),
    [{op, clear_table, cs2list(Cs)}].

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

add_table_copy(Tab, Node, Storage) ->
    schema_transaction(fun() -> do_add_table_copy(Tab, Node, Storage) end).

do_add_table_copy(Tab, Node, Storage) when atom(Tab), atom(Node) ->
    TidTs = get_tid_ts_and_lock(schema, write),
    insert_schema_ops(TidTs, make_add_table_copy(Tab, Node, Storage));
do_add_table_copy(Tab,Node,_) ->
    mnesia:abort({badarg, Tab, Node}).

make_add_table_copy(Tab, Node, Storage) ->
    ensure_writable(schema),
    Cs = incr_version(val({Tab, cstruct})),
    Ns = mnesia_lib:cs_to_nodes(Cs),
    verify(false, lists:member(Node, Ns), {already_exists, Tab, Node}),
    Cs2 = new_cs(Cs, Node, Storage, add),
    verify_cstruct(Cs2),
    
    %% Check storage and if node is running
    IsRunning = lists:member(Node, val({current, db_nodes})),
    if
	Storage == unknown ->
	    mnesia:abort({badarg, Tab, Storage});
	Tab == schema ->
	    if
		Storage /= ram_copies ->
		    mnesia:abort({badarg, Tab, Storage});
		IsRunning == true ->
		    mnesia:abort({already_exists, Tab, Node});
		true ->
		    ignore
	    end;
	Storage == ram_copies ->
	    ignore;
	IsRunning == true ->
	    ignore;
	IsRunning == false ->
	    mnesia:abort({not_active, schema, Node})
    end,
    [{op, add_table_copy, Storage, Node, cs2list(Cs2)}].

del_table_copy(Tab, Node) ->
    schema_transaction(fun() -> do_del_table_copy(Tab, Node) end).

do_del_table_copy(Tab, Node) when atom(Node)  ->
    TidTs = get_tid_ts_and_lock(schema, write),
%%    get_tid_ts_and_lock(Tab, write), 
    insert_schema_ops(TidTs, make_del_table_copy(Tab, Node));
do_del_table_copy(Tab, Node) ->
    mnesia:abort({badarg, Tab, Node}).
   
make_del_table_copy(Tab, Node) ->
    ensure_writable(schema),
    Cs = incr_version(val({Tab, cstruct})),
    Storage = mnesia_lib:schema_cs_to_storage_type(Node, Cs),
    Cs2 = new_cs(Cs, Node, Storage, del),    
    case mnesia_lib:cs_to_nodes(Cs2) of
        [] when Tab == schema ->
            mnesia:abort({combine_error, Tab, "Last replica"});
        [] ->
	    ensure_active(Cs),	    
            dbg_out("Last replica deleted in table ~p~n",  [Tab]),
            make_delete_table(Tab,  whole_table);
        _ when Tab == schema ->
	    ensure_active(Cs2),
	    ensure_not_active(Tab, Node),
            verify_cstruct(Cs2),
	    Ops = remove_node_from_tabs(val({schema, tables}), Node),
	    [{op, del_table_copy, ram_copies, Node, cs2list(Cs2)} | Ops];
        _ ->
	    ensure_active(Cs),
            verify_cstruct(Cs2),
            [{op, del_table_copy, Storage, Node, cs2list(Cs2)}]
    end.

remove_node_from_tabs([], _Node) ->
    [];
remove_node_from_tabs([schema|Rest], Node) ->
    remove_node_from_tabs(Rest, Node);
remove_node_from_tabs([Tab|Rest], Node) ->    
    {Cs, IsFragModified} = 
	mnesia_frag:remove_node(Node, incr_version(val({Tab, cstruct}))),
    case mnesia_lib:schema_cs_to_storage_type(Node, Cs)  of
	unknown ->
	    case IsFragModified of
		true ->
		    [{op, change_table_frag, {del_node, Node}, cs2list(Cs)} | 
		     remove_node_from_tabs(Rest, Node)];
		false ->
		    remove_node_from_tabs(Rest, Node)
	    end;
	Storage ->
	    Cs2 = new_cs(Cs, Node, Storage, del),
	    case mnesia_lib:cs_to_nodes(Cs2) of
		[] ->
		    [{op, delete_table, cs2list(Cs)} |
		     remove_node_from_tabs(Rest, Node)];