merge default into bug24455

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.
 
 %%----------------------------------------------------------------------------