Merge pull request #540 from rabbitmq/ra-539

ra:start_cluster/3: handle timeouts returned by ra_lib:parallel_partition/3

Author: kjnilsson

src/ra.erl

@@ -432,8 +432,7 @@ start_cluster(System, ServerConfigs)
     {error, cluster_not_formed}.
 start_cluster(System, [#{cluster_name := ClusterName} | _] = ServerConfigs,
               Timeout) when is_atom(System) ->
-    {Started, NotStarted} =
-        ra_lib:partition_parallel(
+    case ra_lib:partition_parallel(
             fun (C) ->
                 case start_server(System, C) of
                     ok  -> true;
@@ -442,39 +441,46 @@ start_cluster(System, [#{cluster_name := ClusterName} | _] = ServerConfigs,
                               [C, Err]),
                         false
                 end
-            end, ServerConfigs),
-    case Started of
-        [] ->
-            ?ERR("ra: failed to form a new cluster ~w. "
-                  "No servers were successfully started.",
-                  [ClusterName]),
-            {error, cluster_not_formed};
-        _ ->
+            end, ServerConfigs) of
+        {ok, Started, NotStarted} ->
+            case Started of
+                [] ->
+                    ?ERR("ra: failed to form a new cluster ~w. "
+                        "No servers were successfully started.",
+                        [ClusterName]),
+                    {error, cluster_not_formed};
+                _ ->
+                    StartedIds = sort_by_local([I || #{id := I} <- Started], []),
+                    NotStartedIds = [I || #{id := I} <- NotStarted],
+                    %% try triggering elections until one succeeds
+                    %% TODO: handle case where no election was successfully triggered
+                    {value, TriggeredId} = lists:search(fun (N) ->
+                                                                ok == trigger_election(N)
+                                                        end, StartedIds),
+                    %% the triggered id is likely to become the leader so try that first
+                    case members(TriggeredId,
+                                length(ServerConfigs) * Timeout) of
+                        {ok, _, Leader} ->
+                            ?INFO("ra: started cluster ~ts with ~b servers. "
+                                "~b servers failed to start: ~w. Leader: ~w",
+                                [ClusterName, length(ServerConfigs),
+                                length(NotStarted), NotStartedIds,
+                                Leader]),
+                            % we have a functioning cluster
+                            {ok, StartedIds, NotStartedIds};
+                        Err ->
+                            ?WARN("ra: failed to form new cluster ~w. "
+                                "Error: ~w", [ClusterName, Err]),
+                            _ = [force_delete_server(System, N) || N <- StartedIds],
+                            % we do not have a functioning cluster
+                            {error, cluster_not_formed}
+                    end
+            end;
+        {error, {partition_parallel_timeout, Started, _}} ->
             StartedIds = sort_by_local([I || #{id := I} <- Started], []),
-            NotStartedIds = [I || #{id := I} <- NotStarted],
-            %% try triggering elections until one succeeds
-            %% TODO: handle case where no election was successfully triggered
-            {value, TriggeredId} = lists:search(fun (N) ->
-                                                        ok == trigger_election(N)
-                                                end, StartedIds),
-            %% the triggered id is likely to become the leader so try that first
-            case members(TriggeredId,
-                         length(ServerConfigs) * Timeout) of
-                {ok, _, Leader} ->
-                    ?INFO("ra: started cluster ~ts with ~b servers. "
-                          "~b servers failed to start: ~w. Leader: ~w",
-                          [ClusterName, length(ServerConfigs),
-                           length(NotStarted), NotStartedIds,
-                           Leader]),
-                    % we have a functioning cluster
-                    {ok, StartedIds, NotStartedIds};
-                Err ->
-                    ?WARN("ra: failed to form new cluster ~w. "
-                          "Error: ~w", [ClusterName, Err]),
-                    _ = [force_delete_server(System, N) || N <- StartedIds],
-                    % we do not have a functioning cluster
-                    {error, cluster_not_formed}
-            end
+            ?WARN("ra: a member of cluster ~w failed to start within the expected time interval (~w)", [ClusterName, Timeout]),
+            _ = [force_delete_server(System, N) || N <- StartedIds],
+            {error, cluster_not_formed}
     end.
 
 %% @doc Starts an individual ra server of a cluster.

src/ra_lib.erl

@@ -298,16 +298,23 @@ derive_safe_string(S, Num) ->
          end,
      string:slice(F(string:next_grapheme(S), []), 0, Num).
 
+-spec partition_parallel(fun((any()) -> boolean()), [any()]) ->
+    {ok, [any()], [any()]} | {error, any()}.
 partition_parallel(F, Es) ->
     partition_parallel(F, Es, 60000).
 
+-spec partition_parallel(fun((any()) -> boolean()), [any()], timeout()) ->
+    {ok, [any()], [any()]} | {error, any()}.
 partition_parallel(F, Es, Timeout) ->
     Parent = self(),
     Running = [{spawn_monitor(fun() ->
                                       Parent ! {self(), F(E)}
                               end), E}
                || E <- Es],
-    collect(Running, {[], []}, Timeout).
+    case collect(Running, {[], []}, Timeout) of
+        {error, _} = E -> E;
+        {Successes, Failures} -> {ok, Successes, Failures}
+    end.
 
 collect([], Acc, _Timeout) ->
     Acc;
@@ -322,7 +329,7 @@ collect([{{Pid, MRef}, E} | Next], {Left, Right}, Timeout) ->
         {'DOWN', MRef, process, Pid, Reason} ->
             collect(Next, {Left, [{E, Reason} | Right]}, Timeout)
     after Timeout ->
-              exit(partition_parallel_timeout)
+        {error, {partition_parallel_timeout, Left, Right}}
     end.
 
 retry(Func, Attempts) ->
@@ -579,4 +586,11 @@ lists_detect_sort_test() ->
 
     ok.
 
+partition_parallel_test() ->
+    ?assertMatch({error, {partition_parallel_timeout, [], []}},
+                 partition_parallel(fun(_) ->
+                                        timer:sleep(infinity)
+                                    end, [1, 2, 3], 1000)),
+    ok.
+
 -endif.

src/ra_log_segment_writer.erl

@@ -160,11 +160,12 @@ handle_cast({mem_tables, Ranges, WalFile}, #state{data_dir = Dir,
                    end, [], Ranges),
 
     _ = [begin
-             {_, Failures} = ra_lib:partition_parallel(
-                               fun (TidRange) ->
-                                       ok = flush_mem_table_ranges(TidRange, State),
-                                       true
-                               end, Tabs, infinity),
+             {ok, _, Failures} =
+                ra_lib:partition_parallel(
+                    fun (TidRange) ->
+                            ok = flush_mem_table_ranges(TidRange, State),
+                            true
+                    end, Tabs, infinity),
              case Failures of
                  [] ->
                      %% this is what we expect

test/coordination_SUITE.erl

@@ -390,7 +390,7 @@ shrink_cluster_with_snapshot(Config) ->
     ClusterName = ?config(cluster_name, Config),
     Peers = start_peers([s1,s2,s3], PrivDir),
     ServerIds = server_ids(ClusterName, Peers),
-    [A, B, C] = ServerIds,
+    [_A, _B, _C] = ServerIds,
 
     Machine = {module, ?MODULE, #{}},
     {ok, _, []} = ra:start_cluster(?SYS, ClusterName, Machine, ServerIds),