%% @hidden -module(ra_log_wal). -behaviour(gen_batch_server). -export([start_link/2, 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(MAX_SIZE_BYTES, 512 * 1000 * 1000). -define(METRICS_WINDOW_SIZE, 100). -define(CURRENT_VERSION, 1). -define(MAGIC, "RAWA"). % 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{}}, % {From :: ra_index(), To :: ra_index(), % Term :: ra_term()}}, start_time :: maybe(integer()), pending = [] :: iolist() }). -type wal_write_strategy() :: % writes all pending in one write(2) call then calls fsync(1) default | % like delay writes but tries to open the file using synchronous io % (O_SYNC) rather than a write(2) followed by an fsync. o_sync. -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 = ?MAX_SIZE_BYTES :: non_neg_integer(), write_strategy = default :: wal_write_strategy(), sync_method = datasync :: sync | datasync }). -record(wal, {fd :: maybe(file:io_device()), filename :: maybe(file:filename()), writer_name_cache = {0, #{}} :: writer_name_cache(), max_size = ?MAX_SIZE_BYTES :: 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()}}, metrics_cursor = 0 :: non_neg_integer(), batch :: maybe(#batch{}) }). -type state() :: #state{}. -type wal_conf() :: #{dir => file:filename_all(), max_size_bytes => non_neg_integer(), segment_writer => atom() | pid(), compute_checksums => boolean(), write_strategy => wal_write_strategy(), sync_method => sync | datasync}. -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 ot 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(), Options :: list()) -> {ok, pid()} | {error, {already_started, pid()}}. start_link(Config, Options0) when is_list(Options0) -> Options = [{reversed_batch, true} | Options0], gen_batch_server:start_link({local, ?MODULE}, ?MODULE, Config, Options). %%% Callbacks -spec init(wal_conf()) -> {ok, state()}. init(#{dir := Dir} = Conf0) -> #{max_size_bytes := MaxWalSize, segment_writer := SegWriter, compute_checksums := ComputeChecksums, write_strategy := WriteStrategy, sync_method := SyncMethod} = merge_conf_defaults(Conf0), process_flag(trap_exit, true), % TODO: test that off_heap is actuall beneficial % 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), _ = ets:new(ra_log_wal_metrics, [set, named_table, {read_concurrency, true}, protected]), % seed metrics table with data [true = ets:insert(ra_log_wal_metrics, {I, undefined}) || I <- lists:seq(0, ?METRICS_WINDOW_SIZE-1)], % 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, append, binary], Conf = #conf{file_modes = FileModes, dir = Dir, segment_writer = SegWriter, compute_checksums = ComputeChecksums, max_size_bytes = MaxWalSize, write_strategy = WriteStrategy, sync_method = SyncMethod}, {ok, recover_wal(Dir, Conf)}. -spec handle_batch([wal_op()], state()) -> {ok, [gen_batch_server:action()], state()}. handle_batch(Ops, State0) -> State = lists:foldr(fun handle_op/2, start_batch(State0), Ops), %% process all ops {ok, [garbage_collect], 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 = TblWriter} = 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"))), ?INFO("WAL: recovering ~p", [WalFiles]), % First we recover all the tables using a temporary lookup table. % Then we update the actual lookup tables atomically. _ = ets:new(ra_log_recover_mem_tables, [set, named_table, {read_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 All = [begin Data = open_existing(F), ok = try_recover_records(Data, #{}), recovering_to_closed(F) end || F <- WalFiles], % get all the recovered tables and insert them into closed Closed = lists:append([C || {C, _, _} <- All]), true = ets:insert(ra_log_closed_mem_tables, 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(TblWriter, M, F) || {_, M, F} <- All], FileNum = extract_file_num(lists:reverse(WalFiles)), State = roll_over(ra_log_recover_mem_tables, #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(ra_log_open_mem_tables), true = ets:delete_all_objects(ra_log_open_mem_tables), % delete all open ets tables [true = ets:delete(Tid) || {_, _, _, Tid} <- Open], true = ets:delete(ra_log_recover_mem_tables), %% 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) -> T = case Trunc of true -> 1; false -> 0 end, case Cache of #{UId := BinId} -> {<>, 2, WriterCache}; _ -> % TODO: check overflows of Next % cache the last 23 bits of the header word BinId = <<1:1/unsigned, Next:22/unsigned>>, IdDataLen = byte_size(UId), Prefix = <>, MarkerId = [Prefix, UId], {MarkerId, 4 + IdDataLen, {Next + 1, Cache#{UId => BinId}}} end. write_data({UId, _} = Id, Idx, Term, Data0, Trunc, #state{conf = #conf{compute_checksums = ComputeChecksum}, file_size = FileSize, wal = #wal{max_size = MaxWalSize, writer_name_cache = Cache0} = Wal} = State00) -> EntryData = to_binary(Data0), EntryDataLen = byte_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, % if the next write is going to exceed the configured max wal size % we roll over to a new wal. case FileSize + DataSize > MaxWalSize of true -> State = roll_over(State00), % TODO: there is some redundant computation performed by % recursing here it probably doesn't matter as it only happens % when a wal file fills up write_data(Id, Idx, Term, Data0, Trunc, State); false -> State0 = State00#state{wal = Wal#wal{writer_name_cache = Cache}}, Entry = [<>, EntryData], Checksum = case ComputeChecksum of true -> erlang:adler32(Entry); false -> 0 end, Record = [HeaderData, <>, 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 `~p`, " "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{file_size = FileSize, batch = Batch0, writers = Writers} = State, {UId, Pid}, Idx, Term, Entry, DataSize, Data, Truncate) -> Batch = incr_batch(ra_log_open_mem_tables, Batch0, UId, Pid, {Idx, Term}, Data, Entry, Truncate), State#state{file_size = FileSize + DataSize, batch = Batch, writers = Writers#{UId => {in_seq, Idx}} }. incr_batch(OpnMemTbl, #batch{writes = Writes, waiting = Waiting0, pending = Pend} = Batch, UId, Pid, {Idx, Term}, Data, Entry, Truncate) -> Waiting = case Waiting0 of #{Pid := #batch_writer{tbl_start = TblStart0, tid = _Tid, from = From, inserts = Inserts0} = W} -> % ct:pal("insertingf ~w ~w", [Tid, Idx]), % true = ets:insert(Tid, {Idx, Term, Entry}), TblStart = case Truncate of true -> Idx; false -> % take the min of the First item in % case we are overwriting before % the previously first seen entry min(TblStart0, Idx) end, Inserts = [{Idx, Term, Entry} | Inserts0], 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, _To, T}] -> {T, case Truncate of true -> Idx; false -> min(TblStart0, Idx) end}; _ -> %% there is no table so need %% to open one T = open_mem_table(UId), true = ets:insert_new( OpnMemTbl, {UId, Idx, Idx, T}), {T, Idx} end, % ct:pal("inserting ~w ~w", [Tid, Idx]), % true = ets:insert(Tid, {Idx, Term, Entry}), Writer = #batch_writer{tbl_start = TblStart, from = Idx, to = Idx, tid = Tid, uid = UId, term = Term, inserts = [{Idx, Term, Entry}]}, Waiting0#{Pid => Writer} end, Batch#batch{writes = Writes + 1, waiting = Waiting, pending = [Pend | Data]}. update_mem_table(OpnMemTbl, 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 = case Truncate of true -> Idx; false -> % take the min of the First item in case we are % overwriting before the previously first seen entry min(From0, Idx) end, % 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(UId), true = ets:insert_new(OpnMemTbl, {UId, Idx, Idx, Tid}), true = ets:insert(Tid, {Idx, Term, Entry}) end. roll_over(State0) -> State = complete_batch(State0), roll_over(ra_log_open_mem_tables, start_batch(State)). roll_over(OpnMemTbls, #state{wal = Wal0, file_num = Num0, conf = #conf{dir = Dir, max_size_bytes = MaxBytes, segment_writer = SegWriter} = Conf0} = State0) -> Num = Num0 + 1, Fn = ra_lib:zpad_filename("", "wal", Num), NextFile = filename:join(Dir, Fn), ?DEBUG("wal: opening new file ~p~n", [Fn]), %% if this is the first wal since restart randomise the first %% max wal size to reduce the likelyhood 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), ok = close_open_mem_tables(OpnMemTbls, Wal0#wal.filename, SegWriter), 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 ra_file_handle:open(File, Modes) of {ok, Fd} -> ok = write_header(Fd), % many platforms implement O_SYNC a bit like O_DSYNC % perform a manual sync here to ensure metadata is flushed ok = ra_file_handle:sync(Fd), {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} = Conf) -> {ok, Fd} = ra_file_handle:open(File, Modes), ok = write_header(Fd), {Conf, #wal{fd = Fd, max_size = Max, filename = File}}. write_header(Fd) -> ok = ra_file_handle:write(Fd, <>), ok = ra_file_handle:write(Fd, <>). close_file(undefined) -> ok; close_file(Fd) -> ok = ra_file_handle:sync(Fd), ra_file_handle:close(Fd). close_open_mem_tables(OpnMemTbls, Filename, TblWriter) -> MemTables = ets:tab2list(OpnMemTbls), % 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(ra_log_closed_mem_tables, 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(Filename) -> MemTables = ets:tab2list(ra_log_recover_mem_tables), Closed = [begin M = erlang:unique_integer([monotonic, positive]), erlang:insert_element(2, T, M) end || T <- MemTables], true = ets:delete_all_objects(ra_log_recover_mem_tables), {Closed, MemTables, Filename}. open_mem_table({UId, _Pid}) -> open_mem_table(UId); open_mem_table(UId) -> % lookup the locally registered name of the process to use as ets % name ServerName = ra_directory:name_of(UId), Tid = ets:new(ServerName, [set, {read_concurrency, true}, public]), % immediately give away ownership to ets process true = ra_log_ets:give_away(Tid), Tid. start_batch(State) -> State#state{batch = #batch{start_time = os:system_time(microsecond)}}. flush_pending(#state{wal = #wal{fd = Fd}, batch = #batch{pending = Pend} = Batch, 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; o_sync -> ok = ra_file_handle:write(Fd, Pend) end, State0#state{batch = Batch#batch{pending = []}}. complete_batch(#state{batch = undefined} = State) -> State; complete_batch(#state{batch = #batch{waiting = Waiting, writes = NumWrites, start_time = ST}, metrics_cursor = Cursor } = State00) -> TS = os:system_time(microsecond), State0 = flush_pending(State00), SyncTS = os:system_time(microsecond), _ = ets:update_element(ra_log_wal_metrics, Cursor, {2, {NumWrites, TS-ST, SyncTS-TS}}), NextCursor = (Cursor + 1) rem ?METRICS_WINDOW_SIZE, State = State0#state{metrics_cursor = NextCursor, batch = undefined}, %% process writers _ = maps:map(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(ra_log_open_mem_tables, UId, [{2, TblStart}, {3, To}]), Pid ! {ra_log_event, {written, {From, To, Term}}}, ok end, Waiting), State. wal2list(File) -> Data = open_existing(File), dump_records(Data, []). open_existing(File) -> case file:read_file(File) of {ok, <>} -> %% the only version currently supported Data; {ok, <>} -> exit({unknown_wal_file_format, Magic, UnknownVersion}) end. dump_records(<<_:1/unsigned, 0:1/unsigned, _:22/unsigned, IdDataLen:16/unsigned, _:IdDataLen/binary, _: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 dump_records(Rest, [{Idx, Term, binary_to_term(EntryData)} | Entries]); dump_records(<<_:1/unsigned, 1:1/unsigned, _:22/unsigned, _:32/integer, EntryDataLen:32/unsigned, Idx:64/unsigned, Term:64/unsigned, EntryData:EntryDataLen/binary, Rest/binary>>, Entries) -> dump_records(Rest, [{Idx, Term, binary_to_term(EntryData)} | Entries]); dump_records(<<>>, Entries) -> Entries. try_recover_records(Data, Cache) -> try recover_records(Data, Cache) of ok -> ok catch _:_ = Err -> ?WARN("wal: encountered error during recovery: ~w~n" "Continuing.~n", [Err]), ok end. recover_records(<>, Cache) -> % first writer appearance in WAL true = validate_and_update(UId, Checksum, Idx, Term, EntryData, Trunc), % TODO: recover writers info, i.e. last index seen recover_records(Rest, Cache#{IdRef => {UId, <<1:1/unsigned, IdRef:22/unsigned>>}}); recover_records(<>, Cache) -> #{IdRef := {UId, _}} = Cache, true = validate_and_update(UId, Checksum, Idx, Term, EntryData, Trunc), % TODO: recover writers info, i.e. last index seen recover_records(Rest, Cache); recover_records(<<>>, _Cache) -> ok. validate_and_update(UId, Checksum, Idx, Term, EntryData, Trunc) -> validate_checksum(Checksum, Idx, Term, EntryData), true = update_mem_table(ra_log_recover_mem_tables, UId, Idx, Term, binary_to_term(EntryData), Trunc =:= 1). 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(<>) of Checksum -> ok; _ -> exit(wal_checksum_validation_failure) end. merge_conf_defaults(Conf) -> maps:merge(#{segment_writer => ra_log_segment_writer, max_size_bytes => ?WAL_MAX_SIZE_BYTES, compute_checksums => true, write_strategy => default, sync_method => datasync}, Conf). to_binary(Term) -> term_to_binary(Term).