Current section
Files
Jump to
Current section
Files
src/mstore.erl
%%%-------------------------------------------------------------------
%%% @author Heinz Nikolaus Gies <heinz@licenser.net>
%%% @copyright (C) 2014, Heinz Nikolaus Gies
%%% @doc
%%%
%%% @end
%%% Created : 4 Jun 2014 by Heinz Nikolaus Gies <heinz@licenser.net>
%%%-------------------------------------------------------------------
%% @doc An mstore instance is per directory, consisting of multiple `mstore'
%% and `idx' files. Data points are stored according to their timestamp, and
%% stores typically retain up to one week's (depends on `file_size') worth of
%% data. One master `mstore' file will be created for all metric paths in the
%% directory, and is used for coverage queries.
%%
%% Index (`idx') files are keyed on and offset position and consist of metric
%% names. These metrics are used to calculate the relative position of a
%% metric's values in the `mstore' storage file.
%%
%% The format of an `idx' file is as follows, where `M..' are metric paths:
%% +--------+-----------+----------+----+----------+----+-----+----------+----+
%% | Offset | File Size | Size(M1) | M1 | Size(M2) | M2 | ... | Size(Mn) | Mn |
%% +--------+-----------+----------+----+----------+----+-----+----------+----+
%% |---- Pos 0 ----|---- Pos 2 ----| |---- Pos n ----|
%%
%% Metric names are appended to the index the first time that they are
%% encountered.
%%
%% Storage (mstore) files may store values for more than one metric at a time,
%% and is segmented sequentially. The position for a value within a
%% segment is calculated as follows: P = Base + ((TS - Offset) * Data size),
%% where Base is a function of the metric's position in the index:
%% +-------------+-----+------------------------------------------------+
%% | Pos(M1) = 0 | --> | Base = 0 X File size X Data size = 0 |
%% | Pos(M2) = 1 | --> | Base = 1 X File size X Data size = 4838400 |
%% | ... | | |
%% | Pos(Mn) = N | --> | Base = N X File size X Data size = N * 4838400 |
%% +-------------+-----+------------------------------------------------+
%%
%% Only one process should open the mstore for writing at a given time. At any
%% moment, up to two `mstore' files may be open for writing by the mstore.
%% These files represent the current and previously written to files e.g. the
%% current and previous week.
-module(mstore).
-export([
new/2,
open/1,
open/2,
close/1,
delete/1,
delete/2,
reindex/1,
get/4,
open_mfile/1,
read_idx/1,
put/4,
metrics/1,
count/1,
file_size/1,
fold/3, fold/4,
make_splits/3]).
-export_type([mstore/0]).
-include_lib("mmath/include/mmath.hrl").
-include("mstore.hrl").
-define(SIZE_TYPE, unsigned-integer).
-define(VERSION, 3).
-define(OPTS, [raw, binary]).
-define(MFILE_VER, 1).
-ifdef(TEST).
-include_lib("eunit/include/eunit.hrl").
-compile(export_all).
-endif.
-record(mfile, {
name,
file,
offset,
size,
index = btrie:new(),
next = 0
}).
-record(mstore, {
size,
max_files = 2,
files=[],
dir,
data_size = ?DATA_SIZE,
metrics = btrie:new()
}).
-opaque mstore() :: #mstore{}.
-type open_opt() ::
{max_files, pos_integer()}.
-type new_opt() ::
{file_size, pos_integer()} |
{data_size, pos_integer()}.
-type fold_fun() :: fun((Metric :: binary(),
Offset :: non_neg_integer(),
Data :: binary(), AccIn :: any()) -> AccOut :: any()).
%%--------------------------------------------------------------------
%% Public API
%%--------------------------------------------------------------------
%%--------------------------------------------------------------------
%% @doc
%% Creates a new set or opens an existing one.
%% @end
%%--------------------------------------------------------------------
-spec new(Dir :: string() | binary(),
Opts :: [new_opt() | open_opt()]) ->
{ok, mstore()} |
{error, filesize_missmatch}.
new(Dir, Opts) when (is_list(Dir) orelse is_binary(Dir)) , is_list(Opts) ->
{file_size, FileSize} = proplists:lookup(file_size, Opts),
DataSize = proplists:get_value(data_size, Opts, ?DATA_SIZE),
case new(FileSize, DataSize, Dir) of
{ok, M} ->
{ok, apply_opts(M, Opts)};
E ->
E
end.
new(FileSize, DataSize, Dir) when
is_integer(FileSize), FileSize > 0,
is_binary(Dir) ->
new(FileSize, DataSize, binary_to_list(Dir));
new(FileSize, DataSize, Dir) ->
IdxFile = [Dir | "/mstore"],
case open_mfile(IdxFile) of
{ok, F, _IS, _Metrics} when F =/= FileSize ->
{error, filesize_missmatch};
{ok, _F, IS, _Metrics} when IS =/= DataSize ->
{error, itemsize_missmatch};
{ok, F, IS, Metrics} ->
{ok, #mstore{size=F, dir=Dir, metrics=Metrics, data_size=IS}};
_ ->
file:make_dir(Dir),
MStore = #mstore{size=FileSize, dir=Dir,
data_size=DataSize},
ok = file:write_file(IdxFile, index_header(MStore)),
{ok, MStore}
end.
%%--------------------------------------------------------------------
%% @doc
%% Opens an existing set.
%% @end
%%--------------------------------------------------------------------
-spec open(Dir :: string()) -> {ok, mstore()} | {error, not_found}.
open(Dir) ->
open(Dir, []).
-spec open(Dir :: string(), [open_opt()]) ->
{ok, mstore()} | {error, not_found}.
open(Dir, Opts) ->
case open_mfile([Dir | "/mstore"]) of
{ok, FileSize, DataSize, Metrics} ->
M = #mstore{size=FileSize, dir=Dir, data_size = DataSize,
metrics=Metrics},
{ok, apply_opts(M, Opts)};
E ->
E
end.
%%--------------------------------------------------------------------
%% @doc
%% Closes a metric set.
%% @end
%%--------------------------------------------------------------------
-spec close(mstore()) -> ok.
close(#mstore{files=Files}) ->
[close_store(S) || {_, S} <- Files],
ok.
%%--------------------------------------------------------------------
%% @doc
%% Recursively deletes a metric set.
%% @end
%%--------------------------------------------------------------------
-spec delete(mstore()) -> ok | {error, atom()}.
delete(MStore = #mstore{dir=Dir}) ->
close(MStore),
case file:list_dir(Dir) of
{ok, Files} ->
Files1 = [[Dir, $/ | File] || File <- Files],
[file:delete(F) || F <- Files1],
file:del_dir(Dir);
{error, enoent} ->
ok
end.
%%--------------------------------------------------------------------
%% @doc
%% Deletes entries before *Before*, not exact since it's rounded to
%% the next file.
%% @end
%%--------------------------------------------------------------------
-spec delete(mstore(), pos_integer()) -> {ok, mstore()}.
delete(MStore = #mstore{size = S, dir = Dir, files = Files}, Before) ->
Before1 = ((Before div S) - 1) * S,
Chunks = chunks(Dir, ".mstore"),
Chunks1 = lists:takewhile(fun(C) ->
C < Before1
end, Chunks),
case Chunks1 of
[] ->
{ok, MStore};
_ ->
[close_store(F) || {_, F} <- Files],
[begin
F = [Dir, $/, integer_to_list(C), $.],
file:delete([F | "mstore"]),
file:delete([F | "idx"])
end || C <- Chunks1],
reindex(MStore#mstore{files = []})
end.
%%--------------------------------------------------------------------
%% @doc
%% Reads data from a set.
%% @end
%%--------------------------------------------------------------------
-spec get(
mstore(),
binary(),
non_neg_integer(),
pos_integer()) ->
{'error',atom()} |
{ok, binary()}.
get(#mstore{size=S, files=FS, dir=Dir, data_size=DataSize},
Metric, Time, Count) when
is_binary(Metric),
is_integer(Time), Time >= 0,
is_integer(Count), Count > 0 ->
?DT_READ_ENTRY(Metric, Time, Count),
Parts = make_splits(Time, Count, S),
R = do_get(S, FS, Dir, DataSize, Metric, Parts, <<>>),
?DT_READ_RETURN,
R.
%%--------------------------------------------------------------------
%% @doc
%% Folds over all the metrics in a store. Please be aware that the
%% data for a single metric is not guaranteed to be delivered at a time
%% it is split on non set values and file boundaries!
%% @end
%%--------------------------------------------------------------------
-spec fold(mstore(), FoldFun :: fold_fun(), AccIn :: any()) ->
AccOut :: any().
fold(MStore, FoldFun, Acc) ->
fold(MStore, FoldFun, infinity, Acc).
%%--------------------------------------------------------------------
%% @doc
%% Same as {@link: fold/3} but data is chunked at a maximum size of
%% *ChunkSize*.
%% @end
%%--------------------------------------------------------------------
-spec fold(mstore(), FoldFun :: fold_fun(), ChunkSize :: pos_integer() | infinity,
AccIn :: any()) ->
AccOut :: any().
fold(#mstore{dir=Dir, data_size = DataSize}, Fun, Chunk, Acc) ->
serialize_dir(Dir, DataSize, Fun, Chunk, Acc).
%%--------------------------------------------------------------------
%% @doc
%% Writes data into a mstore.
%% @end
%%--------------------------------------------------------------------
-spec put(mstore(), binary(), non_neg_integer(),
integer() | [integer()] | binary()) ->
mstore().
put(MStore, Metric, Time, [V0 | _] = Values)
when is_integer(V0) ->
put(MStore, Metric, Time, mmath_bin:from_list(Values));
put(MStore, Metric, Time, V) when is_integer(V) ->
put(MStore, Metric, Time, mmath_bin:from_list([V]));
put(MStore = #mstore{size=S, files=CurFiles, metrics=Ms, data_size = DataSize},
Metric, Time, Value)
when is_binary(Value),
is_integer(Time), Time >= 0,
is_binary(Metric) ->
?DT_WRITE_ENTRY(Metric, Time, mmath_bin:length(Value)),
Count = mmath_bin:length(Value),
Parts = make_splits(Time, Count, S),
Parts1 = [{B, round(C * DataSize)} || {B, C} <- Parts],
MStore1 = case btrie:is_key(Metric, Ms) of
true ->
MStore;
false ->
MStorex = MStore#mstore{metrics=btrie:store(Metric, Ms)},
ok = file:write_file([MStorex#mstore.dir | "/mstore"],
<<(byte_size(Metric)):16/integer, Metric/binary>>,
[read, append]),
MStorex
end,
CurFiles1 = do_put(MStore1, Metric, Parts1, Value, CurFiles),
?DT_WRITE_RETURN,
limit_files(MStore1#mstore{files = CurFiles1}).
%%--------------------------------------------------------------------
%% @doc
%% Returns a set of all metrics in the store.
%% @end
%%--------------------------------------------------------------------
-spec metrics(mstore()) -> btrie:btrie().
metrics(#mstore{metrics=M}) ->
M.
%%--------------------------------------------------------------------
%% @doc
%% Returns the chunk in the mstore, each metric row in a file is a
%% chunk. So the total is the sum of the metrics in each file.
%% @end
%%--------------------------------------------------------------------
-spec count(mstore()) -> non_neg_integer().
count(#mstore{dir=Dir}) ->
count(Dir);
count(Dir) when is_list(Dir) ->
{ok, Fs} = file:list_dir(Dir),
Fs1 = [re:split(F, "\\.", [{return, list}]) || F <- Fs],
Idxs = [[Dir, $/, I] || [I, "idx"] <- Fs1],
lists:foldl(fun(File, Cnt) ->
Cnt + count_idx([File | ".idx"])
end, 0, Idxs).
%%--------------------------------------------------------------------
%% @doc
%% Returns the File Size of a store
%% @end
%%--------------------------------------------------------------------
-spec file_size(mstore()) -> non_neg_integer().
file_size(#mstore{size = FileSize}) ->
FileSize.
%%--------------------------------------------------------------------
%% @doc
%% Utility function to split a continous data streams in chunks for a
%% given time range size.
%%
%% * Time - the offset of the first value.
%% * Count - the total number of values.
%% * Size - The size of the chunking.
%% @end
%%--------------------------------------------------------------------
-spec make_splits(Time :: non_neg_integer(),
Count :: pos_integer(),
Size :: pos_integer()) ->
[{StartTime :: non_neg_integer(),
ChunkSize :: pos_integer()}].
make_splits(Time, Count, Size) ->
make_splits(Time, Count, Size, []).
-spec reindex(mstore()) -> {ok, mstore()}.
reindex(MStore = #mstore{dir = Dir}) ->
IdxFileOld = [Dir | "/mstore"],
IdxFileNew = [Dir | "/mstore.new"],
Files = filelib:wildcard([Dir | "/*.idx"]),
{ok, IO} = file:open(IdxFileNew, [write | ?OPTS]),
ok = file:write(IO, index_header(MStore)),
Metrics = lists:foldl(fun (F, Set) ->
reindex_chunk(IO, F, Set)
end, btrie:new(), Files),
ok = file:close(IO),
NewIndex = << <<(byte_size(Metric)):16/integer, Metric/binary>>
|| Metric <- btrie:fetch_keys(Metrics) >>,
ok = file:write_file(IdxFileNew, <<(index_header(MStore))/binary,
NewIndex/binary>>),
ok = file:delete(IdxFileOld),
ok = file:rename(IdxFileNew, IdxFileOld),
{ok, MStore#mstore{metrics = Metrics}}.
%%====================================================================
%% Private functions.
%%====================================================================
apply_opts(MStore, []) ->
MStore;
apply_opts(MStore, [{max_files, N} | R]) when is_integer(N), N >= 0 ->
apply_opts(MStore#mstore{max_files = N}, R);
apply_opts(MStore, [_ | R]) ->
apply_opts(MStore, R).
chunks(Dir, Ext) ->
lists:sort([list_to_integer(filename:rootname(filename:basename(F))) ||
F <- filelib:wildcard([Dir | "/*" ++ Ext])]).
reindex_chunk(IO, File, Set) ->
fold_idx(fun({entry, M}, Acc) ->
ok = file:write(IO, <<(byte_size(M)):16/integer, M/binary>>),
btrie:store(M, Acc);
(_, Acc) ->
Acc
end, Set, File).
index_header(#mstore{size=FileSize, data_size=DataSize}) ->
<<?VERSION:16/?SIZE_TYPE, FileSize:64/?SIZE_TYPE, DataSize:64/?SIZE_TYPE>>.
make_splits(_Time, 0, _Size, Acc) ->
lists:reverse(Acc);
make_splits(Time, Count, Size, Acc) ->
Base = (Time div Size)*Size,
case Time - Base of
D when (D + Count) < Size ->
lists:reverse([{Time, Count} | Acc]);
D ->
Inc = Size-D,
make_splits(Time + Inc, Count - Inc, Size, [{Time, Inc} | Acc])
end.
store_metrics(#mfile{index=M}) ->
btrie:fetch_keys(M).
open_mfile(F) ->
Chunk = 4*1024,
case file:open(F, [read | ?OPTS]) of
{ok, IO} ->
case file:read(IO, Chunk) of
{ok, <<2:16/?SIZE_TYPE, FileSize:64/?SIZE_TYPE, R/binary>>} ->
Set = do_fold_idx(IO, Chunk,
fun({entry, M}, Acc) ->
btrie:store(M, Acc)
end, btrie:new(), R),
{ok, FileSize, 8, Set};
{ok, <<?VERSION:16/?SIZE_TYPE, FileSize:64/?SIZE_TYPE,
DataSize:64/?SIZE_TYPE, R/binary>>} ->
Set = do_fold_idx(IO, Chunk,
fun({entry, M}, Acc) ->
btrie:store(M, Acc)
end, btrie:new(), R),
{ok, FileSize, DataSize, Set};
{ok, _} ->
file:close(IO),
{error, invalid_file};
_ ->
file:close(IO),
{error, invalid_file}
end;
E ->
E
end.
do_put(_, _, [], <<>>, Files) ->
Files;
do_put(MStore = #mstore{size=S, data_size = DataSize}, Metric,
[{Time, Size} | R], InData,
[{FileBase, F} | FileRest]) when
((Time div S)*S) =:= FileBase ->
<<Data:Size/binary, DataRest/binary>> = InData,
{ok, F1} = write(F, DataSize, Metric, Time, Data),
do_put(MStore, Metric, R, DataRest, [{FileBase, F1} | FileRest]);
do_put(MStore = #mstore{size=S, data_size = DataSize}, Metric,
[{Time, Size} | R], InData,
[First, {FileBase, F}]) when
((Time div S)*S) =:= FileBase ->
<<Data:Size/binary, DataRest/binary>> = InData,
{ok, F1} = write(F, DataSize, Metric, Time, Data),
do_put(MStore, Metric, R, DataRest, [{FileBase, F1}, First]);
do_put(MStore = #mstore{size=S, dir=D, data_size = DataSize}, Metric,
[{Time, Size} | R], InData,
[First, {_Other, F}]) ->
<<Data:Size/binary, DataRest/binary>> = InData,
FileBase = (Time div S)*S,
close_store(F),
Base = [D, $/, integer_to_list(FileBase)],
{ok, F1} = open(Base, FileBase, S, write),
{ok, F2} = write(F1, DataSize, Metric, Time, Data),
do_put(MStore, Metric, R, DataRest, [{FileBase, F2}, First]);
do_put(MStore = #mstore{size=S, dir=D, data_size = DataSize}, Metric,
[{Time, Size} | R], InData,
Files) when length(Files) =< 2 ->
<<Data:Size/binary, DataRest/binary>> = InData,
FileBase = (Time div S)*S,
Base = [D, $/, integer_to_list(FileBase)],
{ok, F1} = open(Base, FileBase, S, write),
{ok, F2} = write(F1, DataSize, Metric, Time, Data),
do_put(MStore, Metric, R, DataRest, [{FileBase, F2} | Files]).
limit_files(MStore = #mstore{max_files = MaxFiles,
files = CurFiles = [First, {_FileBase, F} | R]})
when length(CurFiles) > MaxFiles ->
close_store(F),
limit_files(MStore#mstore{files = [First | R]});
limit_files(MStore = #mstore{max_files = MaxFiles,
files = CurFiles = [{_FileBase, F} | R]})
when length(CurFiles) > MaxFiles ->
close_store(F),
limit_files(MStore#mstore{files = R});
limit_files(MStore) ->
MStore.
do_get(_, _, _, _, _, [], Acc) ->
{ok, Acc};
do_get(S, [{FileBase, F} | _] = FS,
Dir, DataSize, Metric, [{Time, Count} | R], Acc)
when ((Time div S)*S) =:= FileBase ->
case read(F, DataSize, Metric, Time, Count) of
{ok, D} ->
Acc1 = <<Acc/binary, D/binary>>,
Acc2 = case mmath_bin:length(D) of
L when L < Count ->
Missing = Count - L,
<<Acc1/binary, (mmath_bin:empty(Missing))/binary>>;
_ ->
Acc1
end,
do_get(S, FS, Dir, DataSize, Metric, R, Acc2);
{error,not_found} ->
Acc1 = <<Acc/binary, (mmath_bin:empty(Count))/binary>>,
do_get(S, FS, Dir, DataSize, Metric, R, Acc1);
eof ->
Acc1 = <<Acc/binary, (mmath_bin:empty(Count))/binary>>,
do_get(S, FS, Dir, DataSize, Metric, R, Acc1);
E ->
E
end;
do_get(S,
[_, {FileBase, F}] = FS,
Dir, DataSize, Metric, [{Time, Count} | R], Acc)
when ((Time div S)*S) =:= FileBase ->
case read(F, DataSize, Metric, Time, Count) of
{ok, D} ->
Acc1 = <<Acc/binary, D/binary>>,
Acc2 = case mmath_bin:length(D) of
L when L < Count ->
Missing = Count - L,
<<Acc1/binary, (mmath_bin:empty(Missing))/binary>>;
_ ->
Acc1
end,
do_get(S, FS, Dir, DataSize, Metric, R, Acc2);
{error,not_found} ->
Acc1 = <<Acc/binary, (mmath_bin:empty(Count))/binary>>,
do_get(S, FS, Dir, DataSize, Metric, R, Acc1);
eof ->
Acc1 = <<Acc/binary, (mmath_bin:empty(Count))/binary>>,
do_get(S, FS, Dir, DataSize, Metric, R, Acc1);
E ->
E
end;
do_get(S, FS, Dir, DataSize, Metric, [{Time, Count} | R], Acc) ->
FileBase = (Time div S)*S,
Base = [Dir, "/", integer_to_list(FileBase)],
{ok, F} = open(Base, FileBase, S, read),
case read(F, DataSize, Metric, Time, Count) of
{ok, D} ->
close_store(F),
Acc1 = <<Acc/binary, D/binary>>,
Acc2 = case mmath_bin:length(D) of
L when L < Count ->
Missing = Count - L,
<<Acc1/binary, (mmath_bin:empty(Missing))/binary>>;
_ ->
Acc1
end,
do_get(S, FS, Dir, DataSize, Metric, R, Acc2);
{error,not_found} ->
close_store(F),
Acc1 = <<Acc/binary, (mmath_bin:empty(Count))/binary>>,
do_get(S, FS, Dir, DataSize, Metric, R, Acc1);
eof ->
close_store(F),
Acc1 = <<Acc/binary, (mmath_bin:empty(Count))/binary>>,
do_get(S, FS, Dir, DataSize, Metric, R, Acc1);
E ->
close_store(F),
E
end.
close_store(#mfile{file=F}) ->
file:close(F).
open(File, Offset, Size, Mode) ->
FileOpts = case Mode of
read ->
[read | ?OPTS];
write ->
[read, write | ?OPTS]
end,
IdxFile = [File | ".idx"],
case read_idx(IdxFile) of
{O, S, Idx, Next} when Offset =:= O,
Size =:= S ->
case file:open([File | ".mstore"], FileOpts) of
{ok, F} ->
{ok, #mfile{index=Idx, name=File, file=F, offset=Offset,
size=Size, next=Next}};
E ->
E
end;
{O, _, _, _} when Offset =/= O ->
{error, offset_missmatch};
{_, S, _, _} when Size =/= S ->
{error, size_missmatch};
_E ->
case file:open([File |".mstore"], [read, write | ?OPTS]) of
{ok, F} ->
M = #mfile{name=File, file=F, offset=Offset,
size=Size, next=0},
file:write_file(IdxFile,
<<Offset:64/?INT_TYPE, Size:64/?INT_TYPE>>),
{ok, M};
E ->
E
end
end.
open_store(File) ->
case read_idx([File | ".idx"]) of
{Offset, Size, Idx, Next} ->
case file:open([File | ".mstore"], [read | ?OPTS]) of
{ok, F} ->
{ok, #mfile{index=Idx, name=File, file=F, offset=Offset,
size=Size, next=Next}};
E ->
E
end;
E ->
E
end.
write(M=#mfile{offset=Offset, size=S},
DataSize, Metric, Position, Value)
when is_binary(Value),
Position >= Offset,
(Position - Offset) + (byte_size(Value) div DataSize) =< S ->
do_write(M, DataSize, Metric, Position, Value).
do_write(M=#mfile{offset=Offset, size=S, file=F, index=Idx},
DataSize, Metric, Position, Value) when
is_binary(Metric),
is_binary(Value) ->
%% As part of a write operation, both the index and mstore files may be
%% updated. There is no atomicity gaurantees, and a fault may occur after
%% one of the writes have been processed. Writing data to the index before
%% writing to the mstore ensures that offsets are calculated so as not to
%% cause data to overlap. In other words, data loss is acceptable, but not
%% data corruption.
{M1, Base} =
case btrie:find(Metric, Idx) of
error ->
Pos = M#mfile.next,
Mx = M#mfile{next=Pos+1, index=btrie:store(Metric, Pos, Idx)},
Bin = <<(byte_size(Metric)):16/integer, Metric/binary>>,
ok = file:write_file([M#mfile.name | ".idx"], Bin,
[read, append]),
{Mx, Pos*S*DataSize};
{ok, Pos} ->
{M, Pos*S*DataSize}
end,
P = Base+((Position - Offset)*DataSize),
R = file:pwrite(F, P, Value),
{R, M1}.
read(#mfile{offset=Offset, size=S, file=F, index=Idx},
DataSize, Metric, Position, Count)
when Position >= Offset,
(Position - Offset) + Count =< S ->
case btrie:find(Metric, Idx) of
error ->
{error, not_found};
{ok, Pos} ->
Base = Pos * S * DataSize,
P = Base+((Position - Offset) * DataSize),
file:pread(F, P, Count * DataSize)
end.
serialize_dir(Dir, DataSize, Fun, Chunk, Acc) ->
{ok, Fs} = file:list_dir(Dir),
Fs1 = [re:split(F, "\\.", [{return, list}]) || F <- Fs],
Idxs = [I || [I, "idx"] <- Fs1],
lists:foldl(fun(I, AccIn) ->
serialize_index([Dir, $/, I], DataSize, Fun, Chunk, AccIn)
end, Acc, Idxs).
serialize_index(Store, DataSize, Fun, Chunk, Acc) ->
{ok, MFile} = open_store(Store),
Res = lists:foldl(fun(M, AccIn) ->
serialize_metric(MFile, DataSize, M, Fun, Chunk, AccIn)
end, Acc, store_metrics(MFile)),
close_store(MFile),
Res.
serialize_metric(MFile, DataSize, Metric, Fun, infinity, Acc) ->
#mfile{offset=O,size=S} = MFile,
Fun1 = fun(Offset, Data, AccIn) ->
Fun(Metric, Offset, Data, AccIn)
end,
case read(MFile, DataSize, Metric, O, S) of
{ok, Data} ->
serialize_binary(DataSize, Data, Fun1, Acc, O, <<>>);
eof ->
Acc
end;
serialize_metric(MFile, DataSize, Metric, Fun, Chunk, Acc) ->
serialize_metric(MFile, DataSize, Metric, Fun, 0, Chunk, Acc).
%% If we've read everything (Start = Size) we just return the acc
serialize_metric(#mfile{size = _Size},
_DataSize, _Metric, _Fun, _Start, _Chunk, Acc) when _Start == _Size ->
Acc;
%% If we have at least 'Chunk' left to read
serialize_metric(MFile = #mfile{offset = O,
size = Size},
DataSize, Metric, Fun, Start, Chunk, Acc)
when Start + Chunk < Size ->
O1 = O + Start,
Fun1 = fun(Offset, Data, AccIn) ->
Fun(Metric, Offset + Start, Data, AccIn)
end,
case read(MFile, DataSize, Metric, O1, Chunk) of
{ok, Data} ->
Acc1 = serialize_binary(DataSize, Data, Fun1, Acc, O1, <<>>),
serialize_metric(MFile, DataSize, Metric, Fun, Start + Chunk, Chunk, Acc1);
eof ->
Acc
end;
%% We don't have a full chunk left to read.
serialize_metric(MFile = #mfile{offset = O,
size = Size},
DataSize, Metric, Fun, Start, _Chunk, Acc) ->
Chunk = Size - Start,
O1 = O + Start,
Fun1 = fun(Offset, Data, AccIn) ->
Fun(Metric, Offset + Start, Data, AccIn)
end,
case read(MFile, DataSize, Metric, O1, Chunk) of
{ok, Data} ->
serialize_binary(DataSize, Data, Fun1, Acc, O1, <<>>);
eof ->
Acc
end.
serialize_binary(_DataSize, <<>>, _Fun, FunAcc, _O, <<>>) ->
FunAcc;
serialize_binary(_DataSize, <<>>, Fun, FunAcc, O, Acc) ->
Fun(O, Acc, FunAcc);
serialize_binary(DataSize, <<?NONE:?TYPE_SIZE, R/binary>>, Fun, FunAcc, O, <<>>) ->
VSize =DataSize - 1,
<<_:VSize/binary, R1/binary>> = R,
serialize_binary(DataSize, R1, Fun, FunAcc, O+1, <<>>);
serialize_binary(DataSize, <<?NONE:?TYPE_SIZE, R/binary>>, Fun, FunAcc, O, Acc) ->
FunAcc1 = Fun(O, Acc, FunAcc),
VSize =DataSize - 1,
<<_:VSize/binary, R1/binary>> = R,
serialize_binary(DataSize, R1, Fun, FunAcc1, O+mmath_bin:length(Acc)+1, <<>>);
serialize_binary(DataSize, R, Fun, FunAcc, O, Acc) ->
<<V:DataSize/binary, R1/binary>> = R,
serialize_binary(DataSize, R1, Fun, FunAcc, O, <<Acc/binary, V/binary>>).
read_idx(F) ->
fold_idx(fun({init, Offset, Size}, undefined) ->
{Offset, Size, btrie:new(), 0};
({entry, M}, {Offset, Size, T, I}) ->
{Offset, Size, btrie:store(M, I, T), I+1}
end, undefined, F).
count_idx(F) ->
R = fold_idx(fun({init, _Offset, _Size}, Acc) ->
Acc;
({entry, _M}, Acc) ->
Acc + 1
end, 0, F),
case R of
{error, invalid_file} ->
lager:error("Ceould not read index file: ~p"),
0;
N ->
N
end.
fold_idx(Fun, Acc0, F) ->
fold_idx(Fun, Acc0, 4*1024, F).
fold_idx(Fun, Acc0, Chunk, F) ->
case file:open(F, [read | ?OPTS]) of
{ok, IO} ->
case file:read(IO, Chunk) of
{ok, <<Offset:64/?INT_TYPE, Size:64/?INT_TYPE, R/binary>>} ->
Acc = Fun({init, Offset, Size}, Acc0),
do_fold_idx(IO, Chunk, Fun, Acc, R);
_ ->
file:close(IO),
{error, invalid_file}
end;
E ->
E
end.
do_fold_idx(IO, Chunk, Fun, AccIn, <<_L:16/integer, M:_L/binary, R/binary>>) ->
Acc = Fun({entry, M}, AccIn),
do_fold_idx(IO, Chunk, Fun, Acc, R);
do_fold_idx(IO, Chunk, Fun, AccIn, R) ->
case file:read(IO, Chunk) of
{ok, Data} ->
do_fold_idx(IO, Chunk, Fun, AccIn, <<R/binary, Data/binary>>);
eof ->
file:close(IO),
case R of
<<>> ->
AccIn;
_ ->
{error, invalid_file}
end;
E ->
file:close(IO),
E
end.