diff options
Diffstat (limited to 'src/rabbit.erl')
| -rw-r--r-- | src/rabbit.erl | 313 |
1 files changed, 209 insertions, 104 deletions
diff --git a/src/rabbit.erl b/src/rabbit.erl index 3293927a..a8b65f46 100644 --- a/src/rabbit.erl +++ b/src/rabbit.erl @@ -39,6 +39,64 @@ -export([log_location/1]). +%%--------------------------------------------------------------------------- +%% Boot steps. +-export([boot_core_processes/0, + boot_recovery/0, + boot_tcp_listeners/0, + boot_ssl_listeners/0]). + +-rabbit_boot_step({database, + [{mfa, {rabbit_mnesia, init, []}}]}). + +-rabbit_boot_step({core_processes, + [{description, "core processes"}, + {mfa, {?MODULE, boot_core_processes, []}}, + {post, database}, + {pre, core_initialized}]}). + +-rabbit_boot_step({core_initialized, + [{description, "core initialized"}]}). + +-rabbit_boot_step({recovery, + [{mfa, {?MODULE, boot_recovery, []}}, + {post, core_initialized}]}). + +-rabbit_boot_step({persister, + [{mfa, {rabbit_sup, start_child, [rabbit_persister]}}, + {post, recovery}]}). + +-rabbit_boot_step({guid_generator, + [{description, "guid generator"}, + {mfa, {rabbit_sup, start_child, [rabbit_guid]}}, + {post, persister}, + {pre, routing_ready}]}). + +-rabbit_boot_step({routing_ready, + [{description, "message delivery logic ready"}]}). + +-rabbit_boot_step({log_relay, + [{description, "error log relay"}, + {mfa, {rabbit_error_logger, boot, []}}, + {post, routing_ready}]}). + +-rabbit_boot_step({tcp_listeners, + [{description, "TCP listeners"}, + {mfa, {?MODULE, boot_tcp_listeners, []}}, + {post, log_relay}, + {pre, networking_listening}]}). + +-rabbit_boot_step({ssl_listeners, + [{description, "SSL listeners"}, + {mfa, {?MODULE, boot_ssl_listeners, []}}, + {post, tcp_listeners}, + {pre, networking_listening}]}). + +-rabbit_boot_step({networking_listening, + [{description, "network listeners available"}]}). + +%%--------------------------------------------------------------------------- + -import(application). -import(mnesia). -import(lists). @@ -79,7 +137,7 @@ prepare() -> start() -> try ok = prepare(), - ok = rabbit_misc:start_applications(?APPS) + ok = rabbit_misc:start_applications(?APPS) after %%give the error loggers some time to catch up timer:sleep(100) @@ -115,99 +173,15 @@ rotate_logs(BinarySuffix) -> %%-------------------------------------------------------------------- start(normal, []) -> - {ok, SupPid} = rabbit_sup:start_link(), print_banner(), - - lists:foreach( - fun ({Msg, Thunk}) -> - io:format("starting ~-20s ...", [Msg]), - Thunk(), - io:format("done~n"); - ({Msg, M, F, A}) -> - io:format("starting ~-20s ...", [Msg]), - apply(M, F, A), - io:format("done~n") - end, - [{"database", - fun () -> ok = rabbit_mnesia:init() end}, - {"core processes", - fun () -> - ok = start_child(rabbit_exchange_type), - ok = start_child(rabbit_log), - ok = rabbit_hooks:start(), - - ok = rabbit_binary_generator: - check_empty_content_body_frame_size(), - - ok = rabbit_alarm:start(), - - {ok, MemoryWatermark} = - application:get_env(vm_memory_high_watermark), - ok = case MemoryWatermark == 0 of - true -> - ok; - false -> - start_child(vm_memory_monitor, [MemoryWatermark]) - end, - - ok = rabbit_amqqueue:start(), - - ok = start_child(rabbit_router), - ok = start_child(rabbit_node_monitor) - end}, - {"recovery", - fun () -> - ok = maybe_insert_default_data(), - ok = rabbit_exchange:recover(), - ok = rabbit_amqqueue:recover() - end}, - {"persister", - fun () -> - ok = start_child(rabbit_persister) - end}, - {"guid generator", - fun () -> - ok = start_child(rabbit_guid) - end}, - {"builtin applications", - fun () -> - {ok, DefaultVHost} = application:get_env(default_vhost), - ok = error_logger:add_report_handler( - rabbit_error_logger, [DefaultVHost]), - ok = start_builtin_amq_applications() - end}, - {"TCP listeners", - fun () -> - ok = rabbit_networking:start(), - {ok, TcpListeners} = application:get_env(tcp_listeners), - lists:foreach( - fun ({Host, Port}) -> - ok = rabbit_networking:start_tcp_listener(Host, Port) - end, - TcpListeners) - end}, - {"SSL listeners", - fun () -> - case application:get_env(ssl_listeners) of - {ok, []} -> - ok; - {ok, SslListeners} -> - ok = rabbit_misc:start_applications([crypto, ssl]), - - {ok, SslOpts} = application:get_env(ssl_options), - - [rabbit_networking:start_ssl_listener - (Host, Port, SslOpts) || {Host, Port} <- SslListeners], - ok - end - end}]), - + [ok = run_boot_step(Step) || Step <- boot_steps()], io:format("~nbroker running~n"), {ok, SupPid}. + stop(_State) -> terminated_ok = error_logger:delete_report_handler(rabbit_error_logger), ok = rabbit_alarm:stop(), @@ -217,10 +191,156 @@ stop(_State) -> end, ok. +%%--------------------------------------------------------------------------- + +boot_error(Format, Args) -> + io:format("BOOT ERROR: " ++ Format, Args), + error_logger:error_msg(Format, Args), + timer:sleep(1000), + exit({?MODULE, failure_during_boot}). + +run_boot_step({StepName, Attributes}) -> + Description = case lists:keysearch(description, 1, Attributes) of + {value, {_, D}} -> D; + false -> StepName + end, + case [MFA || {mfa, MFA} <- Attributes] of + [] -> + io:format("progress -- ~s~n", [Description]); + MFAs -> + io:format("starting ~-20s ...", [Description]), + [case catch apply(M,F,A) of + {'EXIT', Reason} -> + boot_error("FAILED~nReason: ~p~n", [Reason]); + ok -> + ok + end || {M,F,A} <- MFAs], + io:format("done~n"), + ok + end. + +boot_steps() -> + AllApps = [App || {App, _, _} <- application:loaded_applications()], + Modules = lists:usort( + lists:append([Modules + || {ok, Modules} <- + [application:get_key(App, modules) + || App <- AllApps]])), + UnsortedSteps = + lists:flatmap(fun (Module) -> + [{StepName, Attributes} + || {rabbit_boot_step, [{StepName, Attributes}]} + <- Module:module_info(attributes)] + end, Modules), + sort_boot_steps(UnsortedSteps). + +sort_boot_steps(UnsortedSteps) -> + G = digraph:new([acyclic]), + + %% Add vertices, with duplicate checking. + [case digraph:vertex(G, StepName) of + false -> digraph:add_vertex(G, StepName, Step); + _ -> boot_error("Duplicate boot step name: ~w~n", [StepName]) + end || Step = {StepName, _Attrs} <- UnsortedSteps], + + %% Add edges, detecting cycles and missing vertices. + lists:foreach(fun ({StepName, Attributes}) -> + [add_boot_step_dep(G, StepName, PrecedingStepName) + || {post, PrecedingStepName} <- Attributes], + [add_boot_step_dep(G, SucceedingStepName, StepName) + || {pre, SucceedingStepName} <- Attributes] + end, UnsortedSteps), + + %% Use topological sort to find a consistent ordering (if there is + %% one, otherwise fail). + SortedStepsRev = [begin + {StepName, Step} = digraph:vertex(G, StepName), + Step + end || StepName <- digraph_utils:topsort(G)], + SortedSteps = lists:reverse(SortedStepsRev), + + digraph:delete(G), + + %% Check that all mentioned {M,F,A} triples are exported. + case [{StepName, {M,F,A}} + || {StepName, Attributes} <- SortedSteps, + {mfa, {M,F,A}} <- Attributes, + not erlang:function_exported(M, F, length(A))] of + [] -> SortedSteps; + MissingFunctions -> boot_error("Boot step functions not exported: ~p~n", + [MissingFunctions]) + end. + +add_boot_step_dep(G, RunsSecond, RunsFirst) -> + case digraph:add_edge(G, RunsSecond, RunsFirst) of + {error, Reason} -> + boot_error("Could not add boot step dependency of ~w on ~w:~n~s", + [RunsSecond, RunsFirst, + case Reason of + {bad_vertex, V} -> + io_lib:format("Boot step not registered: ~w~n", [V]); + {bad_edge, [First | Rest]} -> + [io_lib:format("Cyclic dependency: ~w", [First]), + [io_lib:format(" depends on ~w", [Next]) + || Next <- Rest], + io_lib:format(" depends on ~w~n", [First])] + end]); + _ -> + ok + end. + +%%--------------------------------------------------------------------------- + +boot_core_processes() -> + ok = rabbit_sup:start_child(rabbit_exchange_type), + ok = rabbit_sup:start_child(rabbit_log), + ok = rabbit_hooks:start(), + + ok = rabbit_binary_generator:check_empty_content_body_frame_size(), + + ok = rabbit_alarm:start(), + + {ok, MemoryWatermark} = application:get_env(vm_memory_high_watermark), + ok = case MemoryWatermark == 0 of + true -> + ok; + false -> + rabbit_sup:start_child(vm_memory_monitor, [MemoryWatermark]) + end, + + ok = rabbit_amqqueue:start(), + + ok = rabbit_sup:start_child(rabbit_router), + ok = rabbit_sup:start_child(rabbit_node_monitor). + +boot_recovery() -> + ok = maybe_insert_default_data(), + ok = rabbit_exchange:recover(), + ok = rabbit_amqqueue:recover(). + +boot_tcp_listeners() -> + ok = rabbit_networking:start(), + {ok, TcpListeners} = application:get_env(tcp_listeners), + [ok = rabbit_networking:start_tcp_listener(Host, Port) + || {Host, Port} <- TcpListeners], + ok. + +boot_ssl_listeners() -> + case application:get_env(ssl_listeners) of + {ok, []} -> + ok; + {ok, SslListeners} -> + ok = rabbit_misc:start_applications([crypto, ssl]), + {ok, SslOpts} = application:get_env(ssl_options), + [rabbit_networking:start_ssl_listener(Host, Port, SslOpts) + || {Host, Port} <- SslListeners], + ok + end. + %--------------------------------------------------------------------------- log_location(Type) -> - case application:get_env(Type, case Type of + case application:get_env(Type, case Type of kernel -> error_logger; sasl -> sasl_error_logger end) of @@ -276,15 +396,6 @@ print_banner() -> lists:foreach(fun ({K, V}) -> io:format(Format, [K, V]) end, Settings), io:nl(). -start_child(Mod) -> - start_child(Mod, []). - -start_child(Mod, Args) -> - {ok,_} = supervisor:start_child(rabbit_sup, - {Mod, {Mod, start_link, Args}, - transient, 100, worker, [Mod]}), - ok. - ensure_working_log_handlers() -> Handlers = gen_event:which_handlers(error_logger), ok = ensure_working_log_handler(error_logger_file_h, @@ -310,7 +421,7 @@ ensure_working_log_handler(OldFHandler, NewFHandler, TTYHandler, throw({error, {cannot_log_to_tty, TTYHandler, not_installed}}) end; - _ -> case lists:member(NewFHandler, Handlers) of + _ -> case lists:member(NewFHandler, Handlers) of true -> ok; false -> case rotate_logs(LogLocation, "", OldFHandler, NewFHandler) of @@ -342,12 +453,6 @@ insert_default_data() -> DefaultReadPerm), ok. -start_builtin_amq_applications() -> - %%TODO: we may want to create a separate supervisor for these so - %%they don't bring down the entire app when they die and fail to - %%restart - ok. - rotate_logs(File, Suffix, Handler) -> rotate_logs(File, Suffix, Handler, Handler). |