nqe 0.5.0 → 0.6.0
raw patch · 9 files changed
+550/−484 lines, 9 filesdep +async
Dependencies added: async
Files
- CHANGELOG.md +6/−0
- nqe.cabal +5/−4
- src/Control/Concurrent/NQE/Conduit.hs +15/−6
- src/Control/Concurrent/NQE/Process.hs +194/−112
- src/Control/Concurrent/NQE/PubSub.hs +0/−80
- src/Control/Concurrent/NQE/Publisher.hs +112/−0
- src/Control/Concurrent/NQE/Supervisor.hs +124/−121
- src/NQE.hs +22/−8
- test/Spec.hs +72/−153
CHANGELOG.md view
@@ -4,6 +4,12 @@ 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.0+### Changed+- Overhaul entire API in a non-backwards-compatible way.+- Separate read/write from write-only mailbox types.+- Improve documentation.+ ## 0.5.0 ### Added - `Inbox` type is now comparable for equality.
nqe.cabal view
@@ -2,10 +2,10 @@ -- -- see: https://github.com/sol/hpack ----- hash: a5599fb449af1ba777de4bf50443ca71032893f717edec6dc78953d607e4145c+-- hash: df57adafd63b61ff84b1e42285daa2463bd77e5347ea2f53fc595f690d5d6108 name: nqe-version: 0.5.0+version: 0.6.0 synopsis: Concurrency library in the style of Erlang/OTP description: Minimalistic actor library inspired by Erlang/OTP with support for supervisor hierarchies and asynchronous messages, as well as abstractions for synchronous communication and easy management of TCP connections. category: Control@@ -29,7 +29,7 @@ exposed-modules: Control.Concurrent.NQE.Conduit Control.Concurrent.NQE.Process- Control.Concurrent.NQE.PubSub+ Control.Concurrent.NQE.Publisher Control.Concurrent.NQE.Supervisor NQE other-modules:@@ -56,7 +56,8 @@ test ghc-options: -threaded -rtsopts -with-rtsopts=-N build-depends:- base >=4.7 && <5+ async+ , base >=4.7 && <5 , bytestring , conduit , conduit-extra
src/Control/Concurrent/NQE/Conduit.hs view
@@ -1,16 +1,25 @@ {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE RankNTypes #-}-module Control.Concurrent.NQE.Conduit- ( conduitMailbox- ) where+{-|+Module : Control.Concurrent.NQE.Conduit+Copyright : No rights reserved+License : UNLICENSE+Maintainer : xenog@protonmail.com+Stability : experimental+Portability : POSIX -import Control.Concurrent.NQE.Process+Mix NQE processes with conduits for easy concurrent IO.+-}+module Control.Concurrent.NQE.Conduit where+ import Conduit+import Control.Concurrent.NQE.Process+import Data.Typeable --- | Consumes messages and sends them to a mailbox.+-- | Consumes messages from a 'Conduit' and sends them to a channel. conduitMailbox ::- (MonadIO m, Mailbox mbox msg)+ (MonadIO m, OutChan mbox, Typeable msg) => mbox msg -> ConduitT msg o m () conduitMailbox mbox = awaitForever (`send` mbox)
src/Control/Concurrent/NQE/Process.hs view
@@ -3,148 +3,204 @@ {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE RankNTypes #-}-module Control.Concurrent.NQE.Process- ( Mailbox(..)- , Inbox- , Reply- , Listen- , newInbox- , send- , receive- , query- , receiveMatch- , receiveMatchSTM- , mailboxEmpty- ) where+{-|+Module : Control.Concurrent.NQE.Process+Copyright : No rights reserved+License : UNLICENSE+Maintainer : xenog@protonmail.com+Stability : experimental+Portability : POSIX +This is the core of the NQE library. It is composed of code to deal with+processes and mailboxes. Processes represent concurrent threads that receive+messages via a mailbox, also referred to as a channel. NQE is inspired by+Erlang/OTP and it stands for “Not Quite Erlang”. A process is analogous to an+actor in Scala, or an object in the original (Alan Kay) sense of the word. To+implement synchronous communication NQE makes use of 'STM' actions embedded in+asynchronous messages.+-}+module Control.Concurrent.NQE.Process where+ import Control.Concurrent.Unique+import Data.Function import Data.Hashable import UnliftIO --- | STM action to reply to a synchronous request.-type Reply a = a -> STM ()---- | STM action for an event listener.+-- | 'STM' function that receives an event and does something with it. type Listen a = a -> STM () --- | Mailboxes are used to communicate with processes (actors). A process will--- usually listen in a loop for events entering its mailbox. A process is its--- mailbox, so it may be named as the process that it communicates with.------ >>> :m + Control.Monad NQE UnliftIO--- >>> registry <- newTQueueIO :: IO (TQueue String)--- >>> let run = receive registry >>= putStrLn . ("Registered: " ++)--- >>> withAsync run $ \a -> "Bruce Wayne" `send` registry >> wait a--- Registered: Bruce Wayne-class Eq (mbox msg) => Mailbox mbox msg where- -- | STM action that responds true if the mailbox is empty. Useful to avoid- -- blocking on an empty mailbox.+-- | Channel that only allows messages to be sent to it.+data Mailbox msg =+ forall mbox. (OutChan mbox) =>+ Mailbox !(mbox msg)+ !Unique++-- | Channel that allows to send or receive messages.+data Inbox msg =+ forall mbox. (OutChan mbox, InChan mbox) =>+ Inbox !(mbox msg)+ !Unique++instance Eq (Mailbox msg) where+ (==) = (==) `on` f+ where+ f (Mailbox _ u) = u++instance Eq (Inbox msg) where+ (==) = (==) `on` f+ where+ f (Inbox _ u) = u++-- | 'Async' handle and 'Mailbox' for a process.+data Process msg = Process+ { getProcessAsync :: Async ()+ , getProcessMailbox :: Mailbox msg+ } deriving Eq++-- | Class for implementation of an 'Inbox'.+class InChan mbox where+ -- | Are there messages queued? mailboxEmptySTM :: mbox msg -> STM Bool- -- | STM action that responds true if the mailbox is full and would block if- -- a new message is received.- mailboxFullSTM :: mbox msg -> STM Bool- -- | STM action to send a message to a mailbox. This is usually called from- -- a process that wishes to communicate with actor that owns the mailbox.- sendSTM :: msg -> mbox msg -> STM ()- -- | STM action to receive a message from a mailbox. This should be called- -- from the process that owns the mailbox.+ -- | Receive a message. receiveSTM :: mbox msg -> STM msg- -- | Put a message back in the mailbox so that it is the next one to be- -- received. Used for pattern matching.+ -- | Put a message in the mailbox such that it is received next. requeueSTM :: msg -> mbox msg -> STM () -instance Mailbox TQueue msg where+-- | Class for implementation of a 'Mailbox'.+class OutChan mbox where+ -- | Is this bounded channel full? Always 'False' for unbounded channels.+ mailboxFullSTM :: mbox msg -> STM Bool+ -- | Send a message to this channel.+ sendSTM :: msg -> mbox msg -> STM ()++instance InChan TQueue where mailboxEmptySTM = isEmptyTQueue- mailboxFullSTM = const $ return False- sendSTM msg = (`writeTQueue` msg) receiveSTM = readTQueue requeueSTM msg = (`unGetTQueue` msg) -instance Mailbox TBQueue msg where+instance OutChan TQueue where+ mailboxFullSTM _ = return False+ sendSTM msg = (`writeTQueue` msg)++instance InChan TBQueue where mailboxEmptySTM = isEmptyTBQueue- mailboxFullSTM = isFullTBQueue- sendSTM msg = (`writeTBQueue` msg) receiveSTM = readTBQueue requeueSTM msg = (`unGetTBQueue` msg) --- | Wrapped 'Mailbox' hiding its implementation.-data Inbox msg = forall mbox. (Mailbox mbox msg) => Inbox !(mbox msg) !Unique---- | Create a new 'Inbox' with a 'Unique' identifier inside. If you run this--- function more than once with the same 'Mailbox', its results will be--- different from the 'Eq' or 'Hashable' point of view.-newInbox :: (MonadIO m, Mailbox mbox msg) => mbox msg -> m (Inbox msg)-newInbox mbox = Inbox mbox <$> liftIO newUnique--instance Eq (Inbox msg) where- Inbox _ u1 == Inbox _ u2 = u1 == u2+instance OutChan TBQueue where+ mailboxFullSTM = isFullTBQueue+ sendSTM msg = (`writeTBQueue` msg) -instance Hashable (Inbox msg) where- hashWithSalt i (Inbox _ u) = hashWithSalt i u- hash (Inbox _ u) = hash u+instance OutChan Mailbox where+ mailboxFullSTM (Mailbox mbox _) = mailboxFullSTM mbox+ sendSTM msg (Mailbox mbox _) = msg `sendSTM` mbox -instance Mailbox Inbox msg where+instance InChan Inbox where mailboxEmptySTM (Inbox mbox _) = mailboxEmptySTM mbox- mailboxFullSTM (Inbox mbox _) = mailboxFullSTM mbox- sendSTM msg (Inbox mbox _) = msg `sendSTM` mbox receiveSTM (Inbox mbox _) = receiveSTM mbox requeueSTM msg (Inbox mbox _) = msg `requeueSTM` mbox --- | Send a message to a mailbox.-send :: (MonadIO m, Mailbox mbox msg) => msg -> mbox msg -> m ()+instance OutChan Inbox where+ mailboxFullSTM (Inbox mbox _) = mailboxFullSTM mbox+ sendSTM msg (Inbox mbox _) = msg `sendSTM` mbox++instance OutChan Process where+ mailboxFullSTM (Process _ mbox) = mailboxFullSTM mbox+ sendSTM msg (Process _ mbox) = msg `sendSTM` mbox++instance Hashable (Process msg) where+ hashWithSalt i (Process _ m) = hashWithSalt i m+ hash (Process _ m) = hash m++instance Hashable (Mailbox msg) where+ hashWithSalt i (Mailbox _ u) = hashWithSalt i u+ hash (Mailbox _ u) = hash u++-- | Get a send-only 'Mailbox' for an 'Inbox'.+inboxToMailbox :: Inbox msg -> Mailbox msg+inboxToMailbox (Inbox m u) = Mailbox m u++-- | Wrap a channel in an 'Inbox'+wrapChannel ::+ (MonadIO m, InChan mbox, OutChan mbox) => mbox msg -> m (Inbox msg)+wrapChannel mbox = Inbox mbox <$> liftIO newUnique++-- | Create an unbounded 'Inbox'.+newInbox :: MonadIO m => m (Inbox msg)+newInbox = newTQueueIO >>= \c -> wrapChannel c++-- | 'Inbox' with upper bound on number of allowed queued messages.+newBoundedInbox :: MonadIO m => Int -> m (Inbox msg)+newBoundedInbox i = newTBQueueIO i >>= \c -> wrapChannel c++-- | Send a message to a channel.+send :: (MonadIO m, OutChan mbox) => msg -> mbox msg -> m () send msg = atomically . sendSTM msg --- | Receive a message from the mailbox. This function should be called only by--- the process that owns the malibox.-receive ::- (MonadIO m, Mailbox mbox msg)- => mbox msg- -> m msg+-- | Receive a message from a channel.+receive :: (InChan mbox, MonadIO m) => mbox msg -> m msg receive mbox = receiveMatch mbox Just --- | Use a partially-applied message type that takes a `Reply a` as its last--- argument. This function will create the STM action for the response, send the--- message to a process and await for the STM action to be fulfilled before--- responding response. It implements synchronous communication with a process.------ Example:------ >>> :m + NQE UnliftIO--- >>> data Message = Square Integer (Reply Integer)--- >>> doubler <- newTQueueIO :: IO (TQueue Message)--- >>> let proc = receive doubler >>= \(Square i r) -> atomically $ r (i * i)--- >>> withAsync proc $ \_ -> Square 2 `query` doubler--- 4------ In this example the @Square@ constructor takes a 'Reply' action as its last--- argument. It is passed partially-applied to @query@, which adds a new 'Reply'--- action before sending it to the @doubler@ and then waiting for it. The--- doubler process will run the @Reply@ action n STM with the reply as its--- argument. In this case @i * i@.+-- | Send request to channel and wait for a response. The @request@ 'STM' action+-- will be created by this function. query ::- (MonadIO m, Mailbox mbox msg)- => (Reply a -> msg)- -> mbox msg- -> m a-query f mbox = do- box <- atomically newEmptyTMVar- f (putTMVar box) `send` mbox- atomically (takeTMVar box)+ (MonadIO m, OutChan mbox)+ => (Listen response -> request)+ -> mbox request+ -> m response+query f m = do+ r <- newEmptyTMVarIO+ f (putTMVar r) `send` m+ atomically $ takeTMVar r --- | Test all the messages in a mailbox against the supplied function and return--- the output of the function only when it is 'Just'. Will block until a message--- matches. All messages that did not match are left in the mailbox. Only call--- from process that owns mailbox.-receiveMatch ::- (MonadIO m, Mailbox mbox msg) => mbox msg -> (msg -> Maybe a) -> m a+-- | Do a 'query' but timeout after @u@ microseconds. Return 'Nothing' if+-- timeout reached.+queryU ::+ (MonadUnliftIO m, OutChan mbox)+ => Int+ -> (Listen response -> request)+ -> mbox request+ -> m (Maybe response)+queryU u f m = timeout u (query f m)++-- | Do a 'query' but timeout after @s@ seconds. Return 'Nothing' if+-- timeout reached.+queryS ::+ (MonadUnliftIO m, OutChan mbox)+ => Int+ -> (Listen response -> request)+ -> mbox request+ -> m (Maybe response)+queryS s f m = timeout (s * 1000 * 1000) (query f m)++-- | Test all messages in a channel against the supplied function and return the+-- first matching message. Will block until a match is found. Messages that do+-- not match remain in the channel.+receiveMatch :: (MonadIO m, InChan mbox) => mbox msg -> (msg -> Maybe a) -> m a receiveMatch mbox = atomically . receiveMatchSTM mbox --- | Match a message in the mailbox as an atomic STM action.-receiveMatchSTM ::- (Mailbox mbox msg)- => mbox msg+-- | Like 'receiveMatch' but with a timeout set at @u@ microseconds. Returns+-- 'Nothing' if timeout is reached.+receiveMatchU ::+ (MonadUnliftIO m, InChan mbox)+ => Int+ -> mbox msg -> (msg -> Maybe a)- -> STM a+ -> m (Maybe a)+receiveMatchU u mbox f = timeout u $ receiveMatch mbox f++-- | Like 'receiveMatch' but with a timeout set at @s@ seconds. Returns+-- 'Nothing' if timeout is reached.+receiveMatchS ::+ (MonadUnliftIO m, InChan mbox)+ => Int+ -> mbox msg+ -> (msg -> Maybe a)+ -> m (Maybe a)+receiveMatchS s mbox f = timeout (s * 1000 * 1000) $ receiveMatch mbox f++-- | Match a message in the channel as an atomic 'STM' action.+receiveMatchSTM :: InChan mbox => mbox msg -> (msg -> Maybe a) -> STM a receiveMatchSTM mbox f = go [] where go acc =@@ -155,10 +211,36 @@ return x Nothing -> go (msg : acc) --- | Check if the mailbox is empty.-mailboxEmpty :: (MonadIO m, Mailbox mbox msg) => mbox msg -> m Bool+-- | Check if the channel is empty.+mailboxEmpty :: (MonadIO m, InChan mbox) => mbox msg -> m Bool mailboxEmpty = atomically . mailboxEmptySTM --- | Put a message at the start of a mailbox, so that it is the next one read.-requeueListSTM :: (Mailbox mbox msg) => [msg] -> mbox msg -> STM ()+-- | Put a list of messages at the start of a channel, so that the last element+-- of the list is the next message to be received.+requeueListSTM :: InChan mbox => [msg] -> mbox msg -> STM () requeueListSTM xs mbox = mapM_ (`requeueSTM` mbox) xs++-- | Run a process in the background and pass it to a function. Stop the+-- background process once the function returns. Background process exceptions+-- are rethrown in the current thread.+withProcess ::+ MonadUnliftIO m => (Inbox msg -> m ()) -> (Process msg -> m a) -> m a+withProcess p f = do+ (i, m) <- newMailbox+ withAsync (p i) (\a -> link a >> f (Process a m))++-- | Run a process in the background and return the 'Process' handle. Background+-- process exceptions are rethrown in the current thread.+process :: MonadUnliftIO m => (Inbox msg -> m ()) -> m (Process msg)+process p = do+ (i, m) <- newMailbox+ a <- async $ p i+ link a+ return (Process a m)++-- | Create an unbounded inbox and corresponding mailbox.+newMailbox :: MonadUnliftIO m => m (Inbox msg, Mailbox msg)+newMailbox = do+ i <- newInbox+ let m = inboxToMailbox i+ return (i, m)
− src/Control/Concurrent/NQE/PubSub.hs
@@ -1,80 +0,0 @@-{-# LANGUAGE FlexibleContexts #-}-module Control.Concurrent.NQE.PubSub- ( Publisher- , publisher- , subscribe- , unsubscribe- , withPubSub- ) where--import Control.Applicative-import Control.Concurrent.NQE.Process-import Control.Monad.Reader-import Data.List-import UnliftIO---- | Subscribe or unsubscribe from an event publisher.-data ControlMsg event- = Subscribe (Inbox event)- | Unsubscribe (Inbox event)---- | Wrapper for a 'Mailbox' that can receive 'ControlMsg'.-type Publisher msg = Inbox (ControlMsg msg)---- | Subscribe an 'Inbox' to events from a 'Publisher'. Unsubscribe when action--- finishes. Pass a mailbox constructor.-withPubSub ::- (MonadUnliftIO m, Mailbox mbox event)- => Publisher event- -> m (mbox event) -- ^ mailbox creator for this subscription- -> (Inbox event -> m a) -- ^ unsubscribe when this action ends- -> m a-withPubSub pub sub = bracket acquire (unsubscribe pub)- where- acquire = do- i <- newInbox =<< sub- subscribe pub i- return i---- | Subscribe an 'Inbox' to a 'Publisher' generating events.-subscribe :: MonadIO m => Publisher event -> Inbox event -> m ()-subscribe pub sub = Subscribe sub `send` pub---- | Unsubscribe an 'Inbox' from a 'Publisher' events.-unsubscribe :: MonadIO m => Publisher event -> Inbox event -> m ()-unsubscribe pub sub = Unsubscribe sub `send` pub---- | Start a publisher that will receive events from an 'STM' action, and--- subscription requests from a 'Publisher' mailbox. Events will be forwarded--- atomically to all subscribers. Full mailboxes will not receive events.-publisher :: MonadIO m => Publisher event -> STM event -> m ()-publisher pub events = do- box <- newTVarIO []- runReaderT go box- where- go =- forever $ do- incoming <-- atomically $- Left <$> receiveSTM pub <|> Right <$> events- process incoming--process ::- (MonadIO m, MonadReader (TVar [Inbox event]) m)- => Either (ControlMsg event) event- -> m ()-process (Left (Subscribe sub)) = do- box <- ask- atomically (modifyTVar box (`union` [sub]))--process (Left (Unsubscribe sub)) = do- box <- ask- atomically (modifyTVar box (delete sub))--process (Right event) = do- box <- ask- atomically $ do- subs <- readTVar box- forM_ subs $ \sub -> do- full <- mailboxFullSTM sub- unless full $ event `sendSTM` sub
+ src/Control/Concurrent/NQE/Publisher.hs view
@@ -0,0 +1,112 @@+{-# LANGUAGE FlexibleContexts #-}+{-|+Module : Control.Concurrent.NQE.Publisher+Copyright : No rights reserved+License : UNLICENSE+Maintainer : xenog@protonmail.com+Stability : experimental+Portability : POSIX++A publisher is a process that forwards messages to subscribers. NQE publishers+are simple, and do not implement filtering directly, although that can be done+on the 'STM' 'Listen' actions that forward messages to subscribers.++If a subscriber has been added to a publisher using the 'subscribe' function, it+needs to be removed later using 'unsubscribe' when it is no longer needed, or+the publisher will continue calling its 'Listen' action in the future, likely+causing memory leaks.+-}+module Control.Concurrent.NQE.Publisher+ ( Subscriber+ , PublisherMessage(..)+ , Publisher+ , withSubscription+ , subscribe+ , unsubscribe+ , withPublisher+ , publisher+ , publisherProcess+ ) where++import Control.Concurrent.NQE.Process+import Control.Concurrent.Unique+import Control.Monad.Reader+import Data.Function+import Data.Hashable+import Data.List+import UnliftIO++-- | Handle of a subscriber to a process. Should be kept in order to+-- unsubscribe.+data Subscriber msg = Subscriber (Listen msg) Unique++instance Eq (Subscriber msg) where+ (==) = (==) `on` f+ where+ f (Subscriber _ u) = u++instance Hashable (Subscriber msg) where+ hashWithSalt i (Subscriber _ u) = hashWithSalt i u++-- | Messages that a publisher will take.+data PublisherMessage msg+ = Subscribe !(Listen msg) !(Listen (Subscriber msg))+ | Unsubscribe !(Subscriber msg)+ | Event msg++-- | Alias for a publisher process.+type Publisher msg = Process (PublisherMessage msg)++-- | Create a mailbox, subscribe it to a publisher and pass it to the supplied+-- function . End subscription when function returns.+withSubscription ::+ MonadUnliftIO m => Publisher msg -> (Inbox msg -> m a) -> m a+withSubscription pub f = do+ inbox <- newInbox+ let sub = subscribe pub (`sendSTM` inbox)+ unsub = unsubscribe pub+ bracket sub unsub $ \_ -> f inbox++-- | 'Listen' to events from a publisher.+subscribe :: MonadIO m => Publisher msg -> Listen msg -> m (Subscriber msg)+subscribe pub sub = Subscribe sub `query` pub++-- | Stop listening to events from a publisher. Must provide 'Subscriber' that+-- was returned from corresponding 'subscribe' action.+unsubscribe :: MonadIO m => Publisher msg -> Subscriber msg -> m ()+unsubscribe pub sub = Unsubscribe sub `send` pub++-- | Start a publisher in the background and pass it to a function. The+-- publisher will be stopped when the function function returns.+withPublisher :: MonadUnliftIO m => (Publisher msg -> m a) -> m a+withPublisher = withProcess publisherProcess++-- | Start a publisher in the background.+publisher :: MonadUnliftIO m => m (Publisher msg)+publisher = process publisherProcess++-- | Start a publisher in the current thread.+publisherProcess :: MonadUnliftIO m => Inbox (PublisherMessage msg) -> m ()+publisherProcess inbox = newTVarIO [] >>= runReaderT go+ where+ go = forever $ receive inbox >>= publisherMessage++-- | Internal function to dispatch a publisher message.+publisherMessage ::+ (MonadIO m, MonadReader (TVar [Subscriber msg]) m)+ => PublisherMessage msg+ -> m ()+publisherMessage (Subscribe sub r) =+ ask >>= \box -> do+ u <- liftIO newUnique+ let s = Subscriber sub u+ atomically $ do+ modifyTVar box (`union` [s])+ r s+publisherMessage (Unsubscribe sub) =+ ask >>= \box -> atomically (modifyTVar box (delete sub))+publisherMessage (Event event) =+ ask >>= \box ->+ atomically $+ readTVar box >>= \subs ->+ forM_ subs $ \(Subscriber sub _) -> sub event
src/Control/Concurrent/NQE/Supervisor.hs view
@@ -1,184 +1,187 @@-{-# LANGUAGE ExistentialQuantification #-}-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE GADTs #-}-{-# LANGUAGE MultiParamTypeClasses #-}+{-# 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++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- ( SupervisorMessage+ ( ChildAction+ , Child+ , SupervisorMessage+ , Supervisor , Strategy(..)+ , withSupervisor , supervisor+ , supervisorProcess , addChild , removeChild- , stopSupervisor ) where import Control.Applicative import Control.Concurrent.NQE.Process import Control.Monad-import Control.Monad.STM (catchSTM)+import Data.List import UnliftIO --- | A supervisor will start, stop and monitor processes.-data SupervisorMessage n- = MonadUnliftIO n =>- AddChild (n ())- (Reply (Async ()))- | RemoveChild (Async ())- | StopSupervisor+-- | Alias for child action to be executed asynchronously by supervisor.+type ChildAction = IO () +-- | Thread handler for child.+type Child = Async ()++-- | Send this message to a supervisor to add or remove a child.+data SupervisorMessage+ = 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 ((Async (), Either SomeException ()) -> STM ())- -- ^ run this 'STM' action when a process stops+ = Notify (Listen (Child, Maybe SomeException))+ -- ^ send a 'SupervisorNotif' to 'Mailbox' when child dies | KillAll- -- ^ kill all processes and propagate exception+ -- ^ kill all processes and propagate exception upstream | IgnoreGraceful- -- ^ ignore processes that stop without raising exceptions+ -- ^ ignore processes that stop without raising an exception | IgnoreAll- -- ^ do nothing and keep running if a process dies+ -- ^ keep running if a child dies and ignore it --- | Run a supervisor with a given 'Strategy' a 'Mailbox' to control it, and a--- list of children to launch. The list can be empty.-supervisor ::- (MonadUnliftIO m, Mailbox mbox (SupervisorMessage n), m ~ n)+-- | 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- -> mbox (SupervisorMessage n)- -> [n ()]+ -> (Supervisor -> m a)+ -> m a+withSupervisor = withProcess . supervisorProcess++-- | Run a supervisor as an asynchronous process.+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 ()-supervisor strat mbox children = do+supervisorProcess strat i = do state <- newTVarIO []- finally (go state) (down state)+ finally (loop state) (stopAll state) where- go state = do- mapM_ (startChild state) children- loop state loop state = do- e <-- atomically $- Right <$> receiveSTM mbox <|> Left <$> waitForChild 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- down state = do- as <- readTVarIO state- mapM_ cancel as --- | Internal action to wait for a child process to finish running.-waitForChild :: TVar [Async ()] -> STM (Async (), Either SomeException ())+-- | 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 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 sup c = RemoveChild c `query` sup++-- | Internal function to stop all children.+stopAll :: MonadUnliftIO m => TVar [Child] -> m ()+stopAll state = mask_ $ do+ 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 +-- | Internal function to process incoming supervisor message. processMessage ::- (MonadUnliftIO m)- => TVar [Async ()]- -> SupervisorMessage m- -> m Bool+ MonadUnliftIO m => TVar [Child] -> SupervisorMessage -> m Bool processMessage state (AddChild ch r) = do- a <- async ch- atomically $ do- modifyTVar' state (a:)- r a+ a <- startChild state ch+ atomically $ r a return True--processMessage state (RemoveChild a) = do- atomically (modifyTVar' state (filter (/= a)))- cancel a+processMessage state (RemoveChild a r) = do+ stopChild state a+ atomically $ r () return True -processMessage state StopSupervisor = do- as <- readTVarIO state- forM_ as (stopChild state)- return False-+-- | Internal function to run when a child process dies. processDead ::- (MonadIO m)- => TVar [Async ()]+ MonadUnliftIO m+ => TVar [Child] -> Strategy- -> (Async (), Either SomeException ())+ -> (Child, Either SomeException ()) -> m Bool processDead state IgnoreAll (a, _) = do- atomically (modifyTVar' state (filter (/= a)))+ atomically . modifyTVar' state $ filter (/= a) return True- processDead state KillAll (a, e) = do- as <- atomically $ do- modifyTVar' state (filter (/= a))- readTVar state- mapM_ (stopChild state) as+ atomically $ modifyTVar' state . filter $ (/= a)+ stopAll state case e of- Left x -> throwIO x+ Left x -> throwIO x Right () -> return False- processDead state IgnoreGraceful (a, Right ()) = do atomically (modifyTVar' state (filter (/= a))) return True- processDead state IgnoreGraceful (a, Left e) = do- as <- atomically $ do- modifyTVar' state (filter (/= a))- readTVar state- mapM_ (stopChild state) as+ atomically $ modifyTVar' state (filter (/= a))+ stopAll state throwIO e--processDead state (Notify notif) (a, e) = do- x <-- atomically $ do- modifyTVar' state (filter (/= a))- catchSTM (notif (a, e) >> return Nothing) $ \x ->- return $ Just (x :: SomeException)- case x of- Nothing -> return True- Just ex -> do- as <- readTVarIO state- forM_ as (stopChild state)- throwIO ex+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+ where+ me =+ case ee of+ Left e -> Just e+ Right () -> Nothing -- | Internal function to start a child process.-startChild ::- (MonadUnliftIO m)- => TVar [Async ()]- -> m ()- -> m (Async ())-startChild state run = do- a <- async run- atomically (modifyTVar' state (a:))+startChild :: MonadUnliftIO m => TVar [Child] -> ChildAction -> m Child+startChild state ch = mask_ $ do+ a <- liftIO $ async ch+ atomically $ modifyTVar' state (a :) return a -- | Internal fuction to stop a child process.-stopChild :: MonadIO m => TVar [Async ()] -> Async () -> m ()-stopChild state a = do+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)---- | Add a new child process to the supervisor. The child process will run in--- the supervisor context. Will return an 'Async' for the child. This function--- will not block or raise an exception if the child dies.-addChild ::- (MonadUnliftIO n, MonadIO m, Mailbox mbox (SupervisorMessage n))- => mbox (SupervisorMessage n)- -> n ()- -> m (Async ())-addChild mbox action = AddChild action `query` mbox---- | Stop a child process controlled by the supervisor. Must pass the child--- 'Async'. Will not wait for the child to die.-removeChild ::- (MonadUnliftIO n, MonadIO m, Mailbox mbox (SupervisorMessage n))- => mbox (SupervisorMessage n)- -> Async ()- -> m ()-removeChild mbox child = RemoveChild child `send` mbox---- | Stop the supervisor and its children.-stopSupervisor ::- (MonadUnliftIO n, MonadIO m, Mailbox mbox (SupervisorMessage n))- => mbox (SupervisorMessage n)- -> m ()-stopSupervisor mbox = StopSupervisor `send` mbox+ when isChild $ cancel a
src/NQE.hs view
@@ -1,11 +1,25 @@+{-|+Module : NQE+Copyright : No rights reserved+License : UNLICENSE+Maintainer : xenog@protonmail.com+Stability : experimental+Portability : POSIX++Concurrency library inspired by Erlang/OTP.+-} module NQE- ( module Control.Concurrent.NQE.Process- , module Control.Concurrent.NQE.Supervisor- , module Control.Concurrent.NQE.PubSub- , module Control.Concurrent.NQE.Conduit+ ( -- * Processes and Mailboxes+ module Process+ -- * Process Supervisors+ , module Supervisor+ -- * Publisher and Subscribers+ , module Publisher+ -- * Conduit Integration+ , module Conduit ) where -import Control.Concurrent.NQE.Conduit-import Control.Concurrent.NQE.Process-import Control.Concurrent.NQE.PubSub-import Control.Concurrent.NQE.Supervisor+import Control.Concurrent.NQE.Conduit as Conduit+import Control.Concurrent.NQE.Process as Process+import Control.Concurrent.NQE.Publisher as Publisher+import Control.Concurrent.NQE.Supervisor as Supervisor
test/Spec.hs view
@@ -3,191 +3,110 @@ {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE ScopedTypeVariables #-} import Conduit-import Control.Concurrent.STM (check)-import Control.Exception+import Control.Concurrent.Async (ExceptionInLinkedThread (..))+import Control.Concurrent.STM (check) import Control.Monad-import Data.ByteString (ByteString)-import Data.Conduit.Text (decode, encode, utf8)-import qualified Data.Conduit.Text as CT-import Data.Conduit.TMChan-import Data.Text (Text) import NQE import Test.Hspec import UnliftIO-import UnliftIO.Concurrent hiding (yield)+import UnliftIO.Concurrent hiding (yield) data Pong = Pong deriving (Eq, Show)-newtype Ping = Ping (Pong -> STM ())+newtype Ping = Ping (Listen Pong) data TestError = TestError1 | TestError2- | TestError3 deriving (Show, Eq) instance Exception TestError -pong :: TQueue Ping -> IO ()-pong mbox =- forever $ do- Ping reply <- receive mbox- atomically (reply Pong)--encoder :: MonadThrow m => ConduitT Text ByteString m ()-encoder = encode utf8+testError :: Selector TestError+testError = const True -decoder :: MonadThrow m => ConduitT ByteString Text m ()-decoder = decode utf8 .| CT.lines+testError1 :: Selector TestError+testError1 e = e == TestError1 -conduits ::- IO ( ConduitT () ByteString IO ()- , ConduitT ByteString Void IO ()- , ConduitT () ByteString IO ()- , ConduitT ByteString Void IO ())-conduits = do- inChan <- newTBMChanIO 2048- outChan <- newTBMChanIO 2048- return ( sourceTBMChan inChan- , sinkTBMChan outChan- , sourceTBMChan outChan- , sinkTBMChan inChan- )+threadError :: Exception e => Selector e -> Selector ExceptionInLinkedThread+threadError e (ExceptionInLinkedThread _ s) = maybe False e (fromException s) -pongServer ::- ConduitT () ByteString IO ()- -> ConduitT ByteString Void IO ()- -> (Async () -> IO a)- -> IO a-pongServer source sink go = do- mbox <- newTQueueIO- withAsync (in_pipe mbox) $ const $ withAsync (out_pipe mbox) go- where- in_pipe mbox = runConduit $ source .| decoder .| conduitMailbox mbox- out_pipe mbox = runConduit $ processor mbox .| encoder .| sink- processor mbox =- forever $- receive mbox >>= \case- ("ping" :: Text) -> yield ("pong\n" :: Text)- _ -> return ()+notifError :: Exception e => Selector e -> Selector (Maybe SomeException)+notifError e s = maybe False e (s >>= fromException) -pongClient :: ConduitT () ByteString IO ()- -> ConduitT ByteString Void IO ()- -> IO Text-pongClient source sink = do- mbox <- newTQueueIO- withAsync out_pipe $- const $ withAsync (in_pipe mbox) $ const $ receive mbox- where- in_pipe mbox = runConduit $ source .| decoder .| conduitMailbox mbox- out_pipe = runConduit $ generator .| encoder .| sink- generator = yield ("ping\n" :: Text)+pongServer :: MonadIO m => Inbox Ping -> m ()+pongServer mbox =+ forever $ do+ Ping r <- receive mbox+ atomically (r Pong) main :: IO () main = hspec $ do describe "two communicating processes" $ it "exchange ping/pong messages" $ do- mbox <- newTQueueIO- g <- withAsync (pong mbox) $ const $ query Ping mbox+ g <- withProcess pongServer (query Ping) g `shouldBe` Pong- describe "network process" $- it "responds to a ping" $ do- (source1, sink1, source2, sink2) <- conduits- msg <-- pongServer source1 sink1 $ const $ pongClient source2 sink2- msg `shouldBe` "pong"- describe "utilities" $ do- it "timeout action fails" $ do- n <- timeout 0xbeef (threadDelay 0xdeadbeef)- n `shouldBe` Nothing- it "timeout action succeeds" $ do- n <- timeout 0xdeadbeef (return (0xbeef :: Integer))- n `shouldBe` Just 0xbeef describe "supervisor" $ do- let p1 m = forever $ receive m >>= \r -> atomically $ r ()- p2 = query id+ let dummy = threadDelay $ 1000 * 1000 it "all processes end without failure" $ do- mbox <- newTQueueIO- sup <- newTQueueIO- g <- async $ supervisor KillAll sup [p1 mbox, p2 mbox]- wait g `shouldReturn` ()+ let action =+ withSupervisor KillAll $ \sup -> do+ addChild sup dummy+ addChild sup dummy+ threadDelay $ 100 * 1000+ action `shouldReturn` () it "one process crashes" $ do- mbox <- newTQueueIO- sup <- newTQueueIO- g <-- async $- supervisor- IgnoreGraceful- sup- [p1 mbox, p2 mbox >> throw TestError1]- wait g `shouldThrow` (== TestError1)+ let action =+ withSupervisor IgnoreGraceful $ \sup -> do+ addChild sup dummy+ addChild sup (throwIO TestError1)+ threadDelay $ 500 * 1000+ action `shouldThrow` threadError testError1 it "both processes crash" $ do- sup <- newTQueueIO- g <-- async $- supervisor- IgnoreGraceful- sup- [throw TestError1, throw TestError2]- wait g `shouldThrow` (\e -> e == TestError1 || e == TestError2)- it "process crashes ignored" $ do- sup <- newTQueueIO- g <-- async $- supervisor- IgnoreAll- sup- [throw TestError1, throw TestError2]- stopSupervisor sup- wait g `shouldReturn` ()+ let action =+ withSupervisor IgnoreGraceful $ \sup -> do+ addChild sup (throwIO TestError1)+ addChild sup (throwIO TestError2)+ threadDelay $ 500 * 1000+ action `shouldThrow` threadError testError it "monitors processes" $ do- sup <- newTQueueIO- mon <- newTQueueIO- g <-- async $- supervisor- (Notify (writeTQueue mon))- sup- [throw TestError1, throw TestError2]+ let rcv i = receive i :: IO (Maybe SomeException)+ (inbox, mailbox) <- newMailbox (t1, t2) <-- atomically $ (,) <$> readTQueue mon <*> readTQueue mon- let er e =- case e of- Right () -> False- Left x ->- case fromException x of- Just TestError1 -> True- Just TestError2 -> True- _ -> False- snd t1 `shouldSatisfy` er- snd t2 `shouldSatisfy` er- stopSupervisor sup- wait g `shouldReturn` ()+ withSupervisor (Notify ((`sendSTM` mailbox) . snd)) $ \sup -> do+ addChild sup (throwIO TestError1)+ addChild sup (throwIO TestError2)+ (,) <$> rcv inbox <*> rcv inbox+ t1 `shouldSatisfy` notifError testError+ t2 `shouldSatisfy` notifError testError describe "pubsub" $ do it "sends messages to all subscribers" $ do let msgs = words "hello world"- pub <- newTQueueIO >>= newInbox- events <- newTQueueIO- withAsync (publisher pub (receiveSTM events)) $- const $- withPubSub pub newTQueueIO $ \sub1 ->- withPubSub pub newTQueueIO $ \sub2 -> do- mapM_ (`send` events) msgs- sub1msgs <- replicateM 2 (receive sub1)- sub2msgs <- replicateM 2 (receive sub2)- sub1msgs `shouldBe` msgs- sub2msgs `shouldBe` msgs+ (inbox1, mbox1) <- newMailbox+ (inbox2, mbox2) <- newMailbox+ (msgs1, msgs2) <-+ withPublisher $ \pub -> do+ subscribe pub (`sendSTM` mbox1)+ subscribe pub (`sendSTM` mbox2)+ mapM_ ((`send` pub) . Event) msgs+ msgs1 <- replicateM 2 (receive inbox1)+ msgs2 <- replicateM 2 (receive inbox2)+ return (msgs1, msgs2)+ msgs1 `shouldBe` msgs+ msgs2 `shouldBe` msgs it "drops messages when bounded queue full" $ do let msgs = words "hello world drop"- pub <- newTQueueIO >>= newInbox- events <- newTQueueIO- withAsync (publisher pub (receiveSTM events)) $- const $- withPubSub pub (newTBQueueIO 2) $ \sub -> do- mapM_ (`send` events) msgs- atomically $ do- check =<< mailboxFullSTM sub- check =<< mailboxEmptySTM events- msgs' <- replicateM 2 (receive sub)- "meh" `send` events- msg <- receive sub- msgs' <> [msg] `shouldBe` words "hello world meh"+ let f mbox msg = mailboxFullSTM mbox >>= \case+ True -> return ()+ False -> msg `sendSTM` mbox+ msgs' <-+ withPublisher $ \pub -> do+ inbox <- newBoundedInbox 2+ subscribe pub (f (inboxToMailbox inbox))+ mapM_ ((`send` pub) . Event) msgs+ atomically $ check =<< mailboxFullSTM inbox+ threadDelay $ 250 * 1000+ msgs' <- replicateM 2 (receive inbox)+ Event "meh" `send` pub+ msg <- receive inbox+ return $ msgs' <> [msg]+ msgs' `shouldBe` words "hello world meh"