diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,5 +1,11 @@ # Changelog for extensible-effects-concurrent +## 0.2.0.3 + + * Improve 'Api' documentation + * Improve `LogChannel` API + * Reorganize unit tests + * Hopefully tune travis ci test parameter enough to get a stable build result ## 0.2.0.2 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.2.0.2 +version: 0.2.0.3 description: Please see the README on GitHub at synopsis: Message passing concurrency as extensible-effect homepage: https://github.com/sheyll/extensible-effects-concurrent#readme @@ -90,6 +90,7 @@ , ForkIOScheduler , SingleThreadedScheduler , ProcessBehaviourTestCases + , Common ghc-options: -threaded -rtsopts -with-rtsopts=-N build-depends: base >=4.9 && <5 diff --git a/src/Control/Eff/Concurrent/Api.hs b/src/Control/Eff/Concurrent/Api.hs --- a/src/Control/Eff/Concurrent/Api.hs +++ b/src/Control/Eff/Concurrent/Api.hs @@ -13,8 +13,21 @@ {-# LANGUAGE TypeApplications #-} {-# LANGUAGE GADTs #-} --- | Type safe /server/ API processes - +-- | This module contains a mechanisms to specify what kind of messages a +-- process can receive and possible answer by sending a message back to the +-- orginator. +-- +-- A message can be either a blocking or a non-blocking request. +-- +-- The type magic in the 'Api' type familiy allows to define a related set of messages along +-- with the corresponding responses. +-- +-- A process can /serve/ a specific 'Api' instance by using the functions provided by +-- the "Control.Eff.Concurrent.Api.Server" module. +-- +-- To enable a process to use such a /service/, the functions provided by +-- the "Control.Eff.Concurrent.Api.Client" should be used. +-- module Control.Eff.Concurrent.Api ( Api , Synchronicity(..) @@ -30,14 +43,43 @@ import Data.Typeable (Typeable, typeRep) import Control.Eff.Concurrent.Process --- | This data family defines an API implemented by a server. --- The first parameter is the API /index/ and the second parameter --- (the @* -> *@) +-- | This data family defines an API, a communication interface description +-- between to processes, where one process acts as a __server__ and the other(s) +-- as __client(s)__. +-- +-- The first parameter is usually a user defined phantom type that identifies +-- the 'Api' instance. +-- +-- The second parameter specifies if a specific constructor of an (GADT-like) +-- @Api@ instance is 'Synchronous', i.e. returns a result and blocks the caller +-- or if it is 'Asynchronous' +-- +-- Example: +-- +-- > +-- > data BookShop deriving Typeable +-- > +-- > data instance Api BookShop r where +-- > RentBook :: BookId -> Api BookShop ('Synchronous (Either RentalError RentalId)) +-- > BringBack :: RentalId -> Api BookShop 'Asynchronous +-- > +-- > type BookId = Int +-- > type RentalId = Int +-- > type RentalError = String +-- > data family Api (api :: Type) (reply :: Synchronicity) -data Synchronicity = Synchronous Type | Asynchronous + +-- | This __data kind__ is used to indicate at the type level if a specific +-- constructor of an @Api@ instance has a result for which some caller has to +-- wait, or if it is asynchronous. +data Synchronicity = + Synchronous Type -- ^ Blocking operation with a specific return type, e.g. @('Synchronous (Either RentalError RentalId))@ + | Asynchronous -- ^ Non-blocking async operation deriving (Typeable) +-- | This is a tag-type that wraps around a 'ProcessId' and holds an 'Api' index +-- type. newtype Server api = Server { _fromServer :: ProcessId } deriving (Eq,Ord,Typeable) @@ -54,8 +96,12 @@ makeLenses ''Server +-- | Tag a 'ProcessId' with an 'Api' type index to mark it a 'Server' process +-- handling that API proxyAsServer :: proxy api -> ProcessId -> Server api proxyAsServer = const Server +-- | Tag a 'ProcessId' with an 'Api' type index to mark it a 'Server' process +-- handling that API asServer :: forall api . ProcessId -> Server api asServer = Server diff --git a/src/Control/Eff/Concurrent/Api/Client.hs b/src/Control/Eff/Concurrent/Api/Client.hs --- a/src/Control/Eff/Concurrent/Api/Client.hs +++ b/src/Control/Eff/Concurrent/Api/Client.hs @@ -13,7 +13,9 @@ {-# LANGUAGE TypeApplications #-} {-# LANGUAGE GADTs #-} --- | Type safe /server/ API processes +-- | Functions for 'Api' clients. +-- +-- This modules is required to write clients that consume an 'Api'. module Control.Eff.Concurrent.Api.Client ( cast @@ -37,6 +39,11 @@ import Data.Typeable (Typeable, typeRep) import GHC.Stack +-- | Send an 'Api' request that has no return value and return as fast as +-- possible. The type signature enforces that the corresponding 'Api' clause is +-- 'Asynchronous'. Return @True@ if the message was sent to the process. Note +-- that this is totally not the same as that the request was successfully +-- handled. If that is important, use 'call' instead. castChecked :: forall r q o . ( HasCallStack @@ -51,6 +58,9 @@ castChecked px (Server pid) callMsg = sendMessage px pid (toDyn (Cast callMsg)) +-- | Send an 'Api' request that has no return value and return as fast as +-- possible. The type signature enforces that the corresponding 'Api' clause is +-- 'Asynchronous'. cast :: forall r q o . ( HasCallStack @@ -66,6 +76,10 @@ do _ <- castChecked px toServer apiRequest return () +-- | Send an 'Api' request and wait for the server to return a result value. +-- +-- The type signature enforces that the corresponding 'Api' clause is +-- 'Synchronous'. call :: forall result api r q . ( SetMember Process (Process q) r @@ -90,22 +104,36 @@ else raiseError px ("Could not send request message " ++ show (typeRep requestMessage)) +-- * Registered Servers +-- | Instead of passing around a 'Server' value and passing to functions like +-- 'cast' or 'call', a 'Server' can provided by a 'Reader' effect, if there is +-- only a __single server__ for a given 'Api' instance. This type alias is +-- convenience to express that an effect has 'Process' and a reader for a +-- 'Server'. type ServesApi o r q = ( Typeable o , SetMember Process (Process q) r , Member (Reader (Server o)) r ) +-- | Run a reader effect that contains __the one__ server handling a specific +-- 'Api' instance. registerServer :: Server o -> Eff ( Reader (Server o) ': r ) a -> Eff r a registerServer = flip runReader +-- | Like 'call' but take the 'Server' from the reader provided by +-- 'registerServer'. callRegistered :: (Typeable reply, ServesApi o r q) => SchedulerProxy q -> Api o ('Synchronous reply) -> Eff r reply callRegistered px method = do serverPid <- ask call px serverPid method +-- | Like 'callRegistered' but also catch errors raised if e.g. the server +-- crashed. By allowing 'Alternative' instances to contain the reply, +-- application level errors can be combined with errors rising from inter +-- process communication. callRegisteredA :: forall r q o f reply . (Alternative f, Typeable f, Typeable reply, ServesApi o r q) @@ -116,6 +144,8 @@ (const (return (empty @f))) (callRegistered px method) +-- | Like 'cast' but take the 'Server' from the reader provided by +-- 'registerServer'. castRegistered :: (Typeable o, ServesApi o r q) => SchedulerProxy q -> Api o 'Asynchronous -> Eff r () castRegistered px method = do diff --git a/src/Control/Eff/Concurrent/Api/Observer.hs b/src/Control/Eff/Concurrent/Api/Observer.hs --- a/src/Control/Eff/Concurrent/Api/Observer.hs +++ b/src/Control/Eff/Concurrent/Api/Observer.hs @@ -1,9 +1,8 @@ -- | 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.Api.Api' endpoints +-- This module supports the implementation of observers and observables. One use +-- case is event propagation. The tools in this module are tailored towards +-- 'Api' servers/clients. {-# LANGUAGE IncoherentInstances #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE ConstraintKinds #-} diff --git a/src/Control/Eff/Concurrent/Api/Server.hs b/src/Control/Eff/Concurrent/Api/Server.hs --- a/src/Control/Eff/Concurrent/Api/Server.hs +++ b/src/Control/Eff/Concurrent/Api/Server.hs @@ -13,14 +13,18 @@ {-# LANGUAGE TypeApplications #-} {-# LANGUAGE GADTs #-} --- | Type safe /server/ API processes +-- | Functions to implement 'Api' __servers__. module Control.Eff.Concurrent.Api.Server - ( ApiHandler (..) - , serve + ( + -- * Api Server + serve + , ApiHandler (..) + -- * ApiHandler default callbacks , unhandledCallError , unhandledCastError , defaultTermination + -- * Multi Api Server , serveBoth , serve3 , tryApiHandler @@ -43,8 +47,7 @@ import Data.Dynamic import GHC.Stack --- | Receive messages until the process exits and invoke the callback on each --- message. +-- | Receive and process incoming requests until the process exits, using an 'ApiHandler'. serve :: forall r q p . (Typeable p, SetMember Process (Process q) r, HasCallStack) @@ -87,7 +90,7 @@ } -> ApiHandler p r -- | Apply either the '_handleCall', '_handleCast' or the '_handleTerminate' --- callback to an incoming 'Request'. +-- callback to an incoming 'Request'. Note, it is unlikely that this function must be used. applyApiHandler :: forall r q p . ( Typeable p , SetMember Process (Process q) r @@ -107,8 +110,6 @@ sendReply reply = sendMessage px fromPid (toDyn (Response (Proxy @p) reply)) --- * ApiHandler default callbacks - -- | A default handler to use in '_handleCall' in 'ApiHandler'. It will call -- 'raiseError' with a nice error message. unhandledCallError @@ -156,7 +157,6 @@ defaultTermination px e = maybe (exitNormally px) (exitWithError px) e --- * Multiple ApiHandler combined -- | 'serve' two 'ApiHandler's at once. The first handler is used for -- termination handling. diff --git a/src/Control/Eff/Concurrent/Process/ForkIOScheduler.hs b/src/Control/Eff/Concurrent/Process/ForkIOScheduler.hs --- a/src/Control/Eff/Concurrent/Process/ForkIOScheduler.hs +++ b/src/Control/Eff/Concurrent/Process/ForkIOScheduler.hs @@ -34,6 +34,7 @@ module Control.Eff.Concurrent.Process.ForkIOScheduler ( schedule , defaultMain + , defaultMainWithLogChannel , SchedulerError(..) , SchedulerIO , forkIoScheduler @@ -64,6 +65,7 @@ import Data.Dynamic import Data.Map ( Map ) import qualified Data.Map as Map +import Data.String -- | Information about a process, needed to implement 'MessagePassing' and -- 'Process' handlers. The message queue is backed by a 'STM.TQueue' and contains @@ -198,11 +200,21 @@ defaultMain c = runLoggingT (logChannelBracket + 128 (Just "~~~~~~ main process started") (Just "====== main process exited") (schedule c)) (print :: String -> IO ()) +-- | Start the message passing concurrency system then execute a 'Process' on +-- top of 'SchedulerIO' effect. All logging is sent to standard output. +defaultMainWithLogChannel :: LogChannel String -> Eff (ConsProcess SchedulerIO) () -> IO () +defaultMainWithLogChannel logC c = + closeLogChannelAfter + (Just (fromString "====== main process exited")) + logC + (schedule c logC) + scheduleProcessWithShutdownAction :: SchedulerVar -> STM.TMVar ProcessId @@ -222,7 +234,7 @@ runProcEffects cleanupVar l = Data.Bifunctor.first Exc.SomeException <$> runLift - (forwardLogsToChannel l + (logToChannel l (runReader (scheduleProcessWithCleanup shutdownAction diff --git a/src/Control/Eff/Log.hs b/src/Control/Eff/Log.hs --- a/src/Control/Eff/Log.hs +++ b/src/Control/Eff/Log.hs @@ -10,20 +10,29 @@ {-# LANGUAGE GADTs #-} {-# LANGUAGE KindSignatures #-} {-# LANGUAGE DataKinds #-} --- | An extensible effect that wraps 'Control.Monad.Log.MonadLog' into an extensible effect. +-- | A logging effect based on 'Control.Monad.Log.MonadLog'. module Control.Eff.Log - ( handleLogsWith - , Logs(..) + ( + -- * Logging Effect + Logs(..) , logMsg + , foldLog + , foldLogFast , module ExtLog - , LogChannel() - , logChannelPutIO , captureLogs , ignoreLogs - , forwardLogsToChannel - , forkLogChannel + , handleLogsWith + -- * Concurrent Logging + , LogChannel() + , logToChannel + , noLogger + , forkLogger + , filterLogChannel , joinLogChannel + , killLogChannel + , closeLogChannelAfter , logChannelBracket + , logChannelPutIO ) where @@ -31,27 +40,81 @@ import Control.Concurrent.STM import Control.Eff as Eff import Control.Exception ( bracket ) +import qualified Control.Exception as Exc import Control.Monad ( void , when + , unless ) 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 +import Data.Foldable ( traverse_ ) +import Data.Kind() import Data.Sequence (Seq()) import qualified Data.Sequence as Seq +import Data.String +import Data.Typeable --- | The 'Eff'ect type to wrap 'ExtLog.MonadLog'. --- This is a +-- | Logging effect type, parameterized by a log message type. data Logs message a where LogMsg :: message -> Logs message () --- | Effectful version of the 'ExtLog.logMessage' function. +-- | Log a message. logMsg :: Member (Logs m) r => m -> Eff r () logMsg msg = send (LogMsg msg) +-- | Change, add or remove log messages. +-- +-- Requirements: +-- +-- * All log meta data for typical prod code can be added without +-- changing much of the code +-- +-- * Add timestamp to a log messages of a sub-computation. +-- +-- * Write some messages to a file. +-- +-- * Log something extra, e.g. runtime memory usage in load tests +-- +-- Approach: Install a callback that sneaks into to log message +-- sending/receiving, to intercept the messages and execute some code and then +-- return a new message. +foldLog :: forall r m a . Member (Logs m) r + => (m -> Eff r ()) -> Eff r a -> Eff r a +foldLog interceptor effect = + interpose return go effect + where + go :: Member (Logs m) r => Logs m x -> (Arr r x y) -> Eff r y + go (LogMsg m) k = + do interceptor m + k () + +-- | Change, add or remove log messages without side effects, faster than +-- 'foldLog'. +-- +-- Requirements: +-- +-- * Tests run fast in unit tests so travis won't time out +-- +-- * Drop debug logs +-- +-- * /Grep like/ log filtering +-- +-- Approach: Install a callback that sneaks into to log message +-- sending/receiving, to intercept the messages and execute some code and then +-- return a new message. +foldLogFast :: forall r m a f . (Foldable f, Member (Logs m) r) + => (m -> f m) -> Eff r a -> Eff r a +foldLogFast interceptor effect = + interpose return go effect + where + go :: Member (Logs m) r => Logs m x -> (Arr r x y) -> Eff r y + go (LogMsg m) k = + do traverse_ logMsg (interceptor m) + k () + -- | Capture the logs in a 'Seq'. captureLogs :: Eff (Logs message ': r) a -> Eff r (a, Seq message) @@ -93,100 +156,183 @@ go (LogMsg m) k = Eff.lift (foldHandler (\doLog -> doLog m)) >>= k --- * Concurrent Logging - --- | Input queue for a concurrent logger. +-- | A log channel processes logs from the 'Logs' effect by en-queuing them in a +-- shared queue read from a seperate processes. A channel can contain log +-- message filters. data LogChannel message = - LogChannel { fromLogChannel :: TQueue message - , logChannelOpen :: TVar Bool - , _logChannelThread :: ThreadId - } + FilteredLogChannel (message -> Bool) (LogChannel message) + -- ^ filter log messages + | DiscardLogs + -- ^ discard all log messages + | ConcurrentLogChannel + { fromLogChannel :: TBQueue message + , _logChannelThread :: ThreadId + } + -- ^ send all log messages to a log process -- | Send the log messages to a 'LogChannel'. -forwardLogsToChannel +logToChannel :: 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 +logToChannel 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) - ) +logChannelPutIO DiscardLogs _ = + return () +logChannelPutIO (FilteredLogChannel f lc) m = + when (f m) (logChannelPutIO lc m) +logChannelPutIO c m = + atomically $ do + dropMessage <- isFullTBQueue (fromLogChannel c) + unless dropMessage (writeTBQueue (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 +-- | Create a 'LogChannel' that will discard all messages sent +-- via 'forwardLogstochannel' or 'logChannelPutIO'. +noLogger :: LogChannel message +noLogger = DiscardLogs + +-- | Fork a new process, that applies a monadic action to all log messages sent +-- via 'logToChannel' or 'logChannelPutIO'. +forkLogger :: forall message - . (message -> IO ()) - -> Maybe message + . (Typeable message, Show message) + => Int -- ^ Size of the log message input queue. If the queue is full, message + -- are dropped silently. + -> (message -> IO ()) -- ^ An IO action to log the messages + -> Maybe message -- ^ Optional __first__ message to log -> IO (LogChannel message) -forkLogChannel handle mFirstMsg = do - (msgQ, isOpenV) <- atomically +forkLogger queueLen handle mFirstMsg = do + msgQ <- atomically (do - tq <- newTQueue - v <- newTVar True - mapM_ (writeTQueue tq) mFirstMsg - return (tq, v) + tq <- newTBQueue queueLen + mapM_ (writeTBQueue tq) mFirstMsg + return tq ) - thread <- forkFinally (logLoop msgQ isOpenV) (const (cleanUp msgQ isOpenV)) - return (LogChannel msgQ isOpenV thread) + thread <- forkFinally (logLoop msgQ) (writeLastLogs msgQ) + return (ConcurrentLogChannel msgQ 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 () + writeLastLogs :: TBQueue message -> Either Exc.SomeException () -> IO () + writeLastLogs tq ee = + do logMessages <- atomically $ flushTBQueue tq + case ee of + Right _ -> return () + Left se -> + case Exc.fromException se of + Just (JoinLogChannelException mCloseMsg) -> + do traverse_ handle logMessages + traverse_ handle mCloseMsg + Nothing -> + case Exc.fromException se of + Just (KillLogChannelException mCloseMsg) -> + traverse_ handle mCloseMsg + Nothing -> + mapM_ handle logMessages --- | 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) + logLoop :: TBQueue message -> IO () + logLoop tq = do + m <- atomically $ readTBQueue tq + handle m + logLoop tq +-- | Filter logs sent to a 'LogChannel' using a predicate. +filterLogChannel :: (message -> Bool) -> LogChannel message -> LogChannel message +filterLogChannel = FilteredLogChannel +-- | Run an action and close a 'LogChannel' created by 'noLogger', 'forkLogger' +-- or 'filterLogChannel' afterwards using 'joinLogChannel'. If a +-- 'Exc.SomeException' was thrown, the log channel is killed with +-- 'killLogChannel', and the exception is re-thrown. +closeLogChannelAfter :: (Show message, Typeable message, IsString message) + => Maybe message + -> LogChannel message + -> IO a + -> IO a +closeLogChannelAfter mGoodbye logC ioAction = + do res <- closeLogAndRethrow `Exc.handle` ioAction + closeLogSuccess + return res + where + closeLogAndRethrow :: Exc.SomeException -> IO a + closeLogAndRethrow se = + do let closeMsg = Just (fromString (Exc.displayException se)) + void $ Exc.try @Exc.SomeException + $ killLogChannel closeMsg logC + Exc.throw se + + closeLogSuccess :: IO () + closeLogSuccess = + joinLogChannel mGoodbye logC + +-- | Close a log channel created by e.g. 'forkLogger'. Message already enqueue +-- are handled, as well as an optional final message. Subsequent log message +-- will not be handled anymore. If the log channel must be closed immediately, +-- use 'killLogChannel' instead. +joinLogChannel :: (Show message, Typeable message) + => Maybe message -> LogChannel message -> IO () +joinLogChannel _closeLogMessage DiscardLogs = return () +joinLogChannel Nothing (FilteredLogChannel _f lc) = joinLogChannel Nothing lc +joinLogChannel (Just closeLogMessage) (FilteredLogChannel f lc) = + if f closeLogMessage then + joinLogChannel (Just closeLogMessage) lc + else + joinLogChannel Nothing lc +joinLogChannel closeLogMessage (ConcurrentLogChannel _tq thread) = do + throwTo thread (JoinLogChannelException closeLogMessage) + +-- | Close a log channel quickly, without logging messages already in the queue. +-- Subsequent logging requests will not be handled anymore. If the log channel +-- must be closed without loosing any messages, use 'joinLogChannel' instead. +killLogChannel :: (Show message, Typeable message) + => Maybe message -> LogChannel message -> IO () +killLogChannel _closeLogMessage DiscardLogs = return () +killLogChannel Nothing (FilteredLogChannel _f lc) = killLogChannel Nothing lc +killLogChannel (Just closeLogMessage) (FilteredLogChannel f lc) = + if f closeLogMessage then + killLogChannel (Just closeLogMessage) lc + else + killLogChannel Nothing lc +killLogChannel closeLogMessage (ConcurrentLogChannel _tq thread) = + throwTo thread (KillLogChannelException closeLogMessage) + +-- | Internal exception to shutdown a 'LogChannel' process created by +-- 'forkLogger'. This exception is handled such that all message already +-- en-queued are handled and then an optional final message is written. +newtype JoinLogChannelException m = JoinLogChannelException (Maybe m) + deriving (Show, Typeable) + +instance (Typeable m, Show m) => Exc.Exception (JoinLogChannelException m) + +-- | Internal exception to **immediately** shutdown a 'LogChannel' process +-- created by 'forkLogger', other than 'JoinLogChannelException' the message queue +-- will not be flushed, not further messages will be logged, except for the +-- optional final message. +newtype KillLogChannelException m = KillLogChannelException (Maybe m) + deriving (Show, Typeable) + +instance (Typeable m, Show m) => Exc.Exception (KillLogChannelException m) + -- | Wrap 'LogChannel' creation and destruction around a monad action in --- 'bracket'y manner. +-- 'bracket'y manner. This function uses 'joinLogChannel', so en-queued messages +-- are flushed on exit. The resulting action in in the 'LoggingT' monad, which +-- is essentially a reader for the log handler function. logChannelBracket - :: Maybe message - -> Maybe message - -> (LogChannel message -> IO a) + :: (Show message, Typeable message) + => Int -- ^ Size of the log message input queue. If the queue is full, message + -- are dropped silently. + -> Maybe message -- ^ Optional __first__ message to log + -> Maybe message -- ^ Optional __last__ message to log + -> (LogChannel message -> IO a) -- ^ An IO action that will use the + -- 'LogChannel', after the action returns (even + -- because of an exception) the log channel is + -- destroyed. -> LoggingT message IO a -logChannelBracket mWelcome mGoodbye f = control +logChannelBracket queueLen mWelcome mGoodbye f = control (\runInIO -> do let logHandler = void . runInIO . logMessage - bracket (forkLogChannel logHandler mWelcome) (joinLogChannel mGoodbye) f) + bracket (forkLogger queueLen logHandler mWelcome) (joinLogChannel mGoodbye) f) diff --git a/test/Common.hs b/test/Common.hs new file mode 100644 --- /dev/null +++ b/test/Common.hs @@ -0,0 +1,71 @@ +module Common where + +import Control.Concurrent.STM +import Control.Eff.Concurrent.Process +import Control.Eff +import Control.Eff.Log +import Control.Eff.Lift +import Control.Monad (void) +import Test.Tasty +import Test.Tasty.HUnit +import Test.Tasty.Runners + +setTravisTestOptions :: TestTree -> TestTree +setTravisTestOptions = localOption (timeoutSeconds 300) . localOption (NumThreads 1) + +timeoutSeconds :: Integer -> Timeout +timeoutSeconds seconds = Timeout (seconds * 1000000) (show seconds ++ "s") + +withTestLogC + :: (e -> LogChannel String -> IO ()) -> (IO (e -> IO ()) -> TestTree) -> TestTree +withTestLogC doSchedule k = + withResource + testLogC + testLogJoin + (\ logCFactory -> + k (return + (\e -> + do logC <- logCFactory + doSchedule e logC))) + +testLogC :: IO (LogChannel String) +testLogC = + filterLogChannel (\m -> take (length logPrefix) m == logPrefix) + <$> + forkLogger 1 (putStrLn) (Just (logPrefix ++ "^^^^^^^^^^^^^^ LogChannel Start ^^^^^^^^^^^^^^")) + +testLogJoin :: LogChannel String -> IO () +testLogJoin = joinLogChannel (Just (logPrefix ++ "^^^^^^^^^^^^^^ LogChannel Done ^^^^^^^^^^^^^^")) + +tlog :: Member (Logs String) r => String -> Eff r () +tlog = logMsg . (logPrefix ++) + +logPrefix :: String +logPrefix = "[TEST] " + +untilShutdown + :: Member t r => t (ResumeProcess v) -> Eff r () +untilShutdown pa = do + r <- send pa + case r of + ShutdownRequested -> return () + _ -> untilShutdown pa + +scheduleAndAssert :: forall r . + (SetMember Lift (Lift IO) r, Member (Logs String) r) + => IO (Eff (Process r ': r) () -> IO ()) + -> ((String -> Bool -> Eff (Process r ': r) ()) -> Eff (Process r ': r) ()) + -> IO () +scheduleAndAssert schedulerFactory testCaseAction = + do resultVar <- newEmptyTMVarIO + void (applySchedulerFactory schedulerFactory + (testCaseAction (\ title cond -> + do lift (atomically (putTMVar resultVar (title, cond)))))) + (title, result) <- atomically (takeTMVar resultVar) + assertBool title result + +applySchedulerFactory :: forall r . (Member (Logs String) r, SetMember Lift (Lift IO) r) + => IO (Eff (Process r ': r) () -> IO ()) -> Eff (Process r ': r) () -> IO () +applySchedulerFactory factory procAction = + do scheduler <- factory + scheduler procAction diff --git a/test/Driver.hs b/test/Driver.hs --- a/test/Driver.hs +++ b/test/Driver.hs @@ -1,1 +1,1 @@ -{-# OPTIONS_GHC -F -pgmF tasty-discover -optF --tree-display #-} +{-# OPTIONS_GHC -F -pgmF tasty-discover #-} diff --git a/test/ForkIOScheduler.hs b/test/ForkIOScheduler.hs --- a/test/ForkIOScheduler.hs +++ b/test/ForkIOScheduler.hs @@ -11,19 +11,20 @@ import Test.Tasty import Test.Tasty.HUnit import Data.Dynamic +import Common timeoutSeconds :: Integer -> Timeout timeoutSeconds seconds = Timeout (seconds * 1000000) (show seconds ++ "s") test_IOExceptionsIsolated :: TestTree test_IOExceptionsIsolated = - localOption - (timeoutSeconds 30) + setTravisTestOptions $ testGroup "one process throws an IO exception, the other continues unimpaired" [ testCase ("process 2 exits with: "++ howToExit ++ " - while process 1 is busy with: " ++ busyWith) $ do aVar <- newEmptyTMVarIO - Scheduler.defaultMain + lc <- testLogC + Scheduler.defaultMainWithLogChannel lc $ do p1 <- spawn $ forever busyEffect lift (threadDelay 1000) void $ spawn $ do lift (threadDelay 1000) @@ -52,59 +53,59 @@ test_mainProcessSpawnsAChildAndReturns :: TestTree test_mainProcessSpawnsAChildAndReturns = - localOption - (timeoutSeconds 30) + setTravisTestOptions (testCase "spawn a child and return" - (Scheduler.defaultMain - (void (spawn (void (receiveMessage forkIoScheduler)))))) + (do lc <- testLogC + Scheduler.defaultMainWithLogChannel lc + (void (spawn (void (receiveMessage forkIoScheduler)))))) test_mainProcessSpawnsAChildAndExitsNormally :: TestTree test_mainProcessSpawnsAChildAndExitsNormally = - localOption - (timeoutSeconds 30) + setTravisTestOptions (testCase "spawn a child and exit normally" - (Scheduler.defaultMain - (do void (spawn (void (receiveMessage forkIoScheduler))) - void (exitNormally forkIoScheduler) - fail "This should not happen!!" - ))) + (do lc <- testLogC + Scheduler.defaultMainWithLogChannel lc + (do void (spawn (void (receiveMessage forkIoScheduler))) + void (exitNormally forkIoScheduler) + fail "This should not happen!!" + ))) test_mainProcessSpawnsAChildInABusySendLoopAndExitsNormally :: TestTree test_mainProcessSpawnsAChildInABusySendLoopAndExitsNormally = - localOption - (timeoutSeconds 300) + setTravisTestOptions (testCase "spawn a child with a busy send loop and exit normally" - (Scheduler.defaultMain - (do void (spawn (forever (void (sendMessage forkIoScheduler 1000 (toDyn "test"))))) - void (exitNormally forkIoScheduler) - fail "This should not happen!!" - ))) + (do lc <- testLogC + Scheduler.defaultMainWithLogChannel lc + (do void (spawn (forever (void (sendMessage forkIoScheduler 1000 (toDyn "test"))))) + void (exitNormally forkIoScheduler) + fail "This should not happen!!" + ))) test_mainProcessSpawnsAChildBothReturn :: TestTree test_mainProcessSpawnsAChildBothReturn = - localOption - (timeoutSeconds 30) + setTravisTestOptions (testCase "spawn a child and let it return and return" - (Scheduler.defaultMain - (do child <- spawn (void (receiveMessageAs @String forkIoScheduler)) - True <- sendMessage forkIoScheduler child (toDyn "test") - return () - ))) + (do lc <- testLogC + Scheduler.defaultMainWithLogChannel lc + (do child <- spawn (void (receiveMessageAs @String forkIoScheduler)) + True <- sendMessage forkIoScheduler child (toDyn "test") + return () + ))) test_mainProcessSpawnsAChildBothExitNormally :: TestTree test_mainProcessSpawnsAChildBothExitNormally = - localOption - (timeoutSeconds 30) + setTravisTestOptions (testCase "spawn a child and let it exit and exit" - (Scheduler.defaultMain - (do child <- spawn - (do void (receiveMessageAs @String forkIoScheduler) - void (exitNormally forkIoScheduler) - error "This should not happen (child)!!" - ) - True <- sendMessage forkIoScheduler child (toDyn "test") - void (exitNormally forkIoScheduler) - error "This should not happen!!" - ))) + (do lc <- testLogC + Scheduler.defaultMainWithLogChannel lc + (do child <- spawn + (do void (receiveMessageAs @String forkIoScheduler) + void (exitNormally forkIoScheduler) + error "This should not happen (child)!!" + ) + True <- sendMessage forkIoScheduler child (toDyn "test") + void (exitNormally forkIoScheduler) + error "This should not happen!!" + ))) diff --git a/test/ProcessBehaviourTestCases.hs b/test/ProcessBehaviourTestCases.hs --- a/test/ProcessBehaviourTestCases.hs +++ b/test/ProcessBehaviourTestCases.hs @@ -14,30 +14,24 @@ import Control.Eff.Lift import Control.Monad (void, replicateM, forever, when) import Test.Tasty -import Test.Tasty.Runners import Test.Tasty.HUnit - -timeoutSeconds :: Integer -> Timeout -timeoutSeconds seconds = Timeout (seconds * 1000000) (show seconds ++ "s") +import Common test_forkIo :: TestTree test_forkIo = - localOption (NumThreads 1) - (withResource - (forkLogChannel (const (return ())) Nothing) - (joinLogChannel Nothing) - (\ logCFactory -> + setTravisTestOptions + (withTestLogC ForkIO.schedule + (\factory -> testGroup "ForkIOScheduler" - [allTests - (return - (\e -> - do logC <- logCFactory - ForkIO.schedule e logC))])) + [allTests factory])) test_singleThreaded :: TestTree test_singleThreaded = - testGroup "SingleThreadedScheduler" - [allTests (return SingleThreaded.defaultMain)] + setTravisTestOptions + (withTestLogC (const . SingleThreaded.defaultMain) + (\factory -> + testGroup "SingleThreadedScheduler" + [allTests factory])) allTests :: forall r . (Member (Logs String) r, SetMember Lift (Lift IO) r) => IO (Eff (Process r ': r) () -> IO ()) @@ -295,6 +289,7 @@ $ \assertEff -> do p1 <- spawn $ forever busyEffect lift (threadDelay 1000) + tlog "Yeah" void $ spawn $ do lift (threadDelay 1000) doExit lift (threadDelay 100000) @@ -396,30 +391,3 @@ assertEff "" (a == "OK") ] ] - -untilShutdown - :: Member t r => t (ResumeProcess v) -> Eff r () -untilShutdown pa = do - r <- send pa - case r of - ShutdownRequested -> return () - _ -> untilShutdown pa - -scheduleAndAssert :: forall r . - (SetMember Lift (Lift IO) r, Member (Logs String) r) - => IO (Eff (Process r ': r) () -> IO ()) - -> ((String -> Bool -> Eff (Process r ': r) ()) -> Eff (Process r ': r) ()) - -> IO () -scheduleAndAssert schedulerFactory testCaseAction = - do resultVar <- newEmptyTMVarIO - void (applySchedulerFactory schedulerFactory - (testCaseAction (\ title cond -> - do lift (atomically (putTMVar resultVar (title, cond)))))) - (title, result) <- atomically (takeTMVar resultVar) - assertBool title result - -applySchedulerFactory :: forall r . (Member (Logs String) r, SetMember Lift (Lift IO) r) - => IO (Eff (Process r ': r) () -> IO ()) -> Eff (Process r ': r) () -> IO () -applySchedulerFactory factory procAction = - do scheduler <- factory - scheduler procAction diff --git a/test/SingleThreadedScheduler.hs b/test/SingleThreadedScheduler.hs --- a/test/SingleThreadedScheduler.hs +++ b/test/SingleThreadedScheduler.hs @@ -6,14 +6,14 @@ import Test.Tasty import Test.Tasty.HUnit import Data.Dynamic +import Common timeoutSeconds :: Integer -> Timeout timeoutSeconds seconds = Timeout (seconds * 1000000) (show seconds ++ "s") test_mainProcessSpawnsAChildAndExitsNormally :: TestTree test_mainProcessSpawnsAChildAndExitsNormally = - localOption - (timeoutSeconds 2) + setTravisTestOptions (testCase "spawn a child and exit normally" (Scheduler.defaultMain (do void (spawn (void (receiveMessage singleThreadedIoScheduler))) @@ -23,8 +23,7 @@ test_mainProcessSpawnsAChildBothExitNormally :: TestTree test_mainProcessSpawnsAChildBothExitNormally = - localOption - (timeoutSeconds 30) + setTravisTestOptions (testCase "spawn a child and let it exit and exit" (Scheduler.defaultMain (do child <- spawn