mnesia_schema.erl

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

		_Ns ->
		    verify_cstruct(Cs2),
		    [{op, del_table_copy, ram_copies, Node, cs2list(Cs2)}|
		     remove_node_from_tabs(Rest, Node)]
	    end
    end.

new_cs(Cs, Node, ram_copies, add) ->
    Cs#cstruct{ram_copies = opt_add(Node, Cs#cstruct.ram_copies)};
new_cs(Cs, Node, disc_copies, add) ->
    Cs#cstruct{disc_copies = opt_add(Node, Cs#cstruct.disc_copies)};
new_cs(Cs, Node, disc_only_copies, add) ->
    Cs#cstruct{disc_only_copies = opt_add(Node, Cs#cstruct.disc_only_copies)};
new_cs(Cs, Node, ram_copies, del) ->
    Cs#cstruct{ram_copies = lists:delete(Node , Cs#cstruct.ram_copies)};
new_cs(Cs, Node, disc_copies, del) ->
    Cs#cstruct{disc_copies = lists:delete(Node , Cs#cstruct.disc_copies)};
new_cs(Cs, Node, disc_only_copies, del) ->
    Cs#cstruct{disc_only_copies = 
               lists:delete(Node , Cs#cstruct.disc_only_copies)};
new_cs(Cs, _Node, Storage, _Op) ->
    mnesia:abort({badarg, Cs#cstruct.name, Storage}).


opt_add(N, L) -> [N | lists:delete(N, L)].

move_table(Tab, FromNode, ToNode) ->
    schema_transaction(fun() -> do_move_table(Tab, FromNode, ToNode) end).

do_move_table(schema, _FromNode, _ToNode) ->
    mnesia:abort({bad_type, schema});
do_move_table(Tab, FromNode, ToNode) when atom(FromNode), atom(ToNode) ->
    TidTs = get_tid_ts_and_lock(schema, write),
    insert_schema_ops(TidTs, make_move_table(Tab, FromNode, ToNode));
do_move_table(Tab, FromNode, ToNode) ->
    mnesia:abort({badarg, Tab, FromNode, ToNode}).

make_move_table(Tab, FromNode, ToNode) ->
    ensure_writable(schema),
    Cs = incr_version(val({Tab, cstruct})),
    Ns = mnesia_lib:cs_to_nodes(Cs),
    verify(false, lists:member(ToNode, Ns), {already_exists, Tab, ToNode}),
    verify(true, lists:member(FromNode, val({Tab, where_to_write})),
           {not_active, Tab, FromNode}),
    verify(false, val({Tab,local_content}),
           {"Cannot move table with local content", Tab}),
    ensure_active(Cs),
    Running = val({current, db_nodes}),
    Storage = mnesia_lib:schema_cs_to_storage_type(FromNode, Cs),
    verify(true, lists:member(ToNode, Running), {not_active, schema, ToNode}),
    
    Cs2 = new_cs(Cs, ToNode, Storage, add),
    Cs3 = new_cs(Cs2, FromNode, Storage, del),
    verify_cstruct(Cs3),
    [{op, add_table_copy, Storage, ToNode, cs2list(Cs2)},
     {op, sync_trans},
     {op, del_table_copy, Storage, FromNode, cs2list(Cs3)}].

%% end of functions to add and delete nodes to tables
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%

change_table_copy_type(Tab, Node, ToS) ->
    schema_transaction(fun() -> do_change_table_copy_type(Tab, Node, ToS) end).

do_change_table_copy_type(Tab, Node, ToS) when atom(Node) ->
    TidTs = get_tid_ts_and_lock(schema, write),
    get_tid_ts_and_lock(Tab, write), % ensure global sync
    %% get_tid_ts_and_lock(Tab, read),
    insert_schema_ops(TidTs, make_change_table_copy_type(Tab, Node, ToS));
do_change_table_copy_type(Tab, Node, _ToS) ->
    mnesia:abort({badarg, Tab, Node}).

make_change_table_copy_type(Tab, Node, unknown) ->
    make_del_table_copy(Tab, Node);
make_change_table_copy_type(Tab, Node, ToS) ->
    ensure_writable(schema),
    Cs = incr_version(val({Tab, cstruct})),
    FromS = mnesia_lib:storage_type_at_node(Node, Tab), 

    case compare_storage_type(false, FromS, ToS) of
	{same, _} ->
	    mnesia:abort({already_exists, Tab, Node, ToS});
	{diff, _} ->
	    ignore;
	incompatible ->
	    ensure_active(Cs)
    end,

    Cs2 = new_cs(Cs, Node, FromS, del),
    Cs3 = new_cs(Cs2, Node, ToS, add),
    verify_cstruct(Cs3),
    
    if
	FromS == unknown ->
	    make_add_table_copy(Tab, Node, ToS);
	true ->
	    ignore
    end,

    [{op, change_table_copy_type, Node, FromS, ToS, cs2list(Cs3)}].

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% change index functions ....
%% Pos is allready added by 1 in both of these functions 

add_table_index(Tab, Pos) ->
    schema_transaction(fun() -> do_add_table_index(Tab, Pos) end).

do_add_table_index(schema, _Attr) ->
    mnesia:abort({bad_type, schema});
do_add_table_index(Tab, Attr) ->
    TidTs = get_tid_ts_and_lock(schema, write),
    get_tid_ts_and_lock(Tab, read),
    Pos = attr_tab_to_pos(Tab, Attr),
    insert_schema_ops(TidTs, make_add_table_index(Tab, Pos)).

make_add_table_index(Tab, Pos) ->
    ensure_writable(schema),
    Cs = incr_version(val({Tab, cstruct})),
    ensure_active(Cs),
    Ix = Cs#cstruct.index,
    verify(false, lists:member(Pos, Ix), {already_exists, Tab, Pos}),
    Ix2 = lists:sort([Pos | Ix]),
    Cs2 = Cs#cstruct{index = Ix2},
    verify_cstruct(Cs2),
    [{op, add_index, Pos, cs2list(Cs2)}].

del_table_index(Tab, Pos) ->
    schema_transaction(fun() -> do_del_table_index(Tab, Pos) end).

do_del_table_index(schema, _Attr) ->
    mnesia:abort({bad_type, schema});
do_del_table_index(Tab, Attr) ->
    TidTs = get_tid_ts_and_lock(schema, write),
    get_tid_ts_and_lock(Tab, read),
    Pos = attr_tab_to_pos(Tab, Attr),
    insert_schema_ops(TidTs, make_del_table_index(Tab, Pos)).

make_del_table_index(Tab, Pos) ->
    ensure_writable(schema),
    Cs = incr_version(val({Tab, cstruct})),
    ensure_active(Cs),
    Ix = Cs#cstruct.index,
    verify(true, lists:member(Pos, Ix), {no_exists, Tab, Pos}),
    Cs2 = Cs#cstruct{index = lists:delete(Pos, Ix)},
    verify_cstruct(Cs2),
    [{op, del_index, Pos, cs2list(Cs2)}].

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

add_snmp(Tab, Ustruct) ->
    schema_transaction(fun() -> do_add_snmp(Tab, Ustruct) end).

do_add_snmp(schema, _Ustruct) ->
    mnesia:abort({bad_type, schema});
do_add_snmp(Tab, Ustruct) ->
    TidTs = get_tid_ts_and_lock(schema, write),
    get_tid_ts_and_lock(Tab, read),
    insert_schema_ops(TidTs, make_add_snmp(Tab, Ustruct)).

make_add_snmp(Tab, Ustruct) ->
    ensure_writable(schema),
    Cs = incr_version(val({Tab, cstruct})),
    ensure_active(Cs),
    verify([], Cs#cstruct.snmp, {already_exists, Tab, snmp}),
    Error = {badarg, Tab, snmp, Ustruct},
    verify(true, mnesia_snmp_hook:check_ustruct(Ustruct), Error),
    Cs2 = Cs#cstruct{snmp = Ustruct},
    verify_cstruct(Cs2),
    [{op, add_snmp, Ustruct, cs2list(Cs2)}].

del_snmp(Tab) ->
    schema_transaction(fun() -> do_del_snmp(Tab) end).

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

make_del_snmp(Tab) ->
    ensure_writable(schema),
    Cs = incr_version(val({Tab, cstruct})),
    ensure_active(Cs),
    Cs2 = Cs#cstruct{snmp = []},
    verify_cstruct(Cs2),
    [{op, del_snmp, cs2list(Cs2)}].

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

transform_table(Tab, Fun, NewAttrs, NewRecName) 
  when function(Fun), list(NewAttrs), atom(NewRecName) -> 
    schema_transaction(fun() -> do_transform_table(Tab, Fun, NewAttrs, NewRecName) end);

transform_table(Tab, ignore, NewAttrs, NewRecName) 
  when list(NewAttrs), atom(NewRecName) -> 
    schema_transaction(fun() -> do_transform_table(Tab, ignore, NewAttrs, NewRecName) end);

transform_table(Tab, Fun, NewAttrs, NewRecName) ->
    {aborted,{bad_type, Tab, Fun, NewAttrs, NewRecName}}.

do_transform_table(schema, _Fun, _NewAttrs, _NewRecName) ->
    mnesia:abort({bad_type, schema});
do_transform_table(Tab, Fun, NewAttrs, NewRecName) ->
    TidTs = get_tid_ts_and_lock(schema, write),
    get_tid_ts_and_lock(Tab, write),
    insert_schema_ops(TidTs, make_transform(Tab, Fun, NewAttrs, NewRecName)).

make_transform(Tab, Fun, NewAttrs, NewRecName) ->
    ensure_writable(schema),
    Cs = incr_version(val({Tab, cstruct})),
    ensure_active(Cs),
    ensure_writable(Tab),
    case mnesia_lib:val({Tab, index}) of
	[] -> 
	    Cs2 = Cs#cstruct{attributes = NewAttrs, record_name = NewRecName},
	    verify_cstruct(Cs2),
	    [{op, transform, Fun, cs2list(Cs2)}];
	PosList ->
	    DelIdx = fun(Pos, Ncs) ->
			     Ix = Ncs#cstruct.index,
			     Ncs1 = Ncs#cstruct{index = lists:delete(Pos, Ix)},
			     Op = {op, del_index, Pos, cs2list(Ncs1)},
			     {Op, Ncs1}
		     end,
	    AddIdx = fun(Pos, Ncs) ->
			     Ix = Ncs#cstruct.index,
			     Ix2 = lists:sort([Pos | Ix]),
			     Ncs1 = Ncs#cstruct{index = Ix2},
			     Op = {op, add_index, Pos, cs2list(Ncs1)},
			     {Op, Ncs1}
		     end,
            {DelOps, Cs1} = lists:mapfoldl(DelIdx, Cs, PosList),
	    Cs2 = Cs1#cstruct{attributes = NewAttrs, record_name = NewRecName},
            {AddOps, Cs3} = lists:mapfoldl(AddIdx, Cs2, PosList),
	    verify_cstruct(Cs3),
	    lists:flatten([DelOps, {op, transform, Fun, cs2list(Cs2)}, AddOps])
    end.

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

change_table_access_mode(Tab, Mode) ->
    schema_transaction(fun() -> do_change_table_access_mode(Tab, Mode) end).

do_change_table_access_mode(Tab, Mode) ->
    {_Mod, Tid, Ts} = get_tid_ts_and_lock(schema, write),
    Store = Ts#tidstore.store,
    mnesia_locker:wlock_no_exist(Tid, Store, schema, val({schema, active_replicas})),
    mnesia_locker:wlock_no_exist(Tid, Store, Tab, val({Tab, active_replicas})),
    do_insert_schema_ops(Store, make_change_table_access_mode(Tab, Mode)).

make_change_table_access_mode(Tab, Mode) ->
    ensure_writable(schema),
    Cs = incr_version(val({Tab, cstruct})),
    ensure_active(Cs),
    OldMode = Cs#cstruct.access_mode,
    verify(false, OldMode ==  Mode, {already_exists, Tab, Mode}),
    Cs2 = Cs#cstruct{access_mode = Mode},
    verify_cstruct(Cs2),
    [{op, change_table_access_mode, cs2list(Cs2), OldMode, Mode}].

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

change_table_load_order(Tab, LoadOrder) ->
    schema_transaction(fun() -> do_change_table_load_order(Tab, LoadOrder) end).

do_change_table_load_order(schema, _LoadOrder) ->
    mnesia:abort({bad_type, schema});
do_change_table_load_order(Tab, LoadOrder) ->
    TidTs = get_tid_ts_and_lock(schema, write),
    get_tid_ts_and_lock(Tab, none),
    insert_schema_ops(TidTs, make_change_table_load_order(Tab, LoadOrder)).

make_change_table_load_order(Tab, LoadOrder) ->
    ensure_writable(schema),
    Cs = incr_version(val({Tab, cstruct})),
    ensure_active(Cs),
    OldLoadOrder = Cs#cstruct.load_order,
    Cs2 = Cs#cstruct{load_order = LoadOrder},
    verify_cstruct(Cs2),
    [{op, change_table_load_order, cs2list(Cs2), OldLoadOrder, LoadOrder}].

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

write_table_property(Tab, Prop) when tuple(Prop), size(Prop) >= 1 ->
    schema_transaction(fun() -> do_write_table_property(Tab, Prop) end);
write_table_property(Tab, Prop) ->
    {aborted, {bad_type, Tab, Prop}}.
do_write_table_property(Tab, Prop) ->
    TidTs = get_tid_ts_and_lock(schema, write),
    {_, _, Ts} = TidTs,
    Store = Ts#tidstore.store,
    case change_prop_in_existing_op(Tab, Prop, write_property, Store) of
	true ->
	    dbg_out("change_prop_in_existing_op"
		    "(~p,~p,write_property,Store) -> true~n",
		    [Tab,Prop]),
	    %% we have merged the table prop into the create_table op
	    ok;
	false ->
	    dbg_out("change_prop_in_existing_op"
		    "(~p,~p,write_property,Store) -> false~n",
		    [Tab,Prop]),
	    %% this must be an existing table
	    get_tid_ts_and_lock(Tab, none),
	    insert_schema_ops(TidTs, make_write_table_properties(Tab, [Prop]))
    end.

make_write_table_properties(Tab, Props) ->
    ensure_writable(schema),
    Cs = incr_version(val({Tab, cstruct})),
    ensure_active(Cs),
    make_write_table_properties(Tab, Props, Cs).

make_write_table_properties(Tab, [Prop | Props], Cs) ->
    OldProps = Cs#cstruct.user_properties,
    PropKey = element(1, Prop),
    DelProps = lists:keydelete(PropKey, 1, OldProps),
    MergedProps = lists:merge(DelProps, [Prop]),
    Cs2 = Cs#cstruct{user_properties = MergedProps},
    verify_cstruct(Cs2),
    [{op, write_property, cs2list(Cs2), Prop} |
     make_write_table_properties(Tab, Props, Cs2)];
make_write_table_properties(_Tab, [], _Cs) ->
    [].

change_prop_in_existing_op(Tab, Prop, How, Store) ->
    Ops = ets:match_object(Store, '_'),
    case update_existing_op(Ops, Tab, Prop, How, []) of
	{true, Ops1} ->
	    ets:match_delete(Store, '_'),
	    [ets:insert(Store, Op) || Op <- Ops1],
	    true;
	false ->
	    false
    end.
 
update_existing_op([{op, Op, L = [{name,Tab}|_], _OldProp}|Ops], 
		   Tab, Prop, How, Acc) when Op == write_property; 
					     Op == delete_property ->
    %% Apparently, mnesia_dumper doesn't care about OldProp here -- just L,
    %% so we will throw away OldProp (not that it matters...) and insert Prop.
    %% as element 3.
    L1 = insert_prop(Prop, L, How),
    NewOp = {op, How, L1, Prop},
    {true, lists:reverse(Acc) ++ [NewOp|Ops]};
update_existing_op([Op = {op, create_table, L}|Ops], Tab, Prop, How, Acc) ->
    case lists:keysearch(name, 1, L) of
	{value, {_, Tab}} ->
	    %% Tab is being created here -- insert Prop into L
	    L1 = insert_prop(Prop, L, How),
	    {true, lists:reverse(Acc) ++ [{op, create_table, L1}|Ops]};
	_ ->
	    update_existing_op(Ops, Tab, Prop, How, [Op|Acc])
    end;
update_existing_op([Op|Ops], Tab, Prop, How, Acc) ->
    update_existing_op(Ops, Tab, Prop, How, [Op|Acc]);
update_existing_op([], _, _, _, _) ->
    false.

do_read_table_property(Tab, Key) ->
    TidTs = get_tid_ts_and_lock(schema, read),
    {_, _, Ts} = TidTs,
    Store = Ts#tidstore.store, 
    Props = ets:foldl(
	      fun({op, create_table, [{name, T}|Opts]}, _Acc)
		 when T==Tab ->
		      find_props(Opts);
		 ({op, Op, [{name,T}|Opts], _Prop}, _Acc)
		 when T==Tab, Op==write_property; Op==delete_property ->
		      find_props(Opts);
		 ({op, delete_table, [{name,T}|_]}, _Acc)
		 when T==Tab ->
		      [];
		 (_Other, Acc) ->
		      Acc
	      end, [], Store),
    case lists:keysearch(Key, 1, Props) of
	{value, Property} ->
	    Property;
	false ->
	    undefined
    end.


%% perhaps a misnomer. How could also be delete_property... never mind.
%% Returns the modified L.
insert_prop(Prop, L, How) ->
    Prev = find_props(L),
    MergedProps = merge_with_previous(How, Prop, Prev),
    replace_props(L, MergedProps).

find_props([{user_properties, P}|_]) -> P;
find_props([_H|T]) -> find_props(T).
%% we shouldn't reach []

replace_props([{user_properties, _}|T], P) -> [{user_properties, P}|T];
replace_props([H|T], P) -> [H|replace_props(T, P)].
%% again, we shouldn't reach []