WAL: remove write strategies.

These have not proved useful o_sync perform very badly and
sync_after_notify only provides benefit when there is hardly
any activity. It is better to settle for a more predictable mode
like the default write + sync provides.

Author: kjnilsson

README.md

@@ -348,20 +348,6 @@ is available in a separate repository.
         <td>Indicate whether the wal should compute and validate checksums. Default: `true`</td>
         <td>Boolean</td>
     </tr>
-    <tr>
-        <td>wal_write_strategy</td>
-        <td>
-            <ul>
-                <li>
-                    <code>default</code>: used by default. <code>write(2)</code> system calls are delayed until a buffer is due to be flushed. Then it writes all the data in a single call then <code>fsync</code>s. Fastest option but incurs some additional memory use.
-                </li>
-                <li>
-                    <code>o_sync</code>: Like default but will try to open the file with O_SYNC and thus won't need the additional <code>fsync(2)</code> system call. If it fails to open the file with this flag this mode falls back to default.
-                </li>
-            </ul>
-        </td>
-        <td>Enumeration: <code>default</code> | <code>o_sync</code></td>
-    </tr>
     <tr>
         <td>wal_sync_method</td>
         <td>

src/ra.erl

@@ -85,8 +85,7 @@
     {wal_data_dir, file:filename()} |
     {segment_max_entries, non_neg_integer()} |
     {wal_max_size_bytes, non_neg_integer()} |
-    {wal_compute_checksums, boolean()} |
-    {wal_write_strategy, default | o_sync}.
+    {wal_compute_checksums, boolean()}.
 
 -type query_fun() :: fun((term()) -> term()) |
                      {M :: module(), F :: atom(), A :: list()}.

src/ra_log_sup.erl

@@ -65,7 +65,6 @@ make_wal_conf(#{data_dir := DataDir,
     MaxBatchSize = maps:get(wal_max_batch_size, Cfg,
                             ?WAL_DEFAULT_MAX_BATCH_SIZE),
     MaxEntries = maps:get(wal_max_entries, Cfg, undefined),
-    Strategy = maps:get(wal_write_strategy, Cfg, default),
     SyncMethod = maps:get(wal_sync_method, Cfg, datasync),
     HibAfter = maps:get(wal_hibernate_after, Cfg, undefined),
     Gc = maps:get(wal_garbage_collect, Cfg, false),
@@ -78,7 +77,6 @@ make_wal_conf(#{data_dir := DataDir,
       dir => WalDir,
       segment_writer => SegWriterName,
       compute_checksums => ComputeChecksums,
-      write_strategy => Strategy,
       max_size_bytes => MaxSizeBytes,
       max_entries => MaxEntries,
       sync_method => SyncMethod,

src/ra_log_wal.erl

@@ -67,16 +67,6 @@
                 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()}}.
 
@@ -86,7 +76,6 @@
                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(),
                mem_tables_tid :: ets:tid(),
@@ -139,7 +128,6 @@
                       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(),
@@ -149,8 +137,7 @@
                       min_bin_vheap_size => non_neg_integer()
                      }.
 
--export_type([wal_conf/0,
-              wal_write_strategy/0]).
+-export_type([wal_conf/0]).
 
 -type wal_command() ::
     {append | truncate, writer_id(), ra_index(), ra_term(), term()}.
@@ -261,7 +248,6 @@ init(#{dir := Dir} = Conf0) ->
       segment_writer := SegWriter,
       compute_checksums := ComputeChecksums,
       pre_allocate := PreAllocate,
-      write_strategy := WriteStrategy,
       sync_method := SyncMethod,
       garbage_collect := Gc,
       min_heap_size := MinHeapSize,
@@ -285,7 +271,6 @@ init(#{dir := Dir} = Conf0) ->
                  max_size_bytes = max(?WAL_MIN_SIZE, MaxWalSize),
                  max_entries = MaxEntries,
                  recovery_chunk_size = RecoveryChunkSize,
-                 write_strategy = WriteStrategy,
                  sync_method = SyncMethod,
                  counter = CRef,
                  mem_tables_tid = ets:whereis(MemTablesName),
@@ -299,7 +284,9 @@ init(#{dir := Dir} = Conf0) ->
             % generated during recovery
             ok = ra_log_segment_writer:await(SegWriter),
             {ok, Result}
-    catch _:Err:_Stack ->
+    catch _:Err:Stack ->
+              ?ERROR("WAL: failed to initialise with ~p, stack ~p",
+                     [Err, Stack]),
               {stop, Err}
     end.
 
@@ -322,16 +309,14 @@ terminate(Reason, State) ->
     _ = cleanup(State),
     ok.
 
-format_status(#state{conf = #conf{write_strategy = Strat,
-                                  sync_method = SyncMeth,
+format_status(#state{conf = #conf{sync_method = SyncMeth,
                                   compute_checksums = Cs,
                                   names = #{wal := WalName},
                                   max_size_bytes = MaxSize},
                      writers = Writers,
                      wal = #wal{file_size = FSize,
                                 filename = Fn}}) ->
-    #{write_strategy => Strat,
-      sync_method => SyncMeth,
+    #{sync_method => SyncMeth,
       compute_checksums => Cs,
       writers => maps:size(Writers),
       filename => filename:basename(Fn),
@@ -595,20 +580,6 @@ roll_over(#state{wal = Wal0, file_num = Num0,
                  wal = Wal,
                  file_num = Num}.
 
-open_wal(File, Max, #conf{write_strategy = o_sync} = Conf) ->
-        Modes = [sync | ?FILE_MODES],
-        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{} = Conf0) ->
     {ok, Fd} = prepare_file(File, ?FILE_MODES),
     Conf = maybe_pre_allocate(Conf0, Fd, Max),
@@ -637,9 +608,7 @@ make_tmp(File) ->
     ok = file:close(Fd),
     Tmp.
 
-maybe_pre_allocate(#conf{pre_allocate = true,
-                         write_strategy = Strat} = Conf, Fd, Max0)
-  when Strat /= o_sync ->
+maybe_pre_allocate(#conf{pre_allocate = true} = Conf, Fd, Max0) ->
     Max = Max0 - ?HEADER_SIZE,
     case file:allocate(Fd, ?HEADER_SIZE, Max) of
         ok ->
@@ -654,7 +623,7 @@ maybe_pre_allocate(#conf{pre_allocate = true,
                   " falling back to fsync instead of fdatasync", []),
             Conf#conf{pre_allocate = false}
     end;
-maybe_pre_allocate(Conf, _Fd, _Max) ->
+maybe_pre_allocate(Conf, _Fd, _Max0) ->
     Conf.
 
 close_file(undefined) ->
@@ -666,26 +635,11 @@ 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}}) ->
-    sync(Fd, SyncMeth);
-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 = file:write(Fd, Pend),
-            sync(Fd, SyncMeth);
-        _ ->
-            ok = file:write(Fd, Pend)
-    end,
+                     conf = #conf{sync_method = SyncMeth}} = State0) ->
+    ok = file:write(Fd, Pend),
+    sync(Fd, SyncMeth),
     State0#state{batch = undefined}.
 
 sync(_Fd, none) ->
@@ -709,7 +663,6 @@ complete_batch(#state{batch = #batch{waiting = Waiting,
     Ranges = maps:fold(fun (Pid, BatchWriter, Acc) ->
                                complete_batch_writer(Pid, BatchWriter, Acc)
                        end, Wal#wal.ranges, Waiting),
-    ok = post_notify_flush(State),
     State#state{wal = Wal#wal{ranges = Ranges}}.
 
 complete_batch_writer(Pid, #batch_writer{snap_idx = SnapIdx,
@@ -955,7 +908,6 @@ merge_conf_defaults(Conf) ->
                  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 => ?MIN_BIN_VHEAP_SIZE,

src/ra_system.erl

@@ -36,7 +36,6 @@
                     wal_compute_checksums => boolean(),
                     wal_max_batch_size => non_neg_integer(),
                     wal_max_entries => undefined | non_neg_integer(),
-                    wal_write_strategy => default | o_sync | sync_after_notify,
                     wal_sync_method => datasync | sync | none,
                     wal_hibernate_after => non_neg_integer(),
                     wal_garbage_collect => boolean(),
@@ -91,7 +90,6 @@ default_config() ->
     WalMaxBatchSize = application:get_env(ra, wal_max_batch_size,
                                           ?WAL_DEFAULT_MAX_BATCH_SIZE),
     WalMaxEntries = application:get_env(ra, wal_max_entries, undefined),
-    WalWriteStrategy = application:get_env(ra, wal_write_strategy, default),
     WalSyncMethod = application:get_env(ra, wal_sync_method, datasync),
     DataDir = ra_env:data_dir(),
     WalDataDir = application:get_env(ra, wal_data_dir, DataDir),
@@ -126,7 +124,6 @@ default_config() ->
       wal_compute_checksums => WalComputeChecksums,
       wal_max_batch_size => WalMaxBatchSize,
       wal_max_entries => WalMaxEntries,
-      wal_write_strategy => WalWriteStrategy,
       wal_garbage_collect => WalGarbageCollect,
       wal_pre_allocate => WalPreAllocate,
       wal_sync_method => WalSyncMethod,

test/ra_log_wal_SUITE.erl

@@ -17,8 +17,6 @@ all() ->
     [
      {group, default},
      {group, fsync},
-     {group, o_sync},
-     {group, sync_after_notify},
      {group, no_sync}
     ].
 
@@ -64,8 +62,6 @@ groups() ->
      {default, [], all_tests()},
      %% uses fsync instead of the default fdatasync
      {fsync, [], all_tests()},
-     {o_sync, [], all_tests()},
-     {sync_after_notify, [], all_tests()},
      {no_sync, [], all_tests()}
     ].
 
@@ -83,16 +79,16 @@ init_per_group(Group, Config) ->
     ra_directory:init(?SYS),
     ra_counters:init(),
     % application:ensure_all_started(lg),
-    {SyncMethod, WriteStrat} =
+    SyncMethod =
         case Group of
             fsync ->
-                {sync, default};
+                sync;
             no_sync ->
-                {none, default};
+                none;
             _ ->
-                {datasync, Group}
+                datasync
         end,
-    [{write_strategy, WriteStrat},
+    [
      {sys_cfg, SysCfg},
      {sync_method,  SyncMethod} | Config].
 
@@ -101,10 +97,9 @@ end_per_group(_, Config) ->
 
 init_per_testcase(TestCase, Config) ->
     PrivDir = ?config(priv_dir, Config),
-    G = ?config(write_strategy, Config),
     M = ?config(sync_method, Config),
     Sys = ?config(sys_cfg, Config),
-    Dir = filename:join([PrivDir, G, M, TestCase]),
+    Dir = filename:join([PrivDir, M, TestCase]),
     {ok, Ets} = ra_log_ets:start_link(Sys),
     ra_counters:init(),
     UId = atom_to_binary(TestCase, utf8),
@@ -114,7 +109,6 @@ init_per_testcase(TestCase, Config) ->
     WalConf = #{dir => Dir,
                 name => ra_log_wal,
                 names => Names,
-                write_strategy => G,
                 max_size_bytes => ?MAX_SIZE_BYTES},
     _ = ets:new(ra_log_snapshot_state,
                 [named_table, public, {write_concurrency, true}]),
@@ -758,8 +752,16 @@ sys_get_status(Config) ->
     Conf = ?config(wal_conf, Config),
     {_UId, _} = ?config(writer_id, Config),
     {ok, Pid} = ra_log_wal:start_link(Conf),
-    {_, _, _, [_, _, _, _, [_, _ ,S]]} = sys:get_status(ra_log_wal),
-    #{write_strategy := _} = S,
+    {_, _, _, [_, _, _, _, [_, _ , S]]} = sys:get_status(ra_log_wal),
+    ?assert(is_map(S)),
+
+    ?assertMatch(#{sync_method := _,
+                   compute_checksums := _,
+                   writers := _,
+                   filename := _,
+                   current_size := _,
+                   max_size_bytes := _,
+                   counters := _ }, S),
     proc_lib:stop(Pid),
     ok.