diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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. diff --git a/nqe.cabal b/nqe.cabal --- a/nqe.cabal +++ b/nqe.cabal @@ -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 diff --git a/src/Control/Concurrent/NQE/Conduit.hs b/src/Control/Concurrent/NQE/Conduit.hs --- a/src/Control/Concurrent/NQE/Conduit.hs +++ b/src/Control/Concurrent/NQE/Conduit.hs @@ -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) diff --git a/src/Control/Concurrent/NQE/Process.hs b/src/Control/Concurrent/NQE/Process.hs --- a/src/Control/Concurrent/NQE/Process.hs +++ b/src/Control/Concurrent/NQE/Process.hs @@ -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) diff --git a/src/Control/Concurrent/NQE/PubSub.hs b/src/Control/Concurrent/NQE/PubSub.hs deleted file mode 100644 --- a/src/Control/Concurrent/NQE/PubSub.hs +++ /dev/null @@ -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 diff --git a/src/Control/Concurrent/NQE/Publisher.hs b/src/Control/Concurrent/NQE/Publisher.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/NQE/Publisher.hs @@ -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 diff --git a/src/Control/Concurrent/NQE/Supervisor.hs b/src/Control/Concurrent/NQE/Supervisor.hs --- a/src/Control/Concurrent/NQE/Supervisor.hs +++ b/src/Control/Concurrent/NQE/Supervisor.hs @@ -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 diff --git a/src/NQE.hs b/src/NQE.hs --- a/src/NQE.hs +++ b/src/NQE.hs @@ -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 diff --git a/test/Spec.hs b/test/Spec.hs --- a/test/Spec.hs +++ b/test/Spec.hs @@ -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"