extensible-effects-concurrent 0.1.1.0 → 0.1.2.0
raw patch · 11 files changed
+1103/−201 lines, 11 filesdep +logging-effectdep +monad-control
Dependencies added: logging-effect, monad-control
Files
- ChangeLog.md +12/−5
- README.md +1/−1
- extensible-effects-concurrent.cabal +10/−3
- src/Control/Eff/Concurrent/Dispatcher.hs +342/−102
- src/Control/Eff/Concurrent/Examples.hs +82/−15
- src/Control/Eff/Concurrent/Examples2.hs +125/−0
- src/Control/Eff/Concurrent/GenServer.hs +45/−51
- src/Control/Eff/Concurrent/MessagePassing.hs +55/−24
- src/Control/Eff/Concurrent/Observer.hs +215/−0
- src/Control/Eff/ExceptionExtra.hs +43/−0
- src/Control/Eff/Log.hs +173/−0
ChangeLog.md view
@@ -1,15 +1,22 @@ # Changelog for extensible-effects-concurrent +## 0.1.2.0++ * Add Observer module+ * Implement Exception handling+ * Improve Dispatcher shutdown+ * Add logging support via the logging-effect library+ ## 0.1.1.0 -* Substantial API reoganization-* Rename/Move modules+ * Substantial API reoganization+ * Rename/Move modules ## 0.1.0.1 -* Stack/Cabal/Github Cosmetics-* Travis build job+ * Stack/Cabal/Github Cosmetics+ * Travis build job ## 0.1.0.0 -* Initial Version+ * Initial Version
README.md view
@@ -2,7 +2,7 @@ Message passing concurrency with 'forkIO' and 'extensible-effects' inspired by Erlang. -[](https://travis-ci.org/sheyll/extensible-effects-concurrent) +[](https://travis-ci.org/sheyll/extensible-effects-concurrent) [](http://hackage.haskell.org/package/extensible-effects-concurrent)
extensible-effects-concurrent.cabal view
@@ -1,5 +1,5 @@ name: extensible-effects-concurrent-version: 0.1.1.0+version: 0.1.2.0 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@@ -32,21 +32,27 @@ containers, QuickCheck <2.11, lens,+ logging-effect, transformers, parallel, process,+ monad-control, random, extensible-effects, stm, tagged exposed-modules:+ Control.Eff.ExceptionExtra, Control.Eff.Interactive,+ Control.Eff.Log,+ Control.Eff.Concurrent.Dispatcher, Control.Eff.Concurrent.GenServer, Control.Eff.Concurrent.MessagePassing,- Control.Eff.Concurrent.Dispatcher+ Control.Eff.Concurrent.Observer other-modules: Paths_extensible_effects_concurrent,- Control.Eff.Concurrent.Examples+ Control.Eff.Concurrent.Examples,+ Control.Eff.Concurrent.Examples2 other-extensions: ConstraintKinds, DeriveFoldable,@@ -54,6 +60,7 @@ DeriveFunctor, DeriveGeneric, DeriveTraversable,+ GADTs, FlexibleContexts, GeneralizedNewtypeDeriving, RankNTypes,
src/Control/Eff/Concurrent/Dispatcher.hs view
@@ -1,6 +1,8 @@ {-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE ConstraintKinds #-} {-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE TypeFamilies #-} {-# LANGUAGE KindSignatures #-} {-# LANGUAGE DataKinds #-}@@ -10,32 +12,42 @@ {-# LANGUAGE StandaloneDeriving #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TypeOperators #-}+{-# LANGUAGE TypeApplications #-} {-# LANGUAGE GADTs #-} module Control.Eff.Concurrent.Dispatcher- ( ProcIO- , HasProcesses- , Scheduler- -- , nextPid, processTable+ ( runMainProcess+ , DispatcherVar+ , withDispatcher+ , Dispatcher+ , DispatcherIO+ , HasDispatcherIO+ , ProcIO+ , ConsProcIO+ , HasProcIO , ProcessInfo , processId- -- , messageQ- , runProcesses , getProcessInfo+ , getLogChannel , spawn- , dispatchMessages- -- , withMessageQueue ) where import GHC.Stack+import Data.Maybe import Data.Kind+import qualified Control.Exception as Exc import Control.Concurrent as Concurrent import Control.Concurrent.STM as STM import Control.Eff-import Control.Eff.Lift import Control.Eff.Concurrent.MessagePassing+import Control.Eff.ExceptionExtra+import Control.Eff.Lift+import Control.Eff.Log import Control.Eff.Reader.Strict as Reader import Control.Lens+import Control.Monad ( when+ , void+ ) import qualified Control.Monad.State as Mtl import Data.Dynamic import Data.Typeable ( typeRep )@@ -47,86 +59,249 @@ data ProcessInfo = ProcessInfo { _processId :: ProcessId , _messageQ :: STM.TQueue Dynamic+ , _exitOnShutdown :: Bool } makeLenses ''ProcessInfo instance Show ProcessInfo where show p =- "ProcessInfo: " ++ show (p ^. processId)+ "ProcessInfo: " ++ show (p ^. processId) ++ " trapExit: "+ ++ show (not (p^.exitOnShutdown)) -data Scheduler =- Scheduler { _nextPid :: ProcessId- , _processTable :: Map ProcessId ProcessInfo- }- deriving Show-makeLenses ''Scheduler+data Dispatcher =+ Dispatcher { _nextPid :: ProcessId+ , _processTable :: Map ProcessId ProcessInfo+ , _threadIdTable :: Map ProcessId ThreadId+ , _schedulerShuttingDown :: Bool+ , _logChannel :: LogChannel String+ } +makeLenses ''Dispatcher++newtype DispatcherVar = DispatcherVar { fromDispatcherVar :: STM.TVar Dispatcher }+ deriving Typeable++data DispatcherError =+ ProcessShutdown String ProcessId+ | ProcessNotFound ProcessId+ | ProcessException String ProcessId+ | DispatcherShuttingDown+ | LowLevelIOException Exc.SomeException+ 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) -type HasProcesses r = ( HasCallStack- , SetMember Lift (Lift IO) r- , Member (Reader (STM.TVar Scheduler)) r)+type HasDispatcherIO r = ( HasCallStack+ , SetMember Lift (Lift IO) r+ , Member (Exc DispatcherError) r+ , Member (Logs String) r+ , Member (Reader DispatcherVar) r) -type ProcIO = '[MessagePassing, Process, Reader (STM.TVar Scheduler), Lift IO]+type DispatcherIO =+ '[ Exc DispatcherError+ , Reader DispatcherVar+ , Logs String+ , Lift IO+ ] -runProcesses- :: (SetMember Lift (Lift IO) r, HasCallStack)- => Eff (Reader (STM.TVar Scheduler) ': r) ()- -> Eff r ()-runProcesses e = do- v <- lift (newTVarIO (Scheduler 1 Map.empty))- runReader e v+type HasProcIO r = ( Member MessagePassing r+ , Member Process r+ , HasDispatcherIO r+ ) -getProcessInfo :: HasProcesses r => ProcessId -> Eff r (Maybe ProcessInfo)+type ConsProcIO r = MessagePassing ': Process ': r++type ProcIO = ConsProcIO DispatcherIO+++instance MonadLog String (Eff ProcIO) where+ logMessageFree = logMessageFreeEff++runMainProcess :: Eff ProcIO a -> LogChannel String -> IO a+runMainProcess e = withDispatcher+ (dispatchMessages+ (\cleanup -> do+ mt <- lift myThreadId+ mp <- self+ logMessage (show mp ++ " main process started in thread " ++ show mt)+ res <- try e+ case res of+ Left ex ->+ do+ logMessage+ ( show mp+ ++ " main process exception: "+ ++ ((show :: DispatcherError -> String) ex)+ )+ lift (runCleanUpAction cleanup)+ >> throwError ex+ Right rres -> do+ logMessage (show mp ++ " main process exited")+ lift (runCleanUpAction cleanup)+ return rres+ )+ )++runChildProcess+ :: DispatcherVar+ -> (CleanUpAction -> Eff ProcIO a)+ -> IO (Either DispatcherError a)+runChildProcess s procAction = do+ l <- (view logChannel) <$> atomically (readTVar (fromDispatcherVar s))+ runLift+ (forwardLogsToChannel+ l+ (runReader (runError (dispatchMessages procAction)) s)+ )++getLogChannel :: HasDispatcherIO r => Eff r (LogChannel String)+getLogChannel = do+ 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- p <- getScheduler+ res <- getProcessInfoSafe pid+ case res of+ Nothing -> throwError (ProcessShutdown "process table not found" pid)+ Just i -> return i++overProcessInfo+ :: HasDispatcherIO r+ => ProcessId+ -> Mtl.StateT ProcessInfo STM.STM a+ -> Eff r a+overProcessInfo pid stAction = liftEither =<< overDispatcher+ (do+ res <- use (processTable . at pid)+ case res of+ Nothing -> return (Left (ProcessNotFound pid))+ Just pinfo -> do+ (x, pinfoOut) <- Mtl.lift (Mtl.runStateT stAction pinfo)+ processTable . at pid . _Just .= pinfoOut+ return (Right x)+ )++getProcessInfoSafe+ :: HasDispatcherIO r => ProcessId -> Eff r (Maybe ProcessInfo)+getProcessInfoSafe pid = do+ p <- getDispatcher return (p ^. processTable . at pid) -- ** MessagePassing execution -spawn :: HasProcesses r => Eff ProcIO () -> Eff r ProcessId+spawn :: HasDispatcherIO r => Eff ProcIO () -> Eff r ProcessId spawn mfa = do- processes <- ask- pidVar <- lift newEmptyTMVarIO- _threadId <- lift- (Concurrent.forkIO- (runLift- (runReader- (dispatchMessages- (do- pid <- self- lift (atomically (STM.putTMVar pidVar pid))+ schedulerVar <- ask+ pidVar <- lift newEmptyTMVarIO+ cleanupVar <- lift newEmptyTMVarIO+ lc <- getLogChannel+ void+ (lift+ (Concurrent.forkFinally+ (runChildProcess+ schedulerVar+ (\cleanUpAction -> do+ lift (atomically (STM.putTMVar cleanupVar cleanUpAction))+ pid <- self+ lift (atomically (STM.putTMVar pidVar (Just pid)))+ catchError mfa- )+ ( logMessage+ . ("process exception: " ++)+ . (show :: DispatcherError -> String)+ ) )- processes )+ (\eres -> do+ mcleanup <- atomically (STM.tryTakeTMVar cleanupVar)+ void (atomically (tryPutTMVar pidVar Nothing))+ case mcleanup of+ Nothing -> return ()+ Just ca -> do+ runCleanUpAction ca+ mt <- myThreadId+ case eres of+ Left se -> logChannelPutIO+ lc+ ("thread " ++ show mt ++ " killed by exception: " ++ show se)+ Right _ ->+ logChannelPutIO lc ("thread " ++ show mt ++ " exited")+ ) ) )- lift (atomically (STM.takeTMVar pidVar))+ mPid <- lift (atomically (STM.takeTMVar pidVar))+ maybe (throwError DispatcherShuttingDown) return mPid +newtype CleanUpAction = CleanUpAction { runCleanUpAction :: IO () }+ dispatchMessages :: forall r a- . (HasProcesses r, HasCallStack)- => Eff (MessagePassing ': Process ': r) a+ . (HasDispatcherIO r, HasCallStack)+ => (CleanUpAction -> Eff (ConsProcIO r) a) -> Eff r a dispatchMessages processAction = withMessageQueue- (\pinfo ->- handle_relay return (goProc pinfo) (handle_relay return (go pinfo) processAction))+ (\cleanUpAction pinfo -> handle_relay+ return+ (goProc (pinfo ^. processId))++ (handle_relay return (go (pinfo ^. processId)) (processAction cleanUpAction)+ )+ ) where go :: forall v . HasCallStack- => ProcessInfo+ => ProcessId -> MessagePassing v -> (v -> Eff (Process ': r) a) -> Eff (Process ': r) a- go _ (SendMessage toPid reqIn) k = do- psVar <- getSchedulerTVar- lift+ go _pid (SendMessage toPid reqIn) k = do+ psVar <- getDispatcherTVar+ liftRethrow+ LowLevelIOException (atomically (do p <- readTVar psVar@@ -141,66 +316,131 @@ ) ) >>= k- go (ProcessInfo selfPidInt channel) (ReceiveMessage onMsg) k = do- mDynMsg <- lift (atomically (readTQueue channel))- case fromDynamic mDynMsg of- Just req -> let result = onMsg req in k (Just result)- nix -> do- lift- (putStrLn- ( show selfPidInt- ++ " got unexpected msg: "- ++ show mDynMsg- ++ " expected: "- ++ show (typeRep nix)- )+ go pid (ReceiveMessage onMsg) k = do+ catchError+ (do+ mDynMsg <- overProcessInfo+ pid+ (do+ mq <- use messageQ+ Mtl.lift (readTQueue mq) )- k Nothing++ case fromDynamic mDynMsg of+ Just req -> let result = onMsg req in k (Message result)+ nix@Nothing ->+ let+ msg =+ "unexpected message: " ++ show mDynMsg ++ " expected: " ++ show+ (typeRep nix)+ in do+ isExitOnShutdown <- overProcessInfo pid (use exitOnShutdown)+ if isExitOnShutdown+ then throwError (ProcessShutdown msg pid)+ else k (ProcessControlMessage msg)+ )+ (\(se :: DispatcherError) -> do+ isExitOnShutdown <- overProcessInfo pid (use exitOnShutdown)+ if isExitOnShutdown+ then throwError se+ else k (ProcessControlMessage (show se))+ )+ goProc- :: forall v+ :: forall v x . HasCallStack- => ProcessInfo+ => ProcessId -> Process v- -> (v -> Eff r a)- -> Eff r a- goProc (ProcessInfo selfPidInt _) SelfPid k = k selfPidInt+ -> (v -> Eff r x)+ -> Eff r x+ goProc pid SelfPid k = k pid+ goProc pid (TrapExit s) k =+ 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) -withMessageQueue :: HasProcesses r => (ProcessInfo -> Eff r a) -> Eff r a+withMessageQueue+ :: HasDispatcherIO r => (CleanUpAction -> ProcessInfo -> Eff r a) -> Eff r a withMessageQueue m = do- pinfo <- createQueue- res <- m pinfo- destroyQueue pinfo- return res+ mpinfo <- createQueue+ lc <- getLogChannel+ case mpinfo of+ Just pinfo -> do+ cleanUpAction <-+ getDispatcherTVar >>= return . CleanUpAction . destroyQueue+ lc+ (pinfo ^. processId)+ m cleanUpAction pinfo+ Nothing -> throwError DispatcherShuttingDown where- createQueue = overScheduler- (do- pid <- nextPid <<+= 1- channel <- Mtl.lift newTQueue- let pinfo = ProcessInfo pid channel- processTable . at pid .= Just pinfo- return pinfo- )- destroyQueue pinfo =- overScheduler (processTable . at (pinfo ^. processId) .= Nothing)---overScheduler :: HasProcesses r => Mtl.StateT Scheduler STM.STM a -> Eff r a-overScheduler stAction = do- psVar <- ask- lift- (STM.atomically+ createQueue = do+ myTId <- lift myThreadId+ overDispatcher (do- ps <- STM.readTVar psVar- (result, psModified) <- Mtl.runStateT stAction ps- STM.writeTVar psVar psModified- return result+ abortNow <- use schedulerShuttingDown+ if abortNow+ then return Nothing+ else do+ pid <- nextPid <<+= 1+ channel <- Mtl.lift newTQueue+ let pinfo = ProcessInfo pid channel True+ threadIdTable . at pid .= Just myTId+ processTable . at pid .= Just pinfo+ return (Just pinfo) )- )+ destroyQueue lc pid psVar = do+ didWork <- Exc.try+ (overDispatcherIO+ psVar+ (do+ abortNow <- use schedulerShuttingDown+ if abortNow+ then return (Nothing, False)+ else do+ os <- processTable . at pid <<.= Nothing+ ot <- threadIdTable . at pid <<.= Nothing+ return (os, isJust os || isJust ot)+ )+ )+ let getCause =+ Exc.try @Exc.SomeException+ (overDispatcherIO psVar (preuse (processTable . at pid)))+ >>= either+ (return . (show pid ++) . show)+ (return . (maybe (show pid) show)) -getSchedulerTVar :: HasProcesses r => Eff r (TVar Scheduler)-getSchedulerTVar = ask+ case didWork of+ Right (pinfo, True) ->+ logChannelPutIO lc ("destroying queue: " ++ show pinfo)+ Right (pinfo, False) ->+ logChannelPutIO lc ("queue already destroyed: " ++ show pinfo)+ Left (e :: Exc.SomeException) ->+ getCause+ >>= logChannelPutIO lc+ . (("failed to destroy queue: " ++ show e ++ " ") ++) -getScheduler :: HasProcesses r => Eff r Scheduler-getScheduler = do- processesVar <- getSchedulerTVar++overDispatcher+ :: HasDispatcherIO r => Mtl.StateT Dispatcher STM.STM a -> Eff r a+overDispatcher stAction = do+ psVar <- getDispatcherTVar+ liftRethrow LowLevelIOException (overDispatcherIO psVar stAction)++overDispatcherIO+ :: STM.TVar Dispatcher -> Mtl.StateT Dispatcher STM.STM a -> IO a+overDispatcherIO psVar stAction = STM.atomically+ (do+ ps <- STM.readTVar psVar+ (result, psModified) <- Mtl.runStateT stAction ps+ STM.writeTVar psVar psModified+ return result+ )++getDispatcherTVar :: HasDispatcherIO r => Eff r (TVar Dispatcher)+getDispatcherTVar = fromDispatcherVar <$> ask++getDispatcher :: HasDispatcherIO r => Eff r Dispatcher+getDispatcher = do+ processesVar <- getDispatcherTVar lift (atomically (readTVar processesVar))
src/Control/Eff/Concurrent/Examples.hs view
@@ -1,3 +1,4 @@+{-# LANGUAGE IncoherentInstances #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE ConstraintKinds #-} {-# LANGUAGE FlexibleContexts #-}@@ -20,10 +21,11 @@ import Control.Eff.Lift import Control.Monad import Data.Dynamic- import Control.Eff.Concurrent.MessagePassing import Control.Eff.Concurrent.GenServer import Control.Eff.Concurrent.Dispatcher+import Control.Eff.Log+import qualified Control.Exception as Exc data TestApi deriving Typeable@@ -31,45 +33,110 @@ data instance Api TestApi x where SayHello :: String -> Api TestApi ('Synchronous Bool) Shout :: String -> Api TestApi 'Asynchronous+ SetTrapExit :: Bool -> Api TestApi ('Synchronous ())+ Terminate :: Api TestApi ('Synchronous ()) deriving (Typeable) +data MyException = MyException+ deriving Show++instance Exc.Exception MyException+ deriving instance Show (Api TestApi x) -runExample :: IO ()-runExample = runLift $ runProcesses $ dispatchMessages example+runExample :: IO (Either Exc.SomeException ())+runExample =+ Exc.try+ (runLoggingT+ (logChannelBracket (Just "hello") (Just "KTHXBY") (runMainProcess example))+ (print :: String -> IO ())) + example :: ( HasCallStack- , HasProcesses r+ , Member (Logs String) r+ , HasDispatcherIO r , Member MessagePassing r , Member Process r+ , MonadLog String (Eff r) , SetMember Lift (Lift IO) r) => Eff r () example = do me <- self- lift (putStrLn ("I am " ++ show me))+ trapExit True+ logMessage ("I am " ++ show me) server <- asServer @TestApi <$> spawn testServerLoop- lift (putStrLn ("Started server " ++ show server))+ logMessage ("Started server " ++ show server) let go = do- x <- lift (putStr ">>> " >> getLine)- res <- call server (SayHello x)- lift (putStrLn ("Result: " ++ show res))+ x <- lift getLine+ res <- ignoreProcessError (call server (SayHello x))+ logMessage ("Result: " ++ show res) case x of- ('q':_) -> return ()- _ -> go+ ('k':_) -> do+ call server Terminate+ logMessage ("terminated: " ++ show server)+ go+ ('t':'0':_) -> do+ call server (SetTrapExit False)+ go+ ('t':'1':_) -> do+ call server (SetTrapExit True)+ go+ ('q':_) -> logMessage "Done."+ _ ->+ go go testServerLoop- :: forall r. (HasCallStack, Member MessagePassing r, Member Process r, SetMember Lift (Lift IO) r)+ :: forall r. (HasCallStack, Member MessagePassing r, Member Process r+ , MonadLog String (Eff r)+ , SetMember Lift (Lift IO) r) => Eff r ()-testServerLoop = forever $ serve_ $ ApiHandler handleCast handleCall+testServerLoop =+ -- trapExit True+ -- >>+ (forever $ serve_ $ ApiHandler handleCast handleCall handleTerminate) where handleCast :: Api TestApi 'Asynchronous -> Eff r () handleCast (Shout x) = do me <- self- lift (putStrLn (show me ++ " Shouting: " ++ x))+ logMessage (show me ++ " Shouting: " ++ x) handleCall :: Api TestApi ('Synchronous x) -> (x -> Eff r Bool) -> Eff r ()+ handleCall (SayHello "e1") _reply = do+ me <- self+ logMessage (show me ++ " raising an error")+ raiseError "No body loves me... :,("+ handleCall (SayHello "e2") _reply = do+ me <- self+ logMessage (show me ++ " throwing a MyException ")+ lift (Exc.throw MyException)+ handleCall (SayHello "self") reply = do+ me <- self+ logMessage (show me ++ " casting to self")+ cast_ (asServer @TestApi me) (Shout "from me")+ void (reply False)+ handleCall (SayHello "die") reply = do+ me <- self+ logMessage (show me ++ " throwing and catching ")+ catchProcessError+ (\ er -> logMessage ("WOW: " ++ show er ++ " - No. This is wrong!"))+ (raiseError "No body loves me... :,(")+ void (reply True) handleCall (SayHello x) reply = do me <- self- lift (putStrLn (show me ++ " Got Hello: " ++ x))+ logMessage (show me ++ " Got Hello: " ++ x) void (reply (length x > 3))+ handleCall (SetTrapExit x) reply = do+ me <- self+ logMessage (show me ++ " setting trap exit to " ++ show x)+ trapExit x+ void (reply ())+ handleCall Terminate reply = do+ me <- self+ logMessage (show me ++ " exitting")+ void (reply ())+ raiseError "DONE"+ handleTerminate msg = do+ me <- self+ logMessage (show me ++ " is exiting: " ++ msg)+ raiseError msg
+ src/Control/Eff/Concurrent/Examples2.hs view
@@ -0,0 +1,125 @@+{-# LANGUAGE IncoherentInstances #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE KindSignatures #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE DeriveFunctor #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeOperators #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE GADTs #-}+module Control.Eff.Concurrent.Examples2 where++import Data.Dynamic+import Control.Eff+import Control.Eff.Concurrent.Dispatcher+import Control.Eff.Concurrent.GenServer+import Control.Eff.Concurrent.MessagePassing+import Control.Eff.Concurrent.Observer+import Control.Eff.Log+import Control.Eff.State.Lazy+import Control.Monad++data Counter deriving Typeable++data instance Api Counter x where+ Inc :: Api Counter 'Asynchronous+ Cnt :: Api Counter ('Synchronous Integer)+ ObserveCounter :: SomeObserver Counter -> Api Counter 'Asynchronous+ UnobserveCounter :: SomeObserver Counter -> Api Counter 'Asynchronous++deriving instance Show (Api Counter x)++instance Observable Counter where+ data Observation Counter where+ CountChanged :: Integer -> Observation Counter+ deriving (Show, Typeable)+ registerObserverMessage os = ObserveCounter os+ forgetObserverMessage os = UnobserveCounter os++logCounterObservations :: Eff ProcIO (Server (CallbackObserver Counter))+logCounterObservations =+ spawnCallbackObserver+ (\fromSvr msg ->+ do me <- self+ logMsg (show me ++ " observed on: " ++ show fromSvr ++ ": " ++ show msg)+ return True)++type CounterEff = State (Observers Counter) ': State Integer ': ProcIO++data ServerState st a where+ ServerState :: State st a -> ServerState st a++counterServerLoop :: Eff ProcIO ()+counterServerLoop = do+ trapExit True+ evalState (manageObservers+ $ forever+ $ serve_+ $ ApiHandler @Counter handleCast handleCall error) 0+ where+ handleCast :: Api Counter 'Asynchronous -> Eff CounterEff ()+ handleCast (ObserveCounter o) = do+ addObserver o+ handleCast (UnobserveCounter o) = do+ removeObserver o+ handleCast Inc = do+ logMsg "Inc"+ modify (+ (1 :: Integer))+ currentCount <- get+ notifyObservers (CountChanged currentCount)+ handleCall :: Api Counter ('Synchronous x) -> (x -> Eff CounterEff Bool) -> Eff CounterEff ()+ handleCall Cnt reply = do+ c <- get+ logMsg ("Cnt is " ++ show c)+ _ <- reply c+ return ()++-- ** Counter client++counterExample :: Eff ProcIO ()+counterExample = do+ let cnt sv = do r <- call sv Cnt+ logMsg (show sv ++ " " ++ show r)+ server1 <- asServer @Counter <$> spawn counterServerLoop+ server2 <- asServer @Counter <$> spawn counterServerLoop+ cast_ server1 Inc+ cnt server1+ cnt server2+ co1 <- logCounterObservations+ co2 <- logCounterObservations+ registerObserver co1 server1+ registerObserver co2 server2+ cast_ server1 Inc+ cnt server1+ cast_ server2 Inc+ cnt server2+ registerObserver co2 server1+ registerObserver co1 server2+ cast_ server1 Inc+ cnt server1+ cast_ server2 Inc+ cnt server2+ forgetObserver co2 server1+ cast_ server1 Inc+ 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
@@ -47,43 +47,43 @@ -- | This data family defines an API implemented by a server. -- The first parameter is the API /index/ and the second parameter -- (the @* -> *@)-data family Api o :: Synchronicity -> *+data family Api (genServerModule :: Type) (replyType :: Synchronicity) data Synchronicity = Synchronous Type | Asynchronous deriving (Typeable) -newtype Server o = Server { _fromServer :: ProcessId }+newtype Server genServerModule = Server { _fromServer :: ProcessId } deriving (Eq,Ord,Typeable) -instance Read (Server o) where+instance Read (Server genServerModule) where readsPrec _ ('[':'#':rest1) =- case reads (dropWhile (/= '⇒') rest1) of+ case reads (dropWhile (/= '#') rest1) of [(c, ']':rest2)] -> [(Server c, rest2)] _ -> [] readsPrec _ _ = [] -instance Typeable o => Show (Server o) where+instance Typeable genServerModule => Show (Server genServerModule) where show s@(Server c) =- "[#" ++ show (typeRep s) ++ "⇒" ++ show c ++ "]"+ "[#" ++ show (typeRep s) ++ "#" ++ show c ++ "]" makeLenses ''Server -proxyAsServer :: proxy api -> ProcessId -> Server api+proxyAsServer :: proxy genServerModule -> ProcessId -> Server genServerModule proxyAsServer = const Server -asServer :: ProcessId -> Server api+asServer :: forall genServerModule . ProcessId -> Server genServerModule asServer = Server -data Request p where- Call :: forall p x . (Typeable p, Typeable x, Typeable (Api p ('Synchronous x)))- => ProcessId -> Api p ('Synchronous x) -> Request p- Cast :: forall p . (Typeable p, Typeable (Api p 'Asynchronous))- => Api p 'Asynchronous -> Request p+data Request genServerModule where+ Call :: forall genServerModule apiCallReplyType . (Typeable genServerModule, Typeable apiCallReplyType, Typeable (Api genServerModule ('Synchronous apiCallReplyType)))+ => ProcessId -> Api genServerModule ('Synchronous apiCallReplyType) -> Request genServerModule+ Cast :: forall genServerModule . (Typeable genServerModule, Typeable (Api genServerModule 'Asynchronous))+ => Api genServerModule 'Asynchronous -> Request genServerModule deriving Typeable -data Response p x where- Response :: (Typeable p, Typeable x) => Proxy p -> x -> Response p x+data Response genServerModule apiCallReplyType where+ Response :: (Typeable genServerModule, Typeable apiCallReplyType) => Proxy genServerModule -> apiCallReplyType -> Response genServerModule apiCallReplyType deriving Typeable cast@@ -111,28 +111,30 @@ cast_ = ((.) . (.)) void cast call- :: forall result o r+ :: forall result genServerModule r . ( Member MessagePassing r , Member Process r- , Typeable o- , Typeable (Api o ('Synchronous result))+ , Typeable genServerModule+ , Typeable (Api genServerModule ( 'Synchronous result)) , Typeable result , HasCallStack )- => Server o- -> Api o ('Synchronous result)- -> Eff r (Maybe result)+ => Server genServerModule+ -> Api genServerModule ( 'Synchronous result)+ -> Eff r (Message result) call (Server pidInt) req = do fromPid <- self let requestMessage = Call fromPid req wasSent <- sendMessage pidInt requestMessage if wasSent then- let extractResult :: Response o result -> result+ let extractResult :: Response genServerModule result -> result extractResult (Response _pxResult result) = result- in do mResp <- receiveMessage (Proxy @(Response o result))+ in do+ mResp <- receiveMessage (Proxy @(Response genServerModule result)) return (extractResult <$> mResp)- else fail "Could not send request message " >> return Nothing+ else raiseError+ ("Could not send request message " ++ show (typeRep requestMessage)) data ApiHandler p r e where ApiHandler ::@@ -142,6 +144,9 @@ , _handleCall :: forall x . (Typeable p, Typeable (Api p ('Synchronous x)), Typeable x, HasCallStack) => Api p ('Synchronous x) -> (x -> Eff r Bool) -> Eff r e+ , _handleTerminate+ :: (Typeable p, HasCallStack)+ => String -> Eff r () } -> ApiHandler p r e serve_@@ -155,40 +160,37 @@ :: forall r p e . (Typeable p, Member MessagePassing r, Member Process r, HasCallStack) => ApiHandler p r e- -> Eff r (Maybe e)-serve (ApiHandler handleCast handleCall) = do+ -> Eff r (Message e)+serve (ApiHandler handleCast handleCall handleTerminate) = do mReq <- receiveMessage (Proxy @(Request p))- mapM receiveCallReq mReq+ mapM receiveCallReq mReq >>= catchProcessControlMessage where+ catchProcessControlMessage :: Message e -> Eff r (Message e)+ catchProcessControlMessage s@(ProcessControlMessage msg) =+ handleTerminate msg >> return s+ catchProcessControlMessage s = return s+ receiveCallReq :: Request p -> Eff r e receiveCallReq (Cast request ) = handleCast request receiveCallReq (Call fromPid request) = handleCall request (sendReply request) where- sendReply :: Typeable x => Api p ('Synchronous x) -> x -> Eff r Bool+ sendReply :: Typeable x => Api p ( 'Synchronous x) -> x -> Eff r Bool sendReply _ reply = sendMessage fromPid (Response (Proxy :: Proxy p) reply) unhandledCallError- :: ( Show (Api p ('Synchronous x))+ :: ( Show (Api p ( 'Synchronous x)) , Typeable p- , Typeable (Api p ('Synchronous x))+ , Typeable (Api p ( 'Synchronous x)) , Typeable x , HasCallStack , Member Process r )- => Api p ('Synchronous x)+ => Api p ( 'Synchronous x) -> (x -> Eff r Bool) -> Eff r e-unhandledCallError api _ = do- me <- self- fail- ( show me- ++ " Unhandled call: ("- ++ show api- ++ " :: "- ++ show (typeRep api)- ++ ")"- )+unhandledCallError api _ = raiseError+ ("Unhandled call: (" ++ show api ++ " :: " ++ show (typeRep api) ++ ")") unhandledCastError :: ( Show (Api p 'Asynchronous)@@ -199,13 +201,5 @@ ) => Api p 'Asynchronous -> Eff r e-unhandledCastError api = do- me <- self- fail- ( show me- ++ " Unhandled cast: ("- ++ show api- ++ " :: "- ++ show (typeRep api)- ++ ")"- )+unhandledCastError api = raiseError+ ("Unhandled cast: (" ++ show api ++ " :: " ++ show (typeRep api) ++ ")")
src/Control/Eff/Concurrent/MessagePassing.hs view
@@ -1,3 +1,5 @@+{-# LANGUAGE DeriveTraversable #-}+{-# LANGUAGE DeriveFoldable #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE ConstraintKinds #-} {-# LANGUAGE FlexibleContexts #-}@@ -10,15 +12,21 @@ {-# LANGUAGE StandaloneDeriving #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TypeOperators #-}+{-# LANGUAGE TypeApplications #-} {-# LANGUAGE GADTs #-} module Control.Eff.Concurrent.MessagePassing ( ProcessId(..) , fromProcessId , Process(..) , self+ , trapExit+ , getTrapExit+ , raiseError+ , catchProcessError+ , ignoreProcessError , MessagePassing(..)+ , Message(..) , sendMessage- , kill , receiveMessage ) where@@ -27,8 +35,8 @@ import Control.Eff import Control.Lens import Data.Dynamic+import Data.Kind import Data.Proxy-import Data.Void import Text.Printf @@ -36,11 +44,35 @@ data Process b where SelfPid :: Process ProcessId+ TrapExit :: Bool -> Process ()+ GetTrapExit :: Process Bool+ RaiseError :: String -> Process b -- LinkProcesses :: ProcessId -> ProcessId -> Process () self :: Member Process r => Eff r ProcessId self = send SelfPid +trapExit :: Member Process r => Bool -> Eff r ()+trapExit = send . TrapExit++getTrapExit :: Member Process r => Eff r Bool+getTrapExit = send GetTrapExit++raiseError :: Member Process r => String -> Eff r b+raiseError = send . RaiseError++catchProcessError+ :: forall r w . Member Process r => (String -> Eff r w) -> Eff r w -> Eff r w+catchProcessError onErr = interpose return go+ where+ go :: forall b . Process b -> (b -> Eff r w) -> Eff r w+ go (RaiseError emsg) _k = onErr emsg+ go s k = send s >>= k++ignoreProcessError+ :: (HasCallStack, Member Process r) => Eff r a -> Eff r (Either String a)+ignoreProcessError = catchProcessError (return . Left) . fmap Right+ newtype ProcessId = ProcessId { _fromProcessId :: Int } deriving (Eq,Ord,Typeable,Bounded,Num, Enum, Integral, Real) @@ -62,18 +94,17 @@ data MessagePassing b where SendMessage :: Typeable m- => ProcessId- -> Message m- -> MessagePassing Bool- ReceiveMessage- :: forall e m . (Typeable m, Typeable (Message m))- => (Message m -> e)- -> MessagePassing (Maybe e)+ => ProcessId+ -> m+ -> MessagePassing Bool+ ReceiveMessage :: forall e m . (Typeable m, Typeable (Message m))+ => (m -> e)+ -> MessagePassing (Message e) data Message m where- Shutdown :: Message Void+ ProcessControlMessage :: String -> Message m Message :: m -> Message m- deriving Typeable+ deriving (Typeable, Functor, Show, Eq, Ord, Foldable, Traversable) sendMessage :: forall o r@@ -81,19 +112,19 @@ => ProcessId -> o -> Eff r Bool-sendMessage pid message = send (SendMessage pid (Message message))--kill :: (HasCallStack, Member MessagePassing r)- => ProcessId -> Eff r Bool-kill pid = send (SendMessage pid Shutdown)+sendMessage pid message = send (SendMessage pid message) receiveMessage- :: forall o r . (HasCallStack, Member MessagePassing r, Member Process r, Typeable o)- => Proxy o -> Eff r (Maybe o)+ :: forall o r+ . (HasCallStack, Member MessagePassing r, Member Process r, Typeable o)+ => Proxy o+ -> Eff r (Message o) receiveMessage _ = do- me <- self- mRes <- send (ReceiveMessage id)- case mRes of- Just Shutdown -> fail (show me ++ " SHUTDOWN")- Just (Message m) -> return (Just m)- Nothing -> return Nothing+ res <- send (ReceiveMessage id)+ case res of+ Message _ -> return res+ ProcessControlMessage msg -> do+ isTrapExit <- getTrapExit+ if isTrapExit+ then return res+ else raiseError ("received exit message: " ++ msg)
+ src/Control/Eff/Concurrent/Observer.hs view
@@ -0,0 +1,215 @@+-- | Observer Effects+--+-- This module supports the implementation of observers and observables. One+-- more concrete perspective might be to understand observers as event listeners+-- and observables as event sources. The tools in this module are tailored+-- towards 'Control.Eff.Concurrent.GenServer.Api' endpoints+{-# LANGUAGE IncoherentInstances #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE KindSignatures #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE DeriveFunctor #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeOperators #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE GADTs #-}+module Control.Eff.Concurrent.Observer+ ( Observer(..)+ , Observable(..)+ , notifyObserver+ , registerObserver+ , forgetObserver+ , SomeObserver(..)+ , notifySomeObserver+ , Observers()+ , manageObservers+ , addObserver+ , removeObserver+ , notifyObservers+ , CallbackObserver+ , spawnCallbackObserver+ ) where++import GHC.Stack+import Data.Dynamic+import Data.Set (Set)+import qualified Data.Set as Set+import Control.Eff+import Control.Eff.Concurrent.MessagePassing+import Control.Eff.Concurrent.GenServer+import Control.Eff.Concurrent.Dispatcher+import Control.Eff.Log+import Control.Eff.State.Lazy+import Control.Lens+import Control.Monad++-- * Observation API++-- | An 'Api' index that support observation of the+-- another 'Api' that is 'Observable'.+class (Typeable p, Observable o) => Observer p o where+ -- | Wrap the 'Observation' and the 'ProcessId' (i.e. the 'Server')+ -- that caused the observation into a 'Api' value that the+ -- 'Observable' understands.+ observationMessage :: Server o -> Observation o -> Api p 'Asynchronous++-- | An 'Api' index that supports registration and de-registration of+-- 'Observer's.+class (Typeable o, Typeable (Observation o)) => Observable o where+ -- | Type of observations visible on this observable+ data Observation o+ -- | Return the 'Api' value for the 'cast_' that registeres an observer+ registerObserverMessage :: SomeObserver o -> Api o 'Asynchronous+ -- | Return the 'Api' value for the 'cast_' that de-registeres an observer+ forgetObserverMessage :: SomeObserver o -> Api o 'Asynchronous++-- | Send an 'Observation' to an 'Observer'+notifyObserver :: ( Member Process r+ , Member MessagePassing r+ , Observable o+ , Observer p o+ , HasCallStack+ )+ => Server p -> Server o -> Observation o -> Eff r ()+notifyObserver observer observed observation =+ cast_ observer (observationMessage observed observation)++-- | Send the 'registerObserverMessage'+registerObserver :: ( Member Process r+ , Member MessagePassing r+ , Observable o+ , Observer p o+ , HasCallStack+ )+ => Server p -> Server o -> Eff r ()+registerObserver observer observed =+ cast_ observed (registerObserverMessage (SomeObserver observer))++-- | Send the 'forgetObserverMessage'+forgetObserver :: ( Member Process r+ , Member MessagePassing r+ , Observable o+ , Observer p o)+ => Server p -> Server o -> Eff r ()+forgetObserver observer observed =+ cast_ observed (forgetObserverMessage (SomeObserver observer))++-- ** Generalized observation++-- | An existential wrapper around a 'Server' of an 'Observer'.+-- Needed to support different types of observers to observe the+-- same 'Observable' in a general fashion.+data SomeObserver o where+ SomeObserver :: (Show (Server p), Typeable p, Observer p o) => Server p -> SomeObserver o++deriving instance Show (SomeObserver o)++instance Ord (SomeObserver o) where+ compare (SomeObserver (Server o1)) (SomeObserver (Server o2)) =+ compare o1 o2++instance Eq (SomeObserver o) where+ (==) (SomeObserver (Server o1)) (SomeObserver (Server o2)) =+ o1 == o2++-- | Send an 'Observation' to 'SomeObserver'.+notifySomeObserver :: ( Member Process r+ , Member MessagePassing r+ , Observable o+ , HasCallStack+ )+ => Server o+ -> Observation o+ -> SomeObserver o+ -> Eff r ()+notifySomeObserver observed observation (SomeObserver observer) =+ notifyObserver observer observed observation++-- ** Manage 'Observers's++-- | Internal state for 'manageobservers'+data Observers o =+ Observers { _observers :: Set (SomeObserver o) }++observers :: Iso' (Observers o) (Set (SomeObserver o))+observers = iso _observers Observers++-- | Keep track of registered 'Observer's Observers can be added and removed,+-- and an 'Observation' can be sent to all registerd observers at once.+manageObservers :: Eff (State (Observers o) ': r) a -> Eff r a+manageObservers = flip evalState (Observers Set.empty)++-- | Add an 'Observer' to the 'Observers' managed by 'manageObservers'.+addObserver+ :: ( Member MessagePassing r+ , Member (State (Observers o)) r+ , Observable o)+ => SomeObserver o -> Eff r ()+addObserver = modify . over observers . Set.insert++-- | Delete an 'Observer' from the 'Observers' managed by 'manageObservers'.+removeObserver+ :: ( Member MessagePassing r+ , Member (State (Observers o)) r+ , Observable o)+ => SomeObserver o -> Eff r ()+removeObserver = modify . over observers . Set.delete++-- | Send an 'Observation' to all 'SomeObserver's in the 'Observers' state.+notifyObservers+ :: forall o r . ( Observable o+ , Member MessagePassing r+ , Member Process r+ , Member (State (Observers o)) r)+ => Observation o -> Eff r ()+notifyObservers observation = do+ me <- asServer @o <$> self+ os <- view observers <$> get+ mapM_ (notifySomeObserver me observation) os++-- * Callback 'Observer'++-- | An 'Observer' that dispatches the observations to an effectful callback.+data CallbackObserver o+ deriving Typeable++data instance Api (CallbackObserver o) r where+ CbObserved :: (Typeable o, Typeable (Observation o)) =>+ Server o -> Observation o -> Api (CallbackObserver o) 'Asynchronous+ deriving Typeable++deriving instance Show (Observation o) => Show (Api (CallbackObserver o) r)++instance (Observable o) => Observer (CallbackObserver o) o where+ observationMessage = CbObserved++-- | 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)+ => (Server o -> Observation o -> Eff ProcIO Bool)+ -> Eff r (Server (CallbackObserver o))+spawnCallbackObserver onObserve =+ asServer @(CallbackObserver o)+ <$>+ (spawn $ do+ trapExit True+ me <- asServer @(CallbackObserver o) <$> self+ let loopUntil =+ serve_ (ApiHandler @(CallbackObserver o)+ (handleCast loopUntil)+ unhandledCallError+ (logMsg . ((show me ++ " observer terminating ") ++)))+ loopUntil+ )+ where+ handleCast :: Eff ProcIO () -> Api (CallbackObserver o) 'Asynchronous -> Eff ProcIO ()+ handleCast k (CbObserved fromSvr v) = onObserve fromSvr v >>= flip when k
+ src/Control/Eff/ExceptionExtra.hs view
@@ -0,0 +1,43 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE MonoLocalBinds #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE TypeOperators #-}+-- | Add-ons to 'Control.Eff.Exception'+module Control.Eff.ExceptionExtra+ ( try+ , liftRethrow+ , runErrorRethrowIO+ , module X+ )+where++import Control.Monad ( (>=>) )+import qualified Control.Exception as Exc+import Control.Eff+import Control.Eff.Lift+import Control.Eff.Exception as X++-- | Catch an exception and return it in an 'Either'.+try :: forall e r a . Member (Exc e) r => Eff r a -> Eff r (Either e a)+try e = catchError (Right <$> e) (return . Left)++-- | Lift an IO action and catch all error using 'Exc.try' then wrap the+-- 'Exc.Exception' using a given wrapper function and rethrow it using+-- 'X.throwError'.+liftRethrow+ :: forall e r a+ . (Exc.Exception e, SetMember Lift (Lift IO) r, Member (Exc e) r)+ => (Exc.SomeException -> e)+ -> IO a+ -> Eff r a+liftRethrow liftE m = lift (Exc.try m) >>= either (throwError . liftE) return++-- | Run an effect with exceptions like 'X.runError' and rethrow it as+-- 'Exc.SomeException' using 'Exc.throw'+runErrorRethrowIO+ :: forall e r a+ . (Exc.Exception e, SetMember Lift (Lift IO) r)+ => Eff (Exc e ': r) a+ -> Eff r a+runErrorRethrowIO = runError >=> either (Exc.throw . Exc.SomeException) return
+ src/Control/Eff/Log.hs view
@@ -0,0 +1,173 @@+{-# LANGUAGE UndecidableInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE TypeOperators #-}+{-# LANGUAGE TypeInType #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE KindSignatures #-}+{-# LANGUAGE DataKinds #-}+-- | An extensible effect that wraps 'Control.Monad.Log.MonadLog' into an extensible effect.+module Control.Eff.Log+ ( handleLogsWith+ , Logs(..)+ , logMessageFreeEff+ , logMsg+ , module ExtLog+ , LogChannel()+ , logChannelPutIO+ , forwardLogsToChannel+ , forkLogChannel+ , joinLogChannel+ , logChannelBracket+ )+where++import Control.Concurrent+import Control.Concurrent.STM+import Control.Eff as Eff+import Control.Exception ( bracket )+import Control.Monad ( void+ , when+ )+import Control.Monad.Log as ExtLog+ hiding ( )+import Control.Monad.Trans.Control+import Data.Kind+import qualified Control.Eff.Lift as Eff+import qualified Control.Monad.Log as Log++-- | The 'Eff'ect type to wrap 'ExtLog.MonadLog'.+-- This is a+data Logs message a where+ LogMessageFree :: (forall n . Monoid n => (message -> n) -> n) -> Logs message ()++-- | Effectful version of the /strange/ 'ExtLog.logMessageFree' function.+logMessageFreeEff+ :: Member (Logs message) r+ => (forall n . Monoid n => (message -> n) -> n)+ -> Eff r ()+logMessageFreeEff foldMapish = send (LogMessageFree foldMapish)++-- | Effectful version of the 'ExtLog.logMessage' function.+logMsg :: Member (Logs m) r => m -> Eff r ()+logMsg msg = logMessageFreeEff ($ msg)++-- | Handle 'Logs' effects using 'Log.LoggingT' 'Log.Handler's.+handleLogsWith+ :: forall m r message a+ . (Monad m, SetMember Eff.Lift (Eff.Lift m) r)+ => Eff (Logs message ': r) a+ -> (forall b . (Log.Handler m message -> m b) -> m b)+ -> Eff r a+handleLogsWith actionThatLogs foldHandler = Eff.handle_relay return+ go+ actionThatLogs+ where+ go :: Logs message b -> (b -> Eff r c) -> Eff r c+ go (LogMessageFree foldMapish) k =+ Eff.lift (foldHandler (Log.runLoggingT (Log.logMessageFree foldMapish)))+ >>= k++-- * Concurrent Logging++-- | Input queue for a concurrent logger.+data LogChannel message =+ LogChannel { fromLogChannel :: TQueue message+ , logChannelOpen :: TVar Bool+ , logChannelThread :: ThreadId+ }++-- | Send the log messages to a 'LogChannel'.+forwardLogsToChannel+ :: forall r message a+ . (SetMember Eff.Lift (Eff.Lift IO) r)+ => LogChannel message+ -> Eff (Logs message ': r) a+ -> Eff r a+forwardLogsToChannel logChan actionThatLogs = do+ handleLogsWith actionThatLogs+ (\withHandler -> withHandler (logChannelPutIO logChan))++-- | Enqueue a log message into a log channel+logChannelPutIO :: LogChannel message -> message -> IO ()+logChannelPutIO c m = atomically+ (do+ isOpen <- readTVar (logChannelOpen c)+ when isOpen (writeTQueue (fromLogChannel c) m)+ )++-- | Fork 'LogChannel' backed by a process that repeatedly receives log messages+-- sent by 'forwardLogstochannel' or 'logChannelPutIO'. The process logs by+-- invoken the given IO action. To stop and terminate a 'LogChannel' invoke+-- 'joinLogChannel'.+forkLogChannel+ :: forall message+ . (message -> IO ())+ -> Maybe message+ -> IO (LogChannel message)+forkLogChannel handle mFirstMsg = do+ (msgQ, isOpenV) <- atomically+ (do+ tq <- newTQueue+ v <- newTVar True+ mapM_ (writeTQueue tq) mFirstMsg+ return (tq, v)+ )+ thread <- forkFinally (logLoop msgQ isOpenV) (const (cleanUp msgQ isOpenV))+ return (LogChannel msgQ isOpenV thread)+ where+ cleanUp :: TQueue message -> TVar Bool -> IO ()+ cleanUp tq isOpenVar =+ atomically+ (do+ writeTVar isOpenVar False+ flushTQueue tq+ )+ >>= mapM_ handle+ logLoop :: TQueue message -> TVar Bool -> IO ()+ logLoop tq isOpenVar = do+ mMsg <- atomically+ (do+ isOpen <- readTVar isOpenVar+ if isOpen then Just <$> readTQueue tq else return Nothing+ )+ case mMsg of+ Just msg -> do+ handle msg+ logLoop tq isOpenVar+ Nothing -> return ()++-- | Close a log channel. Subsequent loggin requests will no be handled any+-- more.+joinLogChannel :: Maybe message -> LogChannel message -> IO ()+joinLogChannel closeLogMessage (LogChannel tq isOpenVar thread) = do+ wasOpen <- atomically+ (do+ isOpen <- readTVar isOpenVar+ if isOpen+ then do+ writeTVar isOpenVar False+ mapM_ (writeTQueue tq) closeLogMessage+ return True+ else return False+ )+ when wasOpen (killThread thread)+++-- | Wrap 'LogChannel' creation and destruction around a monad action in+-- 'bracket'y manner.+logChannelBracket+ :: Maybe message+ -> Maybe message+ -> (LogChannel message -> IO a)+ -> LoggingT message IO a+logChannelBracket mWelcome mGoodbye f = control+ (\runInIO -> do+ myTId <- myThreadId+ let logHandler = void . runInIO . logMessage+ bracket (forkLogChannel logHandler mWelcome) (joinLogChannel mGoodbye) f+ )