diff options
Diffstat (limited to 'lib/kernel/src/disk_log_1.erl')
| -rw-r--r-- | lib/kernel/src/disk_log_1.erl | 1551 |
1 files changed, 1551 insertions, 0 deletions
diff --git a/lib/kernel/src/disk_log_1.erl b/lib/kernel/src/disk_log_1.erl new file mode 100644 index 0000000000..7103417149 --- /dev/null +++ b/lib/kernel/src/disk_log_1.erl @@ -0,0 +1,1551 @@ +%% +%% %CopyrightBegin% +%% +%% Copyright Ericsson AB 1997-2009. All Rights Reserved. +%% +%% The contents of this file are subject to the Erlang Public License, +%% Version 1.1, (the "License"); you may not use this file except in +%% compliance with the License. You should have received a copy of the +%% Erlang Public License along with this software. If not, it can be +%% retrieved online at http://www.erlang.org/. +%% +%% 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. +%% +%% %CopyrightEnd% +%% +-module(disk_log_1). + +%% Efficient file based log - implementation part + +-export([int_open/4, ext_open/4, logl/1, close/3, truncate/3, chunk/5, + sync/2, write_cache/2]). +-export([mf_int_open/7, mf_int_log/3, mf_int_close/2, mf_int_inc/2, + mf_ext_inc/2, mf_int_chunk/4, mf_int_chunk_step/3, + mf_sync/1, mf_write_cache/1]). +-export([mf_ext_open/7, mf_ext_log/3, mf_ext_close/2]). + +-export([print_index_file/1]). +-export([read_index_file/1]). +-export([read_size_file/1, read_size_file_version/1]). +-export([chunk_read_only/5]). +-export([mf_int_chunk_read_only/4]). +-export([change_size_wrap/3]). +-export([get_wrap_size/1]). +-export([is_head/1]). +-export([position/3, truncate_at/3, fwrite/4, fclose/2]). + +-compile({inline,[{scan_f2,7}]}). + +-import(lists, [concat/1, reverse/1, sum/1]). + +-include("disk_log.hrl"). + +%%% At the head of a LOG file we have [?LOGMAGIC, ?OPENED | ?CLOSED]. +%%% Otherwise it's not a LOG file. Following that, the head, come the +%%% logged items. +%%% +%%% There are four formats of wrap log files (so far). Only the size +%%% file and the index file differ between versions between the first +%%% three version. The fourth version 2(a), has some protection +%%% against damaged item sizes. +%%% Version 0: no "siz" file +%%% Version 1: "siz" file, 4 byte sizes +%%% Version 2: 8 byte sizes (support for large files) +%%% Version 2(a): Change of the format of logged items: +%%% if the size of a term binary is greater than or equal to +%%% ?MIN_MD5_TERM, a logged item looks like +%%% <<Size:32, ?BIGMAGICHEAD:32, MD5:128, Term/binary>>, +%%% otherwise <<Size:32, ?BIGMAGICHEAD:32, Term/binary>>. + +%%%---------------------------------------------------------------------- +%%% API +%%%---------------------------------------------------------------------- + +%% -> {ok, NoBytes, NewFdC} | {Error, NewFdC} +log(FdC, FileName, X) -> + {Bs, Size} = logl(X, [], 0), + case fwrite(FdC, FileName, Bs, Size) of + {ok, NewFdC} -> + {ok, Size, NewFdC}; + Error -> + Error + end. + +-spec logl([binary()]) -> {iolist(), non_neg_integer()}. +logl(X) -> + logl(X, [], 0). + +logl([X | T], Bs, Size) -> + Sz = byte_size(X), + BSz = <<Sz:?SIZESZ/unit:8>>, + NBs = case Sz < ?MIN_MD5_TERM of + true -> + [Bs, BSz, ?BIGMAGICHEAD | X]; + false -> + MD5 = erlang:md5(BSz), + [Bs, BSz, ?BIGMAGICHEAD, MD5 | X] + end, + logl(T, NBs, Size + ?HEADERSZ + Sz); +logl([], Bs, Size) -> + {Bs, Size}. + +%% -> {ok, NewFdC} | {Error, NewFdC} +write_cache(#cache{fd = Fd, c = C}, FName) -> + erase(write_cache_timer_is_running), + write_cache(Fd, FName, C). + +%% -> {Reply, NewFdC}; Reply = ok | Error +sync(FdC, FName) -> + fsync(FdC, FName). + +%% -> {Reply, NewFdC}; Reply = ok | Error +truncate(FdC, FileName, Head) -> + Reply = truncate_at(FdC, FileName, ?HEADSZ), + case Reply of + {ok, _} when Head =:= none -> + Reply; + {ok, FdC1} -> + {ok, B} = Head, + case log(FdC1, FileName, [B]) of + {ok, _NoBytes, NewFdC} -> + {ok, NewFdC}; + Reply2 -> + Reply2 + end; + _ -> + Reply + end. + +%% -> {NewFdC, Reply}, Reply = {Cont, Binaries} | {error, Reason} | eof +chunk(FdC, FileName, Pos, B, N) when is_binary(B) -> + true = byte_size(B) >= ?HEADERSZ, + do_handle_chunk(FdC, FileName, Pos, B, N); +chunk(FdC, FileName, Pos, NoBytes, N) -> + MaxNoBytes = case NoBytes of + [] -> ?MAX_CHUNK_SIZE; + _ -> erlang:max(NoBytes, ?MAX_CHUNK_SIZE) + end, + case read_chunk(FdC, FileName, Pos, MaxNoBytes) of + {NewFdC, {ok, Bin}} when byte_size(Bin) < ?HEADERSZ -> + {NewFdC, {error, {corrupt_log_file, FileName}}}; + {NewFdC, {ok, Bin}} when NoBytes =:= []; byte_size(Bin) >= NoBytes -> + NewPos = Pos + byte_size(Bin), + do_handle_chunk(NewFdC, FileName, NewPos, Bin, N); + {NewFdC, {ok, _Bin}} -> + {NewFdC, {error, {corrupt_log_file, FileName}}}; + {NewFdC, eof} when is_integer(NoBytes) -> % "cannot happen" + {NewFdC, {error, {corrupt_log_file, FileName}}}; + Other -> % eof or error + Other + end. + +do_handle_chunk(FdC, FileName, Pos, B, N) -> + case handle_chunk(B, Pos, N, []) of + corrupt -> + {FdC, {error, {corrupt_log_file, FileName}}}; + {C, []} -> + chunk(FdC, FileName, C#continuation.pos, C#continuation.b, N); + C_Ack -> + {FdC, C_Ack} + end. + +handle_chunk(B, Pos, 0, Ack) when byte_size(B) >= ?HEADERSZ -> + {#continuation{pos = Pos, b = B}, Ack}; +handle_chunk(B= <<Size:?SIZESZ/unit:8, ?BIGMAGICINT:?MAGICSZ/unit:8, + Tail/binary>>, Pos, N, Ack) when Size < ?MIN_MD5_TERM -> + case Tail of + <<BinTerm:Size/binary, Tail2/binary>> -> + %% The client calls binary_to_term/1. + handle_chunk(Tail2, Pos, N-1, [BinTerm | Ack]); + _ -> + BytesToRead = Size + ?HEADERSZ, + {#continuation{pos = Pos - byte_size(B), b = BytesToRead}, Ack} + end; +handle_chunk(B= <<Size:?SIZESZ/unit:8, ?BIGMAGICINT:?MAGICSZ/unit:8, + Tail/binary>>, Pos, _N, Ack) -> % when Size >= ?MIN_MD5_TERM + MD5 = erlang:md5(<<Size:?SIZESZ/unit:8>>), + case Tail of + %% The requested object is always bigger than a chunk. + <<MD5:16/binary, Bin:Size/binary>> -> + {#continuation{pos = Pos, b = []}, [Bin | Ack]}; + <<MD5:16/binary, _/binary>> -> + BytesToRead = Size + ?HEADERSZ + 16, + {#continuation{pos = Pos - byte_size(B), b = BytesToRead}, Ack}; + _ when byte_size(Tail) >= 16 -> + corrupt; + _ -> + {#continuation{pos = Pos - byte_size(B), b = []}, Ack} + end; +handle_chunk(B= <<Size:?SIZESZ/unit:8, ?MAGICINT:?MAGICSZ/unit:8, Tail/binary>>, + Pos, N, Ack) -> + %% Version 2, before 2(a). + case Tail of + <<BinTerm:Size/binary, Tail2/binary>> -> + handle_chunk(Tail2, Pos, N-1, [BinTerm | Ack]); + _ -> + %% We read the whole thing into one binary, even if Size is huge. + BytesToRead = Size + ?HEADERSZ, + {#continuation{pos = Pos - byte_size(B), b = BytesToRead}, Ack} + end; +handle_chunk(B, _Pos, _N, _Ack) when byte_size(B) >= ?HEADERSZ -> + corrupt; +handle_chunk(B, Pos, _N, Ack) -> + {#continuation{pos = Pos-byte_size(B), b = []}, Ack}. + +read_chunk(FdC, FileName, Pos, MaxBytes) -> + {FdC1, R} = pread(FdC, FileName, Pos + ?HEADSZ, MaxBytes), + case position(FdC1, FileName, eof) of + {ok, NewFdC, _Pos} -> + {NewFdC, R}; + {Error, NewFdC} -> + {NewFdC, Error} + end. + +%% Used by wrap_log_reader. +%% -> {NewFdC, Reply}, +%% Reply = {Cont, Binaries, Bad} (Bad >= 0) | {error, Reason} | eof +chunk_read_only(FdC = #cache{}, FileName, Pos, B, N) -> + do_chunk_read_only(FdC, FileName, Pos, B, N); +chunk_read_only(Fd, FileName, Pos, B, N) -> + %% wrap_log_reader calling... + FdC = #cache{fd = Fd}, + {_NFdC, Reply} = do_chunk_read_only(FdC, FileName, Pos, B, N), + Reply. + +do_chunk_read_only(FdC, FileName, Pos, B, N) when is_binary(B) -> + true = byte_size(B) >= ?HEADERSZ, + do_handle_chunk_ro(FdC, FileName, Pos, B, N); +do_chunk_read_only(FdC, FileName, Pos, NoBytes, N) -> + MaxNoBytes = case NoBytes of + [] -> ?MAX_CHUNK_SIZE; + _ -> erlang:max(NoBytes, ?MAX_CHUNK_SIZE) + end, + case read_chunk_ro(FdC, FileName, Pos, MaxNoBytes) of + {NewFdC, {ok, Bin}} when byte_size(Bin) < ?HEADERSZ -> + NewCont = #continuation{pos = Pos+byte_size(Bin), b = []}, + {NewFdC, {NewCont, [], byte_size(Bin)}}; + {NewFdC, {ok, Bin}} when NoBytes =:= []; byte_size(Bin) >= NoBytes -> + NewPos = Pos + byte_size(Bin), + do_handle_chunk_ro(NewFdC, FileName, NewPos, Bin, N); + {NewFdC, {ok, Bin}} -> + NewCont = #continuation{pos = Pos+byte_size(Bin), b = []}, + {NewFdC, {NewCont, [], byte_size(Bin)-?HEADERSZ}}; + {NewFdC, eof} when is_integer(NoBytes) -> % "cannot happen" + {NewFdC, eof}; % what else? + Other -> + Other + end. + +do_handle_chunk_ro(FdC, FileName, Pos, B, N) -> + case handle_chunk_ro(B, Pos, N, [], 0) of + {C, [], 0} -> + #continuation{pos = NewPos, b = NoBytes} = C, + do_chunk_read_only(FdC, FileName, NewPos, NoBytes, N); + C_Ack_Bad -> + {FdC, C_Ack_Bad} + end. + +handle_chunk_ro(B, Pos, 0, Ack, Bad) when byte_size(B) >= ?HEADERSZ -> + {#continuation{pos = Pos, b = B}, Ack, Bad}; +handle_chunk_ro(B= <<Size:?SIZESZ/unit:8, ?BIGMAGICINT:?MAGICSZ/unit:8, + Tail/binary>>, Pos, N, Ack, Bad) when Size < ?MIN_MD5_TERM -> + case Tail of + <<BinTerm:Size/binary, Tail2/binary>> -> + handle_chunk_ro(Tail2, Pos, N-1, [BinTerm | Ack], Bad); + _ -> + BytesToRead = Size + ?HEADERSZ, + {#continuation{pos = Pos - byte_size(B), b = BytesToRead}, Ack, Bad} + end; +handle_chunk_ro(B= <<Size:?SIZESZ/unit:8, ?BIGMAGICINT:?MAGICSZ/unit:8, + Tail/binary>>, Pos, N, Ack, Bad) -> % when Size>=?MIN_MD5_TERM + MD5 = erlang:md5(<<Size:?SIZESZ/unit:8>>), + case Tail of + <<MD5:16/binary, Bin:Size/binary>> -> + %% The requested object is always bigger than a chunk. + {#continuation{pos = Pos, b = []}, [Bin | Ack], Bad}; + <<MD5:16/binary, _/binary>> -> + BytesToRead = Size + ?HEADERSZ + 16, + {#continuation{pos = Pos - byte_size(B), b = BytesToRead}, Ack, Bad}; + <<_BadMD5:16/binary, _:1/unit:8, Tail2/binary>> -> + handle_chunk_ro(Tail2, Pos, N-1, Ack, Bad+1); + _ -> + {#continuation{pos = Pos - byte_size(B), b = []}, Ack, Bad} + end; +handle_chunk_ro(B= <<Size:?SIZESZ/unit:8, ?MAGICINT:?MAGICSZ/unit:8, + Tail/binary>>, Pos, N, Ack, Bad) -> + %% Version 2, before 2(a). + case Tail of + <<BinTerm:Size/binary, Tail2/binary>> -> + handle_chunk_ro(Tail2, Pos, N-1, [BinTerm | Ack], Bad); + _ -> + %% We read the whole thing into one binary, even if Size is huge. + BytesToRead = Size + ?HEADERSZ, + {#continuation{pos = Pos - byte_size(B), b = BytesToRead}, Ack, Bad} + end; +handle_chunk_ro(B, Pos, N, Ack, Bad) when byte_size(B) >= ?HEADERSZ -> + <<_:1/unit:8, B2/binary>> = B, + handle_chunk_ro(B2, Pos, N-1, Ack, Bad+1); +handle_chunk_ro(B, Pos, _N, Ack, Bad) -> + {#continuation{pos = Pos-byte_size(B), b = []}, Ack, Bad}. + +read_chunk_ro(FdC, FileName, Pos, MaxBytes) -> + pread(FdC, FileName, Pos + ?HEADSZ, MaxBytes). + +%% -> ok | throw(Error) +close(#cache{fd = Fd, c = []}, _FileName, read_only) -> + file:close(Fd); +close(#cache{fd = Fd, c = C}, FileName, read_write) -> + {Reply, _NewFdC} = write_cache(Fd, FileName, C), + mark(Fd, FileName, ?CLOSED), + file:close(Fd), + if Reply =:= ok -> ok; true -> throw(Reply) end. + +%% Open an internal file. Head is ignored if Mode is read_only. +%% int_open(FileName, Repair, Mode, Head) -> +%% {ok, {Alloc, FdC, HeadSize, FileSize}} +%% | {repaired, FdC, Terms, BadBytes, FileSize} +%% | throw(Error) +%% Alloc = new | existed +%% HeadSize = {NumberOfItemsWritten, NumberOfBytesWritten} +%% (HeadSize is equal {0, 0} if Alloc =:= existed, or no header written.) +int_open(FName, truncate, read_write, Head) -> + new_int_file(FName, Head); +int_open(FName, Repair, read_write, Head) -> + case open_read(FName) of + {ok, Fd} -> %% File exists + case file:read(Fd, ?HEADSZ) of + {ok, FileHead} -> + case is_head(FileHead) of + yes -> + file:close(Fd), + case open_update(FName) of + {ok, Fd2} -> + mark(Fd2, FName, ?OPENED), + FdC1 = #cache{fd = Fd2}, + {FdC, P} = position_close(FdC1, FName,eof), + {ok, {existed, FdC, {0, 0}, P}}; + Error -> + file_error(FName, Error) + end; + yes_not_closed when Repair -> + repair(Fd, FName); + yes_not_closed when not Repair -> + file:close(Fd), + throw({error, {need_repair, FName}}); + no -> + file:close(Fd), + throw({error, {not_a_log_file, FName}}) + end; + eof -> + file:close(Fd), + throw({error, {not_a_log_file, FName}}); + Error -> + file_error_close(Fd, FName, Error) + end; + _Other -> + new_int_file(FName, Head) + end; +int_open(FName, _Repair, read_only, _Head) -> + case open_read(FName) of + {ok, Fd} -> %% File exists + case file:read(Fd, ?HEADSZ) of + {ok, Head} -> + case is_head(Head) of + yes -> + {ok, P} = position_close2(Fd, FName, eof), + FdC = #cache{fd = Fd}, + {ok, {existed, FdC, {0, 0}, P}}; + yes_not_closed -> + {ok, P} = position_close2(Fd, FName, eof), + FdC = #cache{fd = Fd}, + {ok, {existed, FdC, {0, 0}, P}}; + no -> + file:close(Fd), + throw({error, {not_a_log_file, FName}}) + end; + eof -> + file:close(Fd), + throw({error, {not_a_log_file, FName}}); + Error -> + file_error_close(Fd, FName, Error) + end; + Error -> + file_error(FName, Error) + end. + +new_int_file(FName, Head) -> + case open_update(FName) of + {ok, Fd} -> + ok = truncate_at_close2(Fd, FName, bof), + fwrite_close2(Fd, FName, [?LOGMAGIC, ?OPENED]), + {FdC1, Nh, HeadSz} = int_log_head(Fd, Head), + {FdC, FileSize} = position_close(FdC1, FName, cur), + {ok, {new, FdC, {Nh, ?HEADERSZ + HeadSz}, FileSize}}; + Error -> + file_error(FName, Error) + end. + +%% -> {FdC, NoItemsWritten, NoBytesWritten} | throw(Error) +int_log_head(Fd, Head) -> + case lh(Head, internal) of + {ok, BinHead} -> + {Bs, Size} = logl([BinHead]), + {ok, FdC} = fwrite_header(Fd, Bs, Size), + {FdC, 1, Size}; + none -> + {#cache{fd = Fd}, 0, 0}; + Error -> + file:close(Fd), + throw(Error) + end. + +%% Open an external file. +%% -> {ok, {Alloc, FdC, HeadSize}, FileSize} | throw(Error) +ext_open(FName, truncate, read_write, Head) -> + new_ext_file(FName, Head); +ext_open(FName, _Repair, read_write, Head) -> + case file:read_file_info(FName) of + {ok, _FileInfo} -> + case open_update(FName) of + {ok, Fd} -> + {ok, P} = position_close2(Fd, FName, eof), + FdC = #cache{fd = Fd}, + {ok, {existed, FdC, {0, 0}, P}}; + Error -> + file_error(FName, Error) + end; + _Other -> + new_ext_file(FName, Head) + end; +ext_open(FName, _Repair, read_only, _Head) -> + case open_read(FName) of + {ok, Fd} -> + {ok, P} = position_close2(Fd, FName, eof), + FdC = #cache{fd = Fd}, + {ok, {existed, FdC, {0, 0}, P}}; + Error -> + file_error(FName, Error) + end. + +new_ext_file(FName, Head) -> + case open_truncate(FName) of + {ok, Fd} -> + {FdC1, HeadSize} = ext_log_head(Fd, Head), + {FdC, FileSize} = position_close(FdC1, FName, cur), + {ok, {new, FdC, HeadSize, FileSize}}; + Error -> + file_error(FName, Error) + end. + +%% -> {FdC, {NoItemsWritten, NoBytesWritten}} | throw(Error) +ext_log_head(Fd, Head) -> + case lh(Head, external) of + {ok, BinHead} -> + Size = byte_size(BinHead), + {ok, FdC} = fwrite_header(Fd, BinHead, Size), + {FdC, {1, Size}}; + none -> + {#cache{fd = Fd}, {0, 0}}; + Error -> + file:close(Fd), + throw(Error) + end. + +%% -> _Any | throw() +mark(Fd, FileName, What) -> + position_close2(Fd, FileName, 4), + fwrite_close2(Fd, FileName, What). + +%% -> {ok, Bin} | Error +lh({ok, Bin}, _Format) -> + {ok, Bin}; +lh({M, F, A}, Format) when is_list(A) -> + case catch apply(M, F, A) of + {ok, Head} when Format =:= internal -> + {ok, term_to_binary(Head)}; + {ok, Bin} when is_binary(Bin) -> + {ok, Bin}; + {ok, Bytes} -> + case catch list_to_binary(Bytes) of + {'EXIT', _} -> + {error, {invalid_header, {{M,F,A}, {ok, Bytes}}}}; + Bin -> + {ok, Bin} + end; + {'EXIT', Error} -> + {error, {invalid_header, {{M,F,A}, Error}}}; + Error -> + {error, {invalid_header, {{M,F,A}, Error}}} + end; +lh({M, F, A}, _Format) -> % cannot happen + {error, {invalid_header, {M, F, A}}}; +lh(none, _Format) -> + none; +lh(H, _F) -> % cannot happen + {error, {invalid_header, H}}. + +repair(In, File) -> + FSz = file_size(File), + error_logger:info_msg("disk_log: repairing ~p ...\n", [File]), + Tmp = add_ext(File, "TMP"), + {ok, {_Alloc, Out, {0, _}, _FileSize}} = new_int_file(Tmp, none), + scan_f_read(<<>>, In, Out, File, FSz, Tmp, ?MAX_CHUNK_SIZE, 0, 0). + +scan_f_read(B, In, Out, File, FSz, Tmp, MaxBytes, No, Bad) -> + case file:read(In, MaxBytes) of + eof -> + done_scan(In, Out, Tmp, File, No, Bad+byte_size(B)); + {ok, Bin} -> + NewBin = list_to_binary([B, Bin]), + {NB, NMax, Ack, NNo, NBad} = + scan_f(NewBin, FSz, [], No, Bad), + case log(Out, Tmp, lists:reverse(Ack)) of + {ok, _Size, NewOut} -> + scan_f_read(NB, In, NewOut, File, FSz, Tmp, NMax,NNo,NBad); + {{error, {file_error, _Filename, Error}}, NewOut} -> + repair_err(In, NewOut, Tmp, File, {error, Error}) + end; + Error -> + repair_err(In, Out, Tmp, File, Error) + end. + +scan_f(B = <<Size:?SIZESZ/unit:8, ?BIGMAGICINT:?MAGICSZ/unit:8, Tail/binary>>, + FSz, Ack, No, Bad) when Size < ?MIN_MD5_TERM -> + scan_f2(B, FSz, Ack, No, Bad, Size, Tail); +scan_f(B = <<Size:?SIZESZ/unit:8, ?BIGMAGICINT:?MAGICSZ/unit:8, Tail/binary>>, + FSz, Ack, No, Bad) -> % when Size >= ?MIN_MD5_TERM + MD5 = erlang:md5(<<Size:?SIZESZ/unit:8>>), + case Tail of + <<MD5:16/binary, BinTerm:Size/binary, Tail2/binary>> -> + case catch binary_to_term(BinTerm) of + {'EXIT', _} -> + scan_f(Tail2, FSz, Ack, No, Bad+Size); + _Term -> + scan_f(Tail2, FSz, [BinTerm | Ack], No+1, Bad) + end; + <<MD5:16/binary, _/binary>> -> + {B, Size-byte_size(Tail)+16, Ack, No, Bad}; + _ when byte_size(Tail) < 16 -> + {B, Size-byte_size(Tail)+16, Ack, No, Bad}; + _ -> + <<_:8, B2/binary>> = B, + scan_f(B2, FSz, Ack, No, Bad+1) + end; +scan_f(B = <<Size:?SIZESZ/unit:8, ?MAGICINT:?MAGICSZ/unit:8, Tail/binary>>, + FSz, Ack, No, Bad) when Size =< FSz -> + %% Since the file is not compressed, the item size cannot exceed + %% the file size. + scan_f2(B, FSz, Ack, No, Bad, Size, Tail); +scan_f(B = <<_:?HEADERSZ/unit:8, _/binary>>, FSz, Ack, No, Bad) -> + <<_:8, B2/binary>> = B, + scan_f(B2, FSz, Ack, No, Bad + 1); +scan_f(B, _FSz, Ack, No, Bad) -> + {B, ?MAX_CHUNK_SIZE, Ack, No, Bad}. + +scan_f2(B, FSz, Ack, No, Bad, Size, Tail) -> + case Tail of + <<BinTerm:Size/binary, Tail2/binary>> -> + case catch binary_to_term(BinTerm) of + {'EXIT', _} -> + <<_:8, B2/binary>> = B, + scan_f(B2, FSz, Ack, No, Bad+1); + _Term -> + scan_f(Tail2, FSz, [BinTerm | Ack], No+1, Bad) + end; + _ -> + {B, Size-byte_size(Tail), Ack, No, Bad} + end. + +done_scan(In, Out, OutName, FName, RecoveredTerms, BadChars) -> + file:close(In), + case catch fclose(Out, OutName) of + ok -> + case file:rename(OutName, FName) of + ok -> + case open_update(FName) of + {ok, New} -> + {ok, P} = position_close2(New, FName, eof), + FdC = #cache{fd = New}, + {repaired, FdC, RecoveredTerms, BadChars, P}; + Error -> + file_error(FName, Error) + end; + Error -> + file:delete(OutName), + file_error(FName, Error) + end; + Error -> + file:delete(OutName), + throw(Error) + end. + +repair_err(In, Out, OutName, ErrFileName, Error) -> + file:close(In), + catch fclose(Out, OutName), + % OutName is often the culprit, try to remove it anyway... + file:delete(OutName), + file_error(ErrFileName, Error). + +%% Used by wrap_log_reader. +-spec is_head(binary()) -> 'yes' | 'yes_not_closed' | 'no'. +is_head(<<M:4/binary, S:4/binary>>) when ?LOGMAGIC =:= M, ?CLOSED =:= S -> + yes; +is_head(<<M:4/binary, S:4/binary>>) when ?LOGMAGIC =:= M, ?OPENED =:= S -> + yes_not_closed; +is_head(Bin) when is_binary(Bin) -> + no. + +%%----------------------------------------------------------------- +%% Func: mf_int_open/7, mf_ext_open/7 +%% Args: FName = file:filename() +%% MaxB = integer() +%% MaxF = integer() +%% Repair = truncate | true | false +%% Mode = read_write | read_only +%% Head = none | {ok, Bin} | {M, F, A} +%% Version = integer() +%% Purpose: An ADT for wrapping logs. mf_int_ writes binaries (mf_ext_ +%% writes bytes) +%% to files called FName.1, FName.2, ..., FName.MaxF. +%% Writes MaxB bytes on each file. +%% Creates a file called Name.idx in the Dir. This +%% file contains the last written FileName as one byte, and +%% follwing that, the sizes of each file (size 0 number of items). +%% On startup, this file is read, and the next available +%% filename is used as first log file. +%% Reports can be browsed with Report Browser Tool (rb), or +%% read with disk_log. +%%----------------------------------------------------------------- +-spec mf_int_open(FName :: file:filename(), + MaxB :: integer(), + MaxF :: integer(), + Repair :: dlog_repair(), + Mode :: dlog_mode(), + Head :: dlog_head(), + Version :: integer()) + -> {'ok', #handle{}, integer()} + | {'repaired', #handle{}, + non_neg_integer(), non_neg_integer(), non_neg_integer()}. +%% | throw(FileError) +mf_int_open(FName, MaxB, MaxF, Repair, Mode, Head, Version) -> + {First, Sz, TotSz, NFiles} = read_index_file(Repair, FName, MaxF), + write_size_file(Mode, FName, MaxB, MaxF, Version), + NewMaxF = if + NFiles > MaxF -> + {MaxF, NFiles}; + true -> + MaxF + end, + case int_file_open(FName, First, 0, 0, Head, Repair, Mode) of + {ok, FdC, FileName, Lost, {NoItems, NoBytes}, FSz} -> + % firstPos = NoBytes is not always correct when the file + % existed, but it will have to do since we don't know + % where the header ends. + CurCnt = Sz + NoItems - Lost, + {ok, #handle{filename = FName, maxB = MaxB, + maxF = NewMaxF, curF = First, cur_fdc = FdC, + cur_name = FileName, cur_cnt = CurCnt, + acc_cnt = -Sz, curB = FSz, + firstPos = NoBytes, noFull = 0, accFull = 0}, + TotSz + CurCnt}; + {repaired, FdC, FileName, Rec, Bad, FSz} -> + {repaired, + #handle{filename = FName, maxB = MaxB, cur_name = FileName, + maxF = NewMaxF, curF = First, cur_fdc = FdC, + cur_cnt = Rec, acc_cnt = -Rec, curB = FSz, + firstPos = 0, noFull = 0, accFull = 0}, + Rec, Bad, TotSz + Rec} + end. + +%% -> {ok, handle(), Lost} | {error, Error, handle()} +mf_int_inc(Handle, Head) -> + #handle{filename = FName, cur_cnt = CurCnt, acc_cnt = AccCnt, + cur_name = FileName, curF = CurF, maxF = MaxF, + cur_fdc = CurFdC, noFull = NoFull} = Handle, + case catch wrap_int_log(FName, CurF, MaxF, CurCnt, Head) of + {NewF, NewMaxF, NewFdC, NewFileName, Nh, FirstPos, Lost} -> + Handle1 = Handle#handle{cur_fdc = NewFdC, curF = NewF, + cur_name = NewFileName, + cur_cnt = Nh, acc_cnt = AccCnt + CurCnt, + maxF = NewMaxF, firstPos = FirstPos, + curB = FirstPos, noFull = NoFull + 1}, + case catch close(CurFdC, FileName, read_write) of + ok -> + {ok, Handle1, Lost}; + Error -> % Error in the last file, new file opened. + {error, Error, Handle1} + end; + Error -> + {error, Error, Handle} + end. + +%% -> {ok, handle(), Logged, Lost, NoWraps} | {ok, handle(), Logged} +%% | {error, Error, handle(), Logged, Lost} +%% The returned handle is not always valid - something may +%% have been written before things went wrong. +mf_int_log(Handle, Bins, Head) -> + mf_int_log(Handle, Bins, Head, 0, []). + +mf_int_log(Handle, [], _Head, No, []) -> + {ok, Handle, No}; +mf_int_log(Handle, [], _Head, No, Wraps0) -> + Wraps = reverse(Wraps0), + {ok, Handle, No, sum(Wraps), Wraps}; +mf_int_log(Handle, Bins, Head, No0, Wraps) -> + #handle{curB = CurB, maxB = MaxB, cur_name = FileName, cur_fdc = CurFdC, + firstPos = FirstPos0, cur_cnt = CurCnt} = Handle, + {FirstBins, LastBins, NoBytes, N} = + int_split_bins(CurB, MaxB, FirstPos0, Bins), + case FirstBins of + [] -> + #handle{filename = FName, curF = CurF, maxF = MaxF, + acc_cnt = AccCnt, noFull = NoFull} = Handle, + case catch wrap_int_log(FName, CurF, MaxF, CurCnt, Head) of + {NewF, NewMaxF, NewFdC, NewFileName, Nh, FirstPos, Lost} -> + Handle1 = Handle#handle{cur_fdc = NewFdC, curF = NewF, + cur_cnt = Nh, + cur_name = NewFileName, + acc_cnt = AccCnt + CurCnt, + maxF = NewMaxF, + curB = FirstPos, + firstPos = FirstPos, + noFull = NoFull + 1}, + case catch close(CurFdC, FileName, read_write) of + ok -> + mf_int_log(Handle1, Bins, Head, No0 + Nh, + [Lost | Wraps]); + Error -> + Lost1 = Lost + sum(Wraps), + {error, Error, Handle1, No0 + Nh, Lost1} + end; + Error -> + {error, Error, Handle, No0, sum(Wraps)} + end; + _ -> + case fwrite(CurFdC, FileName, FirstBins, NoBytes) of + {ok, NewCurFdC} -> + Handle1 = Handle#handle{cur_fdc = NewCurFdC, + curB = CurB + NoBytes, + cur_cnt = CurCnt + N}, + mf_int_log(Handle1, LastBins, Head, No0 + N, Wraps); + {Error, NewCurFdC} -> + Handle1 = Handle#handle{cur_fdc = NewCurFdC}, + {error, Error, Handle1, No0, sum(Wraps)} + end + end. + +wrap_int_log(FName, CurF, MaxF, CurCnt, Head) -> + {NewF, NewMaxF} = inc_wrap(FName, CurF, MaxF), + {ok, NewFdC, NewFileName, Lost, {Nh, FirstPos}, _FileSize} = + int_file_open(FName, NewF, CurF, CurCnt, Head), + {NewF, NewMaxF, NewFdC, NewFileName, Nh, FirstPos, Lost}. + +%% -> {NewHandle, Reply}, Reply = {Cont, Binaries} | {error, Reason} | eof +mf_int_chunk(Handle, 0, Bin, N) -> + FirstF = find_first_file(Handle), + mf_int_chunk(Handle, {FirstF, 0}, Bin, N); +mf_int_chunk(#handle{curF = FileNo, cur_fdc = FdC, cur_name = FileName} + = Handle, {FileNo, Pos}, Bin, N) -> + {NewFdC, Reply} = chunk(FdC, FileName, Pos, Bin, N), + {Handle#handle{cur_fdc = NewFdC}, conv(Reply, FileNo)}; +mf_int_chunk(Handle, {FileNo, Pos}, Bin, N) -> + FName = add_ext(Handle#handle.filename, FileNo), + NFileNo = inc(FileNo, Handle#handle.maxF), + case catch int_open(FName, true, read_only, any) of + {error, _Reason} -> + error_logger:info_msg("disk_log: chunk error. File ~p missing.\n\n", + [FName]), + mf_int_chunk(Handle, {NFileNo, 0}, [], N); + {ok, {_Alloc, FdC, _HeadSize, _FileSize}} -> + case chunk(FdC, FName, Pos, Bin, N) of + {NewFdC, eof} -> + file:close(NewFdC#cache.fd), + mf_int_chunk(Handle, {NFileNo, 0}, [], N); + {NewFdC, Other} -> + file:close(NewFdC#cache.fd), + {Handle, conv(Other, FileNo)} + end + end. + +%% -> {NewHandle, Reply}, +%% Reply = {Cont, Binaries, Bad} (Bad >= 0) | {error, Reason} | eof +mf_int_chunk_read_only(Handle, 0, Bin, N) -> + FirstF = find_first_file(Handle), + mf_int_chunk_read_only(Handle, {FirstF, 0}, Bin, N); +mf_int_chunk_read_only(#handle{curF = FileNo, cur_fdc = FdC, cur_name=FileName} + = Handle, {FileNo, Pos}, Bin, N) -> + {NewFdC, Reply} = do_chunk_read_only(FdC, FileName, Pos, Bin, N), + {Handle#handle{cur_fdc = NewFdC}, conv(Reply, FileNo)}; +mf_int_chunk_read_only(Handle, {FileNo, Pos}, Bin, N) -> + FName = add_ext(Handle#handle.filename, FileNo), + NFileNo = inc(FileNo, Handle#handle.maxF), + case catch int_open(FName, true, read_only, any) of + {error, _Reason} -> + error_logger:info_msg("disk_log: chunk error. File ~p missing.\n\n", + [FName]), + mf_int_chunk_read_only(Handle, {NFileNo, 0}, [], N); + {ok, {_Alloc, FdC, _HeadSize, _FileSize}} -> + case do_chunk_read_only(FdC, FName, Pos, Bin, N) of + {NewFdC, eof} -> + file:close(NewFdC#cache.fd), + mf_int_chunk_read_only(Handle, {NFileNo,0}, [], N); + {NewFdC, Other} -> + file:close(NewFdC#cache.fd), + {Handle, conv(Other, FileNo)} + end + end. + +%% -> {ok, Cont} | Error +mf_int_chunk_step(Handle, 0, Step) -> + FirstF = find_first_file(Handle), + mf_int_chunk_step(Handle, {FirstF, 0}, Step); +mf_int_chunk_step(Handle, {FileNo, _Pos}, Step) -> + NFileNo = inc(FileNo, Handle#handle.maxF, Step), + FileName = add_ext(Handle#handle.filename, NFileNo), + case file:read_file_info(FileName) of + {ok, _FileInfo} -> + {ok, #continuation{pos = {NFileNo, 0}, b = []}}; + _Error -> + {error, end_of_log} + end. + +%% -> {Reply, handle()}; Reply = ok | Error +mf_write_cache(#handle{filename = FName, cur_fdc = FdC} = Handle) -> + erase(write_cache_timer_is_running), + #cache{fd = Fd, c = C} = FdC, + {Reply, NewFdC} = write_cache(Fd, FName, C), + {Reply, Handle#handle{cur_fdc = NewFdC}}. + +%% -> {Reply, handle()}; Reply = ok | Error +mf_sync(#handle{filename = FName, cur_fdc = FdC} = Handle) -> + {Reply, NewFdC} = fsync(FdC, FName), + {Reply, Handle#handle{cur_fdc = NewFdC}}. + +%% -> ok | throw(FileError) +mf_int_close(#handle{filename = FName, curF = CurF, cur_name = FileName, + cur_fdc = CurFdC, cur_cnt = CurCnt}, Mode) -> + close(CurFdC, FileName, Mode), + write_index_file(Mode, FName, CurF, CurF, CurCnt), + ok. + +%% -> {ok, handle(), Cnt} | throw(FileError) +mf_ext_open(FName, MaxB, MaxF, Repair, Mode, Head, Version) -> + {First, Sz, TotSz, NFiles} = read_index_file(Repair, FName, MaxF), + write_size_file(Mode, FName, MaxB, MaxF, Version), + NewMaxF = if + NFiles > MaxF -> + {MaxF, NFiles}; + true -> + MaxF + end, + {ok, FdC, FileName, Lost, {NoItems, NoBytes}, CurB} = + ext_file_open(FName, First, 0, 0, Head, Repair, Mode), + CurCnt = Sz + NoItems - Lost, + {ok, #handle{filename = FName, maxB = MaxB, cur_name = FileName, + maxF = NewMaxF, cur_cnt = CurCnt, acc_cnt = -Sz, + curF = First, cur_fdc = FdC, firstPos = NoBytes, + curB = CurB, noFull = 0, accFull = 0}, + TotSz + CurCnt}. + +%% -> {ok, handle(), Lost} +%% | {error, Error, handle()} +%% | throw(FatalError) +%% Fatal errors should always terminate the log. +mf_ext_inc(Handle, Head) -> + #handle{filename = FName, cur_cnt = CurCnt, cur_name = FileName, + acc_cnt = AccCnt, curF = CurF, maxF = MaxF, cur_fdc = CurFdC, + noFull = NoFull} = Handle, + case catch wrap_ext_log(FName, CurF, MaxF, CurCnt, Head) of + {NewF, NewMaxF, NewFdC, NewFileName, Nh, FirstPos, Lost} -> + Handle1 = Handle#handle{cur_fdc = NewFdC, curF = NewF, + cur_name = NewFileName, + cur_cnt = Nh, acc_cnt = AccCnt + CurCnt, + maxF = NewMaxF, firstPos = FirstPos, + curB = FirstPos, noFull = NoFull + 1}, + case catch fclose(CurFdC, FileName) of + ok -> + {ok, Handle1, Lost}; + Error -> % Error in the last file, new file opened. + {error, Error, Handle1} + end; + Error -> + {error, Error, Handle} + end. + +%% -> {ok, handle(), Logged, Lost, NoWraps} | {ok, handle(), Logged} +%% | {error, Error, handle(), Logged, Lost} + +%% The returned handle is not always valid - +%% something may have been written before things went wrong. +mf_ext_log(Handle, Bins, Head) -> + mf_ext_log(Handle, Bins, Head, 0, []). + +mf_ext_log(Handle, [], _Head, No, []) -> + {ok, Handle, No}; +mf_ext_log(Handle, [], _Head, No, Wraps0) -> + Wraps = reverse(Wraps0), + {ok, Handle, No, sum(Wraps), Wraps}; +mf_ext_log(Handle, Bins, Head, No0, Wraps) -> + #handle{curB = CurB, maxB = MaxB, cur_name = FileName, cur_fdc = CurFdC, + firstPos = FirstPos0, cur_cnt = CurCnt} = Handle, + {FirstBins, LastBins, NoBytes, N} = + ext_split_bins(CurB, MaxB, FirstPos0, Bins), + case FirstBins of + [] -> + #handle{filename = FName, curF = CurF, maxF = MaxF, + acc_cnt = AccCnt, noFull = NoFull} = Handle, + case catch wrap_ext_log(FName, CurF, MaxF, CurCnt, Head) of + {NewF, NewMaxF, NewFdC, NewFileName, Nh, FirstPos, Lost} -> + Handle1 = Handle#handle{cur_fdc = NewFdC, curF = NewF, + cur_cnt = Nh, + cur_name = NewFileName, + acc_cnt = AccCnt + CurCnt, + maxF = NewMaxF, + curB = FirstPos, + firstPos = FirstPos, + noFull = NoFull + 1}, + case catch fclose(CurFdC, FileName) of + ok -> + mf_ext_log(Handle1, Bins, Head, No0 + Nh, + [Lost | Wraps]); + Error -> + Lost1 = Lost + sum(Wraps), + {error, Error, Handle1, No0 + Nh, Lost1} + end; + Error -> + {error, Error, Handle, No0, sum(Wraps)} + end; + _ -> + case fwrite(CurFdC, FileName, FirstBins, NoBytes) of + {ok, NewCurFdC} -> + Handle1 = Handle#handle{cur_fdc = NewCurFdC, + curB = CurB + NoBytes, + cur_cnt = CurCnt + N}, + mf_ext_log(Handle1, LastBins, Head, No0 + N, Wraps); + {Error, NewCurFdC} -> + Handle1 = Handle#handle{cur_fdc = NewCurFdC}, + {error, Error, Handle1, No0, sum(Wraps)} + end + end. + +wrap_ext_log(FName, CurF, MaxF, CurCnt, Head) -> + {NewF, NewMaxF} = inc_wrap(FName, CurF, MaxF), + {ok, NewFdC, NewFileName, Lost, {Nh, FirstPos}, _FileSize} = + ext_file_open(FName, NewF, CurF, CurCnt, Head), + {NewF, NewMaxF, NewFdC, NewFileName, Nh, FirstPos, Lost}. + +%% -> ok | throw(FileError) +mf_ext_close(#handle{filename = FName, curF = CurF, + cur_fdc = CurFdC, cur_cnt = CurCnt}, Mode) -> + Res = (catch fclose(CurFdC, FName)), + write_index_file(Mode, FName, CurF, CurF, CurCnt), + Res. + +%% -> {ok, handle()} | throw(FileError) +change_size_wrap(Handle, {NewMaxB, NewMaxF}, Version) -> + FName = Handle#handle.filename, + {_MaxB, MaxF} = get_wrap_size(Handle), + write_size_file(read_write, FName, NewMaxB, NewMaxF, Version), + if + NewMaxF > MaxF -> + remove_files(FName, MaxF + 1, NewMaxF), + {ok, Handle#handle{maxB = NewMaxB, maxF = NewMaxF}}; + NewMaxF < MaxF -> + {ok, Handle#handle{maxB = NewMaxB, maxF = {NewMaxF, MaxF}}}; + true -> + {ok, Handle#handle{maxB = NewMaxB, maxF = NewMaxF}} + end. + +%%----------------------------------------------------------------- +%% Misc functions +%%----------------------------------------------------------------- +%% -> {ok, FdC, FileName, Lost, HeadSize, FileSize} | throw(Error) +int_file_open(FName, NewFile, OldFile, OldCnt, Head) -> + Repair = truncate, Mode = read_write, + int_file_open(FName, NewFile, OldFile, OldCnt, Head, Repair, Mode). + +%% -> {ok, FdC, FileName, Lost, HeadSize, FileSize} +%% | {repaired, FdC, FileName, Rec, Bad, FileSize} +%% | throw(Error) +int_file_open(FName, NewFile, OldFile, OldCnt, Head, Repair, Mode) -> + N = add_ext(FName, NewFile), + case int_open(N, Repair, Mode, Head) of + {ok, {_Alloc, FdC, HeadSize, FileSize}} -> + Lost = write_index_file(Mode, FName, NewFile, OldFile, OldCnt), + {ok, FdC, N, Lost, HeadSize, FileSize}; + {repaired, FdC, Recovered, BadBytes, FileSize} -> + write_index_file(Mode, FName, NewFile, OldFile, OldCnt), + {repaired, FdC, N, Recovered, BadBytes, FileSize} + end. + +%% -> {ok, FdC, FileName, Lost, HeadSize, FileSize} | throw(Error) +ext_file_open(FName, NewFile, OldFile, OldCnt, Head) -> + Repair = truncate, Mode = read_write, + ext_file_open(FName, NewFile, OldFile, OldCnt, Head, Repair, Mode). + +ext_file_open(FName, NewFile, OldFile, OldCnt, Head, Repair, Mode) -> + FileName = add_ext(FName, NewFile), + {ok, {_Alloc, FdC, HeadSize, FileSize}} = + ext_open(FileName, Repair, Mode, Head), + Lost = write_index_file(Mode, FName, NewFile, OldFile, OldCnt), + {ok, FdC, FileName, Lost, HeadSize, FileSize}. + +%%----------------------------------------------------------------- +%% The old file format for index file (CurFileNo > 0), Version 0: +%% +%% CurFileNo SizeFile1 SizeFile2 ... SizeFileN +%% 1 byte 4 bytes 4 bytes 4 bytes +%% +%% The new file format for index file (NewFormat = 0), version 1: +%% +%% NewFormat CurFileNo SizeFile1 SizeFile2 ... SizeFileN +%% 1 byte 4 bytes 4 bytes 4 bytes +%% +%% The current file format for index file (sizes in bytes), version 2: +%% +%% 0 (1) 0 (4) FileFormatVersion (1) CurFileNo (4) SizeFile1 (8) ... +%% +%% (SizeFileI refers to number of items on the log file.) +%%----------------------------------------------------------------- + +-define(index_file_name(F), add_ext(F, "idx")). + +read_index_file(truncate, FName, MaxF) -> + remove_files(FName, 2, MaxF), + file:delete(?index_file_name(FName)), + {1, 0, 0, 0}; +read_index_file(_, FName, _MaxF) -> + read_index_file(FName). + +%% Used by wrap_log_reader. +%% -> {CurFileNo, CurFileSz, TotSz, NoFiles} | throw(FileError) +%% where TotSz does not include CurFileSz. + +read_index_file(FName) -> + FileName = ?index_file_name(FName), + case open_read(FileName) of + {ok, Fd} -> + R = case file:read(Fd, ?MAX_CHUNK_SIZE) of + {ok, <<0, 0:32, Version, CurF:32, Tail/binary>>} + when Version =:= ?VERSION, + 0 < CurF, CurF < ?MAX_FILES -> + parse_index(CurF, Version, 1, Tail, Fd, 0, 0, 0); + {ok, <<0, CurF:32, Tail/binary>>} + when 0 < CurF, CurF < ?MAX_FILES -> + parse_index(CurF, 1, 1, Tail, Fd, 0, 0, 0); + {ok, <<CurF, Tail/binary>>} when 0 < CurF -> + parse_index(CurF, 1, 1, Tail, Fd, 0, 0, 0); + _ErrorOrEof -> + {1, 0, 0, 0} + end, + file:close(Fd), + R; + _Error -> + {1, 0, 0, 0} + end. + +parse_index(CurF, V, CurF, <<CurSz:64, Tail/binary>>, Fd, _, TotSz, NFiles) + when V =:= ?VERSION -> + parse_index(CurF, V, CurF+1, Tail, Fd, CurSz, TotSz, NFiles+1); +parse_index(CurF, V, N, <<Sz:64, Tail/binary>>, Fd, CurSz, TotSz, NFiles) + when V =:= ?VERSION -> + parse_index(CurF, V, N+1, Tail, Fd, CurSz, TotSz + Sz, NFiles+1); +parse_index(CurF, V, CurF, <<CurSz:32, Tail/binary>>, Fd, _, TotSz, NFiles) + when V < ?VERSION -> + parse_index(CurF, V, CurF+1, Tail, Fd, CurSz, TotSz, NFiles+1); +parse_index(CurF, V, N, <<Sz:32, Tail/binary>>, Fd, CurSz, TotSz, NFiles) + when V < ?VERSION -> + parse_index(CurF, V, N+1, Tail, Fd, CurSz, TotSz + Sz, NFiles+1); +parse_index(CurF, V, N, B, Fd, CurSz, TotSz, NFiles) -> + case file:read(Fd, ?MAX_CHUNK_SIZE) of + eof when 0 =:= byte_size(B) -> + {CurF, CurSz, TotSz, NFiles}; + {ok, Bin} -> + NewB = list_to_binary([B, Bin]), + parse_index(CurF, V, N, NewB, Fd, CurSz, TotSz, NFiles); + _ErrorOrEof -> + {1, 0, 0, 0} + end. + +%% Returns: Number of lost items (if an old file was truncated) +%% -> integer() | throw(FileError) +write_index_file(read_only, _FName, _NewFile, _OldFile, _OldCnt) -> + 0; +write_index_file(read_write, FName, NewFile, OldFile, OldCnt) -> + FileName = ?index_file_name(FName), + case open_update(FileName) of + {ok, Fd} -> + {Offset, SzSz} = + case file:read(Fd, 6) of + eof -> + Bin = <<0, 0:32, ?VERSION, NewFile:32>>, + fwrite_close2(Fd, FileName, Bin), + {10, 8}; + {ok, <<0, 0:32, _Version>>} -> + pwrite_close2(Fd, FileName, 6, <<NewFile:32>>), + {10, 8}; + {ok, <<0, _/binary>>} -> + pwrite_close2(Fd, FileName, 1, <<NewFile:32>>), + {5, 4}; + {ok, <<_,_/binary>>} -> + %% Very old format, convert to the latest format! + case file:read_file(FileName) of + {ok, <<_CurF, Tail/binary>>} -> + position_close2(Fd, FileName, bof), + Bin = <<0, 0:32, ?VERSION, NewFile:32>>, + NewTail = to_8_bytes(Tail, [], FileName, Fd), + fwrite_close2(Fd, FileName, [Bin | NewTail]), + {10, 8}; + Error -> + file_error_close(Fd, FileName, Error) + end; + Error -> + file_error_close(Fd, FileName, Error) + end, + + NewPos = Offset + (NewFile - 1)*SzSz, + OldCntBin = <<OldCnt:SzSz/unit:8>>, + if + OldFile > 0 -> + R = file:pread(Fd, NewPos, SzSz), + OldPos = Offset + (OldFile - 1)*SzSz, + pwrite_close2(Fd, FileName, OldPos, OldCntBin), + file:close(Fd), + case R of + {ok, <<Lost:SzSz/unit:8>>} -> Lost; + {ok, _} -> + throw({error, {invalid_index_file, FileName}}); + eof -> 0; + Error2 -> file_error(FileName, Error2) + end; + true -> + pwrite_close2(Fd, FileName, NewPos, OldCntBin), + file:close(Fd), + 0 + end; + E -> + file_error(FileName, E) + end. + +to_8_bytes(<<N:32,T/binary>>, NT, FileName, Fd) -> + to_8_bytes(T, [NT | <<N:64>>], FileName, Fd); +to_8_bytes(B, NT, _FileName, _Fd) when byte_size(B) =:= 0 -> + NT; +to_8_bytes(_B, _NT, FileName, Fd) -> + file:close(Fd), + throw({error, {invalid_index_file, FileName}}). + +%% -> ok | throw(FileError) +index_file_trunc(FName, N) -> + FileName = ?index_file_name(FName), + case open_update(FileName) of + {ok, Fd} -> + case file:read(Fd, 6) of + eof -> + file:close(Fd), + ok; + {ok, <<0, 0:32, Version>>} when Version =:= ?VERSION -> + truncate_index_file(Fd, FileName, 10, 8, N); + {ok, <<0, _/binary>>} -> + truncate_index_file(Fd, FileName, 5, 4, N); + {ok, <<_, _/binary>>} -> % cannot happen + truncate_index_file(Fd, FileName, 1, 4, N); + Error -> + file_error_close(Fd, FileName, Error) + end; + Error -> + file_error(FileName, Error) + end. + +truncate_index_file(Fd, FileName, Offset, N, SzSz) -> + Pos = Offset + N*SzSz, + case Pos > file_size(FileName) of + true -> + file:close(Fd); + false -> + truncate_at_close2(Fd, FileName, {bof, Pos}), + file:close(Fd) + end, + ok. + +print_index_file(File) -> + io:format("-- Index begin --~n"), + case file:read_file(File) of + {ok, <<0, 0:32, Version, CurF:32, Tail/binary>>} + when Version =:= ?VERSION, 0 < CurF, CurF < ?MAX_FILES -> + io:format("cur file: ~w~n", [CurF]), + loop_index(1, Version, Tail); + {ok, <<0, CurF:32, Tail/binary>>} when 0 < CurF, CurF < ?MAX_FILES -> + io:format("cur file: ~w~n", [CurF]), + loop_index(1, 1, Tail); + {ok, <<CurF, Tail/binary>>} when 0 < CurF -> + io:format("cur file: ~w~n", [CurF]), + loop_index(1, 1, Tail); + _Else -> + ok + end, + io:format("-- end --~n"). + +loop_index(N, V, <<Sz:64, Tail/binary>>) when V =:= ?VERSION -> + io:format(" ~p items: ~w~n", [N, Sz]), + loop_index(N+1, V, Tail); +loop_index(N, V, <<Sz:32, Tail/binary>>) when V < ?VERSION -> + io:format(" ~p items: ~w~n", [N, Sz]), + loop_index(N+1, V, Tail); +loop_index(_, _, _) -> + ok. + +-define(size_file_name(F), add_ext(F, "siz")). + +%% Version 0: no size file +%% Version 1: <<MaxSize:32, MaxFiles:32>> +%% Version 2: <<Version:8, MaxSize:64, MaxFiles:32>> + +%% -> ok | throw(FileError) +write_size_file(read_only, _FName, _NewSize, _NewMaxFiles, _Version) -> + ok; +write_size_file(read_write, FName, NewSize, NewMaxFiles, Version) -> + FileName = ?size_file_name(FName), + Bin = if + Version =:= ?VERSION -> + <<Version, NewSize:64, NewMaxFiles:32>>; + true -> + <<NewSize:32, NewMaxFiles:32>> + end, + case file:write_file(FileName, Bin) of + ok -> + ok; + E -> + file_error(FileName, E) + end. + +%% -> {NoBytes, NoFiles}. +read_size_file(FName) -> + {Size,_Version} = read_size_file_version(FName), + Size. + +%% -> {{NoBytes, NoFiles}, Version}, Version = integer() | undefined +read_size_file_version(FName) -> + case file:read_file(?size_file_name(FName)) of + {ok, <<Version, Size:64, MaxFiles:32>>} when Version =:= ?VERSION -> + {{Size, MaxFiles}, Version}; + {ok, <<Size:32, MaxFiles:32>>} -> + {{Size, MaxFiles}, 1}; + _ -> + %% The oldest version too... + {{0, 0}, ?VERSION} + end. + +conv({More, Terms}, FileNo) when is_record(More, continuation) -> + Cont = More#continuation{pos = {FileNo, More#continuation.pos}}, + {Cont, Terms}; +conv({More, Terms, Bad}, FileNo) when is_record(More, continuation) -> + Cont = More#continuation{pos = {FileNo, More#continuation.pos}}, + {Cont, Terms, Bad}; +conv(Other, _) -> + Other. + +find_first_file(#handle{filename = FName, curF = CurF, maxF = MaxF}) -> + fff(FName, inc(CurF, MaxF), CurF, MaxF). + +fff(_FName, CurF, CurF, _MaxF) -> CurF; +fff(FName, MaybeFirstF, CurF, MaxF) -> + N = add_ext(FName, MaybeFirstF), + case file:read_file_info(N) of + {ok, _} -> MaybeFirstF; + _ -> fff(FName, inc(MaybeFirstF, MaxF), CurF, MaxF) + end. + +%% -> {iolist(), LastBins, NoBytes, NoTerms} +ext_split_bins(CurB, MaxB, FirstPos, Bins) -> + MaxBs = MaxB - CurB, IsFirst = CurB =:= FirstPos, + ext_split_bins(MaxBs, IsFirst, [], Bins, 0, 0). + +ext_split_bins(MaxBs, IsFirst, First, [X | Last], Bs, N) -> + NBs = Bs + byte_size(X), + if + NBs =< MaxBs -> + ext_split_bins(MaxBs, IsFirst, [First | X], Last, NBs, N+1); + IsFirst, First =:= [] -> + % To avoid infinite loop - we allow the file to be + % too big if it's just one item on the file. + {[X], Last, NBs, N+1}; + true -> + {First, [X | Last], Bs, N} + end; +ext_split_bins(_, _, First, [], Bs, N) -> + {First, [], Bs, N}. + +%% -> {iolist(), LastBins, NoBytes, NoTerms} +int_split_bins(CurB, MaxB, FirstPos, Bins) -> + MaxBs = MaxB - CurB, IsFirst = CurB =:= FirstPos, + int_split_bins(MaxBs, IsFirst, [], Bins, 0, 0). + +int_split_bins(MaxBs, IsFirst, First, [X | Last], Bs, N) -> + Sz = byte_size(X), + NBs = Bs + Sz + ?HEADERSZ, + BSz = <<Sz:?SIZESZ/unit:8>>, + XB = case Sz < ?MIN_MD5_TERM of + true -> + [BSz, ?BIGMAGICHEAD | X]; + false -> + MD5 = erlang:md5(BSz), + [BSz, ?BIGMAGICHEAD, MD5 | X] + end, + if + NBs =< MaxBs -> + int_split_bins(MaxBs, IsFirst, [First | XB], Last, NBs, N+1); + IsFirst, First =:= [] -> + % To avoid infinite loop - we allow the file to be + % too big if it's just one item on the file. + {[XB], Last, NBs, N+1}; + true -> + {First, [X | Last], Bs, N} + end; +int_split_bins(_, _, First, [], Bs, N) -> + {First, [], Bs, N}. + +%% -> {NewCurrentFileNo, MaxFilesToBe} | throw(FileError) +inc_wrap(FName, CurF, MaxF) -> + case MaxF of + %% Number of max files has changed + {NewMaxF, OldMaxF} -> + if + CurF >= NewMaxF -> + %% We are at or above the new number of files + remove_files(FName, CurF + 1, OldMaxF), + if + CurF > NewMaxF -> + %% The change was done while the current file was + %% greater than the new number of files. + %% The index file is not trunctated here, since + %% writing the index file while opening the file + %% with index 1 will write the value for the file + %% with extension CurF as well. Next time the + %% limit is reached, the index file will be + %% truncated. + {1, {NewMaxF, CurF}}; + true -> + %% The change was done while the current file was + %% less than the new number of files. + %% Remove the files from the index file too + index_file_trunc(FName, NewMaxF), + {1, NewMaxF} + end; + true -> + %% We haven't reached the new limit yet + NewFt = inc(CurF, NewMaxF), + {NewFt, MaxF} + end; + MaxF -> + %% Normal case. + NewFt = inc(CurF, MaxF), + {NewFt, MaxF} + end. + +inc(N, {_NewMax, OldMax}) -> inc(N, OldMax, 1); +inc(N, Max) -> inc(N, Max, 1). + +inc(N, Max, Step) -> + Nx = (N + Step) rem Max, + if + Nx > 0 -> Nx; + true -> Nx + Max + end. + + +file_size(Fname) -> + {ok, Fi} = file:read_file_info(Fname), + Fi#file_info.size. + +%% -> ok | throw(FileError) +%% Tries to remove each file with name FName.I, N<=I<=Max. +remove_files(FName, N, Max) -> + remove_files(FName, N, Max, ok). + +remove_files(_FName, N, Max, ok) when N > Max -> + ok; +remove_files(_FName, N, Max, {FileName, Error}) when N > Max -> + file_error(FileName, Error); +remove_files(FName, N, Max, Reply) -> + FileName = add_ext(FName, N), + NewReply = case file:delete(FileName) of + ok -> Reply; + {error, enoent} -> Reply; + Error -> {FileName, Error} + end, + remove_files(FName, N + 1, Max, NewReply). + +%% -> {MaxBytes, MaxFiles} +get_wrap_size(#handle{maxB = MaxB, maxF = MaxF}) -> + case MaxF of + {NewMaxF,_} -> {MaxB, NewMaxF}; + MaxF -> {MaxB, MaxF} + end. + +add_ext(Name, Ext) -> + concat([Name, ".", Ext]). + +open_read(FileName) -> + file:open(FileName, [raw, binary, read]). + +open_update(FileName) -> + file:open(FileName, [raw, binary, read, write]). + +open_truncate(FileName) -> + file:open(FileName, [raw, binary, write]). + +%%% Functions that access files, and throw on error. + +-define(MAX, 16384). % bytes +-define(TIMEOUT, 2000). % ms + +%% -> {Reply, cache()}; Reply = ok | Error +fwrite(#cache{c = []} = FdC, _FN, B, Size) -> + case get(write_cache_timer_is_running) of + true -> + ok; + _ -> + put(write_cache_timer_is_running, true), + erlang:send_after(?TIMEOUT, self(), {self(), write_cache}) + end, + {ok, FdC#cache{sz = Size, c = B}}; +fwrite(#cache{sz = Sz, c = C} = FdC, _FN, B, Size) when Sz < ?MAX -> + {ok, FdC#cache{sz = Sz+Size, c = [C | B]}}; +fwrite(#cache{fd = Fd, c = C}, FileName, B, _Size) -> + write_cache(Fd, FileName, [C | B]). + +fwrite_header(Fd, B, Size) -> + {ok, #cache{fd = Fd, sz = Size, c = B}}. + +%% -> {NewFdC, Reply}; Reply = ok | Error +pread(#cache{fd = Fd, c = C}, FileName, Position, MaxBytes) -> + Reply = write_cache(Fd, FileName, C), + case Reply of + {ok, NewFdC} -> + case file:pread(Fd, Position, MaxBytes) of + {error, Error} -> + {NewFdC, catch file_error(FileName, {error, Error})}; + R -> + {NewFdC, R} + end; + {Error, NewFdC} -> + {NewFdC, Error} + end. + +%% -> {ok, cache(), Pos} | {Error, cache()} +position(#cache{fd = Fd, c = C}, FileName, Pos) -> + Reply = write_cache(Fd, FileName, C), + case Reply of + {ok, NewFdC} -> + case position2(Fd, FileName, Pos) of + {ok, Loc} -> + {ok, NewFdC, Loc}; + Error -> + {Error, NewFdC} + end; + _Error -> + Reply + end. + +position_close(#cache{fd = Fd, c = C}, FileName, Pos) -> + NewFdC = write_cache_close(Fd, FileName, C), + {ok, Loc} = position_close2(Fd, FileName, Pos), + {NewFdC, Loc}. + +fsync(#cache{fd = Fd, c = C}, FileName) -> + Reply = write_cache(Fd, FileName, C), + case Reply of + {ok, NewFdC} -> + case file:sync(Fd) of + ok -> + Reply; + Error -> + {catch file_error(FileName, Error), NewFdC} + end; + _Error -> + Reply + end. + +%% -> {Reply, NewFdC}; Reply = ok | Error +truncate_at(FdC, FileName, Pos) -> + case position(FdC, FileName, Pos) of + {ok, NewFdC, _Pos} -> + case file:truncate(NewFdC#cache.fd) of + ok -> + {ok, NewFdC}; + Error -> + {catch file_error(FileName, Error), NewFdC} + end; + Reply -> + Reply + end. + +fwrite_close2(Fd, FileName, B) -> + case file:write(Fd, B) of + ok -> ok; + Error -> file_error_close(Fd, FileName, Error) + end. + +pwrite_close2(Fd, FileName, Position, B) -> + case file:pwrite(Fd, Position, B) of + ok -> ok; + Error -> file_error(FileName, {error, Error}) + end. + +position2(Fd, FileName, Pos) -> + case file:position(Fd, Pos) of + {error, Error} -> catch file_error(FileName, {error, Error}); + OK -> OK + end. + +position_close2(Fd, FileName, Pos) -> + case file:position(Fd, Pos) of + {error, Error} -> file_error_close(Fd, FileName, {error, Error}); + OK -> OK + end. + +truncate_at_close2(Fd, FileName, Pos) -> + position_close2(Fd, FileName, Pos), + case file:truncate(Fd) of + ok -> ok; + Error -> file_error_close(Fd, FileName, Error) + end. + +fclose(#cache{fd = Fd, c = C}, FileName) -> + %% The cache is empty if the file was opened in read_only mode. + write_cache_close(Fd, FileName, C), + file:close(Fd). + +%% -> {Reply, #cache{}}; Reply = ok | Error +write_cache(Fd, _FileName, []) -> + {ok, #cache{fd = Fd}}; +write_cache(Fd, FileName, C) -> + case file:write(Fd, C) of + ok -> {ok, #cache{fd = Fd}}; + Error -> {catch file_error(FileName, Error), #cache{fd = Fd}} + end. + +-spec write_cache_close(fd(), file:filename(), iodata()) -> #cache{}. % | throw(Error) + +write_cache_close(Fd, _FileName, []) -> + #cache{fd = Fd}; +write_cache_close(Fd, FileName, C) -> + case file:write(Fd, C) of + ok -> #cache{fd = Fd}; + Error -> file_error_close(Fd, FileName, Error) + end. + +-spec file_error(file:filename(), {'error', atom()}) -> no_return(). + +file_error(FileName, {error, Error}) -> + throw({error, {file_error, FileName, Error}}). + +-spec file_error_close(fd(), file:filename(), {'error', atom()}) -> no_return(). + +file_error_close(Fd, FileName, {error, Error}) -> + file:close(Fd), + throw({error, {file_error, FileName, Error}}). |