diff options
| -rw-r--r-- | src/rabbit_disk_queue.erl | 747 |
1 files changed, 437 insertions, 310 deletions
diff --git a/src/rabbit_disk_queue.erl b/src/rabbit_disk_queue.erl index c56b14d5b8..190c06f00b 100644 --- a/src/rabbit_disk_queue.erl +++ b/src/rabbit_disk_queue.erl @@ -59,22 +59,23 @@ -define(SERVER, ?MODULE). --record(dqstate, {msg_location, - file_summary, - sequences, - current_file_num, - current_file_name, - current_file_handle, - current_offset, - file_size_limit, - read_file_handles, - read_file_handles_limit +-record(dqstate, {msg_location, % where are messages? + file_summary, %% what's in the files? + sequences, %% next read and write for each q + current_file_num, %% current file name as number + current_file_name, %% current file name + current_file_handle, %% current file handle + current_offset, %% current offset within current file + file_size_limit, %% how big can our files get? + read_file_handles, %% file handles for reading (LRU) + read_file_handles_limit %% how many file handles can we open? }). %% ---- PUBLIC API ---- start_link(FileSizeLimit, ReadFileHandlesLimit) -> - gen_server:start_link({local, ?SERVER}, ?MODULE, [FileSizeLimit, ReadFileHandlesLimit], []). + gen_server:start_link({local, ?SERVER}, ?MODULE, + [FileSizeLimit, ReadFileHandlesLimit], []). publish(Q, MsgId, Msg) when is_binary(Msg) -> gen_server:cast(?SERVER, {publish, Q, MsgId, Msg}). @@ -106,28 +107,35 @@ init([FileSizeLimit, ReadFileHandlesLimit]) -> process_flag(trap_exit, true), ok = filelib:ensure_dir(form_filename("nothing")), InitName = "0" ++ ?FILE_EXTENSION, - {ok, MsgLocation} = dets:open_file(?MSG_LOC_DETS_NAME, - [{file, form_filename(atom_to_list(?MSG_LOC_DETS_NAME) ++ ?FILE_EXTENSION_DETS)}, - {min_no_slots, 1024*1024}, - % man says this should be <= 32M. But it works... - {max_no_slots, 1024*1024*1024}, - {type, set} - ]), - State = #dqstate { msg_location = MsgLocation, - file_summary = ets:new(?FILE_SUMMARY_ETS_NAME, [set, private]), - sequences = ets:new(?SEQUENCE_ETS_NAME, [set, private]), - current_file_num = 0, - current_file_name = InitName, - current_file_handle = undefined, - current_offset = 0, - file_size_limit = FileSizeLimit, - read_file_handles = {dict:new(), gb_trees:empty()}, - read_file_handles_limit = ReadFileHandlesLimit - }, + {ok, MsgLocation} + = dets:open_file(?MSG_LOC_DETS_NAME, + [{file, form_filename(atom_to_list(?MSG_LOC_DETS_NAME) + ++ ?FILE_EXTENSION_DETS)}, + {min_no_slots, 1024*1024}, + %% man says this should be <= 32M. But it works... + {max_no_slots, 1024*1024*1024}, + {type, set} + ]), + State + = #dqstate { msg_location = MsgLocation, + file_summary = ets:new(?FILE_SUMMARY_ETS_NAME, + [set, private]), + sequences = ets:new(?SEQUENCE_ETS_NAME, + [set, private]), + current_file_num = 0, + current_file_name = InitName, + current_file_handle = undefined, + current_offset = 0, + file_size_limit = FileSizeLimit, + read_file_handles = {dict:new(), gb_trees:empty()}, + read_file_handles_limit = ReadFileHandlesLimit + }, {ok, State1 = #dqstate { current_file_name = CurrentName, - current_offset = Offset } } = load_from_disk(State), + current_offset = Offset } } + = load_from_disk(State), Path = form_filename(CurrentName), - {ok, FileHdl} = file:open(Path, [read, write, raw, binary, delayed_write]), %% read only needed so that we can seek + %% read is only needed so that we can seek + {ok, FileHdl} = file:open(Path, [read, write, raw, binary, delayed_write]), {ok, Offset} = file:position(FileHdl, {bof, Offset}), {ok, State1 # dqstate { current_file_handle = FileHdl }}. @@ -147,8 +155,9 @@ handle_call(clean_stop, _From, State) -> true = ets:delete(FileSummary), true = ets:delete(Sequences), lists:foreach(fun file:delete/1, filelib:wildcard(form_filename("*"))), - {stop, normal, ok, State1 # dqstate { current_file_handle = undefined, - read_file_handles = {dict:new(), gb_trees:empty()}}}. + {stop, normal, ok, + State1 # dqstate { current_file_handle = undefined, + read_file_handles = {dict:new(), gb_trees:empty()}}}. %% gen_server now calls terminate, which then calls shutdown handle_cast({publish, Q, MsgId, MsgBody}, State) -> @@ -174,9 +183,10 @@ shutdown(State = #dqstate { msg_location = MsgLocation, current_file_handle = FileHdl, read_file_handles = {ReadHdls, _ReadHdlsAge} }) -> - % deliberately ignoring return codes here + %% deliberately ignoring return codes here dets:close(MsgLocation), - file:delete(form_filename(atom_to_list(?MSG_LOC_DETS_NAME) ++ ?FILE_EXTENSION_DETS)), + file:delete(form_filename(atom_to_list(?MSG_LOC_DETS_NAME) + ++ ?FILE_EXTENSION_DETS)), if FileHdl =:= undefined -> ok; true -> file:sync(FileHdl), file:close(FileHdl) @@ -200,42 +210,56 @@ base_directory() -> %% ---- INTERNAL RAW FUNCTIONS ---- -internal_deliver(Q, State = #dqstate { msg_location = MsgLocation, - sequences = Sequences, - read_file_handles_limit = ReadFileHandlesLimit, - read_file_handles = {ReadHdls, ReadHdlsAge} - }) -> +internal_deliver(Q, State + = #dqstate { msg_location = MsgLocation, + sequences = Sequences, + read_file_handles_limit = ReadFileHandlesLimit, + read_file_handles = {ReadHdls, ReadHdlsAge} + }) -> case ets:lookup(Sequences, Q) of [] -> {ok, empty, State}; [{Q, ReadSeqId, WriteSeqId}] -> case mnesia:dirty_read(rabbit_disk_queue, {Q, ReadSeqId}) of [] -> {ok, empty, State}; - [Obj = #dq_msg_loc {is_delivered = Delivered, msg_id = MsgId}] -> - [{MsgId, _RefCount, File, Offset, TotalSize}] = dets:lookup(MsgLocation, MsgId), + [Obj + = #dq_msg_loc {is_delivered = Delivered, msg_id = MsgId}] -> + [{MsgId, _RefCount, File, Offset, TotalSize}] + = dets:lookup(MsgLocation, MsgId), {FileHdl, ReadHdls1, ReadHdlsAge1} = case dict:find(File, ReadHdls) of error -> - {ok, Hdl} = file:open(form_filename(File), [read, raw, binary, read_ahead]), + {ok, Hdl} = file:open(form_filename(File), + [read, raw, binary, + read_ahead]), Now = now(), case dict:size(ReadHdls) < ReadFileHandlesLimit of true -> - {Hdl, dict:store(File, {Hdl, Now}, ReadHdls), gb_trees:enter(Now, File, ReadHdlsAge)}; + {Hdl, + dict:store(File, {Hdl, Now}, ReadHdls), + gb_trees:enter(Now, File, ReadHdlsAge)}; _False -> - {_Then, OldFile, ReadHdlsAge2} = gb_trees:take_smallest(ReadHdlsAge), - {ok, {OldHdl, _Then}} = dict:find(OldFile, ReadHdls), + {_Then, OldFile, ReadHdlsAge2} + = gb_trees:take_smallest(ReadHdlsAge), + {ok, {OldHdl, _Then}} + = dict:find(OldFile, ReadHdls), ok = file:close(OldHdl), ReadHdls2 = dict:erase(OldFile, ReadHdls), - {Hdl, dict:store(File, {Hdl, Now}, ReadHdls2), gb_trees:enter(Now, File, ReadHdlsAge2)} + {Hdl, + dict:store(File, {Hdl, Now}, ReadHdls2), + gb_trees:enter(Now, File, ReadHdlsAge2)} end; {ok, {Hdl, Then}} -> Now = now(), {Hdl, dict:store(File, {Hdl, Now}, ReadHdls), - gb_trees:enter(Now, File, gb_trees:delete(Then, ReadHdlsAge))} + gb_trees:enter(Now, File, + gb_trees:delete(Then, ReadHdlsAge))} end, - % read the message - {ok, {MsgBody, BodySize}} = read_message_at_offset(FileHdl, Offset, TotalSize), + %% read the message + {ok, {MsgBody, BodySize}} + = read_message_at_offset(FileHdl, Offset, TotalSize), if Delivered -> ok; - true -> ok = mnesia:dirty_write(rabbit_disk_queue, Obj #dq_msg_loc {is_delivered = true}) + true -> ok = mnesia:dirty_write(rabbit_disk_queue, + Obj #dq_msg_loc {is_delivered = true}) end, true = ets:insert(Sequences, {Q, ReadSeqId + 1, WriteSeqId}), {ok, {MsgId, MsgBody, BodySize, Delivered, {MsgId, ReadSeqId}}, @@ -249,71 +273,83 @@ internal_ack(Q, MsgIds, State) -> %% Q is only needed if MnesiaDelete = true %% called from tx_cancel with MnesiaDelete = false %% called from ack with MnesiaDelete = true -remove_messages(Q, MsgSeqIds, MnesiaDelete, State = # dqstate { msg_location = MsgLocation, - file_summary = FileSummary, - current_file_name = CurName - }) -> +remove_messages(Q, MsgSeqIds, MnesiaDelete, + State = # dqstate { msg_location = MsgLocation, + file_summary = FileSummary, + current_file_name = CurName + }) -> Files - = lists:foldl(fun ({MsgId, SeqId}, Files2) -> - [{MsgId, RefCount, File, Offset, TotalSize}] - = dets:lookup(MsgLocation, MsgId), - Files3 = - if 1 =:= RefCount -> - ok = dets:delete(MsgLocation, MsgId), - [{File, ValidTotalSize, ContiguousTop, Left, Right}] - = ets:lookup(FileSummary, File), - ContiguousTop1 = lists:min([ContiguousTop, Offset]), - true = ets:insert(FileSummary, - {File, (ValidTotalSize - TotalSize - ?FILE_PACKING_ADJUSTMENT), - ContiguousTop1, Left, Right}), - if CurName =:= File -> Files2; - true -> sets:add_element(File, Files2) - end; - 1 < RefCount -> - ok = dets:insert(MsgLocation, {MsgId, RefCount - 1, File, Offset, TotalSize}), - Files2 - end, - if MnesiaDelete -> - ok = mnesia:dirty_delete(rabbit_disk_queue, {Q, SeqId}); - true -> ok - end, - Files3 - end, sets:new(), MsgSeqIds), + = lists:foldl( + fun ({MsgId, SeqId}, Files2) -> + [{MsgId, RefCount, File, Offset, TotalSize}] + = dets:lookup(MsgLocation, MsgId), + Files3 + = if 1 =:= RefCount -> + ok = dets:delete(MsgLocation, MsgId), + [{File, ValidTotalSize, ContiguousTop, Left, Right}] + = ets:lookup(FileSummary, File), + ContiguousTop1 = lists:min([ContiguousTop, Offset]), + true = ets:insert(FileSummary, + {File, (ValidTotalSize - TotalSize + - ?FILE_PACKING_ADJUSTMENT), + ContiguousTop1, Left, Right}), + if CurName =:= File -> Files2; + true -> sets:add_element(File, Files2) + end; + 1 < RefCount -> + ok = dets:insert(MsgLocation, {MsgId, RefCount - 1, + File, Offset, TotalSize}), + Files2 + end, + if MnesiaDelete -> + ok = mnesia:dirty_delete(rabbit_disk_queue, {Q, SeqId}); + true -> ok + end, + Files3 + end, sets:new(), MsgSeqIds), State2 = compact(Files, State), {ok, State2}. -internal_tx_publish(MsgId, MsgBody, State = #dqstate { msg_location = MsgLocation, - current_file_handle = CurHdl, - current_file_name = CurName, - current_offset = CurOffset, - file_summary = FileSummary - }) -> +internal_tx_publish(MsgId, MsgBody, + State = #dqstate { msg_location = MsgLocation, + current_file_handle = CurHdl, + current_file_name = CurName, + current_offset = CurOffset, + file_summary = FileSummary + }) -> case dets:lookup(MsgLocation, MsgId) of [] -> - % New message, lots to do + %% New message, lots to do {ok, TotalSize} = append_message(CurHdl, MsgId, MsgBody), - true = dets:insert_new(MsgLocation, {MsgId, 1, CurName, CurOffset, TotalSize}), + true = dets:insert_new(MsgLocation, {MsgId, 1, CurName, + CurOffset, TotalSize}), [{CurName, ValidTotalSize, ContiguousTop, Left, undefined}] = ets:lookup(FileSummary, CurName), - ValidTotalSize1 = ValidTotalSize + TotalSize + ?FILE_PACKING_ADJUSTMENT, + ValidTotalSize1 = ValidTotalSize + TotalSize + + ?FILE_PACKING_ADJUSTMENT, ContiguousTop1 = if CurOffset =:= ContiguousTop -> - ValidTotalSize1; % can't be any holes in this file + %% can't be any holes in this file + ValidTotalSize1; true -> ContiguousTop end, - true = ets:insert(FileSummary, {CurName, ValidTotalSize1, ContiguousTop1, Left, undefined}), - maybe_roll_to_new_file(CurOffset + TotalSize + ?FILE_PACKING_ADJUSTMENT, - State # dqstate {current_offset = CurOffset + TotalSize + ?FILE_PACKING_ADJUSTMENT}); + true = ets:insert(FileSummary, {CurName, ValidTotalSize1, + ContiguousTop1, Left, undefined}), + NextOffset = CurOffset + TotalSize + ?FILE_PACKING_ADJUSTMENT, + maybe_roll_to_new_file(NextOffset, + State # dqstate {current_offset = NextOffset}); [{MsgId, RefCount, File, Offset, TotalSize}] -> - % We already know about it, just update counter - ok = dets:insert(MsgLocation, {MsgId, RefCount + 1, File, Offset, TotalSize}), + %% We already know about it, just update counter + ok = dets:insert(MsgLocation, {MsgId, RefCount + 1, File, + Offset, TotalSize}), {ok, State} end. -internal_tx_commit(Q, MsgIds, State = #dqstate { msg_location = MsgLocation, - current_file_handle = CurHdl, - current_file_name = CurName, - sequences = Sequences - }) -> +internal_tx_commit(Q, MsgIds, + State = #dqstate { msg_location = MsgLocation, + current_file_handle = CurHdl, + current_file_name = CurName, + sequences = Sequences + }) -> {ReadSeqId, InitWriteSeqId} = case ets:lookup(Sequences, Q) of [] -> {0,0}; @@ -322,15 +358,18 @@ internal_tx_commit(Q, MsgIds, State = #dqstate { msg_location = MsgLocation, {atomic, {Sync, WriteSeqId}} = mnesia:transaction( fun() -> ok = mnesia:write_lock_table(rabbit_disk_queue), - lists:foldl(fun (MsgId, {Acc, NextWriteSeqId}) -> - [{MsgId, _RefCount, File, _Offset, _TotalSize}] = - dets:lookup(MsgLocation, MsgId), - ok = mnesia:write(rabbit_disk_queue, - #dq_msg_loc { queue_and_seq_id = {Q, NextWriteSeqId}, - msg_id = MsgId, is_delivered = false}, - write), - {Acc or (CurName =:= File), NextWriteSeqId + 1} - end, {false, InitWriteSeqId}, MsgIds) + lists:foldl( + fun (MsgId, {Acc, NextWriteSeqId}) -> + [{MsgId, _RefCount, File, _Offset, _TotalSize}] + = dets:lookup(MsgLocation, MsgId), + ok = mnesia:write(rabbit_disk_queue, + #dq_msg_loc { queue_and_seq_id + = {Q, NextWriteSeqId}, + msg_id = MsgId, + is_delivered = false}, + write), + {Acc or (CurName =:= File), NextWriteSeqId + 1} + end, {false, InitWriteSeqId}, MsgIds) end), true = ets:insert(Sequences, {Q, ReadSeqId, WriteSeqId}), if Sync -> ok = file:sync(CurHdl); @@ -339,41 +378,46 @@ internal_tx_commit(Q, MsgIds, State = #dqstate { msg_location = MsgLocation, {ok, State}. internal_publish(Q, MsgId, MsgBody, State) -> - {ok, State1 = #dqstate { sequences = Sequences }} = internal_tx_publish(MsgId, MsgBody, State), + {ok, State1 = #dqstate { sequences = Sequences }} + = internal_tx_publish(MsgId, MsgBody, State), WriteSeqId = case ets:lookup(Sequences, Q) of - [] -> % previously unseen queue + [] -> %% previously unseen queue true = ets:insert_new(Sequences, {Q, 0, 1}), 0; [{Q, ReadSeqId, WriteSeqId2}] -> - true = ets:insert(Sequences, {Q, ReadSeqId, WriteSeqId2 +1}), + true = ets:insert(Sequences, {Q, ReadSeqId, + WriteSeqId2 +1}), WriteSeqId2 end, - ok = mnesia:dirty_write(rabbit_disk_queue, #dq_msg_loc { queue_and_seq_id = {Q, WriteSeqId}, - msg_id = MsgId, - is_delivered = false}), + ok = mnesia:dirty_write(rabbit_disk_queue, + #dq_msg_loc { queue_and_seq_id = {Q, WriteSeqId}, + msg_id = MsgId, + is_delivered = false}), {ok, State1}. internal_tx_cancel(MsgIds, State) -> - % we don't need seq ids because we're not touching mnesia, because seqids were - % never assigned + %% we don't need seq ids because we're not touching mnesia, + %% because seqids were never assigned MsgSeqIds = lists:zip(MsgIds, lists:duplicate(length(MsgIds), undefined)), remove_messages(undefined, MsgSeqIds, false, State). %% ---- ROLLING OVER THE APPEND FILE ---- -maybe_roll_to_new_file(Offset, State = #dqstate { file_size_limit = FileSizeLimit, - current_file_name = CurName, - current_file_handle = CurHdl, - current_file_num = CurNum, - file_summary = FileSummary - } +maybe_roll_to_new_file(Offset, + State = #dqstate { file_size_limit = FileSizeLimit, + current_file_name = CurName, + current_file_handle = CurHdl, + current_file_num = CurNum, + file_summary = FileSummary + } ) when Offset >= FileSizeLimit -> ok = file:sync(CurHdl), ok = file:close(CurHdl), NextNum = CurNum + 1, NextName = integer_to_list(NextNum) ++ ?FILE_EXTENSION, - {ok, NextHdl} = file:open(form_filename(NextName), [write, raw, binary, delayed_write]), - true = ets:update_element(FileSummary, CurName, {5, NextName}), % 5 is Right + {ok, NextHdl} = file:open(form_filename(NextName), + [write, raw, binary, delayed_write]), + true = ets:update_element(FileSummary, CurName, {5, NextName}), %% 5 is Right true = ets:insert_new(FileSummary, {NextName, 0, 0, CurName, undefined}), {ok, State # dqstate { current_file_name = NextName, current_file_handle = NextHdl, @@ -386,55 +430,70 @@ maybe_roll_to_new_file(_, State) -> %% ---- GARBAGE COLLECTION / COMPACTION / AGGREGATION ---- compact(FilesSet, State) -> - % smallest number, hence eldest, hence left-most, first + %% smallest number, hence eldest, hence left-most, first Files = lists:sort(sets:to_list(FilesSet)), - % foldl reverses, so now youngest/right-most first - RemainingFiles = lists:foldl(fun (File, Acc) -> delete_empty_files(File, Acc, State) end, [], Files), + %% foldl reverses, so now youngest/right-most first + RemainingFiles = lists:foldl(fun (File, Acc) -> + delete_empty_files(File, Acc, State) + end, [], Files), lists:foldl(fun combineFile/2, State, lists:reverse(RemainingFiles)). combineFile(File, State = #dqstate { file_size_limit = FileSizeLimit, file_summary = FileSummary, current_file_name = CurName }) -> - % the file we're looking at may no longer exist as it may have been deleted - % within the current GC run + %% the file we're looking at may no longer exist as it may have been deleted + %% within the current GC run case ets:lookup(FileSummary, File) of [] -> State; [FileObj = {File, ValidData, _ContiguousTop, Left, Right}] -> - GoRight = fun() -> - case Right of - undefined -> State; - _ when not(CurName =:= Right) -> - [RightObj = {Right, RightValidData, _RightContiguousTop, File, RightRight}] - = ets:lookup(FileSummary, Right), - RightSumData = ValidData + RightValidData, - if FileSizeLimit >= RightSumData -> - % here, Right will be the source and so will be deleted, - % File will be the destination - State1 = combineFiles(RightObj, FileObj, State), - % this could fail if RightRight is undefined - ets:update_element(FileSummary, RightRight, {4, File}), % left is the 4th field - true = ets:insert(FileSummary, {File, RightSumData, RightSumData, Left, RightRight}), - true = ets:delete(FileSummary, Right), - State1; - true -> State - end; - _ -> State - end - end, + GoRight + = fun() -> + case Right of + undefined -> State; + _ when not(CurName =:= Right) -> + [RightObj = {Right, RightValidData, + _RightContiguousTop, File, RightRight}] + = ets:lookup(FileSummary, Right), + RightSumData = ValidData + RightValidData, + if FileSizeLimit >= RightSumData -> + %% here, Right will be the source and so will be deleted, + %% File will be the destination + State1 = combineFiles(RightObj, FileObj, + State), + %% this could fail if RightRight is undefined + %% left is the 4th field + ets:update_element(FileSummary, + RightRight, {4, File}), + true = ets:insert(FileSummary, {File, + RightSumData, + RightSumData, + Left, + RightRight}), + true = ets:delete(FileSummary, Right), + State1; + true -> State + end; + _ -> State + end + end, case Left of undefined -> GoRight(); - _ -> [LeftObj = {Left, LeftValidData, _LeftContiguousTop, LeftLeft, File}] + _ -> [LeftObj + = {Left, LeftValidData, _LeftContiguousTop, LeftLeft, File}] = ets:lookup(FileSummary, Left), LeftSumData = ValidData + LeftValidData, if FileSizeLimit >= LeftSumData -> - % here, File will be the source and so will be deleted, - % Left will be the destination + %% here, File will be the source and so will be deleted, + %% Left will be the destination State1 = combineFiles(FileObj, LeftObj, State), - % this could fail if Right is undefined - ets:update_element(FileSummary, Right, {4, Left}), % left is the 4th field - true = ets:insert(FileSummary, {Left, LeftSumData, LeftSumData, LeftLeft, Right}), + %% this could fail if Right is undefined + %% left is the 4th field + ets:update_element(FileSummary, Right, {4, Left}), + true = ets:insert(FileSummary, {Left, LeftSumData, + LeftSumData, + LeftLeft, Right}), true = ets:delete(FileSummary, File), State1; true -> @@ -443,113 +502,148 @@ combineFile(File, State = #dqstate { file_size_limit = FileSizeLimit, end end. -combineFiles({Source, SourceValid, _SourceContiguousTop, _SourceLeft, _SourceRight}, - {Destination, DestinationValid, DestinationContiguousTop, _DestinationLeft, _DestinationRight}, +combineFiles({Source, SourceValid, _SourceContiguousTop, + _SourceLeft, _SourceRight}, + {Destination, DestinationValid, DestinationContiguousTop, + _DestinationLeft, _DestinationRight}, State1) -> (State = #dqstate { msg_location = MsgLocation }) = closeFile(Source, closeFile(Destination, State1)), - {ok, SourceHdl} = file:open(form_filename(Source), [read, write, raw, binary, delayed_write, read_ahead]), - {ok, DestinationHdl} = file:open(form_filename(Destination), [read, write, raw, binary, delayed_write, read_ahead]), + {ok, SourceHdl} + = file:open(form_filename(Source), + [read, write, raw, binary, delayed_write, read_ahead]), + {ok, DestinationHdl} + = file:open(form_filename(Destination), + [read, write, raw, binary, delayed_write, read_ahead]), ExpectedSize = SourceValid + DestinationValid, - % if DestinationValid =:= DestinationContiguousTop then we don't need a tmp file - % if they're not equal, then we need to write out everything past the DestinationContiguousTop to a tmp file - % then truncate, copy back in, and then copy over from Source - % otherwise we just truncate straight away and copy over from Source + %% if DestinationValid =:= DestinationContiguousTop then we don't need a tmp file + %% if they're not equal, then we need to write out everything past the DestinationContiguousTop to a tmp file + %% then truncate, copy back in, and then copy over from Source + %% otherwise we just truncate straight away and copy over from Source if DestinationContiguousTop =:= DestinationValid -> - {ok, DestinationValid} = file:position(DestinationHdl, {bof, DestinationValid}), + {ok, DestinationValid} = file:position(DestinationHdl, + {bof, DestinationValid}), ok = file:truncate(DestinationHdl), - {ok, ExpectedSize} = file:position(DestinationHdl, {cur, SourceValid}), + {ok, ExpectedSize} = file:position(DestinationHdl, + {cur, SourceValid}), ok = file:truncate(DestinationHdl), - {ok, DestinationValid} = file:position(DestinationHdl, {bof, DestinationValid}); + {ok, DestinationValid} = file:position(DestinationHdl, + {bof, DestinationValid}); true -> Tmp = filename:rootname(Destination) ++ ?FILE_EXTENSION_TMP, - {ok, TmpHdl} = file:open(form_filename(Tmp), [read, write, raw, binary, delayed_write, read_ahead]), + {ok, TmpHdl} + = file:open(form_filename(Tmp), + [read, write, raw, binary, delayed_write, read_ahead]), Worklist - = lists:dropwhile(fun ({_, _, _, Offset, _}) when Offset /= DestinationContiguousTop -> - % it cannot be that Offset == DestinationContiguousTop - % because if it was then DestinationContiguousTop would have been - % extended by TotalSize - Offset < DestinationContiguousTop - % Given expected access patterns, I suspect that the list should be - % naturally sorted as we require, however, we need to enforce it anyway - end, lists:sort(fun ({_, _, _, OffA, _}, {_, _, _, OffB, _}) -> - OffA < OffB - end, - dets:match_object(MsgLocation, {'_', '_', Destination, '_', '_'}))), + = lists:dropwhile( + fun ({_, _, _, Offset, _}) + when Offset /= DestinationContiguousTop -> + %% it cannot be that Offset == DestinationContiguousTop + %% because if it was then DestinationContiguousTop would have been + %% extended by TotalSize + Offset < DestinationContiguousTop + %% Given expected access patterns, I suspect that the list should be + %% naturally sorted as we require, however, we need to enforce it anyway + end, lists:sort(fun ({_, _, _, OffA, _}, {_, _, _, OffB, _}) -> + OffA < OffB + end, + dets:match_object(MsgLocation, + {'_', '_', Destination, + '_', '_'}))), TmpSize = DestinationValid - DestinationContiguousTop, - {TmpSize, BlockStart1, BlockEnd1} = - lists:foldl(fun ({MsgId, RefCount, _Destination, Offset, TotalSize}, {CurOffset, BlockStart, BlockEnd}) -> - % CurOffset is in the TmpFile. - % Offset, BlockStart and BlockEnd are in the DestinationFile (which is currently the source!) - Size = TotalSize + ?FILE_PACKING_ADJUSTMENT, - % this message is going to end up back in Destination, at DestinationContiguousTop + CurOffset - FinalOffset = DestinationContiguousTop + CurOffset, - ok = dets:insert(MsgLocation, {MsgId, RefCount, Destination, FinalOffset, TotalSize}), - - NextOffset = CurOffset + Size, - if BlockStart =:= undefined -> - % base case, called only for the first list elem - {NextOffset, Offset, Offset + Size}; - Offset =:= BlockEnd -> - % extend the current block because the next msg follows straight on - {NextOffset, BlockStart, BlockEnd + Size}; - true -> - % found a gap, so actually do the work for the previous block - BSize = BlockEnd - BlockStart, - {ok, BlockStart} = file:position(DestinationHdl, {bof, BlockStart}), - {ok, BSize} = file:copy(DestinationHdl, TmpHdl, BSize), - {NextOffset, Offset, Offset + Size} - end - end, {0, undefined, undefined}, Worklist), - % do the last remaining block + {TmpSize, BlockStart1, BlockEnd1} + = lists:foldl( + fun ({MsgId, RefCount, _Destination, Offset, TotalSize}, + {CurOffset, BlockStart, BlockEnd}) -> + %% CurOffset is in the TmpFile. + %% Offset, BlockStart and BlockEnd are in the DestinationFile (which is currently the source!) + Size = TotalSize + ?FILE_PACKING_ADJUSTMENT, + %% this message is going to end up back in Destination, at DestinationContiguousTop + CurOffset + FinalOffset = DestinationContiguousTop + CurOffset, + ok = dets:insert(MsgLocation, {MsgId, RefCount, Destination, + FinalOffset, TotalSize}), + + NextOffset = CurOffset + Size, + if BlockStart =:= undefined -> + %% base case, called only for the first list elem + {NextOffset, Offset, Offset + Size}; + Offset =:= BlockEnd -> + %% extend the current block because the next msg follows straight on + {NextOffset, BlockStart, BlockEnd + Size}; + true -> + %% found a gap, so actually do the work for the previous block + BSize = BlockEnd - BlockStart, + {ok, BlockStart} + = file:position(DestinationHdl, + {bof, BlockStart}), + {ok, BSize} = file:copy(DestinationHdl, + TmpHdl, BSize), + {NextOffset, Offset, Offset + Size} + end + end, {0, undefined, undefined}, Worklist), + %% do the last remaining block BSize1 = BlockEnd1 - BlockStart1, {ok, BlockStart1} = file:position(DestinationHdl, {bof, BlockStart1}), {ok, BSize1} = file:copy(DestinationHdl, TmpHdl, BSize1), - % so now Tmp contains everything we need to salvage from Destination, - % and MsgLocation has been updated to reflect compaction of Destination - % so truncate Destination and copy from Tmp back to the end + %% so now Tmp contains everything we need to salvage from Destination, + %% and MsgLocation has been updated to reflect compaction of Destination + %% so truncate Destination and copy from Tmp back to the end {ok, 0} = file:position(TmpHdl, {bof, 0}), - {ok, DestinationContiguousTop} = file:position(DestinationHdl, {bof, DestinationContiguousTop}), + {ok, DestinationContiguousTop} + = file:position(DestinationHdl, + {bof, DestinationContiguousTop}), ok = file:truncate(DestinationHdl), - {ok, ExpectedSize} = file:position(DestinationHdl, {bof, ExpectedSize}), + {ok, ExpectedSize} + = file:position(DestinationHdl, + {bof, ExpectedSize}), ok = file:truncate(DestinationHdl), - {ok, DestinationContiguousTop} = file:position(DestinationHdl, {bof, DestinationContiguousTop}), + {ok, DestinationContiguousTop} + = file:position(DestinationHdl, + {bof, DestinationContiguousTop}), {ok, TmpSize} = file:copy(TmpHdl, DestinationHdl, TmpSize), - % position in DestinationHdl should now be DestinationValid + %% position in DestinationHdl should now be DestinationValid ok = file:sync(DestinationHdl), ok = file:close(TmpHdl), ok = file:delete(form_filename(Tmp)) end, - SourceWorkList = lists:sort(fun ({_, _, _, OffA, _}, {_, _, _, OffB, _}) -> - OffA < OffB - end, dets:match_object(MsgLocation, {'_', '_', Source, '_', '_'})), - {ExpectedSize, BlockStart2, BlockEnd2} = - lists:foldl(fun ({MsgId, RefCount, _Source, Offset, TotalSize}, {CurOffset, BlockStart, BlockEnd}) -> - % CurOffset is in the DestinationFile. - % Offset, BlockStart and BlockEnd are in the SourceFile - Size = TotalSize + ?FILE_PACKING_ADJUSTMENT, - % update MsgLocation to reflect change of file and offset - ok = dets:insert(MsgLocation, {MsgId, RefCount, Destination, CurOffset, TotalSize}), - NextOffset = CurOffset + Size, - if BlockStart =:= undefined -> - % base case, called only for the first list elem - {NextOffset, Offset, Offset + Size}; - Offset =:= BlockEnd -> - % extend the current block because the next msg follows straight on - {NextOffset, BlockStart, BlockEnd + Size}; - true -> - % found a gap, so actually do the work for the previous block - BSize = BlockEnd - BlockStart, - {ok, BlockStart} = file:position(SourceHdl, {bof, BlockStart}), - {ok, BSize} = file:copy(SourceHdl, DestinationHdl, BSize), - {NextOffset, Offset, Offset + Size} - end - end, {DestinationValid, undefined, undefined}, SourceWorkList), - % do the last remaining block + SourceWorkList + = lists:sort(fun ({_, _, _, OffA, _}, {_, _, _, OffB, _}) -> + OffA < OffB + end, dets:match_object(MsgLocation, {'_', '_', Source, + '_', '_'})), + {ExpectedSize, BlockStart2, BlockEnd2} + = lists:foldl(fun ({MsgId, RefCount, _Source, Offset, TotalSize}, + {CurOffset, BlockStart, BlockEnd}) -> + %% CurOffset is in the DestinationFile. + %% Offset, BlockStart and BlockEnd are in the SourceFile + Size = TotalSize + ?FILE_PACKING_ADJUSTMENT, + %% update MsgLocation to reflect change of file and offset + ok = dets:insert(MsgLocation, {MsgId, RefCount, Destination, + CurOffset, TotalSize}), + NextOffset = CurOffset + Size, + if BlockStart =:= undefined -> + %% base case, called only for the first list elem + {NextOffset, Offset, Offset + Size}; + Offset =:= BlockEnd -> + %% extend the current block because the next msg follows straight on + {NextOffset, BlockStart, BlockEnd + Size}; + true -> + %% found a gap, so actually do the work for the previous block + BSize = BlockEnd - BlockStart, + {ok, BlockStart} + = file:position(SourceHdl, + {bof, BlockStart}), + {ok, BSize} + = file:copy(SourceHdl, DestinationHdl, + BSize), + {NextOffset, Offset, Offset + Size} + end + end, {DestinationValid, undefined, undefined}, SourceWorkList), + %% do the last remaining block BSize2 = BlockEnd2 - BlockStart2, {ok, BlockStart2} = file:position(SourceHdl, {bof, BlockStart2}), {ok, BSize2} = file:copy(SourceHdl, DestinationHdl, BSize2), - % tidy up + %% tidy up ok = file:sync(DestinationHdl), ok = file:close(SourceHdl), ok = file:close(DestinationHdl), @@ -562,20 +656,27 @@ closeFile(File, State = #dqstate { read_file_handles = {ReadHdls, ReadHdlsAge} } State; {ok, {Hdl, Then}} -> ok = file:close(Hdl), - State #dqstate { read_file_handles = { dict:erase(File, ReadHdls), gb_trees:delete(Then, ReadHdlsAge) } } + State #dqstate { read_file_handles + = { dict:erase(File, ReadHdls), + gb_trees:delete(Then, ReadHdlsAge) } } end. delete_empty_files(File, Acc, #dqstate { file_summary = FileSummary }) -> - [{File, ValidData, _ContiguousTop, Left, Right}] = ets:lookup(FileSummary, File), + [{File, ValidData, _ContiguousTop, Left, Right}] + = ets:lookup(FileSummary, File), case ValidData of - % we should NEVER find the current file in here - hence right should always be a file, not undefined + %% we should NEVER find the current file in here + %% hence right should always be a file, not undefined 0 -> case {Left, Right} of {undefined, _} when not(is_atom(Right)) -> - % the eldest file is empty. YAY! - true = ets:update_element(FileSummary, Right, {4, undefined}); % left is the 4th field + %% the eldest file is empty. YAY! + %% left is the 4th field + true = ets:update_element(FileSummary, Right, {4, undefined}); {_, _} when not(is_atom(Right)) -> - true = ets:update_element(FileSummary, Right, {4, Left}), % left is the 4th field - true = ets:update_element(FileSummary, Left, {5, Right}) % right is the 5th field + %% left is the 4th field + true = ets:update_element(FileSummary, Right, {4, Left}), + %% right is the 5th field + true = ets:update_element(FileSummary, Left, {5, Right}) end, true = ets:delete(FileSummary, File), ok = file:delete(form_filename(File)), @@ -586,24 +687,27 @@ delete_empty_files(File, Acc, #dqstate { file_summary = FileSummary }) -> %% ---- DISK RECOVERY ---- load_from_disk(State) -> - % sorted so that smallest number is first. which also means eldest file (left-most) first + %% sorted so that smallest number is first. which also means eldest file (left-most) first {Files, TmpFiles} = get_disk_queue_files(), ok = recover_crashed_compactions(Files, TmpFiles), - % There should be no more tmp files now, so go ahead and load the whole lot - (State1 = #dqstate{ msg_location = MsgLocation }) = load_messages(undefined, Files, State), - % Finally, check there is nothing in mnesia which we haven't loaded + %% There should be no more tmp files now, so go ahead and load the whole lot + (State1 = #dqstate{ msg_location = MsgLocation }) + = load_messages(undefined, Files, State), + %% Finally, check there is nothing in mnesia which we haven't loaded {atomic, true} = mnesia:transaction( fun() -> ok = mnesia:read_lock_table(rabbit_disk_queue), mnesia:foldl(fun (#dq_msg_loc { msg_id = MsgId }, true) -> - true = 1 =:= length(dets:lookup(MsgLocation, MsgId)) end, + true = 1 + =:= length(dets:lookup(MsgLocation, MsgId)) + end, true, rabbit_disk_queue) end), State2 = extract_sequence_numbers(State1), {ok, State2}. extract_sequence_numbers(State = #dqstate { sequences = Sequences }) -> - % next-seqid-to-read is the lowest seqid which has is_delivered = false + %% next-seqid-to-read is the lowest seqid which has is_delivered = false {atomic, true} = mnesia:transaction( fun() -> ok = mnesia:read_lock_table(rabbit_disk_queue), @@ -612,7 +716,8 @@ extract_sequence_numbers(State = #dqstate { sequences = Sequences }) -> NextWrite = SeqId + 1, case ets:lookup(Sequences, Q) of [] -> - true = ets:insert_new(Sequences, {Q, SeqId, NextWrite}); + true = ets:insert_new(Sequences, + {Q, SeqId, NextWrite}); [Orig = {Q, Read, Write}] -> Repl = {Q, lists:min([Read, SeqId]), lists:max([Write, NextWrite])}, @@ -634,7 +739,8 @@ load_messages(Left, [], State = #dqstate { msg_location = MsgLocation }) -> Offset = case dets:match_object(MsgLocation, {'_', '_', Left, '_', '_'}) of [] -> 0; L -> [{_MsgId, _RefCount, Left, MaxOffset, TotalSize}|_] - = lists:sort(fun ({_, _, _, OffA, _}, {_, _, _, OffB, _}) -> + = lists:sort(fun ({_, _, _, OffA, _}, + {_, _, _, OffB, _}) -> OffB < OffA end, L), MaxOffset + TotalSize + ?FILE_PACKING_ADJUSTMENT @@ -645,24 +751,26 @@ load_messages(Left, [File|Files], State = #dqstate { msg_location = MsgLocation, file_summary = FileSummary }) -> - % [{MsgId, TotalSize, FileOffset}] + %% [{MsgId, TotalSize, FileOffset}] {ok, Messages} = scan_file_for_valid_messages(form_filename(File)), {ValidMessagesRev, ValidTotalSize} = lists:foldl( fun ({MsgId, TotalSize, Offset}, {VMAcc, VTSAcc}) -> - case length(mnesia:dirty_match_object(rabbit_disk_queue, - #dq_msg_loc { msg_id = MsgId, - queue_and_seq_id = '_', - is_delivered = '_'})) of + case length(mnesia:dirty_match_object + (rabbit_disk_queue, + #dq_msg_loc { msg_id = MsgId, + queue_and_seq_id = '_', + is_delivered = '_'})) of 0 -> {VMAcc, VTSAcc}; RefCount -> - true = dets:insert_new(MsgLocation, {MsgId, RefCount, File, Offset, TotalSize}), + true = dets:insert_new(MsgLocation, {MsgId, RefCount, File, + Offset, TotalSize}), {[{MsgId, TotalSize, Offset}|VMAcc], VTSAcc + TotalSize + ?FILE_PACKING_ADJUSTMENT } end end, {[], 0}, Messages), - % foldl reverses lists and find_contiguous_block_prefix needs elems in the same order - % as from scan_file_for_valid_messages + %% foldl reverses lists and find_contiguous_block_prefix needs elems in the same order + %% as from scan_file_for_valid_messages {ContiguousTop, _} = find_contiguous_block_prefix(lists:reverse(ValidMessagesRev)), Right = case Files of [] -> undefined; @@ -682,17 +790,20 @@ recover_crashed_compactions1(Files, TmpFile) -> GrabMsgId = fun ({MsgId, _TotalSize, _FileOffset}) -> MsgId end, NonTmpRelatedFile = filename:rootname(TmpFile) ++ ?FILE_EXTENSION, true = lists:member(NonTmpRelatedFile, Files), - % [{MsgId, TotalSize, FileOffset}] - {ok, UncorruptedMessagesTmp} = scan_file_for_valid_messages(form_filename(TmpFile)), + %% [{MsgId, TotalSize, FileOffset}] + {ok, UncorruptedMessagesTmp} + = scan_file_for_valid_messages(form_filename(TmpFile)), MsgIdsTmp = lists:map(GrabMsgId, UncorruptedMessagesTmp), - % all of these messages should appear in the mnesia table, otherwise they wouldn't have been copied out + %% all of these messages should appear in the mnesia table, otherwise they wouldn't have been copied out lists:foreach(fun (MsgId) -> - true = 0 < length(mnesia:dirty_match_object(rabbit_disk_queue, - #dq_msg_loc { msg_id = MsgId, - queue_and_seq_id = '_', - is_delivered = '_'})) + true = 0 < length(mnesia:dirty_match_object + (rabbit_disk_queue, + #dq_msg_loc { msg_id = MsgId, + queue_and_seq_id = '_', + is_delivered = '_'})) end, MsgIdsTmp), - {ok, UncorruptedMessages} = scan_file_for_valid_messages(form_filename(NonTmpRelatedFile)), + {ok, UncorruptedMessages} + = scan_file_for_valid_messages(form_filename(NonTmpRelatedFile)), MsgIds = lists:map(GrabMsgId, UncorruptedMessages), %% 1) It's possible that everything in the tmp file is also in the main file %% such that the main file is (prefix ++ tmpfile). This means that compaction @@ -714,66 +825,74 @@ recover_crashed_compactions1(Files, TmpFile) -> %% Plan: Truncate the main file back to before any of the files in the tmp file and copy %% them over again case lists:all(fun (MsgId) -> lists:member(MsgId, MsgIds) end, MsgIdsTmp) of - true -> % we're in case 1, 2 or 3 above. Just delete the tmp file - % note this also catches the case when the tmp file is empty + true -> %% we're in case 1, 2 or 3 above. Just delete the tmp file + %% note this also catches the case when the tmp file is empty ok = file:delete(TmpFile); _False -> - % we're in case 4 above. - % check that everything in the main file is a valid message in mnesia + %% we're in case 4 above. + %% check that everything in the main file is a valid message in mnesia lists:foreach(fun (MsgId) -> - true = 0 < - length(mnesia:dirty_match_object(rabbit_disk_queue, - #dq_msg_loc { msg_id = MsgId, - queue_and_seq_id = '_', - is_delivered = '_'})) + true = 0 < length(mnesia:dirty_match_object + (rabbit_disk_queue, + #dq_msg_loc { msg_id = MsgId, + queue_and_seq_id = '_', + is_delivered = '_'})) end, MsgIds), - % The main file should be contiguous + %% The main file should be contiguous {Top, MsgIds} = find_contiguous_block_prefix(UncorruptedMessages), - % we should have that none of the messages in the prefix are in the tmp file - true = lists:all(fun (MsgId) -> not(lists:member(MsgId, MsgIdsTmp)) end, MsgIds), + %% we should have that none of the messages in the prefix are in the tmp file + true = lists:all(fun (MsgId) -> not(lists:member(MsgId, MsgIdsTmp)) end, + MsgIds), - {ok, MainHdl} = file:open(form_filename(NonTmpRelatedFile), [write, raw, binary, delayed_write]), + {ok, MainHdl} = file:open(form_filename(NonTmpRelatedFile), + [write, raw, binary, delayed_write]), {ok, Top} = file:position(MainHdl, Top), - ok = file:truncate(MainHdl), % wipe out any rubbish at the end of the file - % there really could be rubbish at the end of the file - we could have failed after the - % extending truncate. - % Remember the head of the list will be the highest entry in the file + ok = file:truncate(MainHdl), %% wipe out any rubbish at the end of the file + %% there really could be rubbish at the end of the file - we could have failed after the + %% extending truncate. + %% Remember the head of the list will be the highest entry in the file [{_, TmpTopTotalSize, TmpTopOffset}|_] = UncorruptedMessagesTmp, TmpSize = TmpTopOffset + TmpTopTotalSize + ?FILE_PACKING_ADJUSTMENT, ExpectedAbsPos = Top + TmpSize, {ok, ExpectedAbsPos} = file:position(MainHdl, {cur, TmpSize}), - ok = file:truncate(MainHdl), % and now extend the main file as big as necessary in a single move - % if we run out of disk space, this truncate could fail, but we still - % aren't risking losing data - {ok, TmpHdl} = file:open(form_filename(TmpFile), [read, raw, binary, read_ahead]), + ok = file:truncate(MainHdl), %% and now extend the main file as big as necessary in a single move + %% if we run out of disk space, this truncate could fail, but we still + %% aren't risking losing data + {ok, TmpHdl} = file:open(form_filename(TmpFile), + [read, raw, binary, read_ahead]), {ok, TmpSize} = file:copy(TmpHdl, MainHdl, TmpSize), ok = file:close(MainHdl), ok = file:close(TmpHdl), ok = file:delete(TmpFile), - {ok, MainMessages} = scan_file_for_valid_messages(form_filename(NonTmpRelatedFile)), + {ok, MainMessages} + = scan_file_for_valid_messages(form_filename(NonTmpRelatedFile)), MsgIdsMain = lists:map(GrabMsgId, MainMessages), - % check that everything in MsgIds is in MsgIdsMain - true = lists:all(fun (MsgId) -> lists:member(MsgId, MsgIdsMain) end, MsgIds), - % check that everything in MsgIdsTmp is in MsgIdsMain - true = lists:all(fun (MsgId) -> lists:member(MsgId, MsgIdsMain) end, MsgIdsTmp) + %% check that everything in MsgIds is in MsgIdsMain + true = lists:all(fun (MsgId) -> lists:member(MsgId, MsgIdsMain) end, + MsgIds), + %% check that everything in MsgIdsTmp is in MsgIdsMain + true = lists:all(fun (MsgId) -> lists:member(MsgId, MsgIdsMain) end, + MsgIdsTmp) end, ok. -% this assumes that the messages are ordered such that the highest address is at -% the head of the list. -% this matches what scan_file_for_valid_messages produces +%% this assumes that the messages are ordered such that the highest address is at +%% the head of the list. +%% this matches what scan_file_for_valid_messages produces find_contiguous_block_prefix([]) -> {0, []}; find_contiguous_block_prefix([{MsgId, TotalSize, Offset}|Tail]) -> case find_contiguous_block_prefix(Tail, Offset, [MsgId]) of - {ok, Acc} -> {Offset + TotalSize + ?FILE_PACKING_ADJUSTMENT, lists:reverse(Acc)}; + {ok, Acc} -> {Offset + TotalSize + ?FILE_PACKING_ADJUSTMENT, + lists:reverse(Acc)}; Res -> Res end. find_contiguous_block_prefix([], 0, Acc) -> {ok, Acc}; find_contiguous_block_prefix([], _N, _Acc) -> {0, []}; -find_contiguous_block_prefix([{MsgId, TotalSize, Offset}|Tail], ExpectedOffset, Acc) +find_contiguous_block_prefix([{MsgId, TotalSize, Offset}|Tail], + ExpectedOffset, Acc) when ExpectedOffset =:= Offset + TotalSize + ?FILE_PACKING_ADJUSTMENT -> find_contiguous_block_prefix(Tail, Offset, [MsgId|Acc]); find_contiguous_block_prefix(List, _ExpectedOffset, _Acc) -> @@ -800,7 +919,8 @@ append_message(FileHdl, MsgId, MsgBody) when is_binary(MsgBody) -> TotalSize = BodySize + MsgIdBinSize, case file:write(FileHdl, <<TotalSize:?INTEGER_SIZE_BITS, MsgIdBinSize:?INTEGER_SIZE_BITS, - MsgIdBin:MsgIdBinSize/binary, MsgBody:BodySize/binary, + MsgIdBin:MsgIdBinSize/binary, + MsgBody:BodySize/binary, ?WRITE_OK:?WRITE_OK_SIZE_BITS>>) of ok -> {ok, TotalSize}; KO -> KO @@ -811,9 +931,12 @@ read_message_at_offset(FileHdl, Offset, TotalSize) -> case file:position(FileHdl, {bof, Offset}) of {ok, Offset} -> case file:read(FileHdl, TotalSize + ?FILE_PACKING_ADJUSTMENT) of - {ok, <<TotalSize:?INTEGER_SIZE_BITS, MsgIdBinSize:?INTEGER_SIZE_BITS, Rest:TotalSizeWriteOkBytes/binary>>} -> + {ok, <<TotalSize:?INTEGER_SIZE_BITS, + MsgIdBinSize:?INTEGER_SIZE_BITS, + Rest:TotalSizeWriteOkBytes/binary>>} -> BodySize = TotalSize - MsgIdBinSize, - <<_MsgId:MsgIdBinSize/binary, MsgBody:BodySize/binary, ?WRITE_OK:?WRITE_OK_SIZE_BITS>> = Rest, + <<_MsgId:MsgIdBinSize/binary, MsgBody:BodySize/binary, + ?WRITE_OK:?WRITE_OK_SIZE_BITS>> = Rest, {ok, {MsgBody, BodySize}}; KO -> KO end; @@ -823,7 +946,7 @@ read_message_at_offset(FileHdl, Offset, TotalSize) -> scan_file_for_valid_messages(File) -> {ok, Hdl} = file:open(File, [raw, binary, read]), Valid = scan_file_for_valid_messages(Hdl, 0, []), - _ = file:close(Hdl), % if something really bad's happened, the close could fail, but ignore + _ = file:close(Hdl), %% if something really bad's happened, the close could fail, but ignore Valid. scan_file_for_valid_messages(FileHdl, Offset, Acc) -> @@ -832,7 +955,8 @@ scan_file_for_valid_messages(FileHdl, Offset, Acc) -> {ok, {corrupted, NextOffset}} -> scan_file_for_valid_messages(FileHdl, NextOffset, Acc); {ok, {ok, MsgId, TotalSize, NextOffset}} -> - scan_file_for_valid_messages(FileHdl, NextOffset, [{MsgId, TotalSize, Offset}|Acc]); + scan_file_for_valid_messages(FileHdl, NextOffset, + [{MsgId, TotalSize, Offset}|Acc]); _KO -> {ok, Acc} %% bad message, but we may still have recovered some valid messages end. @@ -854,14 +978,17 @@ read_next_file_entry(FileHdl, Offset) -> case file:read(FileHdl, MsgIdBinSize) of {ok, <<MsgId:MsgIdBinSize/binary>>} -> ExpectedAbsPos = Offset + TwoIntegers + TotalSize, - case file:position(FileHdl, {cur, TotalSize - MsgIdBinSize}) of + case file:position(FileHdl, + {cur, TotalSize - MsgIdBinSize}) of {ok, ExpectedAbsPos} -> + NextOffset = Offset + TotalSize + + ?FILE_PACKING_ADJUSTMENT, case file:read(FileHdl, 1) of {ok, <<?WRITE_OK:?WRITE_OK_SIZE_BITS>>} -> - {ok, {ok, binary_to_term(MsgId), TotalSize, - Offset + ?FILE_PACKING_ADJUSTMENT + TotalSize}}; + {ok, {ok, binary_to_term(MsgId), + TotalSize, NextOffset}}; {ok, _SomeOtherData} -> - {ok, {corrupted, Offset + ?FILE_PACKING_ADJUSTMENT + TotalSize}}; + {ok, {corrupted, NextOffset}}; KO -> KO end; {ok, _SomeOtherPos} -> {ok, eof}; %% seek failed, so give up |