extensible-effects-concurrent 0.1.2.0 → 0.1.2.1
raw patch · 9 files changed
+206/−124 lines, 9 filesdep ~extensible-effects
Dependency ranges changed: extensible-effects
Files
- ChangeLog.md +5/−0
- extensible-effects-concurrent.cabal +2/−2
- src/Control/Eff/Concurrent/Dispatcher.hs +131/−87
- src/Control/Eff/Concurrent/Examples.hs +10/−11
- src/Control/Eff/Concurrent/Examples2.hs +0/−11
- src/Control/Eff/Concurrent/GenServer.hs +1/−2
- src/Control/Eff/Concurrent/MessagePassing.hs +44/−1
- src/Control/Eff/Concurrent/Observer.hs +1/−1
- src/Control/Eff/Interactive.hs +12/−9
ChangeLog.md view
@@ -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
extensible-effects-concurrent.cabal view
@@ -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 <https://github.com/sheyll/extensible-effects-concurrent#readme> 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:
src/Control/Eff/Concurrent/Dispatcher.hs view
@@ -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
src/Control/Eff/Concurrent/Examples.hs view
@@ -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
src/Control/Eff/Concurrent/Examples2.hs view
@@ -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 ())
src/Control/Eff/Concurrent/GenServer.hs view
@@ -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
src/Control/Eff/Concurrent/MessagePassing.hs view
@@ -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)
src/Control/Eff/Concurrent/Observer.hs view
@@ -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 =
src/Control/Eff/Interactive.hs view
@@ -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