diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -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 diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ Message passing concurrency with 'forkIO' and 'extensible-effects' inspired by Erlang. -[![Build Status](https://travis-ci.org/sheyll/extensible-effects-concurrent.svg?branch=0.5)](https://travis-ci.org/sheyll/extensible-effects-concurrent) +[![Build Status](https://travis-ci.org/sheyll/extensible-effects-concurrent.svg?branch=master)](https://travis-ci.org/sheyll/extensible-effects-concurrent) [![Hackage](https://img.shields.io/badge/hackage-extensible-effects-concurrent-green.svg?style=flat)](http://hackage.haskell.org/package/extensible-effects-concurrent) 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.1.0 +version: 0.1.2.0 description: Please see the README on GitHub at 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, 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,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)) 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 @@ -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 diff --git a/src/Control/Eff/Concurrent/Examples2.hs b/src/Control/Eff/Concurrent/Examples2.hs new file mode 100644 --- /dev/null +++ b/src/Control/Eff/Concurrent/Examples2.hs @@ -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 ()) 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 @@ -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) ++ ")") 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,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) diff --git a/src/Control/Eff/Concurrent/Observer.hs b/src/Control/Eff/Concurrent/Observer.hs new file mode 100644 --- /dev/null +++ b/src/Control/Eff/Concurrent/Observer.hs @@ -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 diff --git a/src/Control/Eff/ExceptionExtra.hs b/src/Control/Eff/ExceptionExtra.hs new file mode 100644 --- /dev/null +++ b/src/Control/Eff/ExceptionExtra.hs @@ -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 diff --git a/src/Control/Eff/Log.hs b/src/Control/Eff/Log.hs new file mode 100644 --- /dev/null +++ b/src/Control/Eff/Log.hs @@ -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 + )