diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,5 +1,10 @@ # Changelog for extensible-effects-concurrent +## 0.1.2.1 + * Add more documentation + * Simplify Dispatcher API + * Make more exception safe + ## 0.1.2.0 * Add Observer module diff --git a/extensible-effects-concurrent.cabal b/extensible-effects-concurrent.cabal --- a/extensible-effects-concurrent.cabal +++ b/extensible-effects-concurrent.cabal @@ -1,5 +1,5 @@ name: extensible-effects-concurrent -version: 0.1.2.0 +version: 0.1.2.1 description: Please see the README on GitHub at synopsis: Message passing concurrency as extensible-effect homepage: https://github.com/sheyll/extensible-effects-concurrent#readme @@ -38,7 +38,7 @@ process, monad-control, random, - extensible-effects, + extensible-effects >= 2.4 && < 2.5, stm, tagged exposed-modules: diff --git a/src/Control/Eff/Concurrent/Dispatcher.hs b/src/Control/Eff/Concurrent/Dispatcher.hs --- a/src/Control/Eff/Concurrent/Dispatcher.hs +++ b/src/Control/Eff/Concurrent/Dispatcher.hs @@ -1,3 +1,20 @@ +-- | Implement Erlang style message passing concurrency. +-- +-- This handles the 'MessagePassing' and 'Process' effects, using +-- 'STM.TQueue's and 'forkIO'. +-- +-- This aims to be a pragmatic implementation, so even logging is +-- supported. +-- +-- At the core is a /main process/ that enters 'runMainProcess' +-- and creates all of the internal state stored in 'STM.TVar's +-- to manage processes with message queues. +-- +-- The 'Eff' handler for 'Process' and 'MessagePassing' use +-- are implemented and available through 'spawn'. +-- +-- 'spawn' uses 'forkFinally' and 'STM.TQueue's and tries to catch +-- most exceptions. {-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE ConstraintKinds #-} {-# LANGUAGE FlexibleContexts #-} @@ -16,19 +33,12 @@ {-# LANGUAGE GADTs #-} module Control.Eff.Concurrent.Dispatcher ( runMainProcess - , DispatcherVar - , withDispatcher - , Dispatcher + , defaultMain + , spawn + , DispatcherError(..) , DispatcherIO , HasDispatcherIO , ProcIO - , ConsProcIO - , HasProcIO - , ProcessInfo - , processId - , getProcessInfo - , getLogChannel - , spawn ) where @@ -54,8 +64,8 @@ import Data.Map ( Map ) import qualified Data.Map as Map --- * MessagePassing Scheduling - +-- | Information about a process, needed to implement 'MessagePassing' and +-- 'Process' handlers. The message queue is backed by a 'STM.TQueue'. data ProcessInfo = ProcessInfo { _processId :: ProcessId , _messageQ :: STM.TQueue Dynamic @@ -69,6 +79,9 @@ "ProcessInfo: " ++ show (p ^. processId) ++ " trapExit: " ++ show (not (p^.exitOnShutdown)) +-- | Contains all 'ProcessInfo' elements, as well as the state needed to +-- implement inter process communication. It contains also a 'LogChannel' to +-- which the logs of all processes are forwarded to. data Dispatcher = Dispatcher { _nextPid :: ProcessId , _processTable :: Map ProcessId ProcessInfo @@ -79,29 +92,43 @@ makeLenses ''Dispatcher +-- | A newtype wrapper around an 'STM.TVar' holding a 'Dispatcher' state. +-- This is needed by 'spawn' and provided by 'runDispatcher'. newtype DispatcherVar = DispatcherVar { fromDispatcherVar :: STM.TVar Dispatcher } deriving Typeable +-- | A sum-type with errors that can occur when dispatching messages. data DispatcherError = - ProcessShutdown String ProcessId + UnhandledMessageReceived Dynamic ProcessId + -- ^ A process message queue contained a bad message and the 'Dynamic' value + -- could not be converted to the expected value using 'fromDynamic'. | ProcessNotFound ProcessId + -- ^ No 'ProcessInfo' was found for a 'ProcessId' during internal + -- processing. NOTE: This is **ONLY** caused by internal errors, probably by + -- an incorrect 'MessagePassing' handler in this module. **Sending a message + -- to a process ALWAYS succeeds!** Even if the process does not exist. | ProcessException String ProcessId + -- ^ A process called 'raiseError'. | DispatcherShuttingDown + -- ^ An action was not performed while the dispatcher was exiting. | LowLevelIOException Exc.SomeException + -- ^ 'Control.Exception.SomeException' was caught while dispatching + -- messages. deriving (Typeable, Show) instance Exc.Exception DispatcherError -type family Members (es :: [* -> *]) (r :: [* -> *]) :: Constraint where - Members '[] r = () - Members (e ': es) r = (Member e r, Members es r) - +-- | An alias for the constraints for the effects essential to this dispatcher +-- implementation, i.e. these effects allow 'spawn'ing new 'Process'es. +-- @see DispatcherIO type HasDispatcherIO r = ( HasCallStack , SetMember Lift (Lift IO) r , Member (Exc DispatcherError) r , Member (Logs String) r , Member (Reader DispatcherVar) r) +-- | The concrete list of 'Eff'ects for this scheduler implementation. +-- @see HasDispatcherIO type DispatcherIO = '[ Exc DispatcherError , Reader DispatcherVar @@ -109,21 +136,21 @@ , Lift IO ] -type HasProcIO r = ( Member MessagePassing r - , Member Process r - , HasDispatcherIO r - ) - -type ConsProcIO r = MessagePassing ': Process ': r - +-- | The concrete list of 'Eff'ects that provide 'MessagePassing' and +-- 'Process'es ontop of 'DispatcherIO' type ProcIO = ConsProcIO DispatcherIO +-- | /Cons/ 'ProcIO' onto a list of effects. +type ConsProcIO r = MessagePassing ': Process ': r instance MonadLog String (Eff ProcIO) where logMessageFree = logMessageFreeEff +-- | This is the main entry point to running a message passing concurrency +-- application. This function takes a 'ProcIO' effect and a 'LogChannel' for +-- concurrent logging. runMainProcess :: Eff ProcIO a -> LogChannel String -> IO a -runMainProcess e = withDispatcher +runMainProcess e logC = withDispatcher (dispatchMessages (\cleanup -> do mt <- lift myThreadId @@ -146,7 +173,63 @@ return rres ) ) + where + withDispatcher :: Eff DispatcherIO a -> IO a + withDispatcher mainProcessAction = do + myTId <- myThreadId + Exc.bracket + (newTVarIO (Dispatcher myPid Map.empty Map.empty False logC)) + (tearDownDispatcher myTId) + (\sch -> runLift + (forwardLogsToChannel + logC + (runReader (runErrorRethrowIO mainProcessAction) (DispatcherVar sch)) + ) + ) + where + myPid = 1 + tearDownDispatcher myTId v = do + logChannelPutIO logC (show myTId ++" begin dispatcher tear down") + sch <- + (atomically + (do + sch <- readTVar v + let sch' = sch & schedulerShuttingDown .~ True + writeTVar v sch' + return sch + ) + ) + logChannelPutIO logC (show myTId ++ " killing threads: " ++ + show (sch ^.. threadIdTable. traversed)) + imapM_ (killProcThread myTId) (sch ^. threadIdTable) + atomically + (do + dispatcher <- readTVar v + let allThreadsDead = dispatcher^.threadIdTable.to Map.null + && dispatcher^.processTable.to Map.null + STM.check allThreadsDead) + logChannelPutIO logC "all threads dead" + + killProcThread myTId pid tid = when + (myTId /= tid) + ( logChannelPutIO logC ("killing thread " ++ show pid) + >> killThread tid + ) + + +-- | Start the message passing concurrency system then execute a 'ProcIO' effect. +-- All logging is sent to standard output. +defaultMain :: Eff ProcIO a -> IO a +defaultMain c = + runLoggingT + (logChannelBracket + (Just "~~~~~~ main process started") + (Just "====== main process exited") + (runMainProcess c)) + (print :: String -> IO ()) + + runChildProcess :: DispatcherVar -> (CleanUpAction -> Eff ProcIO a) @@ -164,50 +247,7 @@ s <- getDispatcherTVar lift ((view logChannel) <$> atomically (readTVar s)) -withDispatcher :: Eff DispatcherIO a -> LogChannel String -> IO a -withDispatcher mainProcessAction logC = do - myTId <- myThreadId - Exc.bracket - (newTVarIO (Dispatcher 1 Map.empty Map.empty False logC)) - (tearDownDispatcher myTId) - (\sch -> runLift - (forwardLogsToChannel - logC - (runReader (runErrorRethrowIO mainProcessAction) (DispatcherVar sch)) - ) - ) - where - tearDownDispatcher myTId v = do - sch <- - (atomically - (do - sch <- readTVar v - let sch' = - sch - & schedulerShuttingDown - .~ True - & processTable - .~ Map.empty - & threadIdTable - .~ Map.empty - writeTVar v sch' - return sch - ) - ) - imapM_ (killProcThread myTId) (sch ^. threadIdTable) - killProcThread myTId pid tid = when - (myTId /= tid) - ( logChannelPutIO logC ("killing thread of process " ++ show pid) - >> killThread tid - ) -getProcessInfo :: HasDispatcherIO r => ProcessId -> Eff r ProcessInfo -getProcessInfo pid = do - res <- getProcessInfoSafe pid - case res of - Nothing -> throwError (ProcessShutdown "process table not found" pid) - Just i -> return i - overProcessInfo :: HasDispatcherIO r => ProcessId @@ -224,12 +264,6 @@ return (Right x) ) -getProcessInfoSafe - :: HasDispatcherIO r => ProcessId -> Eff r (Maybe ProcessInfo) -getProcessInfoSafe pid = do - p <- getDispatcher - return (p ^. processTable . at pid) - -- ** MessagePassing execution spawn :: HasDispatcherIO r => Eff ProcIO () -> Eff r ProcessId @@ -283,12 +317,23 @@ => (CleanUpAction -> Eff (ConsProcIO r) a) -> Eff r a dispatchMessages processAction = withMessageQueue - (\cleanUpAction pinfo -> handle_relay - return - (goProc (pinfo ^. processId)) - - (handle_relay return (go (pinfo ^. processId)) (processAction cleanUpAction) - ) + (\cleanUpAction pinfo -> + try + (handle_relay + return + (goProc (pinfo ^. processId)) + (handle_relay return (go (pinfo ^. processId)) + (processAction cleanUpAction) + )) + >>= + either + (\(e :: DispatcherError) -> + do logMsg (show (pinfo^.processId) + ++ " cleanup on exception: " + ++ show e) + lift (runCleanUpAction cleanUpAction) + throwError e) + return ) where go @@ -319,13 +364,10 @@ go pid (ReceiveMessage onMsg) k = do catchError (do - mDynMsg <- overProcessInfo - pid + mDynMsg <- overProcessInfo pid (do mq <- use messageQ - Mtl.lift (readTQueue mq) - ) - + Mtl.lift (readTQueue mq)) case fromDynamic mDynMsg of Just req -> let result = onMsg req in k (Message result) nix@Nothing -> @@ -336,7 +378,7 @@ in do isExitOnShutdown <- overProcessInfo pid (use exitOnShutdown) if isExitOnShutdown - then throwError (ProcessShutdown msg pid) + then throwError (UnhandledMessageReceived mDynMsg pid) else k (ProcessControlMessage msg) ) (\(se :: DispatcherError) -> do @@ -358,7 +400,9 @@ overProcessInfo pid (exitOnShutdown .= (not s)) >>= k goProc pid GetTrapExit k = overProcessInfo pid (use exitOnShutdown) >>= k . not - goProc pid (RaiseError msg) _k = throwError (ProcessException msg pid) + goProc pid (RaiseError msg) _k = do + logMsg (show pid ++ " error raised: " ++ msg) + throwError (ProcessException msg pid) withMessageQueue :: HasDispatcherIO r => (CleanUpAction -> ProcessInfo -> Eff r a) -> Eff r a diff --git a/src/Control/Eff/Concurrent/Examples.hs b/src/Control/Eff/Concurrent/Examples.hs --- a/src/Control/Eff/Concurrent/Examples.hs +++ b/src/Control/Eff/Concurrent/Examples.hs @@ -44,12 +44,8 @@ deriving instance Show (Api TestApi x) -runExample :: IO (Either Exc.SomeException ()) -runExample = - Exc.try - (runLoggingT - (logChannelBracket (Just "hello") (Just "KTHXBY") (runMainProcess example)) - (print :: String -> IO ())) +main :: IO () +main = defaultMain example example @@ -69,22 +65,25 @@ logMessage ("Started server " ++ show server) let go = do x <- lift getLine - res <- ignoreProcessError (call server (SayHello x)) - logMessage ("Result: " ++ show res) case x of ('k':_) -> do call server Terminate - logMessage ("terminated: " ++ show server) go + ('c':_) -> do + cast_ server (Shout x) + go ('t':'0':_) -> do call server (SetTrapExit False) go ('t':'1':_) -> do call server (SetTrapExit True) go - ('q':_) -> logMessage "Done." + ('q':_) -> + logMessage "Done." _ -> - go + do res <- ignoreProcessError (call server (SayHello x)) + logMessage ("Result: " ++ show res) + go go testServerLoop diff --git a/src/Control/Eff/Concurrent/Examples2.hs b/src/Control/Eff/Concurrent/Examples2.hs --- a/src/Control/Eff/Concurrent/Examples2.hs +++ b/src/Control/Eff/Concurrent/Examples2.hs @@ -112,14 +112,3 @@ cnt server1 cast_ server2 Inc cnt server2 - --- * System stuff - -runInDispatcherWithIO :: Eff ProcIO a -> IO () -runInDispatcherWithIO c = - runLoggingT - (logChannelBracket - (Just "hello") - (Just "KTHXBY") - (runMainProcess c >=> const getLine >=> const (return ()))) - (print :: String -> IO ()) diff --git a/src/Control/Eff/Concurrent/GenServer.hs b/src/Control/Eff/Concurrent/GenServer.hs --- a/src/Control/Eff/Concurrent/GenServer.hs +++ b/src/Control/Eff/Concurrent/GenServer.hs @@ -38,8 +38,7 @@ import Control.Eff import Control.Lens import Control.Monad -import Data.Dynamic -import Data.Typeable ( typeRep ) +import Data.Typeable (Typeable, typeRep) import Data.Proxy import Control.Eff.Concurrent.MessagePassing diff --git a/src/Control/Eff/Concurrent/MessagePassing.hs b/src/Control/Eff/Concurrent/MessagePassing.hs --- a/src/Control/Eff/Concurrent/MessagePassing.hs +++ b/src/Control/Eff/Concurrent/MessagePassing.hs @@ -1,3 +1,10 @@ +-- | The message passing effect. +-- +-- This module describes an abstract message passing effect, and a process +-- effect, mimicking Erlang's process and message semantics. +-- +-- An implementation of a handler for these effects can be found in +-- 'Control.Eff.Concurrent.Dispatcher'. {-# LANGUAGE DeriveTraversable #-} {-# LANGUAGE DeriveFoldable #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-} @@ -42,6 +49,9 @@ -- * Process Effects +-- | The process effect is the basis for message passing concurrency. This binds +-- the semantics of a process with a process-id, and some process flags, and the +-- ability to leave a process early with an error. data Process b where SelfPid :: Process ProcessId TrapExit :: Bool -> Process () @@ -49,18 +59,25 @@ RaiseError :: String -> Process b -- LinkProcesses :: ProcessId -> ProcessId -> Process () +-- | Returns the 'ProcessId' of the current process. self :: Member Process r => Eff r ProcessId self = send SelfPid +-- | Set the flag that controls a process reaction to +-- exit messages from linked/monitored processes. trapExit :: Member Process r => Bool -> Eff r () trapExit = send . TrapExit +-- | Return the 'trapExit' flag. getTrapExit :: Member Process r => Eff r Bool getTrapExit = send GetTrapExit +-- | Thrown an error, can be caught by 'catchProcessError'. raiseError :: Member Process r => String -> Eff r b raiseError = send . RaiseError +-- | Catch and handle an error raised by 'raiseError'. Works independent of the +-- handler implementation. catchProcessError :: forall r w . Member Process r => (String -> Eff r w) -> Eff r w -> Eff r w catchProcessError onErr = interpose return go @@ -69,10 +86,15 @@ go (RaiseError emsg) _k = onErr emsg go s k = send s >>= k +-- | Like 'catchProcessError' it catches 'raiseError', but instead of invoking a +-- user provided handler, the result is wrapped into an 'Either'. ignoreProcessError :: (HasCallStack, Member Process r) => Eff r a -> Eff r (Either String a) ignoreProcessError = catchProcessError (return . Left) . fmap Right +-- | Each process is identified by a single process id, that stays constant +-- throughout the life cycle of a process. Also, message sending relies on these +-- values to address messages to processes. newtype ProcessId = ProcessId { _fromProcessId :: Int } deriving (Eq,Ord,Typeable,Bounded,Num, Enum, Integral, Real) @@ -90,22 +112,39 @@ makeLenses ''ProcessId --- * MessagePassing Effect +-- * 'MessagePassing' Effect +-- | An effect for sending and receiving messages. data MessagePassing b where + -- | Send a message to a process addressed by the 'ProcessId'. Sending a + -- message should **always succeed** and return **immediately**, even if the + -- destination process does not exist, or does not accept messages of the + -- given type. SendMessage :: Typeable m => ProcessId -> m -> MessagePassing Bool + -- | Receive a message. This should block until an a message was received. The + -- pure function may convert the incoming message into something, and the + -- result is returned as 'Message' value. Another reason why this function + -- returns, is if a process control message was sent to the process. This can + -- only occur from inside the runtime system, aka the effect handler + -- implementation. (Currently there is one in 'Control.Eff.Concurrent.Dispatcher'.) ReceiveMessage :: forall e m . (Typeable m, Typeable (Message m)) => (m -> e) -> MessagePassing (Message e) +-- | When a process invokes 'receiveMessage' a value of this type is returned. +-- There are more reasons that 'receiveMessage' might return, one is that a +-- message was sent to the process, another might be that in internal, handler +-- specific, event occurred for which the process should /wake-up/. data Message m where ProcessControlMessage :: String -> Message m Message :: m -> Message m deriving (Typeable, Functor, Show, Eq, Ord, Foldable, Traversable) +-- | Send a message to a process addressed by the 'ProcessId'. +-- @see 'SendMessage'. sendMessage :: forall o r . (HasCallStack, Member MessagePassing r, Typeable o) @@ -114,6 +153,10 @@ -> Eff r Bool sendMessage pid message = send (SendMessage pid message) +-- | Block until a message was received. Expect a message of the type annotated +-- by the 'Proxy'. +-- Depending on 'trapExit', this will 'raiseError'. +-- @see 'ReceiveMessage'. receiveMessage :: forall o r . (HasCallStack, Member MessagePassing r, Member Process r, Typeable o) diff --git a/src/Control/Eff/Concurrent/Observer.hs b/src/Control/Eff/Concurrent/Observer.hs --- a/src/Control/Eff/Concurrent/Observer.hs +++ b/src/Control/Eff/Concurrent/Observer.hs @@ -194,7 +194,7 @@ -- | Start a new process for an 'Observer' that dispatches -- all observations to an effectful callback. spawnCallbackObserver - :: forall o r . (HasProcIO r, Typeable o, Show (Observation o), Observable o) + :: forall o r . (HasDispatcherIO r, Typeable o, Show (Observation o), Observable o) => (Server o -> Observation o -> Eff ProcIO Bool) -> Eff r (Server (CallbackObserver o)) spawnCallbackObserver onObserve = diff --git a/src/Control/Eff/Interactive.hs b/src/Control/Eff/Interactive.hs --- a/src/Control/Eff/Interactive.hs +++ b/src/Control/Eff/Interactive.hs @@ -9,7 +9,7 @@ , promptStep , step , Interactive(..) - , interactiveInterpreter + , interactiveProgram , runInteractionIOE , runInteractionIO ) @@ -50,15 +50,7 @@ singleton (PrintLine m) singleton (ReadLine p) -class Interactive f where - singleSteps :: (Member (Program Interaction) r, HasCallStack) => f a -> Eff r a -interactiveInterpreter - :: (HasCallStack, Member (Program Interaction) r, Interactive f) - => Eff (Program f ': r) a - -> Eff r a -interactiveInterpreter = runProgram singleSteps - runInteractionIOE :: (SetMember Lift (Lift IO) r, HasCallStack) => Eff (Program Interaction ': r) a @@ -72,3 +64,14 @@ runInteractionIO :: Eff '[Program Interaction, Lift IO] a -> IO a runInteractionIO = runLift . runInteractionIOE + +-- ** Interactively work with 'Control.Eff.Operational.Program's. + +class Interactive f where + singleSteps :: (Member (Program Interaction) r, HasCallStack) => f a -> Eff r a + +interactiveProgram + :: (HasCallStack, Member (Program Interaction) r, Interactive f) + => Eff (Program f ': r) a + -> Eff r a +interactiveProgram = runProgram singleSteps