Make all catch Exception call CheckFailFast unless they rethrow (#1620)

* make all catch Exception call CheckFailFast unless they rethrow
asynkron · May 27, 2022 · f47b751 · f47b751
1 parent 2e74b6a
commit f47b751
Show file tree

Hide file tree

Showing 24 changed files with 75 additions and 47 deletions.
diff --git a/src/Proto.Actor/Context/ActorContext.cs b/src/Proto.Actor/Context/ActorContext.cs
@@ -242,6 +242,8 @@ public Task Receive(MessageEnvelope envelope)
 
     public void EscalateFailure(Exception reason, object? message)
     {
+        reason.CheckFailFast();
+
         if (System.Config.DeveloperSupervisionLogging)
         {
             Console.WriteLine($"[Supervision] Actor {Self} : {Actor.GetType().Name} failed with message:{message} exception:{reason}");

diff --git a/src/Proto.Actor/EventStream/EventStream.cs b/src/Proto.Actor/EventStream/EventStream.cs
@@ -232,7 +232,8 @@ public void Publish(T msg)
                     }
                     catch (Exception ex)
                     {
-                        _logger.LogWarning(0, ex, "Exception has occurred when publishing a message.");
+                        ex.CheckFailFast();
+                        _logger.LogError(0, ex, "Exception has occurred when publishing a message");
                     }
 
                     return Task.CompletedTask;

diff --git a/src/Proto.Actor/Extensions.cs b/src/Proto.Actor/Extensions.cs
@@ -3,6 +3,7 @@
 //      Copyright (C) 2015-2022 Asynkron AB All rights reserved
 // </copyright>
 // -----------------------------------------------------------------------
+using System;
 using System.Collections.Generic;
 using JetBrains.Annotations;
 using Proto.Mailbox;
@@ -31,4 +32,12 @@ out TValue value
         key = self.Key;
         value = self.Value;
     }
+
+    public static void CheckFailFast(this Exception? reason)
+    {
+        if (reason is not OutOfMemoryException) return;
+
+        Console.WriteLine("[Fatal] Out of memory exception" + reason);
+        Environment.FailFast(reason.Message, reason);
+    }
 }
diff --git a/src/Proto.Actor/Mailbox/BatchingMailbox.cs b/src/Proto.Actor/Mailbox/BatchingMailbox.cs
@@ -91,6 +91,7 @@ private async Task RunAsync()
         }
         catch (Exception x)
         {
+            x.CheckFailFast();
             _suspended = true;
             _invoker.EscalateFailure(x, currentMessage);
         }

diff --git a/src/Proto.Actor/Mailbox/Mailbox.cs b/src/Proto.Actor/Mailbox/Mailbox.cs
@@ -236,7 +236,7 @@ private ValueTask ProcessMessages()
 
                 if (msg is not null)
                 {
-                    var t= _invoker.InvokeUserMessageAsync(msg);
+                    var t = _invoker.InvokeUserMessageAsync(msg);
 
                     if (!t.IsCompletedSuccessfully)
                     {
@@ -254,6 +254,7 @@ private ValueTask ProcessMessages()
         }
         catch (Exception e)
         {
+            e.CheckFailFast();
             _invoker.EscalateFailure(e, msg);
         }
         return default;
@@ -263,13 +264,15 @@ static async ValueTask Await(object msg, ValueTask task, DefaultMailbox self)
             try
             {
                 await task;
+
                 foreach (var t1 in self._stats)
                 {
                     t1.MessageReceived(msg);
                 }
             }
             catch (Exception e)
             {
+                e.CheckFailFast();
                 self._invoker.EscalateFailure(e, msg);
             }
         }

diff --git a/src/Proto.Actor/Supervision/AlwaysRestartStrategy.cs b/src/Proto.Actor/Supervision/AlwaysRestartStrategy.cs
@@ -17,6 +17,5 @@ public void HandleFailure(
         RestartStatistics rs,
         Exception reason,
         object? message
-    )
-        => supervisor.RestartChildren(reason, child);
+    ) => supervisor.RestartChildren(reason, child);
 }
diff --git a/src/Proto.Actor/Utils/Retry.cs b/src/Proto.Actor/Utils/Retry.cs
@@ -43,6 +43,7 @@ public static async Task<T> TryUntil<T>(Func<Task<T>> body, Func<T?,bool> condit
             }
             catch (Exception x)
             {
+                x.CheckFailFast();
                 onError?.Invoke(i, x);
 
                 if (i == retryCount - 1)
@@ -105,6 +106,7 @@ public static async Task Try(
             }
             catch (Exception x)
             {
+                x.CheckFailFast();
                 onError?.Invoke(i, x);
 
                 if (i == retryCount - 1)

diff --git a/src/Proto.Actor/Utils/TaskFactory.cs b/src/Proto.Actor/Utils/TaskFactory.cs
@@ -29,6 +29,7 @@ public static async Task Run(Func<Task> body, CancellationToken cancellationToke
         }
         catch (Exception x)
         {
+            x.CheckFailFast();
             Logger.LogError(x, "Unhandled exception in async job {Job}", name);
         }
     }

diff --git a/src/Proto.Cluster/DefaultClusterContext.cs b/src/Proto.Cluster/DefaultClusterContext.cs
@@ -168,6 +168,7 @@ private async ValueTask RemoveFromSource(ClusterIdentity clusterIdentity, PidSou
         }
         catch (Exception e)
         {
+            e.CheckFailFast();
             if (context.System.Shutdown.IsCancellationRequested) return default;
 
             if (_requestLogThrottle().IsOpen())
@@ -213,6 +214,7 @@ IFuture future
         }
         catch (Exception x)
         {
+            x.CheckFailFast();
             if (!context.System.Shutdown.IsCancellationRequested && _requestLogThrottle().IsOpen())
                 if (Logger.IsEnabled(LogLevel.Debug)) Logger.LogDebug(x, "TryRequestAsync failed with exception, PID from {Source}", source);
             _pidCache.RemoveByVal(clusterIdentity, pid);

diff --git a/src/Proto.Cluster/ExperimentalClusterContext.cs b/src/Proto.Cluster/ExperimentalClusterContext.cs
@@ -118,6 +118,7 @@ public ExperimentalClusterContext(Cluster cluster)
                 }
                 catch (Exception x)
                 {
+                    x.CheckFailFast();
                     if (!context.System.Shutdown.IsCancellationRequested && _requestLogThrottle().IsOpen())
                         if (Logger.IsEnabled(LogLevel.Debug)) Logger.LogDebug(x, "TryRequestAsync failed with exception, PID from {Source}", source);
                     _pidCache.RemoveByVal(clusterIdentity, pid);
@@ -203,6 +204,7 @@ private async ValueTask RemoveFromSource(ClusterIdentity clusterIdentity, PidSou
         }
         catch (Exception e)
         {
+            e.CheckFailFast();
             if (context.System.Shutdown.IsCancellationRequested) return default;
 
             if (_requestLogThrottle().IsOpen())

diff --git a/src/Proto.Cluster/Gossip/GossipStateManagement.cs b/src/Proto.Cluster/Gossip/GossipStateManagement.cs
@@ -129,53 +129,45 @@ Func<T, TV> extractValue
     {
         var logger = ctx?.Logger()?.BeginMethodScope();
 
-        try
+        if (state.Members.Count == 0)
         {
-            if (state.Members.Count == 0)
-            {
-                logger?.LogDebug("No members found for consensus check");
-                return (false, default);
-            }
+            logger?.LogDebug("No members found for consensus check");
+            return (false, default);
+        }
 
-            logger?.LogDebug("Checking consensus");
+        logger?.LogDebug("Checking consensus");
 
-            if (!state.Members.TryGetValue(myId, out var ownMemberState))
-            {
-                logger?.LogDebug("I can't find myself");
-                return (false, default);
-            }
+        if (!state.Members.TryGetValue(myId, out var ownMemberState))
+        {
+            logger?.LogDebug("I can't find myself");
+            return (false, default);
+        }
+
+        var ownValue = GetConsensusValue(ownMemberState);
 
-            var ownValue = GetConsensusValue(ownMemberState);
+        if (ownValue is null)
+        {
+            logger?.LogDebug("I don't have any value for {Key}", valueKey);
+            return (false, default);
+        }
 
-            if (ownValue is null)
+        foreach (var (memberId, memberState) in state.Members)
+        {
+            //skip blocked members
+            if (!members.Contains(memberId))
             {
-                logger?.LogDebug("I don't have any value for {Key}", valueKey);
-                return (false, default);
+                logger?.LogDebug("Member is not part of cluster {MemberId}", memberId);
+                continue;
             }
 
-            foreach (var (memberId, memberState) in state.Members)
-            {
-                //skip blocked members
-                if (!members.Contains(memberId))
-                {
-                    logger?.LogDebug("Member is not part of cluster {MemberId}", memberId);
-                    continue;
-                }
+            var consensusValue = GetConsensusValue(memberState);
 
-                var consensusValue = GetConsensusValue(memberState);
+            if (consensusValue is null || !ownValue.Equals(consensusValue)) return (false, default);
+        }
 
-                if (consensusValue is null || !ownValue.Equals(consensusValue)) return (false, default);
-            }
+        if (Logger.IsEnabled(LogLevel.Debug)) Logger.LogDebug("Reached Consensus {Key}:{Value} - {State}", valueKey, ownValue, state);
+        return (true, ownValue);
 
-            if (Logger.IsEnabled(LogLevel.Debug)) Logger.LogDebug("Reached Consensus {Key}:{Value} - {State}", valueKey, ownValue, state);
-            return (true, ownValue);
-        }
-        catch (Exception x)
-        {
-            logger?.LogError(x, "Check Consensus failed");
-            Logger.LogError(x, "Check Consensus failed");
-            return (false, default);
-        }
 
         TV? GetConsensusValue(GossipState.Types.GossipMemberState memberState)
         {

diff --git a/src/Proto.Cluster/Gossip/Gossiper.cs b/src/Proto.Cluster/Gossip/Gossiper.cs
@@ -182,6 +182,7 @@ private async Task GossipLoop()
             }
             catch (Exception x)
             {
+                x.CheckFailFast();
                 Logger.LogError(x, "Gossip loop failed");
             }
         }
@@ -381,11 +382,9 @@ private async Task SendStateAsync()
         catch (OperationCanceledException)
         {
         }
-#pragma warning disable RCS1075
-        catch (Exception)
-#pragma warning restore RCS1075
+        catch (Exception x)
         {
-            //TODO: log
+            x.CheckFailFast();
         }
     }
 

diff --git a/src/Proto.Cluster/Member/MemberList.cs b/src/Proto.Cluster/Member/MemberList.cs
@@ -310,9 +310,10 @@ public void BroadcastEvent(object message, bool includeSelf = true)
             {
                 _system.Root.Send(pid, message);
             }
-            catch (Exception)
+            catch (Exception x)
             {
-                Logger.LogError("[MemberList] Failed to broadcast {Message} to {Pid}", message, pid);
+                x.CheckFailFast();
+                Logger.LogError(x, "[MemberList] Failed to broadcast {Message} to {Pid}", message, pid);
             }
         }
     }

diff --git a/src/Proto.Cluster/Partition/PartitionIdentityActor.cs b/src/Proto.Cluster/Partition/PartitionIdentityActor.cs
@@ -616,6 +616,7 @@ TaskCompletionSource<ActivationResponse> setResponse
             }
             catch (Exception x)
             {
+                x.CheckFailFast();
                 Logger.LogError(x, "[PartitionIdentity] Spawn failed");
                 _deltaTopology = null; // Do not use delta handover if we are not sure all spawns are OK.
             }

diff --git a/src/Proto.Cluster/Partition/PartitionIdentityLookup.cs b/src/Proto.Cluster/Partition/PartitionIdentityLookup.cs
@@ -109,6 +109,7 @@ public PartitionIdentityLookup(PartitionConfig? config)
         }
         catch (Exception e)
         {
+            e.CheckFailFast();
             Logger.LogError(e, "[PartitionIdentity] Error occured requesting remote PID {@Request}, identity Owner {Owner}", req, identityOwner);
             return null;
         }

diff --git a/src/Proto.Cluster/Partition/PartitionPlacementActor.cs b/src/Proto.Cluster/Partition/PartitionPlacementActor.cs
@@ -345,6 +345,7 @@ private Task OnActivationRequest(IContext context, ActivationRequest msg)
         }
         catch (Exception e)
         {
+            e.CheckFailFast();
             Logger.LogError(e, "[PartitionIdentity] Failed to spawn {Kind}/{Identity}", msg.Kind, msg.Identity);
             var response = new ActivationResponse
             {

diff --git a/src/Proto.Cluster/PartitionActivator/PartitionActivatorLookup.cs b/src/Proto.Cluster/PartitionActivator/PartitionActivatorLookup.cs
@@ -66,6 +66,7 @@ public PartitionActivatorLookup(TimeSpan getPidTimeout)
         }
         catch (Exception e)
         {
+            e.CheckFailFast();
             Logger.LogError(e, "[PartitionActivator] Error occured requesting remote PID {@Request}, identity Owner {Owner}", req, owner);
             return null;
         }

diff --git a/src/Proto.Cluster/PubSub/BatchingProducer.cs b/src/Proto.Cluster/PubSub/BatchingProducer.cs
@@ -102,6 +102,7 @@ private async Task PublisherLoop(CancellationToken cancel)
         }
         catch (Exception e)
         {
+            e.CheckFailFast();
             if (_logThrottle().IsOpen())
                 Logger.LogError(e, "Error in the publisher loop of Producer for topic {Topic}", _topic);
 

diff --git a/src/Proto.Cluster/Seed/SeedClientNodeActor.cs b/src/Proto.Cluster/Seed/SeedClientNodeActor.cs
@@ -48,6 +48,7 @@ private async Task OnConnect(IContext context)
             }
             catch (Exception x)
             {
+                x.CheckFailFast();
                 Logger.LogError(x, "Failed to connect to seed node {Host}:{Port}", host, port);
             }
         }
@@ -78,6 +79,7 @@ private async Task OnClusterTopology(IContext context, ClusterTopology clusterTo
                 }
                 catch (Exception e)
                 {
+                    e.CheckFailFast();
                     Logger.LogError(e, "Failed to connect to seed node {Member}", member.Address);
                 }
             }

diff --git a/src/Proto.Cluster/Seed/SeedNodeActor.cs b/src/Proto.Cluster/Seed/SeedNodeActor.cs
@@ -61,6 +61,7 @@ private async Task OnConnect(IContext context)
                 }
                 catch (Exception x)
                 {
+                    x.CheckFailFast();
                     Logger.LogError(x, "Failed to connect to seed node {Host}:{Port}", host, port);
                 }
             }

diff --git a/src/Proto.Remote/Endpoints/Endpoint.cs b/src/Proto.Remote/Endpoints/Endpoint.cs
@@ -138,6 +138,7 @@ private int DropMessagesInBatch(RemoteMessage remoteMessage)
                     }
                     catch (Exception ex)
                     {
+                        ex.CheckFailFast();
                         if (_logger.IsEnabled(_deserializationErrorLogLevel))
                             _logger.Log(
                                 _deserializationErrorLogLevel,
@@ -312,6 +313,7 @@ private async Task RunAsync()
             }
             catch (Exception ex)
             {
+                ex.CheckFailFast();
                 _logger.LogError(ex, "[{SystemAddress}] Error in RunAsync", System.Address);
             }
         }

diff --git a/src/Proto.Remote/Endpoints/EndpointReader.cs b/src/Proto.Remote/Endpoints/EndpointReader.cs
@@ -51,8 +51,9 @@ await responseStream.WriteAsync(new RemoteMessage
                                }
                            ).ConfigureAwait(false);
                        }
-                       catch (Exception)
+                       catch (Exception x)
                        {
+                           x.CheckFailFast();
                            Logger.LogWarning("[EndpointReader][{SystemAddress}] Failed to write disconnect message to the stream", _system.Address);
                        }
                    }
@@ -148,6 +149,7 @@ await responseStream.WriteAsync(new RemoteMessage
                             }
                             catch (Exception e)
                             {
+                                e.CheckFailFast();
                                 Logger.LogWarning(e, "[EndpointReader][{SystemAddress}] Writing error to {SystemId}", _system.Address, systemId);
                             }
                         }

diff --git a/src/Proto.Remote/Endpoints/RemoteMessageHandler.cs b/src/Proto.Remote/Endpoints/RemoteMessageHandler.cs
@@ -85,6 +85,7 @@ public void HandleRemoteMessage(RemoteMessage currentMessage)
                     }
                     catch (Exception ex)
                     {
+                        ex.CheckFailFast();
                         if (_logger.IsEnabled(_deserializationErrorLogLevel))
                             _logger.Log(
                                 _deserializationErrorLogLevel,

diff --git a/src/Proto.Remote/Endpoints/ServerConnector.cs b/src/Proto.Remote/Endpoints/ServerConnector.cs
@@ -246,6 +246,7 @@ await call.RequestStream.WriteAsync(new RemoteMessage
             }
             catch (Exception e)
             {
+                e.CheckFailFast();
                 if (actorSystemId is not null && _system.Remote().BlockList.IsBlocked(actorSystemId))
                 {
                     _logger.LogDebug("[ServerConnector][{SystemAddress}] dropped connection to blocked member {ActorSystemId}/{Address}", _system.Address, actorSystemId, _address);
-Original file line number
+Diff line change
@@ Expand Up @@
             }
             catch (Exception x)
             {
+                x.CheckFailFast();
                 Logger.LogError(x, "Unhandled exception in async job {Job}", name);
             }
         }
@@ Expand Down @@