diff options
| -rw-r--r-- | include/rabbit.hrl | 22 | ||||
| -rw-r--r-- | include/rabbit_queue.hrl | 58 | ||||
| -rw-r--r-- | src/file_handle_cache.erl | 455 | ||||
| -rw-r--r-- | src/rabbit.erl | 23 | ||||
| -rw-r--r-- | src/rabbit_amqqueue.erl | 143 | ||||
| -rw-r--r-- | src/rabbit_amqqueue_process.erl | 699 | ||||
| -rw-r--r-- | src/rabbit_amqqueue_sup.erl | 4 | ||||
| -rw-r--r-- | src/rabbit_basic.erl | 17 | ||||
| -rw-r--r-- | src/rabbit_binary_generator.erl | 19 | ||||
| -rw-r--r-- | src/rabbit_channel.erl | 10 | ||||
| -rw-r--r-- | src/rabbit_control.erl | 5 | ||||
| -rw-r--r-- | src/rabbit_guid.erl | 7 | ||||
| -rw-r--r-- | src/rabbit_memory_manager.erl | 404 | ||||
| -rw-r--r-- | src/rabbit_mnesia.erl | 11 | ||||
| -rw-r--r-- | src/rabbit_msg_file.erl | 141 | ||||
| -rw-r--r-- | src/rabbit_msg_store.erl | 1080 | ||||
| -rw-r--r-- | src/rabbit_persister.erl | 523 | ||||
| -rw-r--r-- | src/rabbit_queue_index.erl | 862 | ||||
| -rw-r--r-- | src/rabbit_queue_prefetcher.erl | 295 | ||||
| -rw-r--r-- | src/rabbit_tests.erl | 698 | ||||
| -rw-r--r-- | src/rabbit_variable_queue.erl | 1019 |
21 files changed, 5555 insertions, 940 deletions
diff --git a/include/rabbit.hrl b/include/rabbit.hrl index 5703d0d619..330eef80d7 100644 --- a/include/rabbit.hrl +++ b/include/rabbit.hrl @@ -62,7 +62,10 @@ -record(listener, {node, protocol, host, port}). --record(basic_message, {exchange_name, routing_key, content, persistent_key}). +-record(basic_message, {exchange_name, routing_key, content, + guid, is_persistent}). + +-record(dq_msg_loc, {queue_and_seq_id, is_delivered, is_persistent, msg_id}). -record(ssl_socket, {tcp, ssl}). -record(delivery, {mandatory, immediate, txn, sender, message}). @@ -83,9 +86,12 @@ -type(info_key() :: atom()). -type(info() :: {info_key(), any()}). -type(regexp() :: binary()). +-type(file_path() :: any()). +-type(io_device() :: any()). +-type(file_open_mode() :: any()). %% this is really an abstract type, but dialyzer does not support them --type(guid() :: any()). +-type(guid() :: binary()). -type(txn() :: guid()). -type(pkey() :: guid()). -type(r(Kind) :: @@ -128,17 +134,24 @@ properties :: amqp_properties(), properties_bin :: 'none', payload_fragments_rev :: [binary()]}). +-type(unencoded_content() :: undecoded_content()). -type(decoded_content() :: #content{class_id :: amqp_class_id(), properties :: amqp_properties(), properties_bin :: maybe(binary()), payload_fragments_rev :: [binary()]}). +-type(encoded_content() :: + #content{class_id :: amqp_class_id(), + properties :: maybe(amqp_properties()), + properties_bin :: binary(), + payload_fragments_rev :: [binary()]}). -type(content() :: undecoded_content() | decoded_content()). -type(basic_message() :: #basic_message{exchange_name :: exchange_name(), routing_key :: routing_key(), content :: content(), - persistent_key :: maybe(pkey())}). + guid :: guid(), + is_persistent :: boolean()}). -type(message() :: basic_message()). -type(delivery() :: #delivery{mandatory :: boolean(), @@ -146,9 +159,6 @@ txn :: maybe(txn()), sender :: pid(), message :: message()}). -%% this really should be an abstract type --type(msg_id() :: non_neg_integer()). --type(msg() :: {queue_name(), pid(), msg_id(), boolean(), message()}). -type(listener() :: #listener{node :: erlang_node(), protocol :: atom(), diff --git a/include/rabbit_queue.hrl b/include/rabbit_queue.hrl new file mode 100644 index 0000000000..165a7e7b99 --- /dev/null +++ b/include/rabbit_queue.hrl @@ -0,0 +1,58 @@ +%% The contents of this file are subject to the Mozilla Public License +%% Version 1.1 (the "License"); you may not use this file except in +%% compliance with the License. You may obtain a copy of the License at +%% http://www.mozilla.org/MPL/ +%% +%% 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 Original Code is RabbitMQ. +%% +%% The Initial Developers of the Original Code are LShift Ltd, +%% Cohesive Financial Technologies LLC, and Rabbit Technologies Ltd. +%% +%% Portions created before 22-Nov-2008 00:00:00 GMT by LShift Ltd, +%% Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd +%% are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial +%% Technologies LLC, and Rabbit Technologies Ltd. +%% +%% Portions created by LShift Ltd are Copyright (C) 2007-2009 LShift +%% Ltd. Portions created by Cohesive Financial Technologies LLC are +%% Copyright (C) 2007-2009 Cohesive Financial Technologies +%% LLC. Portions created by Rabbit Technologies Ltd are Copyright +%% (C) 2007-2009 Rabbit Technologies Ltd. +%% +%% All Rights Reserved. +%% +%% Contributor(s): ______________________________________. +%% + +-record(alpha, + { msg, + seq_id, + is_delivered, + msg_on_disk, + index_on_disk + }). + +-record(beta, + { msg_id, + seq_id, + is_persistent, + is_delivered, + index_on_disk + }). + +-record(gamma, + { seq_id, + count + }). + +-ifdef(use_specs). + +-type(gamma() :: #gamma { seq_id :: non_neg_integer(), + count :: non_neg_integer () }). + +-endif. diff --git a/src/file_handle_cache.erl b/src/file_handle_cache.erl new file mode 100644 index 0000000000..5c1c5a83d4 --- /dev/null +++ b/src/file_handle_cache.erl @@ -0,0 +1,455 @@ +%% The contents of this file are subject to the Mozilla Public License +%% Version 1.1 (the "License"); you may not use this file except in +%% compliance with the License. You may obtain a copy of the License at +%% http://www.mozilla.org/MPL/ +%% +%% 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 Original Code is RabbitMQ. +%% +%% The Initial Developers of the Original Code are LShift Ltd, +%% Cohesive Financial Technologies LLC, and Rabbit Technologies Ltd. +%% +%% Portions created before 22-Nov-2008 00:00:00 GMT by LShift Ltd, +%% Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd +%% are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial +%% Technologies LLC, and Rabbit Technologies Ltd. +%% +%% Portions created by LShift Ltd are Copyright (C) 2007-2009 LShift +%% Ltd. Portions created by Cohesive Financial Technologies LLC are +%% Copyright (C) 2007-2009 Cohesive Financial Technologies +%% LLC. Portions created by Rabbit Technologies Ltd are Copyright +%% (C) 2007-2009 Rabbit Technologies Ltd. +%% +%% All Rights Reserved. +%% +%% Contributor(s): ______________________________________. +%% + +-module(file_handle_cache). + +-export([open/3, close/1, read/2, append/2, sync/1, position/2, truncate/1, + last_sync_offset/1, current_virtual_offset/1, current_raw_offset/1, + append_write_buffer/1, copy/3]). + +%%---------------------------------------------------------------------------- + +-record(file, + { reader_count, + has_writer, + path + }). + +-record(handle, + { hdl, + offset, + trusted_offset, + is_dirty, + write_buffer_size, + write_buffer_size_limit, + write_buffer, + at_eof, + is_write, + is_read, + mode, + options, + global_key, + last_used_at + }). + +%%---------------------------------------------------------------------------- +%% Specs + +-ifdef(use_specs). + +-type(ref() :: any()). +-type(error() :: {'error', any()}). +-type(ok_or_error() :: ('ok' | error())). +-type(position() :: ('bof' | 'eof' | {'bof',integer()} | {'eof',integer()} + | {'cur',integer()} | integer())). + +-spec(open/3 :: (string(), [any()], [any()]) -> ({'ok', ref()} | error())). +-spec(close/1 :: (ref()) -> ('ok' | error())). +-spec(read/2 :: (ref(), integer()) -> + ({'ok', ([char()]|binary())} | eof | error())). +-spec(append/2 :: (ref(), iodata()) -> ok_or_error()). +-spec(sync/1 :: (ref()) -> ok_or_error()). +-spec(position/2 :: (ref(), position()) -> + ({'ok', non_neg_integer()} | error())). +-spec(truncate/1 :: (ref()) -> ok_or_error()). +-spec(last_sync_offset/1 :: (ref()) -> ({'ok', integer()} | error())). +-spec(current_virtual_offset/1 :: (ref()) -> ({'ok', integer()} | error())). +-spec(current_raw_offset/1 :: (ref()) -> ({'ok', integer()} | error())). +-spec(append_write_buffer/1 :: (ref()) -> ok_or_error()). +-spec(copy/3 :: (ref(), ref(), non_neg_integer()) -> + ({'ok', integer()} | error())). + +-endif. + +%%---------------------------------------------------------------------------- +%% Public API + +open(Path, Mode, Options) -> + case is_appender(Mode) of + true -> {error, append_not_supported}; + false -> + Path1 = filename:absname(Path), + case get({Path1, fhc_path}) of + {gref, GRef} -> + #file { reader_count = RCount, has_writer = HasWriter } + = File = get({GRef, fhc_file}), + Mode1 = lists:usort(Mode), + IsWriter = is_writer(Mode1), + case IsWriter andalso HasWriter of + true -> + {error, writer_exists}; + false -> + RCount1 = case is_reader(Mode1) of + true -> RCount + 1; + false -> RCount + end, + put({GRef, fhc_file}, + File #file { + reader_count = RCount1, + has_writer = HasWriter orelse IsWriter }), + Ref = make_ref(), + case open1(Path1, Mode1, Options, Ref, GRef) of + {ok, _Handle} -> {ok, Ref}; + Error -> Error + end + end; + undefined -> + GRef = make_ref(), + put({Path1, fhc_path}, {gref, GRef}), + put({GRef, fhc_file}, + #file { reader_count = 0, has_writer = false, + path = Path1 }), + open(Path, Mode, Options) + end + end. + +close(Ref) -> + case erase({Ref, fhc_handle}) of + undefined -> ok; + Handle -> + case write_buffer(Handle) of + {ok, #handle { hdl = Hdl, global_key = GRef, is_dirty = IsDirty, + is_read = IsReader, is_write = IsWriter }} -> + case Hdl of + closed -> ok; + _ -> ok = case IsDirty of + true -> file:sync(Hdl); + false -> ok + end, + ok = file:close(Hdl) + end, + #file { reader_count = RCount, has_writer = HasWriter, + path = Path } = File = get({GRef, fhc_file}), + RCount1 = case IsReader of + true -> RCount - 1; + false -> RCount + end, + HasWriter1 = HasWriter andalso not IsWriter, + case RCount1 =:= 0 andalso not HasWriter1 of + true -> erase({GRef, fhc_file}), + erase({Path, fhc_path}); + false -> put({GRef, fhc_file}, + File #file { reader_count = RCount1, + has_writer = HasWriter1 }) + end, + ok; + {Error, Handle1} -> + put({Ref, fhc_handle}, Handle1), + Error + end + end. + +read(Ref, Count) -> + case get_or_reopen(Ref) of + {ok, #handle { is_read = false }} -> + {error, not_open_for_reading}; + {ok, Handle} -> + {Result, Handle1} = + case write_buffer(Handle) of + {ok, Handle2 = #handle { hdl = Hdl, offset = Offset }} -> + case file:read(Hdl, Count) of + {ok, Data} = Obj -> + Size = iolist_size(Data), + {Obj, + Handle2 #handle { offset = Offset + Size }}; + eof -> {eof, Handle2 #handle { at_eof = true }}; + Error -> {Error, Handle2} + end; + {Error, Handle2} -> {Error, Handle2} + end, + put({Ref, fhc_handle}, Handle1), + Result; + Error -> Error + end. + +append(Ref, Data) -> + case get_or_reopen(Ref) of + {ok, #handle { is_write = false }} -> + {error, not_open_for_writing}; + {ok, Handle} -> + {Result, Handle1} = + case maybe_seek(eof, Handle) of + {{ok, _Offset}, Handle2 = #handle { at_eof = true }} -> + write_to_buffer(Data, Handle2); + {{error, _} = Error, Handle2} -> + {Error, Handle2} + end, + put({Ref, fhc_handle}, Handle1), + Result; + Error -> Error + end. + +sync(Ref) -> + case get_or_reopen(Ref) of + {ok, #handle { is_dirty = false, write_buffer = [] }} -> + ok; + {ok, Handle} -> + %% write_buffer will set is_dirty, or leave it set if buffer empty + {Result, Handle1} = + case write_buffer(Handle) of + {ok, Handle2 = #handle { + hdl = Hdl, offset = Offset, is_dirty = true }} -> + case file:sync(Hdl) of + ok -> {ok, + Handle2 #handle { trusted_offset = Offset, + is_dirty = false }}; + Error -> {Error, Handle2} + end; + Error -> {Error, Handle} + end, + put({Ref, fhc_handle}, Handle1), + Result; + Error -> Error + end. + +position(Ref, NewOffset) -> + case get_or_reopen(Ref) of + {ok, Handle} -> + {Result, Handle1} = + case write_buffer(Handle) of + {ok, Handle2} -> maybe_seek(NewOffset, Handle2); + {Error, Handle2} -> {Error, Handle2} + end, + put({Ref, fhc_handle}, Handle1), + Result; + Error -> Error + end. + +truncate(Ref) -> + case get_or_reopen(Ref) of + {ok, #handle { is_write = false }} -> + {error, not_open_for_writing}; + {ok, Handle} -> + {Result, Handle1} = + case write_buffer(Handle) of + {ok, + Handle2 = #handle { hdl = Hdl, offset = Offset, + trusted_offset = TrustedOffset }} -> + case file:truncate(Hdl) of + ok -> + {ok, + Handle2 #handle { + at_eof = true, + trusted_offset = lists:min([Offset, + TrustedOffset]) + }}; + Error -> {Error, Handle2} + end; + {Error, Handle2} -> {Error, Handle2} + end, + put({Ref, fhc_handle}, Handle1), + Result; + Error -> Error + end. + +last_sync_offset(Ref) -> + case get_or_reopen(Ref) of + {ok, #handle { trusted_offset = TrustedOffset }} -> {ok, TrustedOffset}; + Error -> Error + end. + +current_virtual_offset(Ref) -> + case get_or_reopen(Ref) of + {ok, #handle { at_eof = true, is_write = true, offset = Offset, + write_buffer_size = Size }} -> + {ok, Offset + Size}; + {ok, #handle { offset = Offset }} -> {ok, Offset}; + Error -> Error + end. + +current_raw_offset(Ref) -> + case get_or_reopen(Ref) of + {ok, #handle { offset = Offset }} -> {ok, Offset}; + Error -> Error + end. + +append_write_buffer(Ref) -> + case get_or_reopen(Ref) of + {ok, Handle} -> + {Result, Handle1} = write_buffer(Handle), + put({Ref, fhc_handle}, Handle1), + Result; + Error -> Error + end. + +copy(Src, Dest, Count) -> + case get_or_reopen(Src) of + {ok, SHandle = #handle { is_read = true }} -> + case get_or_reopen(Dest) of + {ok, DHandle = #handle { is_write = true }} -> + {Result, SHandle1, DHandle1} = + case write_buffer(SHandle) of + {ok, SHandle2 = #handle { hdl = SHdl, + offset = SOffset }} -> + case write_buffer(DHandle) of + {ok, + DHandle2 = #handle { hdl = DHdl, + offset = DOffset }} -> + Result1 = file:copy(SHdl, DHdl, Count), + case Result1 of + {ok, Count1} -> + {Result1, + SHandle2 #handle { + offset = SOffset + Count1 }, + DHandle2 #handle { + offset = DOffset + Count1 }}; + Error -> + {Error, SHandle2, DHandle2} + end; + Error -> {Error, SHandle2, DHandle} + end; + Error -> {Error, SHandle, DHandle} + end, + put({Src, fhc_handle}, SHandle1), + put({Dest, fhc_handle}, DHandle1), + Result; + {ok, _} -> {error, destination_not_open_for_writing}; + Error -> Error + end; + {ok, _} -> {error, source_not_open_for_reading}; + Error -> Error + end. + + +%%---------------------------------------------------------------------------- +%% Internal functions +%%---------------------------------------------------------------------------- + +get_or_reopen(Ref) -> + case get({Ref, fhc_handle}) of + undefined -> {error, not_open, Ref}; + #handle { hdl = closed, mode = Mode, global_key = GRef, + options = Options } -> + #file { path = Path } = get({GRef, fhc_file}), + open1(Path, Mode, Options, Ref, GRef); + Handle -> {ok, Handle #handle { last_used_at = now() }} + end. + +open1(Path, Mode, Options, Ref, GRef) -> + case file:open(Path, Mode) of + {ok, Hdl} -> + WriteBufferSize = + case proplists:get_value(write_buffer, Options, unbuffered) of + unbuffered -> 0; + infinity -> infinity; + N when is_integer(N) -> N + end, + Handle = + #handle { hdl = Hdl, offset = 0, trusted_offset = 0, + write_buffer_size = 0, options = Options, + write_buffer_size_limit = WriteBufferSize, + write_buffer = [], at_eof = false, mode = Mode, + is_write = is_writer(Mode), is_read = is_reader(Mode), + global_key = GRef, last_used_at = now(), + is_dirty = false }, + put({Ref, fhc_handle}, Handle), + {ok, Handle}; + {error, Reason} -> + {error, Reason} + end. + +maybe_seek(NewOffset, Handle = #handle { hdl = Hdl, at_eof = AtEoF, + offset = Offset }) -> + {AtEoF1, NeedsSeek} = needs_seek(AtEoF, Offset, NewOffset), + Result = case NeedsSeek of + true -> file:position(Hdl, NewOffset); + false -> {ok, Offset} + end, + case Result of + {ok, Offset1} -> + {Result, Handle #handle { at_eof = AtEoF1, offset = Offset1 }}; + {error, _} = Error -> {Error, Handle} + end. + +write_to_buffer(Data, Handle = #handle { hdl = Hdl, offset = Offset, + write_buffer_size_limit = 0 }) -> + Offset1 = Offset + iolist_size(Data), + {file:write(Hdl, Data), + Handle #handle { is_dirty = true, offset = Offset1 }}; +write_to_buffer(Data, Handle = + #handle { write_buffer = WriteBuffer, + write_buffer_size = Size, + write_buffer_size_limit = Limit }) -> + Size1 = Size + iolist_size(Data), + Handle1 = Handle #handle { write_buffer = [ Data | WriteBuffer ], + write_buffer_size = Size1 }, + case Limit /= infinity andalso Size1 > Limit of + true -> write_buffer(Handle1); + false -> {ok, Handle1} + end. + +write_buffer(Handle = #handle { write_buffer = [] }) -> + {ok, Handle}; +write_buffer(Handle = #handle { hdl = Hdl, offset = Offset, + write_buffer = WriteBuffer, + write_buffer_size = DataSize, + at_eof = true }) -> + case file:write(Hdl, lists:reverse(WriteBuffer)) of + ok -> + Offset1 = Offset + DataSize, + {ok, Handle #handle { offset = Offset1, write_buffer = [], + write_buffer_size = 0, is_dirty = true }}; + {error, _} = Error -> + {Error, Handle} + end. + +is_reader(Mode) -> lists:member(read, Mode). + +is_writer(Mode) -> lists:member(write, Mode). + +is_appender(Mode) -> lists:member(append, Mode). + +needs_seek(AtEof, _CurOffset, DesiredOffset) + when DesiredOffset == cur orelse DesiredOffset == {cur, 0} -> + {AtEof, false}; +needs_seek(true, _CurOffset, DesiredOffset) + when DesiredOffset == eof orelse DesiredOffset == {eof, 0} -> + {true, false}; +needs_seek(false, _CurOffset, DesiredOffset) + when DesiredOffset == eof orelse DesiredOffset == {eof, 0} -> + {true, true}; +needs_seek(AtEof, 0, DesiredOffset) + when DesiredOffset == bof orelse DesiredOffset == {bof, 0} -> + {AtEof, false}; +needs_seek(AtEof, CurOffset, CurOffset) -> + {AtEof, false}; +needs_seek(true, CurOffset, {bof, DesiredOffset}) + when DesiredOffset >= CurOffset -> + {true, true}; +needs_seek(true, _CurOffset, {cur, DesiredOffset}) + when DesiredOffset > 0 -> + {true, true}; +needs_seek(true, CurOffset, DesiredOffset) %% same as {bof, DO} + when is_integer(DesiredOffset) andalso DesiredOffset >= CurOffset -> + {true, true}; +%% because we can't really track size, we could well end up at EoF and not know +needs_seek(_AtEoF, _CurOffset, _DesiredOffset) -> + {false, true}. diff --git a/src/rabbit.erl b/src/rabbit.erl index b17711b473..1c5a07c7b9 100644 --- a/src/rabbit.erl +++ b/src/rabbit.erl @@ -37,7 +37,7 @@ -export([start/2, stop/1]). --export([log_location/1]). +-export([log_location/1, start_child/2]). -import(application). -import(mnesia). @@ -67,6 +67,7 @@ {nodes, [erlang_node()]} | {running_nodes, [erlang_node()]}]). -spec(log_location/1 :: ('sasl' | 'kernel') -> log_location()). +-spec(start_child/2 :: (atom(), [any()]) -> 'ok'). -endif. @@ -154,6 +155,7 @@ start(normal, []) -> ok = rabbit_amqqueue:start(), ok = start_child(rabbit_router), + ok = start_child(rabbit_guid), ok = start_child(rabbit_node_monitor), ok = start_child(rabbit_memory_monitor) end}, @@ -161,15 +163,14 @@ start(normal, []) -> 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) + DurableQueues = rabbit_amqqueue:find_durable_queues(), + ok = rabbit_queue_index:start_msg_store(DurableQueues), + {ok, _RealDurableQueues} = rabbit_amqqueue:recover(DurableQueues) + %% TODO - RealDurableQueues is a subset of + %% DurableQueues. It may have queues removed which + %% have since been recreated on another node in our + %% cluster. We need to remove DurableQueues -- + %% RealDurableQueues somehow. See also bug 20916 end}, {"builtin applications", fun () -> @@ -282,7 +283,7 @@ start_child(Mod) -> start_child(Mod, Args) -> {ok,_} = supervisor:start_child(rabbit_sup, {Mod, {Mod, start_link, Args}, - transient, 100, worker, [Mod]}), + transient, 5000, worker, [Mod]}), ok. ensure_working_log_handlers() -> diff --git a/src/rabbit_amqqueue.erl b/src/rabbit_amqqueue.erl index 4abfcd0ba7..82a0f5b4fa 100644 --- a/src/rabbit_amqqueue.erl +++ b/src/rabbit_amqqueue.erl @@ -31,16 +31,17 @@ -module(rabbit_amqqueue). --export([start/0, recover/0, declare/4, delete/3, purge/1]). --export([internal_declare/2, internal_delete/1]). +-export([start/0, recover/1, find_durable_queues/0, declare/4, delete/3, + purge/1]). +-export([internal_declare/2, internal_delete/1, remeasure_egress_rate/1]). -export([pseudo_queue/2]). -export([lookup/1, with/2, with_or_die/2, stat/1, stat_all/0, deliver/2, redeliver/2, requeue/3, ack/4]). -export([list/1, info/1, info/2, info_all/1, info_all/2]). -export([claim_queue/2]). -export([basic_get/3, basic_consume/8, basic_cancel/4]). --export([notify_sent/2, unblock/2, set_queue_duration/2, - send_memory_monitor_update/1]). +-export([notify_sent/2, unblock/2, tx_commit_msg_store_callback/4, + tx_commit_vq_callback/1]). -export([commit_all/2, rollback_all/2, notify_down_all/2, limit_all/3]). -export([on_node_down/1]). @@ -56,14 +57,19 @@ -ifdef(use_specs). +-type(msg_id() :: non_neg_integer()). +-type(msg() :: {queue_name(), pid(), msg_id(), boolean(), message()}). -type(qstats() :: {'ok', queue_name(), non_neg_integer(), non_neg_integer()}). -type(qlen() :: {'ok', non_neg_integer()}). -type(qfun(A) :: fun ((amqqueue()) -> A)). -type(ok_or_errors() :: 'ok' | {'error', [{'error' | 'exit' | 'throw', any()}]}). +-type(seq_id() :: non_neg_integer()). +-type(acktag() :: ('ack_not_on_disk' | {'ack_index_and_store', msg_id(), seq_id()})). -spec(start/0 :: () -> 'ok'). --spec(recover/0 :: () -> 'ok'). +-spec(recover/1 :: ([amqqueue()]) -> {'ok', [amqqueue()]}). +-spec(find_durable_queues/0 :: () -> [amqqueue()]). -spec(declare/4 :: (queue_name(), boolean(), boolean(), amqp_table()) -> amqqueue()). -spec(lookup/1 :: (queue_name()) -> {'ok', amqqueue()} | not_found()). @@ -102,10 +108,12 @@ -spec(basic_cancel/4 :: (amqqueue(), pid(), ctag(), any()) -> 'ok'). -spec(notify_sent/2 :: (pid(), pid()) -> 'ok'). -spec(unblock/2 :: (pid(), pid()) -> 'ok'). --spec(set_queue_duration/2 :: (pid(), number()) -> 'ok'). --spec(send_memory_monitor_update/1 :: (pid()) -> 'ok'). +-spec(tx_commit_msg_store_callback/4 :: + (pid(), [message()], [acktag()], {pid(), any()}) -> 'ok'). +-spec(tx_commit_vq_callback/1 :: (pid()) -> 'ok'). -spec(internal_declare/2 :: (amqqueue(), boolean()) -> amqqueue()). -spec(internal_delete/1 :: (queue_name()) -> 'ok' | not_found()). +-spec(remeasure_egress_rate/1 :: (pid()) -> 'ok'). -spec(on_node_down/1 :: (erlang_node()) -> 'ok'). -spec(pseudo_queue/2 :: (binary(), pid()) -> amqqueue()). @@ -121,40 +129,47 @@ start() -> transient, infinity, supervisor, [rabbit_amqqueue_sup]}), ok. -recover() -> - ok = recover_durable_queues(), - ok. - -recover_durable_queues() -> +recover(DurableQueues) -> + {ok, _RealDurableQueues} = recover_durable_queues(DurableQueues). + +recover_durable_queues(DurableQueues) -> + RealDurableQueues = + lists:foldl( + fun (RecoveredQ, Acc) -> + Q = start_queue_process(RecoveredQ), + %% We need to catch the case where a client connected to + %% another node has deleted the queue (and possibly + %% re-created it). + case rabbit_misc:execute_mnesia_transaction( + fun () -> + Match = + mnesia:match_object( + rabbit_durable_queue, RecoveredQ, read), + case Match of + [_] -> ok = store_queue(Q), + true; + [] -> false + end + end) of + true -> [Q|Acc]; + false -> exit(Q#amqqueue.pid, shutdown), + Acc + end + end, [], DurableQueues), + {ok, RealDurableQueues}. + +find_durable_queues() -> Node = node(), - lists:foreach( - fun (RecoveredQ) -> - Q = start_queue_process(RecoveredQ), - %% We need to catch the case where a client connected to - %% another node has deleted the queue (and possibly - %% re-created it). - case rabbit_misc:execute_mnesia_transaction( - fun () -> case mnesia:match_object( - rabbit_durable_queue, RecoveredQ, read) of - [_] -> ok = store_queue(Q), - true; - [] -> false - end - end) of - true -> ok; - false -> exit(Q#amqqueue.pid, shutdown) - end - end, - %% TODO: use dirty ops instead - rabbit_misc:execute_mnesia_transaction( - fun () -> - qlc:e(qlc:q([Q || Q = #amqqueue{pid = Pid} - <- mnesia:table(rabbit_durable_queue), - node(Pid) == Node])) - end)), - ok. + %% TODO: use dirty ops instead + rabbit_misc:execute_mnesia_transaction( + fun () -> + qlc:e(qlc:q([Q || Q = #amqqueue{pid = Pid} + <- mnesia:table(rabbit_durable_queue), + node(Pid) == Node])) + end). declare(QueueName, Durable, AutoDelete, Args) -> + prune_queue_childspecs(), Q = start_queue_process(#amqqueue{name = QueueName, durable = Durable, auto_delete = AutoDelete, @@ -165,6 +180,10 @@ declare(QueueName, Durable, AutoDelete, Args) -> internal_declare(Q = #amqqueue{name = QueueName}, WantDefaultBinding) -> case rabbit_misc:execute_mnesia_transaction( fun () -> + %% we could still find that mnesia has another + %% entry here because the queue may exist on + %% another node, beyond the knowledge of our own + %% local queue_sup. case mnesia:wread({rabbit_queue, QueueName}) of [] -> ok = store_queue(Q), case WantDefaultBinding of @@ -188,9 +207,30 @@ store_queue(Q = #amqqueue{durable = false}) -> ok = mnesia:write(rabbit_queue, Q, write), ok. -start_queue_process(Q) -> - {ok, Pid} = supervisor:start_child(rabbit_amqqueue_sup, [Q]), - Q#amqqueue{pid = Pid}. +start_queue_process(Q = #amqqueue{name = QueueName}) -> + case supervisor:start_child( + rabbit_amqqueue_sup, + {QueueName, {rabbit_amqqueue_process, start_link, [Q]}, + %% 4294967295 is 2^32 - 1, which is the highest value allowed + temporary, 4294967295, worker, [rabbit_amqqueue_process]}) of + {ok, Pid} -> + Q#amqqueue{pid = Pid}; + {error, already_present} -> + supervisor:delete_child(rabbit_amqqueue_sup, QueueName), + start_queue_process(Q); + {error, {already_started, _QPid}} -> + case rabbit_misc:execute_mnesia_transaction( + fun () -> + case mnesia:wread({rabbit_queue, QueueName}) of + %% it's vanished in the mean time, try again + [] -> try_again; + [ExistingQ] -> ExistingQ + end + end) of + try_again -> start_queue_process(Q); + ExistingQ -> ExistingQ + end + end. add_default_binding(#amqqueue{name = QueueName}) -> Exchange = rabbit_misc:r(QueueName, exchange, <<>>), @@ -238,6 +278,7 @@ stat_all() -> lists:map(fun stat/1, rabbit_misc:dirty_read_all(rabbit_queue)). delete(#amqqueue{ pid = QPid }, IfUnused, IfEmpty) -> + prune_queue_childspecs(), gen_server2:call(QPid, {delete, IfUnused, IfEmpty}, infinity). purge(#amqqueue{ pid = QPid }) -> gen_server2:call(QPid, purge, infinity). @@ -311,11 +352,12 @@ notify_sent(QPid, ChPid) -> unblock(QPid, ChPid) -> gen_server2:pcast(QPid, 8, {unblock, ChPid}). -set_queue_duration(QPid, Duration) -> - gen_server2:pcast(QPid, 7, {set_queue_duration, Duration}). +tx_commit_msg_store_callback(QPid, Pubs, AckTags, From) -> + gen_server2:pcast(QPid, 8, + {tx_commit_msg_store_callback, Pubs, AckTags, From}). -send_memory_monitor_update(QPid) -> - gen_server2:pcast(QPid, 7, send_memory_monitor_update). +tx_commit_vq_callback(QPid) -> + gen_server2:pcast(QPid, 8, tx_commit_vq_callback). internal_delete(QueueName) -> rabbit_misc:execute_mnesia_transaction( @@ -330,6 +372,17 @@ internal_delete(QueueName) -> end end). +remeasure_egress_rate(QPid) -> + gen_server2:pcast(QPid, 8, remeasure_egress_rate). + +prune_queue_childspecs() -> + lists:foreach( + fun ({Name, undefined, _Type, _Mods}) -> + supervisor:delete_child(rabbit_amqqueue_sup, Name); + (_) -> ok + end, supervisor:which_children(rabbit_amqqueue_sup)), + ok. + on_node_down(Node) -> rabbit_misc:execute_mnesia_transaction( fun () -> diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl index 2d264fc274..cd70979a1f 100644 --- a/src/rabbit_amqqueue_process.erl +++ b/src/rabbit_amqqueue_process.erl @@ -35,13 +35,16 @@ -behaviour(gen_server2). --define(UNSENT_MESSAGE_LIMIT, 100). --define(HIBERNATE_AFTER_MIN, 1000). --define(DESIRED_HIBERNATE, 10000). +-define(UNSENT_MESSAGE_LIMIT, 100). +-define(HIBERNATE_AFTER_MIN, 1000). +-define(DESIRED_HIBERNATE, 10000). +-define(SYNC_INTERVAL, 5). %% milliseconds +-define(EGRESS_REMEASURE_INTERVAL, 5000). -export([start_link/1]). --export([init/1, terminate/2, code_change/3, handle_call/3, handle_cast/2, handle_info/2]). +-export([init/1, terminate/2, code_change/3, handle_call/3, handle_cast/2, + handle_info/2, handle_pre_hibernate/1]). -import(queue). -import(erlang). @@ -52,20 +55,17 @@ owner, exclusive_consumer, has_had_consumers, + variable_queue_state, next_msg_id, - message_buffer, active_consumers, blocked_consumers, - drain_ratio}). + sync_timer_ref, + egress_rate_timer_ref + }). -record(consumer, {tag, ack_required}). --record(tx, {ch_pid, is_persistent, pending_messages, pending_acks}). - --record(ratio, {ratio, %% float. messages/microsecond_us - t0, %% previous timestamp (us) - next_msg_id %% previous next_msg_id - }). +-record(tx, {ch_pid, pending_messages, pending_acks}). %% These are held in our process dictionary -record(cr, {consumer_count, @@ -90,7 +90,8 @@ acks_uncommitted, consumers, transactions, - memory]). + memory + ]). %%---------------------------------------------------------------------------- @@ -98,43 +99,105 @@ start_link(Q) -> gen_server2:start_link(?MODULE, Q, []). %%---------------------------------------------------------------------------- -init(Q) -> + +init(Q = #amqqueue { name = QName }) -> ?LOGDEBUG("Queue starting - ~p~n", [Q]), - rabbit_memory_monitor:register(self(), {rabbit_amqqueue, set_queue_duration, - [self()]}), - %% Beware. This breaks hibernation! - timer:apply_interval(2500, rabbit_amqqueue, send_memory_monitor_update, - [self()]), - {ok, #q{q = Q, - owner = none, - exclusive_consumer = none, - has_had_consumers = false, - next_msg_id = 1, - message_buffer = queue:new(), - active_consumers = queue:new(), - blocked_consumers = queue:new(), - drain_ratio = #ratio{ratio = 0.0, - t0 = now(), - next_msg_id = 1} - }, hibernate, + process_flag(trap_exit, true), + ok = rabbit_memory_manager:register + (self(), false, rabbit_amqqueue, set_storage_mode, [self()]), + VQS = rabbit_variable_queue:init(QName), + State = #q{q = Q, + owner = none, + exclusive_consumer = none, + has_had_consumers = false, + variable_queue_state = VQS, + next_msg_id = 1, + active_consumers = queue:new(), + blocked_consumers = queue:new(), + sync_timer_ref = undefined, + egress_rate_timer_ref = undefined + }, + {ok, State, hibernate, {backoff, ?HIBERNATE_AFTER_MIN, ?HIBERNATE_AFTER_MIN, ?DESIRED_HIBERNATE}}. -terminate(_Reason, State) -> +terminate(shutdown, #q{variable_queue_state = VQS}) -> + _VQS = rabbit_variable_queue:terminate(VQS); +terminate(_Reason, State = #q{variable_queue_state = VQS}) -> %% FIXME: How do we cancel active subscriptions? - QName = qname(State), - lists:foreach(fun (Txn) -> ok = rollback_work(Txn, QName) end, - all_tx()), - ok = purge_message_buffer(QName, State#q.message_buffer), - ok = rabbit_amqqueue:internal_delete(QName). + %% Ensure that any persisted tx messages are removed. + %% TODO: wait for all in flight tx_commits to complete + VQS1 = rabbit_variable_queue:tx_rollback( + lists:concat([PM || #tx { pending_messages = PM } <- + all_tx_record()]), VQS), + %% Delete from disk first. If we crash at this point, when a + %% durable queue, we will be recreated at startup, possibly with + %% partial content. The alternative is much worse however - if we + %% called internal_delete first, we would then have a race between + %% the disk delete and a new queue with the same name being + %% created and published to. + _VQS = rabbit_variable_queue:delete(VQS1), + ok = rabbit_amqqueue:internal_delete(qname(State)). code_change(_OldVsn, State, _Extra) -> {ok, State}. %%---------------------------------------------------------------------------- -reply(Reply, NewState) -> {reply, Reply, NewState, hibernate}. - -noreply(NewState) -> {noreply, NewState, hibernate}. +reply(Reply, NewState) -> + assert_invariant(NewState), + {NewState1, Timeout} = next_state(NewState), + {reply, Reply, NewState1, Timeout}. + +noreply(NewState) -> + assert_invariant(NewState), + {NewState1, Timeout} = next_state(NewState), + {noreply, NewState1, Timeout}. + +next_state(State = #q{variable_queue_state = VQS}) -> + next_state1(ensure_egress_rate_timer(State), + rabbit_variable_queue:needs_sync(VQS)). + +next_state1(State = #q{sync_timer_ref = undefined}, true) -> + {start_sync_timer(State), 0}; +next_state1(State, true) -> + {State, 0}; +next_state1(State = #q{sync_timer_ref = undefined, + variable_queue_state = VQS}, false) -> + {State, case rabbit_variable_queue:can_flush_journal(VQS) of + true -> 0; + false -> hibernate + end}; +next_state1(State, false) -> + {stop_sync_timer(State), 0}. + +ensure_egress_rate_timer(State = #q{egress_rate_timer_ref = undefined}) -> + {ok, TRef} = timer:apply_after(?EGRESS_REMEASURE_INTERVAL, rabbit_amqqueue, + remeasure_egress_rate, [self()]), + State#q{egress_rate_timer_ref = TRef}; +ensure_egress_rate_timer(State = #q{egress_rate_timer_ref = just_measured}) -> + State#q{egress_rate_timer_ref = undefined}; +ensure_egress_rate_timer(State) -> + State. + +stop_egress_rate_timer(State = #q{egress_rate_timer_ref = undefined}) -> + State; +stop_egress_rate_timer(State = #q{egress_rate_timer_ref = just_measured}) -> + State#q{egress_rate_timer_ref = undefined}; +stop_egress_rate_timer(State = #q{egress_rate_timer_ref = TRef}) -> + {ok, cancel} = timer:cancel(TRef), + State#q{egress_rate_timer_ref = undefined}. + +start_sync_timer(State = #q{sync_timer_ref = undefined}) -> + {ok, TRef} = timer:apply_after(?SYNC_INTERVAL, rabbit_amqqueue, + tx_commit_vq_callback, [self()]), + State#q{sync_timer_ref = TRef}. + +stop_sync_timer(State = #q{sync_timer_ref = TRef}) -> + {ok, cancel} = timer:cancel(TRef), + State#q{sync_timer_ref = undefined}. + +assert_invariant(#q{active_consumers = AC, variable_queue_state = VQS}) -> + true = (queue:is_empty(AC) orelse rabbit_variable_queue:is_empty(VQS)). lookup_ch(ChPid) -> case get({ch, ChPid}) of @@ -181,12 +244,12 @@ record_current_channel_tx(ChPid, Txn) -> %% that wasn't happening already) store_ch_record((ch_record(ChPid))#cr{txn = Txn}). -deliver_immediately(Message, Delivered, - State = #q{q = #amqqueue{name = QName}, - active_consumers = ActiveConsumers, - blocked_consumers = BlockedConsumers, - next_msg_id = NextId}) -> - ?LOGDEBUG("AMQQUEUE ~p DELIVERY:~n~p~n", [QName, Message]), +deliver_msgs_to_consumers( + Funs = {PredFun, DeliverFun}, FunAcc, + State = #q{q = #amqqueue{name = QName}, + active_consumers = ActiveConsumers, + blocked_consumers = BlockedConsumers, + next_msg_id = NextId}) -> case queue:out(ActiveConsumers) of {{value, QEntry = {ChPid, #consumer{tag = ConsumerTag, ack_required = AckRequired}}}, @@ -194,15 +257,21 @@ deliver_immediately(Message, Delivered, C = #cr{limiter_pid = LimiterPid, unsent_message_count = Count, unacked_messages = UAM} = ch_record(ChPid), - case rabbit_limiter:can_send(LimiterPid, self(), AckRequired) of + IsMsgReady = PredFun(FunAcc, State), + case (IsMsgReady andalso + rabbit_limiter:can_send( LimiterPid, self(), AckRequired )) of true -> + {{Msg, IsDelivered, AckTag}, FunAcc1, State1} = + DeliverFun(AckRequired, FunAcc, State), + ?LOGDEBUG("AMQQUEUE ~p DELIVERY:~n~p~n", [QName, Msg]), rabbit_channel:deliver( ChPid, ConsumerTag, AckRequired, - {QName, self(), NextId, Delivered, Message}), - NewUAM = case AckRequired of - true -> dict:store(NextId, Message, UAM); - false -> UAM - end, + {QName, self(), NextId, IsDelivered, Msg}), + NewUAM = + case AckRequired of + true -> dict:store(NextId, {Msg, AckTag}, UAM); + false -> UAM + end, NewC = C#cr{unsent_message_count = Count + 1, unacked_messages = NewUAM}, store_ch_record(NewC), @@ -218,54 +287,111 @@ deliver_immediately(Message, Delivered, {ActiveConsumers1, queue:in(QEntry, BlockedConsumers1)} end, - {offered, AckRequired, - State#q{active_consumers = NewActiveConsumers, - blocked_consumers = NewBlockedConsumers, - next_msg_id = NextId + 1}}; - false -> + State2 = State1 #q { + active_consumers = NewActiveConsumers, + blocked_consumers = NewBlockedConsumers, + next_msg_id = NextId + 1 + }, + deliver_msgs_to_consumers(Funs, FunAcc1, State2); + %% if IsMsgReady then we've hit the limiter + false when IsMsgReady -> store_ch_record(C#cr{is_limit_active = true}), {NewActiveConsumers, NewBlockedConsumers} = move_consumers(ChPid, ActiveConsumers, BlockedConsumers), - deliver_immediately( - Message, Delivered, + deliver_msgs_to_consumers( + Funs, FunAcc, State#q{active_consumers = NewActiveConsumers, - blocked_consumers = NewBlockedConsumers}) + blocked_consumers = NewBlockedConsumers}); + false -> + %% no message was ready, so we don't need to block anyone + {FunAcc, State} end; {empty, _} -> - {not_offered, State} + {FunAcc, State} end. -attempt_delivery(none, _ChPid, Message, State) -> - case deliver_immediately(Message, false, State) of - {offered, false, State1} -> - {true, State1}; - {offered, true, State1} -> - persist_message(none, qname(State), Message), - persist_delivery(qname(State), Message, false), - {true, State1}; - {not_offered, State1} -> - {false, State1} - end; -attempt_delivery(Txn, ChPid, Message, State) -> - persist_message(Txn, qname(State), Message), - record_pending_message(Txn, ChPid, Message), - {true, State}. - -deliver_or_enqueue(Txn, ChPid, Message, State) -> - case attempt_delivery(Txn, ChPid, Message, State) of +deliver_from_queue_pred({IsEmpty, _AutoAcks}, _State) -> + not IsEmpty. +deliver_from_queue_deliver(AckRequired, {false, AutoAcks}, + State = #q { variable_queue_state = VQS }) -> + {{Msg, IsDelivered, AckTag, Remaining}, VQS1} = + rabbit_variable_queue:fetch(VQS), + AutoAcks1 = case AckRequired of + true -> AutoAcks; + false -> [AckTag | AutoAcks] + end, + {{Msg, IsDelivered, AckTag}, {0 == Remaining, AutoAcks1}, + State #q { variable_queue_state = VQS1 }}. + +run_message_queue(State = #q { variable_queue_state = VQS }) -> + Funs = { fun deliver_from_queue_pred/2, + fun deliver_from_queue_deliver/3 }, + IsEmpty = rabbit_variable_queue:is_empty(VQS), + {{_IsEmpty1, AutoAcks}, State1} = + deliver_msgs_to_consumers(Funs, {IsEmpty, []}, State), + VQS1 = rabbit_variable_queue:ack(AutoAcks, State1 #q.variable_queue_state), + State1 #q { variable_queue_state = VQS1 }. + +attempt_immediate_delivery(none, _ChPid, Msg, State) -> + PredFun = fun (IsEmpty, _State) -> not IsEmpty end, + DeliverFun = + fun (AckRequired, false, State1) -> + {AckTag, State2} = + case AckRequired of + true -> + {AckTag1, VQS} = + rabbit_variable_queue:publish_delivered( + Msg, State1 #q.variable_queue_state), + {AckTag1, State1 #q { variable_queue_state = VQS }}; + false -> + {noack, State1} + end, + {{Msg, false, AckTag}, true, State2} + end, + deliver_msgs_to_consumers({ PredFun, DeliverFun }, false, State); +attempt_immediate_delivery(Txn, ChPid, Msg, State) -> + VQS = rabbit_variable_queue:tx_publish(Msg, State #q.variable_queue_state), + record_pending_message(Txn, ChPid, Msg), + {true, State #q { variable_queue_state = VQS }}. + +deliver_or_enqueue(Txn, ChPid, Msg, State) -> + case attempt_immediate_delivery(Txn, ChPid, Msg, State) of {true, NewState} -> {true, NewState}; {false, NewState} -> - persist_message(Txn, qname(State), Message), - NewMB = queue:in({Message, false}, NewState#q.message_buffer), - {false, NewState#q{message_buffer = NewMB}} + %% Txn is none and no unblocked channels with consumers + {_SeqId, VQS} = rabbit_variable_queue:publish( + Msg, State #q.variable_queue_state), + {false, NewState #q { variable_queue_state = VQS }} end. -deliver_or_enqueue_n(Messages, State = #q{message_buffer = MessageBuffer}) -> - run_poke_burst(queue:join(MessageBuffer, queue:from_list(Messages)), - State). +%% all these messages have already been delivered at least once and +%% not ack'd, but need to be either redelivered or requeued +deliver_or_requeue_n([], State) -> + State; +deliver_or_requeue_n(MsgsWithAcks, State) -> + Funs = { fun deliver_or_requeue_msgs_pred/2, + fun deliver_or_requeue_msgs_deliver/3 }, + {{_RemainingLengthMinusOne, AutoAcks, OutstandingMsgs}, NewState} = + deliver_msgs_to_consumers( + Funs, {length(MsgsWithAcks), [], MsgsWithAcks}, State), + VQS = rabbit_variable_queue:ack(AutoAcks, NewState #q.variable_queue_state), + case OutstandingMsgs of + [] -> NewState #q { variable_queue_state = VQS }; + _ -> VQS1 = rabbit_variable_queue:requeue(OutstandingMsgs, VQS), + NewState #q { variable_queue_state = VQS1 } + end. + +deliver_or_requeue_msgs_pred({Len, _AcksAcc, _MsgsWithAcks}, _State) -> + 0 < Len. +deliver_or_requeue_msgs_deliver( + false, {Len, AcksAcc, [{Msg, AckTag} | MsgsWithAcks]}, State) -> + {{Msg, true, noack}, {Len - 1, [AckTag | AcksAcc], MsgsWithAcks}, State}; +deliver_or_requeue_msgs_deliver( + true, {Len, AcksAcc, [{Msg, AckTag} | MsgsWithAcks]}, State) -> + {{Msg, true, AckTag}, {Len - 1, AcksAcc, MsgsWithAcks}, State}. add_consumer(ChPid, Consumer, Queue) -> queue:in({ChPid, Consumer}, Queue). @@ -299,7 +425,7 @@ possibly_unblock(State, ChPid, Update) -> move_consumers(ChPid, State#q.blocked_consumers, State#q.active_consumers), - run_poke_burst( + run_message_queue( State#q{active_consumers = NewActiveConsumers, blocked_consumers = NewBlockedeConsumers}) end @@ -316,27 +442,27 @@ handle_ch_down(DownPid, State = #q{exclusive_consumer = Holder}) -> unacked_messages = UAM} -> erlang:demonitor(MonitorRef), erase({ch, ChPid}), - case Txn of - none -> ok; - _ -> ok = rollback_work(Txn, qname(State)), - erase_tx(Txn) - end, - NewState = - deliver_or_enqueue_n( - [{Message, true} || - {_Messsage_id, Message} <- dict:to_list(UAM)], - State#q{ - exclusive_consumer = case Holder of - {ChPid, _} -> none; - Other -> Other - end, - active_consumers = remove_consumers( - ChPid, State#q.active_consumers), - blocked_consumers = remove_consumers( - ChPid, State#q.blocked_consumers)}), - case should_auto_delete(NewState) of - false -> {ok, NewState}; - true -> {stop, NewState} + State1 = State#q{ + exclusive_consumer = case Holder of + {ChPid, _} -> none; + Other -> Other + end, + active_consumers = remove_consumers( + ChPid, State#q.active_consumers), + blocked_consumers = remove_consumers( + ChPid, State#q.blocked_consumers)}, + case should_auto_delete(State1) of + true -> + {stop, State1}; + false -> + State2 = case Txn of + none -> State1; + _ -> rollback_transaction(Txn, State1) + end, + {ok, + deliver_or_requeue_n( + [MsgWithAck || + {_MsgId, MsgWithAck} <- dict:to_list(UAM)], State2)} end end. @@ -359,26 +485,6 @@ check_exclusive_access(none, true, State) -> false -> in_use end. -run_poke_burst(State = #q{message_buffer = MessageBuffer}) -> - run_poke_burst(MessageBuffer, State). - -run_poke_burst(MessageBuffer, State) -> - case queue:out(MessageBuffer) of - {{value, {Message, Delivered}}, BufferTail} -> - case deliver_immediately(Message, Delivered, State) of - {offered, true, NewState} -> - persist_delivery(qname(State), Message, Delivered), - run_poke_burst(BufferTail, NewState); - {offered, false, NewState} -> - persist_auto_ack(qname(State), Message), - run_poke_burst(BufferTail, NewState); - {not_offered, NewState} -> - NewState#q{message_buffer = MessageBuffer} - end; - {empty, _} -> - State#q{message_buffer = MessageBuffer} - end. - is_unused(State) -> queue:is_empty(State#q.active_consumers) andalso queue:is_empty(State#q.blocked_consumers). @@ -387,66 +493,9 @@ maybe_send_reply(ChPid, Msg) -> ok = rabbit_channel:send_command(ChPid, Msg). qname(#q{q = #amqqueue{name = QName}}) -> QName. -persist_message(_Txn, _QName, #basic_message{persistent_key = none}) -> - ok; -persist_message(Txn, QName, Message) -> - M = Message#basic_message{ - %% don't persist any recoverable decoded properties, rebuild from properties_bin on restore - content = rabbit_binary_parser:clear_decoded_content( - Message#basic_message.content)}, - persist_work(Txn, QName, - [{publish, M, {QName, M#basic_message.persistent_key}}]). - -persist_delivery(_QName, _Message, - true) -> - ok; -persist_delivery(_QName, #basic_message{persistent_key = none}, - _Delivered) -> - ok; -persist_delivery(QName, #basic_message{persistent_key = PKey}, - _Delivered) -> - persist_work(none, QName, [{deliver, {QName, PKey}}]). - -persist_acks(Txn, QName, Messages) -> - persist_work(Txn, QName, - [{ack, {QName, PKey}} || - #basic_message{persistent_key = PKey} <- Messages, - PKey =/= none]). - -persist_auto_ack(_QName, #basic_message{persistent_key = none}) -> - ok; -persist_auto_ack(QName, #basic_message{persistent_key = PKey}) -> - %% auto-acks are always non-transactional - rabbit_persister:dirty_work([{ack, {QName, PKey}}]). - -persist_work(_Txn,_QName, []) -> - ok; -persist_work(none, _QName, WorkList) -> - rabbit_persister:dirty_work(WorkList); -persist_work(Txn, QName, WorkList) -> - mark_tx_persistent(Txn), - rabbit_persister:extend_transaction({Txn, QName}, WorkList). - -commit_work(Txn, QName) -> - do_if_persistent(fun rabbit_persister:commit_transaction/1, - Txn, QName). - -rollback_work(Txn, QName) -> - do_if_persistent(fun rabbit_persister:rollback_transaction/1, - Txn, QName). - -%% optimisation: don't do unnecessary work -%% it would be nice if this was handled by the persister -do_if_persistent(F, Txn, QName) -> - case is_tx_persistent(Txn) of - false -> ok; - true -> ok = F({Txn, QName}) - end. - lookup_tx(Txn) -> case get({txn, Txn}) of undefined -> #tx{ch_pid = none, - is_persistent = false, pending_messages = [], pending_acks = []}; V -> V @@ -461,22 +510,10 @@ erase_tx(Txn) -> all_tx_record() -> [T || {{txn, _}, T} <- get()]. -all_tx() -> - [Txn || {{txn, Txn}, _} <- get()]. - -mark_tx_persistent(Txn) -> - Tx = lookup_tx(Txn), - store_tx(Txn, Tx#tx{is_persistent = true}). - -is_tx_persistent(Txn) -> - #tx{is_persistent = Res} = lookup_tx(Txn), - Res. - record_pending_message(Txn, ChPid, Message) -> Tx = #tx{pending_messages = Pending} = lookup_tx(Txn), record_current_channel_tx(ChPid, Txn), - store_tx(Txn, Tx#tx{pending_messages = [{Message, false} | Pending], - ch_pid = ChPid}). + store_tx(Txn, Tx #tx { pending_messages = [Message | Pending] }). record_pending_acks(Txn, ChPid, MsgIds) -> Tx = #tx{pending_acks = Pending} = lookup_tx(Txn), @@ -484,38 +521,42 @@ record_pending_acks(Txn, ChPid, MsgIds) -> store_tx(Txn, Tx#tx{pending_acks = [MsgIds | Pending], ch_pid = ChPid}). -process_pending(Txn, State) -> - #tx{ch_pid = ChPid, - pending_messages = PendingMessages, - pending_acks = PendingAcks} = lookup_tx(Txn), - case lookup_ch(ChPid) of - not_found -> ok; - C = #cr{unacked_messages = UAM} -> - {_Acked, Remaining} = - collect_messages(lists:append(PendingAcks), UAM), - store_ch_record(C#cr{unacked_messages = Remaining}) - end, - deliver_or_enqueue_n(lists:reverse(PendingMessages), State). +commit_transaction(Txn, From, State) -> + #tx { ch_pid = ChPid, + pending_messages = PendingMessages, + pending_acks = PendingAcks + } = lookup_tx(Txn), + PendingMessagesOrdered = lists:reverse(PendingMessages), + PendingAcksOrdered = lists:append(PendingAcks), + Acks = + case lookup_ch(ChPid) of + not_found -> []; + C = #cr { unacked_messages = UAM } -> + {MsgsWithAcks, Remaining} = + collect_messages(PendingAcksOrdered, UAM), + store_ch_record(C#cr{unacked_messages = Remaining}), + [AckTag || {_Msg, AckTag} <- MsgsWithAcks] + end, + {RunQueue, VQS} = + rabbit_variable_queue:tx_commit( + PendingMessagesOrdered, Acks, From, State #q.variable_queue_state), + {RunQueue, State #q { variable_queue_state = VQS }}. + +rollback_transaction(Txn, State) -> + #tx { pending_messages = PendingMessages + } = lookup_tx(Txn), + VQS = rabbit_variable_queue:tx_rollback(PendingMessages, + State #q.variable_queue_state), + erase_tx(Txn), + State #q { variable_queue_state = VQS }. +%% {A, B} = collect_messages(C, D) %% A = C `intersect` D; B = D \\ C +%% err, A = C `intersect` D , via projection through the dict that is C collect_messages(MsgIds, UAM) -> lists:mapfoldl( fun (MsgId, D) -> {dict:fetch(MsgId, D), dict:erase(MsgId, D)} end, UAM, MsgIds). -purge_message_buffer(QName, MessageBuffer) -> - Messages = - [[Message || {Message, _Delivered} <- - queue:to_list(MessageBuffer)] | - lists:map( - fun (#cr{unacked_messages = UAM}) -> - [Message || {_MessageId, Message} <- dict:to_list(UAM)] - end, - all_ch_record())], - %% the simplest, though certainly not the most obvious or - %% efficient, way to purge messages from the persister is to - %% artifically ack them. - persist_acks(none, QName, lists:append(Messages)). - infos(Items, State) -> [{Item, i(Item, State)} || Item <- Items]. i(name, #q{q = #amqqueue{name = Name}}) -> Name; @@ -524,8 +565,8 @@ i(auto_delete, #q{q = #amqqueue{auto_delete = AutoDelete}}) -> AutoDelete; i(arguments, #q{q = #amqqueue{arguments = Arguments}}) -> Arguments; i(pid, _) -> self(); -i(messages_ready, #q{message_buffer = MessageBuffer}) -> - queue:len(MessageBuffer); +i(messages_ready, #q { variable_queue_state = VQS }) -> + rabbit_variable_queue:len(VQS); i(messages_unacknowledged, _) -> lists:sum([dict:size(UAM) || #cr{unacked_messages = UAM} <- all_ch_record()]); @@ -574,7 +615,8 @@ handle_call({deliver_immediately, Txn, Message, ChPid}, _From, State) -> %% just all ready-to-consume queues get the message, with unready %% queues discarding the message? %% - {Delivered, NewState} = attempt_delivery(Txn, ChPid, Message, State), + {Delivered, NewState} = + attempt_immediate_delivery(Txn, ChPid, Message, State), reply(Delivered, NewState); handle_call({deliver, Txn, Message, ChPid}, _From, State) -> @@ -583,12 +625,12 @@ handle_call({deliver, Txn, Message, ChPid}, _From, State) -> reply(Delivered, NewState); handle_call({commit, Txn}, From, State) -> - ok = commit_work(Txn, qname(State)), - %% optimisation: we reply straight away so the sender can continue - gen_server2:reply(From, ok), - NewState = process_pending(Txn, State), + {RunQueue, NewState} = commit_transaction(Txn, From, State), erase_tx(Txn), - noreply(NewState); + noreply(case RunQueue of + true -> run_message_queue(NewState); + false -> NewState + end); handle_call({notify_down, ChPid}, _From, State) -> %% we want to do this synchronously, so that auto_deleted queues @@ -604,25 +646,25 @@ handle_call({notify_down, ChPid}, _From, State) -> handle_call({basic_get, ChPid, NoAck}, _From, State = #q{q = #amqqueue{name = QName}, next_msg_id = NextId, - message_buffer = MessageBuffer}) -> - case queue:out(MessageBuffer) of - {{value, {Message, Delivered}}, BufferTail} -> + variable_queue_state = VQS + }) -> + case rabbit_variable_queue:fetch(VQS) of + {empty, VQS1} -> reply(empty, State #q { variable_queue_state = VQS1 }); + {{Msg, IsDelivered, AckTag, Remaining}, VQS1} -> AckRequired = not(NoAck), - case AckRequired of - true -> - persist_delivery(QName, Message, Delivered), - C = #cr{unacked_messages = UAM} = ch_record(ChPid), - NewUAM = dict:store(NextId, Message, UAM), - store_ch_record(C#cr{unacked_messages = NewUAM}); - false -> - persist_auto_ack(QName, Message) - end, - Msg = {QName, self(), NextId, Delivered, Message}, - reply({ok, queue:len(BufferTail), Msg}, - State#q{message_buffer = BufferTail, - next_msg_id = NextId + 1}); - {empty, _} -> - reply(empty, State) + VQS2 = + case AckRequired of + true -> + C = #cr{unacked_messages = UAM} = ch_record(ChPid), + NewUAM = dict:store(NextId, {Msg, AckTag}, UAM), + store_ch_record(C#cr{unacked_messages = NewUAM}), + VQS1; + false -> + rabbit_variable_queue:ack([AckTag], VQS1) + end, + Message = {QName, self(), NextId, IsDelivered, Msg}, + reply({ok, Remaining, Message}, + State #q { next_msg_id = NextId + 1, variable_queue_state = VQS2 }) end; handle_call({basic_consume, NoAck, ReaderPid, ChPid, LimiterPid, @@ -643,15 +685,14 @@ handle_call({basic_consume, NoAck, ReaderPid, ChPid, LimiterPid, ack_required = not(NoAck)}, store_ch_record(C#cr{consumer_count = ConsumerCount +1, limiter_pid = LimiterPid}), - if ConsumerCount == 0 -> - ok = rabbit_limiter:register(LimiterPid, self()); - true -> - ok + case ConsumerCount of + 0 -> ok = rabbit_limiter:register(LimiterPid, self()); + _ -> ok end, - ExclusiveConsumer = - if ExclusiveConsume -> {ChPid, ConsumerTag}; - true -> ExistingHolder - end, + ExclusiveConsumer = case ExclusiveConsume of + true -> {ChPid, ConsumerTag}; + false -> ExistingHolder + end, State1 = State#q{has_had_consumers = true, exclusive_consumer = ExclusiveConsumer}, ok = maybe_send_reply(ChPid, OkMsg), @@ -662,7 +703,7 @@ handle_call({basic_consume, NoAck, ReaderPid, ChPid, LimiterPid, add_consumer( ChPid, Consumer, State1#q.blocked_consumers)}; - false -> run_poke_burst( + false -> run_message_queue( State1#q{ active_consumers = add_consumer( @@ -681,11 +722,10 @@ handle_call({basic_cancel, ChPid, ConsumerTag, OkMsg}, _From, reply(ok, State); C = #cr{consumer_count = ConsumerCount, limiter_pid = LimiterPid} -> store_ch_record(C#cr{consumer_count = ConsumerCount - 1}), - if ConsumerCount == 1 -> - ok = rabbit_limiter:unregister(LimiterPid, self()); - true -> - ok - end, + ok = case ConsumerCount of + 1 -> rabbit_limiter:unregister(LimiterPid, self()); + _ -> ok + end, ok = maybe_send_reply(ChPid, OkMsg), NewState = State#q{exclusive_consumer = cancel_holder(ChPid, @@ -704,14 +744,15 @@ handle_call({basic_cancel, ChPid, ConsumerTag, OkMsg}, _From, end; handle_call(stat, _From, State = #q{q = #amqqueue{name = Name}, - message_buffer = MessageBuffer, + variable_queue_state = VQS, active_consumers = ActiveConsumers}) -> - reply({ok, Name, queue:len(MessageBuffer), queue:len(ActiveConsumers)}, - State); + Length = rabbit_variable_queue:len(VQS), + reply({ok, Name, Length, queue:len(ActiveConsumers)}, State); handle_call({delete, IfUnused, IfEmpty}, _From, - State = #q{message_buffer = MessageBuffer}) -> - IsEmpty = queue:is_empty(MessageBuffer), + State = #q { variable_queue_state = VQS }) -> + Length = rabbit_variable_queue:len(VQS), + IsEmpty = Length == 0, IsUnused = is_unused(State), if IfEmpty and not(IsEmpty) -> @@ -719,16 +760,15 @@ handle_call({delete, IfUnused, IfEmpty}, _From, IfUnused and not(IsUnused) -> reply({error, in_use}, State); true -> - {stop, normal, {ok, queue:len(MessageBuffer)}, State} + {stop, normal, {ok, Length}, State} end; -handle_call(purge, _From, State = #q{message_buffer = MessageBuffer}) -> - ok = purge_message_buffer(qname(State), MessageBuffer), - reply({ok, queue:len(MessageBuffer)}, - State#q{message_buffer = queue:new()}); +handle_call(purge, _From, State) -> + {Count, VQS} = rabbit_variable_queue:purge(State #q.variable_queue_state), + reply({ok, Count}, State #q { variable_queue_state = VQS }); -handle_call({claim_queue, ReaderPid}, _From, State = #q{owner = Owner, - exclusive_consumer = Holder}) -> +handle_call({claim_queue, ReaderPid}, _From, + State = #q{owner = Owner, exclusive_consumer = Holder}) -> case Owner of none -> case check_exclusive_access(Holder, true, State) of @@ -741,7 +781,9 @@ handle_call({claim_queue, ReaderPid}, _From, State = #q{owner = Owner, %% pid... reply(locked, State); ok -> - reply(ok, State#q{owner = {ReaderPid, erlang:monitor(process, ReaderPid)}}) + reply(ok, + State#q{ owner = {ReaderPid, erlang:monitor( + process, ReaderPid)} }) end; {ReaderPid, _MonitorRef} -> reply(ok, State); @@ -759,24 +801,22 @@ handle_cast({ack, Txn, MsgIds, ChPid}, State) -> not_found -> noreply(State); C = #cr{unacked_messages = UAM} -> - {Acked, Remaining} = collect_messages(MsgIds, UAM), - persist_acks(Txn, qname(State), Acked), case Txn of none -> - store_ch_record(C#cr{unacked_messages = Remaining}); + {MsgWithAcks, Remaining} = collect_messages(MsgIds, UAM), + VQS = rabbit_variable_queue:ack( + [AckTag || {_Msg, AckTag} <- MsgWithAcks], + State #q.variable_queue_state), + store_ch_record(C#cr{unacked_messages = Remaining}), + noreply(State #q { variable_queue_state = VQS }); _ -> - record_pending_acks(Txn, ChPid, MsgIds) - end, - noreply(State) + record_pending_acks(Txn, ChPid, MsgIds), + noreply(State) + end end; handle_cast({rollback, Txn}, State) -> - ok = rollback_work(Txn, qname(State)), - erase_tx(Txn), - noreply(State); - -handle_cast({redeliver, Messages}, State) -> - noreply(deliver_or_enqueue_n(Messages, State)); + noreply(rollback_transaction(Txn, State)); handle_cast({requeue, MsgIds, ChPid}, State) -> case lookup_ch(ChPid) of @@ -785,10 +825,9 @@ handle_cast({requeue, MsgIds, ChPid}, State) -> [ChPid]), noreply(State); C = #cr{unacked_messages = UAM} -> - {Messages, NewUAM} = collect_messages(MsgIds, UAM), + {MsgWithAcks, NewUAM} = collect_messages(MsgIds, UAM), store_ch_record(C#cr{unacked_messages = NewUAM}), - noreply(deliver_or_enqueue_n( - [{Message, true} || Message <- Messages], State)) + noreply(deliver_or_requeue_n(MsgWithAcks, State)) end; handle_cast({unblock, ChPid}, State) -> @@ -803,6 +842,19 @@ handle_cast({notify_sent, ChPid}, State) -> C#cr{unsent_message_count = Count - 1} end)); +handle_cast({tx_commit_msg_store_callback, Pubs, AckTags, From}, + State = #q{variable_queue_state = VQS}) -> + noreply( + State#q{variable_queue_state = + rabbit_variable_queue:tx_commit_from_msg_store( + Pubs, AckTags, From, VQS)}); + +handle_cast(tx_commit_vq_callback, State = #q{variable_queue_state = VQS}) -> + noreply( + run_message_queue( + State#q{variable_queue_state = + rabbit_variable_queue:tx_commit_from_vq(VQS)})); + handle_cast({limit, ChPid, LimiterPid}, State) -> noreply( possibly_unblock( @@ -819,49 +871,10 @@ handle_cast({limit, ChPid, LimiterPid}, State) -> C#cr{limiter_pid = LimiterPid, is_limit_active = NewLimited} end)); -handle_cast(send_memory_monitor_update, State) -> - DrainRatio1 = update_ratio(State#q.drain_ratio, State#q.next_msg_id), - MsgSec = DrainRatio1#ratio.ratio * 1000000, % msg/sec - QueueDuration = - case MsgSec == 0 of - true -> infinity; - false -> queue:len(State#q.message_buffer) / MsgSec % seconds - end, - DesiredQueueDuration = rabbit_memory_monitor:report_queue_duration( - self(), QueueDuration), - ?LOGDEBUG("TIMER ~p Queue length is ~8p, should be ~p~n", - [(State#q.q)#amqqueue.name, queue:len(State#q.message_buffer), - case DesiredQueueDuration of - infinity -> infinity; - _ -> MsgSec * DesiredQueueDuration - end]), - noreply(State#q{drain_ratio = DrainRatio1}); - -handle_cast({set_queue_duration, DesiredQueueDuration}, State) -> - DrainRatio = State#q.drain_ratio, - DesiredBufLength = - case DesiredQueueDuration of - infinity -> infinity; - _ -> DesiredQueueDuration * DrainRatio#ratio.ratio * 1000000 - end, - ?LOGDEBUG("MAGIC ~p Queue length is ~8p, should be ~p~n", - [(State#q.q)#amqqueue.name, queue:len(State#q.message_buffer), - DesiredBufLength]), - noreply(State). - -%% Based on kernel load average, as descibed: -%% http://www.teamquest.com/resources/gunther/display/5/ -calc_load(Load, Exp, N) -> - Load*Exp + N*(1.0-Exp). - -update_ratio(_RatioRec = #ratio{ratio=Ratio, t0 = T0, next_msg_id = MsgCount0}, MsgCount1) -> - T1 = now(), - Td = timer:now_diff(T1, T0), - MsgCount = MsgCount1 - MsgCount0, - MsgUSec = MsgCount / Td, % msg/usec - %% Td is in usec. We're interested in "load average" from last 30 seconds. - Ratio1 = calc_load(Ratio, 1.0/ (math:exp(Td/(30*1000000))), MsgUSec), - #ratio{ratio = Ratio1, t0=T1, next_msg_id = MsgCount1}. +handle_cast(remeasure_egress_rate, State = #q{variable_queue_state = VQS}) -> + noreply(State#q{egress_rate_timer_ref = just_measured, + variable_queue_state = + rabbit_variable_queue:remeasure_egress_rate(VQS)}). handle_info({'DOWN', MonitorRef, process, DownPid, _Reason}, State = #q{owner = {DownPid, MonitorRef}}) -> @@ -882,6 +895,24 @@ handle_info({'DOWN', _MonitorRef, process, DownPid, _Reason}, State) -> {stop, NewState} -> {stop, normal, NewState} end; +handle_info(timeout, State = #q{variable_queue_state = VQS, + sync_timer_ref = undefined}) -> + %% if sync_timer_ref is undefined then we must have set the + %% timeout to zero because we thought we could flush the journal + noreply(State#q{variable_queue_state = + rabbit_variable_queue:flush_journal(VQS)}); + +handle_info(timeout, State = #q{variable_queue_state = VQS}) -> + noreply( + run_message_queue( + State#q{variable_queue_state = + rabbit_variable_queue:tx_commit_from_vq(VQS)})); + handle_info(Info, State) -> ?LOGDEBUG("Info in queue: ~p~n", [Info]), {stop, {unhandled_info, Info}, State}. + +handle_pre_hibernate(State = #q{ variable_queue_state = VQS }) -> + VQS1 = rabbit_variable_queue:maybe_start_prefetcher(VQS), + {hibernate, stop_egress_rate_timer( + State#q{ variable_queue_state = VQS1 })}. diff --git a/src/rabbit_amqqueue_sup.erl b/src/rabbit_amqqueue_sup.erl index 46d23a4075..f06e4c5380 100644 --- a/src/rabbit_amqqueue_sup.erl +++ b/src/rabbit_amqqueue_sup.erl @@ -43,6 +43,4 @@ start_link() -> supervisor:start_link({local, ?SERVER}, ?MODULE, []). init([]) -> - {ok, {{simple_one_for_one, 10, 10}, - [{rabbit_amqqueue, {rabbit_amqqueue_process, start_link, []}, - temporary, brutal_kill, worker, [rabbit_amqqueue_process]}]}}. + {ok, {{one_for_one, 10, 10}, []}}. diff --git a/src/rabbit_basic.erl b/src/rabbit_basic.erl index bec2cd0845..14c655a6d5 100644 --- a/src/rabbit_basic.erl +++ b/src/rabbit_basic.erl @@ -33,8 +33,8 @@ -include("rabbit.hrl"). -include("rabbit_framing.hrl"). --export([publish/1, message/4, properties/1, delivery/4]). --export([publish/4, publish/7]). +-export([publish/1, message/4, message/5, message/6, delivery/4]). +-export([properties/1, publish/4, publish/7]). -export([build_content/2, from_content/1]). %%---------------------------------------------------------------------------- @@ -49,6 +49,10 @@ delivery()). -spec(message/4 :: (exchange_name(), routing_key(), properties_input(), binary()) -> message()). +-spec(message/5 :: (exchange_name(), routing_key(), properties_input(), + binary(), guid()) -> message()). +-spec(message/6 :: (exchange_name(), routing_key(), properties_input(), + binary(), guid(), boolean()) -> message()). -spec(properties/1 :: (properties_input()) -> amqp_properties()). -spec(publish/4 :: (exchange_name(), routing_key(), properties_input(), binary()) -> publish_result()). @@ -92,11 +96,18 @@ from_content(Content) -> {Props, list_to_binary(lists:reverse(FragmentsRev))}. message(ExchangeName, RoutingKeyBin, RawProperties, BodyBin) -> + message(ExchangeName, RoutingKeyBin, RawProperties, BodyBin, rabbit_guid:guid()). + +message(ExchangeName, RoutingKeyBin, RawProperties, BodyBin, MsgId) -> + message(ExchangeName, RoutingKeyBin, RawProperties, BodyBin, MsgId, false). + +message(ExchangeName, RoutingKeyBin, RawProperties, BodyBin, MsgId, IsPersistent) -> Properties = properties(RawProperties), #basic_message{exchange_name = ExchangeName, routing_key = RoutingKeyBin, content = build_content(Properties, BodyBin), - persistent_key = none}. + guid = MsgId, + is_persistent = IsPersistent}. properties(P = #'P_basic'{}) -> P; diff --git a/src/rabbit_binary_generator.erl b/src/rabbit_binary_generator.erl index 6cfa9e6d17..0b68f33f41 100644 --- a/src/rabbit_binary_generator.erl +++ b/src/rabbit_binary_generator.erl @@ -46,6 +46,7 @@ build_heartbeat_frame/0]). -export([generate_table/1, encode_properties/2]). -export([check_empty_content_body_frame_size/0]). +-export([ensure_content_encoded/1, clear_encoded_content/1]). -import(lists). @@ -63,6 +64,8 @@ -spec(generate_table/1 :: (amqp_table()) -> binary()). -spec(encode_properties/2 :: ([amqp_property_type()], [any()]) -> binary()). -spec(check_empty_content_body_frame_size/0 :: () -> 'ok'). +-spec(ensure_content_encoded/1 :: (content()) -> encoded_content()). +-spec(clear_encoded_content/1 :: (content()) -> unencoded_content()). -endif. @@ -275,3 +278,19 @@ check_empty_content_body_frame_size() -> exit({incorrect_empty_content_body_frame_size, ComputedSize, ?EMPTY_CONTENT_BODY_FRAME_SIZE}) end. + +ensure_content_encoded(Content = #content{properties_bin = PropsBin}) + when PropsBin =/= 'none' -> + Content; +ensure_content_encoded(Content = #content{properties = Props}) -> + Content #content{properties_bin = rabbit_framing:encode_properties(Props)}. + +clear_encoded_content(Content = #content{properties_bin = none}) -> + Content; +clear_encoded_content(Content = #content{properties = none}) -> + %% Only clear when we can rebuild the properties later in + %% accordance to the content record definition comment - maximum + %% one of properties and properties_bin can be 'none' + Content; +clear_encoded_content(Content = #content{}) -> + Content#content{properties_bin = none}. diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl index c20cb16ca1..6afd0bc9a7 100644 --- a/src/rabbit_channel.erl +++ b/src/rabbit_channel.erl @@ -54,6 +54,9 @@ -ifdef(use_specs). +-type(msg_id() :: non_neg_integer()). +-type(msg() :: {queue_name(), pid(), msg_id(), boolean(), message()}). + -spec(start_link/5 :: (channel_number(), pid(), pid(), username(), vhost()) -> pid()). -spec(do/2 :: (pid(), amqp_method()) -> 'ok'). @@ -311,14 +314,11 @@ handle_method(#'basic.publish'{exchange = ExchangeNameBin, %% We decode the content's properties here because we're almost %% certain to want to look at delivery-mode and priority. DecodedContent = rabbit_binary_parser:ensure_content_decoded(Content), - PersistentKey = case is_message_persistent(DecodedContent) of - true -> rabbit_guid:guid(); - false -> none - end, Message = #basic_message{exchange_name = ExchangeName, routing_key = RoutingKey, content = DecodedContent, - persistent_key = PersistentKey}, + guid = rabbit_guid:guid(), + is_persistent = is_message_persistent(DecodedContent)}, {RoutingRes, DeliveredQPids} = rabbit_exchange:publish( Exchange, diff --git a/src/rabbit_control.erl b/src/rabbit_control.erl index 1957972990..5e6229b138 100644 --- a/src/rabbit_control.erl +++ b/src/rabbit_control.erl @@ -174,8 +174,8 @@ virtual host parameter for which to display results. The default value is \"/\". <QueueInfoItem> must be a member of the list [name, durable, auto_delete, arguments, node, messages_ready, messages_unacknowledged, messages_uncommitted, -messages, acks_uncommitted, consumers, transactions, memory]. The default is - to display name and (number of) messages. +messages, acks_uncommitted, consumers, transactions, memory, storage_mode]. The +default is to display name and (number of) messages. <ExchangeInfoItem> must be a member of the list [name, type, durable, auto_delete, arguments]. The default is to display name and type. @@ -187,7 +187,6 @@ exchange name, queue name, routing key and arguments, in that order. peer_address, peer_port, state, channels, user, vhost, timeout, frame_max, recv_oct, recv_cnt, send_oct, send_cnt, send_pend]. The default is to display user, peer_address, peer_port and state. - "), halt(1). diff --git a/src/rabbit_guid.erl b/src/rabbit_guid.erl index ea61a679e8..dee21c3421 100644 --- a/src/rabbit_guid.erl +++ b/src/rabbit_guid.erl @@ -67,7 +67,7 @@ update_disk_serial() -> Filename = filename:join(rabbit_mnesia:dir(), ?SERIAL_FILENAME), Serial = case rabbit_misc:read_term_file(Filename) of {ok, [Num]} -> Num; - {error, enoent} -> rabbit_persister:serial(); + {error, enoent} -> 0; {error, Reason} -> throw({error, {cannot_read_serial_file, Filename, Reason}}) end, @@ -99,13 +99,12 @@ guid() -> {S, I} -> {S, I+1} end, put(guid, G), - G. + erlang:md5(term_to_binary(G)). %% generate a readable string representation of a guid. Note that any %% monotonicity of the guid is not preserved in the encoding. string_guid(Prefix) -> - Prefix ++ "-" ++ base64:encode_to_string( - erlang:md5(term_to_binary(guid()))). + Prefix ++ "-" ++ base64:encode_to_string(guid()). binstring_guid(Prefix) -> list_to_binary(string_guid(Prefix)). diff --git a/src/rabbit_memory_manager.erl b/src/rabbit_memory_manager.erl new file mode 100644 index 0000000000..a73f03e27d --- /dev/null +++ b/src/rabbit_memory_manager.erl @@ -0,0 +1,404 @@ +%% The contents of this file are subject to the Mozilla Public License +%% Version 1.1 (the "License"); you may not use this file except in +%% compliance with the License. You may obtain a copy of the License at +%% http://www.mozilla.org/MPL/ +%% +%% 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 Original Code is RabbitMQ. +%% +%% The Initial Developers of the Original Code are LShift Ltd, +%% Cohesive Financial Technologies LLC, and Rabbit Technologies Ltd. +%% +%% Portions created before 22-Nov-2008 00:00:00 GMT by LShift Ltd, +%% Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd +%% are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial +%% Technologies LLC, and Rabbit Technologies Ltd. +%% +%% Portions created by LShift Ltd are Copyright (C) 2007-2009 LShift +%% Ltd. Portions created by Cohesive Financial Technologies LLC are +%% Copyright (C) 2007-2009 Cohesive Financial Technologies +%% LLC. Portions created by Rabbit Technologies Ltd are Copyright +%% (C) 2007-2009 Rabbit Technologies Ltd. +%% +%% All Rights Reserved. +%% +%% Contributor(s): ______________________________________. +%% + +-module(rabbit_memory_manager). + +-behaviour(gen_server2). + +-export([start_link/0]). + +-export([init/1, handle_call/3, handle_cast/2, handle_info/2, + terminate/2, code_change/3]). + +-export([register/5, report_memory/3, info/0, conserve_memory/2]). + +-define(TOTAL_TOKENS, 10000000). +-define(THRESHOLD_MULTIPLIER, 0.05). +-define(THRESHOLD_OFFSET, ?TOTAL_TOKENS * ?THRESHOLD_MULTIPLIER). + +-define(SERVER, ?MODULE). + +%%---------------------------------------------------------------------------- + +-ifdef(use_specs). + +-spec(start_link/0 :: () -> + ({'ok', pid()} | 'ignore' | {'error', any()})). +-spec(register/5 :: (pid(), boolean(), atom(), atom(), list()) -> 'ok'). +-spec(report_memory/3 :: (pid(), non_neg_integer(), boolean()) -> 'ok'). +-spec(info/0 :: () -> [{atom(), any()}]). +-spec(conserve_memory/2 :: (pid(), boolean()) -> 'ok'). + +-endif. + +%%---------------------------------------------------------------------------- + +-record(state, { available_tokens, + processes, + callbacks, + tokens_per_byte, + hibernate, + unoppressable, + alarmed + }). + +%% Token-credit based memory management + +%% Start off by working out the amount of memory available in the +%% system (RAM). Then, work out how many tokens each byte corresponds +%% to. This is the tokens_per_byte field. When a process registers, it +%% must provide an M-F-A triple to a function that needs one further +%% argument, which is the new mode. This will either be 'liberated' or +%% 'oppressed'. +%% +%% Processes then report their own memory usage, in bytes, and the +%% manager takes care of the rest. +%% +%% There are a finite number of tokens in the system. These are +%% allocated to processes as the processes report their memory +%% usage. We keep track of processes which have hibernated. When a +%% process reports memory use which can't be satisfied by the +%% available tokens, we try and oppress processes first from the +%% hibernated group. The hibernated group is a simple queue, and so is +%% implicitly sorted by the order in which processes were added to the +%% queue. This means that when removing from the queue, we evict the +%% sleepiest (and most passive) pid first. +%% +%% If the reported memory use still can't be satisfied after +%% oppressing everyone from those two groups (and note that we check +%% first whether or not oppressing them would make available enough +%% tokens to satisfy the reported use rather than just oppressing all +%% those processes and then going "whoops, didn't help after all"), +%% then we oppress the reporting process. When a process registers, it +%% can declare itself "unoppressable". If a process is unoppressable +%% then it will not be oppressed as a result of other processes +%% needing more tokens. However, if it itself needs additional tokens +%% which aren't available then it is still oppressed as before. This +%% feature is only used by the disk_queue, because if the disk queue +%% is not being used, and hibernates, and then memory pressure gets +%% tight, the disk_queue would typically be one of the first processes +%% to be oppressed (sent to disk_only mode), which cripples +%% performance. Thus by setting it unoppressable, it is only possible +%% for the disk_queue to be oppressed when it is active and attempting +%% to increase its memory allocation. +%% +%% If a process has been oppressed, it continues making memory +%% reports, as if it was liberated. As soon as a reported amount of +%% memory can be satisfied (and this can include oppressing other +%% processes in the way described above), *and* the number of +%% available tokens has changed by ?THRESHOLD_MULTIPLIER since the +%% processes was oppressed, it will be liberated. This later condition +%% prevents processes from continually oppressing each other if they +%% themselves can be liberated by oppressing other processes. +%% +%% Note that the hibernate group can get very out of date. This is +%% fine, and somewhat unavoidable given the absence of useful APIs for +%% queues. Thus we allow them to get out of date (processes will be +%% left in there when they change groups, duplicates can appear, dead +%% processes are not pruned etc etc etc), and when we go through the +%% groups, summing up their allocated tokens, we tidy up at that +%% point. +%% +%% A liberated process, which is reporting a smaller amount of RAM +%% than its last report will remain liberated. A liberated process +%% that is busy but consuming an unchanging amount of RAM will never +%% be oppressed. + +%% Specific notes as applied to queues and the disk_queue: +%% +%% The disk_queue is managed in the same way as queues. This means +%% that a queue that has gone back to mixed mode after being in disk +%% mode now has its messages counted twice as they are counted both in +%% the report made by the queue (even though they may not yet be in +%% RAM (though see the prefetcher)) and also by the disk_queue. Thus +%% the amount of available RAM must be higher when going disk -> mixed +%% than when going mixed -> disk. This is fairly sensible as it +%% reduces the risk of any oscillations occurring. +%% +%% The queue process deliberately reports 4 times its estimated RAM +%% usage, and the disk_queue 2.5 times. In practise, this seems to +%% work well. Note that we are deliberately running out of tokes a +%% little early because of the fact that the mixed -> disk transition +%% can transiently eat a lot of memory and take some time (flushing a +%% few million messages to disk is never going to be instantaneous). + +start_link() -> + gen_server2:start_link({local, ?SERVER}, ?MODULE, [], []). + +register(Pid, Unoppressable, Module, Function, Args) -> + gen_server2:cast(?SERVER, {register, Pid, Unoppressable, + Module, Function, Args}). + +report_memory(Pid, Memory, Hibernating) -> + gen_server2:cast(?SERVER, {report_memory, Pid, Memory, Hibernating}). + +info() -> + gen_server2:call(?SERVER, info). + +conserve_memory(_Pid, Conserve) -> + gen_server2:pcast(?SERVER, 9, {conserve_memory, Conserve}). + +%%---------------------------------------------------------------------------- + +init([]) -> + process_flag(trap_exit, true), + rabbit_alarm:register(self(), {?MODULE, conserve_memory, []}), + {MemTotal, MemUsed, _BigProc} = memsup:get_memory_data(), + MemAvail = MemTotal - MemUsed, + TPB = if MemAvail == 0 -> 0; + true -> ?TOTAL_TOKENS / MemAvail + end, + {ok, #state { available_tokens = ?TOTAL_TOKENS, + processes = dict:new(), + callbacks = dict:new(), + tokens_per_byte = TPB, + hibernate = queue:new(), + unoppressable = sets:new(), + alarmed = false + }}. + +handle_call(info, _From, State) -> + State1 = #state { available_tokens = Avail, + processes = Procs, + hibernate = Sleepy, + unoppressable = Unoppressable } = + free_upto(undefined, 1 + ?TOTAL_TOKENS, State), %% just tidy + {reply, [{ available_tokens, Avail }, + { processes, dict:to_list(Procs) }, + { hibernated_processes, queue:to_list(Sleepy) }, + { unoppressable_processes, sets:to_list(Unoppressable) }], State1}. + +handle_cast({report_memory, Pid, Memory, Hibernating}, + State = #state { processes = Procs, + available_tokens = Avail, + callbacks = Callbacks, + tokens_per_byte = TPB, + alarmed = Alarmed }) -> + Req = rabbit_misc:ceil(TPB * Memory), + LibreActivity = if Hibernating -> hibernate; + true -> active + end, + {StateN = #state { hibernate = Sleepy }, ActivityNew} = + case find_process(Pid, Procs) of + {libre, OAlloc, _OActivity} -> + Avail1 = Avail + OAlloc, + State1 = #state { available_tokens = Avail2, + processes = Procs1 } + = free_upto(Pid, Req, + State #state { available_tokens = Avail1 }), + case Req > Avail2 of + true -> %% nowt we can do, oppress the process + Procs2 = + set_process_mode(Procs1, Callbacks, Pid, oppressed, + {oppressed, Avail2}), + {State1 #state { processes = Procs2 }, oppressed}; + false -> %% keep liberated + {State1 #state + { processes = + dict:store(Pid, {libre, Req, LibreActivity}, Procs1), + available_tokens = Avail2 - Req }, + LibreActivity} + end; + {oppressed, OrigAvail} -> + case Req > 0 andalso + ( Alarmed orelse Hibernating orelse + (Avail > (OrigAvail - ?THRESHOLD_OFFSET) andalso + Avail < (OrigAvail + ?THRESHOLD_OFFSET)) ) of + true -> + {State, oppressed}; + false -> + State1 = #state { available_tokens = Avail1, + processes = Procs1 } = + free_upto(Pid, Req, State), + case Req > Avail1 of + true -> + %% not enough space, so stay oppressed + {State1, oppressed}; + false -> %% can liberate the process + Procs2 = set_process_mode( + Procs1, Callbacks, Pid, liberated, + {libre, Req, LibreActivity}), + {State1 #state { + processes = Procs2, + available_tokens = Avail1 - Req }, + LibreActivity} + end + end + end, + StateN1 = + case ActivityNew of + active -> StateN; + oppressed -> StateN; + hibernate -> + StateN #state { hibernate = queue:in(Pid, Sleepy) } + end, + {noreply, StateN1}; + +handle_cast({register, Pid, IsUnoppressable, Module, Function, Args}, + State = #state { callbacks = Callbacks, + unoppressable = Unoppressable }) -> + _MRef = erlang:monitor(process, Pid), + Unoppressable1 = case IsUnoppressable of + true -> sets:add_element(Pid, Unoppressable); + false -> Unoppressable + end, + {noreply, State #state { callbacks = dict:store + (Pid, {Module, Function, Args}, Callbacks), + unoppressable = Unoppressable1 + }}; + +handle_cast({conserve_memory, Conserve}, State) -> + {noreply, State #state { alarmed = Conserve }}. + +handle_info({'DOWN', _MRef, process, Pid, _Reason}, + State = #state { available_tokens = Avail, + processes = Procs, + callbacks = Callbacks }) -> + State1 = State #state { processes = dict:erase(Pid, Procs), + callbacks = dict:erase(Pid, Callbacks) }, + {noreply, case find_process(Pid, Procs) of + {oppressed, _OrigReq} -> + State1; + {libre, Alloc, _Activity} -> + State1 #state { available_tokens = Avail + Alloc } + end}; +handle_info({'EXIT', _Pid, Reason}, State) -> + {stop, Reason, State}; +handle_info(_Info, State) -> + {noreply, State}. + +terminate(_Reason, State) -> + State. + +code_change(_OldVsn, State, _Extra) -> + {ok, State}. + +%%---------------------------------------------------------------------------- + +find_process(Pid, Procs) -> + case dict:find(Pid, Procs) of + {ok, Value} -> Value; + error -> {oppressed, 0} + end. + +set_process_mode(Procs, Callbacks, Pid, Mode, Record) -> + {Module, Function, Args} = dict:fetch(Pid, Callbacks), + ok = erlang:apply(Module, Function, Args ++ [Mode]), + dict:store(Pid, Record, Procs). + +tidy_and_sum_sleepy(IgnorePids, Sleepy, Procs) -> + tidy_and_sum(hibernate, Procs, fun queue:out/1, + fun (Pid, _Alloc, Queue) -> queue:in(Pid, Queue) end, + IgnorePids, Sleepy, queue:new(), 0). + +tidy_and_sum(AtomExpected, Procs, Generator, Consumer, DupCheckSet, + GenInit, ConInit, AllocAcc) -> + case Generator(GenInit) of + {empty, _GetInit} -> {ConInit, AllocAcc}; + {{value, Pid}, GenInit1} -> + {DupCheckSet1, ConInit1, AllocAcc1} = + case sets:is_element(Pid, DupCheckSet) of + true -> + {DupCheckSet, ConInit, AllocAcc}; + false -> + case find_process(Pid, Procs) of + {libre, Alloc, AtomExpected} -> + {sets:add_element(Pid, DupCheckSet), + Consumer(Pid, Alloc, ConInit), + Alloc + AllocAcc}; + _ -> + {DupCheckSet, ConInit, AllocAcc} + end + end, + tidy_and_sum(AtomExpected, Procs, Generator, Consumer, + DupCheckSet1, GenInit1, ConInit1, AllocAcc1) + end. + +free_upto_sleepy(IgnorePids, Callbacks, Sleepy, Procs, Req, Avail) -> + free_from(Callbacks, + fun(Procs1, Sleepy1, SleepyAcc) -> + case queue:out(Sleepy1) of + {empty, _Sleepy2} -> + empty; + {{value, Pid}, Sleepy2} -> + case sets:is_element(Pid, IgnorePids) of + true -> {skip, Sleepy2, + queue:in(Pid, SleepyAcc)}; + false -> {libre, Alloc, hibernate} = + dict:fetch(Pid, Procs1), + {value, Sleepy2, Pid, Alloc} + end + end + end, fun queue:join/2, Procs, Sleepy, queue:new(), Req, Avail). + +free_from( + Callbacks, Transformer, BaseCase, Procs, DestroyMe, CreateMe, Req, Avail) -> + case Transformer(Procs, DestroyMe, CreateMe) of + empty -> + {CreateMe, Procs, Req}; + {skip, DestroyMe1, CreateMe1} -> + free_from(Callbacks, Transformer, BaseCase, Procs, DestroyMe1, + CreateMe1, Req, Avail); + {value, DestroyMe1, Pid, Alloc} -> + Procs1 = set_process_mode( + Procs, Callbacks, Pid, oppressed, {oppressed, Avail}), + Req1 = Req - Alloc, + case Req1 > 0 of + true -> free_from(Callbacks, Transformer, BaseCase, Procs1, + DestroyMe1, CreateMe, Req1, Avail); + false -> {BaseCase(DestroyMe1, CreateMe), Procs1, Req1} + end + end. + +free_upto(Pid, Req, State = #state { available_tokens = Avail, + processes = Procs, + callbacks = Callbacks, + hibernate = Sleepy, + unoppressable = Unoppressable }) + when Req > Avail -> + Unoppressable1 = sets:add_element(Pid, Unoppressable), + {Sleepy1, SleepySum} = tidy_and_sum_sleepy(Unoppressable1, Sleepy, Procs), + case Req > Avail + SleepySum of + true -> %% not enough in sleepy, just return tidied state + State #state { hibernate = Sleepy1 }; + false -> + %% ReqRem will be <= 0 because it's likely we'll have + %% freed more than we need, thus Req - ReqRem is total + %% freed + {Sleepy2, Procs1, ReqRem} = + free_upto_sleepy(Unoppressable1, Callbacks, + Sleepy1, Procs, Req, Avail), + State #state { available_tokens = Avail + (Req - ReqRem), + processes = Procs1, + hibernate = Sleepy2 } + end; +free_upto(_Pid, _Req, State) -> + State. diff --git a/src/rabbit_mnesia.erl b/src/rabbit_mnesia.erl index 749038dbb1..a1d886bb57 100644 --- a/src/rabbit_mnesia.erl +++ b/src/rabbit_mnesia.erl @@ -162,7 +162,13 @@ table_definitions() -> {disc_copies, [node()]}]}, {rabbit_queue, [{record_name, amqqueue}, - {attributes, record_info(fields, amqqueue)}]}]. + {attributes, record_info(fields, amqqueue)}]}, + {rabbit_disk_queue, + [{record_name, dq_msg_loc}, + {attributes, record_info(fields, dq_msg_loc)}, + {disc_copies, [node()]}, + {local_content, true}]} + ]. table_names() -> [Tab || {Tab, _} <- table_definitions()]. @@ -196,7 +202,8 @@ ensure_mnesia_not_running() -> check_schema_integrity() -> %%TODO: more thorough checks - case catch [mnesia:table_info(Tab, version) || Tab <- table_names()] of + case catch [mnesia:table_info(Tab, version) + || Tab <- table_names()] of {'EXIT', Reason} -> {error, Reason}; _ -> ok end. diff --git a/src/rabbit_msg_file.erl b/src/rabbit_msg_file.erl new file mode 100644 index 0000000000..c08261591d --- /dev/null +++ b/src/rabbit_msg_file.erl @@ -0,0 +1,141 @@ +%% The contents of this file are subject to the Mozilla Public License +%% Version 1.1 (the "License"); you may not use this file except in +%% compliance with the License. You may obtain a copy of the License at +%% http://www.mozilla.org/MPL/ +%% +%% 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 Original Code is RabbitMQ. +%% +%% The Initial Developers of the Original Code are LShift Ltd, +%% Cohesive Financial Technologies LLC, and Rabbit Technologies Ltd. +%% +%% Portions created before 22-Nov-2008 00:00:00 GMT by LShift Ltd, +%% Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd +%% are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial +%% Technologies LLC, and Rabbit Technologies Ltd. +%% +%% Portions created by LShift Ltd are Copyright (C) 2007-2009 LShift +%% Ltd. Portions created by Cohesive Financial Technologies LLC are +%% Copyright (C) 2007-2009 Cohesive Financial Technologies +%% LLC. Portions created by Rabbit Technologies Ltd are Copyright +%% (C) 2007-2009 Rabbit Technologies Ltd. +%% +%% All Rights Reserved. +%% +%% Contributor(s): ______________________________________. +%% + +-module(rabbit_msg_file). + +-export([append/3, read/2, scan/1]). + +%%---------------------------------------------------------------------------- + +-define(INTEGER_SIZE_BYTES, 8). +-define(INTEGER_SIZE_BITS, (8 * ?INTEGER_SIZE_BYTES)). +-define(WRITE_OK_SIZE_BITS, 8). +-define(WRITE_OK_MARKER, 255). +-define(FILE_PACKING_ADJUSTMENT, (1 + ?INTEGER_SIZE_BYTES)). +-define(MSG_ID_SIZE_BYTES, 16). +-define(MSG_ID_SIZE_BITS, (8 * ?MSG_ID_SIZE_BYTES)). +-define(SIZE_AND_MSG_ID_BYTES, (?MSG_ID_SIZE_BYTES + ?INTEGER_SIZE_BYTES)). + +%%---------------------------------------------------------------------------- + +-ifdef(use_specs). + +-type(io_device() :: any()). +-type(msg_id() :: binary()). +-type(msg() :: any()). +-type(position() :: non_neg_integer()). +-type(msg_size() :: non_neg_integer()). + +-spec(append/3 :: (io_device(), msg_id(), msg()) -> + ({'ok', msg_size()} | {'error', any()})). +-spec(read/2 :: (io_device(), msg_size()) -> + ({'ok', {msg_id(), msg()}} | {'error', any()})). +-spec(scan/1 :: (io_device()) -> + {'ok', [{msg_id(), msg_size(), position()}]}). + +-endif. + +%%---------------------------------------------------------------------------- + +append(FileHdl, MsgId, MsgBody) + when is_binary(MsgId) andalso size(MsgId) =< ?MSG_ID_SIZE_BYTES -> + MsgBodyBin = term_to_binary(MsgBody), + MsgBodyBinSize = size(MsgBodyBin), + Size = MsgBodyBinSize + ?MSG_ID_SIZE_BYTES, + case file_handle_cache:append(FileHdl, + <<Size:?INTEGER_SIZE_BITS, + MsgId:?MSG_ID_SIZE_BYTES/binary, + MsgBodyBin:MsgBodyBinSize/binary, + ?WRITE_OK_MARKER:?WRITE_OK_SIZE_BITS>>) of + ok -> {ok, Size + ?FILE_PACKING_ADJUSTMENT}; + KO -> KO + end. + +read(FileHdl, TotalSize) -> + Size = TotalSize - ?FILE_PACKING_ADJUSTMENT, + BodyBinSize = Size - ?MSG_ID_SIZE_BYTES, + case file_handle_cache:read(FileHdl, TotalSize) of + {ok, <<Size:?INTEGER_SIZE_BITS, + MsgId:?MSG_ID_SIZE_BYTES/binary, + MsgBodyBin:BodyBinSize/binary, + ?WRITE_OK_MARKER:?WRITE_OK_SIZE_BITS>>} -> + {ok, {MsgId, binary_to_term(MsgBodyBin)}}; + KO -> KO + end. + +scan(FileHdl) -> scan(FileHdl, 0, []). + +scan(FileHdl, Offset, Acc) -> + case read_next(FileHdl, Offset) of + eof -> {ok, Acc}; + {corrupted, NextOffset} -> + scan(FileHdl, NextOffset, Acc); + {ok, {MsgId, TotalSize, NextOffset}} -> + scan(FileHdl, NextOffset, [{MsgId, TotalSize, Offset} | Acc]); + _KO -> + %% bad message, but we may still have recovered some valid messages + {ok, Acc} + end. + +read_next(FileHdl, Offset) -> + case file_handle_cache:read(FileHdl, ?SIZE_AND_MSG_ID_BYTES) of + %% Here we take option 5 from + %% http://www.erlang.org/cgi-bin/ezmlm-cgi?2:mss:1569 in which + %% we read the MsgId as a number, and then convert it back to + %% a binary in order to work around bugs in Erlang's GC. + {ok, <<Size:?INTEGER_SIZE_BITS, MsgIdNum:?MSG_ID_SIZE_BITS>>} -> + case Size of + 0 -> eof; %% Nothing we can do other than stop + _ -> + TotalSize = Size + ?FILE_PACKING_ADJUSTMENT, + ExpectedAbsPos = Offset + TotalSize - 1, + case file_handle_cache:position( + FileHdl, {cur, Size - ?MSG_ID_SIZE_BYTES}) of + {ok, ExpectedAbsPos} -> + NextOffset = ExpectedAbsPos + 1, + case file_handle_cache:read(FileHdl, 1) of + {ok, + <<?WRITE_OK_MARKER: ?WRITE_OK_SIZE_BITS>>} -> + <<MsgId:?MSG_ID_SIZE_BYTES/binary>> = + <<MsgIdNum:?MSG_ID_SIZE_BITS>>, + {ok, {MsgId, TotalSize, NextOffset}}; + {ok, _SomeOtherData} -> + {corrupted, NextOffset}; + KO -> KO + end; + {ok, _SomeOtherPos} -> + %% seek failed, so give up + eof; + KO -> KO + end + end; + Other -> Other + end. diff --git a/src/rabbit_msg_store.erl b/src/rabbit_msg_store.erl new file mode 100644 index 0000000000..591435ba01 --- /dev/null +++ b/src/rabbit_msg_store.erl @@ -0,0 +1,1080 @@ +%% The contents of this file are subject to the Mozilla Public License +%% Version 1.1 (the "License"); you may not use this file except in +%% compliance with the License. You may obtain a copy of the License at +%% http://www.mozilla.org/MPL/ +%% +%% 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 Original Code is RabbitMQ. +%% +%% The Initial Developers of the Original Code are LShift Ltd, +%% Cohesive Financial Technologies LLC, and Rabbit Technologies Ltd. +%% +%% Portions created before 22-Nov-2008 00:00:00 GMT by LShift Ltd, +%% Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd +%% are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial +%% Technologies LLC, and Rabbit Technologies Ltd. +%% +%% Portions created by LShift Ltd are Copyright (C) 2007-2009 LShift +%% Ltd. Portions created by Cohesive Financial Technologies LLC are +%% Copyright (C) 2007-2009 Cohesive Financial Technologies +%% LLC. Portions created by Rabbit Technologies Ltd are Copyright +%% (C) 2007-2009 Rabbit Technologies Ltd. +%% +%% All Rights Reserved. +%% +%% Contributor(s): ______________________________________. +%% + +-module(rabbit_msg_store). + +-behaviour(gen_server2). + +-export([start_link/3, write/2, read/1, peruse/2, contains/1, remove/1, + release/1, sync/2]). + +-export([sync/0]). %% internal + +-export([init/1, handle_call/3, handle_cast/2, handle_info/2, + terminate/2, code_change/3]). + +-define(SERVER, ?MODULE). + +-define(MAX_READ_FILE_HANDLES, 256). +-define(FILE_SIZE_LIMIT, (256*1024*1024)). +-define(SYNC_INTERVAL, 5). %% milliseconds +-define(HANDLE_CACHE_BUFFER_SIZE, 1048576). %% 1MB + +%%---------------------------------------------------------------------------- + +-ifdef(use_specs). + +-type(msg_id() :: binary()). +-type(msg() :: any()). +-type(file_path() :: any()). + +-spec(start_link/3 :: + (file_path(), + (fun ((A) -> 'finished' | {msg_id(), non_neg_integer(), A})), A) -> + {'ok', pid()} | 'ignore' | {'error', any()}). +-spec(write/2 :: (msg_id(), msg()) -> 'ok'). +-spec(read/1 :: (msg_id()) -> {'ok', msg()} | 'not_found'). +-spec(peruse/2 :: (msg_id(), fun (({'ok', msg()} | 'not_found') -> 'ok')) -> + 'ok'). +-spec(contains/1 :: (msg_id()) -> boolean()). +-spec(remove/1 :: ([msg_id()]) -> 'ok'). +-spec(release/1 :: ([msg_id()]) -> 'ok'). +-spec(sync/2 :: ([msg_id()], fun (() -> any())) -> 'ok'). + +-endif. + +%%---------------------------------------------------------------------------- + +-record(msstate, + {dir, %% store directory + msg_locations, %% where are messages? + file_summary, %% what's in the files? + current_file, %% current file name as number + current_file_handle, %% current file handle + %% since the last fsync? + file_size_limit, %% how big can our files get? + file_handle_cache, %% file handle cache + on_sync, %% pending sync requests + sync_timer_ref, %% TRef for our interval timer + message_cache %% ets message cache + }). + +-record(msg_location, + {msg_id, ref_count, file, offset, total_size}). + +-record(file_summary, + {file, valid_total_size, contiguous_top, left, right}). + +-define(MSG_LOC_NAME, rabbit_disk_queue_msg_location). +-define(FILE_SUMMARY_ETS_NAME, rabbit_disk_queue_file_summary). +-define(FILE_EXTENSION, ".rdq"). +-define(FILE_EXTENSION_TMP, ".rdt"). +-define(CACHE_ETS_NAME, rabbit_disk_queue_cache). + +-define(BINARY_MODE, [raw, binary]). +-define(READ_MODE, [read]). +-define(READ_AHEAD_MODE, [read_ahead | ?READ_MODE]). +-define(WRITE_MODE, [write]). + +%% The components: +%% +%% MsgLocation: this is an ets table which contains: +%% {MsgId, RefCount, File, Offset, TotalSize} +%% FileSummary: this is an ets table which contains: +%% {File, ValidTotalSize, ContiguousTop, Left, Right} +%% +%% The basic idea is that messages are appended to the current file up +%% until that file becomes too big (> file_size_limit). At that point, +%% the file is closed and a new file is created on the _right_ of the +%% old file which is used for new messages. Files are named +%% numerically ascending, thus the file with the lowest name is the +%% eldest file. +%% +%% We need to keep track of which messages are in which files (this is +%% the MsgLocation table); how much useful data is in each file and +%% which files are on the left and right of each other. This is the +%% purpose of the FileSummary table. +%% +%% As messages are removed from files, holes appear in these +%% files. The field ValidTotalSize contains the total amount of useful +%% data left in the file, whilst ContiguousTop contains the amount of +%% valid data right at the start of each file. These are needed for +%% garbage collection. +%% +%% When we discover that either a file is now empty or that it can be +%% combined with the useful data in either its left or right file, we +%% compact the two files together. This keeps disk utilisation high +%% and aids performance. +%% +%% Given the compaction between two files, the left file is considered +%% the ultimate destination for the good data in the right file. If +%% necessary, the good data in the left file which is fragmented +%% throughout the file is written out to a temporary file, then read +%% back in to form a contiguous chunk of good data at the start of the +%% left file. Thus the left file is garbage collected and +%% compacted. Then the good data from the right file is copied onto +%% the end of the left file. MsgLocation and FileSummary tables are +%% updated. +%% +%% On startup, we scan the files we discover, dealing with the +%% possibilites of a crash have occured during a compaction (this +%% consists of tidyup - the compaction is deliberately designed such +%% that data is duplicated on disk rather than risking it being lost), +%% and rebuild the ets tables (MsgLocation, FileSummary). +%% +%% So, with this design, messages move to the left. Eventually, they +%% should end up in a contiguous block on the left and are then never +%% rewritten. But this isn't quite the case. If in a file there is one +%% message that is being ignored, for some reason, and messages in the +%% file to the right and in the current block are being read all the +%% time then it will repeatedly be the case that the good data from +%% both files can be combined and will be written out to a new +%% file. Whenever this happens, our shunned message will be rewritten. +%% +%% So, provided that we combine messages in the right order, +%% (i.e. left file, bottom to top, right file, bottom to top), +%% eventually our shunned message will end up at the bottom of the +%% left file. The compaction/combining algorithm is smart enough to +%% read in good data from the left file that is scattered throughout +%% (i.e. C and D in the below diagram), then truncate the file to just +%% above B (i.e. truncate to the limit of the good contiguous region +%% at the start of the file), then write C and D on top and then write +%% E, F and G from the right file on top. Thus contiguous blocks of +%% good data at the bottom of files are not rewritten (yes, this is +%% the data the size of which is tracked by the ContiguousTop +%% variable. Judicious use of a mirror is required). +%% +%% +-------+ +-------+ +-------+ +%% | X | | G | | G | +%% +-------+ +-------+ +-------+ +%% | D | | X | | F | +%% +-------+ +-------+ +-------+ +%% | X | | X | | E | +%% +-------+ +-------+ +-------+ +%% | C | | F | ===> | D | +%% +-------+ +-------+ +-------+ +%% | X | | X | | C | +%% +-------+ +-------+ +-------+ +%% | B | | X | | B | +%% +-------+ +-------+ +-------+ +%% | A | | E | | A | +%% +-------+ +-------+ +-------+ +%% left right left +%% +%% From this reasoning, we do have a bound on the number of times the +%% message is rewritten. From when it is inserted, there can be no +%% files inserted between it and the head of the queue, and the worst +%% case is that everytime it is rewritten, it moves one position lower +%% in the file (for it to stay at the same position requires that +%% there are no holes beneath it, which means truncate would be used +%% and so it would not be rewritten at all). Thus this seems to +%% suggest the limit is the number of messages ahead of it in the +%% queue, though it's likely that that's pessimistic, given the +%% requirements for compaction/combination of files. +%% +%% The other property is that we have is the bound on the lowest +%% utilisation, which should be 50% - worst case is that all files are +%% fractionally over half full and can't be combined (equivalent is +%% alternating full files and files with only one tiny message in +%% them). +%% +%% Messages are reference-counted. When a message with the same id is +%% written several times we only store it once, and only remove it +%% from the store when it has been removed the same number of times. +%% +%% The reference counts do not persist. Therefore the initialisation +%% function must be provided with a generator that produces ref count +%% deltas for all recovered messages. +%% +%% Read messages with a reference count greater than one are entered +%% into a message cache. The purpose of the cache is not especially +%% performance, though it can help there too, but prevention of memory +%% explosion. It ensures that as messages with a high reference count +%% are read from several processes they are read back as the same +%% binary object rather than multiples of identical binary +%% objects. + +%%---------------------------------------------------------------------------- +%% public API +%%---------------------------------------------------------------------------- + +start_link(Dir, MsgRefDeltaGen, MsgRefDeltaGenInit) -> + gen_server2:start_link({local, ?SERVER}, ?MODULE, + [Dir, MsgRefDeltaGen, MsgRefDeltaGenInit], + [{timeout, infinity}]). + +write(MsgId, Msg) -> gen_server2:cast(?SERVER, {write, MsgId, Msg}). +read(MsgId) -> gen_server2:call(?SERVER, {read, MsgId}, infinity). +peruse(MsgId, Fun) -> gen_server2:pcast(?SERVER, -1, {peruse, MsgId, Fun}). +contains(MsgId) -> gen_server2:call(?SERVER, {contains, MsgId}, infinity). +remove(MsgIds) -> gen_server2:cast(?SERVER, {remove, MsgIds}). +release(MsgIds) -> gen_server2:cast(?SERVER, {release, MsgIds}). +sync(MsgIds, K) -> gen_server2:cast(?SERVER, {sync, MsgIds, K}). +sync() -> gen_server2:pcast(?SERVER, 9, sync). %% internal + +%%---------------------------------------------------------------------------- +%% gen_server callbacks +%%---------------------------------------------------------------------------- + +init([Dir, MsgRefDeltaGen, MsgRefDeltaGenInit]) -> + process_flag(trap_exit, true), + + ok = filelib:ensure_dir(filename:join(Dir, "nothing")), + + MsgLocations = ets:new(?MSG_LOC_NAME, + [set, private, {keypos, #msg_location.msg_id}]), + + InitFile = 0, + FileSummary = ets:new(?FILE_SUMMARY_ETS_NAME, + [set, private, {keypos, #file_summary.file}]), + MessageCache = ets:new(?CACHE_ETS_NAME, [set, private]), + State = + #msstate { dir = Dir, + msg_locations = MsgLocations, + file_summary = FileSummary, + current_file = InitFile, + current_file_handle = undefined, + file_size_limit = ?FILE_SIZE_LIMIT, + file_handle_cache = dict:new(), + on_sync = [], + sync_timer_ref = undefined, + message_cache = MessageCache + }, + + ok = count_msg_refs(MsgRefDeltaGen, MsgRefDeltaGenInit, State), + FileNames = + sort_file_names(filelib:wildcard("*" ++ ?FILE_EXTENSION, Dir)), + TmpFileNames = + sort_file_names(filelib:wildcard("*" ++ ?FILE_EXTENSION_TMP, Dir)), + ok = recover_crashed_compactions(Dir, FileNames, TmpFileNames), + %% There should be no more tmp files now, so go ahead and load the + %% whole lot + Files = [filename_to_num(FileName) || FileName <- FileNames], + {Offset, State1 = #msstate { current_file = CurFile }} = + build_index(Files, State), + + %% read is only needed so that we can seek + {ok, FileHdl} = open_file(Dir, filenum_to_name(CurFile), + [read | ?WRITE_MODE]), + {ok, Offset} = file_handle_cache:position(FileHdl, Offset), + ok = file_handle_cache:truncate(FileHdl), + + {ok, State1 #msstate { current_file_handle = FileHdl }}. + +handle_call({read, MsgId}, _From, State) -> + {Result, State1} = internal_read_message(MsgId, State), + reply(Result, State1); + +handle_call({contains, MsgId}, _From, State) -> + reply(case index_lookup(MsgId, State) of + not_found -> false; + #msg_location {} -> true + end, State). + +handle_cast({write, MsgId, Msg}, + State = #msstate { current_file_handle = CurHdl, + current_file = CurFile, + file_summary = FileSummary }) -> + case index_lookup(MsgId, State) of + not_found -> + %% New message, lots to do + {ok, CurOffset} = file_handle_cache:current_virtual_offset(CurHdl), + {ok, TotalSize} = rabbit_msg_file:append(CurHdl, MsgId, Msg), + ok = index_insert(#msg_location { + msg_id = MsgId, ref_count = 1, file = CurFile, + offset = CurOffset, total_size = TotalSize }, + State), + [FSEntry = #file_summary { valid_total_size = ValidTotalSize, + contiguous_top = ContiguousTop, + right = undefined }] = + ets:lookup(FileSummary, CurFile), + ValidTotalSize1 = ValidTotalSize + TotalSize, + ContiguousTop1 = if CurOffset =:= ContiguousTop -> + %% can't be any holes in this file + ValidTotalSize1; + true -> ContiguousTop + end, + true = ets:insert(FileSummary, FSEntry #file_summary { + valid_total_size = ValidTotalSize1, + contiguous_top = ContiguousTop1 }), + NextOffset = CurOffset + TotalSize, + noreply(maybe_roll_to_new_file(NextOffset, State)); + StoreEntry = #msg_location { ref_count = RefCount } -> + %% We already know about it, just update counter + ok = index_update(StoreEntry #msg_location { + ref_count = RefCount + 1 }, State), + noreply(State) + end; + +handle_cast({peruse, MsgId, Fun}, State) -> + {Result, State1} = internal_read_message(MsgId, State), + Fun(Result), + noreply(State1); + +handle_cast({remove, MsgIds}, State = #msstate { current_file = CurFile }) -> + noreply( + compact(sets:to_list( + lists:foldl( + fun (MsgId, Files1) -> + case remove_message(MsgId, State) of + {compact, File} -> + if CurFile =:= File -> Files1; + true -> sets:add_element(File, Files1) + end; + no_compact -> Files1 + end + end, sets:new(), MsgIds)), + State)); + +handle_cast({release, MsgIds}, State) -> + lists:foreach(fun (MsgId) -> decrement_cache(MsgId, State) end, MsgIds), + noreply(State); + +handle_cast({sync, MsgIds, K}, + State = #msstate { current_file = CurFile, + current_file_handle = CurHdl, + on_sync = Syncs }) -> + {ok, SyncOffset} = file_handle_cache:last_sync_offset(CurHdl), + case lists:any(fun (MsgId) -> + #msg_location { file = File, offset = Offset } = + index_lookup(MsgId, State), + File =:= CurFile andalso Offset >= SyncOffset + end, MsgIds) of + false -> K(), + noreply(State); + true -> noreply(State #msstate { on_sync = [K | Syncs] }) + end; + +handle_cast(sync, State) -> + noreply(sync(State)). + +handle_info(timeout, State) -> + noreply(sync(State)). + +terminate(_Reason, State = #msstate { msg_locations = MsgLocations, + file_summary = FileSummary, + current_file_handle = FileHdl }) -> + State1 = case FileHdl of + undefined -> State; + _ -> State2 = sync(State), + file_handle_cache:close(FileHdl), + State2 + end, + State3 = close_all_handles(State1), + ets:delete(MsgLocations), + ets:delete(FileSummary), + State3 #msstate { msg_locations = undefined, + file_summary = undefined, + current_file_handle = undefined }. + +code_change(_OldVsn, State, _Extra) -> + {ok, State}. + +%%---------------------------------------------------------------------------- +%% general helper functions +%%---------------------------------------------------------------------------- + +noreply(State) -> + {State1, Timeout} = next_state(State), + {noreply, State1, Timeout}. + +reply(Reply, State) -> + {State1, Timeout} = next_state(State), + {reply, Reply, State1, Timeout}. + +next_state(State = #msstate { on_sync = [], sync_timer_ref = undefined }) -> + {State, infinity}; +next_state(State = #msstate { sync_timer_ref = undefined }) -> + {start_sync_timer(State), 0}; +next_state(State = #msstate { on_sync = [] }) -> + {stop_sync_timer(State), infinity}; +next_state(State) -> + {State, 0}. + +start_sync_timer(State = #msstate { sync_timer_ref = undefined }) -> + {ok, TRef} = timer:apply_after(?SYNC_INTERVAL, ?MODULE, sync, []), + State #msstate { sync_timer_ref = TRef }. + +stop_sync_timer(State = #msstate { sync_timer_ref = undefined }) -> + State; +stop_sync_timer(State = #msstate { sync_timer_ref = TRef }) -> + {ok, cancel} = timer:cancel(TRef), + State #msstate { sync_timer_ref = undefined }. + +form_filename(Dir, Name) -> filename:join(Dir, Name). + +filenum_to_name(File) -> integer_to_list(File) ++ ?FILE_EXTENSION. + +filename_to_num(FileName) -> list_to_integer(filename:rootname(FileName)). + +sort_file_names(FileNames) -> + lists:sort(fun (A, B) -> filename_to_num(A) < filename_to_num(B) end, + FileNames). + +preallocate(Hdl, FileSizeLimit, FinalPos) -> + {ok, FileSizeLimit} = file_handle_cache:position(Hdl, FileSizeLimit), + ok = file_handle_cache:truncate(Hdl), + {ok, FinalPos} = file_handle_cache:position(Hdl, FinalPos), + ok. + +truncate_and_extend_file(FileHdl, Lowpoint, Highpoint) -> + {ok, Lowpoint} = file_handle_cache:position(FileHdl, Lowpoint), + ok = file_handle_cache:truncate(FileHdl), + ok = preallocate(FileHdl, Highpoint, Lowpoint). + +sync(State = #msstate { current_file_handle = CurHdl, + on_sync = Syncs }) -> + State1 = stop_sync_timer(State), + case Syncs of + [] -> State1; + _ -> + ok = file_handle_cache:sync(CurHdl), + lists:foreach(fun (K) -> K() end, lists:reverse(Syncs)), + State1 #msstate { on_sync = [] } + end. + +remove_message(MsgId, State = #msstate { file_summary = FileSummary }) -> + StoreEntry = #msg_location { ref_count = RefCount, file = File, + offset = Offset, total_size = TotalSize } = + index_lookup(MsgId, State), + case RefCount of + 1 -> + ok = index_delete(MsgId, State), + ok = remove_cache_entry(MsgId, State), + [FSEntry = #file_summary { valid_total_size = ValidTotalSize, + contiguous_top = ContiguousTop }] = + ets:lookup(FileSummary, File), + ContiguousTop1 = lists:min([ContiguousTop, Offset]), + ValidTotalSize1 = ValidTotalSize - TotalSize, + true = ets:insert(FileSummary, FSEntry #file_summary { + valid_total_size = ValidTotalSize1, + contiguous_top = ContiguousTop1 }), + {compact, File}; + _ when 1 < RefCount -> + ok = decrement_cache(MsgId, State), + ok = index_update(StoreEntry #msg_location { + ref_count = RefCount - 1 }, State), + no_compact + end. + +internal_read_message(MsgId, + State = #msstate { current_file = CurFile, + current_file_handle = CurHdl }) -> + case index_lookup(MsgId, State) of + not_found -> {not_found, State}; + #msg_location { ref_count = RefCount, + file = File, + offset = Offset, + total_size = TotalSize } -> + case fetch_and_increment_cache(MsgId, State) of + not_found -> + {ok, CurOffset} = + file_handle_cache:current_raw_offset(CurHdl), + ok = case CurFile =:= File andalso Offset >= CurOffset of + true -> + file_handle_cache:append_write_buffer(CurHdl); + false -> + ok + end, + {Hdl, State1} = get_read_handle(File, State), + {ok, Offset} = file_handle_cache:position(Hdl, Offset), + {ok, {MsgId, Msg}} = + case rabbit_msg_file:read(Hdl, TotalSize) of + {ok, {MsgId, _}} = Obj -> Obj; + Rest -> + throw({error, {misread, [{old_state, State}, + {file_num, File}, + {offset, Offset}, + {read, Rest}, + {proc_dict, get()}]}}) + end, + ok = if RefCount > 1 -> + insert_into_cache(MsgId, Msg, State1); + true -> ok + %% it's not in the cache and we + %% only have one reference to the + %% message. So don't bother + %% putting it in the cache. + end, + {{ok, Msg}, State1}; + {Msg, _RefCount} -> + {{ok, Msg}, State} + end + end. + +close_handle(Key, State = #msstate { file_handle_cache = FHC }) -> + case dict:find(Key, FHC) of + {ok, Hdl} -> + ok = close_file(Hdl), + State #msstate { file_handle_cache = dict:erase(Key, FHC) }; + error -> State + end. + +close_all_handles(State = #msstate { file_handle_cache = FHC }) -> + ok = dict:fold(fun (_Key, Hdl, ok) -> + file_handle_cache:close(Hdl) + end, ok, FHC), + State #msstate { file_handle_cache = dict:new() }. + +get_read_handle(FileNum, State = #msstate { file_handle_cache = FHC }) -> + case dict:find(FileNum, FHC) of + {ok, Hdl} -> {Hdl, State}; + error -> new_handle(FileNum, filenum_to_name(FileNum), + [read | ?BINARY_MODE], State) + end. + +new_handle(Key, FileName, Mode, State = #msstate { file_handle_cache = FHC, + dir = Dir }) -> + {ok, Hdl} = open_file(Dir, FileName, Mode), + {Hdl, State #msstate { file_handle_cache = dict:store(Key, Hdl, FHC) }}. + +open_file(Dir, FileName, Mode) -> + file_handle_cache:open(form_filename(Dir, FileName), ?BINARY_MODE ++ Mode, + [{write_buffer, ?HANDLE_CACHE_BUFFER_SIZE}]). + +close_file(Hdl) -> + file_handle_cache:close(Hdl). + +%%---------------------------------------------------------------------------- +%% message cache helper functions +%%---------------------------------------------------------------------------- + +remove_cache_entry(MsgId, #msstate { message_cache = Cache }) -> + true = ets:delete(Cache, MsgId), + ok. + +fetch_and_increment_cache(MsgId, #msstate { message_cache = Cache }) -> + case ets:lookup(Cache, MsgId) of + [] -> + not_found; + [{MsgId, Msg, _RefCount}] -> + NewRefCount = ets:update_counter(Cache, MsgId, {3, 1}), + {Msg, NewRefCount} + end. + +decrement_cache(MsgId, #msstate { message_cache = Cache }) -> + true = try case ets:update_counter(Cache, MsgId, {3, -1}) of + N when N =< 0 -> true = ets:delete(Cache, MsgId); + _N -> true + end + catch error:badarg -> + %% MsgId is not in there because although it's been + %% delivered, it's never actually been read (think: + %% persistent message in mixed queue) + true + end, + ok. + +insert_into_cache(MsgId, Msg, #msstate { message_cache = Cache }) -> + true = ets:insert_new(Cache, {MsgId, Msg, 1}), + ok. + +%%---------------------------------------------------------------------------- +%% index +%%---------------------------------------------------------------------------- + +index_lookup(Key, #msstate { msg_locations = MsgLocations }) -> + case ets:lookup(MsgLocations, Key) of + [] -> not_found; + [Entry] -> Entry + end. + +index_insert(Obj, #msstate { msg_locations = MsgLocations }) -> + true = ets:insert_new(MsgLocations, Obj), + ok. + +index_update(Obj, #msstate { msg_locations = MsgLocations }) -> + true = ets:insert(MsgLocations, Obj), + ok. + +index_delete(Key, #msstate { msg_locations = MsgLocations }) -> + true = ets:delete(MsgLocations, Key), + ok. + +index_search_by_file(File, #msstate { msg_locations = MsgLocations }) -> + lists:sort(fun (#msg_location { offset = OffA }, + #msg_location { offset = OffB }) -> + OffA < OffB + end, ets:match_object(MsgLocations, + #msg_location { file = File, _ = '_' })). + + +index_delete_by_file(File, #msstate { msg_locations = MsgLocations }) -> + MatchHead = #msg_location { file = File, _ = '_' }, + ets:select_delete(MsgLocations, [{MatchHead, [], [true]}]), + ok. + +%%---------------------------------------------------------------------------- +%% recovery +%%---------------------------------------------------------------------------- + +count_msg_refs(Gen, Seed, State) -> + case Gen(Seed) of + finished -> ok; + {_MsgId, 0, Next} -> count_msg_refs(Gen, Next, State); + {MsgId, Delta, Next} -> + ok = case index_lookup(MsgId, State) of + not_found -> + index_insert(#msg_location { msg_id = MsgId, + ref_count = Delta }, + State); + StoreEntry = #msg_location { ref_count = RefCount } -> + NewRefCount = RefCount + Delta, + case NewRefCount of + 0 -> index_delete(MsgId, State); + _ -> index_update(StoreEntry #msg_location { + ref_count = NewRefCount }, + State) + end + end, + count_msg_refs(Gen, Next, State) + end. + +recover_crashed_compactions(Dir, FileNames, TmpFileNames) -> + lists:foreach(fun (TmpFileName) -> + ok = recover_crashed_compactions1( + Dir, FileNames, TmpFileName) + end, TmpFileNames), + ok. + +recover_crashed_compactions1(Dir, FileNames, TmpFileName) -> + NonTmpRelatedFileName = filename:rootname(TmpFileName) ++ ?FILE_EXTENSION, + true = lists:member(NonTmpRelatedFileName, FileNames), + {ok, UncorruptedMessagesTmp, MsgIdsTmp} = + scan_file_for_valid_messages_msg_ids(Dir, TmpFileName), + {ok, UncorruptedMessages, MsgIds} = + scan_file_for_valid_messages_msg_ids(Dir, NonTmpRelatedFileName), + %% 1) It's possible that everything in the tmp file is also in the + %% main file such that the main file is (prefix ++ + %% tmpfile). This means that compaction failed immediately + %% prior to the final step of deleting the tmp file. Plan: just + %% delete the tmp file + %% 2) It's possible that everything in the tmp file is also in the + %% main file but with holes throughout (or just somthing like + %% main = (prefix ++ hole ++ tmpfile)). This means that + %% compaction wrote out the tmp file successfully and then + %% failed. Plan: just delete the tmp file and allow the + %% compaction to eventually be triggered later + %% 3) It's possible that everything in the tmp file is also in the + %% main file but such that the main file does not end with tmp + %% file (and there are valid messages in the suffix; main = + %% (prefix ++ tmpfile[with extra holes?] ++ suffix)). This + %% means that compaction failed as we were writing out the tmp + %% file. Plan: just delete the tmp file and allow the + %% compaction to eventually be triggered later + %% 4) It's possible that there are messages in the tmp file which + %% are not in the main file. This means that writing out the + %% tmp file succeeded, but then we failed as we were copying + %% them back over to the main file, after truncating the main + %% file. As the main file has already been truncated, it should + %% consist only of valid messages. Plan: Truncate the main file + %% back to before any of the files in the tmp file and copy + %% them over again + TmpPath = form_filename(Dir, TmpFileName), + case is_sublist(MsgIdsTmp, MsgIds) of + true -> %% we're in case 1, 2 or 3 above. Just delete the tmp file + %% note this also catches the case when the tmp file + %% is empty + ok = file:delete(TmpPath); + false -> + %% We're in case 4 above. We only care about the inital + %% msgs in main file that are not in the tmp file. If + %% there are no msgs in the tmp file then we would be in + %% the 'true' branch of this case, so we know the + %% lists:last call is safe. + EldestTmpMsgId = lists:last(MsgIdsTmp), + {MsgIds1, UncorruptedMessages1} + = case lists:splitwith( + fun (MsgId) -> MsgId /= EldestTmpMsgId end, MsgIds) of + {_MsgIds, []} -> %% no msgs from tmp in main + {MsgIds, UncorruptedMessages}; + {Dropped, [EldestTmpMsgId | Rest]} -> + %% Msgs in Dropped are in tmp, so forget them. + %% *cry*. Lists indexed from 1. + {Rest, lists:sublist(UncorruptedMessages, + 2 + length(Dropped), + length(Rest))} + end, + %% The main file prefix should be contiguous + {Top, MsgIds1} = find_contiguous_block_prefix( + lists:reverse(UncorruptedMessages1)), + %% we should have that none of the messages in the prefix + %% are in the tmp file + true = is_disjoint(MsgIds1, MsgIdsTmp), + %% must open with read flag, otherwise will stomp over contents + {ok, MainHdl} = open_file(Dir, NonTmpRelatedFileName, + [read | ?WRITE_MODE]), + %% Wipe out any rubbish at the end of the file. Remember + %% the head of the list will be the highest entry in the + %% file. + [{_, TmpTopTotalSize, TmpTopOffset}|_] = UncorruptedMessagesTmp, + TmpSize = TmpTopOffset + TmpTopTotalSize, + %% Extend the main file as big as necessary in a single + %% move. If we run out of disk space, this truncate could + %% fail, but we still aren't risking losing data + ok = truncate_and_extend_file(MainHdl, Top, Top + TmpSize), + {ok, TmpHdl} = open_file(Dir, TmpFileName, ?READ_AHEAD_MODE), + {ok, TmpSize} = file_handle_cache:copy(TmpHdl, MainHdl, TmpSize), + ok = file_handle_cache:close(MainHdl), + ok = file_handle_cache:close(TmpHdl), + ok = file:delete(TmpPath), + + {ok, _MainMessages, MsgIdsMain} = + scan_file_for_valid_messages_msg_ids( + Dir, NonTmpRelatedFileName), + %% check that everything in MsgIds1 is in MsgIdsMain + true = is_sublist(MsgIds1, MsgIdsMain), + %% check that everything in MsgIdsTmp is in MsgIdsMain + true = is_sublist(MsgIdsTmp, MsgIdsMain) + end, + ok. + +is_sublist(SmallerL, BiggerL) -> + lists:all(fun (Item) -> lists:member(Item, BiggerL) end, SmallerL). + +is_disjoint(SmallerL, BiggerL) -> + lists:all(fun (Item) -> not lists:member(Item, BiggerL) end, SmallerL). + +scan_file_for_valid_messages_msg_ids(Dir, FileName) -> + {ok, Messages} = scan_file_for_valid_messages(Dir, FileName), + {ok, Messages, [MsgId || {MsgId, _TotalSize, _FileOffset} <- Messages]}. + +scan_file_for_valid_messages(Dir, FileName) -> + case open_file(Dir, FileName, ?READ_MODE) of + {ok, Hdl} -> + Valid = rabbit_msg_file:scan(Hdl), + %% if something really bad's happened, the close could fail, + %% but ignore + file_handle_cache:close(Hdl), + Valid; + {error, enoent} -> {ok, []}; + {error, Reason} -> throw({error, + {unable_to_scan_file, FileName, Reason}}) + end. + +%% Takes the list in *ascending* order (i.e. eldest message +%% first). This is the opposite of what scan_file_for_valid_messages +%% produces. The list of msgs that is produced is youngest first. +find_contiguous_block_prefix([]) -> {0, []}; +find_contiguous_block_prefix(List) -> + find_contiguous_block_prefix(List, 0, []). + +find_contiguous_block_prefix([], ExpectedOffset, MsgIds) -> + {ExpectedOffset, MsgIds}; +find_contiguous_block_prefix([{MsgId, TotalSize, ExpectedOffset} | Tail], + ExpectedOffset, MsgIds) -> + ExpectedOffset1 = ExpectedOffset + TotalSize, + find_contiguous_block_prefix(Tail, ExpectedOffset1, [MsgId | MsgIds]); +find_contiguous_block_prefix([_MsgAfterGap | _Tail], ExpectedOffset, MsgIds) -> + {ExpectedOffset, MsgIds}. + +build_index([], State) -> + CurFile = State #msstate.current_file, + build_index(undefined, [CurFile], [], State); +build_index(Files, State) -> + build_index(undefined, Files, [], State). + +build_index(Left, [], FilesToCompact, State) -> + ok = index_delete_by_file(undefined, State), + Offset = case lists:reverse(index_search_by_file(Left, State)) of + [] -> 0; + [#msg_location { offset = MaxOffset, + total_size = TotalSize } | _] -> + MaxOffset + TotalSize + end, + {Offset, compact(FilesToCompact, %% this never includes the current file + State #msstate { current_file = Left })}; +build_index(Left, [File|Files], FilesToCompact, + State = #msstate { dir = Dir, file_summary = FileSummary }) -> + {ok, Messages} = scan_file_for_valid_messages(Dir, filenum_to_name(File)), + {ValidMessages, ValidTotalSize, AllValid} = + lists:foldl( + fun (Obj = {MsgId, TotalSize, Offset}, + {VMAcc, VTSAcc, AVAcc}) -> + case index_lookup(MsgId, State) of + not_found -> {VMAcc, VTSAcc, false}; + StoreEntry -> + ok = index_update(StoreEntry #msg_location { + file = File, offset = Offset, + total_size = TotalSize }, + State), + {[Obj | VMAcc], VTSAcc + TotalSize, AVAcc} + end + end, {[], 0, Messages =/= []}, Messages), + %% foldl reverses lists, find_contiguous_block_prefix needs + %% msgs eldest first, so, ValidMessages is the right way round + {ContiguousTop, _} = find_contiguous_block_prefix(ValidMessages), + Right = case Files of + [] -> undefined; + [F|_] -> F + end, + true = ets:insert_new(FileSummary, #file_summary { + file = File, valid_total_size = ValidTotalSize, + contiguous_top = ContiguousTop, + left = Left, right = Right }), + FilesToCompact1 = case AllValid orelse Right =:= undefined of + true -> FilesToCompact; + false -> [File | FilesToCompact] + end, + build_index(File, Files, FilesToCompact1, State). + +%%---------------------------------------------------------------------------- +%% garbage collection / compaction / aggregation +%%---------------------------------------------------------------------------- + +maybe_roll_to_new_file(Offset, + State = #msstate { dir = Dir, + file_size_limit = FileSizeLimit, + current_file_handle = CurHdl, + current_file = CurFile, + file_summary = FileSummary }) + when Offset >= FileSizeLimit -> + State1 = sync(State), + ok = close_file(CurHdl), + NextFile = CurFile + 1, + {ok, NextHdl} = open_file(Dir, filenum_to_name(NextFile), ?WRITE_MODE), + true = ets:update_element(FileSummary, CurFile, + {#file_summary.right, NextFile}), + true = ets:insert_new( + FileSummary, #file_summary { + file = NextFile, valid_total_size = 0, contiguous_top = 0, + left = CurFile, right = undefined }), + State2 = State1 #msstate { current_file_handle = NextHdl, + current_file = NextFile }, + compact([CurFile], State2); +maybe_roll_to_new_file(_, State) -> + State. + +compact(Files, State) -> + %% smallest number, hence eldest, hence left-most, first + SortedFiles = lists:sort(Files), + %% foldl reverses, so now youngest/right-most first + RemainingFiles = + lists:foldl(fun (File, Acc) -> + case delete_file_if_empty(File, State) of + true -> Acc; + false -> [File | Acc] + end + end, [], SortedFiles), + lists:foldl(fun combine_file/2, State, lists:reverse(RemainingFiles)). + +%% At this stage, we simply know that the file has had msgs removed +%% from it. However, we don't know if we need to merge it left (which +%% is what we would prefer), or merge it right. If we merge left, then +%% this file is the source, and the left file is the destination. If +%% we merge right then this file is the destination and the right file +%% is the source. +combine_file(File, State = #msstate { file_summary = FileSummary, + current_file = CurFile }) -> + %% the file we're looking at may no longer exist as it may have + %% been deleted within the current GC run + case ets:lookup(FileSummary, File) of + [] -> State; + [FSEntry = #file_summary { left = Left, right = Right }] -> + GoRight = + fun() -> + case Right of + undefined -> State; + _ when not (CurFile == Right) -> + [FSRight] = ets:lookup(FileSummary, Right), + {_, State1} = adjust_meta_and_combine( + FSEntry, FSRight, State), + State1; + _ -> State + end + end, + case Left of + undefined -> + GoRight(); + _ -> [FSLeft] = ets:lookup(FileSummary, Left), + case adjust_meta_and_combine(FSLeft, FSEntry, State) of + {true, State1} -> State1; + {false, State} -> GoRight() + end + end + end. + +adjust_meta_and_combine( + LeftObj = #file_summary { + file = LeftFile, valid_total_size = LeftValidData, right = RightFile }, + RightObj = #file_summary { + file = RightFile, valid_total_size = RightValidData, left = LeftFile, + right = RightRight }, + State = #msstate { file_size_limit = FileSizeLimit, + file_summary = FileSummary }) -> + TotalValidData = LeftValidData + RightValidData, + if FileSizeLimit >= TotalValidData -> + State1 = combine_files(RightObj, LeftObj, State), + %% this could fail if RightRight is undefined + ets:update_element(FileSummary, RightRight, + {#file_summary.left, LeftFile}), + true = ets:insert(FileSummary, LeftObj #file_summary { + valid_total_size = TotalValidData, + contiguous_top = TotalValidData, + right = RightRight }), + true = ets:delete(FileSummary, RightFile), + {true, State1}; + true -> {false, State} + end. + +combine_files(#file_summary { file = Source, + valid_total_size = SourceValid, + left = Destination }, + #file_summary { file = Destination, + valid_total_size = DestinationValid, + contiguous_top = DestinationContiguousTop, + right = Source }, + State = #msstate { dir = Dir }) -> + State1 = close_handle(Source, close_handle(Destination, State)), + SourceName = filenum_to_name(Source), + DestinationName = filenum_to_name(Destination), + {ok, SourceHdl} = open_file(Dir, SourceName, ?READ_AHEAD_MODE), + {ok, DestinationHdl} = open_file(Dir, DestinationName, + ?READ_AHEAD_MODE ++ ?WRITE_MODE), + ExpectedSize = SourceValid + DestinationValid, + %% if DestinationValid =:= DestinationContiguousTop then we don't + %% need a tmp file + %% if they're not equal, then we need to write out everything past + %% the DestinationContiguousTop to a tmp file then truncate, + %% copy back in, and then copy over from Source + %% otherwise we just truncate straight away and copy over from Source + if DestinationContiguousTop =:= DestinationValid -> + ok = truncate_and_extend_file(DestinationHdl, + DestinationValid, ExpectedSize); + true -> + Tmp = filename:rootname(DestinationName) ++ ?FILE_EXTENSION_TMP, + {ok, TmpHdl} = open_file(Dir, Tmp, ?READ_AHEAD_MODE ++ ?WRITE_MODE), + Worklist = + lists:dropwhile( + fun (#msg_location { offset = Offset }) + when Offset /= DestinationContiguousTop -> + %% it cannot be that Offset == + %% DestinationContiguousTop because if it + %% was then DestinationContiguousTop would + %% have been extended by TotalSize + Offset < DestinationContiguousTop + %% Given expected access patterns, I suspect + %% that the list should be naturally sorted + %% as we require, however, we need to + %% enforce it anyway + end, index_search_by_file(Destination, State1)), + ok = copy_messages( + Worklist, DestinationContiguousTop, DestinationValid, + DestinationHdl, TmpHdl, Destination, State1), + TmpSize = DestinationValid - DestinationContiguousTop, + %% so now Tmp contains everything we need to salvage from + %% Destination, and MsgLocationDets has been updated to + %% reflect compaction of Destination so truncate + %% Destination and copy from Tmp back to the end + {ok, 0} = file_handle_cache:position(TmpHdl, 0), + ok = truncate_and_extend_file( + DestinationHdl, DestinationContiguousTop, ExpectedSize), + {ok, TmpSize} = + file_handle_cache:copy(TmpHdl, DestinationHdl, TmpSize), + %% position in DestinationHdl should now be DestinationValid + ok = file_handle_cache:sync(DestinationHdl), + ok = close_file(TmpHdl), + ok = file:delete(form_filename(Dir, Tmp)) + end, + SourceWorkList = index_search_by_file(Source, State1), + ok = copy_messages(SourceWorkList, DestinationValid, ExpectedSize, + SourceHdl, DestinationHdl, Destination, State1), + %% tidy up + ok = close_file(SourceHdl), + ok = close_file(DestinationHdl), + ok = file:delete(form_filename(Dir, SourceName)), + State1. + +copy_messages(WorkList, InitOffset, FinalOffset, SourceHdl, DestinationHdl, + Destination, State) -> + {FinalOffset, BlockStart1, BlockEnd1} = + lists:foldl( + fun (StoreEntry = #msg_location { offset = Offset, + total_size = TotalSize }, + {CurOffset, BlockStart, BlockEnd}) -> + %% CurOffset is in the DestinationFile. + %% Offset, BlockStart and BlockEnd are in the SourceFile + %% update MsgLocationDets to reflect change of file and offset + ok = index_update(StoreEntry #msg_location { + file = Destination, + offset = CurOffset }, State), + NextOffset = CurOffset + TotalSize, + if BlockStart =:= undefined -> + %% base case, called only for the first list elem + {NextOffset, Offset, Offset + TotalSize}; + Offset =:= BlockEnd -> + %% extend the current block because the next + %% msg follows straight on + {NextOffset, BlockStart, BlockEnd + TotalSize}; + true -> + %% found a gap, so actually do the work for + %% the previous block + BSize = BlockEnd - BlockStart, + {ok, BlockStart} = + file_handle_cache:position(SourceHdl, BlockStart), + {ok, BSize} = file_handle_cache:copy( + SourceHdl, DestinationHdl, BSize), + {NextOffset, Offset, Offset + TotalSize} + end + end, {InitOffset, undefined, undefined}, WorkList), + %% do the last remaining block + BSize1 = BlockEnd1 - BlockStart1, + {ok, BlockStart1} = file_handle_cache:position(SourceHdl, BlockStart1), + {ok, BSize1} = file_handle_cache:copy(SourceHdl, DestinationHdl, BSize1), + ok = file_handle_cache:sync(DestinationHdl), + ok. + +delete_file_if_empty(File, + #msstate { dir = Dir, file_summary = FileSummary }) -> + [#file_summary { valid_total_size = ValidData, + left = Left, right = Right }] = + ets:lookup(FileSummary, File), + case ValidData of + %% we should NEVER find the current file in here hence right + %% should always be a file, not undefined + 0 -> case {Left, Right} of + {undefined, _} when not is_atom(Right) -> + %% the eldest file is empty. + true = ets:update_element( + FileSummary, Right, + {#file_summary.left, undefined}); + {_, _} when not (is_atom(Right)) -> + true = ets:update_element(FileSummary, Right, + {#file_summary.left, Left}), + true = + ets:update_element(FileSummary, Left, + {#file_summary.right, Right}) + end, + true = ets:delete(FileSummary, File), + ok = file:delete(form_filename(Dir, filenum_to_name(File))), + true; + _ -> false + end. diff --git a/src/rabbit_persister.erl b/src/rabbit_persister.erl deleted file mode 100644 index d0d60ddf3d..0000000000 --- a/src/rabbit_persister.erl +++ /dev/null @@ -1,523 +0,0 @@ -%% The contents of this file are subject to the Mozilla Public License -%% Version 1.1 (the "License"); you may not use this file except in -%% compliance with the License. You may obtain a copy of the License at -%% http://www.mozilla.org/MPL/ -%% -%% 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 Original Code is RabbitMQ. -%% -%% The Initial Developers of the Original Code are LShift Ltd, -%% Cohesive Financial Technologies LLC, and Rabbit Technologies Ltd. -%% -%% Portions created before 22-Nov-2008 00:00:00 GMT by LShift Ltd, -%% Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd -%% are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial -%% Technologies LLC, and Rabbit Technologies Ltd. -%% -%% Portions created by LShift Ltd are Copyright (C) 2007-2009 LShift -%% Ltd. Portions created by Cohesive Financial Technologies LLC are -%% Copyright (C) 2007-2009 Cohesive Financial Technologies -%% LLC. Portions created by Rabbit Technologies Ltd are Copyright -%% (C) 2007-2009 Rabbit Technologies Ltd. -%% -%% All Rights Reserved. -%% -%% Contributor(s): ______________________________________. -%% - --module(rabbit_persister). - --behaviour(gen_server). - --export([start_link/0]). - --export([init/1, handle_call/3, handle_cast/2, handle_info/2, - terminate/2, code_change/3]). - --export([transaction/1, extend_transaction/2, dirty_work/1, - commit_transaction/1, rollback_transaction/1, - force_snapshot/0, serial/0]). - --include("rabbit.hrl"). - --define(SERVER, ?MODULE). - --define(LOG_BUNDLE_DELAY, 5). --define(COMPLETE_BUNDLE_DELAY, 2). - --define(HIBERNATE_AFTER, 10000). - --define(MAX_WRAP_ENTRIES, 500). - --define(PERSISTER_LOG_FORMAT_VERSION, {2, 4}). - --record(pstate, {log_handle, entry_count, deadline, - pending_logs, pending_replies, - snapshot}). - -%% two tables for efficient persistency -%% one maps a key to a message -%% the other maps a key to one or more queues. -%% The aim is to reduce the overload of storing a message multiple times -%% when it appears in several queues. --record(psnapshot, {serial, transactions, messages, queues}). - -%%---------------------------------------------------------------------------- - --ifdef(use_specs). - --type(qmsg() :: {amqqueue(), pkey()}). --type(work_item() :: - {publish, message(), qmsg()} | - {deliver, qmsg()} | - {ack, qmsg()}). - --spec(start_link/0 :: () -> {'ok', pid()} | 'ignore' | {'error', any()}). --spec(transaction/1 :: ([work_item()]) -> 'ok'). --spec(extend_transaction/2 :: (txn(), [work_item()]) -> 'ok'). --spec(dirty_work/1 :: ([work_item()]) -> 'ok'). --spec(commit_transaction/1 :: (txn()) -> 'ok'). --spec(rollback_transaction/1 :: (txn()) -> 'ok'). --spec(force_snapshot/0 :: () -> 'ok'). --spec(serial/0 :: () -> non_neg_integer()). - --endif. - -%%---------------------------------------------------------------------------- - -start_link() -> - gen_server:start_link({local, ?SERVER}, ?MODULE, [], []). - -transaction(MessageList) -> - ?LOGDEBUG("transaction ~p~n", [MessageList]), - TxnKey = rabbit_guid:guid(), - gen_server:call(?SERVER, {transaction, TxnKey, MessageList}, infinity). - -extend_transaction(TxnKey, MessageList) -> - ?LOGDEBUG("extend_transaction ~p ~p~n", [TxnKey, MessageList]), - gen_server:cast(?SERVER, {extend_transaction, TxnKey, MessageList}). - -dirty_work(MessageList) -> - ?LOGDEBUG("dirty_work ~p~n", [MessageList]), - gen_server:cast(?SERVER, {dirty_work, MessageList}). - -commit_transaction(TxnKey) -> - ?LOGDEBUG("commit_transaction ~p~n", [TxnKey]), - gen_server:call(?SERVER, {commit_transaction, TxnKey}, infinity). - -rollback_transaction(TxnKey) -> - ?LOGDEBUG("rollback_transaction ~p~n", [TxnKey]), - gen_server:cast(?SERVER, {rollback_transaction, TxnKey}). - -force_snapshot() -> - gen_server:call(?SERVER, force_snapshot, infinity). - -serial() -> - gen_server:call(?SERVER, serial, infinity). - -%%-------------------------------------------------------------------- - -init(_Args) -> - process_flag(trap_exit, true), - FileName = base_filename(), - ok = filelib:ensure_dir(FileName), - Snapshot = #psnapshot{serial = 0, - transactions = dict:new(), - messages = ets:new(messages, []), - queues = ets:new(queues, [])}, - LogHandle = - case disk_log:open([{name, rabbit_persister}, - {head, current_snapshot(Snapshot)}, - {file, FileName}]) of - {ok, LH} -> LH; - {repaired, LH, {recovered, Recovered}, {badbytes, Bad}} -> - WarningFun = if - Bad > 0 -> fun rabbit_log:warning/2; - true -> fun rabbit_log:info/2 - end, - WarningFun("Repaired persister log - ~p recovered, ~p bad~n", - [Recovered, Bad]), - LH - end, - {Res, LoadedSnapshot} = internal_load_snapshot(LogHandle, Snapshot), - NewSnapshot = LoadedSnapshot#psnapshot{ - serial = LoadedSnapshot#psnapshot.serial + 1}, - case Res of - ok -> - ok = take_snapshot(LogHandle, NewSnapshot); - {error, Reason} -> - rabbit_log:error("Failed to load persister log: ~p~n", [Reason]), - ok = take_snapshot_and_save_old(LogHandle, NewSnapshot) - end, - State = #pstate{log_handle = LogHandle, - entry_count = 0, - deadline = infinity, - pending_logs = [], - pending_replies = [], - snapshot = NewSnapshot}, - {ok, State}. - -handle_call({transaction, Key, MessageList}, From, State) -> - NewState = internal_extend(Key, MessageList, State), - do_noreply(internal_commit(From, Key, NewState)); -handle_call({commit_transaction, TxnKey}, From, State) -> - do_noreply(internal_commit(From, TxnKey, State)); -handle_call(force_snapshot, _From, State) -> - do_reply(ok, flush(true, State)); -handle_call(serial, _From, - State = #pstate{snapshot = #psnapshot{serial = Serial}}) -> - do_reply(Serial, State); -handle_call(_Request, _From, State) -> - {noreply, State}. - -handle_cast({rollback_transaction, TxnKey}, State) -> - do_noreply(internal_rollback(TxnKey, State)); -handle_cast({dirty_work, MessageList}, State) -> - do_noreply(internal_dirty_work(MessageList, State)); -handle_cast({extend_transaction, TxnKey, MessageList}, State) -> - do_noreply(internal_extend(TxnKey, MessageList, State)); -handle_cast(_Msg, State) -> - {noreply, State}. - -handle_info(timeout, State = #pstate{deadline = infinity}) -> - State1 = flush(true, State), - %% TODO: Once we drop support for R11B-5, we can change this to - %% {noreply, State1, hibernate}; - proc_lib:hibernate(gen_server2, enter_loop, [?MODULE, [], State1]); -handle_info(timeout, State) -> - do_noreply(flush(State)); -handle_info(_Info, State) -> - {noreply, State}. - -terminate(_Reason, State = #pstate{log_handle = LogHandle}) -> - flush(State), - disk_log:close(LogHandle), - ok. - -code_change(_OldVsn, State, _Extra) -> - {ok, flush(State)}. - -%%-------------------------------------------------------------------- - -internal_extend(Key, MessageList, State) -> - log_work(fun (ML) -> {extend_transaction, Key, ML} end, - MessageList, State). - -internal_dirty_work(MessageList, State) -> - log_work(fun (ML) -> {dirty_work, ML} end, - MessageList, State). - -internal_commit(From, Key, State = #pstate{snapshot = Snapshot}) -> - Unit = {commit_transaction, Key}, - NewSnapshot = internal_integrate1(Unit, Snapshot), - complete(From, Unit, State#pstate{snapshot = NewSnapshot}). - -internal_rollback(Key, State = #pstate{snapshot = Snapshot}) -> - Unit = {rollback_transaction, Key}, - NewSnapshot = internal_integrate1(Unit, Snapshot), - log(State#pstate{snapshot = NewSnapshot}, Unit). - -complete(From, Item, State = #pstate{deadline = ExistingDeadline, - pending_logs = Logs, - pending_replies = Waiting}) -> - State#pstate{deadline = compute_deadline( - ?COMPLETE_BUNDLE_DELAY, ExistingDeadline), - pending_logs = [Item | Logs], - pending_replies = [From | Waiting]}. - -%% This is made to limit disk usage by writing messages only once onto -%% disk. We keep a table associating pkeys to messages, and provided -%% the list of messages to output is left to right, we can guarantee -%% that pkeys will be a backreference to a message in memory when a -%% "tied" is met. -log_work(CreateWorkUnit, MessageList, - State = #pstate{ - snapshot = Snapshot = #psnapshot{ - messages = Messages}}) -> - Unit = CreateWorkUnit( - rabbit_misc:map_in_order( - fun(M = {publish, Message, QK = {_QName, PKey}}) -> - case ets:lookup(Messages, PKey) of - [_] -> {tied, QK}; - [] -> ets:insert(Messages, {PKey, Message}), - M - end; - (M) -> M - end, - MessageList)), - NewSnapshot = internal_integrate1(Unit, Snapshot), - log(State#pstate{snapshot = NewSnapshot}, Unit). - -log(State = #pstate{deadline = ExistingDeadline, pending_logs = Logs}, - Message) -> - State#pstate{deadline = compute_deadline(?LOG_BUNDLE_DELAY, - ExistingDeadline), - pending_logs = [Message | Logs]}. - -base_filename() -> - rabbit_mnesia:dir() ++ "/rabbit_persister.LOG". - -take_snapshot(LogHandle, OldFileName, Snapshot) -> - ok = disk_log:sync(LogHandle), - %% current_snapshot is the Head (ie. first thing logged) - ok = disk_log:reopen(LogHandle, OldFileName, current_snapshot(Snapshot)). - -take_snapshot(LogHandle, Snapshot) -> - OldFileName = lists:flatten(base_filename() ++ ".previous"), - file:delete(OldFileName), - rabbit_log:info("Rolling persister log to ~p~n", [OldFileName]), - ok = take_snapshot(LogHandle, OldFileName, Snapshot). - -take_snapshot_and_save_old(LogHandle, Snapshot) -> - {MegaSecs, Secs, MicroSecs} = erlang:now(), - Timestamp = MegaSecs * 1000000 + Secs * 1000 + MicroSecs, - OldFileName = lists:flatten(io_lib:format("~s.saved.~p", - [base_filename(), Timestamp])), - rabbit_log:info("Saving persister log in ~p~n", [OldFileName]), - ok = take_snapshot(LogHandle, OldFileName, Snapshot). - -maybe_take_snapshot(Force, State = #pstate{entry_count = EntryCount, - log_handle = LH, - snapshot = Snapshot}) - when Force orelse EntryCount >= ?MAX_WRAP_ENTRIES -> - ok = take_snapshot(LH, Snapshot), - State#pstate{entry_count = 0}; -maybe_take_snapshot(_Force, State) -> - State. - -later_ms(DeltaMilliSec) -> - {MegaSec, Sec, MicroSec} = now(), - %% Note: not normalised. Unimportant for this application. - {MegaSec, Sec, MicroSec + (DeltaMilliSec * 1000)}. - -%% Result = B - A, more or less -time_diff({B1, B2, B3}, {A1, A2, A3}) -> - (B1 - A1) * 1000000 + (B2 - A2) + (B3 - A3) / 1000000.0 . - -compute_deadline(TimerDelay, infinity) -> - later_ms(TimerDelay); -compute_deadline(_TimerDelay, ExistingDeadline) -> - ExistingDeadline. - -compute_timeout(infinity) -> - ?HIBERNATE_AFTER; -compute_timeout(Deadline) -> - DeltaMilliSec = time_diff(Deadline, now()) * 1000.0, - if - DeltaMilliSec =< 1 -> - 0; - true -> - round(DeltaMilliSec) - end. - -do_noreply(State = #pstate{deadline = Deadline}) -> - {noreply, State, compute_timeout(Deadline)}. - -do_reply(Reply, State = #pstate{deadline = Deadline}) -> - {reply, Reply, State, compute_timeout(Deadline)}. - -flush(State) -> flush(false, State). - -flush(ForceSnapshot, State = #pstate{pending_logs = PendingLogs, - pending_replies = Waiting, - log_handle = LogHandle}) -> - State1 = if PendingLogs /= [] -> - disk_log:alog(LogHandle, lists:reverse(PendingLogs)), - State#pstate{entry_count = State#pstate.entry_count + 1}; - true -> - State - end, - State2 = maybe_take_snapshot(ForceSnapshot, State1), - if Waiting /= [] -> - ok = disk_log:sync(LogHandle), - lists:foreach(fun (From) -> gen_server:reply(From, ok) end, - Waiting); - true -> - ok - end, - State2#pstate{deadline = infinity, - pending_logs = [], - pending_replies = []}. - -current_snapshot(_Snapshot = #psnapshot{serial = Serial, - transactions= Ts, - messages = Messages, - queues = Queues}) -> - %% Avoid infinite growth of the table by removing messages not - %% bound to a queue anymore - prune_table(Messages, ets:foldl( - fun ({{_QName, PKey}, _Delivered}, S) -> - sets:add_element(PKey, S) - end, sets:new(), Queues)), - InnerSnapshot = {{serial, Serial}, - {txns, Ts}, - {messages, ets:tab2list(Messages)}, - {queues, ets:tab2list(Queues)}}, - ?LOGDEBUG("Inner snapshot: ~p~n", [InnerSnapshot]), - {persist_snapshot, {vsn, ?PERSISTER_LOG_FORMAT_VERSION}, - term_to_binary(InnerSnapshot)}. - -prune_table(Tab, Keys) -> - true = ets:safe_fixtable(Tab, true), - ok = prune_table(Tab, Keys, ets:first(Tab)), - true = ets:safe_fixtable(Tab, false). - -prune_table(_Tab, _Keys, '$end_of_table') -> ok; -prune_table(Tab, Keys, Key) -> - case sets:is_element(Key, Keys) of - true -> ok; - false -> ets:delete(Tab, Key) - end, - prune_table(Tab, Keys, ets:next(Tab, Key)). - -internal_load_snapshot(LogHandle, - Snapshot = #psnapshot{messages = Messages, - queues = Queues}) -> - {K, [Loaded_Snapshot | Items]} = disk_log:chunk(LogHandle, start), - case check_version(Loaded_Snapshot) of - {ok, StateBin} -> - {{serial, Serial}, {txns, Ts}, {messages, Ms}, {queues, Qs}} = - binary_to_term(StateBin), - true = ets:insert(Messages, Ms), - true = ets:insert(Queues, Qs), - Snapshot1 = replay(Items, LogHandle, K, - Snapshot#psnapshot{ - serial = Serial, - transactions = Ts}), - Snapshot2 = requeue_messages(Snapshot1), - %% uncompleted transactions are discarded - this is TRTTD - %% since we only get into this code on node restart, so - %% any uncompleted transactions will have been aborted. - {ok, Snapshot2#psnapshot{transactions = dict:new()}}; - {error, Reason} -> {{error, Reason}, Snapshot} - end. - -check_version({persist_snapshot, {vsn, ?PERSISTER_LOG_FORMAT_VERSION}, - StateBin}) -> - {ok, StateBin}; -check_version({persist_snapshot, {vsn, Vsn}, _StateBin}) -> - {error, {unsupported_persister_log_format, Vsn}}; -check_version(_Other) -> - {error, unrecognised_persister_log_format}. - -requeue_messages(Snapshot = #psnapshot{messages = Messages, - queues = Queues}) -> - Work = ets:foldl(fun accumulate_requeues/2, dict:new(), Queues), - %% unstable parallel map, because order doesn't matter - L = lists:append( - rabbit_misc:upmap( - %% we do as much work as possible in spawned worker - %% processes, but we need to make sure the ets:inserts are - %% performed in self() - fun ({QName, Requeues}) -> - requeue(QName, Requeues, Messages) - end, dict:to_list(Work))), - NewMessages = [{K, M} || {{_Q, K}, M, _D} <- L], - NewQueues = [{QK, D} || {QK, _M, D} <- L], - ets:delete_all_objects(Messages), - ets:delete_all_objects(Queues), - true = ets:insert(Messages, NewMessages), - true = ets:insert(Queues, NewQueues), - %% contains the mutated messages and queues tables - Snapshot. - -accumulate_requeues({{QName, PKey}, Delivered}, Acc) -> - Requeue = {PKey, Delivered}, - dict:update(QName, - fun (Requeues) -> [Requeue | Requeues] end, - [Requeue], - Acc). - -requeue(QName, Requeues, Messages) -> - case rabbit_amqqueue:lookup(QName) of - {ok, #amqqueue{pid = QPid}} -> - RequeueMessages = - [{{QName, PKey}, Message, Delivered} || - {PKey, Delivered} <- Requeues, - {_, Message} <- ets:lookup(Messages, PKey)], - rabbit_amqqueue:redeliver( - QPid, - %% Messages published by the same process receive - %% persistence keys that are monotonically - %% increasing. Since message ordering is defined on a - %% per-channel basis, and channels are bound to specific - %% processes, sorting the list does provide the correct - %% ordering properties. - [{Message, Delivered} || {_, Message, Delivered} <- - lists:sort(RequeueMessages)]), - RequeueMessages; - {error, not_found} -> - [] - end. - -replay([], LogHandle, K, Snapshot) -> - case disk_log:chunk(LogHandle, K) of - {K1, Items} -> - replay(Items, LogHandle, K1, Snapshot); - {K1, Items, Badbytes} -> - rabbit_log:warning("~p bad bytes recovering persister log~n", - [Badbytes]), - replay(Items, LogHandle, K1, Snapshot); - eof -> Snapshot - end; -replay([Item | Items], LogHandle, K, Snapshot) -> - NewSnapshot = internal_integrate_messages(Item, Snapshot), - replay(Items, LogHandle, K, NewSnapshot). - -internal_integrate_messages(Items, Snapshot) -> - lists:foldl(fun (Item, Snap) -> internal_integrate1(Item, Snap) end, - Snapshot, Items). - -internal_integrate1({extend_transaction, Key, MessageList}, - Snapshot = #psnapshot {transactions = Transactions}) -> - NewTransactions = - dict:update(Key, - fun (MessageLists) -> [MessageList | MessageLists] end, - [MessageList], - Transactions), - Snapshot#psnapshot{transactions = NewTransactions}; -internal_integrate1({rollback_transaction, Key}, - Snapshot = #psnapshot{transactions = Transactions}) -> - Snapshot#psnapshot{transactions = dict:erase(Key, Transactions)}; -internal_integrate1({commit_transaction, Key}, - Snapshot = #psnapshot{transactions = Transactions, - messages = Messages, - queues = Queues}) -> - case dict:find(Key, Transactions) of - {ok, MessageLists} -> - ?LOGDEBUG("persist committing txn ~p~n", [Key]), - lists:foreach(fun (ML) -> perform_work(ML, Messages, Queues) end, - lists:reverse(MessageLists)), - Snapshot#psnapshot{transactions = dict:erase(Key, Transactions)}; - error -> - Snapshot - end; -internal_integrate1({dirty_work, MessageList}, - Snapshot = #psnapshot {messages = Messages, - queues = Queues}) -> - perform_work(MessageList, Messages, Queues), - Snapshot. - -perform_work(MessageList, Messages, Queues) -> - lists:foreach( - fun (Item) -> perform_work_item(Item, Messages, Queues) end, - MessageList). - -perform_work_item({publish, Message, QK = {_QName, PKey}}, Messages, Queues) -> - ets:insert(Messages, {PKey, Message}), - ets:insert(Queues, {QK, false}); - -perform_work_item({tied, QK}, _Messages, Queues) -> - ets:insert(Queues, {QK, false}); - -perform_work_item({deliver, QK}, _Messages, Queues) -> - %% from R12B-2 onward we could use ets:update_element/3 here - ets:delete(Queues, QK), - ets:insert(Queues, {QK, true}); - -perform_work_item({ack, QK}, _Messages, Queues) -> - ets:delete(Queues, QK). diff --git a/src/rabbit_queue_index.erl b/src/rabbit_queue_index.erl new file mode 100644 index 0000000000..a198ba51ee --- /dev/null +++ b/src/rabbit_queue_index.erl @@ -0,0 +1,862 @@ +%% The contents of this file are subject to the Mozilla Public License +%% Version 1.1 (the "License"); you may not use this file except in +%% compliance with the License. You may obtain a copy of the License at +%% http://www.mozilla.org/MPL/ +%% +%% 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 Original Code is RabbitMQ. +%% +%% The Initial Developers of the Original Code are LShift Ltd, +%% Cohesive Financial Technologies LLC, and Rabbit Technologies Ltd. +%% +%% Portions created before 22-Nov-2008 00:00:00 GMT by LShift Ltd, +%% Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd +%% are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial +%% Technologies LLC, and Rabbit Technologies Ltd. +%% +%% Portions created by LShift Ltd are Copyright (C) 2007-2009 LShift +%% Ltd. Portions created by Cohesive Financial Technologies LLC are +%% Copyright (C) 2007-2009 Cohesive Financial Technologies +%% LLC. Portions created by Rabbit Technologies Ltd are Copyright +%% (C) 2007-2009 Rabbit Technologies Ltd. +%% +%% All Rights Reserved. +%% +%% Contributor(s): ______________________________________. +%% + +-module(rabbit_queue_index). + +-export([init/1, terminate/1, terminate_and_erase/1, write_published/4, + write_delivered/2, write_acks/2, sync_seq_ids/3, can_flush_journal/1, + flush_journal/1, read_segment_entries/2, next_segment_boundary/1, + segment_size/0, find_lowest_seq_id_seg_and_next_seq_id/1, + start_msg_store/1]). + +%%---------------------------------------------------------------------------- +%% The queue disk index +%% +%% The queue disk index operates over a journal, and a number of +%% segment files. Each segment is the same size, both in max number of +%% entries, and max file size, owing to fixed sized records. +%% +%% Publishes are written directly to the segment files. The segment is +%% found by dividing the sequence id by the the max number of entries +%% per segment. Only the relative sequence within the segment is +%% recorded as the sequence id within a segment file (i.e. sequence id +%% modulo max number of entries per segment). This is keeps entries +%% as small as possible. Publishes are only ever going to be received +%% in contiguous ascending order. +%% +%% Acks and deliveries are written to a bounded journal and are also +%% held in memory, each in a dict with the segment as the key. Again, +%% the records are fixed size: the entire sequence id is written and +%% is limited to a 63-bit unsigned integer. The remaining bit +%% indicates whether the journal entry is for a delivery or an +%% ack. When the journal gets too big, or flush_journal is called, the +%% journal is (possibly incrementally) flushed out to the segment +%% files. As acks and delivery notes can be received in any order +%% (this is not obvious for deliveries, but consider what happens when +%% eg msgs are *re*queued - you'll publish and then mark the msgs +%% delivered immediately, which may be out of order), this journal +%% reduces seeking, and batches writes to the segment files, keeping +%% performance high. +%% +%% On startup, the journal is read along with all the segment files, +%% and the journal is fully flushed out to the segment files. Care is +%% taken to ensure that no message can be delivered or ack'd twice. +%% +%%---------------------------------------------------------------------------- + +-define(CLEAN_FILENAME, "clean.dot"). + +-define(MAX_ACK_JOURNAL_ENTRY_COUNT, 32768). +-define(ACK_JOURNAL_FILENAME, "journal.jif"). + +-define(DEL_BIT, 0). +-define(ACK_BIT, 1). +-define(SEQ_BYTES, 8). +-define(SEQ_BITS, ((?SEQ_BYTES * 8) - 1)). +-define(SEGMENT_EXTENSION, ".idx"). + +-define(REL_SEQ_BITS, 14). +-define(REL_SEQ_BITS_BYTE_ALIGNED, (?REL_SEQ_BITS + 8 - (?REL_SEQ_BITS rem 8))). +-define(SEGMENT_ENTRIES_COUNT, 16384). %% trunc(math:pow(2,?REL_SEQ_BITS))). + +%% seq only is binary 00 followed by 14 bits of rel seq id +%% (range: 0 - 16383) +-define(REL_SEQ_ONLY_PREFIX, 00). +-define(REL_SEQ_ONLY_PREFIX_BITS, 2). +-define(REL_SEQ_ONLY_ENTRY_LENGTH_BYTES, 2). + +%% publish record is binary 1 followed by a bit for is_persistent, +%% then 14 bits of rel seq id, and 128 bits of md5sum msg id +-define(PUBLISH_PREFIX, 1). +-define(PUBLISH_PREFIX_BITS, 1). + +-define(MSG_ID_BYTES, 16). %% md5sum is 128 bit or 16 bytes +-define(MSG_ID_BITS, (?MSG_ID_BYTES * 8)). +%% 16 bytes for md5sum + 2 for seq, bits and prefix +-define(PUBLISH_RECORD_LENGTH_BYTES, ?MSG_ID_BYTES + 2). + +%% 1 publish, 1 deliver, 1 ack per msg +-define(SEGMENT_TOTAL_SIZE, ?SEGMENT_ENTRIES_COUNT * + (?PUBLISH_RECORD_LENGTH_BYTES + + (2 * ?REL_SEQ_ONLY_ENTRY_LENGTH_BYTES))). + +%%---------------------------------------------------------------------------- + +-record(qistate, + { dir, + seg_num_handles, + journal_count, + journal_ack_dict, + journal_del_dict, + seg_ack_counts, + publish_handle, + partial_segments + }). + +-include("rabbit.hrl"). + +%%---------------------------------------------------------------------------- + +-ifdef(use_specs). + +-type(hdl() :: ('undefined' | any())). +-type(msg_id() :: binary()). +-type(seq_id() :: integer()). +-type(hdl_and_count() :: ('undefined' | + {non_neg_integer(), hdl(), non_neg_integer()})). +-type(qistate() :: #qistate { dir :: file_path(), + seg_num_handles :: dict(), + journal_count :: integer(), + journal_ack_dict :: dict(), + journal_del_dict :: dict(), + seg_ack_counts :: dict(), + publish_handle :: hdl_and_count(), + partial_segments :: dict() + }). + +-spec(init/1 :: (queue_name()) -> {non_neg_integer(), qistate()}). +-spec(terminate/1 :: (qistate()) -> qistate()). +-spec(terminate_and_erase/1 :: (qistate()) -> qistate()). +-spec(write_published/4 :: (msg_id(), seq_id(), boolean(), qistate()) + -> qistate()). +-spec(write_delivered/2 :: (seq_id(), qistate()) -> qistate()). +-spec(write_acks/2 :: ([seq_id()], qistate()) -> qistate()). +-spec(sync_seq_ids/3 :: ([seq_id()], boolean(), qistate()) -> qistate()). +-spec(can_flush_journal/1 :: (qistate()) -> boolean()). +-spec(flush_journal/1 :: (qistate()) -> qistate()). +-spec(read_segment_entries/2 :: (seq_id(), qistate()) -> + {[{msg_id(), seq_id(), boolean(), boolean()}], qistate()}). +-spec(next_segment_boundary/1 :: (seq_id()) -> seq_id()). +-spec(segment_size/0 :: () -> non_neg_integer()). +-spec(find_lowest_seq_id_seg_and_next_seq_id/1 :: (qistate()) -> + {non_neg_integer(), non_neg_integer(), qistate()}). +-spec(start_msg_store/1 :: ([amqqueue()]) -> 'ok'). + +-endif. + + +%%---------------------------------------------------------------------------- +%% Public API +%%---------------------------------------------------------------------------- + +init(Name) -> + State = blank_state(Name), + {TotalMsgCount, State1} = read_and_prune_segments(State), + scatter_journal(TotalMsgCount, State1). + +terminate(State = #qistate { seg_num_handles = SegHdls }) -> + case 0 == dict:size(SegHdls) of + true -> State; + false -> State1 = #qistate { dir = Dir } = close_all_handles(State), + store_clean_shutdown(Dir), + State1 #qistate { publish_handle = undefined } + end. + +terminate_and_erase(State) -> + State1 = terminate(State), + ok = delete_queue_directory(State1 #qistate.dir), + State1. + +write_published(MsgId, SeqId, IsPersistent, State) + when is_binary(MsgId) -> + ?MSG_ID_BYTES = size(MsgId), + {SegNum, RelSeq} = seq_id_to_seg_and_rel_seq_id(SeqId), + {Hdl, State1} = get_pub_handle(SegNum, State), + ok = file_handle_cache:append(Hdl, + <<?PUBLISH_PREFIX:?PUBLISH_PREFIX_BITS, + (bool_to_int(IsPersistent)):1, + RelSeq:?REL_SEQ_BITS, MsgId/binary>>), + State1. + +write_delivered(SeqId, State = #qistate { journal_del_dict = JDelDict }) -> + {JDelDict1, State1} = + write_to_journal([<<?DEL_BIT:1, SeqId:?SEQ_BITS>>], + [SeqId], JDelDict, State), + maybe_full_flush(State1 #qistate { journal_del_dict = JDelDict1 }). + +write_acks(SeqIds, State = #qistate { journal_ack_dict = JAckDict }) -> + {JAckDict1, State1} = + write_to_journal([<<?ACK_BIT:1, SeqId:?SEQ_BITS>> || SeqId <- SeqIds], + SeqIds, JAckDict, State), + maybe_full_flush(State1 #qistate { journal_ack_dict = JAckDict1 }). + +sync_seq_ids(SeqIds, SyncAckJournal, State) -> + State1 = case SyncAckJournal of + true -> {Hdl, State2} = get_journal_handle(State), + ok = file_handle_cache:sync(Hdl), + State2; + false -> State + end, + SegNumsSet = + lists:foldl( + fun (SeqId, Set) -> + {SegNum, _RelSeq} = seq_id_to_seg_and_rel_seq_id(SeqId), + sets:add_element(SegNum, Set) + end, sets:new(), SeqIds), + sets:fold( + fun (SegNum, StateN) -> + {Hdl1, StateM} = get_seg_handle(SegNum, StateN), + ok = file_handle_cache:sync(Hdl1), + StateM + end, State1, SegNumsSet). + +can_flush_journal(#qistate { journal_count = 0 }) -> + false; +can_flush_journal(_) -> + true. + +flush_journal(State = #qistate { journal_count = 0 }) -> + State; +flush_journal(State = #qistate { journal_ack_dict = JAckDict, + journal_del_dict = JDelDict, + journal_count = JCount }) -> + SegNum = case dict:fetch_keys(JAckDict) of + [] -> hd(dict:fetch_keys(JDelDict)); + [N|_] -> N + end, + Dels = seg_entries_from_dict(SegNum, JDelDict), + Acks = seg_entries_from_dict(SegNum, JAckDict), + State1 = append_dels_to_segment(SegNum, Dels, State), + State2 = append_acks_to_segment(SegNum, Acks, State1), + JCount1 = JCount - length(Dels) - length(Acks), + State3 = State2 #qistate { journal_del_dict = dict:erase(SegNum, JDelDict), + journal_ack_dict = dict:erase(SegNum, JAckDict), + journal_count = JCount1 }, + if + JCount1 == 0 -> + {Hdl, State4} = get_journal_handle(State3), + {ok, 0} = file_handle_cache:position(Hdl, bof), + ok = file_handle_cache:truncate(Hdl), + ok = file_handle_cache:sync(Hdl), + State4; + JCount1 > ?MAX_ACK_JOURNAL_ENTRY_COUNT -> + flush_journal(State3); + true -> + State3 + end. + +read_segment_entries(InitSeqId, State) -> + {SegNum, 0} = seq_id_to_seg_and_rel_seq_id(InitSeqId), + {SDict, _PubCount, _AckCount, _HighRelSeq, State1} = + load_segment(SegNum, State), + %% deliberately sort the list desc, because foldl will reverse it + RelSeqs = rev_sort(dict:fetch_keys(SDict)), + {lists:foldl(fun (RelSeq, Acc) -> + {MsgId, IsDelivered, IsPersistent} = + dict:fetch(RelSeq, SDict), + [ {MsgId, reconstruct_seq_id(SegNum, RelSeq), + IsPersistent, IsDelivered} | Acc] + end, [], RelSeqs), + State1}. + +next_segment_boundary(SeqId) -> + {SegNum, _RelSeq} = seq_id_to_seg_and_rel_seq_id(SeqId), + reconstruct_seq_id(SegNum + 1, 0). + +segment_size() -> + ?SEGMENT_ENTRIES_COUNT. + +find_lowest_seq_id_seg_and_next_seq_id(State = #qistate { dir = Dir }) -> + SegNums = all_segment_nums(Dir), + %% We don't want the lowest seq_id, merely the seq_id of the start + %% of the lowest segment. That seq_id may not actually exist, but + %% that's fine. The important thing is that the segment exists and + %% the seq_id reported is on a segment boundary. + + %% SegNums is sorted, ascending. + LowSeqIdSeg = + case SegNums of + [] -> 0; + [MinSegNum|_] -> reconstruct_seq_id(MinSegNum, 0) + end, + {NextSeqId, State1} = + case SegNums of + [] -> {0, State}; + _ -> MaxSegNum = lists:last(SegNums), + {_SDict, PubCount, _AckCount, HighRelSeq, State2} = + load_segment(MaxSegNum, State), + NextSeqId1 = reconstruct_seq_id(MaxSegNum, HighRelSeq), + NextSeqId2 = case PubCount of + 0 -> NextSeqId1; + _ -> NextSeqId1 + 1 + end, + {NextSeqId2, State2} + end, + {LowSeqIdSeg, NextSeqId, State1}. + +start_msg_store(DurableQueues) -> + DurableDict = + dict:from_list([ {queue_name_to_dir_name(Queue #amqqueue.name), + Queue #amqqueue.name} || Queue <- DurableQueues ]), + QueuesDir = queues_dir(), + Directories = case file:list_dir(QueuesDir) of + {ok, Entries} -> + [ Entry || Entry <- Entries, + filelib:is_dir( + filename:join(QueuesDir, Entry)) ]; + {error, enoent} -> + [] + end, + DurableDirectories = sets:from_list(dict:fetch_keys(DurableDict)), + {DurableQueueNames, TransientDirs} = + lists:foldl( + fun (QueueDir, {DurableAcc, TransientAcc}) -> + case sets:is_element(QueueDir, DurableDirectories) of + true -> + {[dict:fetch(QueueDir, DurableDict) | DurableAcc], + TransientAcc}; + false -> + {DurableAcc, [QueueDir | TransientAcc]} + end + end, {[], []}, Directories), + MsgStoreDir = filename:join(rabbit_mnesia:dir(), "msg_store"), + ok = rabbit:start_child(rabbit_msg_store, [MsgStoreDir, + fun queue_index_walker/1, + DurableQueueNames]), + lists:foreach(fun (DirName) -> + Dir = filename:join(queues_dir(), DirName), + ok = delete_queue_directory(Dir) + end, TransientDirs), + ok. + + +%%---------------------------------------------------------------------------- +%% Minor Helpers +%%---------------------------------------------------------------------------- + +write_to_journal(BinList, SeqIds, Dict, + State = #qistate { journal_count = JCount }) -> + {Hdl, State1} = get_journal_handle(State), + ok = file_handle_cache:append(Hdl, BinList), + {Dict1, JCount1} = + lists:foldl( + fun (SeqId, {Dict2, JCount2}) -> + {add_seqid_to_dict(SeqId, Dict2), JCount2 + 1} + end, {Dict, JCount}, SeqIds), + {Dict1, State1 #qistate { journal_count = JCount1 }}. + +maybe_full_flush(State = #qistate { journal_count = JCount }) -> + case JCount > ?MAX_ACK_JOURNAL_ENTRY_COUNT of + true -> full_flush_journal(State); + false -> State + end. + +full_flush_journal(State) -> + case can_flush_journal(State) of + true -> State1 = flush_journal(State), + full_flush_journal(State1); + false -> State + end. + +queue_name_to_dir_name(Name = #resource { kind = queue }) -> + Bin = term_to_binary(Name), + Size = 8*size(Bin), + <<Num:Size>> = Bin, + lists:flatten(io_lib:format("~.36B", [Num])). + +queues_dir() -> + filename:join(rabbit_mnesia:dir(), "queues"). + +rev_sort(List) -> + lists:sort(fun (A, B) -> B < A end, List). + +get_journal_handle(State = #qistate { dir = Dir, seg_num_handles = SegHdls }) -> + case dict:find(journal, SegHdls) of + {ok, Hdl} -> {Hdl, State}; + error -> + Path = filename:join(Dir, ?ACK_JOURNAL_FILENAME), + Mode = [raw, binary, delayed_write, write, read, read_ahead], + new_handle(journal, Path, Mode, State) + end. + +get_pub_handle(SegNum, State = #qistate { publish_handle = PubHandle }) -> + {State1, PubHandle1 = {_SegNum, Hdl, _Count}} = + get_counted_handle(SegNum, State, PubHandle), + {Hdl, State1 #qistate { publish_handle = PubHandle1 }}. + +get_counted_handle(SegNum, State, undefined) -> + get_counted_handle(SegNum, State, {SegNum, undefined, 0}); +get_counted_handle(SegNum, State = #qistate { partial_segments = Partials }, + {SegNum, undefined, Count}) -> + {Hdl, State1} = get_seg_handle(SegNum, State), + {CountExtra, Partials1} = + case dict:find(SegNum, Partials) of + {ok, CountExtra1} -> {CountExtra1, dict:erase(SegNum, Partials)}; + error -> {0, Partials} + end, + Count1 = Count + 1 + CountExtra, + {State1 #qistate { partial_segments = Partials1 }, {SegNum, Hdl, Count1}}; +get_counted_handle(SegNum, State, {SegNum, Hdl, Count}) + when Count < ?SEGMENT_ENTRIES_COUNT -> + {State, {SegNum, Hdl, Count + 1}}; +get_counted_handle(SegNumA, State, {SegNumB, Hdl, ?SEGMENT_ENTRIES_COUNT}) + when SegNumA == SegNumB + 1 -> + ok = file_handle_cache:append_write_buffer(Hdl), + get_counted_handle(SegNumA, State, undefined); +get_counted_handle(SegNumA, State = #qistate { partial_segments = Partials, + seg_ack_counts = AckCounts, + dir = Dir }, + {SegNumB, Hdl, Count}) -> + %% don't flush here because it's possible SegNumB has been deleted + State1 = + case dict:find(SegNumB, AckCounts) of + {ok, Count} -> + %% #acks == #pubs, and we're moving to different + %% segment, so delete. + delete_segment(SegNumB, State); + _ -> + State #qistate { + partial_segments = dict:store(SegNumB, Count, Partials) } + end, + get_counted_handle(SegNumA, State1, undefined). + +get_seg_handle(SegNum, State = #qistate { dir = Dir, seg_num_handles = SegHdls }) -> + case dict:find(SegNum, SegHdls) of + {ok, Hdl} -> {Hdl, State}; + error -> + new_handle(SegNum, seg_num_to_path(Dir, SegNum), + [binary, raw, read, write, + {delayed_write, ?SEGMENT_TOTAL_SIZE, 1000}, + {read_ahead, ?SEGMENT_TOTAL_SIZE}], + State) + end. + +delete_segment(SegNum, State = #qistate { dir = Dir, + seg_ack_counts = AckCounts, + partial_segments = Partials }) -> + State1 = close_handle(SegNum, State), + ok = case file:delete(seg_num_to_path(Dir, SegNum)) of + ok -> ok; + {error, enoent} -> ok + end, + State1 #qistate {seg_ack_counts = dict:erase(SegNum, AckCounts), + partial_segments = dict:erase(SegNum, Partials) }. + +new_handle(Key, Path, Mode, State = #qistate { seg_num_handles = SegHdls }) -> + {ok, Hdl} = file_handle_cache:open(Path, Mode, [{write_buffer, infinity}]), + {Hdl, State #qistate { seg_num_handles = dict:store(Key, Hdl, SegHdls) }}. + +close_handle(Key, State = #qistate { seg_num_handles = SegHdls }) -> + case dict:find(Key, SegHdls) of + {ok, Hdl} -> + ok = file_handle_cache:close(Hdl), + State #qistate { seg_num_handles = dict:erase(Key, SegHdls) }; + error -> State + end. + +close_all_handles(State = #qistate { seg_num_handles = SegHdls }) -> + ok = dict:fold(fun (_Key, Hdl, ok) -> + file_handle_cache:close(Hdl) + end, ok, SegHdls), + State #qistate { seg_num_handles = dict:new() }. + +bool_to_int(true ) -> 1; +bool_to_int(false) -> 0. + +seq_id_to_seg_and_rel_seq_id(SeqId) -> + { SeqId div ?SEGMENT_ENTRIES_COUNT, SeqId rem ?SEGMENT_ENTRIES_COUNT }. + +reconstruct_seq_id(SegNum, RelSeq) -> + (SegNum * ?SEGMENT_ENTRIES_COUNT) + RelSeq. + +seg_num_to_path(Dir, SegNum) -> + SegName = integer_to_list(SegNum), + filename:join(Dir, SegName ++ ?SEGMENT_EXTENSION). + +delete_queue_directory(Dir) -> + {ok, Entries} = file:list_dir(Dir), + ok = lists:foldl(fun (Entry, ok) -> + file:delete(filename:join(Dir, Entry)) + end, ok, Entries), + ok = file:del_dir(Dir). + +add_seqid_to_dict(SeqId, Dict) -> + {SegNum, RelSeq} = seq_id_to_seg_and_rel_seq_id(SeqId), + add_seqid_to_dict(SegNum, RelSeq, Dict). + +add_seqid_to_dict(SegNum, RelSeq, Dict) -> + dict:update(SegNum, fun(Lst) -> [RelSeq|Lst] end, [RelSeq], Dict). + +all_segment_nums(Dir) -> + lists:sort( + [list_to_integer( + lists:takewhile(fun(C) -> $0 =< C andalso C =< $9 end, SegName)) + || SegName <- filelib:wildcard("*" ++ ?SEGMENT_EXTENSION, Dir)]). + +blank_state(QueueName) -> + StrName = queue_name_to_dir_name(QueueName), + Dir = filename:join(queues_dir(), StrName), + ok = filelib:ensure_dir(filename:join(Dir, "nothing")), + #qistate { dir = Dir, + seg_num_handles = dict:new(), + journal_count = 0, + journal_ack_dict = dict:new(), + journal_del_dict = dict:new(), + seg_ack_counts = dict:new(), + publish_handle = undefined, + partial_segments = dict:new() + }. + +detect_clean_shutdown(Dir) -> + case file:delete(filename:join(Dir, ?CLEAN_FILENAME)) of + ok -> true; + {error, enoent} -> false + end. + +store_clean_shutdown(Dir) -> + {ok, Hdl} = file_handle_cache:open(filename:join(Dir, ?CLEAN_FILENAME), + [write, raw, binary], + [{write_buffer, unbuffered}]), + ok = file_handle_cache:close(Hdl). + +seg_entries_from_dict(SegNum, Dict) -> + case dict:find(SegNum, Dict) of + {ok, Entries} -> Entries; + error -> [] + end. + + +%%---------------------------------------------------------------------------- +%% Msg Store Startup Delta Function +%%---------------------------------------------------------------------------- + +queue_index_walker([]) -> + finished; +queue_index_walker([QueueName|QueueNames]) -> + State = blank_state(QueueName), + {Hdl, State1} = get_journal_handle(State), + {_JDelDict, JAckDict} = load_journal(Hdl, dict:new(), dict:new()), + State2 = #qistate { dir = Dir } = + close_handle(journal, State1 #qistate { journal_ack_dict = JAckDict }), + SegNums = all_segment_nums(Dir), + queue_index_walker({SegNums, State2, QueueNames}); + +queue_index_walker({[], State, QueueNames}) -> + _State = terminate(State), + queue_index_walker(QueueNames); +queue_index_walker({[SegNum | SegNums], State, QueueNames}) -> + {SDict, _PubCount, _AckCount, _HighRelSeq, State1} = + load_segment(SegNum, State), + queue_index_walker({dict:to_list(SDict), State1, SegNums, QueueNames}); + +queue_index_walker({[], State, SegNums, QueueNames}) -> + queue_index_walker({SegNums, State, QueueNames}); +queue_index_walker({[{_RelSeq, {MsgId, _IsDelivered, IsPersistent}} | Msgs], + State, SegNums, QueueNames}) -> + case IsPersistent of + true -> {MsgId, 1, {Msgs, State, SegNums, QueueNames}}; + false -> queue_index_walker({Msgs, State, SegNums, QueueNames}) + end. + + +%%---------------------------------------------------------------------------- +%% Startup Functions +%%---------------------------------------------------------------------------- + +read_and_prune_segments(State = #qistate { dir = Dir }) -> + SegNums = all_segment_nums(Dir), + CleanShutdown = detect_clean_shutdown(Dir), + {TotalMsgCount, State1} = + lists:foldl( + fun (SegNum, {TotalMsgCount1, StateN = + #qistate { publish_handle = PublishHandle, + partial_segments = Partials }}) -> + {SDict, PubCount, AckCount, _HighRelSeq, StateM} = + load_segment(SegNum, StateN), + StateL = #qistate { seg_ack_counts = AckCounts } = + drop_and_deliver(SegNum, SDict, CleanShutdown, StateM), + %% ignore the effect of drop_and_deliver on + %% TotalMsgCount and AckCounts, as drop_and_deliver + %% will add to the journal dicts, which will then + %% effect TotalMsgCount when we scatter the journal + TotalMsgCount2 = TotalMsgCount1 + dict:size(SDict), + AckCounts1 = case AckCount of + 0 -> AckCounts; + N -> dict:store(SegNum, N, AckCounts) + end, + %% In the following, whilst there may be several + %% partial segments, we only remember the last + %% one. All other partial segments get added into + %% the partial_segments dict + {PublishHandle1, Partials1} = + case PubCount of + ?SEGMENT_ENTRIES_COUNT -> + {PublishHandle, Partials}; + 0 -> + {PublishHandle, Partials}; + _ -> + {{SegNum, undefined, PubCount}, + case PublishHandle of + undefined -> + Partials; + {SegNumOld, undefined, PubCountOld} -> + dict:store(SegNumOld, PubCountOld, + Partials) + end} + end, + {TotalMsgCount2, + StateL #qistate { seg_ack_counts = AckCounts1, + publish_handle = PublishHandle1, + partial_segments = Partials1 }} + end, {0, State}, SegNums), + {TotalMsgCount, State1}. + +scatter_journal(TotalMsgCount, State = #qistate { dir = Dir }) -> + {Hdl, State1 = #qistate { journal_del_dict = JDelDict, + journal_ack_dict = JAckDict }} = + get_journal_handle(State), + %% ADict and DDict may well contain duplicates. However, this is + %% ok, because we use sets to eliminate dups before writing to + %% segments + {ADict, DDict} = load_journal(Hdl, JAckDict, JDelDict), + State2 = close_handle(journal, State1), + {TotalMsgCount1, ADict1, State3} = + dict:fold(fun replay_journal_to_segment/3, + {TotalMsgCount, ADict, + %% supply empty dicts so that when + %% replay_journal_to_segment loads segments, it + %% gets all msgs, and ignores anything we've found + %% in the journal. + State2 #qistate { journal_del_dict = dict:new(), + journal_ack_dict = dict:new() }}, DDict), + %% replay for segments which only had acks, and no deliveries + {TotalMsgCount2, State4} = + dict:fold(fun replay_journal_acks_to_segment/3, + {TotalMsgCount1, State3}, ADict1), + JournalPath = filename:join(Dir, ?ACK_JOURNAL_FILENAME), + ok = file:delete(JournalPath), + {TotalMsgCount2, State4}. + +load_journal(Hdl, ADict, DDict) -> + case file_handle_cache:read(Hdl, ?SEQ_BYTES) of + {ok, <<?DEL_BIT:1, SeqId:?SEQ_BITS>>} -> + load_journal(Hdl, ADict, add_seqid_to_dict(SeqId, DDict)); + {ok, <<?ACK_BIT:1, SeqId:?SEQ_BITS>>} -> + load_journal(Hdl, add_seqid_to_dict(SeqId, ADict), DDict); + _ErrOrEoF -> {ADict, DDict} + end. + +replay_journal_to_segment(_SegNum, [], {TotalMsgCount, ADict, State}) -> + {TotalMsgCount, ADict, State}; +replay_journal_to_segment(SegNum, Dels, {TotalMsgCount, ADict, State}) -> + {SDict, _PubCount, _AckCount, _HighRelSeq, State1} = + load_segment(SegNum, State), + ValidDels = sets:to_list( + sets:filter( + fun (RelSeq) -> + case dict:find(RelSeq, SDict) of + {ok, {_MsgId, false, _IsPersistent}} -> true; + _ -> false + end + end, sets:from_list(Dels))), + State2 = append_dels_to_segment(SegNum, ValidDels, State1), + Acks = seg_entries_from_dict(SegNum, ADict), + case Acks of + [] -> {TotalMsgCount, ADict, State2}; + _ -> + ADict1 = dict:erase(SegNum, ADict), + {Count, State3} = filter_acks_and_append_to_segment(SegNum, SDict, + Acks, State2), + {TotalMsgCount - Count, ADict1, State3} + end. + +replay_journal_acks_to_segment(_SegNum, [], {TotalMsgCount, State}) -> + {TotalMsgCount, State}; +replay_journal_acks_to_segment(SegNum, Acks, {TotalMsgCount, State}) -> + {SDict, _PubCount, _AckCount, _HighRelSeq, State1} = + load_segment(SegNum, State), + {Count, State2} = + filter_acks_and_append_to_segment(SegNum, SDict, Acks, State1), + {TotalMsgCount - Count, State2}. + +filter_acks_and_append_to_segment(SegNum, SDict, Acks, State) -> + ValidRelSeqIds = dict:fetch_keys(SDict), + ValidAcks = sets:to_list(sets:intersection(sets:from_list(ValidRelSeqIds), + sets:from_list(Acks))), + {length(ValidAcks), append_acks_to_segment(SegNum, ValidAcks, State)}. + +drop_and_deliver(SegNum, SDict, CleanShutdown, + State = #qistate { journal_del_dict = JDelDict, + journal_ack_dict = JAckDict }) -> + {JDelDict1, JAckDict1} = + dict:fold( + fun (RelSeq, {MsgId, IsDelivered, true}, {JDelDict2, JAckDict2}) -> + %% msg is persistent, keep only if the msg_store has it + case {IsDelivered, rabbit_msg_store:contains(MsgId)} of + {false, true} when not CleanShutdown -> + %% not delivered, but dirty shutdown => mark delivered + {add_seqid_to_dict(SegNum, RelSeq, JDelDict2), + JAckDict2}; + {_, true} -> + {JDelDict2, JAckDict2}; + {true, false} -> + {JDelDict2, + add_seqid_to_dict(SegNum, RelSeq, JAckDict2)}; + {false, false} -> + {add_seqid_to_dict(SegNum, RelSeq, JDelDict2), + add_seqid_to_dict(SegNum, RelSeq, JAckDict2)} + end; + (RelSeq, {_MsgId, false, false}, {JDelDict2, JAckDict2}) -> + %% not persistent and not delivered => deliver and ack it + {add_seqid_to_dict(SegNum, RelSeq, JDelDict2), + add_seqid_to_dict(SegNum, RelSeq, JAckDict2)}; + (RelSeq, {_MsgId, true, false}, {JDelDict2, JAckDict2}) -> + %% not persistent but delivered => ack it + {JDelDict2, + add_seqid_to_dict(SegNum, RelSeq, JAckDict2)} + end, {JDelDict, JAckDict}, SDict), + State #qistate { journal_del_dict = JDelDict1, + journal_ack_dict = JAckDict1 }. + + +%%---------------------------------------------------------------------------- +%% Loading Segments +%%---------------------------------------------------------------------------- + +load_segment(SegNum, State = #qistate { seg_num_handles = SegHdls, + dir = Dir }) -> + SegmentExists = case dict:find(SegNum, SegHdls) of + {ok, _} -> true; + error -> filelib:is_file(seg_num_to_path(Dir, SegNum)) + end, + case SegmentExists of + false -> {dict:new(), 0, 0, 0, State}; + true -> + {Hdl, State1 = #qistate { journal_del_dict = JDelDict, + journal_ack_dict = JAckDict }} = + get_seg_handle(SegNum, State), + {ok, 0} = file_handle_cache:position(Hdl, bof), + {SDict, PubCount, AckCount, HighRelSeq} = + load_segment_entries(Hdl, dict:new(), 0, 0, 0), + %% delete ack'd msgs first + {SDict1, AckCount1} = + lists:foldl(fun (RelSeq, {SDict2, AckCount2}) -> + {dict:erase(RelSeq, SDict2), AckCount2 + 1} + end, {SDict, AckCount}, + seg_entries_from_dict(SegNum, JAckDict)), + %% ensure remaining msgs are delivered as necessary + SDict3 = + lists:foldl( + fun (RelSeq, SDict4) -> + case dict:find(RelSeq, SDict4) of + {ok, {MsgId, false, IsPersistent}} -> + dict:store(RelSeq, {MsgId, true, IsPersistent}, + SDict4); + _ -> SDict4 + end + end, SDict1, seg_entries_from_dict(SegNum, JDelDict)), + + {SDict3, PubCount, AckCount1, HighRelSeq, State1} + end. + +load_segment_entries(Hdl, SDict, PubCount, AckCount, HighRelSeq) -> + case file_handle_cache:read(Hdl, 1) of + {ok, <<?REL_SEQ_ONLY_PREFIX:?REL_SEQ_ONLY_PREFIX_BITS, + MSB:(8-?REL_SEQ_ONLY_PREFIX_BITS)>>} -> + {ok, LSB} = file_handle_cache:read( + Hdl, ?REL_SEQ_ONLY_ENTRY_LENGTH_BYTES - 1), + <<RelSeq:?REL_SEQ_BITS_BYTE_ALIGNED>> = <<MSB, LSB/binary>>, + {SDict1, AckCount1} = deliver_or_ack_msg(SDict, AckCount, RelSeq), + load_segment_entries(Hdl, SDict1, PubCount, AckCount1, HighRelSeq); + {ok, <<?PUBLISH_PREFIX:?PUBLISH_PREFIX_BITS, + IsPersistentNum:1, MSB:(7-?PUBLISH_PREFIX_BITS)>>} -> + %% because we specify /binary, and binaries are complete + %% bytes, the size spec is in bytes, not bits. + {ok, <<LSB:1/binary, MsgId:?MSG_ID_BYTES/binary>>} = + file_handle_cache:read( + Hdl, ?PUBLISH_RECORD_LENGTH_BYTES - 1), + <<RelSeq:?REL_SEQ_BITS_BYTE_ALIGNED>> = <<MSB, LSB/binary>>, + HighRelSeq1 = lists:max([RelSeq, HighRelSeq]), + load_segment_entries( + Hdl, dict:store(RelSeq, {MsgId, false, 1 == IsPersistentNum}, + SDict), PubCount + 1, AckCount, HighRelSeq1); + _ErrOrEoF -> {SDict, PubCount, AckCount, HighRelSeq} + end. + +deliver_or_ack_msg(SDict, AckCount, RelSeq) -> + case dict:find(RelSeq, SDict) of + {ok, {MsgId, false, IsPersistent}} -> + {dict:store(RelSeq, {MsgId, true, IsPersistent}, SDict), AckCount}; + {ok, {_MsgId, true, _IsPersistent}} -> + {dict:erase(RelSeq, SDict), AckCount + 1} + end. + + +%%---------------------------------------------------------------------------- +%% Appending Acks or Dels to Segments +%%---------------------------------------------------------------------------- + +append_acks_to_segment(SegNum, Acks, + State = #qistate { seg_ack_counts = AckCounts, + partial_segments = Partials }) -> + AckCount = case dict:find(SegNum, AckCounts) of + {ok, AckCount1} -> AckCount1; + error -> 0 + end, + AckTarget = case dict:find(SegNum, Partials) of + {ok, PubCount} -> PubCount; + error -> ?SEGMENT_ENTRIES_COUNT + end, + AckCount2 = AckCount + length(Acks), + append_acks_to_segment(SegNum, AckCount2, Acks, AckTarget, State). + +append_acks_to_segment(SegNum, AckCount, _Acks, AckCount, State = + #qistate { publish_handle = PubHdl }) -> + PubHdl1 = case PubHdl of + %% If we're adjusting the pubhdl here then there + %% will be no entry in partials, thus the target ack + %% count must be SEGMENT_ENTRIES_COUNT + {SegNum, Hdl, AckCount = ?SEGMENT_ENTRIES_COUNT} + when Hdl /= undefined -> + {SegNum + 1, undefined, 0}; + _ -> PubHdl + end, + delete_segment(SegNum, State #qistate { publish_handle = PubHdl1 }); +append_acks_to_segment(_SegNum, _AckCount, [], _AckTarget, State) -> + State; +append_acks_to_segment(SegNum, AckCount, Acks, AckTarget, State = + #qistate { seg_ack_counts = AckCounts }) + when AckCount < AckTarget -> + {Hdl, State1} = append_to_segment(SegNum, Acks, State), + ok = file_handle_cache:sync(Hdl), + State1 #qistate { seg_ack_counts = + dict:store(SegNum, AckCount, AckCounts) }. + +append_dels_to_segment(SegNum, Dels, State) -> + {_Hdl, State1} = append_to_segment(SegNum, Dels, State), + State1. + +append_to_segment(SegNum, AcksOrDels, State) -> + {Hdl, State1} = get_seg_handle(SegNum, State), + ok = file_handle_cache:append( + Hdl, [<<?REL_SEQ_ONLY_PREFIX:?REL_SEQ_ONLY_PREFIX_BITS, + RelSeq:?REL_SEQ_BITS>> || RelSeq <- AcksOrDels ]), + {Hdl, State1}. diff --git a/src/rabbit_queue_prefetcher.erl b/src/rabbit_queue_prefetcher.erl new file mode 100644 index 0000000000..f5e717f55b --- /dev/null +++ b/src/rabbit_queue_prefetcher.erl @@ -0,0 +1,295 @@ +%% The contents of this file are subject to the Mozilla Public License +%% Version 1.1 (the "License"); you may not use this file except in +%% compliance with the License. You may obtain a copy of the License at +%% http://www.mozilla.org/MPL/ +%% +%% 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 Original Code is RabbitMQ. +%% +%% The Initial Developers of the Original Code are LShift Ltd, +%% Cohesive Financial Technologies LLC, and Rabbit Technologies Ltd. +%% +%% Portions created before 22-Nov-2008 00:00:00 GMT by LShift Ltd, +%% Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd +%% are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial +%% Technologies LLC, and Rabbit Technologies Ltd. +%% +%% Portions created by LShift Ltd are Copyright (C) 2007-2009 LShift +%% Ltd. Portions created by Cohesive Financial Technologies LLC are +%% Copyright (C) 2007-2009 Cohesive Financial Technologies +%% LLC. Portions created by Rabbit Technologies Ltd are Copyright +%% (C) 2007-2009 Rabbit Technologies Ltd. +%% +%% All Rights Reserved. +%% +%% Contributor(s): ______________________________________. +%% + +-module(rabbit_queue_prefetcher). + +-behaviour(gen_server2). + +-export([start_link/1]). + +-export([init/1, handle_call/3, handle_cast/2, handle_info/2, + terminate/2, code_change/3]). + +-export([publish/2, drain/1, drain_and_stop/1, stop/1]). + +-include("rabbit.hrl"). +-include("rabbit_queue.hrl"). + +-define(HIBERNATE_AFTER_MIN, 1000). +-define(DESIRED_HIBERNATE, 10000). + +-record(pstate, + { alphas, + betas, + queue_mref, + peruse_cb + }). + +%%---------------------------------------------------------------------------- +%% Novel +%%---------------------------------------------------------------------------- + +%% The design of the prefetcher is based on the following: +%% +%% a) It must issue low-priority (-ve) requests to the disk queue for +%% the next message. +%% b) If the prefetcher is empty and the amqqueue_process +%% (mixed_queue) asks it for a message, it must exit immediately, +%% telling the mixed_queue that it is empty so that the mixed_queue +%% can then take the more efficient path and communicate with the +%% disk_queue directly +%% c) No message can accidentally be delivered twice, or lost +%% d) The prefetcher must only cause load when the disk_queue is +%% otherwise idle, and must not worsen performance in a loaded +%% situation. +%% +%% As such, it's a little tricky. It must never issue a call to the +%% disk_queue - if it did, then that could potentially block, thus +%% causing pain to the mixed_queue that needs fast answers as to +%% whether the prefetcher has prefetched content or not. It behaves as +%% follows: +%% +%% 1) disk_queue:prefetch(Q) +%% This is a low priority cast +%% +%% 2) The disk_queue may pick up the cast, at which point it'll read +%% the next message and invoke prefetcher:publish(Msg) - normal +%% priority cast. Note that in the mean time, the mixed_queue could +%% have come along, found the prefetcher empty, asked it to +%% exit. This means the effective "reply" from the disk_queue will +%% go no where. As a result, the disk_queue should not advance the +%% queue. However, it does mark the messages as delivered. The +%% reasoning is that if it didn't, there would be the possibility +%% that the message was delivered without it being marked as such +%% on disk. We must maintain the property that a message which is +%% marked as non-redelivered really hasn't been delivered anywhere +%% before. The downside is that should the prefetcher not receive +%% this message, the queue will then fetch the message from the +%% disk_queue directly, and this message will have its delivered +%% bit set. The queue will not be advanced though - if it did +%% advance the queue and the msg was then lost, then the queue +%% would have lost a msg that the mixed_queue would not pick up. +%% +%% 3) The prefetcher hopefully receives the call from +%% prefetcher:publish(Msg). It replies immediately, and then adds +%% to its internal queue. A cast is not sufficient as a pseudo +%% "reply" here because the mixed_queue could come along, drain the +%% prefetcher, thus catching the msg just sent by the disk_queue +%% and then call disk_queue:fetch(Q) which is normal priority call, +%% which could overtake a reply cast from the prefetcher to the +%% disk queue, resulting in the same message being delivered +%% twice. Thus when the disk_queue calls prefetcher:publish(Msg), +%% it is briefly blocked. However, a) the prefetcher replies +%% immediately, and b) the prefetcher should never have more than +%% two items in its mailbox anyway (one from the queue process / +%% mixed_queue and one from the disk_queue), so this should not +%% cause a problem to the disk_queue. +%% +%% 4) The disk_queue receives the reply, and advances the Q to the +%% next msg. +%% +%% 5) If the prefetcher has not met its target then it goes back to +%% 1). Otherwise it just sits and waits for the mixed_queue to +%% drain it. +%% +%% Now at some point, the mixed_queue will come along and will call +%% prefetcher:drain() - normal priority call. The prefetcher then +%% replies with its internal queue and a flag saying if the prefetcher +%% has finished or is continuing; if the prefetch target was reached, +%% the prefetcher stops normally at this point. If it hasn't been +%% reached, then the prefetcher continues to hang around (it almost +%% certainly has issued a disk_queue:prefetch(Q) cast and is waiting +%% for a reply from the disk_queue). +%% +%% If the mixed_queue calls prefetcher:drain() and the prefetcher's +%% internal queue is empty then the prefetcher replies with 'empty', +%% and it exits. This informs the mixed_queue that it should from now +%% on talk directly with the disk_queue and not via the +%% prefetcher. This is more efficient and the mixed_queue will use +%% normal priority blocking calls to the disk_queue and thus get +%% better service. +%% +%% The prefetcher may at this point have issued a +%% disk_queue:prefetch(Q) cast which has not yet been picked up by the +%% disk_queue. This msg won't go away and the disk_queue will +%% eventually find it. However, when it does, it'll simply read the +%% next message from the queue (which could now be empty), possibly +%% populate the cache (no harm done), mark the message as delivered +%% (oh well, not a spec violation, and better than the alternative) +%% and try and call prefetcher:publish(Msg) which will result in an +%% error, which the disk_queue catches, as the publish call is to a +%% non-existant process. However, the state of the queue has not been +%% altered so the mixed_queue will be able to fetch this message as if +%% it had never been prefetched. +%% +%% The only point at which the queue is advanced is when the +%% prefetcher replies to the publish call. At this point the message +%% has been received by the prefetcher and so we guarantee it will be +%% passed to the mixed_queue when the mixed_queue tries to drain the +%% prefetcher. We must therefore ensure that this msg can't also be +%% delivered to the mixed_queue directly by the disk_queue through the +%% mixed_queue calling disk_queue:fetch(Q) which is why the +%% prefetcher:publish function is a call and not a cast, thus blocking +%% the disk_queue. +%% +%% Finally, the prefetcher is only created when the mixed_queue is +%% operating in mixed mode and it sees that the next N messages are +%% all on disk, and the queue process is about to hibernate. During +%% this phase, the mixed_queue can be asked to go back to disk_only +%% mode. When this happens, it calls prefetcher:drain_and_stop() which +%% behaves like two consecutive calls to drain() - i.e. replies with +%% all prefetched messages and causes the prefetcher to exit. +%% +%% Note there is a flaw here in that we end up marking messages which +%% have come through the prefetcher as delivered even if they don't +%% get delivered (e.g. prefetcher fetches them, then broker +%% dies). However, the alternative is that the mixed_queue must do a +%% call to the disk_queue when it effectively passes them out to the +%% rabbit_writer. This would hurt performance, and even at that stage, +%% we have no guarantee that the message will really go out of the +%% socket. What we do still have is that messages which have the +%% redelivered bit set false really are guaranteed to have not been +%% delivered already. + +%%---------------------------------------------------------------------------- + +-ifdef(use_specs). + +-spec(start_link/1 :: (queue()) -> + ({'ok', pid()} | 'ignore' | {'error', any()})). +-spec(publish/2 :: (pid(), (message()| 'not_found')) -> 'ok'). +-spec(drain/1 :: (pid()) -> ({('finished' | 'continuing' | 'empty'), queue()})). +-spec(drain_and_stop/1 :: (pid()) -> ({('empty' | queue()), queue()})). +-spec(stop/1 :: (pid()) -> 'ok'). + +-endif. + +%%---------------------------------------------------------------------------- + +start_link(Betas) -> + false = queue:is_empty(Betas), %% ASSERTION + gen_server2:start_link(?MODULE, [Betas, self()], []). + +publish(Prefetcher, Obj = #basic_message {}) -> + gen_server2:call(Prefetcher, {publish, Obj}, infinity); +publish(Prefetcher, not_found) -> + gen_server2:call(Prefetcher, publish_empty, infinity). + +drain(Prefetcher) -> + gen_server2:call(Prefetcher, drain, infinity). + +drain_and_stop(Prefetcher) -> + gen_server2:call(Prefetcher, drain_and_stop, infinity). + +stop(Prefetcher) -> + gen_server2:call(Prefetcher, stop, infinity). + +%%---------------------------------------------------------------------------- + +init([Betas, QPid]) when is_pid(QPid) -> + %% link isn't enough because the signal will not appear if the + %% queue exits normally. Thus have to use monitor. + MRef = erlang:monitor(process, QPid), + Self = self(), + CB = fun (Result) -> + rabbit_misc:with_exit_handler( + fun () -> ok end, + fun () -> case Result of + {ok, Msg} -> publish(Self, Msg); + not_found -> publish(Self, not_found) + end + end) + end, + State = #pstate { alphas = queue:new(), + betas = Betas, + queue_mref = MRef, + peruse_cb = CB + }, + {ok, prefetch(State), infinity, {backoff, ?HIBERNATE_AFTER_MIN, + ?HIBERNATE_AFTER_MIN, ?DESIRED_HIBERNATE}}. + +handle_call({publish, Msg = #basic_message { guid = MsgId, + is_persistent = IsPersistent }}, + DiskQueue, State = #pstate { alphas = Alphas, betas = Betas }) -> + gen_server2:reply(DiskQueue, ok), + {{value, #beta { msg_id = MsgId, seq_id = SeqId, + is_persistent = IsPersistent, + is_delivered = IsDelivered, + index_on_disk = IndexOnDisk}}, Betas1} = queue:out(Betas), + Alphas1 = queue:in(#alpha { msg = Msg, seq_id = SeqId, + is_delivered = IsDelivered, msg_on_disk = true, + index_on_disk = IndexOnDisk }, Alphas), + State1 = State #pstate { alphas = Alphas1, betas = Betas1 }, + {Timeout, State2} = case queue:is_empty(Betas1) of + true -> {hibernate, State1}; + false -> {infinity, prefetch(State1)} + end, + {noreply, State2, Timeout}; +handle_call(publish_empty, _From, State) -> + %% Very odd. This could happen if the queue is deleted or purged + %% and the mixed queue fails to shut us down. + {reply, ok, State, hibernate}; +handle_call(drain, _From, State = #pstate { alphas = Alphas, betas = Betas }) -> + case {queue:is_empty(Betas), queue:is_empty(Alphas)} of + {true , _ } -> {stop, normal, {finished, Alphas}, State}; + {false, true } -> {stop, normal, {empty, Betas}, State}; + {false, false} -> {reply, {continuing, Alphas}, + State #pstate { alphas = queue:new() }} + end; +handle_call(drain_and_stop, _From, State = #pstate { alphas = Alphas, + betas = Betas }) -> + Res = case queue:is_empty(Alphas) of + true -> {empty, Betas}; + false -> {Alphas, Betas} + end, + {stop, normal, Res, State}; +handle_call(stop, _From, State) -> + {stop, normal, ok, State}. + +handle_cast(Msg, State) -> + exit({unexpected_message_cast_to_prefetcher, Msg, State}). + +handle_info({'DOWN', MRef, process, _Pid, _Reason}, + State = #pstate { queue_mref = MRef }) -> + %% this is the amqqueue_process going down, so we should go down + %% too + {stop, normal, State}. + +terminate(_Reason, _State) -> + ok. + +code_change(_OldVsn, State, _Extra) -> + {ok, State}. + +prefetch(State = #pstate { betas = Betas, peruse_cb = CB }) -> + {{value, #beta { msg_id = MsgId }}, _Betas1} = queue:out(Betas), + ok = rabbit_msg_store:peruse(MsgId, CB), + State. diff --git a/src/rabbit_tests.erl b/src/rabbit_tests.erl index c5a7d05eb4..d74f998e33 100644 --- a/src/rabbit_tests.erl +++ b/src/rabbit_tests.erl @@ -31,6 +31,8 @@ -module(rabbit_tests). +-compile(export_all). + -export([all_tests/0, test_parsing/0]). %% Exported so the hook mechanism can call back @@ -48,6 +50,9 @@ test_content_prop_roundtrip(Datum, Binary) -> Binary = rabbit_binary_generator:encode_properties(Types, Values). %% assertion all_tests() -> + passed = test_msg_store(), + passed = test_queue_index(), + passed = test_variable_queue(), passed = test_priority_queue(), passed = test_unfold(), passed = test_parsing(), @@ -168,7 +173,6 @@ priority_queue_out_all(Q) -> {empty, _} -> []; {{value, V}, Q1} -> [V | priority_queue_out_all(Q1)] end. - test_priority_queue(Q) -> {priority_queue:is_queue(Q), priority_queue:is_empty(Q), @@ -819,3 +823,695 @@ bad_handle_hook(_, _, _) -> bad:bad(). extra_arg_hook(Hookname, Handler, Args, Extra1, Extra2) -> handle_hook(Hookname, Handler, {Args, Extra1, Extra2}). + +msg_store_dir() -> + filename:join(rabbit_mnesia:dir(), "msg_store"). + +start_msg_store_empty() -> + start_msg_store(fun (ok) -> finished end, ok). + +start_msg_store(MsgRefDeltaGen, MsgRefDeltaGenInit) -> + rabbit:start_child(rabbit_msg_store, [msg_store_dir(), MsgRefDeltaGen, + MsgRefDeltaGenInit]). + +stop_msg_store() -> + case supervisor:terminate_child(rabbit_sup, rabbit_msg_store) of + ok -> supervisor:delete_child(rabbit_sup, rabbit_msg_store); + E -> E + end. + +msg_id_bin(X) -> + erlang:md5(term_to_binary(X)). + +msg_store_contains(Atom, MsgIds) -> + Atom = lists:foldl( + fun (MsgId, Atom1) when Atom1 =:= Atom -> + rabbit_msg_store:contains(MsgId) end, Atom, MsgIds). + +msg_store_sync(MsgIds) -> + Ref = make_ref(), + Self = self(), + ok = rabbit_msg_store:sync(MsgIds, + fun () -> Self ! {sync, Ref} end), + receive + {sync, Ref} -> ok + after + 10000 -> + io:format("Sync from msg_store missing for msg_ids ~p~n", [MsgIds]), + throw(timeout) + end. + +msg_store_read(MsgIds) -> + ok = + lists:foldl( + fun (MsgId, ok) -> {ok, MsgId} = rabbit_msg_store:read(MsgId), ok end, + ok, MsgIds). + +msg_store_write(MsgIds) -> + ok = lists:foldl( + fun (MsgId, ok) -> rabbit_msg_store:write(MsgId, MsgId) end, + ok, MsgIds). + +test_msg_store() -> + stop_msg_store(), + ok = start_msg_store_empty(), + Self = self(), + MsgIds = [msg_id_bin(M) || M <- lists:seq(1,100)], + {MsgIds1stHalf, MsgIds2ndHalf} = lists:split(50, MsgIds), + %% check we don't contain any of the msgs we're about to publish + false = msg_store_contains(false, MsgIds), + %% publish the first half + ok = msg_store_write(MsgIds1stHalf), + %% sync on the first half + ok = msg_store_sync(MsgIds1stHalf), + %% publish the second half + ok = msg_store_write(MsgIds2ndHalf), + %% sync on the first half again - the msg_store will be dirty, but + %% we won't need the fsync + ok = msg_store_sync(MsgIds1stHalf), + %% check they're all in there + true = msg_store_contains(true, MsgIds), + %% publish the latter half twice so we hit the caching and ref count code + ok = msg_store_write(MsgIds2ndHalf), + %% check they're still all in there + true = msg_store_contains(true, MsgIds), + %% sync on the 2nd half, but do lots of individual syncs to try + %% and cause coalescing to happen + ok = lists:foldl( + fun (MsgId, ok) -> rabbit_msg_store:sync( + [MsgId], fun () -> Self ! {sync, MsgId} end) + end, ok, MsgIds2ndHalf), + lists:foldl( + fun(MsgId, ok) -> + receive + {sync, MsgId} -> ok + after + 10000 -> + io:format("Sync from msg_store missing (msg_id: ~p)~n", + [MsgId]), + throw(timeout) + end + end, ok, MsgIds2ndHalf), + %% it's very likely we're not dirty here, so the 1st half sync + %% should hit a different code path + ok = msg_store_sync(MsgIds1stHalf), + %% read them all + ok = msg_store_read(MsgIds), + %% read them all again - this will hit the cache, not disk + ok = msg_store_read(MsgIds), + %% remove them all + ok = rabbit_msg_store:remove(MsgIds), + %% check first half doesn't exist + false = msg_store_contains(false, MsgIds1stHalf), + %% check second half does exist + true = msg_store_contains(true, MsgIds2ndHalf), + %% read the second half again + ok = msg_store_read(MsgIds2ndHalf), + %% release the second half, just for fun (aka code coverage) + ok = rabbit_msg_store:release(MsgIds2ndHalf), + %% read the second half again, just for fun (aka code coverage) + ok = msg_store_read(MsgIds2ndHalf), + %% read the second half via peruse + lists:foldl( + fun (MsgId, ok) -> + rabbit_msg_store:peruse(MsgId, + fun ({ok, MsgId1}) when MsgId1 == MsgId -> + Self ! {peruse, MsgId1} + end), + receive + {peruse, MsgId} -> + ok + after + 10000 -> + io:format("Failed to receive response via peruse~n"), + throw(timeout) + end + end, ok, MsgIds2ndHalf), + %% stop and restart, preserving every other msg in 2nd half + ok = stop_msg_store(), + ok = start_msg_store(fun ([]) -> finished; + ([MsgId|MsgIdsTail]) + when length(MsgIdsTail) rem 2 == 0 -> + {MsgId, 1, MsgIdsTail}; + ([MsgId|MsgIdsTail]) -> + {MsgId, 0, MsgIdsTail} + end, MsgIds2ndHalf), + %% check we have the right msgs left + lists:foldl( + fun (MsgId, Bool) -> + not(Bool = rabbit_msg_store:contains(MsgId)) + end, false, MsgIds2ndHalf), + %% restart empty + ok = stop_msg_store(), + ok = start_msg_store_empty(), + %% check we don't contain any of the msgs + false = msg_store_contains(false, MsgIds), + %% publish the first half again + ok = msg_store_write(MsgIds1stHalf), + %% this should force some sort of sync internally otherwise misread + ok = msg_store_read(MsgIds1stHalf), + ok = rabbit_msg_store:remove(MsgIds1stHalf), + %% push a lot of msgs in... + BigCount = 100000, + MsgIdsBig = lists:seq(1, BigCount), + Payload = << 0:65536 >>, + ok = lists:foldl( + fun (MsgId, ok) -> + rabbit_msg_store:write(msg_id_bin(MsgId), Payload) + end, ok, MsgIdsBig), + %% .., then remove even numbers ascending, and odd numbers + %% descending. This hits the GC. + ok = lists:foldl( + fun (MsgId, ok) -> + rabbit_msg_store:remove([msg_id_bin( + case MsgId rem 2 of + 0 -> MsgId; + 1 -> BigCount - MsgId + end)]) + end, ok, MsgIdsBig), + %% ensure empty + false = msg_store_contains(false, [msg_id_bin(M) || M <- MsgIdsBig]), + %% restart empty + ok = stop_msg_store(), + ok = start_msg_store_empty(), + passed. + +queue_name(Name) -> + rabbit_misc:r(<<"/">>, queue, term_to_binary(Name)). + +test_queue() -> + queue_name(test). + +test_amqqueue(Durable) -> + #amqqueue{name = test_queue(), + durable = Durable, + auto_delete = true, + arguments = [], + pid = none}. + +empty_test_queue() -> + ok = rabbit_queue_index:start_msg_store([]), + {0, Qi1} = rabbit_queue_index:init(test_queue()), + _Qi2 = rabbit_queue_index:terminate_and_erase(Qi1), + ok. + +queue_index_publish(SeqIds, Persistent, Qi) -> + lists:foldl( + fun (SeqId, {QiN, SeqIdsMsgIdsAcc}) -> + MsgId = rabbit_guid:guid(), + QiM = rabbit_queue_index:write_published(MsgId, SeqId, Persistent, + QiN), + ok = rabbit_msg_store:write(MsgId, MsgId), + {QiM, [{SeqId, MsgId} | SeqIdsMsgIdsAcc]} + end, {Qi, []}, SeqIds). + +queue_index_deliver(SeqIds, Qi) -> + lists:foldl( + fun (SeqId, QiN) -> + rabbit_queue_index:write_delivered(SeqId, QiN) + end, Qi, SeqIds). + +queue_index_flush_journal(Qi) -> + {_Oks, {false, Qi1}} = + rabbit_misc:unfold( + fun ({true, QiN}) -> + QiM = rabbit_queue_index:flush_journal(QiN), + {true, ok, {rabbit_queue_index:can_flush_journal(QiM), QiM}}; + ({false, _QiN}) -> + false + end, {true, Qi}), + Qi1. + +verify_read_with_published(_Delivered, _Persistent, [], _) -> + ok; +verify_read_with_published(Delivered, Persistent, + [{MsgId, SeqId, Persistent, Delivered}|Read], + [{SeqId, MsgId}|Published]) -> + verify_read_with_published(Delivered, Persistent, Read, Published); +verify_read_with_published(_Delivered, _Persistent, _Read, _Published) -> + ko. + +test_queue_index() -> + stop_msg_store(), + ok = empty_test_queue(), + SeqIdsA = lists:seq(0,9999), + SeqIdsB = lists:seq(10000,19999), + {0, Qi0} = rabbit_queue_index:init(test_queue()), + {0, 0, Qi1} = + rabbit_queue_index:find_lowest_seq_id_seg_and_next_seq_id(Qi0), + {Qi2, SeqIdsMsgIdsA} = queue_index_publish(SeqIdsA, false, Qi1), + {0, 10000, Qi3} = + rabbit_queue_index:find_lowest_seq_id_seg_and_next_seq_id(Qi2), + {ReadA, Qi4} = rabbit_queue_index:read_segment_entries(0, Qi3), + ok = verify_read_with_published(false, false, ReadA, + lists:reverse(SeqIdsMsgIdsA)), + %% call terminate twice to prove it's idempotent + _Qi5 = rabbit_queue_index:terminate(rabbit_queue_index:terminate(Qi4)), + ok = stop_msg_store(), + ok = rabbit_queue_index:start_msg_store([test_amqqueue(true)]), + %% should get length back as 0, as all the msgs were transient + {0, Qi6} = rabbit_queue_index:init(test_queue()), + false = rabbit_queue_index:can_flush_journal(Qi6), + {0, 10000, Qi7} = + rabbit_queue_index:find_lowest_seq_id_seg_and_next_seq_id(Qi6), + {Qi8, SeqIdsMsgIdsB} = queue_index_publish(SeqIdsB, true, Qi7), + {0, 20000, Qi9} = + rabbit_queue_index:find_lowest_seq_id_seg_and_next_seq_id(Qi8), + {ReadB, Qi10} = rabbit_queue_index:read_segment_entries(0, Qi9), + ok = verify_read_with_published(false, true, ReadB, + lists:reverse(SeqIdsMsgIdsB)), + _Qi11 = rabbit_queue_index:terminate(Qi10), + ok = stop_msg_store(), + ok = rabbit_queue_index:start_msg_store([test_amqqueue(true)]), + %% should get length back as 10000 + LenB = length(SeqIdsB), + {LenB, Qi12} = rabbit_queue_index:init(test_queue()), + {0, 20000, Qi13} = + rabbit_queue_index:find_lowest_seq_id_seg_and_next_seq_id(Qi12), + Qi14 = queue_index_deliver(SeqIdsB, Qi13), + {ReadC, Qi15} = rabbit_queue_index:read_segment_entries(0, Qi14), + ok = verify_read_with_published(true, true, ReadC, + lists:reverse(SeqIdsMsgIdsB)), + Qi16 = rabbit_queue_index:write_acks(SeqIdsB, Qi15), + true = rabbit_queue_index:can_flush_journal(Qi16), + Qi17 = rabbit_queue_index:flush_journal(Qi16), + %% the entire first segment will have gone as they were firstly + %% transient, and secondly ack'd + SegmentSize = rabbit_queue_index:segment_size(), + {SegmentSize, 20000, Qi18} = + rabbit_queue_index:find_lowest_seq_id_seg_and_next_seq_id(Qi17), + _Qi19 = rabbit_queue_index:terminate(Qi18), + ok = stop_msg_store(), + ok = rabbit_queue_index:start_msg_store([test_amqqueue(true)]), + %% should get length back as 0 because all persistent msgs have been acked + {0, Qi20} = rabbit_queue_index:init(test_queue()), + _Qi21 = rabbit_queue_index:terminate_and_erase(Qi20), + ok = stop_msg_store(), + ok = empty_test_queue(), + + %% These next bits are just to hit the auto deletion of segment files. + %% First, partials: + %% a) partial pub+del+ack, then move to new segment + SeqIdsC = lists:seq(0,trunc(SegmentSize/2)), + {0, Qi22} = rabbit_queue_index:init(test_queue()), + {Qi23, _SeqIdsMsgIdsC} = queue_index_publish(SeqIdsC, false, Qi22), + Qi24 = queue_index_deliver(SeqIdsC, Qi23), + Qi25 = rabbit_queue_index:write_acks(SeqIdsC, Qi24), + Qi26 = queue_index_flush_journal(Qi25), + {Qi27, _SeqIdsMsgIdsC1} = queue_index_publish([SegmentSize], false, Qi26), + _Qi28 = rabbit_queue_index:terminate_and_erase(Qi27), + ok = stop_msg_store(), + ok = empty_test_queue(), + + %% b) partial pub+del, then move to new segment, then ack all in old segment + {0, Qi29} = rabbit_queue_index:init(test_queue()), + {Qi30, _SeqIdsMsgIdsC2} = queue_index_publish(SeqIdsC, false, Qi29), + Qi31 = queue_index_deliver(SeqIdsC, Qi30), + {Qi32, _SeqIdsMsgIdsC3} = queue_index_publish([SegmentSize], false, Qi31), + Qi33 = rabbit_queue_index:write_acks(SeqIdsC, Qi32), + Qi34 = queue_index_flush_journal(Qi33), + _Qi35 = rabbit_queue_index:terminate_and_erase(Qi34), + ok = stop_msg_store(), + ok = empty_test_queue(), + + %% c) just fill up several segments of all pubs, then +dels, then +acks + SeqIdsD = lists:seq(0,SegmentSize*4), + {0, Qi36} = rabbit_queue_index:init(test_queue()), + {Qi37, _SeqIdsMsgIdsD} = queue_index_publish(SeqIdsD, false, Qi36), + Qi38 = queue_index_deliver(SeqIdsD, Qi37), + Qi39 = rabbit_queue_index:write_acks(SeqIdsD, Qi38), + Qi40 = queue_index_flush_journal(Qi39), + _Qi41 = rabbit_queue_index:terminate_and_erase(Qi40), + ok = stop_msg_store(), + ok = rabbit_queue_index:start_msg_store([]), + ok = stop_msg_store(), + passed. + +variable_queue_publish(IsPersistent, Count, VQ) -> + lists:foldl( + fun (_N, {Acc, VQ1}) -> + {SeqId, VQ2} = rabbit_variable_queue:publish( + rabbit_basic:message( + rabbit_misc:r(<<>>, exchange, <<>>), + <<>>, [], <<>>, rabbit_guid:guid(), + IsPersistent), VQ1), + {[SeqId | Acc], VQ2} + end, {[], VQ}, lists:seq(1, Count)). + +variable_queue_fetch(Count, IsPersistent, IsDelivered, Len, VQ) -> + lists:foldl(fun (N, {VQN, AckTagsAcc}) -> + Rem = Len - N, + {{#basic_message { is_persistent = IsPersistent }, + IsDelivered, AckTagN, Rem}, VQM} = + rabbit_variable_queue:fetch(VQN), + {VQM, [AckTagN | AckTagsAcc]} + end, {VQ, []}, lists:seq(1, Count)). + +assert_prop(List, Prop, Value) -> + Value = proplists:get_value(Prop, List). + +fresh_variable_queue() -> + stop_msg_store(), + ok = empty_test_queue(), + VQ = rabbit_variable_queue:init(test_queue()), + S0 = rabbit_variable_queue:status(VQ), + assert_prop(S0, len, 0), + assert_prop(S0, prefetching, false), + assert_prop(S0, q1, 0), + assert_prop(S0, q2, 0), + assert_prop(S0, gamma, {gamma, undefined, 0}), + assert_prop(S0, q3, 0), + assert_prop(S0, q4, 0), + VQ. + +test_variable_queue() -> + passed = test_variable_queue_prefetching_and_gammas_to_betas(), + passed = test_variable_queue_prefetching_during_publish(0), + passed = test_variable_queue_prefetching_during_publish(5000), + passed = test_variable_queue_prefetch_evicts_q1(), + passed = test_variable_queue_dynamic_duration_change(), + passed. + +test_variable_queue_dynamic_duration_change() -> + SegmentSize = rabbit_queue_index:segment_size(), + VQ0 = fresh_variable_queue(), + %% start by sending in a couple of segments worth + Len1 = 2*SegmentSize, + {_SeqIds, VQ1} = variable_queue_publish(false, Len1, VQ0), + VQ2 = rabbit_variable_queue:remeasure_egress_rate(VQ1), + {ok, _TRef} = timer:send_after(1000, {duration, 60, + fun (V) -> (V*0.75)-1 end}), + VQ3 = test_variable_queue_dynamic_duration_change_f(Len1, VQ2), + {VQ4, AckTags} = variable_queue_fetch(Len1, false, false, Len1, VQ3), + VQ5 = rabbit_variable_queue:ack(AckTags, VQ4), + {empty, VQ6} = rabbit_variable_queue:fetch(VQ5), + + %% just publish and fetch some persistent msgs, this hits the the + %% partial segment path in queue_index due to the period when + %% duration was 0 and the entire queue was gamma. + {_SeqIds1, VQ7} = variable_queue_publish(true, 20, VQ6), + {VQ8, AckTags1} = variable_queue_fetch(20, true, false, 20, VQ7), + VQ9 = rabbit_variable_queue:ack(AckTags1, VQ8), + VQ10 = rabbit_variable_queue:flush_journal(VQ9), + VQ11 = rabbit_variable_queue:flush_journal(VQ10), + {empty, VQ12} = rabbit_variable_queue:fetch(VQ11), + + rabbit_variable_queue:terminate(VQ12), + + passed. + +test_variable_queue_dynamic_duration_change_f(Len, VQ0) -> + {_SeqIds, VQ1} = variable_queue_publish(false, 1, VQ0), + {{_Msg, false, AckTag, Len}, VQ2} = rabbit_variable_queue:fetch(VQ1), + VQ3 = rabbit_variable_queue:ack([AckTag], VQ2), + receive + {duration, _, stop} -> + VQ3; + {duration, N, Fun} -> + N1 = lists:max([Fun(N), 0]), + Fun1 = case N1 of + 0 -> fun (V) -> (V+1)/0.75 end; + _ when N1 > 400 -> stop; + _ -> Fun + end, + {ok, _TRef} = timer:send_after(1000, {duration, N1, Fun1}), + VQ4 = rabbit_variable_queue:remeasure_egress_rate(VQ3), + VQ5 = %% /37 otherwise the duration is just to high to stress things + rabbit_variable_queue:set_queue_ram_duration_target(N/37, VQ4), + io:format("~p:~n~p~n~n", [N, rabbit_variable_queue:status(VQ5)]), + test_variable_queue_dynamic_duration_change_f(Len, VQ5) + after 0 -> + test_variable_queue_dynamic_duration_change_f(Len, VQ3) + end. + +test_variable_queue_prefetch_evicts_q1() -> + SegmentSize = rabbit_queue_index:segment_size(), + VQ0 = fresh_variable_queue(), + VQ1 = rabbit_variable_queue:set_queue_ram_duration_target(0, VQ0), + assert_prop(rabbit_variable_queue:status(VQ1), target_ram_msg_count, 0), + Len1 = 2*SegmentSize, + {_SeqIds, VQ2} = variable_queue_publish(true, Len1, VQ1), + %% one segment will be in q3, the other in gamma. We want to fetch + %% all of q3 so that gamma is then moved into q3, emptying gamma + + VQ3 = rabbit_variable_queue:remeasure_egress_rate(VQ2), + Start = now(), + {VQ4, AckTags} = variable_queue_fetch(SegmentSize, true, false, Len1, VQ3), + End = now(), + VQ5 = rabbit_variable_queue:ack(AckTags, VQ4), + S5 = rabbit_variable_queue:status(VQ5), + assert_prop(S5, q4, 0), + assert_prop(S5, q3, SegmentSize), + assert_prop(S5, gamma, {gamma, undefined, 0}), + assert_prop(S5, len, SegmentSize), + assert_prop(S5, prefetching, false), + + VQ6 = rabbit_variable_queue:remeasure_egress_rate(VQ5), + %% half the seconds taken to fetch one segment + Duration = timer:now_diff(End, Start) / 2000000, + VQ7 = rabbit_variable_queue:set_queue_ram_duration_target(Duration, VQ6), + S7 = rabbit_variable_queue:status(VQ7), + assert_prop(S7, q4, 0), + Q3 = proplists:get_value(q3, S7), + true = Q3 > 0, %% not prefetching everything + assert_prop(S7, gamma, {gamma, undefined, 0}), + assert_prop(S7, len, SegmentSize), + assert_prop(S7, prefetching, true), + + %% now publish a segment, this'll go half in q1, half in q3, in + %% theory. + {_SeqIds1, VQ8} = variable_queue_publish(true, SegmentSize, VQ7), + S8 = rabbit_variable_queue:status(VQ8), + assert_prop(S8, q4, 0), + assert_prop(S8, q2, 0), + assert_prop(S8, len, Len1), + assert_prop(S8, prefetching, true), + Q3a = proplists:get_value(q3, S8), + Q3a_new = Q3a - Q3, + Q1a = proplists:get_value(q1, S8), + true = (Q3a_new + Q1a == SegmentSize) andalso Q1a < SegmentSize, + + %% wait a bit, to let the prefetcher do its thing + timer:sleep(2000), + %% fetch a msg. The prefetcher *should* have finished, but can't + %% guarantee it. + Len2 = Len1-1, + {{_Msg, false, AckTag, Len2}, VQ9} = rabbit_variable_queue:fetch(VQ8), + S9 = rabbit_variable_queue:status(VQ9), + case proplists:get_value(prefetching, S9) of + true -> + %% bits of q1 could have moved into q3, and the prefetcher + %% won't have returned any betas for q3. So q3 can not + %% have shrunk. + Q3b = proplists:get_value(q3, S9), + Q1b = proplists:get_value(q1, S9), + true = (Q1a + Q3a) == (Q1b + Q3b) andalso Q3b >= Q3a; + false -> + %% there should be content in q4 and q3 (we only did 1 + %% fetch. This is not sufficient to kill the prefetcher + %% through draining it when it's empty, thus if it's not + %% running, it must have finished, not been killed, thus + %% q4 will not be empty), and q1 should have gone into q3. + Q1b = proplists:get_value(q1, S9), + Q3b = proplists:get_value(q3, S9), + Q4b = proplists:get_value(q4, S9), + NotPrefetched = Q3b - (SegmentSize - Q1b), + SegmentSize = NotPrefetched + Q4b + 1 %% we fetched one + end, + + %% just for the fun of it, set duration to 0. This should push + %% everything back into gamma, except the eldest (partial) segment + %% in q3 + VQ10 = rabbit_variable_queue:set_queue_ram_duration_target(0, VQ9), + S10 = rabbit_variable_queue:status(VQ10), + assert_prop(S10, len, Len2), + assert_prop(S10, prefetching, false), + assert_prop(S10, q1, 0), + assert_prop(S10, q2, 0), + assert_prop(S10, gamma, {gamma, Len1, SegmentSize}), + assert_prop(S10, q3, (Len2 - SegmentSize)), + assert_prop(S10, q4, 0), + + {VQ11, AckTags1} = variable_queue_fetch(Len2, true, false, Len2, VQ10), + VQ12 = rabbit_variable_queue:ack([AckTag|AckTags1], VQ11), + {empty, VQ13} = rabbit_variable_queue:fetch(VQ12), + rabbit_variable_queue:terminate(VQ13), + + passed. + +test_variable_queue_prefetching_during_publish(PrefetchDelay) -> + SegmentSize = rabbit_queue_index:segment_size(), + VQ0 = fresh_variable_queue(), + VQ1 = rabbit_variable_queue:set_queue_ram_duration_target(0, VQ0), + assert_prop(rabbit_variable_queue:status(VQ1), target_ram_msg_count, 0), + + Len1 = 2*SegmentSize, + {_SeqIds, VQ2} = variable_queue_publish(true, Len1, VQ1), + %% one segment will be in q3, the other in gamma. We want to fetch + %% all of q3 so that gamma is then moved into q3, emptying gamma + + VQ3 = rabbit_variable_queue:remeasure_egress_rate(VQ2), + {VQ4, AckTags} = variable_queue_fetch(SegmentSize, true, false, Len1, VQ3), + VQ5 = rabbit_variable_queue:ack(AckTags, VQ4), + S5 = rabbit_variable_queue:status(VQ5), + assert_prop(S5, q4, 0), + assert_prop(S5, q3, SegmentSize), + assert_prop(S5, gamma, {gamma, undefined, 0}), + assert_prop(S5, len, SegmentSize), + assert_prop(S5, prefetching, false), + + %% we assume that we can fetch at > 1 msg a second + VQ6 = rabbit_variable_queue:remeasure_egress_rate(VQ5), + VQ7 = rabbit_variable_queue:set_queue_ram_duration_target(Len1, VQ6), + S7 = rabbit_variable_queue:status(VQ7), + assert_prop(S7, q4, 0), + assert_prop(S7, q3, 0), + assert_prop(S7, gamma, {gamma, undefined, 0}), + assert_prop(S7, len, SegmentSize), + assert_prop(S7, prefetching, true), + + timer:sleep(PrefetchDelay), + + {_SeqIds1, VQ8} = variable_queue_publish(true, SegmentSize, VQ7), + S8 = rabbit_variable_queue:status(VQ8), + assert_prop(S8, q4, 0), + assert_prop(S8, q2, 0), + assert_prop(S8, q1, SegmentSize), + assert_prop(S8, len, Len1), + assert_prop(S8, prefetching, true), + + {VQ9, AckTags1} = + variable_queue_fetch(SegmentSize-1, true, false, Len1, VQ8), + VQ10 = rabbit_variable_queue:ack(AckTags1, VQ9), + %% can't guarantee the prefetcher has stopped here. If it is still + %% running, then we must have SegmentSize is q1. If it's not + %% running, and it completed, then we'll find SegmentSize + 1 in + %% q4 (q1 will have been joined to q4), otherwise, we'll find + %% SegmentSize in q1 and 1 in q3 and q4 empty. + S10 = rabbit_variable_queue:status(VQ10), + assert_prop(S10, q2, 0), + assert_prop(S10, len, (SegmentSize+1)), + case proplists:get_value(prefetching, S10) of + true -> assert_prop(S10, q1, SegmentSize), + assert_prop(S10, q3, 0), + assert_prop(S10, q4, 0); + false -> case proplists:get_value(q3, S10) of + 0 -> assert_prop(S10, q4, SegmentSize+1), + assert_prop(S10, q1, 0); + 1 -> assert_prop(S10, q4, 0), + assert_prop(S10, q1, SegmentSize) + end + end, + + {VQ11, AckTags2} = + variable_queue_fetch(SegmentSize+1, true, false, SegmentSize+1, VQ10), + VQ12 = rabbit_variable_queue:ack(AckTags2, VQ11), + + {empty, VQ13} = rabbit_variable_queue:fetch(VQ12), + rabbit_variable_queue:terminate(VQ13), + + passed. + +test_variable_queue_prefetching_and_gammas_to_betas() -> + SegmentSize = rabbit_queue_index:segment_size(), + VQ0 = fresh_variable_queue(), + + VQ1 = rabbit_variable_queue:set_queue_ram_duration_target(10, VQ0), + assert_prop(rabbit_variable_queue:status(VQ1), target_ram_msg_count, 0), + + {_SeqIds, VQ2} = variable_queue_publish(false, 3 * SegmentSize, VQ1), + S2 = rabbit_variable_queue:status(VQ2), + assert_prop(S2, gamma, {gamma, SegmentSize, 2*SegmentSize}), + assert_prop(S2, q3, SegmentSize), + assert_prop(S2, len, 3*SegmentSize), + + VQ3 = rabbit_variable_queue:remeasure_egress_rate(VQ2), + Len1 = 3*SegmentSize - 1, + {{_Msg, false, AckTag, Len1}, VQ4} = rabbit_variable_queue:fetch(VQ3), + timer:sleep(1000), + VQ5 = rabbit_variable_queue:remeasure_egress_rate(VQ4), + VQ6 = rabbit_variable_queue:set_queue_ram_duration_target(10, VQ5), + timer:sleep(1000), %% let the prefetcher run and grab enough - about 4 msgs + S6 = rabbit_variable_queue:status(VQ6), + RamCount = proplists:get_value(target_ram_msg_count, S6), + assert_prop(S6, prefetching, true), + assert_prop(S6, q4, 0), + assert_prop(S6, q3, (Len1 - RamCount)), + assert_prop(S6, gamma, {gamma, undefined, 0}), + + Len2 = Len1 - 1, + %% this should be enough to stop + drain the prefetcher + {{_Msg1, false, AckTag1, Len2}, VQ7} = rabbit_variable_queue:fetch(VQ6), + S7 = rabbit_variable_queue:status(VQ7), + assert_prop(S7, prefetching, false), + assert_prop(S7, q4, (RamCount - 1)), + assert_prop(S7, q3, (Len1 - RamCount)), + + %% now fetch SegmentSize - 1 which will exhaust q4 and work through a bit of q3 + %% bringing in a segment from gamma: + {VQ8, AckTags} = variable_queue_fetch(SegmentSize-1, false, false, Len2, VQ7), + Len3 = Len2 - (SegmentSize - 1), + S8 = rabbit_variable_queue:status(VQ8), + assert_prop(S8, prefetching, false), + assert_prop(S8, q4, 0), + assert_prop(S8, q3, Len3), + assert_prop(S8, len, Len3), + + VQ9 = rabbit_variable_queue:remeasure_egress_rate(VQ8), + VQ10 = rabbit_variable_queue:ack(AckTags, VQ9), + + S10 = rabbit_variable_queue:status(VQ10), + assert_prop(S10, prefetching, true), + %% egress rate should be really high, so it's likely if we wait a + %% little bit, lots of msgs will be brought in + timer:sleep(2000), + PrefetchCount = lists:min([proplists:get_value(target_ram_msg_count, S10) - + proplists:get_value(ram_msg_count, S10), + Len3]), + Len4 = Len3 - 1, + {{_Msg2, false, AckTag2, Len4}, VQ11} = rabbit_variable_queue:fetch(VQ10), + S11 = rabbit_variable_queue:status(VQ11), + %% prefetcher will stop if it's fast enough and has completed by + %% now, or may still be running if PrefetchCount > 1 + Prefetched = proplists:get_value(q4, S11), + true = PrefetchCount > Prefetched, %% already fetched 1, thus >, not >= + %% q3 will contain whatever the prefetcher was not allowed to + %% prefetch, due to memory constraints. If the prefetcher is still + %% running, this will be less than (Len4 - Prefetched) because + %% Prefetched will not reflect the true number of msgs that it's + %% trying to prefetch. + case proplists:get_value(prefetching, S11) of + true -> true = (Len4 - Prefetched) > proplists:get_value(q3, S11); + false -> assert_prop(S11, q3, Len4 - Prefetched) + end, + assert_prop(S11, gamma, {gamma, undefined, 0}), + assert_prop(S11, q2, 0), + assert_prop(S11, q1, 0), + + VQ12 = rabbit_variable_queue:maybe_start_prefetcher(VQ11), + S12 = rabbit_variable_queue:status(VQ12), + assert_prop(S12, prefetching, (Len4 - Prefetched) > 0), + timer:sleep(2000), + %% we have to fetch all of q4 before the prefetcher will be drained + {VQ13, AckTags1} = + variable_queue_fetch(Prefetched, false, false, Len4, VQ12), + {VQ16, Acks} = + case Len4 == Prefetched of + true -> + {VQ13, [AckTag2, AckTag1, AckTag, AckTags1]}; + false -> + Len5 = Len4 - Prefetched - 1, + {{_Msg3, false, AckTag3, Len5}, VQ14} = + rabbit_variable_queue:fetch(VQ13), + assert_prop(rabbit_variable_queue:status(VQ14), + prefetching, false), + {VQ15, AckTags2} = + variable_queue_fetch(Len5, false, false, Len5, VQ14), + {VQ15, [AckTag3, AckTag2, AckTag1, AckTag, AckTags1, AckTags2]} + end, + VQ17 = rabbit_variable_queue:ack(lists:flatten(Acks), VQ16), + + {empty, VQ18} = rabbit_variable_queue:fetch(VQ17), + + rabbit_variable_queue:terminate(VQ18), + passed. diff --git a/src/rabbit_variable_queue.erl b/src/rabbit_variable_queue.erl new file mode 100644 index 0000000000..15caf81bb3 --- /dev/null +++ b/src/rabbit_variable_queue.erl @@ -0,0 +1,1019 @@ +%% The contents of this file are subject to the Mozilla Public License +%% Version 1.1 (the "License"); you may not use this file except in +%% compliance with the License. You may obtain a copy of the License at +%% http://www.mozilla.org/MPL/ +%% +%% 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 Original Code is RabbitMQ. +%% +%% The Initial Developers of the Original Code are LShift Ltd, +%% Cohesive Financial Technologies LLC, and Rabbit Technologies Ltd. +%% +%% Portions created before 22-Nov-2008 00:00:00 GMT by LShift Ltd, +%% Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd +%% are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial +%% Technologies LLC, and Rabbit Technologies Ltd. +%% +%% Portions created by LShift Ltd are Copyright (C) 2007-2009 LShift +%% Ltd. Portions created by Cohesive Financial Technologies LLC are +%% Copyright (C) 2007-2009 Cohesive Financial Technologies +%% LLC. Portions created by Rabbit Technologies Ltd are Copyright +%% (C) 2007-2009 Rabbit Technologies Ltd. +%% +%% All Rights Reserved. +%% +%% Contributor(s): ______________________________________. +%% + +-module(rabbit_variable_queue). + +-export([init/1, terminate/1, publish/2, publish_delivered/2, + set_queue_ram_duration_target/2, remeasure_egress_rate/1, fetch/1, + ack/2, len/1, is_empty/1, maybe_start_prefetcher/1, purge/1, delete/1, + requeue/2, tx_publish/2, tx_rollback/2, tx_commit/4, + tx_commit_from_msg_store/4, tx_commit_from_vq/1, needs_sync/1, + can_flush_journal/1, flush_journal/1, status/1]). + +%%---------------------------------------------------------------------------- + +-record(vqstate, + { q1, + q2, + gamma, + q3, + q4, + duration_target, + target_ram_msg_count, + ram_msg_count, + queue, + index_state, + next_seq_id, + out_counter, + egress_rate, + avg_egress_rate, + egress_rate_timestamp, + prefetcher, + len, + on_sync + }). + +-include("rabbit.hrl"). +-include("rabbit_queue.hrl"). + +%%---------------------------------------------------------------------------- + +%% Basic premise is that msgs move from q1 -> q2 -> gamma -> q3 -> q4 +%% but they can only do so in the right form. q1 and q4 only hold +%% alphas (msgs in ram), q2 and q3 only hold betas (msg on disk, index +%% in ram), and gamma is just a count of the number of index entries +%% on disk at that stage (msg on disk, index on disk). +%% +%% When a msg arrives, we decide in which form it should be. It is +%% then added to the right-most appropriate queue, maintaining +%% order. Thus if the msg is to be an alpha, it will be added to q1, +%% unless all of q2, gamma and q3 are empty, in which case it will go +%% to q4. If it is to be a beta, it will be added to q2 unless gamma +%% is empty, in which case it will go to q3. +%% +%% The major invariant is that if the msg is to be a beta, q1 will be +%% empty, and if it is to be a gamma then both q1 and q2 will be empty. +%% +%% When taking msgs out of the queue, if q4 is empty then we drain the +%% prefetcher. If that doesn't help then we read directly from q3, or +%% gamma, if q3 is empty. If q3 and gamma are empty then we have an +%% invariant that q2 must be empty because q2 can only grow if gamma +%% is non empty. +%% +%% A further invariant is that if the queue is non empty, either q4 or +%% q3 contains at least one entry. I.e. we never allow gamma to +%% contain all msgs in the queue. Also, if q4 is non empty and gamma +%% is non empty then q3 must be non empty. + +%%---------------------------------------------------------------------------- + +-ifdef(use_specs). + +-type(msg_id() :: binary()). +-type(seq_id() :: non_neg_integer()). +-type(ack() :: {'ack_index_and_store', msg_id(), seq_id()} + | 'ack_not_on_disk'). +-type(vqstate() :: #vqstate { + q1 :: queue(), + q2 :: queue(), + gamma :: gamma(), + q3 :: queue(), + q4 :: queue(), + duration_target :: non_neg_integer(), + target_ram_msg_count :: non_neg_integer(), + queue :: queue_name(), + index_state :: any(), + next_seq_id :: seq_id(), + out_counter :: non_neg_integer(), + egress_rate :: float(), + avg_egress_rate :: float(), + egress_rate_timestamp :: {integer(), integer(), integer()}, + prefetcher :: ('undefined' | pid()), + len :: non_neg_integer(), + on_sync :: {[ack()], [msg_id()], [{pid(), any()}]} + }). + +-spec(init/1 :: (queue_name()) -> vqstate()). +-spec(terminate/1 :: (vqstate()) -> vqstate()). +-spec(publish/2 :: (basic_message(), vqstate()) -> + {seq_id(), vqstate()}). +-spec(publish_delivered/2 :: (basic_message(), vqstate()) -> + {ack(), vqstate()}). +-spec(set_queue_ram_duration_target/2 :: + (('undefined' | number()), vqstate()) -> vqstate()). +-spec(remeasure_egress_rate/1 :: (vqstate()) -> vqstate()). +-spec(fetch/1 :: (vqstate()) -> + {('empty'|{basic_message(), boolean(), ack(), non_neg_integer()}), + vqstate()}). +-spec(ack/2 :: ([ack()], vqstate()) -> vqstate()). +-spec(len/1 :: (vqstate()) -> non_neg_integer()). +-spec(is_empty/1 :: (vqstate()) -> boolean()). +-spec(maybe_start_prefetcher/1 :: (vqstate()) -> vqstate()). +-spec(purge/1 :: (vqstate()) -> {non_neg_integer(), vqstate()}). +-spec(delete/1 :: (vqstate()) -> vqstate()). +-spec(requeue/2 :: ([{basic_message(), ack()}], vqstate()) -> vqstate()). +-spec(tx_publish/2 :: (basic_message(), vqstate()) -> vqstate()). +-spec(tx_rollback/2 :: ([msg_id()], vqstate()) -> vqstate()). +-spec(tx_commit/4 :: ([msg_id()], [ack()], {pid(), any()}, vqstate()) -> + {boolean(), vqstate()}). +-spec(tx_commit_from_msg_store/4 :: + ([msg_id()], [ack()], {pid(), any()}, vqstate()) -> vqstate()). +-spec(tx_commit_from_vq/1 :: (vqstate()) -> vqstate()). +-spec(needs_sync/1 :: (vqstate()) -> boolean()). +-spec(can_flush_journal/1 :: (vqstate()) -> boolean()). +-spec(flush_journal/1 :: (vqstate()) -> vqstate()). +-spec(status/1 :: (vqstate()) -> [{atom(), any()}]). + +-endif. + +%%---------------------------------------------------------------------------- +%% Public API +%%---------------------------------------------------------------------------- + +init(QueueName) -> + {GammaCount, IndexState} = + rabbit_queue_index:init(QueueName), + {GammaSeqId, NextSeqId, IndexState1} = + rabbit_queue_index:find_lowest_seq_id_seg_and_next_seq_id(IndexState), + Gamma = case GammaCount of + 0 -> #gamma { seq_id = undefined, count = 0 }; + _ -> #gamma { seq_id = GammaSeqId, count = GammaCount } + end, + State = + #vqstate { q1 = queue:new(), q2 = queue:new(), + gamma = Gamma, + q3 = queue:new(), q4 = queue:new(), + target_ram_msg_count = undefined, + duration_target = undefined, + ram_msg_count = 0, + queue = QueueName, + index_state = IndexState1, + next_seq_id = NextSeqId, + out_counter = 0, + egress_rate = 0, + avg_egress_rate = 0, + egress_rate_timestamp = now(), + prefetcher = undefined, + len = GammaCount, + on_sync = {[], [], []} + }, + maybe_gammas_to_betas(State). + +terminate(State = #vqstate { index_state = IndexState }) -> + State #vqstate { index_state = rabbit_queue_index:terminate(IndexState) }. + +publish(Msg, State) -> + publish(Msg, false, false, State). + +publish_delivered(Msg = #basic_message { guid = MsgId, + is_persistent = IsPersistent }, + State = #vqstate { len = 0, index_state = IndexState, + next_seq_id = SeqId }) -> + case maybe_write_msg_to_disk(false, false, Msg) of + true -> + {true, IndexState1} = + maybe_write_index_to_disk(false, IsPersistent, MsgId, SeqId, + true, IndexState), + {{ack_index_and_store, MsgId, SeqId}, + State #vqstate { index_state = IndexState1, + next_seq_id = SeqId + 1 }}; + false -> + {ack_not_on_disk, State} + end. + +set_queue_ram_duration_target(undefined, State) -> + State; +set_queue_ram_duration_target( + DurationTarget, State = #vqstate { avg_egress_rate = EgressRate, + target_ram_msg_count = TargetRamMsgCount + }) -> + TargetRamMsgCount1 = trunc(DurationTarget * EgressRate), %% msgs = sec * msgs/sec + State1 = State #vqstate { target_ram_msg_count = TargetRamMsgCount1, + duration_target = DurationTarget }, + if TargetRamMsgCount == TargetRamMsgCount1 -> + State1; + TargetRamMsgCount == undefined orelse + TargetRamMsgCount < TargetRamMsgCount1 -> + maybe_start_prefetcher(State1); + true -> + reduce_memory_use(State1) + end. + +remeasure_egress_rate(State = #vqstate { egress_rate = OldEgressRate, + egress_rate_timestamp = Timestamp, + out_counter = OutCount, + duration_target = DurationTarget }) -> + %% We do an average over the last two values, but also hold the + %% current value separately so that the average always only + %% incorporates the last two values, and not the current value and + %% the last average. Averaging helps smooth out spikes. + Now = now(), + %% EgressRate is in seconds, and now_diff is in microseconds + EgressRate = 1000000 * OutCount / timer:now_diff(Now, Timestamp), + AvgEgressRate = (EgressRate + OldEgressRate) / 2, + set_queue_ram_duration_target( + DurationTarget, + State #vqstate { egress_rate = EgressRate, + avg_egress_rate = AvgEgressRate, + egress_rate_timestamp = Now, + out_counter = 0 }). + +fetch(State = + #vqstate { q4 = Q4, ram_msg_count = RamMsgCount, + out_counter = OutCount, prefetcher = Prefetcher, + index_state = IndexState, len = Len }) -> + case queue:out(Q4) of + {empty, _Q4} when Prefetcher == undefined -> + fetch_from_q3_or_gamma(State); + {empty, _Q4} -> + fetch(drain_prefetcher(drain, State)); + {{value, + #alpha { msg = Msg = #basic_message { guid = MsgId, + is_persistent = IsPersistent }, + seq_id = SeqId, is_delivered = IsDelivered, + msg_on_disk = MsgOnDisk, index_on_disk = IndexOnDisk }}, + Q4a} -> + {IndexState1, IndexOnDisk1} = + case IndexOnDisk of + true -> + IndexState2 = + case IsDelivered of + false -> rabbit_queue_index:write_delivered( + SeqId, IndexState); + true -> IndexState + end, + case IsPersistent of + true -> {IndexState2, true}; + false -> {rabbit_queue_index:write_acks( + [SeqId], IndexState2), false} + end; + false -> + {IndexState, false} + end, + _MsgOnDisk1 = IndexOnDisk1 = + case IndexOnDisk1 of + true -> true = IsPersistent, %% ASSERTION + true = MsgOnDisk; %% ASSERTION + false -> ok = case MsgOnDisk andalso not IsPersistent of + true -> rabbit_msg_store:remove([MsgId]); + false -> ok + end, + false + end, + AckTag = case IndexOnDisk1 of + true -> {ack_index_and_store, MsgId, SeqId}; + false -> ack_not_on_disk + end, + Len1 = Len - 1, + {{Msg, IsDelivered, AckTag, Len1}, + State #vqstate { q4 = Q4a, out_counter = OutCount + 1, + ram_msg_count = RamMsgCount - 1, + index_state = IndexState1, len = Len1 }} + end. + +ack(AckTags, State = #vqstate { index_state = IndexState }) -> + {MsgIds, SeqIds} = + lists:foldl( + fun (ack_not_on_disk, Acc) -> Acc; + ({ack_index_and_store, MsgId, SeqId}, {MsgIds, SeqIds}) -> + {[MsgId | MsgIds], [SeqId | SeqIds]} + end, {[], []}, AckTags), + IndexState1 = case SeqIds of + [] -> IndexState; + _ -> rabbit_queue_index:write_acks(SeqIds, IndexState) + end, + ok = case MsgIds of + [] -> ok; + _ -> rabbit_msg_store:remove(MsgIds) + end, + State #vqstate { index_state = IndexState1 }. + +len(#vqstate { len = Len }) -> + Len. + +is_empty(State) -> + 0 == len(State). + +maybe_start_prefetcher(State = #vqstate { target_ram_msg_count = 0 }) -> + State; +maybe_start_prefetcher(State = #vqstate { prefetcher = undefined }) -> + %% ensure we have as much index in RAM as we can + State1 = #vqstate { ram_msg_count = RamMsgCount, + target_ram_msg_count = TargetRamMsgCount, + q1 = Q1, q3 = Q3 } = maybe_gammas_to_betas(State), + case queue:is_empty(Q3) of + true -> %% nothing to do + State1; + false -> + %% prefetched content takes priority over q1 + AvailableSpace = + case TargetRamMsgCount of + undefined -> queue:len(Q3); + _ -> (TargetRamMsgCount - RamMsgCount) + queue:len(Q1) + end, + PrefetchCount = lists:min([queue:len(Q3), AvailableSpace]), + case PrefetchCount =< 0 of + true -> State1; + false -> + {PrefetchQueue, Q3a} = queue:split(PrefetchCount, Q3), + {ok, Prefetcher} = + rabbit_queue_prefetcher:start_link(PrefetchQueue), + State1 #vqstate { q3 = Q3a, prefetcher = Prefetcher } + end + end; +maybe_start_prefetcher(State) -> + State. + +purge(State = #vqstate { prefetcher = undefined, q4 = Q4, + index_state = IndexState, len = Len }) -> + {Q4Count, IndexState1} = remove_queue_entries(Q4, IndexState), + {Len, State1} = + purge1(Q4Count, State #vqstate { index_state = IndexState1, + q4 = queue:new() }), + {Len, State1 #vqstate { len = 0 }}; +purge(State) -> + purge(drain_prefetcher(stop, State)). + +%% the only difference between purge and delete is that delete also +%% needs to delete everything that's been delivered and not ack'd. +delete(State) -> + {_PurgeCount, State1 = #vqstate { index_state = IndexState }} = purge(State), + IndexState1 = + case rabbit_queue_index:find_lowest_seq_id_seg_and_next_seq_id( + IndexState) of + {N, N, IndexState2} -> + IndexState2; + {GammaSeqId, NextSeqId, IndexState2} -> + {_DeleteCount, IndexState3} = + delete1(NextSeqId, 0, GammaSeqId, IndexState2), + IndexState3 + end, + IndexState4 = rabbit_queue_index:terminate_and_erase(IndexState1), + State1 #vqstate { index_state = IndexState4 }. + +%% [{Msg, AckTag}] +%% We guarantee that after fetch, only persistent msgs are left on +%% disk. This means that in a requeue, we set +%% PersistentMsgsAlreadyOnDisk to true, thus avoiding calls to +%% msg_store:write for persistent msgs. It also means that we don't +%% need to worry about calling msg_store:remove (as ack would do) +%% because transient msgs won't be on disk anyway, thus they won't +%% need to be removed. However, we do call msg_store:release so that +%% the cache isn't held full of msgs which are now at the tail of the +%% queue. +requeue(MsgsWithAckTags, State) -> + {SeqIds, MsgIds, State1 = #vqstate { index_state = IndexState }} = + lists:foldl( + fun ({Msg = #basic_message { guid = MsgId }, AckTag}, + {SeqIdsAcc, MsgIdsAcc, StateN}) -> + {_SeqId, StateN1} = publish(Msg, true, true, StateN), + {SeqIdsAcc1, MsgIdsAcc1} = + case AckTag of + ack_not_on_disk -> + {SeqIdsAcc, MsgIdsAcc}; + {ack_index_and_store, MsgId, SeqId} -> + {[SeqId | SeqIdsAcc], [MsgId | MsgIdsAcc]} + end, + {SeqIdsAcc1, MsgIdsAcc1, StateN1} + end, {[], [], State}, MsgsWithAckTags), + IndexState1 = case SeqIds of + [] -> IndexState; + _ -> rabbit_queue_index:write_acks(SeqIds, IndexState) + end, + ok = case MsgIds of + [] -> ok; + _ -> rabbit_msg_store:release(MsgIds) + end, + State1 #vqstate { index_state = IndexState1 }. + +tx_publish(Msg = #basic_message { is_persistent = true }, State) -> + true = maybe_write_msg_to_disk(true, false, Msg), + State; +tx_publish(_Msg, State) -> + State. + +tx_rollback(Pubs, State) -> + ok = case persistent_msg_ids(Pubs) of + [] -> ok; + PP -> rabbit_msg_store:remove(PP) + end, + State. + +tx_commit(Pubs, AckTags, From, State) -> + case persistent_msg_ids(Pubs) of + [] -> + {true, tx_commit_from_msg_store(Pubs, AckTags, From, State)}; + PersistentMsgIds -> + Self = self(), + ok = rabbit_msg_store:sync( + PersistentMsgIds, + fun () -> ok = rabbit_amqqueue:tx_commit_msg_store_callback( + Self, Pubs, AckTags, From) + end), + {false, State} + end. + +tx_commit_from_msg_store(Pubs, AckTags, From, + State = #vqstate { on_sync = {SAcks, SPubs, SFroms} }) -> + DiskAcks = + lists:filter(fun (AckTag) -> AckTag /= ack_not_on_disk end, AckTags), + State #vqstate { on_sync = { [DiskAcks | SAcks], + [Pubs | SPubs], + [From | SFroms] }}. + +tx_commit_from_vq(State = #vqstate { on_sync = {SAcks, SPubs, SFroms} }) -> + State1 = ack(lists:flatten(SAcks), State), + {PubSeqIds, State2 = #vqstate { index_state = IndexState }} = + lists:foldl( + fun (Msg = #basic_message { is_persistent = IsPersistent }, + {SeqIdsAcc, StateN}) -> + {SeqId, StateN1} = publish(Msg, false, true, StateN), + SeqIdsAcc1 = case IsPersistent of + true -> [SeqId | SeqIdsAcc]; + false -> SeqIdsAcc + end, + {SeqIdsAcc1, StateN1} + end, {[], State1}, lists:flatten(lists:reverse(SPubs))), + IndexState1 = + rabbit_queue_index:sync_seq_ids(PubSeqIds, [] /= SAcks, IndexState), + [ gen_server2:reply(From, ok) || From <- lists:reverse(SFroms) ], + State2 #vqstate { index_state = IndexState1, on_sync = {[], [], []} }. + +needs_sync(#vqstate { on_sync = {_, _, []} }) -> + false; +needs_sync(_) -> + true. + +can_flush_journal(#vqstate { index_state = IndexState }) -> + rabbit_queue_index:can_flush_journal(IndexState). + +flush_journal(State = #vqstate { index_state = IndexState }) -> + State #vqstate { index_state = + rabbit_queue_index:flush_journal(IndexState) }. + +status(#vqstate { q1 = Q1, q2 = Q2, gamma = Gamma, q3 = Q3, q4 = Q4, + len = Len, on_sync = {_, _, From}, + target_ram_msg_count = TargetRamMsgCount, + ram_msg_count = RamMsgCount, prefetcher = Prefetcher, + avg_egress_rate = AvgEgressRate }) -> + [ {q1, queue:len(Q1)}, + {q2, queue:len(Q2)}, + {gamma, Gamma}, + {q3, queue:len(Q3)}, + {q4, queue:len(Q4)}, + {len, Len}, + {outstanding_txns, length(From)}, + {target_ram_msg_count, TargetRamMsgCount}, + {ram_msg_count, RamMsgCount}, + {avg_egress_rate, AvgEgressRate}, + {prefetching, Prefetcher /= undefined} ]. + +%%---------------------------------------------------------------------------- +%% Minor helpers +%%---------------------------------------------------------------------------- + +persistent_msg_ids(Pubs) -> + [MsgId || Obj = #basic_message { guid = MsgId } <- Pubs, + Obj #basic_message.is_persistent]. + +entry_salient_details(#alpha { msg = #basic_message { guid = MsgId }, + seq_id = SeqId, is_delivered = IsDelivered, + msg_on_disk = MsgOnDisk, + index_on_disk = IndexOnDisk }) -> + {MsgId, SeqId, IsDelivered, MsgOnDisk, IndexOnDisk}; +entry_salient_details(#beta { msg_id = MsgId, seq_id = SeqId, + is_delivered = IsDelivered, + index_on_disk = IndexOnDisk }) -> + {MsgId, SeqId, IsDelivered, true, IndexOnDisk}. + +betas_from_segment_entries(List) -> + queue:from_list([#beta { msg_id = MsgId, seq_id = SeqId, + is_persistent = IsPersistent, + is_delivered = IsDelivered, + index_on_disk = true } + || {MsgId, SeqId, IsPersistent, IsDelivered} <- List]). + +read_index_segment(SeqId, IndexState) -> + SeqId1 = SeqId + rabbit_queue_index:segment_size(), + case rabbit_queue_index:read_segment_entries(SeqId, IndexState) of + {[], IndexState1} -> read_index_segment(SeqId1, IndexState1); + {List, IndexState1} -> {List, IndexState1, SeqId1} + end. + +ensure_binary_properties(Msg = #basic_message { content = Content }) -> + Msg #basic_message { + content = rabbit_binary_parser:clear_decoded_content( + rabbit_binary_generator:ensure_content_encoded(Content)) }. + +%% the first arg is the older gamma +combine_gammas(#gamma { count = 0 }, #gamma { count = 0 }) -> + #gamma { seq_id = undefined, count = 0 }; +combine_gammas(#gamma { count = 0 }, #gamma { } = B) -> B; +combine_gammas(#gamma { } = A, #gamma { count = 0 }) -> A; +combine_gammas(#gamma { seq_id = SeqIdLow, count = CountLow }, + #gamma { seq_id = SeqIdHigh, count = CountHigh}) -> + true = SeqIdLow =< SeqIdHigh, %% ASSERTION + #gamma { seq_id = SeqIdLow, count = CountLow + CountHigh}. + +%%---------------------------------------------------------------------------- +%% Internal major helpers for Public API +%%---------------------------------------------------------------------------- + +delete1(NextSeqId, Count, GammaSeqId, IndexState) + when GammaSeqId >= NextSeqId -> + {Count, IndexState}; +delete1(NextSeqId, Count, GammaSeqId, IndexState) -> + Gamma1SeqId = GammaSeqId + rabbit_queue_index:segment_size(), + case rabbit_queue_index:read_segment_entries(GammaSeqId, IndexState) of + {[], IndexState1} -> + delete1(NextSeqId, Count, Gamma1SeqId, IndexState1); + {List, IndexState1} -> + Q = betas_from_segment_entries(List), + {QCount, IndexState2} = remove_queue_entries(Q, IndexState1), + delete1(NextSeqId, Count + QCount, Gamma1SeqId, IndexState2) + end. + +purge1(Count, State = #vqstate { q3 = Q3, index_state = IndexState }) -> + case queue:is_empty(Q3) of + true -> + {Q1Count, IndexState1} = + remove_queue_entries(State #vqstate.q1, IndexState), + {Count + Q1Count, State #vqstate { q1 = queue:new(), + index_state = IndexState1 }}; + false -> + {Q3Count, IndexState1} = remove_queue_entries(Q3, IndexState), + purge1(Count + Q3Count, + maybe_gammas_to_betas( + State #vqstate { index_state = IndexState1, + q3 = queue:new() })) + end. + +remove_queue_entries(Q, IndexState) -> + {Count, MsgIds, SeqIds, IndexState1} = + lists:foldl( + fun (Entry, {CountN, MsgIdsAcc, SeqIdsAcc, IndexStateN}) -> + {MsgId, SeqId, IsDelivered, MsgOnDisk, IndexOnDisk} = + entry_salient_details(Entry), + IndexStateN1 = case IndexOnDisk andalso not IsDelivered of + true -> rabbit_queue_index:write_delivered( + SeqId, IndexStateN); + false -> IndexStateN + end, + SeqIdsAcc1 = case IndexOnDisk of + true -> [SeqId | SeqIdsAcc]; + false -> SeqIdsAcc + end, + MsgIdsAcc1 = case MsgOnDisk of + true -> [MsgId | MsgIdsAcc]; + false -> MsgIdsAcc + end, + {CountN + 1, MsgIdsAcc1, SeqIdsAcc1, IndexStateN1} + %% we need to write the delivered records in order otherwise + %% we upset the qi. So don't reverse. + end, {0, [], [], IndexState}, queue:to_list(Q)), + ok = case MsgIds of + [] -> ok; + _ -> rabbit_msg_store:remove(MsgIds) + end, + IndexState2 = + case SeqIds of + [] -> IndexState1; + _ -> rabbit_queue_index:write_acks(SeqIds, IndexState1) + end, + {Count, IndexState2}. + +fetch_from_q3_or_gamma(State = #vqstate { + q1 = Q1, q2 = Q2, gamma = #gamma { count = GammaCount }, + q3 = Q3, q4 = Q4, ram_msg_count = RamMsgCount }) -> + case queue:out(Q3) of + {empty, _Q3} -> + 0 = GammaCount, %% ASSERTION + true = queue:is_empty(Q2), %% ASSERTION + true = queue:is_empty(Q1), %% ASSERTION + {empty, State}; + {{value, + #beta { msg_id = MsgId, seq_id = SeqId, is_delivered = IsDelivered, + is_persistent = IsPersistent, index_on_disk = IndexOnDisk }}, + Q3a} -> + {ok, Msg = #basic_message { is_persistent = IsPersistent, + guid = MsgId }} = + rabbit_msg_store:read(MsgId), + Q4a = queue:in( + #alpha { msg = Msg, seq_id = SeqId, + is_delivered = IsDelivered, msg_on_disk = true, + index_on_disk = IndexOnDisk }, Q4), + State1 = State #vqstate { q3 = Q3a, q4 = Q4a, + ram_msg_count = RamMsgCount + 1 }, + State2 = + case {queue:is_empty(Q3a), 0 == GammaCount} of + {true, true} -> + %% q3 is now empty, it wasn't before; gamma is + %% still empty. So q2 must be empty, and q1 + %% can now be joined onto q4 + true = queue:is_empty(Q2), %% ASSERTION + State1 #vqstate { q1 = queue:new(), + q4 = queue:join(Q4a, Q1) }; + {true, false} -> + maybe_gammas_to_betas(State1); + {false, _} -> + %% q3 still isn't empty, we've not touched + %% gamma, so the invariants between q1, q2, + %% gamma and q3 are maintained + State1 + end, + fetch(State2) + end. + +drain_prefetcher(_DrainOrStop, State = #vqstate { prefetcher = undefined }) -> + State; +drain_prefetcher(DrainOrStop, + State = #vqstate { prefetcher = Prefetcher, q1 = Q1, q2 = Q2, + gamma = #gamma { count = GammaCount }, + q3 = Q3, q4 = Q4, + ram_msg_count = RamMsgCount }) -> + Fun = case DrainOrStop of + drain -> fun rabbit_queue_prefetcher:drain/1; + stop -> fun rabbit_queue_prefetcher:drain_and_stop/1 + end, + {Q3a, Q4a, Prefetcher1, RamMsgCountAdj} = + case Fun(Prefetcher) of + {empty, Betas} -> %% drain or drain_and_stop + {queue:join(Betas, Q3), Q4, undefined, 0}; + {finished, Alphas} -> %% just drain + {Q3, queue:join(Q4, Alphas), undefined, queue:len(Alphas)}; + {continuing, Alphas} -> %% just drain + {Q3, queue:join(Q4, Alphas), Prefetcher, queue:len(Alphas)}; + {Alphas, Betas} -> %% just drain_and_stop + {queue:join(Betas, Q3), queue:join(Q4, Alphas), undefined, + queue:len(Alphas)} + end, + State1 = State #vqstate { prefetcher = Prefetcher1, q3 = Q3a, q4 = Q4a, + ram_msg_count = RamMsgCount + RamMsgCountAdj }, + %% don't join up with q1/q2 unless the prefetcher has stopped + State2 = case GammaCount == 0 andalso Prefetcher1 == undefined of + true -> case queue:is_empty(Q3a) andalso queue:is_empty(Q2) of + true -> + State1 #vqstate { q1 = queue:new(), + q4 = queue:join(Q4a, Q1) }; + false -> + State1 #vqstate { q3 = queue:join(Q3a, Q2) } + end; + false -> State1 + end, + maybe_push_q1_to_betas(State2). + +reduce_memory_use(State = #vqstate { ram_msg_count = RamMsgCount, + target_ram_msg_count = TargetRamMsgCount }) + when TargetRamMsgCount == undefined orelse TargetRamMsgCount >= RamMsgCount -> + State; +reduce_memory_use(State = + #vqstate { target_ram_msg_count = TargetRamMsgCount }) -> + %% strictly, it's not necessary to stop the prefetcher this early, + %% but because of its potential effect on q1 and the + %% ram_msg_count, it's just much simpler to stop it sooner and + %% relaunch when we next hibernate. + State1 = maybe_push_q4_to_betas(maybe_push_q1_to_betas( + drain_prefetcher(stop, State))), + case TargetRamMsgCount of + 0 -> push_betas_to_gammas(State1); + _ -> State1 + end. + +%%---------------------------------------------------------------------------- +%% Internal gubbins for publishing +%%---------------------------------------------------------------------------- + +test_keep_msg_in_ram(SeqId, #vqstate { target_ram_msg_count = TargetRamMsgCount, + ram_msg_count = RamMsgCount, + q1 = Q1, q3 = Q3 }) -> + case TargetRamMsgCount of + undefined -> + msg; + 0 -> + case queue:out(Q3) of + {empty, _Q3} -> + %% if TargetRamMsgCount == 0, we know we have no + %% alphas. If q3 is empty then gamma must be empty + %% too, so create a beta, which should end up in + %% q3 + index; + {{value, #beta { seq_id = OldSeqId }}, _Q3a} -> + %% Don't look at the current gamma as it may be + %% empty. If the SeqId is still within the current + %% segment, it'll be a beta, else it'll go into + %% gamma + case SeqId >= rabbit_queue_index:next_segment_boundary(OldSeqId) of + true -> neither; + false -> index + end + end; + _ when TargetRamMsgCount > RamMsgCount -> + msg; + _ -> + case queue:is_empty(Q1) of + true -> index; + false -> msg %% can push out elders to disk + end + end. + +publish(Msg, IsDelivered, PersistentMsgsAlreadyOnDisk, + State = #vqstate { next_seq_id = SeqId, len = Len }) -> + {SeqId, publish(test_keep_msg_in_ram(SeqId, State), Msg, SeqId, IsDelivered, + PersistentMsgsAlreadyOnDisk, + State #vqstate { next_seq_id = SeqId + 1, len = Len + 1 })}. + +publish(msg, Msg = #basic_message { guid = MsgId, + is_persistent = IsPersistent }, + SeqId, IsDelivered, PersistentMsgsAlreadyOnDisk, + State = #vqstate { index_state = IndexState, + ram_msg_count = RamMsgCount }) -> + MsgOnDisk = + maybe_write_msg_to_disk(false, PersistentMsgsAlreadyOnDisk, Msg), + {IndexOnDisk, IndexState1} = + maybe_write_index_to_disk(false, IsPersistent, MsgId, SeqId, + IsDelivered, IndexState), + Entry = #alpha { msg = Msg, seq_id = SeqId, is_delivered = IsDelivered, + msg_on_disk = MsgOnDisk, index_on_disk = IndexOnDisk }, + State1 = State #vqstate { ram_msg_count = RamMsgCount + 1, + index_state = IndexState1 }, + store_alpha_entry(Entry, State1); + +publish(index, Msg = #basic_message { guid = MsgId, + is_persistent = IsPersistent }, + SeqId, IsDelivered, PersistentMsgsAlreadyOnDisk, + State = #vqstate { index_state = IndexState, q1 = Q1 }) -> + true = maybe_write_msg_to_disk(true, PersistentMsgsAlreadyOnDisk, Msg), + {IndexOnDisk, IndexState1} = + maybe_write_index_to_disk(false, IsPersistent, MsgId, SeqId, + IsDelivered, IndexState), + Entry = #beta { msg_id = MsgId, seq_id = SeqId, is_delivered = IsDelivered, + is_persistent = IsPersistent, index_on_disk = IndexOnDisk }, + State1 = State #vqstate { index_state = IndexState1 }, + true = queue:is_empty(Q1), %% ASSERTION + store_beta_entry(Entry, State1); + +publish(neither, Msg = #basic_message { guid = MsgId, + is_persistent = IsPersistent }, + SeqId, IsDelivered, PersistentMsgsAlreadyOnDisk, + State = #vqstate { index_state = IndexState, q1 = Q1, q2 = Q2, + gamma = Gamma }) -> + true = maybe_write_msg_to_disk(true, PersistentMsgsAlreadyOnDisk, Msg), + {true, IndexState1} = + maybe_write_index_to_disk(true, IsPersistent, MsgId, SeqId, + IsDelivered, IndexState), + true = queue:is_empty(Q1) andalso queue:is_empty(Q2), %% ASSERTION + %% gamma may be empty, seq_id > next_segment_boundary from q3 + %% head, so we need to find where the segment boundary is before + %% or equal to seq_id + GammaSeqId = rabbit_queue_index:next_segment_boundary(SeqId) - + rabbit_queue_index:segment_size(), + Gamma1 = #gamma { seq_id = GammaSeqId, count = 1 }, + State #vqstate { index_state = IndexState1, + gamma = combine_gammas(Gamma, Gamma1) }. + +store_alpha_entry(Entry = #alpha {}, State = + #vqstate { q1 = Q1, q2 = Q2, + gamma = #gamma { count = GammaCount }, + q3 = Q3, q4 = Q4, prefetcher = Prefetcher }) -> + case queue:is_empty(Q2) andalso GammaCount == 0 andalso + queue:is_empty(Q3) andalso Prefetcher == undefined of + true -> + State #vqstate { q4 = queue:in(Entry, Q4) }; + false -> + maybe_push_q1_to_betas(State #vqstate { q1 = queue:in(Entry, Q1) }) + end. + +store_beta_entry(Entry = #beta {}, State = + #vqstate { q2 = Q2, gamma = #gamma { count = GammaCount }, + q3 = Q3 }) -> + case GammaCount == 0 of + true -> State #vqstate { q3 = queue:in(Entry, Q3) }; + false -> State #vqstate { q2 = queue:in(Entry, Q2) } + end. + +%% Bool IsPersistent PersistentMsgsAlreadyOnDisk | WriteToDisk? +%% -----------------------------------------------+------------- +%% false false false | false 1 +%% false true false | true 2 +%% false false true | false 3 +%% false true true | false 4 +%% true false false | true 5 +%% true true false | true 6 +%% true false true | true 7 +%% true true true | false 8 + +%% (Bool and not (IsPersistent and PersistentMsgsAlreadyOnDisk)) or | 5 6 7 +%% (IsPersistent and (not PersistentMsgsAlreadyOnDisk)) | 2 6 +maybe_write_msg_to_disk(Bool, PersistentMsgsAlreadyOnDisk, + Msg = #basic_message { guid = MsgId, + is_persistent = IsPersistent }) + when (Bool andalso not (IsPersistent andalso PersistentMsgsAlreadyOnDisk)) + orelse (IsPersistent andalso not PersistentMsgsAlreadyOnDisk) -> + ok = rabbit_msg_store:write(MsgId, ensure_binary_properties(Msg)), + true; +maybe_write_msg_to_disk(_Bool, true, #basic_message { is_persistent = true }) -> + true; +maybe_write_msg_to_disk(_Bool, _PersistentMsgsAlreadyOnDisk, _Msg) -> + false. + +maybe_write_index_to_disk(Bool, IsPersistent, MsgId, SeqId, IsDelivered, + IndexState) when Bool orelse IsPersistent -> + IndexState1 = rabbit_queue_index:write_published( + MsgId, SeqId, IsPersistent, IndexState), + {true, case IsDelivered of + true -> rabbit_queue_index:write_delivered(SeqId, IndexState1); + false -> IndexState1 + end}; +maybe_write_index_to_disk(_Bool, _IsPersistent, _MsgId, _SeqId, _IsDelivered, + IndexState) -> + {false, IndexState}. + +%%---------------------------------------------------------------------------- +%% Phase changes +%%---------------------------------------------------------------------------- + +maybe_gammas_to_betas(State = #vqstate { gamma = #gamma { count = 0 } }) -> + State; +maybe_gammas_to_betas(State = + #vqstate { index_state = IndexState, q2 = Q2, q3 = Q3, + target_ram_msg_count = TargetRamMsgCount, + gamma = #gamma { seq_id = GammaSeqId, + count = GammaCount }}) -> + case (not queue:is_empty(Q3)) andalso 0 == TargetRamMsgCount of + true -> + State; + false -> + %% either q3 is empty, in which case we load at least one + %% segment, or TargetRamMsgCount > 0, meaning we should + %% really be holding all the betas in memory. + {List, IndexState1, Gamma1SeqId} = + read_index_segment(GammaSeqId, IndexState), + State1 = State #vqstate { index_state = IndexState1 }, + %% length(List) may be < segment_size because of acks. But + %% it can't be [] + Q3a = queue:join(Q3, betas_from_segment_entries(List)), + case GammaCount - length(List) of + 0 -> + %% gamma is now empty, but it wasn't before, so + %% can now join q2 onto q3 + State1 #vqstate { gamma = #gamma { seq_id = undefined, + count = 0 }, + q2 = queue:new(), + q3 = queue:join(Q3a, Q2) }; + N when N > 0 -> + maybe_gammas_to_betas( + State1 #vqstate { q3 = Q3a, + gamma = #gamma { seq_id = Gamma1SeqId, + count = N } }) + end + end. + +maybe_push_q1_to_betas(State = #vqstate { q1 = Q1 }) -> + maybe_push_alphas_to_betas( + fun queue:out/1, + fun (Beta, Q1a, State1) -> + %% these could legally go to q3 if gamma and q2 are empty + store_beta_entry(Beta, State1 #vqstate { q1 = Q1a }) + end, Q1, State). + +maybe_push_q4_to_betas(State = #vqstate { q4 = Q4 }) -> + maybe_push_alphas_to_betas( + fun queue:out_r/1, + fun (Beta, Q4a, State1 = #vqstate { q3 = Q3 }) -> + %% these must go to q3 + State1 #vqstate { q3 = queue:in_r(Beta, Q3), q4 = Q4a } + end, Q4, State). + +maybe_push_alphas_to_betas(_Generator, _Consumer, _Q, State = + #vqstate { ram_msg_count = RamMsgCount, + target_ram_msg_count = TargetRamMsgCount }) + when TargetRamMsgCount == undefined orelse TargetRamMsgCount >= RamMsgCount -> + State; +maybe_push_alphas_to_betas(Generator, Consumer, Q, State = + #vqstate { ram_msg_count = RamMsgCount }) -> + case Generator(Q) of + {empty, _Q} -> State; + {{value, + #alpha { msg = Msg = #basic_message { guid = MsgId, + is_persistent = IsPersistent }, + seq_id = SeqId, is_delivered = IsDelivered, + index_on_disk = IndexOnDisk }}, + Qa} -> + true = maybe_write_msg_to_disk(true, true, Msg), + Beta = #beta { msg_id = MsgId, seq_id = SeqId, + is_persistent = IsPersistent, + is_delivered = IsDelivered, + index_on_disk = IndexOnDisk }, + State1 = State #vqstate { ram_msg_count = RamMsgCount - 1 }, + maybe_push_alphas_to_betas(Generator, Consumer, Qa, + Consumer(Beta, Qa, State1)) + end. + +push_betas_to_gammas(State = #vqstate { q2 = Q2, gamma = Gamma, q3 = Q3, + index_state = IndexState }) -> + %% HighSeqId is high in the sense that it must be higher than the + %% seq_id in Gamma, but it's also the lowest of the betas that we + %% transfer from q2 to gamma. + {HighSeqId, Len1, Q2a, IndexState1} = + push_betas_to_gammas(fun queue:out/1, undefined, Q2, IndexState), + Gamma1 = #gamma { seq_id = Gamma1SeqId } = + combine_gammas(Gamma, #gamma { seq_id = HighSeqId, count = Len1 }), + State1 = State #vqstate { q2 = Q2a, gamma = Gamma1, + index_state = IndexState1 }, + case queue:out(Q3) of + {empty, _Q3} -> State1; + {{value, #beta { seq_id = SeqId }}, _Q3a} -> + Limit = rabbit_queue_index:next_segment_boundary(SeqId), + case Gamma1SeqId of + Limit -> %% already only holding the minimum, nothing to do + State1; + _ when Gamma1SeqId == undefined orelse + (is_integer(Gamma1SeqId) andalso Gamma1SeqId > Limit) -> + %% ASSERTION (sadly large!) + %% This says that if Gamma1SeqId /= undefined then + %% the gap from Limit to Gamma1SeqId is an integer + %% multiple of segment_size + 0 = case Gamma1SeqId of + undefined -> 0; + _ -> (Gamma1SeqId - Limit) rem + rabbit_queue_index:segment_size() + end, + %% LowSeqId is low in the sense that it must be + %% lower than the seq_id in gamma1, in fact either + %% gamma1 has undefined as its seq_id or there + %% does not exist a seq_id X s.t. X > LowSeqId and + %% X < gamma1's seq_id (would be +1 if it wasn't + %% for the possibility of gaps in the seq_ids). + %% But because we use queue:out_r, LowSeqId is + %% actually also the highest seq_id of the betas we + %% transfer from q3 to gammas. + {LowSeqId, Len2, Q3b, IndexState2} = + push_betas_to_gammas(fun queue:out_r/1, Limit, Q3, + IndexState1), + true = Gamma1SeqId > LowSeqId, %% ASSERTION + Gamma2 = combine_gammas( + #gamma { seq_id = Limit, count = Len2}, Gamma1), + State1 #vqstate { q3 = Q3b, gamma = Gamma2, + index_state = IndexState2 } + end + end. + +push_betas_to_gammas(Generator, Limit, Q, IndexState) -> + case Generator(Q) of + {empty, Qa} -> {undefined, 0, Qa, IndexState}; + {{value, #beta { seq_id = SeqId }}, _Qa} -> + {Count, Qb, IndexState1} = + push_betas_to_gammas(Generator, Limit, Q, 0, IndexState), + {SeqId, Count, Qb, IndexState1} + end. + +push_betas_to_gammas(Generator, Limit, Q, Count, IndexState) -> + case Generator(Q) of + {empty, Qa} -> {Count, Qa, IndexState}; + {{value, #beta { seq_id = SeqId }}, _Qa} + when Limit /= undefined andalso SeqId < Limit -> + {Count, Q, IndexState}; + {{value, #beta { msg_id = MsgId, seq_id = SeqId, + is_persistent = IsPersistent, + is_delivered = IsDelivered, + index_on_disk = IndexOnDisk}}, Qa} -> + IndexState1 = + case IndexOnDisk of + true -> IndexState; + false -> + {true, IndexState2} = + maybe_write_index_to_disk( + true, IsPersistent, MsgId, + SeqId, IsDelivered, IndexState), + IndexState2 + end, + push_betas_to_gammas(Generator, Limit, Qa, Count + 1, IndexState1) + end. |