diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/credit_flow.erl | 3 | ||||
| -rw-r--r-- | src/rabbit_amqqueue.erl | 51 | ||||
| -rw-r--r-- | src/rabbit_amqqueue_process.erl | 83 | ||||
| -rw-r--r-- | src/rabbit_channel.erl | 5 | ||||
| -rw-r--r-- | src/rabbit_guid.erl | 10 | ||||
| -rw-r--r-- | src/rabbit_mnesia.erl | 2 | ||||
| -rw-r--r-- | src/rabbit_nodes.erl | 16 | ||||
| -rw-r--r-- | src/rabbit_tests.erl | 2 | ||||
| -rw-r--r-- | src/rabbit_upgrade.erl | 12 | ||||
| -rw-r--r-- | src/rabbit_version.erl | 5 | ||||
| -rw-r--r-- | src/supervisor2.erl | 63 |
11 files changed, 148 insertions, 104 deletions
diff --git a/src/credit_flow.erl b/src/credit_flow.erl index 072f4d9dc5..ba99811f70 100644 --- a/src/credit_flow.erl +++ b/src/credit_flow.erl @@ -96,7 +96,8 @@ peer_down(Peer) -> %% credit_deferred and thus send messages into the void... unblock(Peer), erase({credit_from, Peer}), - erase({credit_to, Peer}). + erase({credit_to, Peer}), + ok. %% -------------------------------------------------------------------------- diff --git a/src/rabbit_amqqueue.erl b/src/rabbit_amqqueue.erl index 9b6f14ca06..c2724a12e2 100644 --- a/src/rabbit_amqqueue.erl +++ b/src/rabbit_amqqueue.erl @@ -32,7 +32,7 @@ %% internal --export([internal_declare/2, internal_delete/1, run_backing_queue/3, +-export([internal_declare/2, internal_delete/2, run_backing_queue/3, set_ram_duration_target/2, set_maximum_since_use/2]). -include("rabbit.hrl"). @@ -144,11 +144,11 @@ -spec(notify_sent_queue_down/1 :: (pid()) -> 'ok'). -spec(unblock/2 :: (pid(), pid()) -> 'ok'). -spec(flush_all/2 :: (qpids(), pid()) -> 'ok'). --spec(internal_delete/1 :: - (name()) -> rabbit_types:ok_or_error('not_found') | - rabbit_types:connection_exit() | - fun (() -> rabbit_types:ok_or_error('not_found') | - rabbit_types:connection_exit())). +-spec(internal_delete/2 :: + (name(), pid()) -> rabbit_types:ok_or_error('not_found') | + rabbit_types:connection_exit() | + fun (() -> rabbit_types:ok_or_error('not_found') | + rabbit_types:connection_exit())). -spec(run_backing_queue/3 :: (pid(), atom(), (fun ((atom(), A) -> {[rabbit_types:msg_id()], A}))) -> 'ok'). @@ -231,7 +231,7 @@ internal_declare(Q = #amqqueue{name = QueueName}, false) -> [ExistingQ = #amqqueue{pid = QPid}] -> case rabbit_misc:is_process_alive(QPid) of true -> rabbit_misc:const(ExistingQ); - false -> TailFun = internal_delete(QueueName), + false -> TailFun = internal_delete(QueueName, QPid), fun () -> TailFun(), ExistingQ end end end @@ -534,13 +534,19 @@ internal_delete1(QueueName) -> %% after the transaction. rabbit_binding:remove_for_destination(QueueName). -internal_delete(QueueName) -> +internal_delete(QueueName, QPid) -> rabbit_misc:execute_mnesia_tx_with_tail( fun () -> case mnesia:wread({rabbit_queue, QueueName}) of [] -> rabbit_misc:const({error, not_found}); [_] -> Deletions = internal_delete1(QueueName), - rabbit_binding:process_deletions(Deletions) + T = rabbit_binding:process_deletions(Deletions), + fun() -> + ok = T(), + ok = rabbit_event:notify(queue_deleted, + [{pid, QPid}, + {name, QueueName}]) + end end end). @@ -555,14 +561,25 @@ set_maximum_since_use(QPid, Age) -> on_node_down(Node) -> rabbit_misc:execute_mnesia_tx_with_tail( - fun () -> Dels = qlc:e(qlc:q([delete_queue(QueueName) || - #amqqueue{name = QueueName, pid = Pid, - slave_pids = []} - <- mnesia:table(rabbit_queue), - node(Pid) == Node])), - rabbit_binding:process_deletions( - lists:foldl(fun rabbit_binding:combine_deletions/2, - rabbit_binding:new_deletions(), Dels)) + fun () -> QsDels = + qlc:e(qlc:q([{{QName, Pid}, delete_queue(QName)} || + #amqqueue{name = QName, pid = Pid, + slave_pids = []} + <- mnesia:table(rabbit_queue), + node(Pid) == Node])), + {Qs, Dels} = lists:unzip(QsDels), + T = rabbit_binding:process_deletions( + lists:foldl(fun rabbit_binding:combine_deletions/2, + rabbit_binding:new_deletions(), Dels)), + fun () -> + T(), + lists:foreach( + fun({QName, QPid}) -> + ok = rabbit_event:notify(queue_deleted, + [{pid, QPid}, + {name, QName}]) + end, Qs) + end end). delete_queue(QueueName) -> diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl index 106a99607c..e1fd9bbca5 100644 --- a/src/rabbit_amqqueue_process.erl +++ b/src/rabbit_amqqueue_process.erl @@ -24,9 +24,6 @@ -define(SYNC_INTERVAL, 25). %% milliseconds -define(RAM_DURATION_UPDATE_INTERVAL, 5000). --define(BASE_MESSAGE_PROPERTIES, - #message_properties{expiry = undefined, needs_confirming = false}). - -export([start_link/1, info_keys/0]). -export([init_with_backing_queue_state/7]). @@ -183,16 +180,13 @@ terminate({shutdown, _} = R, State = #q{backing_queue = BQ}) -> terminate(Reason, State = #q{q = #amqqueue{name = QName}, backing_queue = BQ}) -> %% FIXME: How do we cancel active subscriptions? - terminate_shutdown(fun (BQS) -> - rabbit_event:notify( - queue_deleted, [{pid, self()}, - {name, QName}]), - BQS1 = BQ:delete_and_terminate(Reason, BQS), - %% don't care if the internal delete - %% doesn't return 'ok'. - rabbit_amqqueue:internal_delete(qname(State)), - BQS1 - end, State). + terminate_shutdown( + fun (BQS) -> + BQS1 = BQ:delete_and_terminate(Reason, BQS), + %% don't care if the internal delete doesn't return 'ok'. + rabbit_amqqueue:internal_delete(QName, self()), + BQS1 + end, State). code_change(_OldVsn, State, _Extra) -> {ok, State}. @@ -333,12 +327,10 @@ ensure_expiry_timer(State = #q{expires = undefined}) -> State; ensure_expiry_timer(State = #q{expires = Expires}) -> case is_unused(State) of - true -> - NewState = stop_expiry_timer(State), - TRef = erlang:send_after(Expires, self(), maybe_expire), - NewState#q{expiry_timer_ref = TRef}; - false -> - State + true -> NewState = stop_expiry_timer(State), + TRef = erlang:send_after(Expires, self(), maybe_expire), + NewState#q{expiry_timer_ref = TRef}; + false -> State end. ensure_stats_timer(State) -> @@ -532,15 +524,10 @@ attempt_delivery(Delivery = #delivery{sender = SenderPid, {false, BQS1} -> DeliverFun = fun (AckRequired, State1 = #q{backing_queue_state = BQS2}) -> - %% we don't need an expiry here because - %% messages are not being enqueued, so we use - %% an empty message_properties. - {AckTag, BQS3} = - BQ:publish_delivered( - AckRequired, Message, - (?BASE_MESSAGE_PROPERTIES)#message_properties{ - needs_confirming = needs_confirming(Confirm)}, - SenderPid, BQS2), + Props = message_properties(Confirm, State1), + {AckTag, BQS3} = BQ:publish_delivered( + AckRequired, Message, Props, + SenderPid, BQS2), {{Message, false, AckTag}, true, State1#q{backing_queue_state = BQS3}} end, @@ -567,8 +554,7 @@ deliver_or_enqueue(Delivery = #delivery{message = Message, maybe_record_confirm_message(Confirm, State1), case Delivered of true -> State2; - false -> Props = (message_properties(State)) #message_properties{ - needs_confirming = needs_confirming(Confirm)}, + false -> Props = message_properties(Confirm, State), BQS1 = BQ:publish(Message, Props, SenderPid, BQS), ensure_ttl_timer(State2#q{backing_queue_state = BQS1}) end. @@ -696,8 +682,9 @@ discard_delivery(#delivery{sender = SenderPid, backing_queue_state = BQS}) -> State#q{backing_queue_state = BQ:discard(Message, SenderPid, BQS)}. -message_properties(#q{ttl=TTL}) -> - #message_properties{expiry = calculate_msg_expiry(TTL)}. +message_properties(Confirm, #q{ttl = TTL}) -> + #message_properties{expiry = calculate_msg_expiry(TTL), + needs_confirming = needs_confirming(Confirm)}. calculate_msg_expiry(undefined) -> undefined; calculate_msg_expiry(TTL) -> now_micros() + (TTL * 1000). @@ -735,10 +722,9 @@ dead_letter_fun(Reason, _State) -> dead_letter_msg(Msg, AckTag, Reason, State = #q{dlx = DLX}) -> case rabbit_exchange:lookup(DLX) of - {error, not_found} -> - noreply(State); - _ -> - dead_letter_msg_existing_dlx(Msg, AckTag, Reason, State) + {error, not_found} -> noreply(State); + _ -> dead_letter_msg_existing_dlx(Msg, AckTag, Reason, + State) end. dead_letter_msg_existing_dlx(Msg, AckTag, Reason, @@ -755,7 +741,7 @@ dead_letter_msg_existing_dlx(Msg, AckTag, Reason, State1 = lists:foldl(fun monitor_queue/2, State, QPids), State2 = State1#q{publish_seqno = MsgSeqNo + 1}, case QPids of - [] -> {_, BQS1} = BQ:ack([AckTag], BQS), + [] -> {_Guids, BQS1} = BQ:ack([AckTag], BQS), cleanup_after_confirm(State2#q{backing_queue_state = BQS1}); _ -> State3 = lists:foldl( @@ -792,11 +778,12 @@ handle_queue_down(QPid, Reason, State = #q{queue_monitors = QMons, error -> noreply(State); {ok, _} -> - #resource{name = QName} = qname(State), - rabbit_log:info("DLQ ~p (for ~p) died~n", [QPid, QName]), + rabbit_log:info("DLQ ~p (for ~s) died~n", + [QPid, rabbit_misc:rs(qname(State))]), + State1 = State#q{queue_monitors = dict:erase(QPid, QMons)}, case gb_trees:lookup(QPid, UQM) of none -> - noreply(State); + noreply(State1); {value, MsgSeqNosSet} -> case rabbit_misc:is_abnormal_termination(Reason) of true -> rabbit_log:warning( @@ -804,9 +791,7 @@ handle_queue_down(QPid, Reason, State = #q{queue_monitors = QMons, [gb_sets:size(MsgSeqNosSet)]); false -> ok end, - handle_confirm(gb_sets:to_list(MsgSeqNosSet), QPid, - State#q{queue_monitors = - dict:erase(QPid, QMons)}) + handle_confirm(gb_sets:to_list(MsgSeqNosSet), QPid, State1) end end. @@ -1285,14 +1270,14 @@ handle_cast({ack, AckTags, ChPid}, State) -> handle_cast({reject, AckTags, Requeue, ChPid}, State) -> noreply(subtract_acks( ChPid, AckTags, State, - fun (State1 = #q{backing_queue = BQ, - backing_queue_state = BQS}) -> - case Requeue of - true -> requeue_and_run(AckTags, State1); - false -> Fun = dead_letter_fun(rejected, State), + case Requeue of + true -> fun (State1) -> requeue_and_run(AckTags, State1) end; + false -> Fun = dead_letter_fun(rejected, State), + fun (State1 = #q{backing_queue = BQ, + backing_queue_state = BQS}) -> BQS1 = BQ:fold(Fun, BQS, AckTags), State1#q{backing_queue_state = BQS1} - end + end end)); handle_cast(delete_immediately, State) -> diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl index bb6636eb80..cac622f856 100644 --- a/src/rabbit_channel.erl +++ b/src/rabbit_channel.erl @@ -1411,11 +1411,10 @@ process_routing_result(routed, QPids, XName, MsgSeqNo, _, State) -> send_nacks([], State) -> State; send_nacks(MXs, State = #ch{tx_status = none}) -> - MsgSeqNos = [ MsgSeqNo || {MsgSeqNo, _} <- MXs ], - coalesce_and_send(MsgSeqNos, + coalesce_and_send([MsgSeqNo || {MsgSeqNo, _} <- MXs], fun(MsgSeqNo, Multiple) -> #'basic.nack'{delivery_tag = MsgSeqNo, - multiple = Multiple} + multiple = Multiple} end, State); send_nacks(_, State) -> maybe_complete_tx(State#ch{tx_status = failed}). diff --git a/src/rabbit_guid.erl b/src/rabbit_guid.erl index f4c425ca20..ba0cb04f71 100644 --- a/src/rabbit_guid.erl +++ b/src/rabbit_guid.erl @@ -19,6 +19,7 @@ -behaviour(gen_server). -export([start_link/0]). +-export([filename/0]). -export([gen/0, gen_secure/0, string/2, binary/2]). -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, @@ -38,6 +39,7 @@ -type(guid() :: binary()). -spec(start_link/0 :: () -> rabbit_types:ok_pid_or_error()). +-spec(filename/0 :: () -> string()). -spec(gen/0 :: () -> guid()). -spec(gen_secure/0 :: () -> guid()). -spec(string/2 :: (guid(), any()) -> string()). @@ -51,8 +53,14 @@ start_link() -> gen_server:start_link({local, ?SERVER}, ?MODULE, [update_disk_serial()], []). +%% We use this to detect a (possibly rather old) Mnesia directory, +%% since it has existed since at least 1.7.0 (as far back as I cared +%% to go). +filename() -> + filename:join(rabbit_mnesia:dir(), ?SERIAL_FILENAME). + update_disk_serial() -> - Filename = filename:join(rabbit_mnesia:dir(), ?SERIAL_FILENAME), + Filename = filename(), Serial = case rabbit_file:read_term_file(Filename) of {ok, [Num]} -> Num; {error, enoent} -> 0; diff --git a/src/rabbit_mnesia.erl b/src/rabbit_mnesia.erl index 4d419fd9a9..c714d3a777 100644 --- a/src/rabbit_mnesia.erl +++ b/src/rabbit_mnesia.erl @@ -508,7 +508,7 @@ init_db(ClusterNodes, Force) -> ok -> ok; %% If we're just starting up a new node we won't have a %% version - version_not_available -> ok = rabbit_version:record_desired() + starting_from_scratch -> ok = rabbit_version:record_desired() end end). diff --git a/src/rabbit_nodes.erl b/src/rabbit_nodes.erl index 329c07dc9f..9a972d9e78 100644 --- a/src/rabbit_nodes.erl +++ b/src/rabbit_nodes.erl @@ -39,15 +39,15 @@ names(Hostname) -> Self = self(), - process_flag(trap_exit, true), - Pid = spawn_link(fun () -> Self ! {names, net_adm:names(Hostname)} end), + Ref = make_ref(), + {Pid, MRef} = spawn_monitor( + fun () -> Self ! {Ref, net_adm:names(Hostname)} end), timer:exit_after(?EPMD_TIMEOUT, Pid, timeout), - Res = receive - {names, Names} -> Names; - {'EXIT', Pid, Reason} -> {error, Reason} - end, - process_flag(trap_exit, false), - Res. + receive + {Ref, Names} -> erlang:demonitor(MRef, [flush]), + Names; + {'DOWN', MRef, process, Pid, Reason} -> {error, Reason} + end. diagnostics(Nodes) -> Hosts = lists:usort([element(2, parts(Node)) || Node <- Nodes]), diff --git a/src/rabbit_tests.erl b/src/rabbit_tests.erl index 85fe54263a..55e4a6f854 100644 --- a/src/rabbit_tests.erl +++ b/src/rabbit_tests.erl @@ -2551,7 +2551,7 @@ test_queue_recover() -> {{_Msg1, true, _AckTag1, CountMinusOne}, VQ2} = rabbit_variable_queue:fetch(true, VQ1), _VQ3 = rabbit_variable_queue:delete_and_terminate(shutdown, VQ2), - rabbit_amqqueue:internal_delete(QName) + rabbit_amqqueue:internal_delete(QName, QPid1) end), passed. diff --git a/src/rabbit_upgrade.erl b/src/rabbit_upgrade.erl index 80f50b38b3..e1a7bcaea8 100644 --- a/src/rabbit_upgrade.erl +++ b/src/rabbit_upgrade.erl @@ -28,7 +28,9 @@ -ifdef(use_specs). -spec(maybe_upgrade_mnesia/0 :: () -> 'ok'). --spec(maybe_upgrade_local/0 :: () -> 'ok' | 'version_not_available'). +-spec(maybe_upgrade_local/0 :: () -> 'ok' | + 'version_not_available' | + 'starting_from_scratch'). -endif. @@ -119,8 +121,13 @@ remove_backup() -> info("upgrades: Mnesia backup removed~n", []). maybe_upgrade_mnesia() -> - AllNodes = rabbit_mnesia:all_clustered_nodes(), + %% rabbit_mnesia:all_clustered_nodes/0 will return [] at this point + %% if we are a RAM node since Mnesia has not started yet. + AllNodes = lists:usort(rabbit_mnesia:all_clustered_nodes() ++ + rabbit_mnesia:read_cluster_nodes_config()), case rabbit_version:upgrades_required(mnesia) of + {error, starting_from_scratch} -> + ok; {error, version_not_available} -> case AllNodes of [_] -> ok; @@ -235,6 +242,7 @@ nodes_running(Nodes) -> maybe_upgrade_local() -> case rabbit_version:upgrades_required(local) of {error, version_not_available} -> version_not_available; + {error, starting_from_scratch} -> starting_from_scratch; {error, _} = Err -> throw(Err); {ok, []} -> ensure_backup_removed(), ok; diff --git a/src/rabbit_version.erl b/src/rabbit_version.erl index 7545d81362..1cc7d6c8da 100644 --- a/src/rabbit_version.erl +++ b/src/rabbit_version.erl @@ -96,7 +96,10 @@ record_desired_for_scope(Scope) -> upgrades_required(Scope) -> case recorded_for_scope(Scope) of {error, enoent} -> - {error, version_not_available}; + case filelib:is_file(rabbit_guid:filename()) of + false -> {error, starting_from_scratch}; + true -> {error, version_not_available} + end; {ok, CurrentHeads} -> with_upgrade_graph( fun (G) -> diff --git a/src/supervisor2.erl b/src/supervisor2.erl index a2f4fae980..8dd8aba829 100644 --- a/src/supervisor2.erl +++ b/src/supervisor2.erl @@ -9,15 +9,15 @@ %% terminated as per the shutdown component of the child_spec. %% %% 3) child specifications can contain, as the restart type, a tuple -%% {permanent, Delay} | {transient, Delay} where Delay >= 0. The -%% delay, in seconds, indicates what should happen if a child, upon -%% being restarted, exceeds the MaxT and MaxR parameters. Thus, if -%% a child exits, it is restarted as normal. If it exits -%% sufficiently quickly and often to exceed the boundaries set by -%% the MaxT and MaxR parameters, and a Delay is specified, then -%% rather than stopping the supervisor, the supervisor instead -%% continues and tries to start up the child again, Delay seconds -%% later. +%% {permanent, Delay} | {transient, Delay} | {intrinsic, Delay} +%% where Delay >= 0 (see point (4) below for intrinsic). The delay, +%% in seconds, indicates what should happen if a child, upon being +%% restarted, exceeds the MaxT and MaxR parameters. Thus, if a +%% child exits, it is restarted as normal. If it exits sufficiently +%% quickly and often to exceed the boundaries set by the MaxT and +%% MaxR parameters, and a Delay is specified, then rather than +%% stopping the supervisor, the supervisor instead continues and +%% tries to start up the child again, Delay seconds later. %% %% Note that you can never restart more frequently than the MaxT %% and MaxR parameters allow: i.e. you must wait until *both* the @@ -31,6 +31,16 @@ %% the MaxT and MaxR parameters to permit the child to be %% restarted. This may require waiting for longer than Delay. %% +%% Sometimes, you may wish for a transient or intrinsic child to +%% exit abnormally so that it gets restarted, but still log +%% nothing. gen_server will log any exit reason other than +%% 'normal', 'shutdown' or {'shutdown', _}. Thus the exit reason of +%% {'shutdown', 'restart'} is interpreted to mean you wish the +%% child to be restarted according to the delay parameters, but +%% gen_server will not log the error. Thus from gen_server's +%% perspective it's a normal exit, whilst from supervisor's +%% perspective, it's an abnormal exit. +%% %% 4) Added an 'intrinsic' restart type. Like the transient type, this %% type means the child should only be restarted if the child exits %% abnormally. Unlike the transient type, if the child exits @@ -529,25 +539,23 @@ restart_child(Pid, Reason, State) -> {ok, State} end. -do_restart({RestartType, Delay}, Reason, Child, State) -> - case restart1(Child, State) of - {ok, NState} -> - {ok, NState}; - {terminate, NState} -> - _TRef = erlang:send_after(trunc(Delay*1000), self(), - {delayed_restart, - {{RestartType, Delay}, Reason, Child}}), - {ok, state_del_child(Child, NState)} - end; +do_restart({permanent = RestartType, Delay}, Reason, Child, State) -> + do_restart_delay({RestartType, Delay}, Reason, Child, State); do_restart(permanent, Reason, Child, State) -> report_error(child_terminated, Reason, Child, State#state.name), restart(Child, State); do_restart(Type, normal, Child, State) -> del_child_and_maybe_shutdown(Type, Child, State); +do_restart({RestartType, Delay}, {shutdown, restart} = Reason, Child, State) + when RestartType =:= transient orelse RestartType =:= intrinsic -> + do_restart_delay({RestartType, Delay}, Reason, Child, State); do_restart(Type, {shutdown, _}, Child, State) -> del_child_and_maybe_shutdown(Type, Child, State); do_restart(Type, shutdown, Child = #child{child_type = supervisor}, State) -> del_child_and_maybe_shutdown(Type, Child, State); +do_restart({RestartType, Delay}, Reason, Child, State) + when RestartType =:= transient orelse RestartType =:= intrinsic -> + do_restart_delay({RestartType, Delay}, Reason, Child, State); do_restart(Type, Reason, Child, State) when Type =:= transient orelse Type =:= intrinsic -> report_error(child_terminated, Reason, Child, State#state.name), @@ -557,8 +565,21 @@ do_restart(temporary, Reason, Child, State) -> NState = state_del_child(Child, State), {ok, NState}. +do_restart_delay({RestartType, Delay}, Reason, Child, State) -> + case restart1(Child, State) of + {ok, NState} -> + {ok, NState}; + {terminate, NState} -> + _TRef = erlang:send_after(trunc(Delay*1000), self(), + {delayed_restart, + {{RestartType, Delay}, Reason, Child}}), + {ok, state_del_child(Child, NState)} + end. + del_child_and_maybe_shutdown(intrinsic, Child, State) -> {shutdown, state_del_child(Child, State)}; +del_child_and_maybe_shutdown({intrinsic, _Delay}, Child, State) -> + {shutdown, state_del_child(Child, State)}; del_child_and_maybe_shutdown(_, Child, State) -> {ok, state_del_child(Child, State)}. @@ -911,7 +932,8 @@ supname(N,_) -> N. %%% Func is {Mod, Fun, Args} == {atom, atom, list} %%% RestartType is permanent | temporary | transient | %%% intrinsic | {permanent, Delay} | -%%% {transient, Delay} where Delay >= 0 +%%% {transient, Delay} | {intrinsic, Delay} +%% where Delay >= 0 %%% Shutdown = integer() | infinity | brutal_kill %%% ChildType = supervisor | worker %%% Modules = [atom()] | dynamic @@ -962,6 +984,7 @@ validRestartType(temporary) -> true; validRestartType(transient) -> true; validRestartType(intrinsic) -> true; validRestartType({permanent, Delay}) -> validDelay(Delay); +validRestartType({intrinsic, Delay}) -> validDelay(Delay); validRestartType({transient, Delay}) -> validDelay(Delay); validRestartType(RestartType) -> throw({invalid_restart_type, RestartType}). |