mnesia_log.erl

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

confirm_decision_log_dump() ->
    case mnesia_monitor:close_log(previous_decision_log) of
	ok ->
	    file:delete(previous_decision_log_file());
	{error, Reason} ->
	    fatal("Cannot confirm decision log dump: ~p~n",
		  [Reason])
    end.

save_decision_tab(Decisions) ->
    Log = decision_tab,
    Tmp = mnesia_lib:dir("DECISION_TAB.TMP"),
    file:delete(Tmp),
    open_log(Log, decision_tab_header(), Tmp),
    append(Log, Decisions),
    close_log(Log),
    TabFile = decision_tab_file(),
    ok = file:rename(Tmp, TabFile).

open_decision_tab() ->
    TabFile = decision_tab_file(),
    open_log(decision_tab, decision_tab_header(), TabFile),
    start.

close_decision_tab() ->
    close_log(decision_tab).

chunk_decision_tab(Cont) ->
    %% dbg_out("chunk tab ~p~n", [Cont]),
    chunk_log(decision_tab, Cont).

close_decision_log() ->
    close_log(decision_log).

log_decision(Decision) ->
    append(decision_log, Decision).

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% Debug functions

view() ->
    lists:foreach(fun(F) -> view(F) end, log_files()).

view(File) ->
    mnesia_lib:show("*****  ~p ***** ~n", [File]),
    case exists(File) of
	false ->
	    nolog;
	true ->
	    N = view_only,
	    Args = [{file, File}, {name, N}, {mode, read_only}],
	    case disk_log:open(Args) of
		{ok, N} ->
		    view_file(start, N);
		{repaired, _, _, _} ->
		    view_file(start, N);
		{error, Reason} ->
		    error("Cannot open log ~p: ~p~n", [File, Reason])
	    end
    end.

view_file(C, Log) ->
    case disk_log:chunk(Log, C) of
	{error, Reason} ->
	    error("** Possibly truncated FILE ~p~n", [Reason]),
	    error;
	eof ->
	    disk_log:close(Log),
	    eof;
	{C2, Terms, _BadBytes} ->
	    dbg_out("Lost ~p bytes in ~p ~n", [_BadBytes, Log]),
	    lists:foreach(fun(X) -> mnesia_lib:show("~p~n", [X]) end, 
			  Terms),
	    view_file(C2, Log);
	{C2, Terms} ->
	    lists:foreach(fun(X) -> mnesia_lib:show("~p~n", [X]) end, 
			  Terms),
	    view_file(C2, Log)
    end.

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% Backup

-record(backup_args, {name, module, opaque, scope, prev_name, tables, cookie}).

backup(Opaque) ->
    backup(Opaque, []).

backup(Opaque, Mod) when atom(Mod) ->
    backup(Opaque, [{module, Mod}]);
backup(Opaque, Args) when list(Args) ->
    %% Backup all tables with max redundancy
    CpArgs = [{ram_overrides_dump, false}, {max, val({schema, tables})}],
    case mnesia_checkpoint:activate(CpArgs) of
	{ok, Name, _Nodes} ->
	    Res = backup_checkpoint(Name, Opaque, Args),
	    mnesia_checkpoint:deactivate(Name),
	    Res;
	{error, Reason} ->
	    {error, Reason}
    end.

backup_checkpoint(Name, Opaque) ->
    backup_checkpoint(Name, Opaque, []).

backup_checkpoint(Name, Opaque, Mod) when atom(Mod) ->
    backup_checkpoint(Name, Opaque, [{module, Mod}]);
backup_checkpoint(Name, Opaque, Args) when list(Args) ->
    DefaultMod = mnesia_monitor:get_env(backup_module),
    B = #backup_args{name = Name,
		     module = DefaultMod,
		     opaque = Opaque,
		     scope = global,
		     tables = all,
		     prev_name = Name},
    case check_backup_args(Args, B) of
	{ok, B2} ->
	    %% Decentralized backup
	    %% Incremental

	    Self = self(),
	    Pid = spawn_link(?MODULE, backup_master, [Self, B2]),
	    receive
		{Pid, Self, Res} -> Res
	    end;
	{error, Reason} ->
	    {error, Reason}
    end.

check_backup_args([Arg | Tail], B) ->
    case catch check_backup_arg_type(Arg, B) of
	{'EXIT', _Reason} ->
	    {error, {badarg, Arg}};
	B2 ->
	    check_backup_args(Tail, B2)
    end;

check_backup_args([], B) ->
    {ok, B}.

check_backup_arg_type(Arg, B) ->    
    case Arg of
	{scope, global} ->
	    B#backup_args{scope = global};
	{scope, local} ->
	    B#backup_args{scope = local};
	{module, Mod} ->
	    Mod2 = mnesia_monitor:do_check_type(backup_module, Mod),
	    B#backup_args{module = Mod2};
	{incremental, Name} ->
	    B#backup_args{prev_name = Name};
	{tables, Tabs} when list(Tabs) ->
	    B#backup_args{tables = Tabs}
    end.

backup_master(ClientPid, B) ->
    process_flag(trap_exit, true),
    case catch do_backup_master(B) of
	{'EXIT', Reason} ->
	    ClientPid ! {self(), ClientPid, {error, {'EXIT', Reason}}};
	Res ->
	    ClientPid ! {self(), ClientPid, Res}
    end,
    unlink(ClientPid),
    exit(normal).

do_backup_master(B) ->
    Name = B#backup_args.name,
    B2 = safe_apply(B, open_write, [B#backup_args.opaque]),
    B3 = safe_write(B2, [backup_log_header()]),
    case mnesia_checkpoint:tables_and_cookie(Name) of
	{ok, AllTabs, Cookie} ->
	    Tabs = select_tables(AllTabs, B3),
	    B4 = B3#backup_args{cookie = Cookie},
	    %% Always put schema first in backup file
	    B5 = backup_schema(B4, Tabs),
	    B6 = lists:foldl(fun backup_tab/2, B5, Tabs -- [schema]),
	    safe_apply(B6, commit_write, [B6#backup_args.opaque]),
	    ok;
	{error, Reason} ->
	    abort_write(B3, {?MODULE, backup_master}, [B], {error, Reason})
    end.

select_tables(AllTabs, B) ->
    Tabs =
	case B#backup_args.tables of
	    all -> AllTabs;
	    SomeTabs when list(SomeTabs) -> SomeTabs
	end,
    case B#backup_args.scope of
	global ->
	    Tabs;
	local ->
	    Name = B#backup_args.name,
	    [T || T <- Tabs, mnesia_checkpoint:most_local_node(Name, T) == node()]
    end.

safe_write(B, []) ->
    B;
safe_write(B, Recs) ->	
    safe_apply(B, write, [B#backup_args.opaque, Recs]).

backup_schema(B, Tabs) ->
    case lists:member(schema, Tabs) of
	true ->
	    backup_tab(schema, B);
	false ->
	    Defs = [{schema, T, mnesia_schema:get_create_list(T)} || T <- Tabs],
	    safe_write(B, Defs)
    end.

safe_apply(B, write, [_, Items]) when Items == [] ->
    B;
safe_apply(B, What, Args) ->
    Abort = fun(R) -> abort_write(B, What, Args, R) end,
    receive
	{'EXIT', Pid, R} -> Abort({'EXIT', Pid, R})
    after 0 ->
	    Mod = B#backup_args.module,
	    case catch apply(Mod, What, Args) of
		{ok, Opaque} -> B#backup_args{opaque=Opaque};
		{error, R} -> Abort(R);
		R -> Abort(R)
	    end
    end.

abort_write(B, What, Args, Reason) ->
    Mod = B#backup_args.module,
    Opaque = B#backup_args.opaque,
    dbg_out("Failed to perform backup. M=~p:F=~p:A=~p -> ~p~n",
	    [Mod, What, Args, Reason]),
    case catch apply(Mod, abort_write, [Opaque]) of
	{ok, _Res} ->
	    throw({error, Reason});
	Other ->
	    error("Failed to abort backup. ~p:~p~p -> ~p~n",
		  [Mod, abort_write, [Opaque], Other]),
	    throw({error, Reason})
    end.
	
backup_tab(Tab, B) ->
    Name = B#backup_args.name,
    case mnesia_checkpoint:most_local_node(Name, Tab) of
	{ok, Node} when Node == node() ->
	    tab_copier(self(), B, Tab);
	{ok, Node} ->
	    RemoteB = B,
	    Pid = spawn_link(Node, ?MODULE, tab_copier, [self(), RemoteB, Tab]),
	    RecName = val({Tab, record_name}),
	    tab_receiver(Pid, B, Tab, RecName, 0);
	{error, Reason} ->
	    abort_write(B, {?MODULE, backup_tab}, [Tab, B], {error, Reason})
    end.
    
tab_copier(Pid, B, Tab) when record(B, backup_args) ->
    %% Intentional crash at exit
    Name = B#backup_args.name,
    PrevName = B#backup_args.prev_name,
    {FirstName, FirstSource} = select_source(Tab, Name, PrevName),

    ?eval_debug_fun({?MODULE, tab_copier, pre}, [{name, Name}, {tab, Tab}]),
    Res = handle_more(Pid, B, Tab, FirstName, FirstSource, Name),
    ?eval_debug_fun({?MODULE, tab_copier, post}, [{name, Name}, {tab, Tab}]),

    handle_last(Pid, Res).

select_source(Tab, Name, PrevName) ->
    if
	Tab == schema ->
	    %% Always full backup of schema
	    {Name, table};
	Name == PrevName ->
	    %% Full backup
	    {Name, table};
	true ->
	    %% Wants incremental backup
	    case mnesia_checkpoint:most_local_node(PrevName, Tab) of
		{ok, Node} when Node == node() ->
		    %% Accept incremental backup
		    {PrevName, retainer};
		_ ->
		    %% Do a full backup anyway
		    dbg_out("Incremental backup escalated to full backup: ~p~n", [Tab]),
		    {Name, table}
	    end
    end.

handle_more(Pid, B, Tab, FirstName, FirstSource, Name) ->
    Acc = {0, B},
    case {mnesia_checkpoint:really_retain(Name, Tab),
	  mnesia_checkpoint:really_retain(FirstName, Tab)} of
	{true, true} ->
	    Acc2 = iterate(B, FirstName, Tab, Pid, FirstSource, latest, first, Acc),
	    iterate(B, Name, Tab, Pid, retainer, checkpoint, last, Acc2);
	{false, false}->
	    %% Put the dumped file in the backup
	    %% instead of the ram table. Does
	    %% only apply to ram_copies.
	    iterate(B, Name, Tab, Pid, retainer, checkpoint, last, Acc);
	Bad ->
	    Reason = {"Checkpoints for incremental backup must have same "
		      "setting of ram_overrides_dump",
		      Tab, Name, FirstName, Bad},
	    abort_write(B, {?MODULE, backup_tab}, [Tab, B], {error, Reason})
    end.

handle_last(Pid, {_Count, B}) when Pid == self() ->
    B;
handle_last(Pid, _Acc) ->
    unlink(Pid),
    Pid ! {self(), {last, {ok, dummy}}},
    exit(normal).

iterate(B, Name, Tab, Pid, Source, Age, Pass, Acc) ->
    Fun = 
	if
	    Pid == self() ->
		RecName = val({Tab, record_name}),
		fun(Recs, A) -> copy_records(RecName, Tab, Recs, A) end;
	    true ->
		fun(Recs, A) -> send_records(Pid, Tab, Recs, Pass, A) end
	end,
    case mnesia_checkpoint:iterate(Name, Tab, Fun, Acc, Source, Age) of
	{ok, Acc2} ->
	    Acc2;
	{error, Reason} ->
	    R = {error, {"Tab copier iteration failed", Reason}},
	    abort_write(B, {?MODULE, iterate}, [self(), B, Tab], R)
    end.

copy_records(_RecName, _Tab, [], Acc) ->
    Acc;
copy_records(RecName, Tab, Recs, {Count, B}) ->
    Recs2 = rec_filter(B, Tab, RecName, Recs),
    B2 = safe_write(B, Recs2),
    {Count + 1, B2}.

send_records(Pid, Tab, Recs, Pass, {Count, B}) ->
    receive
	{Pid, more, Count} ->
	    if
		Pass == last, Recs == [] ->
		    {Count, B};
		true ->
		    Next = Count + 1,
		    Pid ! {self(), {more, Next, Recs}},
		    {Next, B}
	    end;
	Msg ->
	    exit({send_records_unexpected_msg, Tab, Msg})
    end.

tab_receiver(Pid, B, Tab, RecName, Slot) ->
    Pid ! {self(), more, Slot},
    receive
	{Pid, {more, Next, Recs}} ->
	    Recs2 = rec_filter(B, Tab, RecName, Recs),
	    B2 = safe_write(B, Recs2),
	    tab_receiver(Pid, B2, Tab, RecName, Next);
	
	{Pid, {last, {ok,_}}} ->
	    B;

	{'EXIT', Pid, {error, R}} ->
	    Reason = {error, {"Tab copier crashed", R}},
	    abort_write(B, {?MODULE, remote_tab_sender}, [self(), B, Tab], Reason);
	{'EXIT', Pid, R} ->
	    Reason = {error, {"Tab copier crashed", {'EXIT', R}}},
	    abort_write(B, {?MODULE, remote_tab_sender}, [self(), B, Tab], Reason);
	Msg ->
	    R = {error, {"Tab receiver got unexpected msg", Msg}}, 
	    abort_write(B, {?MODULE, remote_tab_sender}, [self(), B, Tab], R)
    end.

rec_filter(B, schema, _RecName, Recs) ->
    case catch mnesia_bup:refresh_cookie(Recs, B#backup_args.cookie) of
	Recs2 when list(Recs2) ->
	    Recs2;
	{error, _Reason} ->
	    %% No schema table cookie
	    Recs
    end;
rec_filter(_B, Tab, Tab, Recs) ->
    Recs;
rec_filter(_B, Tab, _RecName, Recs) ->
    [setelement(1, Rec, Tab) || Rec <- Recs].

ets2dcd(Tab) ->
    ets2dcd(Tab, dcd).

ets2dcd(Tab, Ftype) ->
    Fname =
	case Ftype of
	    dcd -> mnesia_lib:tab2dcd(Tab);
	    dmp -> mnesia_lib:tab2dmp(Tab)
	end,    
    TmpF = mnesia_lib:tab2tmp(Tab),
    file:delete(TmpF),    
    Log  = open_log({Tab, ets2dcd}, dcd_log_header(), TmpF, false),
    mnesia_lib:db_fixtable(ram_copies, Tab, true),
    ok   = ets2dcd(mnesia_lib:db_init_chunk(ram_copies, Tab, 1000), Tab, Log),
    mnesia_lib:db_fixtable(ram_copies, Tab, false),
    close_log(Log),
    ok = file:rename(TmpF, Fname),
    %% Remove old log data which is now in the new dcd.
    %% No one else should be accessing this file!
    file:delete(mnesia_lib:tab2dcl(Tab)),
    ok.

ets2dcd('$end_of_table', _Tab, _Log) ->
    ok;
ets2dcd({Recs, Cont}, Tab, Log) -> 
    ok = disk_log:alog_terms(Log, Recs),     
    ets2dcd(mnesia_lib:db_chunk(ram_copies, Cont), Tab, Log).

dcd2ets(Tab) ->
    dcd2ets(Tab, mnesia_monitor:get_env(auto_repair)).

dcd2ets(Tab, Rep) ->
    Dcd = mnesia_lib:tab2dcd(Tab),
    case mnesia_lib:exists(Dcd) of
	true ->
	    Log = open_log({Tab, dcd2ets}, dcd_log_header(), Dcd, 
			   true, Rep, read_only),
	    Data = chunk_log(Log, start),
	    ok = insert_dcdchunk(Data, Log, Tab),
	    close_log(Log),
	    load_dcl(Tab, Rep); 
	false -> %% Handle old dets files, and conversion from disc_only to disc.
	    Fname = mnesia_lib:tab2dat(Tab),
	    Type = val({Tab, setorbag}),
	    case mnesia_lib:dets_to_ets(Tab, Tab, Fname, Type, Rep, yes) of
		loaded ->
		    ets2dcd(Tab),
		    file:delete(Fname),
		    0;
		{error, Error} ->
		    erlang:fault({"Failed to load table from disc", [Tab, Error]})
	    end
    end.

insert_dcdchunk({Cont, [LogH | Rest]}, Log, Tab) 
  when record(LogH, log_header),  
       LogH#log_header.log_kind == dcd_log, 
       LogH#log_header.log_version >= "1.0" ->    
    insert_dcdchunk({Cont, Rest}, Log, Tab);   

insert_dcdchunk({Cont, Recs}, Log, Tab) ->     
    true = ets:insert(Tab, Recs),
    insert_dcdchunk(chunk_log(Log, Cont), Log, Tab);
insert_dcdchunk(eof, _Log, _Tab) ->
    ok.

load_dcl(Tab, Rep) ->
    FName = mnesia_lib:tab2dcl(Tab),
    case mnesia_lib:exists(FName) of
	true -> 	    
	    Name = {load_dcl,Tab},
	    open_log(Name, 
		     dcl_log_header(), 
		     FName, 
		     true,
		     Rep, 
		     read_only),
	    FirstChunk = chunk_log(Name, start),
            N = insert_logchunk(FirstChunk, Name, 0),
	    close_log(Name),
	    N;
	false ->
	    0
    end.

insert_logchunk({C2, Recs}, Tab, C) ->
    N = add_recs(Recs, C),
    insert_logchunk(chunk_log(Tab, C2), Tab, C+N);
insert_logchunk(eof, _Tab, C) ->
    C.

add_recs([{{Tab, _Key}, Val, write} | Rest], N) ->
    true = ets:insert(Tab, Val),
    add_recs(Rest, N+1);
add_recs([{{Tab, Key}, _Val, delete} | Rest], N) ->
    true = ets:delete(Tab, Key),
    add_recs(Rest, N+1);
add_recs([{{Tab, _Key}, Val, delete_object} | Rest], N) ->
    true = ets:match_delete(Tab, Val),
    add_recs(Rest, N+1);
add_recs([{{Tab, Key}, Val, update_counter} | Rest], N) ->
    {RecName, Incr} = Val,
    case catch ets:update_counter(Tab, Key, Incr) of
	CounterVal when integer(CounterVal) ->
	    ok;
	_ when Incr < 0 ->
	    Zero = {RecName, Key, 0},
	    true = ets:insert(Tab, Zero);
	_ ->
	    Zero = {RecName, Key, Incr},
	    true = ets:insert(Tab, Zero)
    end,
    add_recs(Rest, N+1);
add_recs([LogH|Rest], N) 
  when record(LogH, log_header),  
       LogH#log_header.log_kind == dcl_log, 
       LogH#log_header.log_version >= "1.0" ->    
    add_recs(Rest, N);
add_recs([{{Tab, _Key}, _Val, clear_table} | Rest], N) ->
    true = ets:match_delete(Tab, '_'),
    add_recs(Rest, N+ets:info(Tab, size));  
add_recs([], N) ->
    N.