Enable async_dist by default for Ra server processes.

And add a system config option for `async_dist`.

Async dist should reduce activitions of the resend protocol
in a loaded system where the distribution buffer sometimes fills.

src/ra_server_proc.erl

@@ -309,9 +309,11 @@ do_init(#{id := Id,
     MinBinVheapSize = maps:get(server_min_bin_vheap_size, SysConf,
                                ?MIN_BIN_VHEAP_SIZE),
     MinHeapSize = maps:get(server_min_heap_size, SysConf, ?MIN_BIN_VHEAP_SIZE),
+    AsyncDist = maps:get(async_dist, SysConf, true),
     process_flag(message_queue_data, MsgQData),
     process_flag(min_bin_vheap_size, MinBinVheapSize),
     process_flag(min_heap_size, MinHeapSize),
+    process_flag(async_dist, AsyncDist),
     %% wait for wal for a bit before initialising the server state and log
     #{cluster := Cluster} = ServerState = ra_server:init(Config),
     LogId = ra_server:log_id(ServerState),

src/ra_system.erl

@@ -59,7 +59,8 @@
                     server_recovery_strategy => undefined |
                                                registered |
                                                {module(), atom(), list()},
-                    machine_upgrade_strategy => ra_server:machine_upgrade_strategy()
+                    machine_upgrade_strategy => ra_server:machine_upgrade_strategy(),
+                    async_dist => boolean()
                    }.
 
 -export_type([
@@ -146,6 +147,7 @@ default_config() ->
       low_priority_commands_flush_size => LowPriorityCommandsFlushSize,
       low_priority_commands_in_memory_size => LowPriorityInMemSize,
       machine_upgrade_strategy => MachineUpgradeStrategy,
+      async_dist => true,
       names => #{wal => ra_log_wal,
                  wal_sup => ra_log_wal_sup,
                  log_sup => ra_log_sup,

test/ra_SUITE.erl

@@ -128,19 +128,17 @@ end_per_testcase(_TestCase, Config) ->
     Config.
 
 single_server_processes_command(Config) ->
-    % ok = logger:set_primary_config(level, all),
     Name = ?config(test_name, Config),
-    N1 = nth_server_name(Config, 1),
+    {RaName, _} = N1 = nth_server_name(Config, 1),
     ok = ra:start_server(default, Name, N1, add_machine(), []),
     ok = ra:trigger_election(N1),
     monitor(process, element(1, N1)),
-    % index is 2 as leaders commit a no-op entry on becoming leaders
-    % debugger:start(),
-    % int:i(ra_server_proc),
-    % int:break(ra_server_proc, 440),
+    ?assertMatch([{async_dist, true}],
+                 process_info(whereis(RaName), [async_dist])),
     {ok, 5, _} = ra:process_command(N1, 5, 2000),
     {ok, 10, _} = ra:process_command(N1, 5, 2000),
-    terminate_cluster([N1]).
+    terminate_cluster([N1]),
+    ok.
 
 pipeline_commands(Config) ->
     Name = ?config(test_name, Config),