mnesia_monitor.erl

OTP是开放电信平台的简称

%% ``The contents of this file are subject to the Erlang Public License,
%% Version 1.1, (the "License"); you may not use this file except in
%% compliance with the License. You should have received a copy of the
%% Erlang Public License along with this software. If not, it can be
%% retrieved via the world wide web at http://www.erlang.org/.
%% 
%% Software distributed under the License is distributed on an "AS IS"
%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
%% the License for the specific language governing rights and limitations
%% under the License.
%% 
%% The Initial Developer of the Original Code is Ericsson Utvecklings AB.
%% Portions created by Ericsson are Copyright 1999, Ericsson Utvecklings
%% AB. All Rights Reserved.''
%% 
%%     $Id$
%%
-module(mnesia_monitor).

-behaviour(gen_server).

%% Public exports
-export([
	 close_dets/1,
	 close_log/1,
	 detect_inconcistency/2,
	 get_env/1,
	 init/0,
	 mktab/2,
	 unsafe_mktab/2,
	 mnesia_down/2,
	 needs_protocol_conversion/1,
	 negotiate_protocol/1,
	 disconnect/1,
	 open_dets/2,
	 unsafe_open_dets/2,
	 open_log/1,
	 patch_env/2,
	 protocol_version/0,
	 reopen_log/3,
	 set_env/2,
	 start/0,
	 start_proc/4,
	 terminate_proc/3,
	 unsafe_close_dets/1,
	 unsafe_close_log/1,
	 use_dir/0,
	 do_check_type/2
	]).

%% gen_server callbacks
-export([
	 init/1,
	 handle_call/3,
	 handle_cast/2,
	 handle_info/2,
	 terminate/2,
	 code_change/3
	]).

%% Internal exports
-export([
	 call/1,
	 cast/1,
	 detect_partitioned_network/2,
	 has_remote_mnesia_down/1,
	 negotiate_protocol_impl/2
	]).

-import(mnesia_lib, [dbg_out/2, verbose/2, error/2, fatal/2, set/2]).

-include("mnesia.hrl").

-record(state, {supervisor, pending_negotiators = [], 
		going_down = [], tm_started = false, early_connects = [],
		connecting, mq = []}).

-define(current_protocol_version, {7,6}).

-define(previous_protocol_version, {7,5}).

start() ->
    gen_server:start_link({local, ?MODULE}, ?MODULE,
			  [self()], [{timeout, infinity}
				     %% ,{debug, [trace]}
				    ]).

init() ->
    call(init).

mnesia_down(From, Node) ->
    cast({mnesia_down, From, Node}).

mktab(Tab, Args) ->
    unsafe_call({mktab, Tab, Args}).
unsafe_mktab(Tab, Args) ->
    unsafe_call({unsafe_mktab, Tab, Args}).

open_dets(Tab, Args) ->
    unsafe_call({open_dets, Tab, Args}).
unsafe_open_dets(Tab, Args) ->
    unsafe_call({unsafe_open_dets, Tab, Args}).

close_dets(Tab) ->
    unsafe_call({close_dets, Tab}).

unsafe_close_dets(Name) ->
    unsafe_call({unsafe_close_dets, Name}).

open_log(Args) ->
    unsafe_call({open_log, Args}).

reopen_log(Name, Fname, Head) ->
    unsafe_call({reopen_log, Name, Fname, Head}).

close_log(Name) ->
    unsafe_call({close_log, Name}).

unsafe_close_log(Name) ->
    unsafe_call({unsafe_close_log, Name}).


disconnect(Node) ->
    cast({disconnect, Node}).

%% Returns GoodNoodes
%% Creates a link to each compatible monitor and
%% protocol_version to agreed version upon success

negotiate_protocol([]) -> [];
negotiate_protocol(Nodes) ->
    call({negotiate_protocol, Nodes}).

negotiate_protocol_impl(Nodes, Requester) ->
    Version    = mnesia:system_info(version),
    Protocols  = acceptable_protocol_versions(),
    MonitorPid = whereis(?MODULE),
    Msg = {negotiate_protocol, MonitorPid, Version, Protocols},
    {Replies, _BadNodes} = multicall(Nodes, Msg),
    Res = check_protocol(Replies, Protocols),
    ?MODULE ! {protocol_negotiated,Requester,Res},
    unlink(whereis(?MODULE)),
    ok.

check_protocol([{Node, {accept, Mon, Version, Protocol}} | Tail], Protocols) ->
    case lists:member(Protocol, Protocols) of
	true ->
	    case Protocol == protocol_version() of
		true -> 
		    set({protocol, Node}, {Protocol, false});
		false ->
		    set({protocol, Node}, {Protocol, true})
	    end,
	    [node(Mon) | check_protocol(Tail, Protocols)]; 
	false  ->
	    verbose("Failed to connect with ~p. ~p protocols rejected. "
		    "expected version = ~p, expected protocol = ~p~n",
		    [Node, Protocols, Version, Protocol]),
	    unlink(Mon), % Get rid of unneccessary link
	    check_protocol(Tail, Protocols)
    end;
check_protocol([{Node, {reject, _Mon, Version, Protocol}} | Tail], Protocols) ->
    verbose("Failed to connect with ~p. ~p protocols rejected. "
	    "expected version = ~p, expected protocol = ~p~n",
	    [Node, Protocols, Version, Protocol]),
    check_protocol(Tail, Protocols);
check_protocol([{error, _Reason} | Tail], Protocols) ->
    dbg_out("~p connect failed error: ~p~n", [?MODULE, _Reason]),
    check_protocol(Tail, Protocols);
check_protocol([{badrpc, _Reason} | Tail], Protocols) ->
    dbg_out("~p connect failed badrpc: ~p~n", [?MODULE, _Reason]),
    check_protocol(Tail, Protocols);
check_protocol([], [Protocol | _Protocols]) ->
    set(protocol_version, Protocol),
    [].

protocol_version() -> 
    case ?catch_val(protocol_version) of
	{'EXIT', _} -> ?current_protocol_version;
	Version -> Version
    end.

%% A sorted list of acceptable protocols the
%% preferred protocols are first in the list
acceptable_protocol_versions() ->
    [protocol_version(), ?previous_protocol_version].
    
needs_protocol_conversion(Node) ->
    case {?catch_val({protocol, Node}), protocol_version()} of
	{{'EXIT', _}, _} ->
	    false;
	{{_, Bool}, ?current_protocol_version} -> 
	    Bool;
	{{_, Bool}, _} -> 
	    not Bool
    end.

cast(Msg) ->
    case whereis(?MODULE) of
	undefined -> ignore;
	Pid ->  gen_server:cast(Pid, Msg)
    end.

unsafe_call(Msg) ->
    case whereis(?MODULE) of
	undefined -> {error, {node_not_running, node()}};
	Pid -> gen_server:call(Pid, Msg, infinity)
    end.

call(Msg) ->
    case whereis(?MODULE) of
	undefined ->
	    {error, {node_not_running, node()}};
	Pid ->
	    link(Pid),
	    Res = gen_server:call(Pid, Msg, infinity),
	    unlink(Pid),

            %% We get an exit signal if server dies
	    receive
		{'EXIT', Pid, _Reason} ->
		    {error, {node_not_running, node()}}
	    after 0 ->
		    ignore
	    end,
	    Res
    end.

multicall(Nodes, Msg) ->
    rpc:multicall(Nodes, ?MODULE, call, [Msg]).

start_proc(Who, Mod, Fun, Args) ->
    Args2 = [Who, Mod, Fun, Args],
    proc_lib:start_link(mnesia_sp, init_proc, Args2, infinity).

terminate_proc(Who, R, State) when R /= shutdown, R /= killed ->
    fatal("~p crashed: ~p state: ~p~n", [Who, R, State]);

terminate_proc(Who, Reason, _State) ->
    mnesia_lib:verbose("~p terminated: ~p~n", [Who, Reason]),
    ok.

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%% Callback functions from gen_server

%%----------------------------------------------------------------------
%% Func: init/1
%% Returns: {ok, State}          |
%%          {ok, State, Timeout} |
%%          {stop, Reason}
%%----------------------------------------------------------------------
init([Parent]) ->
    process_flag(trap_exit, true),
    ?ets_new_table(mnesia_gvar, [set, public, named_table]), 
    set(subscribers, []),
    mnesia_lib:verbose("~p starting: ~p~n", [?MODULE, self()]),
    Version = mnesia:system_info(version),
    set(version, Version),
    dbg_out("Version: ~p~n", [Version]),
    
    case catch process_config_args(env()) of
	ok ->
	    mnesia_lib:set({'$$$_report', current_pos}, 0),
	    Level = mnesia_lib:val(debug),
	    mnesia_lib:verbose("Mnesia debug level set to ~p\n", [Level]),
	    set(mnesia_status, starting), %%  set start status
	    set({current, db_nodes}, [node()]),
	    set(use_dir, use_dir()),
	    mnesia_lib:create_counter(trans_aborts),
	    mnesia_lib:create_counter(trans_commits),
	    mnesia_lib:create_counter(trans_log_writes),
	    Left = get_env(dump_log_write_threshold),
	    mnesia_lib:set_counter(trans_log_writes_left, Left),
	    mnesia_lib:create_counter(trans_log_writes_prev),
	    mnesia_lib:create_counter(trans_restarts),
	    mnesia_lib:create_counter(trans_failures),
	    ?ets_new_table(mnesia_held_locks, [bag, public, named_table]), 
	    ?ets_new_table(mnesia_tid_locks, [bag, public, named_table]),
	    ?ets_new_table(mnesia_sticky_locks, [set, public, named_table]),
	    ?ets_new_table(mnesia_lock_queue, 
			   [bag, public, named_table, {keypos, 2}]),
	    ?ets_new_table(mnesia_lock_counter, [set, public, named_table]),
	    set(checkpoints, []),
	    set(pending_checkpoints, []),
	    set(pending_checkpoint_pids, []),
	    
	    {ok, #state{supervisor = Parent}};
	{'EXIT', Reason} ->
	    mnesia_lib:report_fatal("Bad configuration: ~p~n", [Reason]),
	    {stop, {bad_config, Reason}}
    end.

use_dir() ->
    case ?catch_val(use_dir) of
	{'EXIT', _} ->
	    case get_env(schema_location) of
		disc -> true;
		opt_disc -> non_empty_dir();
		ram -> false
	    end;
	Bool ->
	    Bool
    end.

%% Returns true if the Mnesia directory contains
%% important files
non_empty_dir() ->
    mnesia_lib:exists(mnesia_bup:fallback_bup()) or
    mnesia_lib:exists(mnesia_lib:tab2dmp(schema)) or
    mnesia_lib:exists(mnesia_lib:tab2dat(schema)).

%%----------------------------------------------------------------------
%% Func: handle_call/3
%% Returns: {reply, Reply, State}          |
%%          {reply, Reply, State, Timeout} |
%%          {noreply, State}               |
%%          {noreply, State, Timeout}      |
%%          {stop, Reason, Reply, State}   | (terminate/2 is called)
%%----------------------------------------------------------------------

handle_call({mktab, Tab, Args}, _From, State) ->
    case catch ?ets_new_table(Tab, Args) of
	{'EXIT', ExitReason} ->
	    Msg = "Cannot create ets table",
	    Reason = {system_limit, Msg, Tab, Args, ExitReason},
	    fatal("~p~n", [Reason]),
	    {noreply, State};
	Reply ->
	    {reply, Reply, State}
    end;

handle_call({unsafe_mktab, Tab, Args}, _From, State) ->
    case catch ?ets_new_table(Tab, Args) of
	{'EXIT', ExitReason} ->
	    {reply, {error, ExitReason}, State};
	Reply ->
	    {reply, Reply, State}
    end;


handle_call({open_dets, Tab, Args}, _From, State) ->
    case mnesia_lib:dets_sync_open(Tab, Args) of
	{ok, Tab} ->
	    {reply, {ok, Tab}, State};

	{error, Reason} ->
	    Msg = "Cannot open dets table",
	    Error = {error, {Msg, Tab, Args, Reason}},
	    fatal("~p~n", [Error]),
	    {noreply, State}
    end;

handle_call({unsafe_open_dets, Tab, Args}, _From, State) ->
    case mnesia_lib:dets_sync_open(Tab, Args) of
	{ok, Tab} ->
	    {reply, {ok, Tab}, State};
	{error, Reason} ->
	    {reply, {error,Reason}, State}
    end;

handle_call({close_dets, Tab}, _From, State) ->
    ok = mnesia_lib:dets_sync_close(Tab),
    {reply, ok, State};

handle_call({unsafe_close_dets, Tab}, _From, State) ->
    mnesia_lib:dets_sync_close(Tab),
    {reply, ok, State};

handle_call({open_log, Args}, _From, State) ->
    Res = disk_log:open([{notify, true}|Args]),
    {reply, Res, State};

handle_call({reopen_log, Name, Fname, Head}, _From, State) ->
    case disk_log:reopen(Name, Fname, Head) of
	ok ->
	    {reply, ok, State};

        {error, Reason} ->
	    Msg = "Cannot rename disk_log file",
            Error = {error, {Msg, Name, Fname, Head, Reason}},
	    fatal("~p~n", [Error]),
 	    {noreply, State}
    end;

handle_call({close_log, Name}, _From, State) ->
    case disk_log:close(Name) of
	ok ->
	    {reply, ok, State};

        {error, Reason} ->
	    Msg = "Cannot close disk_log file",
            Error = {error, {Msg, Name, Reason}},
	    fatal("~p~n", [Error]),
	    {noreply, State}
    end;

handle_call({unsafe_close_log, Name}, _From, State) ->
    disk_log:close(Name),
    {reply, ok, State};

handle_call({negotiate_protocol, Mon, _Version, _Protocols}, _From, State) 
  when State#state.tm_started == false ->
    State2 =  State#state{early_connects = [node(Mon) | State#state.early_connects]},    
    {reply, {node(), {reject, self(), uninitialized, uninitialized}}, State2};

%% From remote monitor..
handle_call({negotiate_protocol, Mon, Version, Protocols}, From, State)
  when node(Mon) /= node() ->
    Protocol = protocol_version(),
    MyVersion = mnesia:system_info(version),
    case lists:member(Protocol, Protocols) of
	true ->
	    accept_protocol(Mon, MyVersion, Protocol, From, State);
	false ->