Simplify peer selection actions

Removed some code which tested peer status variable with various
'wasWarm' 'isCold' 'not notCold' booleans and made the flow
more explicit for dealing with races between peer selection activities
and 'peerMonitoringLoop' which separately call into the relevant
functions in 'PeerSelectionActions'

Author: crocodile-dentist

ouroboros-network/src/Ouroboros/Network/PeerSelection/PeerStateActions.hs

@@ -1,7 +1,9 @@
+{-# LANGUAGE BlockArguments      #-}
 {-# LANGUAGE DataKinds           #-}
 {-# LANGUAGE FlexibleContexts    #-}
 {-# LANGUAGE GADTs               #-}
 {-# LANGUAGE KindSignatures      #-}
+{-# LANGUAGE LambdaCase          #-}
 {-# LANGUAGE NamedFieldPuns      #-}
 {-# LANGUAGE RankNTypes          #-}
 {-# LANGUAGE ScopedTypeVariables #-}
@@ -34,7 +36,7 @@ module Ouroboros.Network.PeerSelection.PeerStateActions
 import Control.Applicative (Alternative)
 import Control.Concurrent.Class.MonadSTM.Strict
 import Control.Exception (SomeAsyncException (..), assert)
-import Control.Monad (when, (<=<))
+import Control.Monad (join, when, (<=<))
 import Control.Monad.Class.MonadAsync
 import Control.Monad.Class.MonadFork
 import Control.Monad.Class.MonadThrow
@@ -622,14 +624,10 @@ withPeerStateActions PeerStateActionsArguments {
          then return False
          else writeTVar stateVar newState >> return True
 
-    isNotCoolingOrCold :: StrictTVar m PeerStatus -> STM m Bool
-    isNotCoolingOrCold stateVar =
-      (> PeerCooling) <$> readTVar stateVar
-
     peerMonitoringLoop
       :: PeerConnectionHandle muxMode responderCtx peerAddr versionData ByteString m a b
       -> m ()
-    peerMonitoringLoop pch@PeerConnectionHandle { pchConnectionId, pchPeerStatus, pchAppHandles } = do
+    peerMonitoringLoop pch@PeerConnectionHandle { pchConnectionId, pchPeerStatus, pchAppHandles, pchPromotedHotVar } = do
         -- A first-to-finish synchronisation on all the bundles; As a result
         -- this is a first-to-finish synchronisation between all the
         -- mini-protocols runs toward the given peer.
@@ -671,15 +669,18 @@ withPeerStateActions PeerStateActionsArguments {
 
           Just (WithSomeProtocolTemperature (WithHot MiniProtocolError{})) -> do
             -- current `pchPeerStatus` must be 'HotPeer'
-            state <- atomically $ do
-              peerState <- readTVar pchPeerStatus
-              _  <- updateUnlessCoolingOrCold pchPeerStatus PeerCooling
-              return peerState
+            (pchPromotedHot, state) <- atomically $ do
+              (,) <$> stateTVar pchPromotedHotVar (, Nothing) <*> readTVar pchPeerStatus <* updateUnlessCoolingOrCold pchPeerStatus PeerCooling
+            case pchPromotedHot of
+              Just t1 -> do
+                dt <- diffTime <$> getMonotonicTime <*> pure t1
+                traceWith spsTracer (PeerHotDuration pchConnectionId dt)
+              Nothing -> pure ()
             case state of
               PeerCold    -> return ()
               PeerCooling -> return ()
-              hotOrWarm -> assert (hotOrWarm == PeerHot) $
-                           traceWith spsTracer (PeerStatusChanged (HotToCooling pchConnectionId))
+              hotOrInDemotion -> assert (hotOrInDemotion >= PeerWarmWait) $
+                                   traceWith spsTracer (PeerStatusChanged (HotToCooling pchConnectionId))
             peerMonitoringLoop pch
           Just (WithSomeProtocolTemperature (WithWarm MiniProtocolError{})) -> do
             -- current `pchPeerStatus` must be 'WarmPeer'
@@ -689,13 +690,21 @@ withPeerStateActions PeerStateActionsArguments {
           Just (WithSomeProtocolTemperature (WithEstablished MiniProtocolError{})) -> do
             -- update 'pchPeerStatus' and log (as the two other transition to
             -- cold state.
-            state <- atomically $ do
-              peerState <- readTVar pchPeerStatus
-              _  <- updateUnlessCoolingOrCold pchPeerStatus PeerCooling
-              pure peerState
+            (state, pchPromotedHot) <- atomically $ do
+              state <- readTVar pchPeerStatus <* updateUnlessCoolingOrCold pchPeerStatus PeerCooling
+              case state of
+                PeerHot -> (state,) <$> stateTVar pchPromotedHotVar (,Nothing)
+                _otherwise -> return (state, Nothing)
+            case pchPromotedHot of
+              Just t1 -> do
+                dt <- diffTime <$> getMonotonicTime <*> pure t1
+                traceWith spsTracer (PeerHotDuration pchConnectionId dt)
+              Nothing -> pure ()
+
             case state of
-              PeerCold    -> return ()
-              PeerCooling -> return ()
+              PeerCold     -> return ()
+              PeerCooling  -> return ()
+              PeerWarmWait -> return () -- ^ the relevant trace will be performed by deactivatePeerConnection
               PeerWarm    -> traceWith spsTracer (PeerStatusChanged (WarmToCooling pchConnectionId))
               PeerHot     -> traceWith spsTracer (PeerStatusChanged (HotToCooling pchConnectionId))
             peerMonitoringLoop pch
@@ -919,35 +928,22 @@ withPeerStateActions PeerStateActionsArguments {
             pchPeerStatus,
             pchAppHandles,
             pchPromotedHotVar } = do
-      -- quiesce warm peer protocols and set hot ones in 'Continue' mode.
-      wasWarm <- atomically $ do
-        -- if the peer is cold we can't activate it.
-        notCold <- isNotCoolingOrCold pchPeerStatus
-        when notCold $ do
-          writeTVar (getControlVar SingHot pchAppHandles) Continue
-          writeTVar (getControlVar SingWarm pchAppHandles) Quiesce
-        return notCold
-      when (not wasWarm) $ do
-        traceWith spsTracer (PeerStatusChangeFailure
-                              (WarmToHot pchConnectionId)
-                              ActiveCold)
-        throwIO $ ColdActivationException pchConnectionId
-
-      -- start hot peer protocols
-      startProtocols SingHot isBigLedgerPeer connHandle
-
-      -- Only set the status to PeerHot if the peer isn't PeerCold.
-      -- This can happen asynchronously between the check above and now.
-      wasWarm' <- atomically $ updateUnlessCoolingOrCold pchPeerStatus PeerHot
-      if wasWarm'
-         then do
-           atomically . writeTVar pchPromotedHotVar . (Just $!) =<< getMonotonicTime
-           traceWith spsTracer (PeerStatusChanged (WarmToHot pchConnectionId))
-         else do
-           traceWith spsTracer (PeerStatusChangeFailure
-                                 (WarmToHot pchConnectionId)
-                                 ActiveCold)
-           throwIO $ ColdActivationException pchConnectionId
+      join . atomically $ do
+        peerStatus <- readTVar pchPeerStatus
+        case peerStatus of
+          PeerWarm -> do
+            writeTVar (getControlVar SingHot pchAppHandles) Continue
+            writeTVar (getControlVar SingWarm pchAppHandles) Quiesce
+            writeTVar pchPeerStatus PeerHot
+            return $ do
+              startProtocols SingHot isBigLedgerPeer connHandle
+              atomically . writeTVar pchPromotedHotVar . (Just $!) =<< getMonotonicTime
+              traceWith spsTracer (PeerStatusChanged (WarmToHot pchConnectionId))
+          _otherwise -> return $ do
+            traceWith spsTracer (PeerStatusChangeFailure
+                                  (WarmToHot pchConnectionId)
+                                  (ActiveCold peerStatus))
+            throwIO $ ColdActivationException pchConnectionId
 
 
     -- Take a hot peer and demote it to a warm one.
@@ -960,75 +956,65 @@ withPeerStateActions PeerStateActionsArguments {
             pchAppHandles,
             pchPromotedHotVar
           } = do
-      wasCold <- atomically $ do
-        notCold <- isNotCoolingOrCold pchPeerStatus
-        when notCold $ do
-          writeTVar (getControlVar SingHot pchAppHandles) Terminate
-          writeTVar (getControlVar SingWarm pchAppHandles) Continue
-        return (not notCold)
-      when wasCold $ do
-        -- The governor attempted to demote an already cold peer.
-        traceWith spsTracer (PeerStatusChangeFailure
-                             (HotToWarm pchConnectionId)
-                             ActiveCold)
-        throwIO $ ColdDeactivationException pchConnectionId
-
-
-      -- Hot protocols should stop within 'spsDeactivateTimeout'.
-      res <-
-        timeout spsDeactivateTimeout
-                (atomically $ awaitAllResults SingHot pchAppHandles)
-
-      pchPromotedHot <- atomically . stateTVar pchPromotedHotVar $ (, Nothing)
-      case pchPromotedHot of
-        Just t1 -> do
-          dt <- diffTime <$> getMonotonicTime <*> pure t1
-          traceWith spsTracer (PeerHotDuration pchConnectionId dt)
-        Nothing -> pure ()
-
-      case res of
-        Nothing -> do
-          Mux.stop pchMux
-          atomically (writeTVar pchPeerStatus PeerCooling)
-          traceWith spsTracer (PeerStatusChangeFailure
-                                (HotToCooling pchConnectionId)
-                                TimeoutError)
-          throwIO (DeactivationTimeout pchConnectionId)
-
-        -- some of the hot mini-protocols errored
-        Just (SomeErrored errs) -> do
-          -- we don't need to notify the connection manager, we can instead
-          -- relay on mux property: if any of the mini-protocols errors, mux
-          -- throws an exception as well.
-          atomically (writeTVar pchPeerStatus PeerCooling)
-          traceWith spsTracer (PeerStatusChangeFailure
-                                (HotToCooling pchConnectionId)
-                                (ApplicationFailure errs))
-          throwIO (MiniProtocolExceptions errs)
-
-        -- all hot mini-protocols succeeded
-        Just (AllSucceeded results) -> do
-          -- we don't notify the connection manager as this connection is still
-          -- useful to the outbound governor (warm peer).
-          wasWarm <- atomically $ do
-            -- Only set the status to PeerWarm if the peer isn't cold
-            -- (can happen asynchronously).
-            notCold <- updateUnlessCoolingOrCold pchPeerStatus PeerWarm
-            when notCold $ do
-              -- We need to update hot protocols to indicate that they are not
-              -- running. Preserve the results returned by their previous
-              -- execution.
-              modifyTVar (getMiniProtocolsVar SingHot pchAppHandles)
-                         (\_ -> Map.map (pure . NotRunning . Right) results)
-            return notCold
-
-          if wasWarm
-             then traceWith spsTracer (PeerStatusChanged (HotToWarm pchConnectionId))
-             else do
-                 traceWith spsTracer (PeerStatusChangeFailure
-                                      (HotToWarm pchConnectionId)
-                                      ActiveCold)
-                 throwIO $ ColdDeactivationException pchConnectionId
+      join . atomically $ do
+        peerStatus <- readTVar pchPeerStatus
+        case peerStatus of
+          PeerHot -> do
+            writeTVar (getControlVar SingHot pchAppHandles) Terminate
+            writeTVar (getControlVar SingWarm pchAppHandles) Continue
+            pchPromotedHot <- stateTVar pchPromotedHotVar (, Nothing)
+            writeTVar pchPeerStatus PeerWarmWait
+            return $ do
+              -- Hot protocols should stop within 'spsDeactivateTimeout'.
+              res <- timeout spsDeactivateTimeout
+                      $ atomically $ do
+                          res <- awaitAllResults SingHot pchAppHandles
+                          res <$ case res of
+                            AllSucceeded results -> do
+                              modifyTVar (getMiniProtocolsVar SingHot pchAppHandles)
+                                         (\_ -> Map.map (pure . NotRunning . Right) results)
+                              writeTVar pchPeerStatus PeerWarm
+                            SomeErrored _ ->
+                              void $ updateUnlessCoolingOrCold pchPeerStatus PeerCooling
+
+              case pchPromotedHot of
+                Just t1 -> do
+                  dt <- diffTime <$> getMonotonicTime <*> pure t1
+                  traceWith spsTracer (PeerHotDuration pchConnectionId dt)
+                Nothing -> pure ()
+
+              case res of
+                Nothing -> do
+                  Mux.stop pchMux
+                  void . atomically $ updateUnlessCoolingOrCold pchPeerStatus PeerCooling
+                  traceWith spsTracer (PeerStatusChangeFailure
+                                        (HotToCooling pchConnectionId)
+                                        TimeoutError)
+                  throwIO (DeactivationTimeout pchConnectionId)
+                Just (SomeErrored errs) -> do
+                  traceWith spsTracer (PeerStatusChangeFailure
+                                        (HotToCooling pchConnectionId)
+                                        (ApplicationFailure errs))
+                  throwIO (MiniProtocolExceptions errs)
+                Just (AllSucceeded {}) -> do
+                  traceWith spsTracer (PeerStatusChanged (HotToWarm pchConnectionId))
+
+          -- either the peer monitoring loop or peer selection demotion lost the race
+          PeerWarmWait -> do
+            peerStatus' <- readTVar pchPeerStatus
+            check (peerStatus' /= PeerWarmWait)
+            return $ do
+              case peerStatus' of
+                PeerWarm -> return () -- ^ successful demotion by the winner
+                -- in this case the race winner traces the error
+                _otherwise -> throwIO $ ColdDeactivationException pchConnectionId
+
+          _otherwise ->
+            return $ do
+              traceWith spsTracer (PeerStatusChangeFailure
+                                    (HotToWarm pchConnectionId)
+                                    (ActiveCold peerStatus))
+              throwIO $ ColdDeactivationException pchConnectionId
 
 
     closePeerConnection :: PeerConnectionHandle muxMode responderCtx peerAddr versionData ByteString m a b
@@ -1041,64 +1027,56 @@ withPeerStateActions PeerStateActionsArguments {
             pchMux,
             pchPromotedHotVar
           } = do
-      atomically $ do
+      peerStatus <- atomically $ do
         writeTVar (getControlVar SingWarm pchAppHandles) Terminate
         writeTVar (getControlVar SingEstablished pchAppHandles) Terminate
         writeTVar (getControlVar SingHot pchAppHandles) Terminate
+        readTVar pchPeerStatus <* updateUnlessCoolingOrCold pchPeerStatus PeerCooling
+
+      case peerStatus of
+        ps@PeerCooling -> return ps
+        ps@PeerCold    -> return ps
+        _otherwise     -> do
+          res <-
+            timeout spsCloseConnectionTimeout
+                    (atomically $
+                      (\a b c -> a <> b <> c)
+                        -- note: we use last to finish on hot, warm and
+                        -- established mini-protocols since 'closePeerConnection'
+                        -- is also used by asynchronous demotions, not just
+                        -- /warm → cold/ transition.
+                        <$> awaitAllResults SingHot pchAppHandles
+                        <*> awaitAllResults SingWarm pchAppHandles
+                        <*> awaitAllResults SingEstablished pchAppHandles)
+
+          PeerCooling <$ case res of
+            Nothing -> do
+              -- timeout fired
+              Mux.stop pchMux
+              traceWith spsTracer (PeerStatusChangeFailure
+                                    (WarmToCooling pchConnectionId)
+                                    TimeoutError)
+
+            Just (SomeErrored errs) -> do
+              -- some mini-protocol errored
+              --
+              -- we don't need to notify the connection manager, we can instead
+              -- rely on mux property: if any of the mini-protocols errors, mux
+              -- throws an exception as well.
+              traceWith spsTracer (PeerStatusChangeFailure
+                                     (WarmToCooling pchConnectionId)
+                                     (ApplicationFailure errs))
+              throwIO (MiniProtocolExceptions errs)
+
+            Just AllSucceeded {} -> do
+              -- all mini-protocols terminated cleanly
+              --
+              -- 'unregisterOutboundConnection' could only fail to demote the peer if
+              -- connection manager would simultaneously promote it, but this is not
+              -- possible.
+              _ <- releaseOutboundConnection spsConnectionManager pchConnectionId
+              traceWith spsTracer (PeerStatusChanged (WarmToCooling pchConnectionId))
 
-      res <-
-        timeout spsCloseConnectionTimeout
-                (atomically $
-                  (\a b c -> a <> b <> c)
-                    -- note: we use last to finish on hot, warm and
-                    -- established mini-protocols since 'closePeerConnection'
-                    -- is also used by asynchronous demotions, not just
-                    -- /warm → cold/ transition.
-                    <$> awaitAllResults SingHot pchAppHandles
-                    <*> awaitAllResults SingWarm pchAppHandles
-                    <*> awaitAllResults SingEstablished pchAppHandles)
-
-
-      pchPromotedHot <- atomically . stateTVar pchPromotedHotVar $ (, Nothing)
-      case pchPromotedHot of
-        Just t1 -> do
-          dt <- diffTime <$> getMonotonicTime <*> pure t1
-          traceWith spsTracer (PeerHotDuration pchConnectionId dt)
-        Nothing -> pure ()
-
-      wasWarm <- atomically (updateUnlessCoolingOrCold pchPeerStatus PeerCooling)
-      case res of
-        Nothing -> do
-          -- timeout fired
-          Mux.stop pchMux
-          when wasWarm $
-            traceWith spsTracer (PeerStatusChangeFailure
-                                  (WarmToCooling pchConnectionId)
-                                  TimeoutError)
-          readTVarIO pchPeerStatus
-
-        Just (SomeErrored errs) -> do
-          -- some mini-protocol errored
-          --
-          -- we don't need to notify the connection manager, we can instead
-          -- rely on mux property: if any of the mini-protocols errors, mux
-          -- throws an exception as well.
-          when wasWarm $
-            traceWith spsTracer (PeerStatusChangeFailure
-                                  (WarmToCooling pchConnectionId)
-                                  (ApplicationFailure errs))
-          throwIO (MiniProtocolExceptions errs)
-
-        Just AllSucceeded {} -> do
-          -- all mini-protocols terminated cleanly
-          --
-          -- 'unregisterOutboundConnection' could only fail to demote the peer if
-          -- connection manager would simultaneously promote it, but this is not
-          -- possible.
-          when wasWarm $ do
-            _ <- releaseOutboundConnection spsConnectionManager pchConnectionId
-            traceWith spsTracer (PeerStatusChanged (WarmToCooling pchConnectionId))
-          readTVarIO pchPeerStatus
 
 --
 -- Utilities
@@ -1203,7 +1181,7 @@ data FailureType versionNumber =
     | HandleFailure !SomeException
     | MuxStoppedFailure
     | TimeoutError
-    | ActiveCold
+    | ActiveCold !PeerStatus
     | ApplicationFailure ![MiniProtocolException]
   deriving Show