Statically configurable fixed width msg id size in msg_store. Associated changes in tests. All tests pass.

2 files changed, 49 insertions, 62 deletions

diff --git a/src/rabbit_msg_file.erl b/src/rabbit_msg_file.erl
index ea50cc6432..84dce90e8c 100644
--- a/src/rabbit_msg_file.erl
+++ b/src/rabbit_msg_file.erl
@@ -39,7 +39,10 @@
 -define(INTEGER_SIZE_BITS,       (8 * ?INTEGER_SIZE_BYTES)).
 -define(WRITE_OK_SIZE_BITS,      8).
 -define(WRITE_OK_MARKER,         255).
--define(FILE_PACKING_ADJUSTMENT, (1 + (2 * (?INTEGER_SIZE_BYTES)))).
+-define(FILE_PACKING_ADJUSTMENT, (1 + ?INTEGER_SIZE_BYTES)).
+-define(MSG_ID_SIZE_BYTES,       16).
+-define(MSG_ID_SIZE_BITS,        (8 * ?MSG_ID_SIZE_BYTES)).
+-define(SIZE_AND_MSG_ID_BYTES,   (?MSG_ID_SIZE_BYTES + ?INTEGER_SIZE_BYTES)).
 
 %%----------------------------------------------------------------------------
 
@@ -62,13 +65,13 @@
 
 %%----------------------------------------------------------------------------
 
-append(FileHdl, MsgId, MsgBody) when is_binary(MsgId) ->
+append(FileHdl, MsgId, MsgBody)
+  when is_binary(MsgId) andalso size(MsgId) =< ?MSG_ID_SIZE_BYTES ->
     MsgBodyBin  = term_to_binary(MsgBody),
-    [MsgIdSize, MsgBodyBinSize] = Sizes = [size(B) || B <- [MsgId, MsgBodyBin]],
-    Size = lists:sum(Sizes),
+    MsgBodyBinSize = size(MsgBodyBin),
+    Size = MsgBodyBinSize + ?MSG_ID_SIZE_BYTES,
     case file:write(FileHdl, <<Size:?INTEGER_SIZE_BITS,
-                               MsgIdSize:?INTEGER_SIZE_BITS,
-                               MsgId:MsgIdSize/binary,
+                               MsgId:?MSG_ID_SIZE_BYTES/binary,
                                MsgBodyBin:MsgBodyBinSize/binary,
                                ?WRITE_OK_MARKER:?WRITE_OK_SIZE_BITS>>) of
         ok -> {ok, Size + ?FILE_PACKING_ADJUSTMENT};
@@ -77,15 +80,12 @@ append(FileHdl, MsgId, MsgBody) when is_binary(MsgId) ->
 
 read(FileHdl, TotalSize) ->
     Size = TotalSize - ?FILE_PACKING_ADJUSTMENT,
-    SizeWriteOkBytes = Size + 1,
+    BodyBinSize = Size - ?MSG_ID_SIZE_BYTES,
     case file:read(FileHdl, TotalSize) of
         {ok, <<Size:?INTEGER_SIZE_BITS,
-               MsgIdSize:?INTEGER_SIZE_BITS,
-               Rest:SizeWriteOkBytes/binary>>} ->
-            BodyBinSize = Size - MsgIdSize,
-            <<MsgId:MsgIdSize/binary,
-              MsgBodyBin:BodyBinSize/binary,
-              ?WRITE_OK_MARKER:?WRITE_OK_SIZE_BITS>> = Rest,
+               MsgId:?MSG_ID_SIZE_BYTES/binary,
+               MsgBodyBin:BodyBinSize/binary,
+               ?WRITE_OK_MARKER:?WRITE_OK_SIZE_BITS>>} ->
             {ok, {MsgId, binary_to_term(MsgBodyBin)}};
         KO -> KO
     end.
@@ -105,50 +105,35 @@ scan(FileHdl, Offset, Acc) ->
     end.
 
 read_next(FileHdl, Offset) ->
-    TwoIntegers = 2 * ?INTEGER_SIZE_BYTES,
-    case file:read(FileHdl, TwoIntegers) of
-        {ok, <<Size:?INTEGER_SIZE_BITS, MsgIdSize:?INTEGER_SIZE_BITS>>} ->
-            if Size == 0 -> eof; %% Nothing we can do other than stop
-               MsgIdSize == 0 ->
-                    %% current message corrupted, try skipping past it
-                    ExpectedAbsPos = Offset + Size + ?FILE_PACKING_ADJUSTMENT,
-                    case file:position(FileHdl, {cur, Size + 1}) of
-                        {ok, ExpectedAbsPos} -> {corrupted, ExpectedAbsPos};
-                        {ok, _SomeOtherPos}  -> eof; %% seek failed, so give up
-                        KO                   -> KO
-                    end;
-               true -> %% all good, let's continue
-                    %% Here we take option 5 from
-                    %% http://www.erlang.org/cgi-bin/ezmlm-cgi?2:mss:1569
-                    %% in which we read the MsgId as a number, and
-                    %% then convert it back to a binary in order to
-                    %% work around bugs in Erlang's GC.
-                    MsgIdSizeBits = MsgIdSize * 8,
-                    case file:read(FileHdl, MsgIdSize) of
-                        {ok, <<MsgIdNum:MsgIdSizeBits>>} ->
-                            TotalSize = Size + ?FILE_PACKING_ADJUSTMENT,
-                            ExpectedAbsPos = Offset + TotalSize - 1,
-                            case file:position(
-                                   FileHdl, {cur, Size - MsgIdSize}) of
-                                {ok, ExpectedAbsPos} ->
-                                    NextOffset = ExpectedAbsPos + 1,
-                                    case file:read(FileHdl, 1) of
-                                        {ok, <<?WRITE_OK_MARKER:
-                                               ?WRITE_OK_SIZE_BITS>>} ->
-                                            <<MsgId:MsgIdSize/binary>> =
-                                                <<MsgIdNum:MsgIdSizeBits>>,
-                                            {ok, {MsgId,
-                                                  TotalSize, NextOffset}};
-                                        {ok, _SomeOtherData} ->
-                                            {corrupted, NextOffset};
-                                        KO -> KO
-                                    end;
-                                {ok, _SomeOtherPos} ->
-                                    %% seek failed, so give up
-                                    eof;
+    case file:read(FileHdl, ?SIZE_AND_MSG_ID_BYTES) of
+        %% Here we take option 5 from
+        %% http://www.erlang.org/cgi-bin/ezmlm-cgi?2:mss:1569 in which
+        %% we read the MsgId as a number, and then convert it back to
+        %% a binary in order to work around bugs in Erlang's GC.
+        {ok, <<Size:?INTEGER_SIZE_BITS, MsgIdNum:?MSG_ID_SIZE_BITS>>} ->
+            case Size of
+                0 -> eof; %% Nothing we can do other than stop
+                _ ->
+                    TotalSize = Size + ?FILE_PACKING_ADJUSTMENT,
+                    ExpectedAbsPos = Offset + TotalSize - 1,
+                    case file:position(
+                           FileHdl, {cur, Size - ?MSG_ID_SIZE_BYTES}) of
+                        {ok, ExpectedAbsPos} ->
+                            NextOffset = ExpectedAbsPos + 1,
+                            case file:read(FileHdl, 1) of
+                                {ok,
+                                 <<?WRITE_OK_MARKER: ?WRITE_OK_SIZE_BITS>>} ->
+                                    <<MsgId:?MSG_ID_SIZE_BYTES/binary>> =
+                                        <<MsgIdNum:?MSG_ID_SIZE_BITS>>,
+                                    {ok, {MsgId, TotalSize, NextOffset}};
+                                {ok, _SomeOtherData} ->
+                                    {corrupted, NextOffset};
                                 KO -> KO
                             end;
-                        Other -> Other
+                        {ok, _SomeOtherPos} ->
+                            %% seek failed, so give up
+                            eof;
+                        KO -> KO
                     end
             end;
         Other -> Other
diff --git a/src/rabbit_tests.erl b/src/rabbit_tests.erl
index 2034cd54b8..56dd3483f8 100644
--- a/src/rabbit_tests.erl
+++ b/src/rabbit_tests.erl
@@ -839,6 +839,9 @@ stop_msg_store() ->
         E -> E
     end.
 
+msg_id_bin(X) ->
+    erlang:md5(term_to_binary(X)).
+
 msg_store_contains(Atom, MsgIds) ->
     Atom = lists:foldl(
               fun (MsgId, Atom1) when Atom1 =:= Atom ->
@@ -872,7 +875,7 @@ test_msg_store() ->
     stop_msg_store(),
     ok = start_msg_store_empty(),
     Self = self(),
-    MsgIds = [term_to_binary(M) || M <- lists:seq(1,100)],
+    MsgIds = [msg_id_bin(M) || M <- lists:seq(1,100)],
     {MsgIds1stHalf, MsgIds2ndHalf} = lists:split(50, MsgIds),
     %% check we don't contain any of the msgs we're about to publish
     false = msg_store_contains(false, MsgIds),
@@ -967,22 +970,21 @@ test_msg_store() ->
     MsgIdsBig = lists:seq(1, BigCount),
     Payload = << 0:65536 >>,
     ok = lists:foldl(
-           fun (MsgId, ok) -> rabbit_msg_store:write(term_to_binary(MsgId),
-                                                     Payload) end,
-           ok, MsgIdsBig),
+           fun (MsgId, ok) ->
+                   rabbit_msg_store:write(msg_id_bin(MsgId), Payload)
+           end, ok, MsgIdsBig),
     %% .., then remove even numbers ascending, and odd numbers
     %% descending. This hits the GC.
     ok = lists:foldl(
            fun (MsgId, ok) ->
-                   rabbit_msg_store:remove([term_to_binary(
+                   rabbit_msg_store:remove([msg_id_bin(
                                               case MsgId rem 2 of
                                                   0 -> MsgId;
                                                   1 -> BigCount - MsgId
                                               end)])
            end, ok, MsgIdsBig),
     %% ensure empty
-    false =
-        msg_store_contains(false, lists:map(fun term_to_binary/1, MsgIdsBig)),
+    false = msg_store_contains(false, [msg_id_bin(M) || M <- MsgIdsBig]),
     %% restart empty
     ok = stop_msg_store(),
     ok = start_msg_store_empty(),