diff options
author | Matthias Radestock <matthias@rabbitmq.com> | 2011-06-24 22:36:22 +0100 |
---|---|---|
committer | Matthias Radestock <matthias@rabbitmq.com> | 2011-06-24 22:36:22 +0100 |
commit | bee6e92332ca8f3de1f7b043e04dd2fa4eea5eab (patch) | |
tree | 404949c84beb96f54b47021213de699e22310e1b | |
parent | 3c63aba85361beae3f5733574fd470222e4f3a8c (diff) | |
download | rabbitmq-server-bee6e92332ca8f3de1f7b043e04dd2fa4eea5eab.tar.gz |
tx gone from everywhere
-rw-r--r-- | docs/rabbitmqctl.1.xml | 12 | ||||
-rw-r--r-- | include/rabbit.hrl | 3 | ||||
-rw-r--r-- | src/rabbit_amqqueue.erl | 23 | ||||
-rw-r--r-- | src/rabbit_amqqueue_process.erl | 7 | ||||
-rw-r--r-- | src/rabbit_basic.erl | 32 | ||||
-rw-r--r-- | src/rabbit_channel.erl | 141 | ||||
-rw-r--r-- | src/rabbit_control.erl | 2 | ||||
-rw-r--r-- | src/rabbit_error_logger.erl | 2 | ||||
-rw-r--r-- | src/rabbit_mirror_queue_coordinator.erl | 39 | ||||
-rw-r--r-- | src/rabbit_mirror_queue_slave.erl | 10 | ||||
-rw-r--r-- | src/rabbit_queue_index.erl | 18 | ||||
-rw-r--r-- | src/rabbit_tests.erl | 15 | ||||
-rw-r--r-- | src/rabbit_types.erl | 6 |
13 files changed, 67 insertions, 243 deletions
diff --git a/docs/rabbitmqctl.1.xml b/docs/rabbitmqctl.1.xml index a0f03192..fdb49912 100644 --- a/docs/rabbitmqctl.1.xml +++ b/docs/rabbitmqctl.1.xml @@ -1196,10 +1196,6 @@ <listitem><para>Virtual host in which the channel operates.</para></listitem> </varlistentry> <varlistentry> - <term>transactional</term> - <listitem><para>True if the channel is in transactional mode, false otherwise.</para></listitem> - </varlistentry> - <varlistentry> <term>consumer_count</term> <listitem><para>Number of logical AMQP consumers retrieving messages via the channel.</para></listitem> @@ -1210,11 +1206,6 @@ yet acknowledged.</para></listitem> </varlistentry> <varlistentry> - <term>acks_uncommitted</term> - <listitem><para>Number of acknowledgements received in an as yet - uncommitted transaction.</para></listitem> - </varlistentry> - <varlistentry> <term>prefetch_count</term> <listitem><para>QoS prefetch count limit in force, 0 if unlimited.</para></listitem> </varlistentry> @@ -1239,8 +1230,7 @@ </variablelist> <para> If no <command>channelinfoitem</command>s are specified then pid, - user, transactional, consumer_count, and - messages_unacknowledged are assumed. + user, consumer_count, and messages_unacknowledged are assumed. </para> <para role="example-prefix"> diff --git a/include/rabbit.hrl b/include/rabbit.hrl index 00b7e6e9..3861df2a 100644 --- a/include/rabbit.hrl +++ b/include/rabbit.hrl @@ -67,8 +67,7 @@ is_persistent}). -record(ssl_socket, {tcp, ssl}). --record(delivery, {mandatory, immediate, txn, sender, message, - msg_seq_no}). +-record(delivery, {mandatory, immediate, sender, message, msg_seq_no}). -record(amqp_error, {name, explanation = "", method = none}). -record(event, {type, props, timestamp}). diff --git a/src/rabbit_amqqueue.erl b/src/rabbit_amqqueue.erl index bacb1d21..4d6aaa18 100644 --- a/src/rabbit_amqqueue.erl +++ b/src/rabbit_amqqueue.erl @@ -20,12 +20,12 @@ -export([pseudo_queue/2]). -export([lookup/1, with/2, with_or_die/2, assert_equivalence/5, check_exclusive_access/2, with_exclusive_access_or_die/3, - stat/1, deliver/2, requeue/3, ack/4, reject/4]). + stat/1, deliver/2, requeue/3, ack/3, reject/4]). -export([list/1, info_keys/0, info/1, info/2, info_all/1, info_all/2]). -export([consumers/1, consumers_all/1, consumer_info_keys/0]). -export([basic_get/3, basic_consume/7, basic_cancel/4]). -export([notify_sent/2, unblock/2, flush_all/2]). --export([commit_all/3, rollback_all/3, notify_down_all/2, limit_all/3]). +-export([notify_down_all/2, limit_all/3]). -export([on_node_down/1]). -export([store_queue/1]). @@ -117,12 +117,8 @@ -spec(purge/1 :: (rabbit_types:amqqueue()) -> qlen()). -spec(deliver/2 :: (pid(), rabbit_types:delivery()) -> boolean()). -spec(requeue/3 :: (pid(), [msg_id()], pid()) -> 'ok'). --spec(ack/4 :: - (pid(), rabbit_types:maybe(rabbit_types:txn()), [msg_id()], pid()) - -> 'ok'). +-spec(ack/3 :: (pid(), [msg_id()], pid()) -> 'ok'). -spec(reject/4 :: (pid(), [msg_id()], boolean(), pid()) -> 'ok'). --spec(commit_all/3 :: ([pid()], rabbit_types:txn(), pid()) -> ok_or_errors()). --spec(rollback_all/3 :: ([pid()], rabbit_types:txn(), pid()) -> 'ok'). -spec(notify_down_all/2 :: ([pid()], pid()) -> ok_or_errors()). -spec(limit_all/3 :: ([pid()], pid(), pid() | 'undefined') -> ok_or_errors()). -spec(basic_get/3 :: (rabbit_types:amqqueue(), pid(), boolean()) -> @@ -436,21 +432,12 @@ deliver(QPid, Delivery) -> requeue(QPid, MsgIds, ChPid) -> delegate_call(QPid, {requeue, MsgIds, ChPid}). -ack(QPid, Txn, MsgIds, ChPid) -> - delegate_cast(QPid, {ack, Txn, MsgIds, ChPid}). +ack(QPid, MsgIds, ChPid) -> + delegate_cast(QPid, {ack, MsgIds, ChPid}). reject(QPid, MsgIds, Requeue, ChPid) -> delegate_cast(QPid, {reject, MsgIds, Requeue, ChPid}). -commit_all(QPids, Txn, ChPid) -> - safe_delegate_call_ok( - fun (QPid) -> gen_server2:call(QPid, {commit, Txn, ChPid}, infinity) end, - QPids). - -rollback_all(QPids, Txn, ChPid) -> - delegate:invoke_no_result( - QPids, fun (QPid) -> gen_server2:cast(QPid, {rollback, Txn, ChPid}) end). - notify_down_all(QPids, ChPid) -> safe_delegate_call_ok( fun (QPid) -> gen_server2:call(QPid, {notify_down, ChPid}, infinity) end, diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl index 28fced98..87cdf925 100644 --- a/src/rabbit_amqqueue_process.erl +++ b/src/rabbit_amqqueue_process.erl @@ -495,8 +495,7 @@ run_message_queue(State) -> {_IsEmpty1, State2} = deliver_msgs_to_consumers(Funs, IsEmpty, State1), State2. -attempt_delivery(Delivery = #delivery{txn = none, - sender = ChPid, +attempt_delivery(Delivery = #delivery{sender = ChPid, message = Message, msg_seq_no = MsgSeqNo}, State = #q{backing_queue = BQ, backing_queue_state = BQS}) -> @@ -801,7 +800,7 @@ prioritise_cast(Msg, _State) -> maybe_expire -> 8; drop_expired -> 8; emit_stats -> 7; - {ack, _Txn, _AckTags, _ChPid} -> 7; + {ack, _AckTags, _ChPid} -> 7; {reject, _AckTags, _Requeue, _ChPid} -> 7; {notify_sent, _ChPid} -> 7; {unblock, _ChPid} -> 7; @@ -1028,7 +1027,7 @@ handle_cast({deliver, Delivery}, State) -> %% Asynchronous, non-"mandatory", non-"immediate" deliver mode. noreply(deliver_or_enqueue(Delivery, State)); -handle_cast({ack, none, AckTags, ChPid}, +handle_cast({ack, AckTags, ChPid}, State = #q{backing_queue = BQ, backing_queue_state = BQS}) -> case lookup_ch(ChPid) of not_found -> diff --git a/src/rabbit_basic.erl b/src/rabbit_basic.erl index fa7e3a5a..ec8ed351 100644 --- a/src/rabbit_basic.erl +++ b/src/rabbit_basic.erl @@ -18,8 +18,8 @@ -include("rabbit.hrl"). -include("rabbit_framing.hrl"). --export([publish/1, message/3, message/4, properties/1, delivery/5]). --export([publish/4, publish/7]). +-export([publish/1, message/3, message/4, properties/1, delivery/4]). +-export([publish/4, publish/6]). -export([build_content/2, from_content/1]). %%---------------------------------------------------------------------------- @@ -37,9 +37,8 @@ -spec(publish/1 :: (rabbit_types:delivery()) -> publish_result()). --spec(delivery/5 :: - (boolean(), boolean(), rabbit_types:maybe(rabbit_types:txn()), - rabbit_types:message(), undefined | integer()) -> +-spec(delivery/4 :: + (boolean(), boolean(), rabbit_types:message(), undefined | integer()) -> rabbit_types:delivery()). -spec(message/4 :: (rabbit_exchange:name(), rabbit_router:routing_key(), @@ -53,10 +52,9 @@ -spec(publish/4 :: (exchange_input(), rabbit_router:routing_key(), properties_input(), body_input()) -> publish_result()). --spec(publish/7 :: +-spec(publish/6 :: (exchange_input(), rabbit_router:routing_key(), boolean(), boolean(), - rabbit_types:maybe(rabbit_types:txn()), properties_input(), - body_input()) -> publish_result()). + properties_input(), body_input()) -> publish_result()). -spec(build_content/2 :: (rabbit_framing:amqp_property_record(), binary() | [binary()]) -> rabbit_types:content()). -spec(from_content/1 :: (rabbit_types:content()) -> @@ -73,9 +71,9 @@ publish(Delivery = #delivery{ Other -> Other end. -delivery(Mandatory, Immediate, Txn, Message, MsgSeqNo) -> - #delivery{mandatory = Mandatory, immediate = Immediate, txn = Txn, - sender = self(), message = Message, msg_seq_no = MsgSeqNo}. +delivery(Mandatory, Immediate, Message, MsgSeqNo) -> + #delivery{mandatory = Mandatory, immediate = Immediate, sender = self(), + message = Message, msg_seq_no = MsgSeqNo}. build_content(Properties, BodyBin) when is_binary(BodyBin) -> build_content(Properties, [BodyBin]); @@ -157,19 +155,17 @@ indexof([_ | Rest], Element, N) -> indexof(Rest, Element, N + 1). %% Convenience function, for avoiding round-trips in calls across the %% erlang distributed network. publish(Exchange, RoutingKeyBin, Properties, Body) -> - publish(Exchange, RoutingKeyBin, false, false, none, Properties, - Body). + publish(Exchange, RoutingKeyBin, false, false, Properties, Body). %% Convenience function, for avoiding round-trips in calls across the %% erlang distributed network. -publish(X = #exchange{name = XName}, RKey, Mandatory, Immediate, Txn, - Props, Body) -> - publish(X, delivery(Mandatory, Immediate, Txn, +publish(X = #exchange{name = XName}, RKey, Mandatory, Immediate, Props, Body) -> + publish(X, delivery(Mandatory, Immediate, message(XName, RKey, properties(Props), Body), undefined)); -publish(XName, RKey, Mandatory, Immediate, Txn, Props, Body) -> +publish(XName, RKey, Mandatory, Immediate, Props, Body) -> case rabbit_exchange:lookup(XName) of - {ok, X} -> publish(X, RKey, Mandatory, Immediate, Txn, Props, Body); + {ok, X} -> publish(X, RKey, Mandatory, Immediate, Props, Body); Err -> Err end. diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl index 991b0b06..36471bf5 100644 --- a/src/rabbit_channel.erl +++ b/src/rabbit_channel.erl @@ -30,8 +30,7 @@ prioritise_cast/2]). -record(ch, {state, protocol, channel, reader_pid, writer_pid, conn_pid, - limiter_pid, start_limiter_fun, transaction_id, tx_participants, - next_tag, uncommitted_ack_q, unacked_message_q, + limiter_pid, start_limiter_fun, next_tag, unacked_message_q, user, virtual_host, most_recently_declared_queue, consumer_mapping, blocking, consumer_monitors, queue_collector_pid, stats_timer, confirm_enabled, publish_seqno, unconfirmed_mq, @@ -41,12 +40,10 @@ -define(STATISTICS_KEYS, [pid, - transactional, confirm, consumer_count, messages_unacknowledged, messages_unconfirmed, - acks_uncommitted, prefetch_count, client_flow_blocked]). @@ -173,10 +170,7 @@ init([Channel, ReaderPid, WriterPid, ConnPid, Protocol, User, VHost, conn_pid = ConnPid, limiter_pid = undefined, start_limiter_fun = StartLimiterFun, - transaction_id = none, - tx_participants = sets:new(), next_tag = 1, - uncommitted_ack_q = queue:new(), unacked_message_q = queue:new(), user = User, virtual_host = VHost, @@ -331,7 +325,7 @@ handle_pre_hibernate(State = #ch{stats_timer = StatsTimer}) -> {hibernate, State#ch{stats_timer = StatsTimer1}}. terminate(Reason, State) -> - {Res, _State1} = rollback_and_notify(State), + {Res, _State1} = notify_queues(State), case Reason of normal -> ok = Res; shutdown -> ok = Res; @@ -386,8 +380,8 @@ send_exception(Reason, State = #ch{protocol = Protocol, rabbit_binary_generator:map_exception(Channel, Reason, Protocol), rabbit_log:error("connection ~p, channel ~p - error:~n~p~n", [ConnPid, Channel, Reason]), - %% something bad's happened: rollback_and_notify may not be 'ok' - {_Result, State1} = rollback_and_notify(State), + %% something bad's happened: notify_queues may not be 'ok' + {_Result, State1} = notify_queues(State), case CloseChannel of Channel -> ok = rabbit_writer:send_command(WriterPid, CloseMethod), {noreply, State1}; @@ -589,7 +583,7 @@ handle_method(_Method, _, State = #ch{state = closing}) -> {noreply, State}; handle_method(#'channel.close'{}, _, State = #ch{reader_pid = ReaderPid}) -> - {ok, State1} = rollback_and_notify(State), + {ok, State1} = notify_queues(State), ReaderPid ! {channel_closing, self()}, {noreply, State1}; @@ -601,7 +595,6 @@ handle_method(#'basic.publish'{exchange = ExchangeNameBin, mandatory = Mandatory, immediate = Immediate}, Content, State = #ch{virtual_host = VHostPath, - transaction_id = TxnKey, confirm_enabled = ConfirmEnabled, trace_state = TraceState}) -> ExchangeName = rabbit_misc:r(VHostPath, exchange, ExchangeNameBin), @@ -623,19 +616,15 @@ handle_method(#'basic.publish'{exchange = ExchangeNameBin, rabbit_trace:tap_trace_in(Message, TraceState), {RoutingRes, DeliveredQPids} = rabbit_exchange:publish( - Exchange, - rabbit_basic:delivery(Mandatory, Immediate, TxnKey, Message, - MsgSeqNo)), + Exchange, rabbit_basic:delivery(Mandatory, Immediate, Message, + MsgSeqNo)), State2 = process_routing_result(RoutingRes, DeliveredQPids, ExchangeName, MsgSeqNo, Message, State1), maybe_incr_stats([{ExchangeName, 1} | [{{QPid, ExchangeName}, 1} || QPid <- DeliveredQPids]], publish, State2), - {noreply, case TxnKey of - none -> State2; - _ -> add_tx_participants(DeliveredQPids, State2) - end}; + {noreply, State2}; {error, Reason} -> rabbit_misc:protocol_error(precondition_failed, "invalid message: ~p", [Reason]) @@ -649,22 +638,12 @@ handle_method(#'basic.nack'{delivery_tag = DeliveryTag, handle_method(#'basic.ack'{delivery_tag = DeliveryTag, multiple = Multiple}, - _, State = #ch{transaction_id = TxnKey, - unacked_message_q = UAMQ}) -> + _, State = #ch{unacked_message_q = UAMQ}) -> {Acked, Remaining} = collect_acks(UAMQ, DeliveryTag, Multiple), - QIncs = ack(TxnKey, Acked), - Participants = [QPid || {QPid, _} <- QIncs], + QIncs = ack(Acked), maybe_incr_stats(QIncs, ack, State), - {noreply, case TxnKey of - none -> ok = notify_limiter(State#ch.limiter_pid, Acked), - State#ch{unacked_message_q = Remaining}; - _ -> NewUAQ = queue:join(State#ch.uncommitted_ack_q, - Acked), - add_tx_participants( - Participants, - State#ch{unacked_message_q = Remaining, - uncommitted_ack_q = NewUAQ}) - end}; + ok = notify_limiter(State#ch.limiter_pid, Acked), + {noreply, State#ch{unacked_message_q = Remaining}}; handle_method(#'basic.get'{queue = QueueNameBin, no_ack = NoAck}, @@ -1048,35 +1027,6 @@ handle_method(#'queue.purge'{queue = QueueNameBin, #'queue.purge_ok'{message_count = PurgedMessageCount}); -handle_method(#'tx.select'{}, _, #ch{confirm_enabled = true}) -> - rabbit_misc:protocol_error( - precondition_failed, "cannot switch from confirm to tx mode", []); - -handle_method(#'tx.select'{}, _, State = #ch{transaction_id = none}) -> - {reply, #'tx.select_ok'{}, new_tx(State)}; - -handle_method(#'tx.select'{}, _, State) -> - {reply, #'tx.select_ok'{}, State}; - -handle_method(#'tx.commit'{}, _, #ch{transaction_id = none}) -> - rabbit_misc:protocol_error( - precondition_failed, "channel is not transactional", []); - -handle_method(#'tx.commit'{}, _, State) -> - {reply, #'tx.commit_ok'{}, internal_commit(State)}; - -handle_method(#'tx.rollback'{}, _, #ch{transaction_id = none}) -> - rabbit_misc:protocol_error( - precondition_failed, "channel is not transactional", []); - -handle_method(#'tx.rollback'{}, _, State) -> - {reply, #'tx.rollback_ok'{}, internal_rollback(State)}; - -handle_method(#'confirm.select'{}, _, #ch{transaction_id = TxId}) - when TxId =/= none -> - rabbit_misc:protocol_error( - precondition_failed, "cannot switch from tx to confirm mode", []); - handle_method(#'confirm.select'{nowait = NoWait}, _, State) -> return_ok(State#ch{confirm_enabled = true}, NoWait, #'confirm.select_ok'{}); @@ -1252,55 +1202,17 @@ collect_acks(ToAcc, PrefixAcc, Q, DeliveryTag, Multiple) -> precondition_failed, "unknown delivery tag ~w", [DeliveryTag]) end. -add_tx_participants(MoreP, State = #ch{tx_participants = Participants}) -> - State#ch{tx_participants = sets:union(Participants, - sets:from_list(MoreP))}. - -ack(TxnKey, UAQ) -> - fold_per_queue( - fun (QPid, MsgIds, L) -> - ok = rabbit_amqqueue:ack(QPid, TxnKey, MsgIds, self()), - [{QPid, length(MsgIds)} | L] - end, [], UAQ). - -make_tx_id() -> rabbit_guid:guid(). - -new_tx(State) -> - State#ch{transaction_id = make_tx_id(), - tx_participants = sets:new(), - uncommitted_ack_q = queue:new()}. - -internal_commit(State = #ch{transaction_id = TxnKey, - tx_participants = Participants}) -> - case rabbit_amqqueue:commit_all(sets:to_list(Participants), - TxnKey, self()) of - ok -> ok = notify_limiter(State#ch.limiter_pid, - State#ch.uncommitted_ack_q), - new_tx(State); - {error, Errors} -> rabbit_misc:protocol_error( - internal_error, "commit failed: ~w", [Errors]) - end. +ack(UAQ) -> + fold_per_queue(fun (QPid, MsgIds, L) -> + ok = rabbit_amqqueue:ack(QPid, MsgIds, self()), + [{QPid, length(MsgIds)} | L] + end, [], UAQ). -internal_rollback(State = #ch{transaction_id = TxnKey, - tx_participants = Participants, - uncommitted_ack_q = UAQ, - unacked_message_q = UAMQ}) -> - ?LOGDEBUG("rollback ~p~n - ~p acks uncommitted, ~p messages unacked~n", - [self(), - queue:len(UAQ), - queue:len(UAMQ)]), - ok = rabbit_amqqueue:rollback_all(sets:to_list(Participants), - TxnKey, self()), - NewUAMQ = queue:join(UAQ, UAMQ), - new_tx(State#ch{unacked_message_q = NewUAMQ}). - -rollback_and_notify(State = #ch{state = closing}) -> +notify_queues(State = #ch{state = closing}) -> {ok, State}; -rollback_and_notify(State = #ch{transaction_id = none}) -> - {notify_queues(State), State#ch{state = closing}}; -rollback_and_notify(State) -> - State1 = internal_rollback(State), - {notify_queues(State1), State1#ch{state = closing}}. +notify_queues(State = #ch{consumer_mapping = Consumers}) -> + {rabbit_amqqueue:notify_down_all(consumer_queues(Consumers), self()), + State#ch{state = closing}}. fold_per_queue(F, Acc0, UAQ) -> D = rabbit_misc:queue_fold( @@ -1319,9 +1231,6 @@ start_limiter(State = #ch{unacked_message_q = UAMQ, start_limiter_fun = SLF}) -> ok = limit_queues(LPid, State), LPid. -notify_queues(#ch{consumer_mapping = Consumers}) -> - rabbit_amqqueue:notify_down_all(consumer_queues(Consumers), self()). - unlimit_queues(State) -> ok = limit_queues(undefined, State), undefined. @@ -1436,17 +1345,13 @@ i(connection, #ch{conn_pid = ConnPid}) -> ConnPid; i(number, #ch{channel = Channel}) -> Channel; i(user, #ch{user = User}) -> User#user.username; i(vhost, #ch{virtual_host = VHost}) -> VHost; -i(transactional, #ch{transaction_id = TxnKey}) -> TxnKey =/= none; i(confirm, #ch{confirm_enabled = CE}) -> CE; i(consumer_count, #ch{consumer_mapping = ConsumerMapping}) -> dict:size(ConsumerMapping); i(messages_unconfirmed, #ch{unconfirmed_mq = UMQ}) -> gb_trees:size(UMQ); -i(messages_unacknowledged, #ch{unacked_message_q = UAMQ, - uncommitted_ack_q = UAQ}) -> - queue:len(UAMQ) + queue:len(UAQ); -i(acks_uncommitted, #ch{uncommitted_ack_q = UAQ}) -> - queue:len(UAQ); +i(messages_unacknowledged, #ch{unacked_message_q = UAMQ}) -> + queue:len(UAMQ); i(prefetch_count, #ch{limiter_pid = LimiterPid}) -> rabbit_limiter:get_limit(LimiterPid); i(client_flow_blocked, #ch{limiter_pid = LimiterPid}) -> diff --git a/src/rabbit_control.erl b/src/rabbit_control.erl index 9eef384a..6eb1aaba 100644 --- a/src/rabbit_control.erl +++ b/src/rabbit_control.erl @@ -301,7 +301,7 @@ action(list_connections, Node, Args, _Opts, Inform) -> action(list_channels, Node, Args, _Opts, Inform) -> Inform("Listing channels", []), - ArgAtoms = default_if_empty(Args, [pid, user, transactional, consumer_count, + ArgAtoms = default_if_empty(Args, [pid, user, consumer_count, messages_unacknowledged]), display_info_list(rpc_call(Node, rabbit_channel, info_all, [ArgAtoms]), ArgAtoms); diff --git a/src/rabbit_error_logger.erl b/src/rabbit_error_logger.erl index 3fb0817a..93aad9e3 100644 --- a/src/rabbit_error_logger.erl +++ b/src/rabbit_error_logger.erl @@ -71,7 +71,7 @@ publish1(RoutingKey, Format, Data, LogExch) -> %% second resolution, not millisecond. Timestamp = rabbit_misc:now_ms() div 1000, {ok, _RoutingRes, _DeliveredQPids} = - rabbit_basic:publish(LogExch, RoutingKey, false, false, none, + rabbit_basic:publish(LogExch, RoutingKey, false, false, #'P_basic'{content_type = <<"text/plain">>, timestamp = Timestamp}, list_to_binary(io_lib:format(Format, Data))), diff --git a/src/rabbit_mirror_queue_coordinator.erl b/src/rabbit_mirror_queue_coordinator.erl index 2727c1d0..d1c3a2e5 100644 --- a/src/rabbit_mirror_queue_coordinator.erl +++ b/src/rabbit_mirror_queue_coordinator.erl @@ -254,45 +254,6 @@ %% sender_death message. The slave will then be able to tidy up its %% state as normal. %% -%% We don't support transactions on mirror queues. To do so is -%% challenging. The underlying bq is free to add the contents of the -%% txn to the queue proper at any point after the tx.commit comes in -%% but before the tx.commit-ok goes out. This means that it is not -%% safe for all mirrors to simply issue the bq:tx_commit at the same -%% time, as the addition of the txn's contents to the queue may -%% subsequently be inconsistently interwoven with other actions on the -%% bq. The solution to this is, in the master, wrap the PostCommitFun -%% and do the gm:broadcast in there: at that point, you're in the bq -%% (well, there's actually nothing to stop that function being invoked -%% by some other process, but let's pretend for now: you could always -%% use run_backing_queue to ensure you really are in the queue process -%% (the _async variant would be unsafe from an ordering pov)), the -%% gm:broadcast is safe because you don't have to worry about races -%% with other gm:broadcast calls (same process). Thus this signal -%% would indicate sufficiently to all the slaves that they must insert -%% the complete contents of the txn at precisely this point in the -%% stream of events. -%% -%% However, it's quite difficult for the slaves to make that happen: -%% they would be forced to issue the bq:tx_commit at that point, but -%% then stall processing any further instructions from gm until they -%% receive the notification from their bq that the tx_commit has fully -%% completed (i.e. they need to treat what is an async system as being -%% fully synchronous). This is not too bad (apart from the -%% vomit-inducing notion of it all): just need a queue of instructions -%% from the GM; but then it gets rather worse when you consider what -%% needs to happen if the master dies at this point and the slave in -%% the middle of this tx_commit needs to be promoted. -%% -%% Finally, we can't possibly hope to make transactions atomic across -%% mirror queues, and it's not even clear that that's desirable: if a -%% slave fails whilst there's an open transaction in progress then -%% when the channel comes to commit the txn, it will detect the -%% failure and destroy the channel. However, the txn will have -%% actually committed successfully in all the other mirrors (including -%% master). To do this bit properly would require 2PC and all the -%% baggage that goes with that. -%% %% Recovery of mirrored queues is straightforward: as nodes die, the %% remaining nodes record this, and eventually a situation is reached %% in which only one node is alive, which is the master. This is the diff --git a/src/rabbit_mirror_queue_slave.erl b/src/rabbit_mirror_queue_slave.erl index 55d61d41..66ff575f 100644 --- a/src/rabbit_mirror_queue_slave.erl +++ b/src/rabbit_mirror_queue_slave.erl @@ -488,7 +488,7 @@ promote_me(From, #state { q = Q, %% %% Everything that's in MA gets requeued. Consequently the new %% master should start with a fresh AM as there are no messages - %% pending acks (txns will have been rolled back). + %% pending acks. MSList = dict:to_list(MS), SS = dict:from_list( @@ -612,8 +612,7 @@ confirm_sender_death(Pid) -> maybe_enqueue_message( Delivery = #delivery { message = #basic_message { id = MsgId }, msg_seq_no = MsgSeqNo, - sender = ChPid, - txn = none }, + sender = ChPid }, EnqueueOnPromotion, State = #state { sender_queues = SQ, msg_id_status = MS }) -> State1 = ensure_monitoring(ChPid, State), @@ -655,10 +654,7 @@ maybe_enqueue_message( SQ1 = remove_from_pending_ch(MsgId, ChPid, SQ), State1 #state { msg_id_status = dict:erase(MsgId, MS), sender_queues = SQ1 } - end; -maybe_enqueue_message(_Delivery, _EnqueueOnPromotion, State) -> - %% We don't support txns in mirror queues. - State. + end. get_sender_queue(ChPid, SQ) -> case dict:find(ChPid, SQ) of diff --git a/src/rabbit_queue_index.erl b/src/rabbit_queue_index.erl index aaf3df78..bf89cdb2 100644 --- a/src/rabbit_queue_index.erl +++ b/src/rabbit_queue_index.erl @@ -76,11 +76,10 @@ %% the segment file combined with the journal, no writing needs to be %% done to the segment file either (in fact it is deleted if it exists %% at all). This is safe given that the set of acks is a subset of the -%% set of publishes. When it's necessary to sync messages because of -%% transactions, it's only necessary to fsync on the journal: when -%% entries are distributed from the journal to segment files, those -%% segments appended to are fsync'd prior to the journal being -%% truncated. +%% set of publishes. When it is necessary to sync messages, it is +%% sufficient to fsync on the journal: when entries are distributed +%% from the journal to segment files, those segments appended to are +%% fsync'd prior to the journal being truncated. %% %% This module is also responsible for scanning the queue index files %% and seeding the message store on start up. @@ -289,14 +288,13 @@ sync(State = #qistate { unsynced_msg_ids = MsgIds }) -> sync_if([] =/= MsgIds, State). sync(SeqIds, State) -> - %% The SeqIds here contains the SeqId of every publish and ack in - %% the transaction. Ideally we should go through these seqids and - %% only sync the journal if the pubs or acks appear in the + %% The SeqIds here contains the SeqId of every publish and ack to + %% be sync'ed. Ideally we should go through these seqids and only + %% sync the journal if the pubs or acks appear in the %% journal. However, this would be complex to do, and given that %% the variable queue publishes and acks to the qi, and then %% syncs, all in one operation, there is no possibility of the - %% seqids not being in the journal, provided the transaction isn't - %% emptied (handled by sync_if anyway). + %% seqids not being in the journal. sync_if([] =/= SeqIds, State). flush(State = #qistate { dirty_count = 0 }) -> State; diff --git a/src/rabbit_tests.erl b/src/rabbit_tests.erl index 6e44c7a0..bc1b00b2 100644 --- a/src/rabbit_tests.erl +++ b/src/rabbit_tests.erl @@ -705,7 +705,6 @@ test_topic_expect_match(X, List) -> Res = rabbit_exchange_type_topic:route( X, #delivery{mandatory = false, immediate = false, - txn = none, sender = self(), message = Message}), ExpectedRes = lists:map( @@ -1669,8 +1668,8 @@ test_backing_queue() -> passed = test_queue_index(), passed = test_queue_index_props(), passed = test_variable_queue(), - %% FIXME: replace the use of tx in these with confirms - %% passed = test_variable_queue_delete_msg_store_files_callback(), + passed = test_variable_queue_delete_msg_store_files_callback(), + %% FIXME: re-enable once fixed %% passed = test_queue_recover(), application:set_env(rabbit, queue_index_max_journal_entries, MaxJournal, infinity), @@ -2319,17 +2318,16 @@ test_variable_queue_all_the_bits_not_covered_elsewhere2(VQ0) -> test_queue_recover() -> Count = 2 * rabbit_queue_index:next_segment_boundary(0), - TxID = rabbit_guid:guid(), {new, #amqqueue { pid = QPid, name = QName } = Q} = rabbit_amqqueue:declare(test_queue(), true, false, [], none), [begin Msg = rabbit_basic:message(rabbit_misc:r(<<>>, exchange, <<>>), <<>>, #'P_basic'{delivery_mode = 2}, <<>>), - Delivery = #delivery{mandatory = false, immediate = false, txn = TxID, + Delivery = #delivery{mandatory = false, immediate = false, sender = self(), message = Msg}, true = rabbit_amqqueue:deliver(QPid, Delivery) end || _ <- lists:seq(1, Count)], - rabbit_amqqueue:commit_all([QPid], TxID, self()), + %% FIXME: wait for confirms of all publishes exit(QPid, kill), MRef = erlang:monitor(process, QPid), receive {'DOWN', MRef, process, QPid, _Info} -> ok @@ -2356,18 +2354,17 @@ test_variable_queue_delete_msg_store_files_callback() -> ok = restart_msg_store_empty(), {new, #amqqueue { pid = QPid, name = QName } = Q} = rabbit_amqqueue:declare(test_queue(), true, false, [], none), - TxID = rabbit_guid:guid(), Payload = <<0:8388608>>, %% 1MB Count = 30, [begin Msg = rabbit_basic:message( rabbit_misc:r(<<>>, exchange, <<>>), <<>>, #'P_basic'{delivery_mode = 2}, Payload), - Delivery = #delivery{mandatory = false, immediate = false, txn = TxID, + Delivery = #delivery{mandatory = false, immediate = false, sender = self(), message = Msg}, true = rabbit_amqqueue:deliver(QPid, Delivery) end || _ <- lists:seq(1, Count)], - rabbit_amqqueue:commit_all([QPid], TxID, self()), + %% FIXME: wait for confirms of all publishes rabbit_amqqueue:set_ram_duration_target(QPid, 0), CountMinusOne = Count - 1, diff --git a/src/rabbit_types.erl b/src/rabbit_types.erl index 03b2c9e8..2db960ac 100644 --- a/src/rabbit_types.erl +++ b/src/rabbit_types.erl @@ -20,7 +20,7 @@ -ifdef(use_specs). --export_type([txn/0, maybe/1, info/0, infos/0, info_key/0, info_keys/0, +-export_type([maybe/1, info/0, infos/0, info_key/0, info_keys/0, message/0, msg_id/0, basic_message/0, delivery/0, content/0, decoded_content/0, undecoded_content/0, unencoded_content/0, encoded_content/0, message_properties/0, @@ -73,16 +73,12 @@ -type(delivery() :: #delivery{mandatory :: boolean(), immediate :: boolean(), - txn :: maybe(txn()), sender :: pid(), message :: message()}). -type(message_properties() :: #message_properties{expiry :: pos_integer() | 'undefined', needs_confirming :: boolean()}). -%% this is really an abstract type, but dialyzer does not support them --type(txn() :: rabbit_guid:guid()). - -type(info_key() :: atom()). -type(info_keys() :: [info_key()]). |