Packages
ra
2.4.5
3.1.9
3.1.8
3.1.7
3.1.6
3.1.5
3.1.4
3.1.3
3.1.2
3.1.1
3.1.0
3.0.2
3.0.1
3.0.0
3.0.0-beta.1
2.17.3
2.17.2
2.17.1
2.17.0
2.16.13
2.16.12
2.16.11
2.16.10
2.16.9
2.16.8
2.16.7
2.16.6
2.16.5
2.16.4
2.16.3
2.16.2
2.16.1
2.16.0
2.16.0-pre.12
2.16.0-pre.11
2.16.0-pre.10
2.16.0-pre.9
2.16.0-pre.8
2.16.0-pre.7
2.16.0-pre.6
2.16.0-pre.5
2.16.0-pre.4
2.16.0-pre.3
2.16.0-pre.2
2.16.0-pre.1
2.15.4
2.15.3
2.15.2
2.15.1
2.15.0
2.14.0
2.13.6
2.13.5
2.13.4
2.13.3
2.13.2
2.13.1
2.13.0
2.13.0-pre.1
2.12.0
2.11.0
2.11.0-pre.1
2.10.2-pre.2
2.10.2-pre.1
2.10.1
2.10.0
2.10.0-pre.3
2.10.0-pre.2
2.10.0-pre.1
2.9.10-pre.1
2.9.1
2.9.1-pre.2
2.9.1-pre.1
2.9.0
2.8.0
retired
2.7.3
2.7.2
2.7.1
2.7.0
2.7.0-pre.3
2.7.0-pre.2
2.7.0-pre.1
2.6.3
2.6.2
2.6.1
2.6.0-pre.1
2.5.1
2.5.1-pre.1
2.5.0
2.4.9
2.4.8
2.4.7
2.4.6
2.4.5
2.4.4
2.4.3
2.4.2
retired
2.4.1
2.4.0
2.3.0
2.2.0
2.1.0
2.0.13
2.0.12
2.0.11
2.0.10
2.0.9
2.0.8
2.0.7
2.0.6
2.0.5
2.0.4
2.0.3
2.0.2
2.0.1
2.0.0
1.1.9
1.1.8
1.1.7
1.1.6
1.1.5
1.1.4
1.1.3
1.1.2
1.1.1
1.1.0
1.0.8
1.0.7
1.0.6
1.0.5
1.0.4
1.0.3
1.0.2
1.0.1
1.0.0
0.9.6
0.9.5
0.9.4
0.9.2
0.3.3
retired
0.3.2
retired
0.3.1
retired
Raft library
Current section
Files
Jump to
Current section
Files
src/ra_log_wal.erl
%% This Source Code Form is subject to the terms of the Mozilla Public
%% License, v. 2.0. If a copy of the MPL was not distributed with this
%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
%%
%% Copyright (c) 2017-2022 VMware, Inc. or its affiliates. All rights reserved.
%%
%% @hidden
-module(ra_log_wal).
-behaviour(gen_batch_server).
-export([start_link/1,
write/5,
write_batch/2,
truncate_write/5,
force_roll_over/1,
init/1,
handle_batch/2,
terminate/2,
format_status/1
]).
-export([wal2list/1]).
-compile([inline_list_funcs]).
-compile(inline).
-include("ra.hrl").
-define(CURRENT_VERSION, 1).
-define(MAGIC, "RAWA").
-define(HEADER_SIZE, 5).
-define(C_WAL_FILES, 1).
-define(C_BATCHES, 2).
-define(C_WRITES, 3).
-define(COUNTER_FIELDS,
[{wal_files, ?C_WAL_FILES, counter, "Number of write-ahead log files created"},
{batches, ?C_BATCHES, counter, "Number of batches written"},
{writes, ?C_WRITES, counter, "Number of entries written"}
]).
% a writer_id consists of a unqique local name (see ra_directory) and a writer's
% current pid().
% The pid is used for the immediate writer notification
% The atom is used by the segment writer to send the segments
% This has the effect that a restarted server has a different identity in terms
% of it's write notification but the same identity in terms of it's ets
% tables and segment notification
-type writer_id() :: {binary(), pid()}.
-record(batch_writer, {tbl_start :: ra_index(),
uid :: term(),
tid :: term(), %% TODO
from :: ra_index(),
to :: ra_index(),
term :: ra_term(),
inserts = [] :: list()}).
-record(batch, {writes = 0 :: non_neg_integer(),
waiting = #{} :: #{pid() => #batch_writer{}},
pending = [] :: iolist()
}).
-type wal_write_strategy() ::
% writes all pending in one write(2) call then calls fsync(1)
default |
% like default but tries to open the file using synchronous io
% (O_SYNC) rather than a write(2) followed by an fsync.
o_sync |
%% low latency mode where writers are notifies _before_ syncing
%% but after writing.
sync_after_notify.
-type writer_name_cache() :: {NextIntId :: non_neg_integer(),
#{writer_id() => binary()}}.
-record(conf, {file_modes :: [term()],
dir :: string(),
segment_writer = ra_log_segment_writer :: atom(),
compute_checksums = false :: boolean(),
max_size_bytes :: non_neg_integer(),
max_entries :: undefined | non_neg_integer(),
recovery_chunk_size = ?WAL_RECOVERY_CHUNK_SIZE :: non_neg_integer(),
write_strategy = default :: wal_write_strategy(),
sync_method = datasync :: sync | datasync | none,
counter :: counters:counters_ref(),
open_mem_tbls_tid :: ets:tid(),
closed_mem_tbls_tid :: ets:tid(),
names :: ra_system:names(),
explicit_gc = false :: boolean(),
pre_allocate = false :: boolean(),
compress_mem_tables = false :: boolean()
}).
-record(wal, {fd :: 'maybe'(file:io_device()),
filename :: 'maybe'(file:filename()),
writer_name_cache = {0, #{}} :: writer_name_cache(),
max_size :: non_neg_integer(),
entry_count = 0 :: non_neg_integer()
}).
-record(state, {conf = #conf{},
file_num = 0 :: non_neg_integer(),
wal :: #wal{} | undefined,
file_size = 0 :: non_neg_integer(),
% writers that have attempted to write an non-truncating
% out of seq % entry.
% No further writes are allowed until the missing
% index has been received.
% out_of_seq are kept after a roll over or until
% a truncating write is received.
% no attempt is made to recover this information after a crash
% beyond the available WAL files
% all writers seen within the lifetime of a WAL file
% and the last index seen
writers = #{} :: #{ra_uid() =>
{in_seq | out_of_seq, ra_index()}},
batch :: 'maybe'(#batch{})
}).
-type state() :: #state{}.
-type wal_conf() :: #{name := atom(), %% the name to register the wal as
names := ra_system:names(),
dir := file:filename_all(),
max_size_bytes => non_neg_integer(),
max_entries => non_neg_integer(),
segment_writer => atom() | pid(),
compute_checksums => boolean(),
pre_allocate => boolean(),
write_strategy => wal_write_strategy(),
sync_method => sync | datasync,
recovery_chunk_size => non_neg_integer(),
hibernate_after => non_neg_integer(),
max_batch_size => non_neg_integer(),
garbage_collect => boolean(),
min_bin_vheap_size => non_neg_integer(),
compress_mem_tables => boolean()
}.
-export_type([wal_conf/0,
wal_write_strategy/0]).
-type wal_command() ::
{append | truncate, writer_id(), ra_index(), ra_term(), term()}.
-type wal_op() :: {cast, wal_command()} |
{call, from(), wal_command()}.
-spec write(writer_id(), atom(), ra_index(), ra_term(), term()) ->
ok | {error, wal_down}.
write(From, Wal, Idx, Term, Entry) ->
named_cast(Wal, {append, From, Idx, Term, Entry}).
-spec truncate_write(writer_id(), atom(), ra_index(), ra_term(), term()) ->
ok | {error, wal_down}.
truncate_write(From, Wal, Idx, Term, Entry) ->
named_cast(Wal, {truncate, From, Idx, Term, Entry}).
-spec write_batch(Wal :: atom() | pid(), [wal_command()]) ->
ok | {error, wal_down}.
write_batch(Wal, WalCommands) when is_pid(Wal) ->
gen_batch_server:cast_batch(Wal, WalCommands);
write_batch(Wal, WalCommands) when is_atom(Wal) ->
case whereis(Wal) of
undefined ->
{error, wal_down};
Pid ->
write_batch(Pid, WalCommands)
end.
named_cast(To, Msg) when is_pid(To) ->
gen_batch_server:cast(To, Msg);
named_cast(Wal, Msg) ->
case whereis(Wal) of
undefined ->
{error, wal_down};
Pid ->
named_cast(Pid, Msg)
end.
% force a wal file to roll over to a new file
% mostly useful for testing
force_roll_over(Wal) ->
ok = gen_batch_server:cast(Wal, rollover),
ok.
%% ra_log_wal
%%
%% Writes Raft entries to shared persistent storage for multiple "writers"
%% fsyncs in batches, typically the write requests
%% received in the mailbox during
%% the previous fsync operation. Notifies all writers after each fsync batch.
%% Also have got a dynamically increasing max writes limit that grows in order
%% to trade-off latency for throughput.
%%
%% Entries are written to the .wal file as well as a per-writer mem table (ETS).
%% In order for writers to locate an entry by an index a lookup ETS table
%% (ra_log_open_mem_tables) keeps the current range of indexes
%% a mem_table as well
%% as the mem_table tid(). This lookup table is updated on every write.
%%
%% Once the current .wal file is full a new one is closed. All the entries in
%% ra_log_open_mem_tables are moved to ra_log_closed_mem_tables so that writers
%% can still locate the tables whilst they are being flushed to disk. The
%% ra_log_segment_writer is notified of all the mem tables written to during
%% the lifetime of the .wal file and will begin writing these to on-disk segment
%% files. Once it has finished the current set of mem_tables it will delete the
%% corresponding .wal file.
-spec start_link(Config :: wal_conf()) ->
{ok, pid()} |
{error, {already_started, pid()}} |
{error, wal_checksum_validation_failure}.
start_link(#{name := Name} = Config)
when is_atom(Name) ->
WalMaxBatchSize = maps:get(max_batch_size, Config,
?WAL_DEFAULT_MAX_BATCH_SIZE),
Options0 = case maps:get(hibernate_after, Config, undefined) of
undefined ->
[{max_batch_size, WalMaxBatchSize}];
Hib ->
[{hibernate_after, Hib},
{max_batch_size, WalMaxBatchSize}]
end,
Options = [{reversed_batch, true} | Options0],
gen_batch_server:start_link({local, Name}, ?MODULE, Config, Options).
%%% Callbacks
-spec init(wal_conf()) -> {ok, state()}.
init(#{dir := Dir} = Conf0) ->
#{max_size_bytes := MaxWalSize,
max_entries := MaxEntries,
recovery_chunk_size := RecoveryChunkSize,
segment_writer := SegWriter,
compute_checksums := ComputeChecksums,
pre_allocate := PreAllocate,
write_strategy := WriteStrategy,
sync_method := SyncMethod,
garbage_collect := Gc,
min_bin_vheap_size := MinBinVheapSize,
compress_mem_tables := CompressMemTables,
names := #{wal := WalName,
open_mem_tbls := OpenTblsName,
closed_mem_tbls := ClosedTblsName} = Names} =
merge_conf_defaults(Conf0),
?NOTICE("WAL: ~ts init, open tbls: ~w, closed tbls: ~w",
[WalName, OpenTblsName, ClosedTblsName]),
process_flag(trap_exit, true),
% given ra_log_wal is effectively a fan-in sink it is likely that it will
% at times receive large number of messages from a large number of
% writers
process_flag(message_queue_data, off_heap),
process_flag(min_bin_vheap_size, MinBinVheapSize),
CRef = ra_counters:new(WalName, ?COUNTER_FIELDS),
% wait for the segment writer to process anything in flight
ok = ra_log_segment_writer:await(SegWriter),
%% TODO: recover wal should return {stop, Reason} if it fails
%% rather than crash
FileModes = [raw, write, read, binary],
Conf = #conf{file_modes = FileModes,
dir = Dir,
segment_writer = SegWriter,
compute_checksums = ComputeChecksums,
max_size_bytes = max(?WAL_MIN_SIZE, MaxWalSize),
max_entries = MaxEntries,
recovery_chunk_size = RecoveryChunkSize,
write_strategy = WriteStrategy,
sync_method = SyncMethod,
counter = CRef,
open_mem_tbls_tid = ets:whereis(OpenTblsName),
closed_mem_tbls_tid = ets:whereis(ClosedTblsName),
names = Names,
explicit_gc = Gc,
pre_allocate = PreAllocate,
compress_mem_tables = CompressMemTables},
{ok, recover_wal(Dir, Conf)}.
-spec handle_batch([wal_op()], state()) ->
{ok, [gen_batch_server:action()], state()}.
handle_batch(Ops, #state{conf = #conf{explicit_gc = Gc}} = State0) ->
State = lists:foldr(fun handle_op/2, start_batch(State0), Ops),
%% process all ops
Actions = case Gc of
true -> [garbage_collect];
false -> []
end,
{ok, Actions, complete_batch(State)}.
terminate(_Reason, State) ->
_ = cleanup(State),
ok.
format_status(#state{conf = #conf{write_strategy = Strat,
compute_checksums = Cs,
max_size_bytes = MaxSize},
writers = Writers,
file_size = FSize,
wal = #wal{filename = Fn}}) ->
#{write_strategy => Strat,
compute_checksums => Cs,
writers => maps:size(Writers),
filename => filename:basename(Fn),
current_size => FSize,
max_size_bytes => MaxSize}.
%% Internal
handle_op({cast, WalCmd}, State) ->
handle_msg(WalCmd, State).
recover_wal(Dir, #conf{segment_writer = SegWriter,
open_mem_tbls_tid = OpenTbl,
closed_mem_tbls_tid = ClosedTbl,
recovery_chunk_size = RecoveryChunkSize} = Conf) ->
% ensure configured directory exists
ok = ra_lib:make_dir(Dir),
% recover each mem table and notify segment writer
% this may result in duplicated segments but that is better than
% losing any data
% As we have waited for the segment writer to finish processing it is
% assumed that any remaining wal files need to be re-processed.
WalFiles = lists:sort(filelib:wildcard(filename:join(Dir, "*.wal"))),
% First we recover all the tables using a temporary lookup table.
% Then we update the actual lookup tables atomically.
RecoverTid = ets:new(ra_log_recover_mem_tables,
[set, {write_concurrency, true}, private]),
% compute all closed mem table lookups required so we can insert them
% all at once, atomically
% It needs to be atomic so that readers don't accidentally
% read partially recovered
% tables mixed with old tables
RecoverConf = Conf#conf{open_mem_tbls_tid = RecoverTid},
All = [begin
FBase = filename:basename(F),
?DEBUG("wal: recovering ~ts", [FBase]),
Fd = open_at_first_record(F),
{Time, ok} = timer:tc(
fun () ->
recover_wal_chunks(RecoverConf, Fd,
RecoveryChunkSize)
end),
?DEBUG("wal: recovered ~ts time taken ~bms",
[FBase, Time div 1000]),
close_existing(Fd),
recovering_to_closed(RecoverTid, F)
end || F <- WalFiles],
% get all the recovered tables and insert them into closed
Closed = lists:append([C || {C, _, _} <- All]),
true = ets:insert(ClosedTbl, Closed),
% send all the mem tables to segment writer for processing
% This could result in duplicate segments
[ok = ra_log_segment_writer:accept_mem_tables(SegWriter, M, F)
|| {_, M, F} <- All],
FileNum = extract_file_num(lists:reverse(WalFiles)),
State = roll_over(RecoverTid, #state{conf = Conf,
file_num = FileNum}),
% we can now delete all open mem tables as should be covered by recovered
% closed tables
Open = ets:tab2list(OpenTbl),
true = ets:delete_all_objects(OpenTbl),
% delete all open ets tables
[true = ets:delete(Tid) || {_, _, _, Tid} <- Open],
true = ets:delete(RecoverTid),
%% force garbage cleanup
true = erlang:garbage_collect(),
State.
extract_file_num([]) ->
0;
extract_file_num([F | _]) ->
ra_lib:zpad_extract_num(filename:basename(F)).
cleanup(#state{wal = #wal{fd = undefined}}) ->
ok;
cleanup(#state{wal = #wal{fd = Fd}}) ->
_ = ra_file_handle:sync(Fd),
ok.
serialize_header(UId, Trunc, {Next, Cache} = WriterCache) ->
case Cache of
#{UId := <<_:1, BinId:23/bitstring>>} when Trunc ->
{<<1:1/unsigned, BinId/bitstring>>, 2, WriterCache};
#{UId := BinId} ->
{BinId, 3, WriterCache};
_ ->
% TODO: check overflows of Next
% cache the header index binary to avoid re-creating it every time
% sets Truncate = false initially as this is the most common case
T = case Trunc of true -> 1; false -> 0 end,
BinId = <<0:1/unsigned, 1:1/unsigned, Next:22/unsigned>>,
IdDataLen = byte_size(UId),
Header = <<T:1/unsigned, 0:1/unsigned, Next:22/unsigned,
IdDataLen:16/unsigned, UId/binary>>,
{Header, byte_size(Header),
{Next + 1, Cache#{UId => BinId}}}
end.
write_data({UId, _} = Id, Idx, Term, Data0, Trunc,
#state{conf = #conf{compute_checksums = ComputeChecksum},
wal = #wal{writer_name_cache = Cache0,
entry_count = Count} = Wal} = State00) ->
% if the next write is going to exceed the configured max wal size
% we roll over to a new wal.
case should_roll_wal(State00) of
true ->
State = roll_over(State00),
write_data(Id, Idx, Term, Data0, Trunc, State);
false ->
EntryData = to_binary(Data0),
EntryDataLen = iolist_size(EntryData),
{HeaderData, HeaderLen, Cache} = serialize_header(UId, Trunc, Cache0),
% fixed overhead =
% 24 bytes 2 * 64bit ints (idx, term) + 2 * 32 bit ints (checksum, datalen)
DataSize = HeaderLen + 24 + EntryDataLen,
State0 = State00#state{wal = Wal#wal{writer_name_cache = Cache,
entry_count = Count + 1}},
Entry = [<<Idx:64/unsigned,
Term:64/unsigned>> |
EntryData],
Checksum = case ComputeChecksum of
true -> erlang:adler32(Entry);
false -> 0
end,
Record = [HeaderData,
<<Checksum:32/integer, EntryDataLen:32/unsigned>> |
Entry],
append_data(State0, Id, Idx, Term, Data0,
DataSize, Record, Trunc)
end.
handle_msg({append, {UId, Pid} = Id, Idx, Term, Entry},
#state{writers = Writers} = State0) ->
case maps:find(UId, Writers) of
{ok, {_, PrevIdx}} when Idx =< PrevIdx + 1 ->
write_data(Id, Idx, Term, Entry, false, State0);
error ->
write_data(Id, Idx, Term, Entry, false, State0);
{ok, {out_of_seq, _}} ->
% writer is out of seq simply ignore drop the write
% TODO: capture metric for dropped writes
State0;
{ok, {in_seq, PrevIdx}} ->
% writer was in seq but has sent an out of seq entry
% notify writer
?DEBUG("WAL: requesting resend from `~w`, "
"last idx ~b idx received ~b",
[UId, PrevIdx, Idx]),
Pid ! {ra_log_event, {resend_write, PrevIdx + 1}},
State0#state{writers = Writers#{UId => {out_of_seq, PrevIdx}}}
end;
handle_msg({truncate, Id, Idx, Term, Entry}, State0) ->
write_data(Id, Idx, Term, Entry, true, State0);
handle_msg(rollover, State) ->
roll_over(State).
append_data(#state{conf = Cfg,
file_size = FileSize,
batch = Batch0,
writers = Writers} = State,
{UId, Pid}, Idx, Term, Entry, DataSize, Data, Truncate) ->
Batch = incr_batch(Cfg, Batch0, UId, Pid,
{Idx, Term, Entry}, Data, Truncate),
State#state{file_size = FileSize + DataSize,
batch = Batch,
writers = Writers#{UId => {in_seq, Idx}} }.
incr_batch(#conf{open_mem_tbls_tid = OpnMemTbl} = Cfg,
#batch{writes = Writes,
waiting = Waiting0,
pending = Pend} = Batch,
UId, Pid, {Idx, Term, _} = Record, Data, Truncate) ->
Waiting = case Waiting0 of
#{Pid := #batch_writer{tbl_start = TblStart0,
tid = _Tid,
from = From,
inserts = Inserts0} = W} ->
TblStart = table_start(Truncate, Idx, TblStart0),
Inserts = case Inserts0 of
[] ->
[Record];
[{PrevIdx, _, _} | RemIdxs] when Idx =< PrevIdx ->
%% we are overwriting, this should rarely,
%% if ever happen within the timeframe of a batch
%% Drop all inserts that are higher or equal
%% to the current Idx
Filtered =
lists:dropwhile(fun ({I, _, _}) ->
Idx =< I
end, RemIdxs),
[Record | Filtered];
_ ->
[Record | Inserts0]
end,
Waiting0#{Pid => W#batch_writer{from = min(Idx, From),
tbl_start = TblStart,
to = Idx,
term = Term,
inserts = Inserts}};
_ ->
%% no batch_writer
{Tid, TblStart} =
case ets:lookup(OpnMemTbl, UId) of
[{_UId, TblStart0, _TblEnd, T}] ->
{T, table_start(Truncate, Idx, TblStart0)};
_ ->
%% there is no table so need
%% to open one
T = open_mem_table(Cfg, UId),
true = ets:insert_new(OpnMemTbl,
{UId, Idx, Idx, T}),
{T, Idx}
end,
Writer = #batch_writer{tbl_start = TblStart,
from = Idx,
to = Idx,
tid = Tid,
uid = UId,
term = Term,
inserts = [Record]},
Waiting0#{Pid => Writer}
end,
Batch#batch{writes = Writes + 1,
waiting = Waiting,
pending = [Pend | Data]}.
update_mem_table(#conf{open_mem_tbls_tid = OpnMemTbl} = Cfg,
UId, Idx, Term, Entry, Truncate) ->
% TODO: if Idx =< First we could truncate the entire table and save
% some disk space when it later is flushed to disk
case ets:lookup(OpnMemTbl, UId) of
[{_UId, From0, _To, Tid}] ->
true = ets:insert(Tid, {Idx, Term, Entry}),
From = table_start(Truncate, Idx, From0),
% update Last idx for current tbl
% this is how followers overwrite previously seen entries
% TODO: OPTIMISATION
% Writers don't need this updated for every entry. As they keep
% a local cache of unflushed entries it is sufficient to update
% ra_log_open_mem_tables before completing the batch.
% Instead the `From` and `To` could be kept in the batch.
_ = ets:update_element(OpnMemTbl, UId, [{2, From}, {3, Idx}]);
[] ->
% open new ets table
Tid = open_mem_table(Cfg, UId),
true = ets:insert_new(OpnMemTbl, {UId, Idx, Idx, Tid}),
true = ets:insert(Tid, {Idx, Term, Entry})
end.
roll_over(#state{conf = #conf{open_mem_tbls_tid = Tbl}} = State0) ->
State = complete_batch(State0),
roll_over(Tbl, start_batch(State)).
roll_over(OpnMemTbls, #state{wal = Wal0, file_num = Num0,
conf = #conf{dir = Dir,
max_size_bytes = MaxBytes
} = Conf0} = State0) ->
counters:add(Conf0#conf.counter, ?C_WAL_FILES, 1),
Num = Num0 + 1,
Fn = ra_lib:zpad_filename("", "wal", Num),
NextFile = filename:join(Dir, Fn),
?DEBUG("wal: opening new file ~ts open mem tables: ~w", [Fn, OpnMemTbls]),
%% if this is the first wal since restart randomise the first
%% max wal size to reduce the likelihood that each erlang node will
%% flush mem tables at the same time
NextMaxBytes = case Wal0 of
undefined ->
Half = MaxBytes div 2,
Half + rand:uniform(Half);
_ ->
ok = close_file(Wal0#wal.fd),
MemTables = ets:tab2list(OpnMemTbls),
ok = close_open_mem_tables(MemTables, Conf0,
Wal0#wal.filename),
MaxBytes
end,
{Conf, Wal} = open_wal(NextFile, NextMaxBytes, Conf0),
State0#state{conf = Conf,
wal = Wal,
file_size = 0,
file_num = Num}.
open_wal(File, Max, #conf{write_strategy = o_sync,
file_modes = Modes0} = Conf) ->
Modes = [sync | Modes0],
case prepare_file(File, Modes) of
{ok, Fd} ->
% many platforms implement O_SYNC a bit like O_DSYNC
% perform a manual sync here to ensure metadata is flushed
{Conf, #wal{fd = Fd,
max_size = Max,
filename = File}};
{error, enotsup} ->
?WARN("wal: o_sync write strategy not supported. "
"Reverting back to default strategy.", []),
open_wal(File, Max, Conf#conf{write_strategy = default})
end;
open_wal(File, Max, #conf{file_modes = Modes} = Conf0) ->
{ok, Fd} = prepare_file(File, Modes),
Conf = maybe_pre_allocate(Conf0, Fd, Max),
{Conf, #wal{fd = Fd,
max_size = Max,
filename = File}}.
prepare_file(File, Modes) ->
Tmp = make_tmp(File),
%% rename is atomic-ish so we will never accidentally write an empty wal file
%% using prim_file here as file:rename/2 uses the file server
ok = prim_file:rename(Tmp, File),
case ra_file_handle:open(File, Modes) of
{ok, Fd2} ->
{ok, ?HEADER_SIZE} = file:position(Fd2, ?HEADER_SIZE),
{ok, Fd2};
{error, _} = Err ->
Err
end.
make_tmp(File) ->
Tmp = filename:rootname(File) ++ ".tmp",
{ok, Fd} = file:open(Tmp, [write, binary, raw]),
ok = file:write(Fd, <<?MAGIC, ?CURRENT_VERSION:8/unsigned>>),
ok = file:sync(Fd),
ok = file:close(Fd),
Tmp.
maybe_pre_allocate(#conf{pre_allocate = true,
write_strategy = Strat} = Conf, Fd, Max0)
when Strat /= o_sync ->
Max = Max0 - ?HEADER_SIZE,
case file:allocate(Fd, ?HEADER_SIZE, Max) of
ok ->
{ok, Max} = file:position(Fd, Max),
ok = file:truncate(Fd),
{ok, ?HEADER_SIZE} = file:position(Fd, ?HEADER_SIZE),
Conf;
{error, _} ->
%% fallocate may not be supported, fall back to fsync instead
%% of fdatasync
?INFO("wal: preallocation may not be supported by the file system"
" falling back to fsync instead of fdatasync", []),
Conf#conf{pre_allocate = false}
end;
maybe_pre_allocate(Conf, _Fd, _Max) ->
Conf.
close_file(undefined) ->
ok;
close_file(Fd) ->
% ok = ra_file_handle:sync(Fd),
ra_file_handle:close(Fd).
close_open_mem_tables(MemTables,
#conf{segment_writer = TblWriter,
open_mem_tbls_tid = OpnMemTbls,
closed_mem_tbls_tid = CloseMemTbls},
Filename) ->
% insert into closed mem tables
% so that readers can still resolve the table whilst it is being
% flushed to persistent tables asynchronously
[begin
% In order to ensure that reads are done in the correct causal order
% we need to append a monotonically increasing value for readers to
% sort by
M = erlang:unique_integer([monotonic, positive]),
_ = ets:insert(CloseMemTbls, erlang:insert_element(2, T, M))
end || T <- MemTables],
% reset open mem tables table
true = ets:delete_all_objects(OpnMemTbls),
% notify segment_writer of new unflushed memtables
ok = ra_log_segment_writer:accept_mem_tables(TblWriter, MemTables, Filename),
ok.
recovering_to_closed(RecoverTid, Filename) ->
MemTables = ets:tab2list(RecoverTid),
Closed = [begin
M = erlang:unique_integer([monotonic, positive]),
erlang:insert_element(2, T, M)
end || T <- MemTables],
true = ets:delete_all_objects(RecoverTid),
{Closed, MemTables, Filename}.
open_mem_table(Cfg, {UId, _Pid}) ->
open_mem_table(Cfg, UId);
open_mem_table(#conf{names = Names,
compress_mem_tables = CompressTbls}, UId) ->
% lookup the locally registered name of the process to use as ets
% name
ServerName = ra_directory:name_of(Names, UId),
Opts = case CompressTbls of
true ->
[set, {write_concurrency, true}, public, compressed];
false ->
[set, {write_concurrency, true}, public]
end,
Tid = ets:new(ServerName, Opts),
% immediately give away ownership to ets process
true = ra_log_ets:give_away(Names, Tid),
Tid.
start_batch(#state{conf = #conf{counter = CRef}} = State) ->
ok = counters:add(CRef, ?C_BATCHES, 1),
State#state{batch = #batch{}}.
post_notify_flush(#state{wal = #wal{fd = Fd},
conf = #conf{write_strategy = sync_after_notify,
sync_method = SyncMeth}}) ->
ok = ra_file_handle:SyncMeth(Fd);
post_notify_flush(_State) ->
ok.
flush_pending(#state{wal = #wal{fd = Fd},
batch = #batch{pending = Pend},
conf = #conf{write_strategy = WriteStrategy,
sync_method = SyncMeth}} = State0) ->
case WriteStrategy of
default ->
ok = ra_file_handle:write(Fd, Pend),
ok = ra_file_handle:SyncMeth(Fd),
ok;
_ ->
ok = ra_file_handle:write(Fd, Pend)
end,
State0#state{batch = undefined}.
complete_batch(#state{batch = undefined} = State) ->
State;
complete_batch(#state{batch = #batch{waiting = Waiting,
writes = NumWrites},
conf = #conf{open_mem_tbls_tid = OpnTbl} = Cfg
} = State0) ->
% TS = erlang:system_time(microsecond),
State = flush_pending(State0),
% SyncTS = erlang:system_time(microsecond),
counters:add(Cfg#conf.counter, ?C_WRITES, NumWrites),
%% process writers
ra_lib:maps_foreach(
fun (Pid, #batch_writer{tbl_start = TblStart,
uid = UId,
from = From,
to = To,
term = Term,
inserts = Inserts,
tid = Tid}) ->
true = ets:insert(Tid, Inserts),
true = ets:update_element(OpnTbl, UId,
[{2, TblStart}, {3, To}]),
Pid ! {ra_log_event, {written, {From, To, Term}}},
ok
end, Waiting),
ok = post_notify_flush(State),
State.
wal2list(File) ->
Data = open_existing(File),
dump_records(Data, []).
open_existing(File) ->
case file:read_file(File) of
{ok, <<?MAGIC, ?CURRENT_VERSION:8/unsigned, Data/binary>>} ->
%% the only version currently supported
Data;
{ok, <<Magic:4/binary, UnknownVersion:8/unsigned, _/binary>>} ->
exit({unknown_wal_file_format, Magic, UnknownVersion})
end.
open_at_first_record(File) ->
{ok, Fd} = file:open(File, [read, binary, raw]),
case file:read(Fd, 5) of
{ok, <<?MAGIC, ?CURRENT_VERSION:8/unsigned>>} ->
%% the only version currently supported
Fd;
{ok, <<Magic:4/binary, UnknownVersion:8/unsigned>>} ->
exit({unknown_wal_file_format, Magic, UnknownVersion})
end.
close_existing(Fd) ->
case file:close(Fd) of
ok ->
ok;
{error, Reason} ->
exit({could_not_close, Reason})
end.
dump_records(<<_:1/unsigned, 0:1/unsigned, _:22/unsigned,
IdDataLen:16/unsigned, _:IdDataLen/binary,
_:32/integer,
0:32/unsigned,
_Idx:64/unsigned, _Term:64/unsigned,
_EntryData:0/binary,
_Rest/binary>>, Entries) ->
Entries;
dump_records(<<_:1/unsigned, 0:1/unsigned, _:22/unsigned,
IdDataLen:16/unsigned, _:IdDataLen/binary,
Crc:32/integer,
EntryDataLen:32/unsigned,
Idx:64/unsigned, Term:64/unsigned,
EntryData:EntryDataLen/binary,
Rest/binary>>, Entries) ->
% TODO: recover writers info, i.e. last index seen
case erlang:adler32(<<Idx:64/unsigned, Term:64/unsigned, EntryData/binary>>) of
Crc ->
dump_records(Rest, [{Idx, Term, binary_to_term(EntryData)} | Entries]);
_ ->
exit({crc_failed_for, Idx, EntryData})
end;
dump_records(<<_:1/unsigned, 1:1/unsigned, _:22/unsigned,
Crc:32/integer,
EntryDataLen:32/unsigned,
Idx:64/unsigned, Term:64/unsigned,
EntryData:EntryDataLen/binary,
Rest/binary>>, Entries) ->
case erlang:adler32(<<Idx:64/unsigned, Term:64/unsigned, EntryData/binary>>) of
Crc ->
dump_records(Rest, [{Idx, Term, binary_to_term(EntryData)} | Entries]);
_ ->
exit({crc_failed_for, Idx, EntryData})
end;
dump_records(<<>>, Entries) ->
Entries.
% TODO: recover writers info, i.e. last index seen
recover_wal_chunks(Conf, Fd, RecoveryChunkSize) ->
Chunk = read_from_wal_file(Fd, RecoveryChunkSize),
recover_records(Conf, Fd, Chunk, #{}, RecoveryChunkSize).
% All zeros indicates end of a pre-allocated wal file
recover_records(_, _Fd, <<0:1/unsigned, 0:1/unsigned, 0:22/unsigned,
IdDataLen:16/unsigned, _:IdDataLen/binary,
0:32/integer, 0:32/unsigned, _/binary>>,
_Cache, _ChunkSize) ->
ok;
% First record or different UID to last record
recover_records(Conf, Fd,
<<Trunc:1/unsigned, 0:1/unsigned, IdRef:22/unsigned,
IdDataLen:16/unsigned, UId:IdDataLen/binary,
Checksum:32/integer,
EntryDataLen:32/unsigned,
Idx:64/unsigned, Term:64/unsigned,
EntryData:EntryDataLen/binary,
Rest/binary>>,
Cache, RecoveryChunkSize) ->
case validate_checksum(Checksum, Idx, Term, EntryData) of
ok ->
true = update_mem_table(Conf, UId, Idx, Term,
binary_to_term(EntryData), Trunc =:= 1),
Cache0 = Cache#{IdRef => {UId, <<1:1/unsigned, IdRef:22/unsigned>>}},
recover_records(Conf, Fd, Rest, Cache0, RecoveryChunkSize);
error ->
?DEBUG("WAL: record failed CRC check. If this is the last record"
" recovery can resume", []),
%% if this is the last entry in the wal we can just drop the
%% record;
is_last_record(Fd, Rest)
end;
% Same UID as last record
recover_records(Conf, Fd,
<<Trunc:1/unsigned, 1:1/unsigned, IdRef:22/unsigned,
Checksum:32/integer,
EntryDataLen:32/unsigned,
Idx:64/unsigned, Term:64/unsigned,
EntryData:EntryDataLen/binary,
Rest/binary>>,
Cache, RecoveryChunkSize) ->
#{IdRef := {UId, _}} = Cache,
case validate_checksum(Checksum, Idx, Term, EntryData) of
ok ->
true = update_mem_table(Conf, UId, Idx, Term,
binary_to_term(EntryData), Trunc =:= 1),
recover_records(Conf, Fd, Rest, Cache, RecoveryChunkSize);
error ->
?DEBUG("WAL: record failed CRC check. If this is the last record"
" recovery can resume", []),
%% if this is the last entry in the wal we can just drop the
%% record;
is_last_record(Fd, Rest)
end;
% Not enough remainder to parse another record, need to read
recover_records(Conf, Fd, Chunk, Cache, RecoveryChunkSize) ->
NextChunk = read_from_wal_file(Fd, RecoveryChunkSize),
case NextChunk of
<<>> ->
%% we have reached the end of the file
ok;
_ ->
%% append this chunk to the remainder of the last chunk
Chunk0 = <<Chunk/binary, NextChunk/binary>>,
recover_records(Conf, Fd, Chunk0, Cache, RecoveryChunkSize)
end.
is_last_record(_Fd, <<0:104, _/binary>>) ->
ok;
is_last_record(Fd, Rest) ->
case byte_size(Rest) < 13 of
true ->
case read_from_wal_file(Fd, 256) of
<<>> ->
ok;
Next ->
is_last_record(Fd, <<Rest/binary, Next/binary>>)
end;
false ->
?ERROR("WAL: record failed CRC check during recovery. "
"Unable to recover WAL data safely", []),
throw({stop, wal_checksum_validation_failure})
end.
read_from_wal_file(Fd, Len) ->
case file:read(Fd, Len) of
{ok, <<Data/binary>>} ->
Data;
eof ->
<<>>;
{error, Reason} ->
exit({could_not_read_wal_chunk, Reason})
end.
validate_checksum(0, _, _, _) ->
% checksum not used
ok;
validate_checksum(Checksum, Idx, Term, Data) ->
% building a binary just for the checksum may feel a bit wasteful
% but this is only called during recovery which should be a rare event
case erlang:adler32(<<Idx:64/unsigned, Term:64/unsigned, Data/binary>>) of
Checksum ->
ok;
_ ->
error
end.
merge_conf_defaults(Conf) ->
maps:merge(#{segment_writer => ra_log_segment_writer,
max_size_bytes => ?WAL_DEFAULT_MAX_SIZE_BYTES,
max_entries => undefined,
recovery_chunk_size => ?WAL_RECOVERY_CHUNK_SIZE,
compute_checksums => true,
pre_allocate => false,
write_strategy => default,
garbage_collect => false,
sync_method => datasync,
min_bin_vheap_size => ?WAL_MIN_BIN_VHEAP_SIZE,
compress_mem_tables => false}, Conf).
to_binary(Term) ->
term_to_iovec(Term).
should_roll_wal(#state{conf = #conf{max_entries = MaxEntries},
file_size = FileSize,
wal = #wal{max_size = MaxWalSize,
entry_count = Count}}) ->
%% Initially, MaxWalSize was a hard limit for the file size: if FileSize +
%% DataSize went over that limit, we would use a new file. This was an
%% issue when DataSize was larger than the limit alone.
%%
%% The chosen solution is to only consider the current file size in the
%% calculation. It means that after DataSize bytes are written, the file
%% will be larger than the configured maximum size. But at least it will
%% accept Ra commands larger than the max WAL size.
FileSize > MaxWalSize orelse case MaxEntries of
undefined -> false;
_ ->
Count + 1 > MaxEntries
end.
table_start(false, Idx, TblStart) ->
%% take the smaller of the existing first item
%% in case we are overwriting a previous entry
min(TblStart, Idx);
table_start(true, Idx, _TblStart) ->
Idx.