summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorMatthias Radestock <matthias@rabbitmq.com>2011-10-15 14:27:24 +0100
committerMatthias Radestock <matthias@rabbitmq.com>2011-10-15 14:27:24 +0100
commit0899cc6af2ddde32ca28e1f8f273b2251a82f94c (patch)
tree9413f29fafa116a78ebc9151f09181488f79820d
parentfa3cfa9742c2be448283cbd36da27cf7357d17ec (diff)
parent0e3b7b08aec68f9ec177d97cf7250bacdb2caa48 (diff)
downloadrabbitmq-server-0899cc6af2ddde32ca28e1f8f273b2251a82f94c.tar.gz
merge default into bug24455
-rw-r--r--src/bpqueue.erl271
-rw-r--r--src/lqueue.erl128
-rw-r--r--src/rabbit_tests.erl149
-rw-r--r--src/rabbit_variable_queue.erl653
4 files changed, 429 insertions, 772 deletions
diff --git a/src/bpqueue.erl b/src/bpqueue.erl
deleted file mode 100644
index 71a34262..00000000
--- a/src/bpqueue.erl
+++ /dev/null
@@ -1,271 +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 Developer of the Original Code is VMware, Inc.
-%% Copyright (c) 2007-2011 VMware, Inc. All rights reserved.
-%%
-
--module(bpqueue).
-
-%% Block-prefixed queue. From the perspective of the queue interface
-%% the datastructure acts like a regular queue where each value is
-%% paired with the prefix.
-%%
-%% This is implemented as a queue of queues, which is more space and
-%% time efficient, whilst supporting the normal queue interface. Each
-%% inner queue has a prefix, which does not need to be unique, and it
-%% is guaranteed that no two consecutive blocks have the same
-%% prefix. len/1 returns the flattened length of the queue and is
-%% O(1).
-
--export([new/0, is_empty/1, len/1, in/3, in_r/3, out/1, out_r/1, join/2,
- foldl/3, foldr/3, from_list/1, to_list/1, map_fold_filter_l/4,
- map_fold_filter_r/4]).
-
-%%----------------------------------------------------------------------------
-
--ifdef(use_specs).
-
--export_type([bpqueue/0]).
-
--type(bpqueue() :: {non_neg_integer(), queue()}).
--type(prefix() :: any()).
--type(value() :: any()).
--type(result() :: ({'empty', bpqueue()} |
- {{'value', prefix(), value()}, bpqueue()})).
-
--spec(new/0 :: () -> bpqueue()).
--spec(is_empty/1 :: (bpqueue()) -> boolean()).
--spec(len/1 :: (bpqueue()) -> non_neg_integer()).
--spec(in/3 :: (prefix(), value(), bpqueue()) -> bpqueue()).
--spec(in_r/3 :: (prefix(), value(), bpqueue()) -> bpqueue()).
--spec(out/1 :: (bpqueue()) -> result()).
--spec(out_r/1 :: (bpqueue()) -> result()).
--spec(join/2 :: (bpqueue(), bpqueue()) -> bpqueue()).
--spec(foldl/3 :: (fun ((prefix(), value(), B) -> B), B, bpqueue()) -> B).
--spec(foldr/3 :: (fun ((prefix(), value(), B) -> B), B, bpqueue()) -> B).
--spec(from_list/1 :: ([{prefix(), [value()]}]) -> bpqueue()).
--spec(to_list/1 :: (bpqueue()) -> [{prefix(), [value()]}]).
--spec(map_fold_filter_l/4 :: ((fun ((prefix()) -> boolean())),
- (fun ((value(), B) ->
- ({prefix(), value(), B} | 'stop'))),
- B,
- bpqueue()) ->
- {bpqueue(), B}).
--spec(map_fold_filter_r/4 :: ((fun ((prefix()) -> boolean())),
- (fun ((value(), B) ->
- ({prefix(), value(), B} | 'stop'))),
- B,
- bpqueue()) ->
- {bpqueue(), B}).
-
--endif.
-
-%%----------------------------------------------------------------------------
-
-new() -> {0, queue:new()}.
-
-is_empty({0, _Q}) -> true;
-is_empty(_BPQ) -> false.
-
-len({N, _Q}) -> N.
-
-in(Prefix, Value, {0, Q}) ->
- {1, queue:in({Prefix, queue:from_list([Value])}, Q)};
-in(Prefix, Value, BPQ) ->
- in1({fun queue:in/2, fun queue:out_r/1}, Prefix, Value, BPQ).
-
-in_r(Prefix, Value, BPQ = {0, _Q}) ->
- in(Prefix, Value, BPQ);
-in_r(Prefix, Value, BPQ) ->
- in1({fun queue:in_r/2, fun queue:out/1}, Prefix, Value, BPQ).
-
-in1({In, Out}, Prefix, Value, {N, Q}) ->
- {N+1, case Out(Q) of
- {{value, {Prefix, InnerQ}}, Q1} ->
- In({Prefix, In(Value, InnerQ)}, Q1);
- {{value, {_Prefix, _InnerQ}}, _Q1} ->
- In({Prefix, queue:in(Value, queue:new())}, Q)
- end}.
-
-in_q(Prefix, Queue, BPQ = {0, Q}) ->
- case queue:len(Queue) of
- 0 -> BPQ;
- N -> {N, queue:in({Prefix, Queue}, Q)}
- end;
-in_q(Prefix, Queue, BPQ) ->
- in_q1({fun queue:in/2, fun queue:out_r/1,
- fun queue:join/2},
- Prefix, Queue, BPQ).
-
-in_q_r(Prefix, Queue, BPQ = {0, _Q}) ->
- in_q(Prefix, Queue, BPQ);
-in_q_r(Prefix, Queue, BPQ) ->
- in_q1({fun queue:in_r/2, fun queue:out/1,
- fun (T, H) -> queue:join(H, T) end},
- Prefix, Queue, BPQ).
-
-in_q1({In, Out, Join}, Prefix, Queue, BPQ = {N, Q}) ->
- case queue:len(Queue) of
- 0 -> BPQ;
- M -> {N + M, case Out(Q) of
- {{value, {Prefix, InnerQ}}, Q1} ->
- In({Prefix, Join(InnerQ, Queue)}, Q1);
- {{value, {_Prefix, _InnerQ}}, _Q1} ->
- In({Prefix, Queue}, Q)
- end}
- end.
-
-out({0, _Q} = BPQ) -> {empty, BPQ};
-out(BPQ) -> out1({fun queue:in_r/2, fun queue:out/1}, BPQ).
-
-out_r({0, _Q} = BPQ) -> {empty, BPQ};
-out_r(BPQ) -> out1({fun queue:in/2, fun queue:out_r/1}, BPQ).
-
-out1({In, Out}, {N, Q}) ->
- {{value, {Prefix, InnerQ}}, Q1} = Out(Q),
- {{value, Value}, InnerQ1} = Out(InnerQ),
- Q2 = case queue:is_empty(InnerQ1) of
- true -> Q1;
- false -> In({Prefix, InnerQ1}, Q1)
- end,
- {{value, Prefix, Value}, {N-1, Q2}}.
-
-join({0, _Q}, BPQ) ->
- BPQ;
-join(BPQ, {0, _Q}) ->
- BPQ;
-join({NHead, QHead}, {NTail, QTail}) ->
- {{value, {Prefix, InnerQHead}}, QHead1} = queue:out_r(QHead),
- {NHead + NTail,
- case queue:out(QTail) of
- {{value, {Prefix, InnerQTail}}, QTail1} ->
- queue:join(
- queue:in({Prefix, queue:join(InnerQHead, InnerQTail)}, QHead1),
- QTail1);
- {{value, {_Prefix, _InnerQTail}}, _QTail1} ->
- queue:join(QHead, QTail)
- end}.
-
-foldl(_Fun, Init, {0, _Q}) -> Init;
-foldl( Fun, Init, {_N, Q}) -> fold1(fun queue:out/1, Fun, Init, Q).
-
-foldr(_Fun, Init, {0, _Q}) -> Init;
-foldr( Fun, Init, {_N, Q}) -> fold1(fun queue:out_r/1, Fun, Init, Q).
-
-fold1(Out, Fun, Init, Q) ->
- case Out(Q) of
- {empty, _Q} ->
- Init;
- {{value, {Prefix, InnerQ}}, Q1} ->
- fold1(Out, Fun, fold1(Out, Fun, Prefix, Init, InnerQ), Q1)
- end.
-
-fold1(Out, Fun, Prefix, Init, InnerQ) ->
- case Out(InnerQ) of
- {empty, _Q} ->
- Init;
- {{value, Value}, InnerQ1} ->
- fold1(Out, Fun, Prefix, Fun(Prefix, Value, Init), InnerQ1)
- end.
-
-from_list(List) ->
- {FinalPrefix, FinalInnerQ, ListOfPQs1, Len} =
- lists:foldl(
- fun ({_Prefix, []}, Acc) ->
- Acc;
- ({Prefix, InnerList}, {Prefix, InnerQ, ListOfPQs, LenAcc}) ->
- {Prefix, queue:join(InnerQ, queue:from_list(InnerList)),
- ListOfPQs, LenAcc + length(InnerList)};
- ({Prefix1, InnerList}, {Prefix, InnerQ, ListOfPQs, LenAcc}) ->
- {Prefix1, queue:from_list(InnerList),
- [{Prefix, InnerQ} | ListOfPQs], LenAcc + length(InnerList)}
- end, {undefined, queue:new(), [], 0}, List),
- ListOfPQs2 = [{FinalPrefix, FinalInnerQ} | ListOfPQs1],
- [{undefined, InnerQ1} | Rest] = All = lists:reverse(ListOfPQs2),
- {Len, queue:from_list(case queue:is_empty(InnerQ1) of
- true -> Rest;
- false -> All
- end)}.
-
-to_list({0, _Q}) -> [];
-to_list({_N, Q}) -> [{Prefix, queue:to_list(InnerQ)} ||
- {Prefix, InnerQ} <- queue:to_list(Q)].
-
-%% map_fold_filter_[lr](FilterFun, Fun, Init, BPQ) -> {BPQ, Init}
-%% where FilterFun(Prefix) -> boolean()
-%% Fun(Value, Init) -> {Prefix, Value, Init} | stop
-%%
-%% The filter fun allows you to skip very quickly over blocks that
-%% you're not interested in. Such blocks appear in the resulting bpq
-%% without modification. The Fun is then used both to map the value,
-%% which also allows you to change the prefix (and thus block) of the
-%% value, and also to modify the Init/Acc (just like a fold). If the
-%% Fun returns 'stop' then it is not applied to any further items.
-map_fold_filter_l(_PFilter, _Fun, Init, BPQ = {0, _Q}) ->
- {BPQ, Init};
-map_fold_filter_l(PFilter, Fun, Init, {N, Q}) ->
- map_fold_filter1({fun queue:out/1, fun queue:in/2,
- fun in_q/3, fun join/2},
- N, PFilter, Fun, Init, Q, new()).
-
-map_fold_filter_r(_PFilter, _Fun, Init, BPQ = {0, _Q}) ->
- {BPQ, Init};
-map_fold_filter_r(PFilter, Fun, Init, {N, Q}) ->
- map_fold_filter1({fun queue:out_r/1, fun queue:in_r/2,
- fun in_q_r/3, fun (T, H) -> join(H, T) end},
- N, PFilter, Fun, Init, Q, new()).
-
-map_fold_filter1(Funs = {Out, _In, InQ, Join}, Len, PFilter, Fun,
- Init, Q, QNew) ->
- case Out(Q) of
- {empty, _Q} ->
- {QNew, Init};
- {{value, {Prefix, InnerQ}}, Q1} ->
- case PFilter(Prefix) of
- true ->
- {Init1, QNew1, Cont} =
- map_fold_filter2(Funs, Fun, Prefix, Prefix,
- Init, InnerQ, QNew, queue:new()),
- case Cont of
- false -> {Join(QNew1, {Len - len(QNew1), Q1}), Init1};
- true -> map_fold_filter1(Funs, Len, PFilter, Fun,
- Init1, Q1, QNew1)
- end;
- false ->
- map_fold_filter1(Funs, Len, PFilter, Fun,
- Init, Q1, InQ(Prefix, InnerQ, QNew))
- end
- end.
-
-map_fold_filter2(Funs = {Out, In, InQ, _Join}, Fun, OrigPrefix, Prefix,
- Init, InnerQ, QNew, InnerQNew) ->
- case Out(InnerQ) of
- {empty, _Q} ->
- {Init, InQ(OrigPrefix, InnerQ,
- InQ(Prefix, InnerQNew, QNew)), true};
- {{value, Value}, InnerQ1} ->
- case Fun(Value, Init) of
- stop ->
- {Init, InQ(OrigPrefix, InnerQ,
- InQ(Prefix, InnerQNew, QNew)), false};
- {Prefix1, Value1, Init1} ->
- {Prefix2, QNew1, InnerQNew1} =
- case Prefix1 =:= Prefix of
- true -> {Prefix, QNew, In(Value1, InnerQNew)};
- false -> {Prefix1, InQ(Prefix, InnerQNew, QNew),
- In(Value1, queue:new())}
- end,
- map_fold_filter2(Funs, Fun, OrigPrefix, Prefix2,
- Init1, InnerQ1, QNew1, InnerQNew1)
- end
- end.
diff --git a/src/lqueue.erl b/src/lqueue.erl
new file mode 100644
index 00000000..9f58b250
--- /dev/null
+++ b/src/lqueue.erl
@@ -0,0 +1,128 @@
+%% 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 Developer of the Original Code is VMware, Inc.
+%% Copyright (c) 2011-2011 VMware, Inc. All rights reserved.
+%%
+
+-module(lqueue).
+
+-export([new/0, is_empty/1, len/1, in/2, in_r/2, out/1, out_r/1, join/2,
+ foldl/3, foldr/3, from_list/1, to_list/1, peek/1, peek_r/1]).
+
+-define(QUEUE, queue).
+
+-ifdef(use_specs).
+
+-export_type([?MODULE/0]).
+
+-opaque(?MODULE() ::
+ {non_neg_integer(), ?QUEUE(), maybe_value(), maybe_value()}).
+-type(value() :: any()).
+-type(maybe_value() :: {'value', value()} | 'empty').
+-type(result() :: {maybe_value(), ?MODULE}).
+
+-spec(new/0 :: () -> ?MODULE()).
+-spec(is_empty/1 :: (?MODULE()) -> boolean()).
+-spec(len/1 :: (?MODULE()) -> non_neg_integer()).
+-spec(in/2 :: (value(), ?MODULE()) -> ?MODULE()).
+-spec(in_r/2 :: (value(), ?MODULE()) -> ?MODULE()).
+-spec(out/1 :: (?MODULE()) -> result()).
+-spec(out_r/1 :: (?MODULE()) -> result()).
+-spec(join/2 :: (?MODULE(), ?MODULE()) -> ?MODULE()).
+-spec(foldl/3 :: (fun ((value(), B) -> B), B, ?MODULE()) -> B).
+-spec(foldr/3 :: (fun ((value(), B) -> B), B, ?MODULE()) -> B).
+-spec(from_list/1 :: ([value()]) -> ?MODULE()).
+-spec(to_list/1 :: (?MODULE()) -> [value()]).
+-spec(peek(?MODULE) -> maybe_value()).
+-spec(peek_r(?MODULE) -> maybe_value()).
+
+-endif.
+
+new() -> {0, ?QUEUE:new(), empty, empty}.
+
+is_empty({0, _Q, _I, _O}) -> true;
+is_empty(_ ) -> false.
+
+in(V, {0, Q, empty, empty}) -> {1, Q, empty, {value, V}};
+in(V, {1, Q, empty, V1 }) -> {2, Q, {value, V}, V1};
+in(V, {1, Q, V1, empty}) -> {2, Q, {value, V}, V1};
+in(V, {N, Q, {value, V1}, V2 }) -> {N+1, ?QUEUE:in(V1, Q), {value, V}, V2}.
+
+in_r(V, {0, Q, empty, empty }) -> {1, Q, {value, V}, empty};
+in_r(V, {1, Q, V1, empty }) -> {2, Q, V1, {value, V}};
+in_r(V, {1, Q, empty, V1 }) -> {2, Q, V1, {value, V}};
+in_r(V, {N, Q, V1, {value, V2}}) -> {N+1, ?QUEUE:in_r(V2, Q), V1, {value, V}}.
+
+out({0, _Q, _I, _O} = Q) -> {empty, Q};
+out({1, Q, empty, V}) -> {V, {0, Q, empty, empty}};
+out({1, Q, V, empty}) -> {V, {0, Q, empty, empty}};
+out({2, Q, V1, V2}) -> {V2, {1, Q, empty, V1}};
+out({N, Q, V1, V2}) -> {V3 = {value, _V}, Q1} = ?QUEUE:out(Q),
+ {V2, {N-1, Q1, V1, V3}}.
+
+out_r({0, _Q, _I, _O} = Q) -> {empty, Q};
+out_r({1, Q, empty, V}) -> {V, {0, Q, empty, empty}};
+out_r({1, Q, V, empty}) -> {V, {0, Q, empty, empty}};
+out_r({2, Q, V1, V2}) -> {V1, {1, Q, V2, empty}};
+out_r({N, Q, V1, V2}) -> {V3 = {value, _V}, Q1} = ?QUEUE:out_r(Q),
+ {V1, {N-1, Q1, V3, V2}}.
+
+join({0, _, _, _}, Q) -> Q;
+join(Q, {0, _, _, _}) -> Q;
+
+join({1, _, empty, {value, V} }, Q) -> in_r(V, Q);
+join({1, _, {value, V}, empty }, Q) -> in_r(V, Q);
+join({2, _, {value, V1}, {value, V2}}, Q) -> in_r(V2, in_r(V1, Q));
+
+join(Q, {1, _, empty, {value, V} }) -> in(V, Q);
+join(Q, {1, _, {value, V}, empty }) -> in(V, Q);
+join(Q, {2, _, {value, V1}, {value, V2}}) -> in(V1, in(V2, Q));
+
+join({L1, Q1, {value, InV}, Out}, {L2, Q2, In, {value, OutV}}) ->
+ {L1+L2, ?QUEUE:join(?QUEUE:in(InV, Q1), ?QUEUE:in_r(OutV, Q2)), In, Out}.
+
+to_list({0, _, _, _}) -> [];
+to_list({1, _, empty, {value, V}}) -> [V];
+to_list({1, _, {value, V}, empty}) -> [V];
+to_list({_, Q, {value, InV}, {value, OutV}}) ->
+ [OutV | ?QUEUE:to_list(Q) ++ [InV]].
+
+from_list([]) -> new();
+from_list([V]) -> in(V, new());
+from_list([V1, V2]) -> in(V2, in(V1, new()));
+from_list([V1 | Vs] = L) -> Q = ?QUEUE:from_list(Vs),
+ {V2, Q1} = ?QUEUE:out_r(Q),
+ {length(L), Q1, V2, V1}.
+
+foldl(Fun, Init, Q) ->
+ case out(Q) of
+ {empty, _Q} -> Init;
+ {{value, V}, Q1} -> foldl(Fun, Fun(V, Init), Q1)
+ end.
+
+foldr(Fun, Init, Q) ->
+ case out_r(Q) of
+ {empty, _Q} -> Init;
+ {{value, V}, Q1} -> foldr(Fun, Fun(V, Init), Q1)
+ end.
+
+len({L, _Q, _I, _O}) ->
+ L.
+
+peek({0, _, _, _}) -> empty;
+peek({1, _, V, empty}) -> V;
+peek({_L, _Q, _I, O}) -> O.
+
+peek_r({0, _, _, _}) -> empty;
+peek_r({1, _, empty, V}) -> V;
+peek_r({_L, _Q, I, _O}) -> I.
diff --git a/src/rabbit_tests.erl b/src/rabbit_tests.erl
index 7a36dd60..78857092 100644
--- a/src/rabbit_tests.erl
+++ b/src/rabbit_tests.erl
@@ -44,7 +44,6 @@ all_tests() ->
passed = test_file_handle_cache(),
passed = test_backing_queue(),
passed = test_priority_queue(),
- passed = test_bpqueue(),
passed = test_pg_local(),
passed = test_unfold(),
passed = test_supervisor_delayed_restart(),
@@ -262,143 +261,6 @@ test_priority_queue(Q) ->
priority_queue:to_list(Q),
priority_queue_out_all(Q)}.
-test_bpqueue() ->
- Q = bpqueue:new(),
- true = bpqueue:is_empty(Q),
- 0 = bpqueue:len(Q),
- [] = bpqueue:to_list(Q),
-
- Q1 = bpqueue_test(fun bpqueue:in/3, fun bpqueue:out/1,
- fun bpqueue:to_list/1,
- fun bpqueue:foldl/3, fun bpqueue:map_fold_filter_l/4),
- Q2 = bpqueue_test(fun bpqueue:in_r/3, fun bpqueue:out_r/1,
- fun (QR) -> lists:reverse(
- [{P, lists:reverse(L)} ||
- {P, L} <- bpqueue:to_list(QR)])
- end,
- fun bpqueue:foldr/3, fun bpqueue:map_fold_filter_r/4),
-
- [{foo, [1, 2]}, {bar, [3]}] = bpqueue:to_list(bpqueue:join(Q, Q1)),
- [{bar, [3]}, {foo, [2, 1]}] = bpqueue:to_list(bpqueue:join(Q2, Q)),
- [{foo, [1, 2]}, {bar, [3, 3]}, {foo, [2,1]}] =
- bpqueue:to_list(bpqueue:join(Q1, Q2)),
-
- [{foo, [1, 2]}, {bar, [3]}, {foo, [1, 2]}, {bar, [3]}] =
- bpqueue:to_list(bpqueue:join(Q1, Q1)),
-
- [{foo, [1, 2]}, {bar, [3]}] =
- bpqueue:to_list(
- bpqueue:from_list(
- [{x, []}, {foo, [1]}, {y, []}, {foo, [2]}, {bar, [3]}, {z, []}])),
-
- [{undefined, [a]}] = bpqueue:to_list(bpqueue:from_list([{undefined, [a]}])),
-
- {4, [a,b,c,d]} =
- bpqueue:foldl(
- fun (Prefix, Value, {Prefix, Acc}) ->
- {Prefix + 1, [Value | Acc]}
- end,
- {0, []}, bpqueue:from_list([{0,[d]}, {1,[c]}, {2,[b]}, {3,[a]}])),
-
- [{bar,3}, {foo,2}, {foo,1}] =
- bpqueue:foldr(fun (P, V, I) -> [{P,V} | I] end, [], Q2),
-
- BPQL = [{foo,[1,2,2]}, {bar,[3,4,5]}, {foo,[5,6,7]}],
- BPQ = bpqueue:from_list(BPQL),
-
- %% no effect
- {BPQL, 0} = bpqueue_mffl([none], {none, []}, BPQ),
- {BPQL, 0} = bpqueue_mffl([foo,bar], {none, [1]}, BPQ),
- {BPQL, 0} = bpqueue_mffl([bar], {none, [3]}, BPQ),
- {BPQL, 0} = bpqueue_mffr([bar], {foo, [5]}, BPQ),
-
- %% process 1 item
- {[{foo,[-1,2,2]}, {bar,[3,4,5]}, {foo,[5,6,7]}], 1} =
- bpqueue_mffl([foo,bar], {foo, [2]}, BPQ),
- {[{foo,[1,2,2]}, {bar,[-3,4,5]}, {foo,[5,6,7]}], 1} =
- bpqueue_mffl([bar], {bar, [4]}, BPQ),
- {[{foo,[1,2,2]}, {bar,[3,4,5]}, {foo,[5,6,-7]}], 1} =
- bpqueue_mffr([foo,bar], {foo, [6]}, BPQ),
- {[{foo,[1,2,2]}, {bar,[3,4]}, {baz,[-5]}, {foo,[5,6,7]}], 1} =
- bpqueue_mffr([bar], {baz, [4]}, BPQ),
-
- %% change prefix
- {[{bar,[-1,-2,-2,-3,-4,-5,-5,-6,-7]}], 9} =
- bpqueue_mffl([foo,bar], {bar, []}, BPQ),
- {[{bar,[-1,-2,-2,3,4,5]}, {foo,[5,6,7]}], 3} =
- bpqueue_mffl([foo], {bar, [5]}, BPQ),
- {[{bar,[-1,-2,-2,3,4,5,-5,-6]}, {foo,[7]}], 5} =
- bpqueue_mffl([foo], {bar, [7]}, BPQ),
- {[{foo,[1,2,2,-3,-4]}, {bar,[5]}, {foo,[5,6,7]}], 2} =
- bpqueue_mffl([bar], {foo, [5]}, BPQ),
- {[{bar,[-1,-2,-2,3,4,5,-5,-6,-7]}], 6} =
- bpqueue_mffl([foo], {bar, []}, BPQ),
- {[{foo,[1,2,2,-3,-4,-5,5,6,7]}], 3} =
- bpqueue_mffl([bar], {foo, []}, BPQ),
-
- %% edge cases
- {[{foo,[-1,-2,-2]}, {bar,[3,4,5]}, {foo,[5,6,7]}], 3} =
- bpqueue_mffl([foo], {foo, [5]}, BPQ),
- {[{foo,[1,2,2]}, {bar,[3,4,5]}, {foo,[-5,-6,-7]}], 3} =
- bpqueue_mffr([foo], {foo, [2]}, BPQ),
-
- passed.
-
-bpqueue_test(In, Out, List, Fold, MapFoldFilter) ->
- Q = bpqueue:new(),
- {empty, _Q} = Out(Q),
-
- ok = Fold(fun (Prefix, Value, ok) -> {error, Prefix, Value} end, ok, Q),
- {Q1M, 0} = MapFoldFilter(fun(_P) -> throw(explosion) end,
- fun(_V, _N) -> throw(explosion) end, 0, Q),
- [] = bpqueue:to_list(Q1M),
-
- Q1 = In(bar, 3, In(foo, 2, In(foo, 1, Q))),
- false = bpqueue:is_empty(Q1),
- 3 = bpqueue:len(Q1),
- [{foo, [1, 2]}, {bar, [3]}] = List(Q1),
-
- {{value, foo, 1}, Q3} = Out(Q1),
- {{value, foo, 2}, Q4} = Out(Q3),
- {{value, bar, 3}, _Q5} = Out(Q4),
-
- F = fun (QN) ->
- MapFoldFilter(fun (foo) -> true;
- (_) -> false
- end,
- fun (2, _Num) -> stop;
- (V, Num) -> {bar, -V, V - Num} end,
- 0, QN)
- end,
- {Q6, 0} = F(Q),
- [] = bpqueue:to_list(Q6),
- {Q7, 1} = F(Q1),
- [{bar, [-1]}, {foo, [2]}, {bar, [3]}] = List(Q7),
-
- Q1.
-
-bpqueue_mffl(FF1A, FF2A, BPQ) ->
- bpqueue_mff(fun bpqueue:map_fold_filter_l/4, FF1A, FF2A, BPQ).
-
-bpqueue_mffr(FF1A, FF2A, BPQ) ->
- bpqueue_mff(fun bpqueue:map_fold_filter_r/4, FF1A, FF2A, BPQ).
-
-bpqueue_mff(Fold, FF1A, FF2A, BPQ) ->
- FF1 = fun (Prefixes) ->
- fun (P) -> lists:member(P, Prefixes) end
- end,
- FF2 = fun ({Prefix, Stoppers}) ->
- fun (Val, Num) ->
- case lists:member(Val, Stoppers) of
- true -> stop;
- false -> {Prefix, -Val, 1 + Num}
- end
- end
- end,
- Queue_to_list = fun ({LHS, RHS}) -> {bpqueue:to_list(LHS), RHS} end,
-
- Queue_to_list(Fold(FF1(FF1A), FF2(FF2A), 0, BPQ)).
-
test_simple_n_element_queue(N) ->
Items = lists:seq(1, N),
Q = priority_queue_in_all(priority_queue:new(), Items),
@@ -2374,12 +2236,7 @@ test_variable_queue_requeue(VQ0) ->
Seq = lists:seq(1, Count),
VQ1 = rabbit_variable_queue:set_ram_duration_target(0, VQ0),
VQ2 = variable_queue_publish(false, Count, VQ1),
- {VQ3, Acks} = lists:foldl(
- fun (_N, {VQN, AckTags}) ->
- {{#basic_message{}, false, AckTag, _}, VQM} =
- rabbit_variable_queue:fetch(true, VQN),
- {VQM, [AckTag | AckTags]}
- end, {VQ2, []}, Seq),
+ {VQ3, Acks} = variable_queue_fetch(Count, false, false, Count, VQ2),
Subset = lists:foldl(fun ({Ack, N}, Acc) when N rem Interval == 0 ->
[Ack | Acc];
(_, Acc) ->
@@ -2507,7 +2364,7 @@ test_variable_queue_partial_segments_delta_thing(VQ0) ->
{_Duration, VQ2} = rabbit_variable_queue:ram_duration(VQ1),
VQ3 = check_variable_queue_status(
rabbit_variable_queue:set_ram_duration_target(0, VQ2),
- %% one segment in q3 as betas, and half a segment in delta
+ %% one segment in q3, and half a segment in delta
[{delta, {delta, SegmentSize, HalfSegment, OneAndAHalfSegment}},
{q3, SegmentSize},
{len, SegmentSize + HalfSegment}]),
@@ -2523,7 +2380,7 @@ test_variable_queue_partial_segments_delta_thing(VQ0) ->
SegmentSize + HalfSegment + 1, VQ5),
VQ7 = check_variable_queue_status(
VQ6,
- %% the half segment should now be in q3 as betas
+ %% the half segment should now be in q3
[{q1, 1},
{delta, {delta, undefined, 0, undefined}},
{q3, HalfSegment},
diff --git a/src/rabbit_variable_queue.erl b/src/rabbit_variable_queue.erl
index b853d983..d418a10f 100644
--- a/src/rabbit_variable_queue.erl
+++ b/src/rabbit_variable_queue.erl
@@ -49,17 +49,15 @@
%% within the queue are always held on disk, *in addition* to being in
%% one of the above classifications.
%%
-%% Also note that within this code, the term gamma never
-%% appears. Instead, gammas are defined by betas who have had their
-%% queue position recorded on disk.
+%% Also note that within this code, the term gamma seldom
+%% appears. It's frequently the case that gammas are defined by betas
+%% who have had their queue position recorded on disk.
%%
%% In general, messages move q1 -> q2 -> delta -> q3 -> q4, though
%% many of these steps are frequently skipped. q1 and q4 only hold
-%% alphas, q2 and q3 hold both betas and gammas (as queues of queues,
-%% using the bpqueue module where the block prefix determines whether
-%% they're betas or gammas). When a message arrives, its
-%% classification is determined. It is then added to the rightmost
-%% appropriate queue.
+%% alphas, q2 and q3 hold both betas and gammas. When a message
+%% arrives, its classification is determined. It is then added to the
+%% rightmost appropriate queue.
%%
%% If a new message is determined to be a beta or gamma, q1 is
%% empty. If a new message is determined to be a delta, q1 and q2 are
@@ -74,14 +72,15 @@
%%
%% The duration indicated to us by the memory_monitor is used to
%% calculate, given our current ingress and egress rates, how many
-%% messages we should hold in RAM. We track the ingress and egress
-%% rates for both messages and pending acks and rates for both are
-%% considered when calculating the number of messages to hold in
-%% RAM. When we need to push alphas to betas or betas to gammas, we
-%% favour writing out messages that are further from the head of the
-%% queue. This minimises writes to disk, as the messages closer to the
-%% tail of the queue stay in the queue for longer, thus do not need to
-%% be replaced as quickly by sending other messages to disk.
+%% messages we should hold in RAM (i.e. as alphas). We track the
+%% ingress and egress rates for both messages and pending acks and
+%% rates for both are considered when calculating the number of
+%% messages to hold in RAM. When we need to push alphas to betas or
+%% betas to gammas, we favour writing out messages that are further
+%% from the head of the queue. This minimises writes to disk, as the
+%% messages closer to the tail of the queue stay in the queue for
+%% longer, thus do not need to be replaced as quickly by sending other
+%% messages to disk.
%%
%% Whilst messages are pushed to disk and forgotten from RAM as soon
%% as requested by a new setting of the queue RAM duration, the
@@ -119,17 +118,60 @@
%% getting rid of messages as fast as possible and remaining
%% responsive, and using only the egress rate impacts that goal.
%%
-%% If a queue is full of transient messages, then the transition from
-%% betas to deltas will be potentially very expensive as millions of
-%% entries must be written to disk by the queue_index module. This can
-%% badly stall the queue. In order to avoid this, the proportion of
-%% gammas / (betas+gammas) must not be lower than (betas+gammas) /
-%% (alphas+betas+gammas). As the queue grows or available memory
-%% shrinks, the latter ratio increases, requiring the conversion of
-%% more gammas to betas in order to maintain the invariant. At the
-%% point at which betas and gammas must be converted to deltas, there
-%% should be very few betas remaining, thus the transition is fast (no
-%% work needs to be done for the gamma -> delta transition).
+%% Once the queue has more alphas than the target_ram_count, the
+%% surplus must be converted to betas, if not gammas, if not rolled
+%% into delta. The conditions under which these transitions occur
+%% reflect the conflicting goals of minimising RAM cost per msg, and
+%% minimising CPU cost per msg. Once the msg has become a beta, its
+%% payload is no longer in RAM, thus a read from the msg_store must
+%% occur before the msg can be delivered, but the RAM cost of a beta
+%% is the same as a gamma, so converting a beta to gamma will not free
+%% up any further RAM. To reduce the RAM cost further, the gamma must
+%% be rolled into delta. Whilst recovering a beta or a gamma to an
+%% alpha requires only one disk read (from the msg_store), recovering
+%% a msg from within delta will require two reads (queue_index and
+%% then msg_store). But delta has a near-0 per-msg RAM cost. So the
+%% conflict is between you using delta more, which will free up more
+%% memory, but require additional CPU and disk ops, versus using delta
+%% less and gammas and betas more, which will cost more memory, but
+%% require fewer disk ops and less CPU overhead.
+%%
+%% The decision taken is that once a gamma is at the head of q2, or at
+%% the tail of q3, it'll be immediately rolled into delta. This
+%% ensures that the memory use of a queue stays relatively flat, even
+%% as it gets longer: we avoid a cliff face where an event forces
+%% potentially millions of gammas to be suddenly rolled into
+%% delta. This comes at the cost of needing to do additional reads
+%% from queue_index as gammas are eventually brought out of delta.
+%%
+%% However, that does not yet explain when a beta is converted to a
+%% gamma. In the case of a persistent msg published to a durable
+%% queue, the msg is immediately written to the msg_store and
+%% queue_index. If then additionally converted from an alpha, it'll
+%% immediately go to a gamma (as it's already in queue_index), and
+%% cannot exist as a beta. Thus a durable queue with a mixture of
+%% persistent and transient msgs in it which has more messages than
+%% permitted by the target_ram_count may contain an interspersed
+%% mixture of betas and gammas in q2 and q3, but with a beta at the
+%% head of q2 and the tail of q3.
+%%
+%% There is then a ratio that controls how many betas and gammas there
+%% can be. This is based on the target_ram_count and thus expresses
+%% the fact that as the number of permitted alphas in the queue falls,
+%% so should the number of betas and gammas fall (i.e. delta
+%% grows). If q2 and q3 contain more than the permitted number of
+%% betas and gammas, then the surplus are forcibly converted to gammas
+%% (as necessary) and then rolled into delta. The ratio is that the
+%% size of delta / (betas+gammas+delta) should equal
+%% (betas+gammas+delta)/(alphas+betas+gammas+delta). I.e. as alphas
+%% shrink to 0, so must betas and gammas. The actual calculation done
+%% is a rearrangement of this, and uses target_ram_count instead of
+%% alphas. The reason for this is that once the queue length starts
+%% falling, any alphas in q4 will be sent out first, thus reducing the
+%% number of alphas. Thus if the real number of alphas was used then
+%% this scenario could result in more conversions towards delta,
+%% simply because the queue is starting to drain. By using the
+%% target_ram_count, we avoid this problem.
%%
%% The conversion of betas to gammas is done in batches of exactly
%% ?IO_BATCH_SIZE. This value should not be too small, otherwise the
@@ -137,8 +179,7 @@
%% effectively amortised (switching the direction of queue access
%% defeats amortisation), nor should it be too big, otherwise
%% converting a batch stalls the queue for too long. Therefore, it
-%% must be just right. ram_index_count is used here and is the number
-%% of betas.
+%% must be just right.
%%
%% The conversion from alphas to betas is also chunked, but only to
%% ensure no more than ?IO_BATCH_SIZE alphas are converted to betas at
@@ -248,7 +289,6 @@
ram_msg_count,
ram_msg_count_prev,
ram_ack_count_prev,
- ram_index_count,
out_counter,
in_counter,
rates,
@@ -280,8 +320,6 @@
end_seq_id %% end_seq_id is exclusive
}).
--record(merge_funs, {new, join, out, in, publish}).
-
%% When we discover, on publish, that we should write some indices to
%% disk for some betas, the IO_BATCH_SIZE sets the number of betas
%% that we must be due to write indices for before we do any work at
@@ -291,6 +329,7 @@
-define(IO_BATCH_SIZE, 64).
-define(PERSISTENT_MSG_STORE, msg_store_persistent).
-define(TRANSIENT_MSG_STORE, msg_store_transient).
+-define(QUEUE, lqueue).
-include("rabbit.hrl").
@@ -315,11 +354,11 @@
end_seq_id :: non_neg_integer() }).
-type(state() :: #vqstate {
- q1 :: queue(),
- q2 :: bpqueue:bpqueue(),
+ q1 :: ?QUEUE:?QUEUE(),
+ q2 :: ?QUEUE:?QUEUE(),
delta :: delta(),
- q3 :: bpqueue:bpqueue(),
- q4 :: queue(),
+ q3 :: ?QUEUE:?QUEUE(),
+ q4 :: ?QUEUE:?QUEUE(),
next_seq_id :: seq_id(),
pending_ack :: gb_tree(),
ram_ack_index :: gb_tree(),
@@ -337,7 +376,6 @@
target_ram_count :: non_neg_integer() | 'infinity',
ram_msg_count :: non_neg_integer(),
ram_msg_count_prev :: non_neg_integer(),
- ram_index_count :: non_neg_integer(),
out_counter :: non_neg_integer(),
in_counter :: non_neg_integer(),
rates :: rates(),
@@ -475,23 +513,22 @@ purge(State = #vqstate { q4 = Q4,
%% we could simply wipe the qi instead of issuing delivers and
%% acks for all the messages.
{LensByStore, IndexState1} = remove_queue_entries(
- fun rabbit_misc:queue_fold/3, Q4,
+ fun ?QUEUE:foldl/3, Q4,
orddict:new(), IndexState, MSCState),
{LensByStore1, State1 = #vqstate { q1 = Q1,
index_state = IndexState2,
msg_store_clients = MSCState1 }} =
purge_betas_and_deltas(LensByStore,
- State #vqstate { q4 = queue:new(),
+ State #vqstate { q4 = ?QUEUE:new(),
index_state = IndexState1 }),
{LensByStore2, IndexState3} = remove_queue_entries(
- fun rabbit_misc:queue_fold/3, Q1,
+ fun ?QUEUE:foldl/3, Q1,
LensByStore1, IndexState2, MSCState1),
PCount1 = PCount - find_persistent_count(LensByStore2),
- {Len, a(State1 #vqstate { q1 = queue:new(),
+ {Len, a(State1 #vqstate { q1 = ?QUEUE:new(),
index_state = IndexState3,
len = 0,
ram_msg_count = 0,
- ram_index_count = 0,
persistent_count = PCount1 })}.
publish(Msg = #basic_message { is_persistent = IsPersistent, id = MsgId },
@@ -507,9 +544,9 @@ publish(Msg = #basic_message { is_persistent = IsPersistent, id = MsgId },
IsPersistent1 = IsDurable andalso IsPersistent,
MsgStatus = msg_status(IsPersistent1, SeqId, Msg, MsgProps),
{MsgStatus1, State1} = maybe_write_to_disk(false, false, MsgStatus, State),
- State2 = case bpqueue:is_empty(Q3) of
- false -> State1 #vqstate { q1 = queue:in(m(MsgStatus1), Q1) };
- true -> State1 #vqstate { q4 = queue:in(m(MsgStatus1), Q4) }
+ State2 = case ?QUEUE:is_empty(Q3) of
+ false -> State1 #vqstate { q1 = ?QUEUE:in(m(MsgStatus1), Q1) };
+ true -> State1 #vqstate { q4 = ?QUEUE:in(m(MsgStatus1), Q4) }
end,
PCount1 = PCount + one_if(IsPersistent1),
UC1 = gb_sets_maybe_insert(NeedsConfirming, MsgId, UC),
@@ -615,11 +652,11 @@ requeue(AckTags, MsgPropsFun, #vqstate { delta = Delta,
len = Len } = State) ->
{SeqIds, Q4a, MsgIds, State1} = queue_merge(lists:sort(AckTags), Q4, [],
beta_limit(Q3),
- alpha_funs(),
+ fun publish_alpha/2,
MsgPropsFun, State),
{SeqIds1, Q3a, MsgIds1, State2} = queue_merge(SeqIds, Q3, MsgIds,
delta_limit(Delta),
- beta_funs(),
+ fun publish_beta/2,
MsgPropsFun, State1),
{Delta1, MsgIds2, State3} = delta_merge(SeqIds1, Delta, MsgIds1,
MsgPropsFun, State2),
@@ -718,7 +755,6 @@ needs_timeout(State) ->
false -> case reduce_memory_use(
fun (_Quota, State1) -> {0, State1} end,
fun (_Quota, State1) -> State1 end,
- fun (State1) -> State1 end,
fun (_Quota, State1) -> {0, State1} end,
State) of
{true, _State} -> idle;
@@ -740,24 +776,22 @@ status(#vqstate {
ram_ack_index = RAI,
target_ram_count = TargetRamCount,
ram_msg_count = RamMsgCount,
- ram_index_count = RamIndexCount,
next_seq_id = NextSeqId,
persistent_count = PersistentCount,
rates = #rates { avg_egress = AvgEgressRate,
avg_ingress = AvgIngressRate },
ack_rates = #rates { avg_egress = AvgAckEgressRate,
avg_ingress = AvgAckIngressRate } }) ->
- [ {q1 , queue:len(Q1)},
- {q2 , bpqueue:len(Q2)},
+ [ {q1 , ?QUEUE:len(Q1)},
+ {q2 , ?QUEUE:len(Q2)},
{delta , Delta},
- {q3 , bpqueue:len(Q3)},
- {q4 , queue:len(Q4)},
+ {q3 , ?QUEUE:len(Q3)},
+ {q4 , ?QUEUE:len(Q4)},
{len , Len},
{pending_acks , gb_trees:size(PA)},
{target_ram_count , TargetRamCount},
{ram_msg_count , RamMsgCount},
{ram_ack_count , gb_trees:size(RAI)},
- {ram_index_count , RamIndexCount},
{next_seq_id , NextSeqId},
{persistent_count , PersistentCount},
{avg_ingress_rate , AvgIngressRate},
@@ -776,15 +810,14 @@ discard(_Msg, _ChPid, State) -> State.
%%----------------------------------------------------------------------------
a(State = #vqstate { q1 = Q1, q2 = Q2, delta = Delta, q3 = Q3, q4 = Q4,
- len = Len,
- persistent_count = PersistentCount,
- ram_msg_count = RamMsgCount,
- ram_index_count = RamIndexCount }) ->
- E1 = queue:is_empty(Q1),
- E2 = bpqueue:is_empty(Q2),
+ len = Len,
+ persistent_count = PersistentCount,
+ ram_msg_count = RamMsgCount }) ->
+ E1 = ?QUEUE:is_empty(Q1),
+ E2 = ?QUEUE:is_empty(Q2),
ED = Delta#delta.count == 0,
- E3 = bpqueue:is_empty(Q3),
- E4 = queue:is_empty(Q4),
+ E3 = ?QUEUE:is_empty(Q3),
+ E4 = ?QUEUE:is_empty(Q4),
LZ = Len == 0,
true = E1 or not E3,
@@ -795,7 +828,6 @@ a(State = #vqstate { q1 = Q1, q2 = Q2, delta = Delta, q3 = Q3, q4 = Q4,
true = Len >= 0,
true = PersistentCount >= 0,
true = RamMsgCount >= 0,
- true = RamIndexCount >= 0,
State.
@@ -891,54 +923,46 @@ betas_from_index_entries(List, TransientThreshold, PA, IndexState) ->
cons_if(not IsDelivered, SeqId, Delivers1),
[SeqId | Acks1]};
false -> case gb_trees:is_defined(SeqId, PA) of
- false -> {[m(#msg_status {
- seq_id = SeqId,
- msg_id = MsgId,
- msg = undefined,
- is_persistent = IsPersistent,
- is_delivered = IsDelivered,
- msg_on_disk = true,
- index_on_disk = true,
- msg_props = MsgProps
- }) | Filtered1],
- Delivers1,
- Acks1};
- true -> Acc
+ false ->
+ {?QUEUE:in_r(
+ m(#msg_status {
+ seq_id = SeqId,
+ msg_id = MsgId,
+ msg = undefined,
+ is_persistent = IsPersistent,
+ is_delivered = IsDelivered,
+ msg_on_disk = true,
+ index_on_disk = true,
+ msg_props = MsgProps
+ }), Filtered1),
+ Delivers1, Acks1};
+ true ->
+ Acc
end
end
- end, {[], [], []}, List),
- {bpqueue:from_list([{true, Filtered}]),
- rabbit_queue_index:ack(Acks,
- rabbit_queue_index:deliver(Delivers, IndexState))}.
+ end, {?QUEUE:new(), [], []}, List),
+ {Filtered, rabbit_queue_index:ack(
+ Acks, rabbit_queue_index:deliver(Delivers, IndexState))}.
-%% the first arg is the older delta
-combine_deltas(?BLANK_DELTA_PATTERN(X), ?BLANK_DELTA_PATTERN(Y)) ->
- ?BLANK_DELTA;
-combine_deltas(?BLANK_DELTA_PATTERN(X), #delta { start_seq_id = Start,
- count = Count,
- end_seq_id = End } = B) ->
- true = Start + Count =< End, %% ASSERTION
- B;
-combine_deltas(#delta { start_seq_id = Start,
- count = Count,
- end_seq_id = End } = A, ?BLANK_DELTA_PATTERN(Y)) ->
- true = Start + Count =< End, %% ASSERTION
- A;
-combine_deltas(#delta { start_seq_id = StartLow,
- count = CountLow,
- end_seq_id = EndLow },
- #delta { start_seq_id = StartHigh,
- count = CountHigh,
- end_seq_id = EndHigh }) ->
- Count = CountLow + CountHigh,
- true = (StartLow =< StartHigh) %% ASSERTIONS
- andalso ((StartLow + CountLow) =< EndLow)
- andalso ((StartHigh + CountHigh) =< EndHigh)
- andalso ((StartLow + Count) =< EndHigh),
- #delta { start_seq_id = StartLow, count = Count, end_seq_id = EndHigh }.
-
-beta_fold(Fun, Init, Q) ->
- bpqueue:foldr(fun (_Prefix, Value, Acc) -> Fun(Value, Acc) end, Init, Q).
+expand_delta(SeqId, ?BLANK_DELTA_PATTERN(X)) ->
+ #delta { start_seq_id = SeqId, count = 1, end_seq_id = SeqId + 1 };
+expand_delta(SeqId, #delta { start_seq_id = StartSeqId,
+ count = Count,
+ end_seq_id = EndSeqId } = Delta)
+ when SeqId < StartSeqId ->
+ true = StartSeqId + Count =< EndSeqId, %% ASSERTION
+ Delta #delta { start_seq_id = SeqId, count = Count + 1 };
+expand_delta(SeqId, #delta { start_seq_id = StartSeqId,
+ count = Count,
+ end_seq_id = EndSeqId } = Delta)
+ when SeqId >= EndSeqId ->
+ true = StartSeqId + Count =< EndSeqId, %% ASSERTION
+ Delta #delta { count = Count + 1, end_seq_id = SeqId + 1 };
+expand_delta(_SeqId, #delta { start_seq_id = StartSeqId,
+ count = Count,
+ end_seq_id = EndSeqId } = Delta) ->
+ true = StartSeqId + Count + 1 =< EndSeqId, %% ASSERTION
+ Delta #delta { count = Count + 1 }.
update_rate(Now, Then, Count, {OThen, OCount}) ->
%% avg over the current period and the previous
@@ -961,11 +985,11 @@ init(IsDurable, IndexState, DeltaCount, Terms, AsyncCallback,
end,
Now = now(),
State = #vqstate {
- q1 = queue:new(),
- q2 = bpqueue:new(),
+ q1 = ?QUEUE:new(),
+ q2 = ?QUEUE:new(),
delta = Delta,
- q3 = bpqueue:new(),
- q4 = queue:new(),
+ q3 = ?QUEUE:new(),
+ q4 = ?QUEUE:new(),
next_seq_id = NextSeqId,
pending_ack = gb_trees:empty(),
ram_ack_index = gb_trees:empty(),
@@ -983,7 +1007,6 @@ init(IsDurable, IndexState, DeltaCount, Terms, AsyncCallback,
ram_msg_count = 0,
ram_msg_count_prev = 0,
ram_ack_count_prev = 0,
- ram_index_count = 0,
out_counter = 0,
in_counter = 0,
rates = blank_rate(Now, DeltaCount1),
@@ -1003,21 +1026,19 @@ blank_rate(Timestamp, IngressLength) ->
avg_ingress = 0.0,
timestamp = Timestamp }.
-in_r(MsgStatus = #msg_status { msg = undefined, index_on_disk = IndexOnDisk },
- State = #vqstate { q3 = Q3, q4 = Q4, ram_index_count = RamIndexCount }) ->
- case queue:is_empty(Q4) of
- true -> State #vqstate {
- q3 = bpqueue:in_r(IndexOnDisk, MsgStatus, Q3),
- ram_index_count = RamIndexCount + one_if(not IndexOnDisk) };
+in_r(MsgStatus = #msg_status { msg = undefined },
+ State = #vqstate { q3 = Q3, q4 = Q4 }) ->
+ case ?QUEUE:is_empty(Q4) of
+ true -> State #vqstate { q3 = ?QUEUE:in_r(MsgStatus, Q3) };
false -> {MsgStatus1, State1 = #vqstate { q4 = Q4a }} =
read_msg(MsgStatus, State),
- State1 #vqstate { q4 = queue:in_r(MsgStatus1, Q4a) }
+ State1 #vqstate { q4 = ?QUEUE:in_r(MsgStatus1, Q4a) }
end;
in_r(MsgStatus, State = #vqstate { q4 = Q4 }) ->
- State #vqstate { q4 = queue:in_r(MsgStatus, Q4) }.
+ State #vqstate { q4 = ?QUEUE:in_r(MsgStatus, Q4) }.
queue_out(State = #vqstate { q4 = Q4 }) ->
- case queue:out(Q4) of
+ case ?QUEUE:out(Q4) of
{empty, _Q4} ->
case fetch_from_q3(State) of
{empty, _State1} = Result -> Result;
@@ -1095,15 +1116,15 @@ purge_betas_and_deltas(LensByStore,
State = #vqstate { q3 = Q3,
index_state = IndexState,
msg_store_clients = MSCState }) ->
- case bpqueue:is_empty(Q3) of
+ case ?QUEUE:is_empty(Q3) of
true -> {LensByStore, State};
false -> {LensByStore1, IndexState1} =
- remove_queue_entries(fun beta_fold/3, Q3,
+ remove_queue_entries(fun ?QUEUE:foldl/3, Q3,
LensByStore, IndexState, MSCState),
purge_betas_and_deltas(LensByStore1,
maybe_deltas_to_betas(
State #vqstate {
- q3 = bpqueue:new(),
+ q3 = ?QUEUE:new(),
index_state = IndexState1 }))
end.
@@ -1321,99 +1342,54 @@ msg_indices_written_to_disk(Callback, MsgIdSet) ->
%% Internal plumbing for requeue
%%----------------------------------------------------------------------------
-alpha_funs() ->
- #merge_funs {
- new = fun queue:new/0,
- join = fun queue:join/2,
- out = fun queue:out/1,
- in = fun queue:in/2,
- publish = fun (#msg_status { msg = undefined } = MsgStatus, State) ->
- read_msg(MsgStatus, State);
- (MsgStatus, #vqstate {
- ram_msg_count = RamMsgCount } = State) ->
- {MsgStatus, State #vqstate {
- ram_msg_count = RamMsgCount + 1 }}
- end}.
-
-beta_funs() ->
- #merge_funs {
- new = fun bpqueue:new/0,
- join = fun bpqueue:join/2,
- out = fun (Q) ->
- case bpqueue:out(Q) of
- {{value, _IndexOnDisk, MsgStatus}, Q1} ->
- {{value, MsgStatus}, Q1};
- {empty, _Q1} = X ->
- X
- end
- end,
- in = fun (#msg_status { index_on_disk = IOD } = MsgStatus, Q) ->
- bpqueue:in(IOD, MsgStatus, Q)
- end,
- publish = fun (#msg_status { msg_on_disk = MsgOnDisk } = MsgStatus,
- State) ->
- {#msg_status { index_on_disk = IndexOnDisk,
- msg = Msg} = MsgStatus1,
- #vqstate { ram_index_count = RamIndexCount,
- ram_msg_count = RamMsgCount } =
- State1} =
- maybe_write_to_disk(not MsgOnDisk, false,
- MsgStatus, State),
- {MsgStatus1, State1 #vqstate {
- ram_msg_count = RamMsgCount +
- one_if(Msg =/= undefined),
- ram_index_count = RamIndexCount +
- one_if(not IndexOnDisk) }}
- end}.
+publish_alpha(#msg_status { msg = undefined } = MsgStatus, State) ->
+ read_msg(MsgStatus, State);
+publish_alpha(MsgStatus, #vqstate {ram_msg_count = RamMsgCount } = State) ->
+ {MsgStatus, State #vqstate { ram_msg_count = RamMsgCount + 1 }}.
+
+publish_beta(MsgStatus, State) ->
+ {#msg_status { msg = Msg} = MsgStatus1,
+ #vqstate { ram_msg_count = RamMsgCount } = State1} =
+ maybe_write_to_disk(true, false, MsgStatus, State),
+ {MsgStatus1, State1 #vqstate {
+ ram_msg_count = RamMsgCount + one_if(Msg =/= undefined) }}.
%% Rebuild queue, inserting sequence ids to maintain ordering
-queue_merge(SeqIds, Q, MsgIds, Limit, #merge_funs { new = QNew } = Funs,
- MsgPropsFun, State) ->
- queue_merge(SeqIds, Q, QNew(), MsgIds, Limit, Funs, MsgPropsFun, State).
+queue_merge(SeqIds, Q, MsgIds, Limit, PubFun, MsgPropsFun, State) ->
+ queue_merge(SeqIds, Q, ?QUEUE:new(), MsgIds,
+ Limit, PubFun, MsgPropsFun, State).
-queue_merge([SeqId | Rest] = SeqIds, Q, Front, MsgIds, Limit,
- #merge_funs { out = QOut, in = QIn, publish = QPublish } = Funs,
- MsgPropsFun, State)
+queue_merge([SeqId | Rest] = SeqIds, Q, Front, MsgIds,
+ Limit, PubFun, MsgPropsFun, State)
when Limit == undefined orelse SeqId < Limit ->
- case QOut(Q) of
+ case ?QUEUE:out(Q) of
{{value, #msg_status { seq_id = SeqIdQ } = MsgStatus}, Q1}
when SeqIdQ < SeqId ->
%% enqueue from the remaining queue
- queue_merge(SeqIds, Q1, QIn(MsgStatus, Front), MsgIds,
- Limit, Funs, MsgPropsFun, State);
+ queue_merge(SeqIds, Q1, ?QUEUE:in(MsgStatus, Front), MsgIds,
+ Limit, PubFun, MsgPropsFun, State);
{_, _Q1} ->
%% enqueue from the remaining list of sequence ids
{MsgStatus, State1} = msg_from_pending_ack(SeqId, MsgPropsFun,
State),
{#msg_status { msg_id = MsgId } = MsgStatus1, State2} =
- QPublish(MsgStatus, State1),
- queue_merge(Rest, Q, QIn(MsgStatus1, Front), [MsgId | MsgIds],
- Limit, Funs, MsgPropsFun, State2)
+ PubFun(MsgStatus, State1),
+ queue_merge(Rest, Q, ?QUEUE:in(MsgStatus1, Front), [MsgId | MsgIds],
+ Limit, PubFun, MsgPropsFun, State2)
end;
-queue_merge(SeqIds, Q, Front, MsgIds, _Limit, #merge_funs { join = QJoin },
- _MsgPropsFun, State) ->
- {SeqIds, QJoin(Front, Q), MsgIds, State}.
+queue_merge(SeqIds, Q, Front, MsgIds,
+ _Limit, _PubFun, _MsgPropsFun, State) ->
+ {SeqIds, ?QUEUE:join(Front, Q), MsgIds, State}.
delta_merge([], Delta, MsgIds, _MsgPropsFun, State) ->
{Delta, MsgIds, State};
-delta_merge(SeqIds, #delta { start_seq_id = StartSeqId,
- count = Count,
- end_seq_id = EndSeqId} = Delta,
- MsgIds, MsgPropsFun, State) ->
+delta_merge(SeqIds, Delta, MsgIds, MsgPropsFun, State) ->
lists:foldl(fun (SeqId, {Delta0, MsgIds0, State0}) ->
- {#msg_status { msg_id = MsgId,
- index_on_disk = IndexOnDisk,
- msg_on_disk = MsgOnDisk} = MsgStatus,
- State1} =
+ {#msg_status { msg_id = MsgId } = MsgStatus, State1} =
msg_from_pending_ack(SeqId, MsgPropsFun, State0),
{_MsgStatus, State2} =
- maybe_write_to_disk(not MsgOnDisk, not IndexOnDisk,
- MsgStatus, State1),
- {Delta0 #delta {
- start_seq_id = lists:min([SeqId, StartSeqId]),
- count = Count + 1,
- end_seq_id = lists:max([SeqId + 1, EndSeqId]) },
- [MsgId | MsgIds0], State2}
+ maybe_write_to_disk(true, true, MsgStatus, State1),
+ {expand_delta(SeqId, Delta0), [MsgId | MsgIds0], State2}
end, {Delta, MsgIds, State}, SeqIds).
%% Mostly opposite of record_pending_ack/2
@@ -1424,13 +1400,13 @@ msg_from_pending_ack(SeqId, MsgPropsFun, State) ->
msg_props = (MsgPropsFun(MsgProps)) #message_properties {
needs_confirming = false } }, State1}.
-beta_limit(BPQ) ->
- case bpqueue:out(BPQ) of
- {{value, _Prefix, #msg_status { seq_id = SeqId }}, _BPQ} -> SeqId;
- {empty, _BPQ} -> undefined
+beta_limit(Q) ->
+ case ?QUEUE:peek(Q) of
+ {value, #msg_status { seq_id = SeqId }} -> SeqId;
+ empty -> undefined
end.
-delta_limit(?BLANK_DELTA_PATTERN(_X)) -> undefined;
+delta_limit(?BLANK_DELTA_PATTERN(_X)) -> undefined;
delta_limit(#delta { start_seq_id = StartSeqId }) -> StartSeqId.
%%----------------------------------------------------------------------------
@@ -1456,10 +1432,10 @@ delta_limit(#delta { start_seq_id = StartSeqId }) -> StartSeqId.
%% one segment's worth of messages in q3 - and thus would risk
%% perpetually reporting the need for a conversion when no such
%% conversion is needed. That in turn could cause an infinite loop.
-reduce_memory_use(_AlphaBetaFun, _BetaGammaFun, _BetaDeltaFun, _AckFun,
+reduce_memory_use(_AlphaBetaFun, _BetaDeltaFun, _AckFun,
State = #vqstate {target_ram_count = infinity}) ->
{false, State};
-reduce_memory_use(AlphaBetaFun, BetaGammaFun, BetaDeltaFun, AckFun,
+reduce_memory_use(AlphaBetaFun, BetaDeltaFun, AckFun,
State = #vqstate {
ram_ack_index = RamAckIndex,
ram_msg_count = RamMsgCount,
@@ -1470,7 +1446,7 @@ reduce_memory_use(AlphaBetaFun, BetaGammaFun, BetaDeltaFun, AckFun,
avg_egress = AvgAckEgress }
}) ->
- {Reduce, State1} =
+ {Reduce, State1 = #vqstate { q2 = Q2, q3 = Q3 }} =
case chunk_size(RamMsgCount + gb_trees:size(RamAckIndex),
TargetRamCount) of
0 -> {false, State};
@@ -1491,13 +1467,10 @@ reduce_memory_use(AlphaBetaFun, BetaGammaFun, BetaDeltaFun, AckFun,
{true, State2}
end,
- case State1 #vqstate.target_ram_count of
- 0 -> {Reduce, BetaDeltaFun(State1)};
- _ -> case chunk_size(State1 #vqstate.ram_index_count,
- permitted_ram_index_count(State1)) of
- ?IO_BATCH_SIZE = S2 -> {true, BetaGammaFun(S2, State1)};
- _ -> {Reduce, State1}
- end
+ case chunk_size(?QUEUE:len(Q2) + ?QUEUE:len(Q3),
+ permitted_beta_count(State1)) of
+ ?IO_BATCH_SIZE = S2 -> {true, BetaDeltaFun(S2, State1)};
+ _ -> {Reduce, State1}
end.
limit_ram_acks(0, State) ->
@@ -1519,54 +1492,22 @@ limit_ram_acks(Quota, State = #vqstate { pending_ack = PA,
ram_ack_index = RAI1 })
end.
-
reduce_memory_use(State) ->
{_, State1} = reduce_memory_use(fun push_alphas_to_betas/2,
- fun limit_ram_index/2,
- fun push_betas_to_deltas/1,
+ fun push_betas_to_deltas/2,
fun limit_ram_acks/2,
State),
State1.
-limit_ram_index(Quota, State = #vqstate { q2 = Q2, q3 = Q3,
- index_state = IndexState,
- ram_index_count = RamIndexCount }) ->
- {Q2a, {Quota1, IndexState1}} = limit_ram_index(
- fun bpqueue:map_fold_filter_r/4,
- Q2, {Quota, IndexState}),
- %% TODO: we shouldn't be writing index entries for messages that
- %% can never end up in delta due them residing in the only segment
- %% held by q3.
- {Q3a, {Quota2, IndexState2}} = limit_ram_index(
- fun bpqueue:map_fold_filter_r/4,
- Q3, {Quota1, IndexState1}),
- State #vqstate { q2 = Q2a, q3 = Q3a,
- index_state = IndexState2,
- ram_index_count = RamIndexCount - (Quota - Quota2) }.
-
-limit_ram_index(_MapFoldFilterFun, Q, {0, IndexState}) ->
- {Q, {0, IndexState}};
-limit_ram_index(MapFoldFilterFun, Q, {Quota, IndexState}) ->
- MapFoldFilterFun(
- fun erlang:'not'/1,
- fun (MsgStatus, {0, _IndexStateN}) ->
- false = MsgStatus #msg_status.index_on_disk, %% ASSERTION
- stop;
- (MsgStatus, {N, IndexStateN}) when N > 0 ->
- false = MsgStatus #msg_status.index_on_disk, %% ASSERTION
- {MsgStatus1, IndexStateN1} =
- maybe_write_index_to_disk(true, MsgStatus, IndexStateN),
- {true, m(MsgStatus1), {N-1, IndexStateN1}}
- end, {Quota, IndexState}, Q).
-
-permitted_ram_index_count(#vqstate { len = 0 }) ->
+permitted_beta_count(#vqstate { len = 0 }) ->
infinity;
-permitted_ram_index_count(#vqstate { len = Len,
- q2 = Q2,
- q3 = Q3,
- delta = #delta { count = DeltaCount } }) ->
- BetaLen = bpqueue:len(Q2) + bpqueue:len(Q3),
- BetaLen - trunc(BetaLen * BetaLen / (Len - DeltaCount)).
+permitted_beta_count(#vqstate { target_ram_count = 0 }) ->
+ rabbit_queue_index:next_segment_boundary(0);
+permitted_beta_count(#vqstate { target_ram_count = TargetRamCount,
+ len = Len }) ->
+ BetaDelta = lists:max([0, Len - TargetRamCount]),
+ lists:max([BetaDelta - ((BetaDelta * BetaDelta) div Len),
+ rabbit_queue_index:next_segment_boundary(0)]).
chunk_size(Current, Permitted)
when Permitted =:= infinity orelse Permitted >= Current ->
@@ -1574,41 +1515,35 @@ chunk_size(Current, Permitted)
chunk_size(Current, Permitted) ->
lists:min([Current - Permitted, ?IO_BATCH_SIZE]).
-fetch_from_q3(State = #vqstate {
- q1 = Q1,
- q2 = Q2,
- delta = #delta { count = DeltaCount },
- q3 = Q3,
- q4 = Q4,
- ram_index_count = RamIndexCount}) ->
- case bpqueue:out(Q3) of
+fetch_from_q3(State = #vqstate { q1 = Q1,
+ q2 = Q2,
+ delta = #delta { count = DeltaCount },
+ q3 = Q3,
+ q4 = Q4 }) ->
+ case ?QUEUE:out(Q3) of
{empty, _Q3} ->
{empty, State};
- {{value, IndexOnDisk, MsgStatus}, Q3a} ->
- RamIndexCount1 = RamIndexCount - one_if(not IndexOnDisk),
- true = RamIndexCount1 >= 0, %% ASSERTION
- State1 = State #vqstate { q3 = Q3a,
- ram_index_count = RamIndexCount1 },
- State2 =
- case {bpqueue:is_empty(Q3a), 0 == DeltaCount} of
- {true, true} ->
- %% q3 is now empty, it wasn't before; delta is
- %% still empty. So q2 must be empty, and we
- %% know q4 is empty otherwise we wouldn't be
- %% loading from q3. As such, we can just set
- %% q4 to Q1.
- true = bpqueue:is_empty(Q2), %% ASSERTION
- true = queue:is_empty(Q4), %% ASSERTION
- State1 #vqstate { q1 = queue:new(),
- q4 = Q1 };
- {true, false} ->
- maybe_deltas_to_betas(State1);
- {false, _} ->
- %% q3 still isn't empty, we've not touched
- %% delta, so the invariants between q1, q2,
- %% delta and q3 are maintained
- State1
- end,
+ {{value, MsgStatus}, Q3a} ->
+ State1 = State #vqstate { q3 = Q3a },
+ State2 = case {?QUEUE:is_empty(Q3a), 0 == DeltaCount} of
+ {true, true} ->
+ %% q3 is now empty, it wasn't before;
+ %% delta is still empty. So q2 must be
+ %% empty, and we know q4 is empty
+ %% otherwise we wouldn't be loading from
+ %% q3. As such, we can just set q4 to Q1.
+ true = ?QUEUE:is_empty(Q2), %% ASSERTION
+ true = ?QUEUE:is_empty(Q4), %% ASSERTION
+ State1 #vqstate { q1 = ?QUEUE:new(), q4 = Q1 };
+ {true, false} ->
+ maybe_deltas_to_betas(State1);
+ {false, _} ->
+ %% q3 still isn't empty, we've not
+ %% touched delta, so the invariants
+ %% between q1, q2, delta and q3 are
+ %% maintained
+ State1
+ end,
{loaded, {MsgStatus, State2}}
end.
@@ -1632,7 +1567,7 @@ maybe_deltas_to_betas(State = #vqstate {
{Q3a, IndexState2} =
betas_from_index_entries(List, TransientThreshold, PA, IndexState1),
State1 = State #vqstate { index_state = IndexState2 },
- case bpqueue:len(Q3a) of
+ case ?QUEUE:len(Q3a) of
0 ->
%% we ignored every message in the segment due to it being
%% transient and below the threshold
@@ -1640,14 +1575,14 @@ maybe_deltas_to_betas(State = #vqstate {
State1 #vqstate {
delta = Delta #delta { start_seq_id = DeltaSeqId1 }});
Q3aLen ->
- Q3b = bpqueue:join(Q3, Q3a),
+ Q3b = ?QUEUE:join(Q3, Q3a),
case DeltaCount - Q3aLen of
0 ->
%% delta is now empty, but it wasn't before, so
%% can now join q2 onto q3
- State1 #vqstate { q2 = bpqueue:new(),
+ State1 #vqstate { q2 = ?QUEUE:new(),
delta = ?BLANK_DELTA,
- q3 = bpqueue:join(Q3b, Q2) };
+ q3 = ?QUEUE:join(Q3b, Q2) };
N when N > 0 ->
Delta1 = #delta { start_seq_id = DeltaSeqId1,
count = N,
@@ -1664,23 +1599,21 @@ push_alphas_to_betas(Quota, State) ->
maybe_push_q1_to_betas(Quota, State = #vqstate { q1 = Q1 }) ->
maybe_push_alphas_to_betas(
- fun queue:out/1,
- fun (MsgStatus = #msg_status { index_on_disk = IndexOnDisk },
- Q1a, State1 = #vqstate { q3 = Q3, delta = #delta { count = 0 } }) ->
+ fun ?QUEUE:out/1,
+ fun (MsgStatus, Q1a,
+ State1 = #vqstate { q3 = Q3, delta = #delta { count = 0 } }) ->
State1 #vqstate { q1 = Q1a,
- q3 = bpqueue:in(IndexOnDisk, MsgStatus, Q3) };
- (MsgStatus = #msg_status { index_on_disk = IndexOnDisk },
- Q1a, State1 = #vqstate { q2 = Q2 }) ->
+ q3 = ?QUEUE:in(MsgStatus, Q3) };
+ (MsgStatus, Q1a, State1 = #vqstate { q2 = Q2 }) ->
State1 #vqstate { q1 = Q1a,
- q2 = bpqueue:in(IndexOnDisk, MsgStatus, Q2) }
+ q2 = ?QUEUE:in(MsgStatus, Q2) }
end, Quota, Q1, State).
maybe_push_q4_to_betas(Quota, State = #vqstate { q4 = Q4 }) ->
maybe_push_alphas_to_betas(
- fun queue:out_r/1,
- fun (MsgStatus = #msg_status { index_on_disk = IndexOnDisk },
- Q4a, State1 = #vqstate { q3 = Q3 }) ->
- State1 #vqstate { q3 = bpqueue:in_r(IndexOnDisk, MsgStatus, Q3),
+ fun ?QUEUE:out_r/1,
+ fun (MsgStatus, Q4a, State1 = #vqstate { q3 = Q3 }) ->
+ State1 #vqstate { q3 = ?QUEUE:in_r(MsgStatus, Q3),
q4 = Q4a }
end, Quota, Q4, State).
@@ -1697,78 +1630,88 @@ maybe_push_alphas_to_betas(Generator, Consumer, Quota, Q, State) ->
{empty, _Q} ->
{Quota, State};
{{value, MsgStatus}, Qa} ->
- {MsgStatus1 = #msg_status { msg_on_disk = true,
- index_on_disk = IndexOnDisk },
- State1 = #vqstate { ram_msg_count = RamMsgCount,
- ram_index_count = RamIndexCount }} =
+ {MsgStatus1 = #msg_status { msg_on_disk = true },
+ State1 = #vqstate { ram_msg_count = RamMsgCount }} =
maybe_write_to_disk(true, false, MsgStatus, State),
MsgStatus2 = m(trim_msg_status(MsgStatus1)),
- RamIndexCount1 = RamIndexCount + one_if(not IndexOnDisk),
- State2 = State1 #vqstate { ram_msg_count = RamMsgCount - 1,
- ram_index_count = RamIndexCount1 },
+ State2 = State1 #vqstate { ram_msg_count = RamMsgCount - 1 },
maybe_push_alphas_to_betas(Generator, Consumer, Quota - 1, Qa,
Consumer(MsgStatus2, Qa, State2))
end.
-push_betas_to_deltas(State = #vqstate { q2 = Q2,
- delta = Delta,
- q3 = Q3,
- index_state = IndexState,
- ram_index_count = RamIndexCount }) ->
- {Delta2, Q2a, RamIndexCount2, IndexState2} =
- push_betas_to_deltas(fun (Q2MinSeqId) -> Q2MinSeqId end,
- fun bpqueue:out/1, Q2,
- RamIndexCount, IndexState),
- {Delta3, Q3a, RamIndexCount3, IndexState3} =
- push_betas_to_deltas(fun rabbit_queue_index:next_segment_boundary/1,
- fun bpqueue:out_r/1, Q3,
- RamIndexCount2, IndexState2),
- Delta4 = combine_deltas(Delta3, combine_deltas(Delta, Delta2)),
- State #vqstate { q2 = Q2a,
- delta = Delta4,
- q3 = Q3a,
- index_state = IndexState3,
- ram_index_count = RamIndexCount3 }.
-
-push_betas_to_deltas(LimitFun, Generator, Q, RamIndexCount, IndexState) ->
- case bpqueue:out(Q) of
- {empty, _Q} ->
- {?BLANK_DELTA, Q, RamIndexCount, IndexState};
- {{value, _IndexOnDisk1, #msg_status { seq_id = MinSeqId }}, _Qa} ->
- {{value, _IndexOnDisk2, #msg_status { seq_id = MaxSeqId }}, _Qb} =
- bpqueue:out_r(Q),
+push_betas_to_deltas(Quota, State = #vqstate { q2 = Q2,
+ delta = Delta,
+ q3 = Q3,
+ index_state = IndexState }) ->
+ PushState = {Quota, Delta, IndexState},
+ {Q3a, PushState1} = push_betas_to_deltas(
+ fun ?QUEUE:out_r/1,
+ fun rabbit_queue_index:next_segment_boundary/1,
+ Q3, PushState),
+ {Q2a, PushState2} = push_betas_to_deltas(
+ fun ?QUEUE:out/1,
+ fun (Q2MinSeqId) -> Q2MinSeqId end,
+ Q2, PushState1),
+ {_, Delta1, IndexState1} = PushState2,
+ State #vqstate { q2 = Q2a,
+ delta = Delta1,
+ q3 = Q3a,
+ index_state = IndexState1 }.
+
+push_betas_to_deltas(Generator, LimitFun, Q, PushState) ->
+ case ?QUEUE:is_empty(Q) of
+ true ->
+ {Q, PushState};
+ false ->
+ {value, #msg_status { seq_id = MinSeqId }} = ?QUEUE:peek(Q),
+ {value, #msg_status { seq_id = MaxSeqId }} = ?QUEUE:peek_r(Q),
Limit = LimitFun(MinSeqId),
case MaxSeqId < Limit of
- true -> {?BLANK_DELTA, Q, RamIndexCount, IndexState};
- false -> {Len, Qc, RamIndexCount1, IndexState1} =
- push_betas_to_deltas(Generator, Limit, Q, 0,
- RamIndexCount, IndexState),
- {#delta { start_seq_id = Limit,
- count = Len,
- end_seq_id = MaxSeqId + 1 },
- Qc, RamIndexCount1, IndexState1}
+ true -> {Q, PushState};
+ false -> {Q1, {Quota, Delta, IndexState}} =
+ push_betas_to_deltas1(
+ Generator, Limit, Q, PushState),
+ {Q2, Delta1} =
+ push_gammas_to_deltas(Generator, Limit, Q1, Delta),
+ {Q2, {Quota, Delta1, IndexState}}
end
end.
-push_betas_to_deltas(Generator, Limit, Q, Count, RamIndexCount, IndexState) ->
+push_betas_to_deltas1(_Generator, _Limit, Q,
+ {0, _Delta, _IndexState} = PushState) ->
+ {Q, PushState};
+push_betas_to_deltas1(Generator, Limit, Q,
+ {Quota, Delta, IndexState} = PushState) ->
case Generator(Q) of
{empty, _Q} ->
- {Count, Q, RamIndexCount, IndexState};
- {{value, _IndexOnDisk, #msg_status { seq_id = SeqId }}, _Qa}
+ {Q, PushState};
+ {{value, #msg_status { seq_id = SeqId }}, _Qa}
when SeqId < Limit ->
- {Count, Q, RamIndexCount, IndexState};
- {{value, IndexOnDisk, MsgStatus}, Qa} ->
- {RamIndexCount1, IndexState1} =
+ {Q, PushState};
+ {{value, MsgStatus = #msg_status { index_on_disk = IndexOnDisk,
+ seq_id = SeqId }}, Qa} ->
+ {Quota1, IndexState1} =
case IndexOnDisk of
- true -> {RamIndexCount, IndexState};
+ true -> {Quota, IndexState};
false -> {#msg_status { index_on_disk = true },
IndexState2} =
maybe_write_index_to_disk(true, MsgStatus,
IndexState),
- {RamIndexCount - 1, IndexState2}
+ {Quota - 1, IndexState2}
end,
- push_betas_to_deltas(
- Generator, Limit, Qa, Count + 1, RamIndexCount1, IndexState1)
+ Delta1 = expand_delta(SeqId, Delta),
+ push_betas_to_deltas1(Generator, Limit, Qa,
+ {Quota1, Delta1, IndexState1})
+ end.
+
+push_gammas_to_deltas(Generator, Limit, Q, Delta) ->
+ case Generator(Q) of
+ {{value, #msg_status { seq_id = SeqId, index_on_disk = true }}, Q1}
+ when SeqId >= Limit ->
+ push_gammas_to_deltas(Generator, Limit, Q1,
+ expand_delta(SeqId, Delta));
+ {_, _Q} ->
+ {Q, Delta}
end.
%%----------------------------------------------------------------------------