nqe 0.6.4 → 0.6.5
raw patch · 3 files changed
+129/−120 lines, 3 files
Files
- CHANGELOG.md +4/−0
- nqe.cabal +3/−3
- src/Control/Concurrent/NQE/Supervisor.hs +122/−117
CHANGELOG.md view
@@ -4,6 +4,10 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/) and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html). +## 0.6.5+### Fixed+- Don't call `waitAnyCatchSTM` with an empty list.+ ## 0.6.4 ### Fixed - Import `Control.Monad` to fix `mtl-2.3` issue.
nqe.cabal view
@@ -1,13 +1,13 @@ cabal-version: 1.12 --- This file has been generated from package.yaml by hpack version 0.34.4.+-- This file has been generated from package.yaml by hpack version 0.36.0. -- -- see: https://github.com/sol/hpack ----- hash: 3792a9b030f7697ae24ef34aef1c83768206ff91344fe288987266195c699f0a+-- hash: 294113b7ba7506e5e1487a0172b6daeaa115d1558af30149e09fa01cfdcfe7fc name: nqe-version: 0.6.4+version: 0.6.5 synopsis: Concurrency library in the style of Erlang/OTP description: Please see the README on GitHub at <https://github.com/jprupp/nqe#readme> category: Control
src/Control/Concurrent/NQE/Supervisor.hs view
@@ -1,41 +1,43 @@-{-# LANGUAGE ExistentialQuantification #-}-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE GADTs #-}-{-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE Rank2Types #-}-{-|-Module : Control.Concurrent.NQE.Supervisor-Copyright : No rights reserved-License : UNLICENSE-Maintainer : xenog@protonmail.com-Stability : experimental-Portability : POSIX+{-# LANGUAGE ExistentialQuantification #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE Rank2Types #-} -Supervisors run and monitor processes, including other supervisors. A supervisor-has a corresponding 'Strategy' that controls its behaviour if a child stops.-Supervisors deal with exceptions in concurrent processes so that their code does-not need to be written in an overly-defensive style. They help prevent problems-caused by processes dying quietly in the background, potentially locking an-entire application.--}+-- |+-- Module : Control.Concurrent.NQE.Supervisor+-- Copyright : No rights reserved+-- License : UNLICENSE+-- Maintainer : xenog@protonmail.com+-- Stability : experimental+-- Portability : POSIX+--+-- Supervisors run and monitor processes, including other supervisors. A supervisor+-- has a corresponding 'Strategy' that controls its behaviour if a child stops.+-- Supervisors deal with exceptions in concurrent processes so that their code does+-- not need to be written in an overly-defensive style. They help prevent problems+-- caused by processes dying quietly in the background, potentially locking an+-- entire application. module Control.Concurrent.NQE.Supervisor- ( ChildAction- , Child- , SupervisorMessage- , Supervisor- , Strategy(..)- , withSupervisor- , supervisor- , supervisorProcess- , addChild- , removeChild- ) where+ ( ChildAction,+ Child,+ SupervisorMessage,+ Supervisor,+ Strategy (..),+ withSupervisor,+ supervisor,+ supervisorProcess,+ addChild,+ removeChild,+ )+where -import Control.Applicative-import Control.Concurrent.NQE.Process-import Control.Monad-import Data.List-import UnliftIO+import Control.Applicative+import Control.Concurrent.NQE.Process+import Control.Concurrent.STM (retry)+import Control.Monad+import Data.List+import UnliftIO -- | Alias for child action to be executed asynchronously by supervisor. type ChildAction = IO ()@@ -45,143 +47,146 @@ -- | Send this message to a supervisor to add or remove a child. data SupervisorMessage- = AddChild !ChildAction- !(Listen Child)- | RemoveChild !Child- !(Listen ())+ = AddChild+ !ChildAction+ !(Listen Child)+ | RemoveChild+ !Child+ !(Listen ()) -- | Alias for supervisor process. type Supervisor = Process SupervisorMessage -- | Supervisor strategies to decide what to do when a child stops. data Strategy- = Notify (Listen (Child, Maybe SomeException))- -- ^ send a 'SupervisorNotif' to 'Mailbox' when child dies- | KillAll- -- ^ kill all processes and propagate exception upstream- | IgnoreGraceful- -- ^ ignore processes that stop without raising an exception- | IgnoreAll- -- ^ keep running if a child dies and ignore it+ = -- | send a 'SupervisorNotif' to 'Mailbox' when child dies+ Notify (Listen (Child, Maybe SomeException))+ | -- | kill all processes and propagate exception upstream+ KillAll+ | -- | ignore processes that stop without raising an exception+ IgnoreGraceful+ | -- | keep running if a child dies and ignore it+ IgnoreAll -- | Run a supervisor asynchronously and pass its mailbox to a function. -- Supervisor will be stopped along with all its children when the function -- ends. withSupervisor ::- MonadUnliftIO m- => Strategy- -> (Supervisor -> m a)- -> m a+ (MonadUnliftIO m) =>+ Strategy ->+ (Supervisor -> m a) ->+ m a withSupervisor = withProcess . supervisorProcess -- | Run a supervisor as an asynchronous process.-supervisor :: MonadUnliftIO m => Strategy -> m Supervisor+supervisor :: (MonadUnliftIO m) => Strategy -> m Supervisor supervisor strat = process (supervisorProcess strat) -- | Run a supervisor in the current thread. supervisorProcess ::- MonadUnliftIO m- => Strategy- -> Inbox SupervisorMessage- -> m ()+ (MonadUnliftIO m) =>+ Strategy ->+ Inbox SupervisorMessage ->+ m () supervisorProcess strat i = do- state <- newTVarIO []- finally (loop state) (stopAll state)+ state <- newTVarIO []+ finally (loop state) (stopAll state) where loop state = do- e <- atomically $ Right <$> receiveSTM i <|> Left <$> waitForChild state- again <-- case e of- Right m -> processMessage state m- Left x -> processDead state strat x- when again $ loop state+ e <- atomically $ Right <$> receiveSTM i <|> Left <$> waitForChild state+ again <-+ case e of+ Right m -> processMessage state m+ Left x -> processDead state strat x+ when again $ loop state -- | Add a new 'ChildAction' to the supervisor. Will return the 'Child' that was -- just started. This function will not block or raise an exception if the child -- dies.-addChild :: MonadIO m => Supervisor -> ChildAction -> m Child+addChild :: (MonadIO m) => Supervisor -> ChildAction -> m Child addChild sup action = AddChild action `query` sup -- | Stop a 'Child' controlled by this supervisor. Will block until the child -- dies.-removeChild :: MonadIO m => Supervisor -> Child -> m ()+removeChild :: (MonadIO m) => Supervisor -> Child -> m () removeChild sup c = RemoveChild c `query` sup -- | Internal function to stop all children.-stopAll :: MonadUnliftIO m => TVar [Child] -> m ()+stopAll :: (MonadUnliftIO m) => TVar [Child] -> m () stopAll state = mask_ $ do- as <- readTVarIO state- mapM_ cancel as+ as <- readTVarIO state+ mapM_ cancel as -- | Internal function to wait for a child process to finish running. waitForChild :: TVar [Child] -> STM (Child, Either SomeException ()) waitForChild state = do- as <- readTVar state- waitAnyCatchSTM as+ as <- readTVar state+ when (null as) retry+ waitAnyCatchSTM as -- | Internal function to process incoming supervisor message. processMessage ::- MonadUnliftIO m => TVar [Child] -> SupervisorMessage -> m Bool+ (MonadUnliftIO m) => TVar [Child] -> SupervisorMessage -> m Bool processMessage state (AddChild ch r) = do- a <- startChild state ch- atomically $ r a- return True+ a <- startChild state ch+ atomically $ r a+ return True processMessage state (RemoveChild a r) = do- stopChild state a- atomically $ r ()- return True+ stopChild state a+ atomically $ r ()+ return True -- | Internal function to run when a child process dies. processDead ::- MonadUnliftIO m- => TVar [Child]- -> Strategy- -> (Child, Either SomeException ())- -> m Bool+ (MonadUnliftIO m) =>+ TVar [Child] ->+ Strategy ->+ (Child, Either SomeException ()) ->+ m Bool processDead state IgnoreAll (a, _) = do- atomically . modifyTVar' state $ filter (/= a)- return True+ atomically . modifyTVar' state $ filter (/= a)+ return True processDead state KillAll (a, e) = do- atomically $ modifyTVar' state . filter $ (/= a)- stopAll state- case e of- Left x -> throwIO x- Right () -> return False+ atomically $ modifyTVar' state . filter $ (/= a)+ stopAll state+ case e of+ Left x -> throwIO x+ Right () -> return False processDead state IgnoreGraceful (a, Right ()) = do- atomically (modifyTVar' state (filter (/= a)))- return True+ atomically (modifyTVar' state (filter (/= a)))+ return True processDead state IgnoreGraceful (a, Left e) = do- atomically $ modifyTVar' state (filter (/= a))- stopAll state- throwIO e+ atomically $ modifyTVar' state (filter (/= a))+ stopAll state+ throwIO e processDead state (Notify notif) (a, ee) = do- atomically $ do- as <- readTVar state- case find (== a) as of- Just p -> notif (p, me)- Nothing -> return ()- modifyTVar state (filter (/= a))- return True+ atomically $ do+ as <- readTVar state+ case find (== a) as of+ Just p -> notif (p, me)+ Nothing -> return ()+ modifyTVar state (filter (/= a))+ return True where me =- case ee of- Left e -> Just e- Right () -> Nothing+ case ee of+ Left e -> Just e+ Right () -> Nothing -- | Internal function to start a child process.-startChild :: MonadUnliftIO m => TVar [Child] -> ChildAction -> m Child+startChild :: (MonadUnliftIO m) => TVar [Child] -> ChildAction -> m Child startChild state ch = mask_ $ do- a <- liftIO $ async ch- atomically $ modifyTVar' state (a :)- return a+ a <- liftIO $ async ch+ atomically $ modifyTVar' state (a :)+ return a -- | Internal fuction to stop a child process.-stopChild :: MonadUnliftIO m => TVar [Child] -> Child -> m ()+stopChild :: (MonadUnliftIO m) => TVar [Child] -> Child -> m () stopChild state a = mask_ $ do- isChild <-- atomically $ do- cur <- readTVar state- let new = filter (/= a) cur- writeTVar state new- return (cur /= new)- when isChild $ cancel a+ isChild <-+ atomically $ do+ cur <- readTVar state+ let new = filter (/= a) cur+ writeTVar state new+ return (cur /= new)+ when isChild $ cancel a