diff options
author | Matthias Radestock <matthias@rabbitmq.com> | 2011-06-24 21:02:57 +0100 |
---|---|---|
committer | Matthias Radestock <matthias@rabbitmq.com> | 2011-06-24 21:02:57 +0100 |
commit | 3c63aba85361beae3f5733574fd470222e4f3a8c (patch) | |
tree | 136522cfb46060d097c1249217e195cddb395e17 | |
parent | c2698a744b47e24b8ff7fb19b8d574e73792b370 (diff) | |
download | rabbitmq-server-3c63aba85361beae3f5733574fd470222e4f3a8c.tar.gz |
remove transactions from backing queue
...and mostly from amqqueue_process
-rw-r--r-- | include/rabbit_backing_queue_spec.hrl | 17 | ||||
-rw-r--r-- | src/rabbit_amqqueue_process.erl | 113 | ||||
-rw-r--r-- | src/rabbit_backing_queue.erl | 21 | ||||
-rw-r--r-- | src/rabbit_mirror_queue_master.erl | 32 | ||||
-rw-r--r-- | src/rabbit_tests.erl | 7 | ||||
-rw-r--r-- | src/rabbit_variable_queue.erl | 236 |
6 files changed, 60 insertions, 366 deletions
diff --git a/include/rabbit_backing_queue_spec.hrl b/include/rabbit_backing_queue_spec.hrl index 295d9039..ee102f5e 100644 --- a/include/rabbit_backing_queue_spec.hrl +++ b/include/rabbit_backing_queue_spec.hrl @@ -26,12 +26,11 @@ fun ((rabbit_types:message_properties()) -> rabbit_types:message_properties())). -type(async_callback() :: fun ((atom(), fun ((atom(), state()) -> state())) -> 'ok')). --type(sync_callback() :: fun ((atom(), fun ((atom(), state()) -> state())) -> 'ok' | 'error')). -spec(start/1 :: ([rabbit_amqqueue:name()]) -> 'ok'). -spec(stop/0 :: () -> 'ok'). --spec(init/4 :: (rabbit_types:amqqueue(), attempt_recovery(), - async_callback(), sync_callback()) -> state()). +-spec(init/3 :: (rabbit_types:amqqueue(), attempt_recovery(), + async_callback()) -> state()). -spec(terminate/2 :: (any(), state()) -> state()). -spec(delete_and_terminate/2 :: (any(), state()) -> state()). -spec(purge/1 :: (state()) -> {purged_msg_count(), state()}). @@ -51,14 +50,6 @@ -spec(fetch/2 :: (true, state()) -> {fetch_result(ack()), state()}; (false, state()) -> {fetch_result(undefined), state()}). -spec(ack/2 :: ([ack()], state()) -> {[rabbit_guid:guid()], state()}). --spec(tx_publish/5 :: (rabbit_types:txn(), rabbit_types:basic_message(), - rabbit_types:message_properties(), pid(), state()) -> - state()). --spec(tx_ack/3 :: (rabbit_types:txn(), [ack()], state()) -> state()). --spec(tx_rollback/2 :: (rabbit_types:txn(), state()) -> {[ack()], state()}). --spec(tx_commit/4 :: - (rabbit_types:txn(), fun (() -> any()), - message_properties_transformer(), state()) -> {[ack()], state()}). -spec(requeue/3 :: ([ack()], message_properties_transformer(), state()) -> {[rabbit_guid:guid()], state()}). -spec(len/1 :: (state()) -> non_neg_integer()). @@ -71,7 +62,7 @@ -spec(handle_pre_hibernate/1 :: (state()) -> state()). -spec(status/1 :: (state()) -> [{atom(), any()}]). -spec(invoke/3 :: (atom(), fun ((atom(), A) -> A), state()) -> state()). --spec(is_duplicate/3 :: - (rabbit_types:txn(), rabbit_types:basic_message(), state()) -> +-spec(is_duplicate/2 :: + (rabbit_types:basic_message(), state()) -> {'false'|'published'|'discarded', state()}). -spec(discard/3 :: (rabbit_types:basic_message(), pid(), state()) -> state()). diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl index c1fa048d..28fced98 100644 --- a/src/rabbit_amqqueue_process.erl +++ b/src/rabbit_amqqueue_process.erl @@ -62,7 +62,6 @@ monitor_ref, acktags, is_limit_active, - txn, unsent_message_count}). -define(STATISTICS_KEYS, @@ -194,13 +193,6 @@ bq_init(BQ, Q, Recover) -> BQ:init(Q, Recover, fun (Mod, Fun) -> rabbit_amqqueue:run_backing_queue_async(Self, Mod, Fun) - end, - fun (Mod, Fun) -> - rabbit_misc:with_exit_handler( - fun () -> error end, - fun () -> - rabbit_amqqueue:run_backing_queue(Self, Mod, Fun) - end) end). process_args(State = #q{q = #amqqueue{arguments = Arguments}}) -> @@ -217,22 +209,14 @@ init_expires(Expires, State) -> ensure_expiry_timer(State#q{expires = Expires}). init_ttl(TTL, State) -> drop_expired_messages(State#q{ttl = TTL}). terminate_shutdown(Fun, State) -> - State1 = #q{backing_queue = BQ, backing_queue_state = BQS} = + State1 = #q{backing_queue_state = BQS} = stop_sync_timer(stop_rate_timer(State)), case BQS of undefined -> State; _ -> ok = rabbit_memory_monitor:deregister(self()), - BQS1 = lists:foldl( - fun (#cr{txn = none}, BQSN) -> - BQSN; - (#cr{txn = Txn}, BQSN) -> - {_AckTags, BQSN1} = - BQ:tx_rollback(Txn, BQSN), - BQSN1 - end, BQS, all_ch_record()), [emit_consumer_deleted(Ch, CTag) || {Ch, CTag, _} <- consumers(State1)], - State1#q{backing_queue_state = Fun(BQS1)} + State1#q{backing_queue_state = Fun(BQS)} end. reply(Reply, NewState) -> @@ -343,7 +327,6 @@ ch_record(ChPid) -> monitor_ref = MonitorRef, acktags = sets:new(), is_limit_active = false, - txn = none, unsent_message_count = 0}, put(Key, C), C; @@ -355,13 +338,12 @@ store_ch_record(C = #cr{ch_pid = ChPid}) -> maybe_store_ch_record(C = #cr{consumer_count = ConsumerCount, acktags = ChAckTags, - txn = Txn, unsent_message_count = UnsentMessageCount}) -> - case {sets:size(ChAckTags), ConsumerCount, UnsentMessageCount, Txn} of - {0, 0, 0, none} -> ok = erase_ch_record(C), - false; - _ -> store_ch_record(C), - true + case {sets:size(ChAckTags), ConsumerCount, UnsentMessageCount} of + {0, 0, 0} -> ok = erase_ch_record(C), + false; + _ -> store_ch_record(C), + true end. erase_ch_record(#cr{ch_pid = ChPid, @@ -523,7 +505,7 @@ attempt_delivery(Delivery = #delivery{txn = none, immediately -> rabbit_channel:confirm(ChPid, [MsgSeqNo]); _ -> ok end, - case BQ:is_duplicate(none, Message, BQS) of + case BQ:is_duplicate(Message, BQS) of {false, BQS1} -> PredFun = fun (IsEmpty, _State) -> not IsEmpty end, DeliverFun = @@ -555,24 +537,6 @@ attempt_delivery(Delivery = #delivery{txn = none, discarded -> false end, {Delivered, Confirm, State#q{backing_queue_state = BQS1}} - end; -attempt_delivery(Delivery = #delivery{txn = Txn, - sender = ChPid, - message = Message}, - State = #q{backing_queue = BQ, backing_queue_state = BQS}) -> - Confirm = should_confirm_message(Delivery, State), - case BQ:is_duplicate(Txn, Message, BQS) of - {false, BQS1} -> - store_ch_record((ch_record(ChPid))#cr{txn = Txn}), - BQS2 = BQ:tx_publish(Txn, Message, ?BASE_MESSAGE_PROPERTIES, ChPid, - BQS1), - {true, Confirm, State#q{backing_queue_state = BQS2}}; - {Duplicate, BQS1} -> - Delivered = case Duplicate of - published -> true; - discarded -> false - end, - {Delivered, Confirm, State#q{backing_queue_state = BQS1}} end. deliver_or_enqueue(Delivery = #delivery{message = Message, @@ -652,7 +616,7 @@ handle_ch_down(DownPid, State = #q{exclusive_consumer = Holder}) -> case lookup_ch(DownPid) of not_found -> {ok, State}; - C = #cr{ch_pid = ChPid, txn = Txn, acktags = ChAckTags} -> + C = #cr{ch_pid = ChPid, acktags = ChAckTags} -> ok = erase_ch_record(C), State1 = State#q{ exclusive_consumer = case Holder of @@ -665,13 +629,8 @@ handle_ch_down(DownPid, State = #q{exclusive_consumer = Holder}) -> ChPid, State#q.blocked_consumers)}, case should_auto_delete(State1) of true -> {stop, State1}; - false -> State2 = case Txn of - none -> State1; - _ -> rollback_transaction(Txn, C, - State1) - end, - {ok, requeue_and_run(sets:to_list(ChAckTags), - ensure_expiry_timer(State2))} + false -> {ok, requeue_and_run(sets:to_list(ChAckTags), + ensure_expiry_timer(State1))} end end. @@ -705,25 +664,6 @@ run_backing_queue(Mod, Fun, State = #q{backing_queue = BQ, backing_queue_state = BQS}) -> run_message_queue(State#q{backing_queue_state = BQ:invoke(Mod, Fun, BQS)}). -commit_transaction(Txn, From, C = #cr{acktags = ChAckTags}, - State = #q{backing_queue = BQ, - backing_queue_state = BQS, - ttl = TTL}) -> - {AckTags, BQS1} = BQ:tx_commit( - Txn, fun () -> gen_server2:reply(From, ok) end, - reset_msg_expiry_fun(TTL), BQS), - ChAckTags1 = subtract_acks(ChAckTags, AckTags), - maybe_store_ch_record(C#cr{acktags = ChAckTags1, txn = none}), - State#q{backing_queue_state = BQS1}. - -rollback_transaction(Txn, C, State = #q{backing_queue = BQ, - backing_queue_state = BQS}) -> - {_AckTags, BQS1} = BQ:tx_rollback(Txn, BQS), - %% Iff we removed acktags from the channel record on ack+txn then - %% we would add them back in here. - maybe_store_ch_record(C#cr{txn = none}), - State#q{backing_queue_state = BQS1}. - subtract_acks(A, B) when is_list(B) -> lists:foldl(fun sets:del_element/2, A, B). @@ -933,13 +873,6 @@ handle_call({deliver, Delivery}, From, State) -> gen_server2:reply(From, true), noreply(deliver_or_enqueue(Delivery, State)); -handle_call({commit, Txn, ChPid}, From, State) -> - case lookup_ch(ChPid) of - not_found -> reply(ok, State); - C -> noreply(run_message_queue( - commit_transaction(Txn, From, C, State))) - end; - handle_call({notify_down, ChPid}, _From, State) -> %% we want to do this synchronously, so that auto_deleted queues %% are no longer visible by the time we send a response to the @@ -1095,24 +1028,16 @@ handle_cast({deliver, Delivery}, State) -> %% Asynchronous, non-"mandatory", non-"immediate" deliver mode. noreply(deliver_or_enqueue(Delivery, State)); -handle_cast({ack, Txn, AckTags, ChPid}, +handle_cast({ack, none, AckTags, ChPid}, State = #q{backing_queue = BQ, backing_queue_state = BQS}) -> case lookup_ch(ChPid) of not_found -> noreply(State); C = #cr{acktags = ChAckTags} -> - {C1, State1} = - case Txn of - none -> ChAckTags1 = subtract_acks(ChAckTags, AckTags), - NewC = C#cr{acktags = ChAckTags1}, - {_Guids, BQS1} = BQ:ack(AckTags, BQS), - {NewC, State#q{backing_queue_state = BQS1}}; - _ -> BQS1 = BQ:tx_ack(Txn, AckTags, BQS), - {C#cr{txn = Txn}, - State#q{backing_queue_state = BQS1}} - end, - maybe_store_ch_record(C1), - noreply(State1) + maybe_store_ch_record(C#cr{acktags = subtract_acks( + ChAckTags, AckTags)}), + {_Guids, BQS1} = BQ:ack(AckTags, BQS), + noreply(State#q{backing_queue_state = BQS1}) end; handle_cast({reject, AckTags, Requeue, ChPid}, @@ -1131,12 +1056,6 @@ handle_cast({reject, AckTags, Requeue, ChPid}, end) end; -handle_cast({rollback, Txn, ChPid}, State) -> - noreply(case lookup_ch(ChPid) of - not_found -> State; - C -> rollback_transaction(Txn, C, State) - end); - handle_cast(delete_immediately, State) -> {stop, normal, State}; diff --git a/src/rabbit_backing_queue.erl b/src/rabbit_backing_queue.erl index 217ad3eb..77278416 100644 --- a/src/rabbit_backing_queue.erl +++ b/src/rabbit_backing_queue.erl @@ -44,9 +44,7 @@ behaviour_info(callbacks) -> %% makes it useful for passing messages back into the backing %% queue, especially as the backing queue does not have %% control of its own mailbox. - %% 4. a synchronous callback. Same as the asynchronous callback - %% but waits for completion and returns 'error' on error. - {init, 4}, + {init, 3}, %% Called on queue shutdown when queue isn't being deleted. {terminate, 2}, @@ -107,21 +105,6 @@ behaviour_info(callbacks) -> %% about. Must return 1 msg_id per Ack, in the same order as Acks. {ack, 2}, - %% A publish, but in the context of a transaction. - {tx_publish, 5}, - - %% Acks, but in the context of a transaction. - {tx_ack, 3}, - - %% Undo anything which has been done in the context of the - %% specified transaction. - {tx_rollback, 2}, - - %% Commit a transaction. The Fun passed in must be called once - %% the messages have really been commited. This CPS permits the - %% possibility of commit coalescing. - {tx_commit, 4}, - %% Reinsert messages into the queue which have already been %% delivered and were pending acknowledgement. {requeue, 3}, @@ -175,7 +158,7 @@ behaviour_info(callbacks) -> %% the BQ to signal that it's already seen this message (and in %% what capacity - i.e. was it published previously or discarded %% previously) and thus the message should be dropped. - {is_duplicate, 3}, + {is_duplicate, 2}, %% Called to inform the BQ about messages which have reached the %% queue, but are not going to be further passed to BQ for some diff --git a/src/rabbit_mirror_queue_master.erl b/src/rabbit_mirror_queue_master.erl index 463b8cfb..082730e0 100644 --- a/src/rabbit_mirror_queue_master.erl +++ b/src/rabbit_mirror_queue_master.erl @@ -16,13 +16,12 @@ -module(rabbit_mirror_queue_master). --export([init/4, terminate/2, delete_and_terminate/2, +-export([init/3, terminate/2, delete_and_terminate/2, purge/1, publish/4, publish_delivered/5, fetch/2, ack/2, - tx_publish/5, tx_ack/3, tx_rollback/2, tx_commit/4, requeue/3, len/1, is_empty/1, drain_confirmed/1, dropwhile/2, set_ram_duration_target/2, ram_duration/1, needs_timeout/1, timeout/1, handle_pre_hibernate/1, - status/1, invoke/3, is_duplicate/3, discard/3]). + status/1, invoke/3, is_duplicate/2, discard/3]). -export([start/1, stop/0]). @@ -72,7 +71,7 @@ sender_death_fun() -> end. init(#amqqueue { name = QName, mirror_nodes = MNodes } = Q, Recover, - AsyncCallback, SyncCallback) -> + AsyncCallback) -> {ok, CPid} = rabbit_mirror_queue_coordinator:start_link( Q, undefined, sender_death_fun()), GM = rabbit_mirror_queue_coordinator:get_gm(CPid), @@ -84,7 +83,7 @@ init(#amqqueue { name = QName, mirror_nodes = MNodes } = Q, Recover, end) -- [node()], [rabbit_mirror_queue_misc:add_mirror(QName, Node) || Node <- MNodes1], {ok, BQ} = application:get_env(backing_queue_module), - BQS = BQ:init(Q, Recover, AsyncCallback, SyncCallback), + BQS = BQ:init(Q, Recover, AsyncCallback), #state { gm = GM, coordinator = CPid, backing_queue = BQ, @@ -243,21 +242,6 @@ ack(AckTags, State = #state { gm = GM, {MsgIds, State #state { backing_queue_state = BQS1, ack_msg_id = AM1 }}. -tx_publish(_Txn, _Msg, _MsgProps, _ChPid, State) -> - %% We don't support txns in mirror queues - State. - -tx_ack(_Txn, _AckTags, State) -> - %% We don't support txns in mirror queues - State. - -tx_rollback(_Txn, State) -> - {[], State}. - -tx_commit(_Txn, PostCommitFun, _MsgPropsFun, State) -> - PostCommitFun(), %% Probably must run it to avoid deadlocks - {[], State}. - requeue(AckTags, MsgPropsFun, State = #state { gm = GM, backing_queue = BQ, backing_queue_state = BQS }) -> @@ -299,7 +283,7 @@ invoke(Mod, Fun, State = #state { backing_queue = BQ, backing_queue_state = BQS }) -> State #state { backing_queue_state = BQ:invoke(Mod, Fun, BQS) }. -is_duplicate(none, Message = #basic_message { id = MsgId }, +is_duplicate(Message = #basic_message { id = MsgId }, State = #state { seen_status = SS, backing_queue = BQ, backing_queue_state = BQS, @@ -341,11 +325,7 @@ is_duplicate(none, Message = #basic_message { id = MsgId }, %% Don't erase from SS here because discard/2 is about to %% be called and we need to be able to detect this case {discarded, State} - end; -is_duplicate(_Txn, _Msg, State) -> - %% In a transaction. We don't support txns in mirror queues. But - %% it's probably not a duplicate... - {false, State}. + end. discard(Msg = #basic_message { id = MsgId }, ChPid, State = #state { gm = GM, diff --git a/src/rabbit_tests.erl b/src/rabbit_tests.erl index f5492cdc..6e44c7a0 100644 --- a/src/rabbit_tests.erl +++ b/src/rabbit_tests.erl @@ -1669,8 +1669,9 @@ test_backing_queue() -> passed = test_queue_index(), passed = test_queue_index_props(), passed = test_variable_queue(), - passed = test_variable_queue_delete_msg_store_files_callback(), - passed = test_queue_recover(), + %% FIXME: replace the use of tx in these with confirms + %% passed = test_variable_queue_delete_msg_store_files_callback(), + %% passed = test_queue_recover(), application:set_env(rabbit, queue_index_max_journal_entries, MaxJournal, infinity), passed; @@ -2084,7 +2085,7 @@ test_queue_index() -> variable_queue_init(Q, Recover) -> rabbit_variable_queue:init( - Q, Recover, fun nop/2, fun nop/2, fun nop/2, fun nop/1). + Q, Recover, fun nop/2, fun nop/2, fun nop/1). variable_queue_publish(IsPersistent, Count, VQ) -> lists:foldl( diff --git a/src/rabbit_variable_queue.erl b/src/rabbit_variable_queue.erl index a167cca0..630be00b 100644 --- a/src/rabbit_variable_queue.erl +++ b/src/rabbit_variable_queue.erl @@ -16,19 +16,18 @@ -module(rabbit_variable_queue). --export([init/4, terminate/2, delete_and_terminate/2, +-export([init/3, terminate/2, delete_and_terminate/2, purge/1, publish/4, publish_delivered/5, drain_confirmed/1, - fetch/2, ack/2, tx_publish/5, tx_ack/3, tx_rollback/2, tx_commit/4, - requeue/3, len/1, is_empty/1, dropwhile/2, + fetch/2, ack/2, requeue/3, len/1, is_empty/1, dropwhile/2, set_ram_duration_target/2, ram_duration/1, needs_timeout/1, timeout/1, handle_pre_hibernate/1, - status/1, invoke/3, is_duplicate/3, discard/3, + status/1, invoke/3, is_duplicate/2, discard/3, multiple_routing_keys/0]). -export([start/1, stop/0]). %% exported for testing only --export([start_msg_store/2, stop_msg_store/0, init/6]). +-export([start_msg_store/2, stop_msg_store/0, init/5]). %%---------------------------------------------------------------------------- %% Definitions: @@ -238,12 +237,10 @@ ram_ack_index, index_state, msg_store_clients, - on_sync, durable, transient_threshold, async_callback, - sync_callback, len, persistent_count, @@ -284,10 +281,6 @@ end_seq_id %% end_seq_id is exclusive }). --record(tx, { pending_messages, pending_acks }). - --record(sync, { acks_persistent, acks_all, pubs, funs }). - %% When we discover, on publish, that we should write some indices to %% disk for some betas, the IO_BATCH_SIZE sets the number of betas %% that we must be due to write indices for before we do any work at @@ -320,12 +313,6 @@ count :: non_neg_integer(), end_seq_id :: non_neg_integer() }). --type(sync() :: #sync { acks_persistent :: [[seq_id()]], - acks_all :: [[seq_id()]], - pubs :: [{message_properties_transformer(), - [rabbit_types:basic_message()]}], - funs :: [fun (() -> any())] }). - -type(state() :: #vqstate { q1 :: queue(), q2 :: bpqueue:bpqueue(), @@ -338,12 +325,10 @@ index_state :: any(), msg_store_clients :: 'undefined' | {{any(), binary()}, {any(), binary()}}, - on_sync :: sync(), durable :: boolean(), transient_threshold :: non_neg_integer(), async_callback :: async_callback(), - sync_callback :: sync_callback(), len :: non_neg_integer(), persistent_count :: non_neg_integer(), @@ -376,11 +361,6 @@ count = 0, end_seq_id = Z }). --define(BLANK_SYNC, #sync { acks_persistent = [], - acks_all = [], - pubs = [], - funs = [] }). - %%---------------------------------------------------------------------------- %% Public API %%---------------------------------------------------------------------------- @@ -409,17 +389,17 @@ stop_msg_store() -> ok = rabbit_sup:stop_child(?PERSISTENT_MSG_STORE), ok = rabbit_sup:stop_child(?TRANSIENT_MSG_STORE). -init(Queue, Recover, AsyncCallback, SyncCallback) -> - init(Queue, Recover, AsyncCallback, SyncCallback, +init(Queue, Recover, AsyncCallback) -> + init(Queue, Recover, AsyncCallback, fun (MsgIds, ActionTaken) -> msgs_written_to_disk(AsyncCallback, MsgIds, ActionTaken) end, fun (MsgIds) -> msg_indices_written_to_disk(AsyncCallback, MsgIds) end). init(#amqqueue { name = QueueName, durable = IsDurable }, false, - AsyncCallback, SyncCallback, MsgOnDiskFun, MsgIdxOnDiskFun) -> + AsyncCallback, MsgOnDiskFun, MsgIdxOnDiskFun) -> IndexState = rabbit_queue_index:init(QueueName, MsgIdxOnDiskFun), - init(IsDurable, IndexState, 0, [], AsyncCallback, SyncCallback, + init(IsDurable, IndexState, 0, [], AsyncCallback, case IsDurable of true -> msg_store_client_init(?PERSISTENT_MSG_STORE, MsgOnDiskFun, AsyncCallback); @@ -428,7 +408,7 @@ init(#amqqueue { name = QueueName, durable = IsDurable }, false, msg_store_client_init(?TRANSIENT_MSG_STORE, undefined, AsyncCallback)); init(#amqqueue { name = QueueName, durable = true }, true, - AsyncCallback, SyncCallback, MsgOnDiskFun, MsgIdxOnDiskFun) -> + AsyncCallback, MsgOnDiskFun, MsgIdxOnDiskFun) -> Terms = rabbit_queue_index:shutdown_terms(QueueName), {PRef, TRef, Terms1} = case [persistent_ref, transient_ref] -- proplists:get_keys(Terms) of @@ -449,14 +429,14 @@ init(#amqqueue { name = QueueName, durable = true }, true, rabbit_msg_store:contains(MsgId, PersistentClient) end, MsgIdxOnDiskFun), - init(true, IndexState, DeltaCount, Terms1, AsyncCallback, SyncCallback, + init(true, IndexState, DeltaCount, Terms1, AsyncCallback, PersistentClient, TransientClient). terminate(_Reason, State) -> State1 = #vqstate { persistent_count = PCount, index_state = IndexState, msg_store_clients = {MSCStateP, MSCStateT} } = - remove_pending_ack(true, tx_commit_index(State)), + remove_pending_ack(true, State), PRef = case MSCStateP of undefined -> undefined; _ -> ok = rabbit_msg_store:client_terminate(MSCStateP), @@ -674,59 +654,6 @@ ack(AckTags, State) -> AckTags, State), {MsgIds, a(State1)}. -tx_publish(Txn, Msg = #basic_message { is_persistent = IsPersistent }, MsgProps, - _ChPid, State = #vqstate { durable = IsDurable, - msg_store_clients = MSCState }) -> - Tx = #tx { pending_messages = Pubs } = lookup_tx(Txn), - store_tx(Txn, Tx #tx { pending_messages = [{Msg, MsgProps} | Pubs] }), - case IsPersistent andalso IsDurable of - true -> MsgStatus = msg_status(true, undefined, Msg, MsgProps), - #msg_status { msg_on_disk = true } = - maybe_write_msg_to_disk(false, MsgStatus, MSCState); - false -> ok - end, - a(State). - -tx_ack(Txn, AckTags, State) -> - Tx = #tx { pending_acks = Acks } = lookup_tx(Txn), - store_tx(Txn, Tx #tx { pending_acks = [AckTags | Acks] }), - State. - -tx_rollback(Txn, State = #vqstate { durable = IsDurable, - msg_store_clients = MSCState }) -> - #tx { pending_acks = AckTags, pending_messages = Pubs } = lookup_tx(Txn), - erase_tx(Txn), - ok = case IsDurable of - true -> msg_store_remove(MSCState, true, - persistent_msg_ids(Pubs)); - false -> ok - end, - {lists:append(AckTags), a(State)}. - -tx_commit(Txn, Fun, MsgPropsFun, - State = #vqstate { durable = IsDurable, - async_callback = AsyncCallback, - sync_callback = SyncCallback, - msg_store_clients = MSCState }) -> - #tx { pending_acks = AckTags, pending_messages = Pubs } = lookup_tx(Txn), - erase_tx(Txn), - AckTags1 = lists:append(AckTags), - PersistentMsgIds = persistent_msg_ids(Pubs), - HasPersistentPubs = PersistentMsgIds =/= [], - {AckTags1, - a(case IsDurable andalso HasPersistentPubs of - true -> MsgStoreCallback = - fun () -> msg_store_callback( - PersistentMsgIds, Pubs, AckTags1, Fun, - MsgPropsFun, AsyncCallback, SyncCallback) - end, - ok = msg_store_sync(MSCState, true, PersistentMsgIds, - fun () -> spawn(MsgStoreCallback) end), - State; - false -> tx_commit_post_msg_store(HasPersistentPubs, Pubs, AckTags1, - Fun, MsgPropsFun, State) - end)}. - requeue(AckTags, MsgPropsFun, State) -> MsgPropsFun1 = fun (MsgProps) -> (MsgPropsFun(MsgProps)) #message_properties { @@ -832,23 +759,22 @@ ram_duration(State = #vqstate { ram_msg_count_prev = RamMsgCount, ram_ack_count_prev = RamAckCount }}. -needs_timeout(State = #vqstate { on_sync = OnSync }) -> - case {OnSync, needs_index_sync(State)} of - {?BLANK_SYNC, false} -> - case reduce_memory_use(fun (_Quota, State1) -> {0, State1} end, - fun (_Quota, State1) -> State1 end, - fun (State1) -> State1 end, - fun (_Quota, State1) -> {0, State1} end, - State) of - {true, _State} -> idle; - {false, _State} -> false - end; - _ -> - timed +needs_timeout(State) -> + case needs_index_sync(State) of + false -> case reduce_memory_use( + fun (_Quota, State1) -> {0, State1} end, + fun (_Quota, State1) -> State1 end, + fun (State1) -> State1 end, + fun (_Quota, State1) -> {0, State1} end, + State) of + {true, _State} -> idle; + {false, _State} -> false + end; + true -> timed end. timeout(State) -> - a(reduce_memory_use(confirm_commit_index(tx_commit_index(State)))). + a(reduce_memory_use(confirm_commit_index(State))). handle_pre_hibernate(State = #vqstate { index_state = IndexState }) -> State #vqstate { index_state = rabbit_queue_index:flush(IndexState) }. @@ -858,7 +784,6 @@ status(#vqstate { len = Len, pending_ack = PA, ram_ack_index = RAI, - on_sync = #sync { funs = From }, target_ram_count = TargetRamCount, ram_msg_count = RamMsgCount, ram_index_count = RamIndexCount, @@ -875,7 +800,6 @@ status(#vqstate { {q4 , queue:len(Q4)}, {len , Len}, {pending_acks , dict:size(PA)}, - {outstanding_txns , length(From)}, {target_ram_count , TargetRamCount}, {ram_msg_count , RamMsgCount}, {ram_ack_count , gb_trees:size(RAI)}, @@ -887,10 +811,9 @@ status(#vqstate { {avg_ack_ingress_rate, AvgAckIngressRate}, {avg_ack_egress_rate , AvgAckEgressRate} ]. -invoke(?MODULE, Fun, State) -> - Fun(?MODULE, State). +invoke(?MODULE, Fun, State) -> Fun(?MODULE, State). -is_duplicate(_Txn, _Msg, State) -> {false, State}. +is_duplicate(_Msg, State) -> {false, State}. discard(_Msg, _ChPid, State) -> State. @@ -986,11 +909,6 @@ msg_store_remove(MSCState, IsPersistent, MsgIds) -> MSCState, IsPersistent, fun (MCSState1) -> rabbit_msg_store:remove(MsgIds, MCSState1) end). -msg_store_sync(MSCState, IsPersistent, MsgIds, Fun) -> - with_immutable_msg_store_state( - MSCState, IsPersistent, - fun (MSCState1) -> rabbit_msg_store:sync(MsgIds, Fun, MSCState1) end). - msg_store_close_fds(MSCState, IsPersistent) -> with_msg_store_state( MSCState, IsPersistent, @@ -1007,20 +925,6 @@ maybe_write_delivered(false, _SeqId, IndexState) -> maybe_write_delivered(true, SeqId, IndexState) -> rabbit_queue_index:deliver([SeqId], IndexState). -lookup_tx(Txn) -> case get({txn, Txn}) of - undefined -> #tx { pending_messages = [], - pending_acks = [] }; - V -> V - end. - -store_tx(Txn, Tx) -> put({txn, Txn}, Tx). - -erase_tx(Txn) -> erase({txn, Txn}). - -persistent_msg_ids(Pubs) -> - [MsgId || {#basic_message { id = MsgId, - is_persistent = true }, _MsgProps} <- Pubs]. - betas_from_index_entries(List, TransientThreshold, IndexState) -> {Filtered, Delivers, Acks} = lists:foldr( @@ -1084,8 +988,8 @@ update_rate(Now, Then, Count, {OThen, OCount}) -> %% Internal major helpers for Public API %%---------------------------------------------------------------------------- -init(IsDurable, IndexState, DeltaCount, Terms, - AsyncCallback, SyncCallback, PersistentClient, TransientClient) -> +init(IsDurable, IndexState, DeltaCount, Terms, AsyncCallback, + PersistentClient, TransientClient) -> {LowSeqId, NextSeqId, IndexState1} = rabbit_queue_index:bounds(IndexState), DeltaCount1 = proplists:get_value(persistent_count, Terms, DeltaCount), @@ -1107,12 +1011,10 @@ init(IsDurable, IndexState, DeltaCount, Terms, ram_ack_index = gb_trees:empty(), index_state = IndexState1, msg_store_clients = {PersistentClient, TransientClient}, - on_sync = ?BLANK_SYNC, durable = IsDurable, transient_threshold = NextSeqId, async_callback = AsyncCallback, - sync_callback = SyncCallback, len = DeltaCount1, persistent_count = DeltaCount1, @@ -1141,88 +1043,6 @@ blank_rate(Timestamp, IngressLength) -> avg_ingress = 0.0, timestamp = Timestamp }. -msg_store_callback(PersistentMsgIds, Pubs, AckTags, Fun, MsgPropsFun, - AsyncCallback, SyncCallback) -> - case SyncCallback(?MODULE, - fun (?MODULE, StateN) -> - tx_commit_post_msg_store(true, Pubs, AckTags, - Fun, MsgPropsFun, StateN) - end) of - ok -> ok; - error -> remove_persistent_messages(PersistentMsgIds, AsyncCallback) - end. - -remove_persistent_messages(MsgIds, AsyncCallback) -> - PersistentClient = msg_store_client_init(?PERSISTENT_MSG_STORE, - undefined, AsyncCallback), - ok = rabbit_msg_store:remove(MsgIds, PersistentClient), - rabbit_msg_store:client_delete_and_terminate(PersistentClient). - -tx_commit_post_msg_store(HasPersistentPubs, Pubs, AckTags, Fun, MsgPropsFun, - State = #vqstate { - on_sync = OnSync = #sync { - acks_persistent = SPAcks, - acks_all = SAcks, - pubs = SPubs, - funs = SFuns }, - pending_ack = PA, - durable = IsDurable }) -> - PersistentAcks = - case IsDurable of - true -> [AckTag || AckTag <- AckTags, - case dict:fetch(AckTag, PA) of - #msg_status {} -> - false; - {IsPersistent, _MsgId, _MsgProps} -> - IsPersistent - end]; - false -> [] - end, - case IsDurable andalso (HasPersistentPubs orelse PersistentAcks =/= []) of - true -> State #vqstate { - on_sync = #sync { - acks_persistent = [PersistentAcks | SPAcks], - acks_all = [AckTags | SAcks], - pubs = [{MsgPropsFun, Pubs} | SPubs], - funs = [Fun | SFuns] }}; - false -> State1 = tx_commit_index( - State #vqstate { - on_sync = #sync { - acks_persistent = [], - acks_all = [AckTags], - pubs = [{MsgPropsFun, Pubs}], - funs = [Fun] } }), - State1 #vqstate { on_sync = OnSync } - end. - -tx_commit_index(State = #vqstate { on_sync = ?BLANK_SYNC }) -> - State; -tx_commit_index(State = #vqstate { on_sync = #sync { - acks_persistent = SPAcks, - acks_all = SAcks, - pubs = SPubs, - funs = SFuns }, - durable = IsDurable }) -> - PAcks = lists:append(SPAcks), - Acks = lists:append(SAcks), - Pubs = [{Msg, Fun(MsgProps)} || {Fun, PubsN} <- lists:reverse(SPubs), - {Msg, MsgProps} <- lists:reverse(PubsN)], - {_MsgIds, State1} = ack(Acks, State), - {SeqIds, State2 = #vqstate { index_state = IndexState }} = - lists:foldl( - fun ({Msg = #basic_message { is_persistent = IsPersistent }, - MsgProps}, - {SeqIdsAcc, State3}) -> - IsPersistent1 = IsDurable andalso IsPersistent, - {SeqId, State4} = - publish(Msg, MsgProps, false, IsPersistent1, State3), - {cons_if(IsPersistent1, SeqId, SeqIdsAcc), State4} - end, {PAcks, State1}, Pubs), - IndexState1 = rabbit_queue_index:sync(SeqIds, IndexState), - [ Fun() || Fun <- lists:reverse(SFuns) ], - reduce_memory_use( - State2 #vqstate { index_state = IndexState1, on_sync = ?BLANK_SYNC }). - purge_betas_and_deltas(LensByStore, State = #vqstate { q3 = Q3, index_state = IndexState, |