extensible-effects-concurrent 0.2.0.2 → 0.2.0.3

raw patch · 13 files changed

+470/−191 lines, 13 filesPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

API changes (from Hackage documentation)

Files

• ChangeLog.md +6/−0
• extensible-effects-concurrent.cabal +2/−1
• src/Control/Eff/Concurrent/Api.hs +52/−6
• src/Control/Eff/Concurrent/Api/Client.hs +31/−1
• src/Control/Eff/Concurrent/Api/Observer.hs +3/−4
• src/Control/Eff/Concurrent/Api/Server.hs +9/−9
• src/Control/Eff/Concurrent/Process/ForkIOScheduler.hs +13/−1
• src/Control/Eff/Log.hs +227/−81
• test/Common.hs +71/−0
• test/Driver.hs +1/−1
• test/ForkIOScheduler.hs +41/−40
• test/ProcessBehaviourTestCases.hs +11/−43
• test/SingleThreadedScheduler.hs +3/−4

@@ -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 

@@ -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 <https://github.com/sheyll/extensible-effects-concurrent#readme> 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

@@ -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

@@ -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

@@ -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 #-}

@@ -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.

@@ -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

@@ -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)

@@ -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

@@ -1,1 +1,1 @@-{-# OPTIONS_GHC -F -pgmF tasty-discover -optF --tree-display #-}+{-# OPTIONS_GHC -F -pgmF tasty-discover #-}

@@ -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!!"+         )))

@@ -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

@@ -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