diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,5 +1,18 @@ # Changelog for extensible-effects-concurrent +## 0.9.2 + +- Try to adapt the dependency versions to make hackage happy again + +## 0.9.1 + +- Add smart constructors for `MessageSelector` +- Remove `ReceiveMessage` `Process` action +- Rename `ReceiveMessageSuchThat` to `ReceiveSelectedMessage` +- Improve some Show instances, e.g. ProcessId +- Rewrite Logging API: + - Vastly simplified API + ## 0.9.0 - Make `ForkIOScheduler` faster and more robust @@ -18,7 +31,7 @@ unique `Int`s - Rename `spawnServer` to `spawnServerWithEffects` and add a simpler version of `spawnServerWithEffects` called `spawnServer` -- Make all `ApiHandler` handler callbacks optional (by changing the type to `Maybe ...`) +- Make all `ApiHandler` handler callbacks optional (by changing the type to `Maybe ...`) - `ApiHandler` must now return an `ApiServerCmd`. - Add `ApiServerCmd` which allows handler functions to leave to server loop without exitting the process diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -4,10 +4,105 @@ [![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) +[![Hackage](https://img.shields.io/hackage/v/extensible-effects-concurrent.svg?style=flat)](http://hackage.haskell.org/package/extensible-effects-concurrent) -[![extensible-effects-concurrent LTS](http://stackage.org/package/extensible-effects-concurrent/badge/lts)](http://stackage.org/lts/package/extensible-effects-concurrent) +Also included: +- Logging + +- Memory Leak Free `forever` + ## Example -TODO +```haskell +module Main where + +import Control.Eff +import Control.Eff.Lift +import Control.Eff.Concurrent +import Data.Dynamic +import Control.Concurrent +import Control.DeepSeq + +main :: IO () +main = defaultMain + (do + lift (threadDelay 100000) -- because of async logging + firstExample forkIoScheduler + lift (threadDelay 100000) -- ... async logging + ) + -- The SchedulerProxy paremeter contains the effects of a specific scheduler + -- implementation. + +newtype WhoAreYou = WhoAreYou ProcessId deriving (Typeable, NFData) + +firstExample :: (HasLoggingIO q) => SchedulerProxy q -> Eff (Process q ': q) () +firstExample px = do + person <- spawn + (do + logInfo "I am waiting for someone to ask me..." + WhoAreYou replyPid <- receiveMessageAs px + sendMessageAs px replyPid "Alice" + logInfo (show replyPid ++ " just needed to know it.") + ) + me <- self px + sendMessageAs px person (WhoAreYou me) + personName <- receiveMessageAs px + logInfo ("I just met " ++ personName) +``` + +**Running** this example causes this output: +(_not entirely true because of async logging, but true enough_) + +```text +2018-11-05T10:50:42 DEBUG scheduler loop entered ForkIOScheduler.hs line 131 +2018-11-05T10:50:42 DEBUG !1 [ThreadId 11] enter process ForkIOScheduler.hs line 437 +2018-11-05T10:50:42 NOTICE !1 [ThreadId 11] ++++++++ main process started ++++++++ ForkIOScheduler.hs line 394 +2018-11-05T10:50:42 DEBUG !2 [ThreadId 12] enter process ForkIOScheduler.hs line 437 +2018-11-05T10:50:42 INFO !2 [ThreadId 12] I am waiting for someone to ask me... Main.hs line 27 +2018-11-05T10:50:42 INFO !2 [ThreadId 12] !1 just needed to know it. Main.hs line 30 +2018-11-05T10:50:42 DEBUG !2 [ThreadId 12] returned ForkIOScheduler.hs line 440 +2018-11-05T10:50:42 INFO !1 [ThreadId 11] I just met Alice Main.hs line 35 +2018-11-05T10:50:42 NOTICE !1 [ThreadId 11] ++++++++ main process returned ++++++++ ForkIOScheduler.hs line 396 +2018-11-05T10:50:42 DEBUG !1 [ThreadId 11] returned ForkIOScheduler.hs line 440 +2018-11-05T10:50:42 DEBUG scheduler loop returned ForkIOScheduler.hs line 133 +2018-11-05T10:50:42 DEBUG scheduler cleanup begin ForkIOScheduler.hs line 137 +2018-11-05T10:50:42 NOTICE cancelling processes: [] ForkIOScheduler.hs line 149 +2018-11-05T10:50:42 DEBUG scheduler cleanup done ForkIOScheduler.hs line 141 +``` + +## TODO + +### Stackage + +Still todo... + +[![extensible-effects-concurrent LTS](http://stackage.org/package/extensible-effects-concurrent/badge/lts)](http://stackage.org/lts/package/extensible-effects-concurrent) + +### Scheduler Variation + +The ambiguity-flexibility trade-off introduced by using extensible effects +kicks in because the `Process` type has a `Spawn` clause, which needs to +know the `Eff`ects. + +That is resolved by an omnipresent scheduler proxy parameter. + +I will resolve this issue in one of these ways, but haven't decided: + +- By using backpack - best option, apart from missing stack support + +- By duplicating the code for each scheduler implementation + +- By using implicit parameters (experimental use of that technique is in + the logging part) - problem is that implicit parameter sometimes act weired + and also might break compiler inlineing. + +### Other + +- Process Linking/Monitoring + +- Scheduler `ekg` Monitoring + +- Timers and Timeouts (e.g. in `receive`) + +- Rename stuff diff --git a/examples/example-1/Main.hs b/examples/example-1/Main.hs --- a/examples/example-1/Main.hs +++ b/examples/example-1/Main.hs @@ -38,10 +38,8 @@ example :: ( HasCallStack , SetMember Process (Process q) r - , Member (Logs LogMessage) r - , Member (Logs LogMessage) q - , SetMember Lift (Lift IO) q - , SetMember Lift (Lift IO) r + , HasLoggingIO r + , HasLoggingIO q ) => SchedulerProxy q -> Eff r () @@ -75,15 +73,11 @@ testServerLoop :: forall r q - . ( HasCallStack - , Member (Logs LogMessage) q - , SetMember Lift (Lift IO) q - , SetMember Process (Process q) r - ) + . (HasCallStack, SetMember Process (Process q) r, HasLoggingIO q) => SchedulerProxy q -> Eff r (Server TestApi) testServerLoop px = spawnServer px - $ ApiHandler (Just handleCastTest) (Just handleCallTest) Nothing -- (Just handleTerminateTest) + $ apiHandler handleCastTest handleCallTest handleTerminateTest where handleCastTest :: Api TestApi 'Asynchronous -> Eff (Process q ': q) ApiServerCmd @@ -143,7 +137,7 @@ logInfo (show me ++ " exiting with error: " ++ msg) void (reply ()) exitWithError px msg - -- handleTerminateTest msg = do - -- me <- self px - -- logInfo (show me ++ " is exiting: " ++ show msg) - -- maybe (exitNormally px) (exitWithError px) msg + handleTerminateTest msg = do + me <- self px + logInfo (show me ++ " is exiting: " ++ show msg) + maybe (exitNormally px) (exitWithError px) msg diff --git a/examples/example-2/Main.hs b/examples/example-2/Main.hs --- a/examples/example-2/Main.hs +++ b/examples/example-2/Main.hs @@ -3,10 +3,6 @@ import Data.Dynamic import Control.Eff -import qualified Control.Eff.Concurrent.Process.ForkIOScheduler - as ForkIO -import qualified Control.Eff.Concurrent.Process.SingleThreadedScheduler - as Pure import Control.Eff.Concurrent import Control.Eff.State.Strict import Control.Eff.Lift @@ -14,8 +10,8 @@ main :: IO () main = do - ForkIO.defaultMain (counterExample SchedulerProxy) - Pure.defaultMain (counterExample SchedulerProxy) + defaultMain (void (counterExample SchedulerProxy)) + print (schedulePure (counterExample SchedulerProxy)) -- * First API @@ -37,7 +33,7 @@ forgetObserverMessage = UnobserveCounter logCounterObservations - :: (SetMember Process (Process q) r, Member (Logs LogMessage) q, Lifted IO q) + :: (SetMember Process (Process q) r, HasLogging m q) => SchedulerProxy q -> Eff r (Server (CallbackObserver Counter)) logCounterObservations px = spawnCallbackObserver @@ -49,12 +45,13 @@ ) counterHandler - :: forall r q + :: forall r q m . ( Member (State (Observers Counter)) r , Member (State Integer) r - , Member (Logs LogMessage) r + , HasLogging m q + , HasLogging m r , SetMember Process (Process q) r - , Lifted IO r + , Lifted m r ) => SchedulerProxy q -> ApiHandler Counter r @@ -90,11 +87,10 @@ deriving instance Show (Api PocketCalc x) pocketCalcHandler - :: forall r q + :: forall r q m . ( Member (State Integer) r - , Member (Logs LogMessage) r + , HasLogging m r , SetMember Process (Process q) r - , Lifted IO r ) => SchedulerProxy q -> ApiHandler PocketCalc r @@ -118,11 +114,8 @@ return HandleNextRequest serverLoop - :: forall r q - . ( Member (Logs LogMessage) r - , SetMember Process (Process q) r - , Lifted IO r - ) + :: forall r q m + . (HasLogging m r, HasLogging m q, SetMember Process (Process q) r) => SchedulerProxy q -> Eff r () serverLoop px = evalState @Integer @@ -130,22 +123,16 @@ (manageObservers @Counter (serve px (counterHandler px, pocketCalcHandler px)) ) - -- ** Counter client counterExample - :: ( SetMember Process (Process q) r - , Member (Logs LogMessage) q - , Member (Logs LogMessage) r - , Lifted IO q - , Lifted IO r - , q <:: r - ) + :: (SetMember Process (Process q) r, HasLogging m r, HasLogging m q, q <:: r) => SchedulerProxy q - -> Eff r () -counterExample px = do + -> Eff r Integer +counterExample px = execState (0 :: Integer) $ do let cnt sv = do r <- call px sv Cnt logInfo (show sv ++ " " ++ show r) + modify (+ r) pid1 <- spawn (serverLoop px) pid2 <- spawn (serverLoop px) let cntServer1 = asServer @Counter pid1 diff --git a/examples/example-3/Main.hs b/examples/example-3/Main.hs new file mode 100644 --- /dev/null +++ b/examples/example-3/Main.hs @@ -0,0 +1,44 @@ +module Main where + +import Control.Concurrent +import Control.Eff.Lift +import Control.Eff.Log +import Control.Exception as IOException +import System.Directory +import System.FilePath +import System.IO + +main :: IO () +main = withAsyncLogChannel + 1000 + (ioLogMessageWriter + (fileAppender "extensible-effects-concurrent-example-3.log") + ) + (handleLoggingAndIO + (do + logInfo "test 1" + lift (threadDelay 1000000) + logDebug "test 2" + lift (threadDelay 1000000) + logCritical "test 3" + lift (threadDelay 1000000) + ) + ) + +fileAppender :: FilePath -> LogWriter String IO +fileAppender fnIn = multiMessageLogWriter + (\writeLogMessageWith -> bracket + (do + fnCanon <- canonicalizePath fnIn + createDirectoryIfMissing True (takeDirectory fnCanon) + h <- openFile fnCanon AppendMode + hSetBuffering h (BlockBuffering (Just 1024)) + return h + ) + (\h -> IOException.try @SomeException (hFlush h) >> hClose h) + (writeLogMessageWith . hPutStrLn) + ) + +fileAppenderSimple :: FilePath -> LogWriter String IO +fileAppenderSimple fnIn = + singleMessageLogWriter (\msg -> withFile fnIn AppendMode (`hPutStrLn` msg)) diff --git a/examples/example-4/Main.hs b/examples/example-4/Main.hs new file mode 100644 --- /dev/null +++ b/examples/example-4/Main.hs @@ -0,0 +1,34 @@ +module Main where + +import Control.Eff +import Control.Eff.Lift +import Control.Eff.Concurrent +import Data.Dynamic +import Control.Concurrent +import Control.DeepSeq + +main :: IO () +main = defaultMain + (do + lift (threadDelay 100000) -- because of async logging + firstExample forkIoScheduler + lift (threadDelay 100000) -- ... async logging + ) + -- The SchedulerProxy paremeter contains the effects of a specific scheduler + -- implementation. + +newtype WhoAreYou = WhoAreYou ProcessId deriving (Typeable, NFData) + +firstExample :: (HasLoggingIO q) => SchedulerProxy q -> Eff (Process q ': q) () +firstExample px = do + person <- spawn + (do + logInfo "I am waiting for someone to ask me..." + WhoAreYou replyPid <- receiveMessageAs px + sendMessageAs px replyPid "Alice" + logInfo (show replyPid ++ " just needed to know it.") + ) + me <- self px + sendMessageAs px person (WhoAreYou me) + personName <- receiveMessageAs px + logInfo ("I just met " ++ personName) 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.9.0 +version: 0.9.2 description: Please see the README on GitHub at synopsis: Message passing concurrency as extensible-effect homepage: https://github.com/sheyll/extensible-effects-concurrent#readme @@ -34,10 +34,9 @@ src build-depends: async >= 2.2 && <3, - base >=4.9 && <5, + base >=4.7 && <4.12, data-default, deepseq, - directory, enclosed-exceptions >= 1.0 && < 1.1, filepath, time, @@ -45,15 +44,11 @@ containers >=0.5.8 && <0.7, QuickCheck <2.12, lens, - logging-effect, - transformers, parallel, process, monad-control, - random, - extensible-effects >= 3, - stm >= 2.4.5 && <2.6, - tagged + extensible-effects >= 3.1 && <4, + stm >= 2.4.5 && <2.6 autogen-modules: Paths_extensible_effects_concurrent exposed-modules: Control.Eff.Loop, @@ -107,7 +102,7 @@ main-is: Main.hs hs-source-dirs: examples/example-1 build-depends: - base >=4.9 && <5 + base >=4.7 && <4.12 , data-default , extensible-effects-concurrent , extensible-effects >= 3 @@ -133,7 +128,7 @@ main-is: Main.hs hs-source-dirs: examples/example-2 build-depends: - base >=4.9 && <5 + base >=4.7 && <4.12 , data-default , extensible-effects-concurrent , extensible-effects >= 3 @@ -155,6 +150,59 @@ , TypeFamilies , TypeOperators +executable extensible-effects-concurrent-example-3 + main-is: Main.hs + hs-source-dirs: examples/example-3 + build-depends: + base >=4.7 && <4.12 + , data-default + , directory + , extensible-effects-concurrent + , extensible-effects >= 3 + , filepath + , lens + ghc-options: -Wall -threaded -fno-full-laziness -rtsopts -with-rtsopts=-N + default-language: Haskell2010 + default-extensions: BangPatterns + , DataKinds + , FlexibleContexts + , FlexibleInstances + , FunctionalDependencies + , GADTs + , GeneralizedNewtypeDeriving + , RankNTypes + , ScopedTypeVariables + , StandaloneDeriving + , TemplateHaskell + , TypeApplications + , TypeFamilies + , TypeOperators + +executable extensible-effects-concurrent-example-4 + main-is: Main.hs + hs-source-dirs: examples/example-4 + build-depends: + base >=4.7 && <4.12 + , extensible-effects-concurrent + , extensible-effects >= 3 + , deepseq + ghc-options: -Wall -threaded -fno-full-laziness -rtsopts -with-rtsopts=-N + default-language: Haskell2010 + default-extensions: BangPatterns + , DataKinds + , FlexibleContexts + , FlexibleInstances + , FunctionalDependencies + , GADTs + , GeneralizedNewtypeDeriving + , RankNTypes + , ScopedTypeVariables + , StandaloneDeriving + , TemplateHaskell + , TypeApplications + , TypeFamilies + , TypeOperators + test-suite extensible-effects-concurrent-test type: exitcode-stdio-1.0 main-is: Driver.hs @@ -170,7 +218,7 @@ , SingleThreadedScheduler ghc-options: -Wall -threaded -fno-full-laziness -rtsopts -with-rtsopts=-N build-depends: - base >=4.9 && <5 + base >=4.7 && <4.12 , data-default , extensible-effects-concurrent , extensible-effects >= 3 diff --git a/src/Control/Eff/Concurrent.hs b/src/Control/Eff/Concurrent.hs --- a/src/Control/Eff/Concurrent.hs +++ b/src/Control/Eff/Concurrent.hs @@ -42,7 +42,20 @@ , ConsProcess , ResumeProcess(..) , SchedulerProxy(..) - , MessageSelector(..) + , MessageSelector + ( runMessageSelector + ) + , selectMessage + , selectMessageLazy + , selectMessageProxy + , selectMessageProxyLazy + , filterMessage + , filterMessageLazy + , selectMessageWith + , selectMessageWithLazy + , selectDynamicMessage + , selectDynamicMessageLazy + , selectAnyMessageLazy , ProcessState(..) , ProcessExitReason(..) , ShutdownRequest(..) @@ -57,7 +70,7 @@ , spawn_ , receiveMessage , receiveMessageAs - , receiveMessageSuchThat + , receiveSelectedMessage , receiveLoop , receiveLoopAs , receiveLoopSuchThat 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 @@ -72,16 +72,11 @@ newtype Server api = Server { _fromServer :: ProcessId } deriving (Eq,Ord,Typeable) -instance Read (Server api) where - readsPrec _ ('[':'#':rest1) = - case reads (dropWhile (/= '#') rest1) of - [(c, ']':rest2)] -> [(Server c, rest2)] - _ -> [] - readsPrec _ _ = [] instance Typeable api => Show (Server api) where - show s@(Server c) = - "[#" ++ show (typeRep s) ++ "#" ++ show c ++ "]" + showsPrec d s@(Server c) = + showParen (d >= 10) + (showsPrec 11 (typeRep s) . showsPrec 11 c) makeLenses ''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 @@ -23,11 +23,11 @@ import Control.Eff.Concurrent.Api import Control.Eff.Concurrent.Api.Internal import Control.Eff.Concurrent.Process -import Control.Monad ( (>=>) ) import Data.Dynamic import Data.Typeable ( Typeable , typeRep ) +import Control.DeepSeq import GHC.Stack -- | Send an 'Api' request that has no return value and return as fast as @@ -78,6 +78,7 @@ , Typeable (Api api ( 'Synchronous result)) , Typeable result , HasCallStack + , NFData result ) => SchedulerProxy q -> Server api @@ -95,8 +96,8 @@ let extractResult :: Response api result -> Maybe result extractResult (Response _pxResult callRefMsg result) = if callRefMsg == callRef then Just result else Nothing - in MessageSelector (fromDynamic >=> extractResult) - in receiveMessageSuchThat px selectResult + in selectMessageWith extractResult + in receiveSelectedMessage px selectResult else raiseError px ( "failed to send a call for: '" @@ -135,7 +136,7 @@ -- | Like 'call' but take the 'Server' from the reader provided by -- 'registerServer'. callRegistered - :: (Typeable reply, ServesApi o r q, HasCallStack) + :: (Typeable reply, ServesApi o r q, HasCallStack, NFData reply) => SchedulerProxy q -> Api o ( 'Synchronous reply) -> Eff r reply @@ -154,6 +155,7 @@ , Typeable reply , ServesApi o r q , HasCallStack + , NFData (f reply) ) => SchedulerProxy q -> Api o ( 'Synchronous (f reply)) 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 @@ -36,7 +36,6 @@ import Control.Eff.Concurrent.Api.Client import Control.Eff.Concurrent.Api.Server import Control.Eff.Log -import Control.Eff.Lift import Control.Eff.State.Strict import Control.Lens @@ -180,12 +179,12 @@ -- | Start a new process for an 'Observer' that schedules -- all observations to an effectful callback. spawnCallbackObserver - :: forall o r q + :: forall o r q logWriter . ( SetMember Process (Process q) r , Typeable o , Show (Observation o) , Observable o - , Member (Logs LogMessage) q + , HasLogging logWriter q , HasCallStack ) => SchedulerProxy q @@ -206,13 +205,12 @@ -- -- @since 0.3.0.0 spawnLoggingObserver - :: forall o r q + :: forall o r q logWriter . ( SetMember Process (Process q) r , Typeable o , Show (Observation o) , Observable o - , Member (Logs LogMessage) q - , Lifted IO q + , HasLogging logWriter q , HasCallStack ) => SchedulerProxy q diff --git a/src/Control/Eff/Concurrent/Api/Observer/Queue.hs b/src/Control/Eff/Concurrent/Api/Observer/Queue.hs --- a/src/Control/Eff/Concurrent/Api/Observer/Queue.hs +++ b/src/Control/Eff/Concurrent/Api/Observer/Queue.hs @@ -35,7 +35,7 @@ type ObservationQueueReader a = Reader (ObservationQueue a) logPrefix :: forall o proxy . (HasCallStack, Typeable o) => proxy o -> String -logPrefix px = "" +logPrefix px = "observation queue: " ++ show (typeRep px) -- | Read queued observations captured by observing a 'Server' that implements -- an 'Observable' 'Api' using 'enqueueObservationsRegistered' or 'enqueueObservations'. @@ -47,7 +47,7 @@ , HasCallStack , MonadIO (Eff r) , Typeable o - , Member (Logs LogMessage) r + , HasLoggingIO r ) => Eff r (Observation o) readObservationQueue = do @@ -65,7 +65,7 @@ , HasCallStack , MonadIO (Eff r) , Typeable o - , Member (Logs LogMessage) r + , HasLoggingIO r ) => Eff r (Maybe (Observation o)) tryReadObservationQueue = do @@ -82,7 +82,7 @@ , HasCallStack , MonadIO (Eff r) , Typeable o - , Member (Logs LogMessage) r + , HasLoggingIO r ) => Eff r [Observation o] flushObservationQueue = do @@ -99,11 +99,9 @@ , Typeable o , Show (Observation o) , Observable o - , Member (Logs LogMessage) q - , MonadIO (Eff (Process q ': q)) - , MonadIO (Eff r) - , SetMember Lift (Lift IO) r - , Member (Logs LogMessage) r + , HasLoggingIO q + , HasLoggingIO r + , Lifted IO r , HasCallStack ) => SchedulerProxy q @@ -126,11 +124,9 @@ , Typeable o , Show (Observation o) , Observable o - , Member (Logs LogMessage) q - , MonadIO (Eff (Process q ': q)) - , MonadIO (Eff r) - , SetMember Lift (Lift IO) r - , Member (Logs LogMessage) r + , HasLoggingIO r + , HasLoggingIO q + , Lifted IO q , HasCallStack ) => SchedulerProxy q @@ -180,13 +176,7 @@ withQueue :: forall a b e - . ( HasCallStack - , MonadIO (Eff e) - , Member (Logs LogMessage) e - , Typeable a - , Show (Observation a) - , SetMember Lift (Lift IO) e - ) + . (HasCallStack, Typeable a, Show (Observation a), HasLoggingIO e) => Int -> Eff (ObservationQueueReader a ': e) b -> Eff e b 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 @@ -44,6 +44,8 @@ import Data.Kind import GHC.Stack import Data.Maybe +import GHC.Generics +import Control.DeepSeq import Data.Default -- | A record of callbacks, handling requests sent to a /server/ 'Process', all @@ -174,8 +176,9 @@ -- optional reason. StopApiServer :: Maybe String -> ApiServerCmd -- SendReply :: reply -> ApiServerCmd () -> ApiServerCmd (reply -> Eff eff ()) - deriving (Show, Typeable) + deriving (Show, Typeable, Generic) +instance NFData ApiServerCmd makeLenses ''ApiHandler @@ -202,7 +205,7 @@ instance Semigroup (ServerCallback eff) where l <> r = l & requestHandlerSelector .~ - MessageSelector (\x -> + selectDynamicMessageLazy (\x -> runMessageSelector (view requestHandlerSelector l) x <|> runMessageSelector (view requestHandlerSelector r) x) & terminationHandler .~ @@ -213,7 +216,7 @@ instance Monoid (ServerCallback eff) where mappend = (<>) mempty = ServerCallback - { _requestHandlerSelector = MessageSelector (const Nothing) + { _requestHandlerSelector = selectDynamicMessageLazy (const Nothing) , _terminationHandler = const (return ()) } @@ -335,7 +338,7 @@ -> ApiHandler api eff -> MessageSelector (Eff eff ApiServerCmd) selectHandlerMethod px handlers = - MessageSelector (fmap (applyHandlerMethod px handlers) . fromDynamic) + selectDynamicMessageLazy (fmap (applyHandlerMethod px handlers) . fromDynamic) -- | Apply either the '_callCallback', '_castCallback' or the '_terminateCallback' -- callback to an incoming request. @@ -392,5 +395,4 @@ => SchedulerProxy q -> Maybe String -> Eff r () -defaultTermination px = - maybe (return ()) (exitWithError px) +defaultTermination px = maybe (return ()) (exitWithError px) diff --git a/src/Control/Eff/Concurrent/Process.hs b/src/Control/Eff/Concurrent/Process.hs --- a/src/Control/Eff/Concurrent/Process.hs +++ b/src/Control/Eff/Concurrent/Process.hs @@ -19,7 +19,18 @@ , ConsProcess , ResumeProcess(..) , SchedulerProxy(..) - , MessageSelector(..) + , MessageSelector(runMessageSelector) + , selectMessage + , selectMessageLazy + , selectMessageProxy + , selectMessageProxyLazy + , filterMessage + , filterMessageLazy + , selectMessageWith + , selectMessageWithLazy + , selectDynamicMessage + , selectDynamicMessageLazy + , selectAnyMessageLazy , ProcessState(..) , ProcessExitReason(..) , ShutdownRequest(..) @@ -34,7 +45,7 @@ , spawn_ , receiveMessage , receiveMessageAs - , receiveMessageSuchThat + , receiveSelectedMessage , receiveLoop , receiveLoopAs , receiveLoopSuchThat @@ -60,8 +71,9 @@ import Control.Eff.Log.Handler import Control.Eff.Log.Message import Control.Lens -import Control.Monad ( void ) -import Control.Monad.IO.Class +import Control.Monad ( void + , (>=>) + ) import Data.Default import Data.Dynamic import Data.Kind @@ -115,12 +127,11 @@ -- destination process does not exist, or does not accept messages of the -- given type. SendMessage :: ProcessId -> Dynamic -> Process r (ResumeProcess Bool) - -- | Receive a message. This should block until an a message was received. The - -- message is returned as a 'ProcessMessage' value. The function should also - -- return if an exception was caught or a shutdown was requested. - ReceiveMessage :: Process r (ResumeProcess Dynamic) - -- | Wait for the next message that matches a criterium. Similar to 'ReceiveMessage'. - ReceiveMessageSuchThat :: MessageSelector a -> Process r (ResumeProcess a) + -- | Receive a message that matches a criterium. + -- This should block until an a message was received. The message is returned + -- as a 'ProcessMessage' value. The function should also return if an exception + -- was caught or a shutdown was requested. + ReceiveSelectedMessage :: forall r a . MessageSelector a -> Process r (ResumeProcess a) -- | Generate a unique 'Int' for the current process. MakeReference :: Process r (ResumeProcess Int) @@ -147,8 +158,7 @@ . showChar ' ' . showsPrec 10 sr ) - ReceiveMessage -> showString "ReceiveMessage" - ReceiveMessageSuchThat _ -> showString "ReceiveMessageSuchThat" + ReceiveSelectedMessage _ -> showString "ReceiveSelectedMessage" MakeReference -> showString "MakeReference" -- | Every 'Process' action returns it's actual result wrapped in this type. It @@ -172,12 +182,99 @@ instance NFData1 ResumeProcess -- | A function that deciced if the next message will be received by --- 'ReceiveMessageSuchThat'. It conveniently is an instance of 'Monoid' +-- 'ReceiveSelectedMessage'. It conveniently is an instance of 'Monoid' -- with first come, first serve bais. newtype MessageSelector a = MessageSelector {runMessageSelector :: Dynamic -> Maybe a } deriving (Semigroup, Monoid, Functor) + +-- | Create a message selector for a value that can be obtained by 'fromDynamic'. +-- It will also 'force' the result. +-- +-- @since 0.9.1 +selectMessage :: (NFData t, Typeable t) => MessageSelector t +selectMessage = selectDynamicMessage fromDynamic + +-- | Create a message selector for a value that can be obtained by 'fromDynamic'. +-- It will also 'force' the result. +-- +-- @since 0.9.1 +selectMessageLazy :: Typeable t => MessageSelector t +selectMessageLazy = selectDynamicMessageLazy fromDynamic + +-- | Create a message selector from a predicate. It will 'force' the result. +-- +-- @since 0.9.1 +filterMessage :: (Typeable a, NFData a) => (a -> Bool) -> MessageSelector a +filterMessage predicate = selectDynamicMessage + (\d -> case fromDynamic d of + Just a | predicate a -> Just a + _ -> Nothing + ) + +-- | Create a message selector from a predicate. It will 'force' the result. +-- +-- @since 0.9.1 +filterMessageLazy :: Typeable a => (a -> Bool) -> MessageSelector a +filterMessageLazy predicate = selectDynamicMessageLazy + (\d -> case fromDynamic d of + Just a | predicate a -> Just a + _ -> Nothing + ) + +-- | Select a message of type @a@ and apply the given function to it. +-- If the function returns 'Just' The 'ReceiveSelectedMessage' function will +-- return the result (sans @Maybe@). It will 'force' the result. +-- +-- @since 0.9.1 +selectMessageWith + :: (Typeable a, NFData b) => (a -> Maybe b) -> MessageSelector b +selectMessageWith f = selectDynamicMessage (fromDynamic >=> f) + +-- | Select a message of type @a@ and apply the given function to it. +-- If the function returns 'Just' The 'ReceiveSelectedMessage' function will +-- return the result (sans @Maybe@). It will 'force' the result. +-- +-- @since 0.9.1 +selectMessageWithLazy :: Typeable a => (a -> Maybe b) -> MessageSelector b +selectMessageWithLazy f = selectDynamicMessageLazy (fromDynamic >=> f) + +-- | Create a message selector. It will 'force' the result. +-- +-- @since 0.9.1 +selectDynamicMessage :: NFData a => (Dynamic -> Maybe a) -> MessageSelector a +selectDynamicMessage = MessageSelector . (force .) + +-- | Create a message selector. +-- +-- @since 0.9.1 +selectDynamicMessageLazy :: (Dynamic -> Maybe a) -> MessageSelector a +selectDynamicMessageLazy = MessageSelector + +-- | Create a message selector that will match every message. This is /lazy/ +-- because the result is not 'force'ed. +-- +-- @since 0.9.1 +selectAnyMessageLazy :: MessageSelector Dynamic +selectAnyMessageLazy = MessageSelector Just + +-- | Create a message selector for a value that can be obtained by 'fromDynamic' +-- with a proxy argument. It will also 'force' the result. +-- +-- @since 0.9.1 +selectMessageProxy + :: forall proxy t . (NFData t, Typeable t) => proxy t -> MessageSelector t +selectMessageProxy _ = selectDynamicMessage fromDynamic + +-- | Create a message selector for a value that can be obtained by 'fromDynamic' +-- with a proxy argument. It will also 'force' the result. +-- +-- @since 0.9.1 +selectMessageProxyLazy + :: forall proxy t . (Typeable t) => proxy t -> MessageSelector t +selectMessageProxyLazy _ = selectDynamicMessageLazy fromDynamic + -- | Every function for 'Process' things needs such a proxy value -- for the low-level effect list, i.e. the effects identified by -- @__r__@ in @'Process' r : r@, this might be dependent on the @@ -376,32 +473,33 @@ . (HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> Eff r Dynamic -receiveMessage _ = executeAndResume ReceiveMessage +receiveMessage _ = + executeAndResume (ReceiveSelectedMessage selectAnyMessageLazy) -- | Block until a message was received, that is not 'Nothing' after applying -- a callback to it. --- See 'ReceiveMessageSuchThat' for more documentation. -receiveMessageSuchThat +-- See 'ReceiveSelectedMessage' for more documentation. +receiveSelectedMessage :: forall r q a . (HasCallStack, Typeable a, SetMember Process (Process q) r) => SchedulerProxy q -> MessageSelector a -> Eff r a -receiveMessageSuchThat _ f = executeAndResume (ReceiveMessageSuchThat f) +receiveSelectedMessage _ f = executeAndResume (ReceiveSelectedMessage f) -- | Receive and cast the message to some 'Typeable' instance. --- See 'ReceiveMessageSuchThat' for more documentation. --- This will wait for a message of the return type using 'receiveMessageSuchThat' +-- See 'ReceiveSelectedMessage' for more documentation. +-- This will wait for a message of the return type using 'receiveSelectedMessage' receiveMessageAs :: forall a r q . (HasCallStack, Typeable a, SetMember Process (Process q) r) => SchedulerProxy q -> Eff r a -receiveMessageAs px = receiveMessageSuchThat px (MessageSelector fromDynamic) +receiveMessageAs px = receiveSelectedMessage px (MessageSelector fromDynamic) -- | Enter a loop to receive messages and pass them to a callback, until the -- function returns 'Just' a result. --- See 'receiveMesage' or 'ReceiveMessage' for more documentation. +-- See 'selectAnyMessageLazy' or 'ReceiveSelectedMessage' for more documentation. receiveLoop :: forall r q endOfLoopResult . (SetMember Process (Process q) r, HasCallStack) @@ -409,7 +507,7 @@ -> (Either (Maybe String) Dynamic -> Eff r (Maybe endOfLoopResult)) -> Eff r endOfLoopResult receiveLoop px handlers = do - mReq <- send (ReceiveMessage @q) + mReq <- send (ReceiveSelectedMessage @q selectAnyMessageLazy) mRes <- case mReq of RetryLastAction -> return Nothing ShutdownRequested ExitNormally -> handlers (Left Nothing) @@ -428,7 +526,7 @@ -> (Either (Maybe String) a -> Eff r (Maybe endOfLoopResult)) -> Eff r endOfLoopResult receiveLoopAs px handlers = do - mReq <- send (ReceiveMessageSuchThat @a @q (MessageSelector fromDynamic)) + mReq <- send (ReceiveSelectedMessage @q @a (MessageSelector fromDynamic)) mRes <- case mReq of RetryLastAction -> return Nothing ShutdownRequested ExitNormally -> handlers (Left Nothing) @@ -448,7 +546,7 @@ -> (Either (Maybe String) a -> Eff r (Maybe endOfLoopResult)) -> Eff r endOfLoopResult receiveLoopSuchThat px selectMesage handlers = do - mReq <- send (ReceiveMessageSuchThat @a @q selectMesage) + mReq <- send (ReceiveSelectedMessage @q @a selectMesage) mRes <- case mReq of RetryLastAction -> return Nothing ShutdownRequested ExitNormally -> handlers (Left Nothing) @@ -541,7 +639,7 @@ -- | Log the 'ProcessExitReaons' logProcessExit - :: ('[Logs LogMessage] <:: e, MonadIO (Eff e), HasCallStack) + :: (HasCallStack, HasLogWriter LogMessage h e) => ProcessExitReason -> Eff e () logProcessExit ex = withFrozenCallStack $ case ex of @@ -550,5 +648,5 @@ ProcessShutDown (ExitWithError m) -> logError ("exit with error: " ++ show m) ProcessCaughtIOException w m -> logError ("runtime exception: " ++ m ++ " caught here: " ++ w) - ProcessRaisedError m -> logError ("unhandled process exception: " ++ show m) - _ -> logError ("scheduler error: " ++ show ex) + ProcessRaisedError m -> logError ("unhandled process exception: " ++ show m) + SchedulerShuttingDown -> logInfo "scheduler schutting down" 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 @@ -38,7 +38,6 @@ import Control.Eff.Lift import Control.Eff.Log import Control.Eff.Reader.Strict as Reader -import Control.Monad.Log ( runLoggingT ) import Control.Lens import Control.Monad ( when , void @@ -123,7 +122,7 @@ -- | Create a new 'SchedulerState' run an IO action, catching all exceptions, -- and when the actions returns, clean up and kill all processes. withNewSchedulerState - :: HasCallStack => Eff SchedulerIO () -> Eff LoggingAndIO () + :: (HasLoggingIO SchedulerIO, HasCallStack) => Eff SchedulerIO () -> Eff LoggingAndIO () withNewSchedulerState mainProcessAction = Exceptions.tryAny (lift (atomically newSchedulerState)) >>= either @@ -182,20 +181,22 @@ -- | Start the message passing concurrency system then execute a 'Process' on -- top of 'SchedulerIO' effect. All logging is sent to standard output. -defaultMain :: HasCallStack => Eff ProcEff () -> IO () -defaultMain c = runLoggingT - (logChannelBracket 128 - Nothing - (handleLoggingAndIO_ (schedule c)) - ) - (printLogMessage :: LogMessage -> IO ()) +defaultMain + :: HasCallStack => (HasLogWriterProxy IO => Eff ProcEff ()) -> IO () +defaultMain c = + withAsyncLogChannel 1024 + (multiMessageLogWriter ($ printLogMessage)) + (handleLoggingAndIO_ (schedule c)) -- | Start the message passing concurrency system then execute a 'Process' on -- top of 'SchedulerIO' effect. All logging is sent to standard output. defaultMainWithLogChannel - :: HasCallStack => LogChannel LogMessage -> Eff ProcEff () -> IO () -defaultMainWithLogChannel logC c = - closeLogChannelAfter logC (handleLoggingAndIO_ (schedule c) logC) + :: HasCallStack + => (HasLogWriterProxy IO => Eff ProcEff ()) + -> LogChannel LogMessage + -> IO () +defaultMainWithLogChannel c = + handleLoggingAndIO_ (schedule c) -- | A 'SchedulerProxy' for 'SchedulerIO' forkIoScheduler :: SchedulerProxy SchedulerIO @@ -221,7 +222,7 @@ -- ** MessagePassing execution handleProcess - :: HasCallStack + :: (HasLoggingIO SchedulerIO, HasCallStack) => ProcessInfo -> Eff ProcEff ProcessExitReason -> Eff SchedulerIO ProcessExitReason @@ -317,8 +318,7 @@ then diskontinue (ProcessShutDown r) else kontinue (ShutdownRequested shutdownRequest) Spawn _ -> kontinue (ShutdownRequested shutdownRequest) - ReceiveMessage -> kontinue (ShutdownRequested shutdownRequest) - ReceiveMessageSuchThat _ -> kontinue (ShutdownRequested shutdownRequest) + ReceiveSelectedMessage _ -> kontinue (ShutdownRequested shutdownRequest) SelfPid -> kontinue (ShutdownRequested shutdownRequest) MakeReference -> kontinue (ShutdownRequested shutdownRequest) YieldProcess -> kontinue (ShutdownRequested shutdownRequest) @@ -341,8 +341,7 @@ then kontinue nextRef (ShutdownRequested msg) else interpretSendShutdown toPid msg >>= kontinue nextRef . ResumeWith Spawn child -> spawnNewProcess child >>= kontinue nextRef . ResumeWith . fst - ReceiveMessage -> interpretReceive (MessageSelector Just) >>= kontinue nextRef - ReceiveMessageSuchThat f -> interpretReceive f >>= kontinue nextRef + ReceiveSelectedMessage f -> interpretReceive f >>= kontinue nextRef SelfPid -> kontinue nextRef (ResumeWith myPid) MakeReference -> kontinue (nextRef + 1) (ResumeWith nextRef) YieldProcess -> kontinue nextRef (ResumeWith ()) @@ -384,7 +383,10 @@ -- | This is the main entry point to running a message passing concurrency -- application. This function takes a 'Process' on top of the 'SchedulerIO' -- effect and a 'LogChannel' for concurrent logging. -schedule :: HasCallStack => Eff ProcEff () -> Eff LoggingAndIO () +schedule + :: (HasLoggingIO SchedulerIO, HasCallStack) + => Eff ProcEff () + -> Eff LoggingAndIO () schedule procEff = liftBaseWith (\runS -> Async.withAsync (runS $ withNewSchedulerState $ do @@ -400,7 +402,10 @@ ) ) >>= restoreM -spawnNewProcess :: Eff ProcEff () -> Eff SchedulerIO (ProcessId, Async ProcessExitReason) +spawnNewProcess + :: (HasLoggingIO SchedulerIO, HasCallStack) + => Eff ProcEff () + -> Eff SchedulerIO (ProcessId, Async ProcessExitReason) spawnNewProcess mfa = do schedulerState <- getSchedulerState liftBaseWith diff --git a/src/Control/Eff/Concurrent/Process/Interactive.hs b/src/Control/Eff/Concurrent/Process/Interactive.hs --- a/src/Control/Eff/Concurrent/Process/Interactive.hs +++ b/src/Control/Eff/Concurrent/Process/Interactive.hs @@ -45,6 +45,7 @@ import Control.Monad import Data.Foldable import Data.Typeable ( Typeable ) +import Control.DeepSeq import System.Timeout -- | Contains the communication channels to interact with a scheduler running in @@ -140,7 +141,7 @@ -- | Combination of 'submit' and 'cast'. submitCall :: forall o q r - . (SetMember Lift (Lift IO) r, Typeable o, Typeable q) + . (SetMember Lift (Lift IO) r, Typeable o, Typeable q, NFData q) => SchedulerSession r -> Server o -> Api o ( 'Synchronous q) diff --git a/src/Control/Eff/Concurrent/Process/SingleThreadedScheduler.hs b/src/Control/Eff/Concurrent/Process/SingleThreadedScheduler.hs --- a/src/Control/Eff/Concurrent/Process/SingleThreadedScheduler.hs +++ b/src/Control/Eff/Concurrent/Process/SingleThreadedScheduler.hs @@ -36,8 +36,16 @@ -- @schedulePure == runIdentity . 'scheduleM' (Identity . run) (return ())@ -- -- @since 0.3.0.2 -schedulePure :: Eff (ConsProcess '[]) a -> Either String a -schedulePure = runIdentity . scheduleM (Identity . run) (return ()) +schedulePure + :: ( HasLogWriterProxy Identity + => Eff (ConsProcess '[LogsM LogMessage Identity, Lift Identity]) a + ) + -> Either String a +schedulePure e = runIdentity + (scheduleM (runLift . handleLogs traceLogMessageWriter) + (return ()) + (usingLogWriterProxy (LogWriterProxy @Identity) e) + ) -- | Invoke 'schedule' with @lift 'Control.Concurrent.yield'@ as yield effect. -- @scheduleIO runEff == 'scheduleM' (runLift . runEff) (liftIO 'yield')@ @@ -69,10 +77,10 @@ -- @since 0.4.0.0 scheduleIOWithLogging :: (NFData l) - => (l -> IO ()) + => LogWriter l IO -> Eff (ConsProcess '[Logs l, Lift IO]) a -> IO (Either String a) -scheduleIOWithLogging handleLog = scheduleIO (handleLogsWith handleLog) +scheduleIOWithLogging h = scheduleIO (handleLogs h) -- | Handle the 'Process' effect, as well as all lower effects using an effect handler function. -- @@ -91,14 +99,16 @@ -- -- @since 0.4.0.0 scheduleM - :: Monad m + :: forall m r a + . Monad m => (forall b . Eff r b -> m b) -> m () -- ^ An that performs a __yield__ w.r.t. the underlying effect -- @r@. E.g. if @Lift IO@ is present, this might be: -- @lift 'Control.Concurrent.yield'. - -> Eff (ConsProcess r) a + -> (HasLogWriterProxy m => Eff (ConsProcess r) a) -> m (Either String a) -scheduleM runEff yieldEff e = do +scheduleM runEff yieldEff e' = do + let e = (usingLogWriterProxy (LogWriterProxy @m) e') y <- runAsCoroutinePure runEff e handleProcess runEff yieldEff @@ -247,7 +257,7 @@ (msgQs & at newPid ?~ Seq.empty) (rest :|> (nextK, pid) :|> (fk, newPid)) - recv@(OnRecv selectMessage k) -> case msgQs ^. at pid of + recv@(OnRecv messageSelector k) -> case msgQs ^. at pid of Nothing -> do nextK <- runEff $ k (OnError (show pid ++ " has no message queue!")) handleProcess runEff @@ -277,7 +287,7 @@ partitionMessages (m :<| msgRest) acc = maybe (partitionMessages msgRest (acc :|> m)) (\res -> Just (res, acc Seq.>< msgRest)) - (runMessageSelector selectMessage m) + (runMessageSelector messageSelector m) in case partitionMessages messages Empty of Nothing -> handleProcess runEff yieldEff @@ -309,8 +319,7 @@ cont (Shutdown !sr ) _k = return (OnShutdown sr) cont (RaiseError !e ) _k = return (OnRaiseError e) cont (SendMessage !tp !msg) k = return (OnSend tp msg k) - cont ReceiveMessage k = return (OnRecv (MessageSelector Just) k) - cont (ReceiveMessageSuchThat f) k = return (OnRecv f k) + cont (ReceiveSelectedMessage f) k = return (OnRecv f k) cont (SendShutdown !pid !sr ) k = return (OnSendShutdown pid sr k) cont MakeReference k = return (OnMakeReference k) @@ -334,5 +343,5 @@ defaultMain = void . runLift - . handleLogsWithLoggingTHandler ($ printLogMessage) + . handleLogs (multiMessageLogWriter ($ printLogMessage)) . scheduleMonadIOEff diff --git a/src/Control/Eff/Log/Channel.hs b/src/Control/Eff/Log/Channel.hs --- a/src/Control/Eff/Log/Channel.hs +++ b/src/Control/Eff/Log/Channel.hs @@ -1,207 +1,102 @@ -- | Concurrent Logging module Control.Eff.Log.Channel ( LogChannel() - , logToChannel - , nullLogChannel - , forkLogger - , filterLogChannel - , joinLogChannel - , killLogChannel - , closeLogChannelAfter - , logChannelBracket - , logChannelPutIO - , JoinLogChannelException() - , KillLogChannelException() + , withAsyncLogChannel , handleLoggingAndIO , handleLoggingAndIO_ ) where -import Control.Concurrent +import Control.Concurrent.Async import Control.Concurrent.STM import Control.Eff as Eff import Control.Eff.Lift -import Control.Exception ( bracket ) -import qualified Control.Exception as Exc +import Control.Exception ( evaluate ) import Control.Monad ( void - , when , unless ) -import Control.Monad.Log as ExtLog - hiding ( ) -import Control.Monad.Trans.Control import Control.Eff.Log.Handler -import qualified Control.Eff.Lift as Eff import Data.Foldable ( traverse_ ) import Data.Kind ( ) -import Data.String -import Data.Typeable +import Control.DeepSeq import GHC.Stack --- | 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 = - 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'. -logToChannel - :: forall r message a - . (SetMember Eff.Lift (Eff.Lift IO) r) - => LogChannel message - -> Eff (Logs message ': r) a - -> Eff r a -logToChannel logChan = - handleLogsWithLoggingTHandler ($ logChannelPutIO logChan) - --- | Enqueue a log message into a log channel -logChannelPutIO :: LogChannel message -> message -> IO () -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) - --- | Create a 'LogChannel' that will discard all messages sent --- via 'forwardLogstochannel' or 'logChannelPutIO'. -nullLogChannel :: LogChannel message -nullLogChannel = DiscardLogs - --- | Fork a new process, that applies a monadic action to all log messages sent --- via 'logToChannel' or 'logChannelPutIO'. -forkLogger - :: forall message - . (Typeable message) +-- | Fork a new process in which the given log message writer, will listen +-- on a message queue in a 'LogChannel', which is passed to the second function. +-- If the function returns or throws, the logging process will be killed. +-- +-- Log messages are deeply evaluated before being sent to the logger process, +-- to prevent that lazy evaluation leads to heavy work being done in the +-- logger process instead of the caller process. +-- +-- Example usage, a super stupid log to file: +-- +-- > +-- > main = +-- > withAsyncLogChannel +-- > 1000 +-- > (singleMessageLogWriter putStrLn) +-- > (handleLoggingAndIO +-- > (do logMsg "test 1" +-- > logMsg "test 2" +-- > logMsg "test 3")) +-- > +-- +withAsyncLogChannel + :: forall message a + . (NFData 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) -forkLogger queueLen handle mFirstMsg = do - msgQ <- atomically - (do - tq <- newTBQueue (fromIntegral @Int queueLen) - mapM_ (writeTBQueue tq) mFirstMsg - return tq - ) - thread <- forkFinally (logLoop msgQ) (writeLastLogs msgQ) - return (ConcurrentLogChannel msgQ thread) + -> LogWriter message IO -- ^ An IO action to write the log messages + -> (LogChannel message -> IO a) + -> IO a +withAsyncLogChannel queueLen ioWriter action = do + msgQ <- newTBQueueIO (fromIntegral @Int queueLen) + withAsync (logLoop msgQ) (action . ConcurrentLogChannel msgQ) where - 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 -> traverse_ handle logMessages - Nothing -> case Exc.fromException se of - Just KillLogChannelException -> return () - Nothing -> mapM_ handle logMessages - - logLoop :: TBQueue message -> IO () logLoop tq = do - m <- atomically $ do + ms <- atomically $ do h <- readTBQueue tq t <- flushTBQueue tq return (h : t) - traverse_ handle m + writeAllLogMessages ioWriter ms 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 'nullLogChannel', '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 - :: (Typeable message, IsString message) => LogChannel message -> IO a -> IO a -closeLogChannelAfter logC ioAction = do - res <- closeLogAndRethrow `Exc.handle` ioAction - closeLogSuccess - return res - where - closeLogAndRethrow :: Exc.SomeException -> IO a - closeLogAndRethrow se = do - void $ Exc.try @Exc.SomeException $ killLogChannel logC - Exc.throw se - - closeLogSuccess :: IO () - closeLogSuccess = joinLogChannel logC - --- | Close a log channel created by e.g. 'forkLogger'. Message already enqueue --- are handled. Subsequent log message --- will not be handled anymore. If the log channel must be closed immediately, --- use 'killLogChannel' instead. -joinLogChannel :: (Typeable message) => LogChannel message -> IO () -joinLogChannel DiscardLogs = return () -joinLogChannel (FilteredLogChannel _f lc) = joinLogChannel lc -joinLogChannel (ConcurrentLogChannel _tq thread) = - throwTo thread JoinLogChannelException - --- | 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 :: (Typeable message) => LogChannel message -> IO () -killLogChannel DiscardLogs = return () -killLogChannel (FilteredLogChannel _f lc) = killLogChannel lc -killLogChannel (ConcurrentLogChannel _tq thread) = - throwTo thread KillLogChannelException - --- | 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. -data JoinLogChannelException = JoinLogChannelException - deriving (Show, Typeable) - -instance Exc.Exception JoinLogChannelException - --- | 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. -data KillLogChannelException = KillLogChannelException - deriving (Show, Typeable) - -instance Exc.Exception KillLogChannelException - --- | Wrap 'LogChannel' creation and destruction around a monad action in --- 'bracket'y manner. This function uses 'joinLogChannel', so en-queued messages --- are flushed on exit. The resulting action is a 'LoggingT' action, which --- is essentially a reader for a log handler function in 'IO'. -logChannelBracket - :: (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 - -> (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 queueLen mWelcome f = control - (\runInIO -> do - let logHandler = void . runInIO . logMessage - bracket (forkLogger queueLen logHandler mWelcome) joinLogChannel f - ) - --- | Handle the 'LoggingAndIO' effects, using a 'LogChannel' for the 'Logs' --- effect. +-- | Fork an IO based log writer thread and set the 'LogWriter' to an action +-- that will send all logs to that thread via a bounded queue. +-- When the queue is full, flush it handleLoggingAndIO - :: HasCallStack => Eff '[Logs m, Lift IO] a -> LogChannel m -> IO a -handleLoggingAndIO e lc = runLift (logToChannel lc e) + :: (NFData m, HasCallStack) + => (HasLogWriterProxy IO => Eff '[Logs m, Lift IO] a) + -> LogChannel m + -> IO a +handleLoggingAndIO e lc = runLift + (handleLogs (foldingLogWriter (traverse_ logChannelPutIO)) e) + where + logQ = fromLogChannel lc + logChannelPutIO (force -> me) = do + !m <- evaluate me + atomically + (do + dropMessage <- isFullTBQueue logQ + unless dropMessage (writeTBQueue logQ m) + ) -- | Like 'handleLoggingAndIO' but return @()@. handleLoggingAndIO_ - :: HasCallStack => Eff '[Logs m, Lift IO] a -> LogChannel m -> IO () -handleLoggingAndIO_ = ((.) . (.)) void handleLoggingAndIO + :: (NFData m, HasCallStack) + => (HasLogWriterProxy IO => Eff '[Logs m, Lift IO] a) + -> LogChannel m + -> IO () +handleLoggingAndIO_ e lc = void (handleLoggingAndIO e lc) + +-- | 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 = + ConcurrentLogChannel + { fromLogChannel :: TBQueue message + , _logChannelThread :: Async () + } + -- ^ send all log messages to a log process diff --git a/src/Control/Eff/Log/Handler.hs b/src/Control/Eff/Log/Handler.hs --- a/src/Control/Eff/Log/Handler.hs +++ b/src/Control/Eff/Log/Handler.hs @@ -1,47 +1,155 @@ +{-# LANGUAGE ImplicitParams #-} + -- | A logging effect based on 'Control.Monad.Log.MonadLog'. module Control.Eff.Log.Handler - ( - -- * Logging Effect - Logs - , LogWriter(..) - , askLogWriter - , logMsg - , interceptLogging - , foldLogMessages + ( logMsg + , logMsgs + , LogsM + , Logs + , handleLogs , ignoreLogs - , handleLogsWith - , handleLogsWithLoggingTHandler + , mapLogMessages + , filterLogMessages + , interceptLogging + , HasLogWriter + , HasLogWriterIO + , LogWriter() + , writeAllLogMessages + , foldingLogWriter + , singleMessageLogWriter + , multiMessageLogWriter + , HasLogWriterProxy + , LogWriterProxy(..) + , usingLogWriterProxy + , logWriterProxy + , askLogWriter ) where import Control.DeepSeq import Control.Eff as Eff import Control.Eff.Reader.Strict -import qualified Control.Eff.Lift as Eff -import qualified Control.Monad.Log as Log -import Control.Monad.IO.Class -import Data.Foldable ( traverse_ ) -import Data.Kind ( ) +import Control.Eff.Lift as Eff +import Data.Foldable ( traverse_, toList ) +import Data.Kind +import Data.Functor.Identity +import GHC.Stack +import Data.Default -- | A function that takes a log message and returns an effect that -- /logs/ the message. -newtype LogWriter message = LogWriter { runLogWriter :: message -> IO () } +newtype LogWriter message writerM = + LogWriter + { runLogWriter + :: forall f. (HasCallStack, Traversable f, Foldable f, Functor f) + => f message + -> writerM () + } +instance Applicative w => Default (LogWriter m w) where + def = LogWriter (const (pure ())) + +-- | Efficiently apply the 'LogWriter' to a 'Foldable' container of log +-- messages. +writeAllLogMessages + :: (Foldable f, Functor f, Traversable f, Applicative writerM, HasCallStack) + => LogWriter message writerM -> f message -> writerM () +writeAllLogMessages = runLogWriter + +-- | Create a 'LogWriter' from a function that can write +-- a 'Foldable' container. +foldingLogWriter + :: (forall f. (Foldable f, Traversable f, HasCallStack) => f message -> writerM ()) + -> LogWriter message writerM +foldingLogWriter = LogWriter + +-- | Create a 'LogWriter' from a function that is applied to each +-- individual log message. NOTE: This is probably the simplest, +-- but also the most inefficient and annoying way to make +-- a 'LogWriter'. Better use 'foldingLogWriter' or even +-- 'multiMessageLogWriter'. +singleMessageLogWriter + :: (Applicative writerM, HasCallStack) + => (message -> writerM ()) + -> LogWriter message writerM +singleMessageLogWriter writeMessage = + foldingLogWriter (traverse_ writeMessage) + +-- | Create a 'LogWriter' from a function that is applied to each +-- individual log message. Don't be scared by the type signature, +-- here is an example file appender that re-opens the log file +-- everytime a bunch of log messages are written: +-- +-- > fileAppender fn = multiMessageLogWriter +-- > (\writeLogMessageWith -> +-- > withFile fn AppendMode (writeLogMessageWith . hPutStrLn)) +-- +multiMessageLogWriter + :: (Applicative writerM, HasCallStack) + => (((message -> writerM ()) -> writerM ()) -> writerM ()) + -> LogWriter message writerM +multiMessageLogWriter withMessageWriter = + foldingLogWriter + (\xs -> withMessageWriter (\writer -> traverse_ writer xs)) + -- | Logging effect type, parameterized by a log message type. This is a reader -- for a 'LogWriter'. -type Logs message = Reader (LogWriter message) +type LogsM message writerM = Reader (LogWriter message writerM) --- | Log a message. -logMsg :: (NFData m, MonadIO (Eff r), Member (Logs m) r) => m -> Eff r () -logMsg !(force -> m) = do - lw <- ask - liftIO (runLogWriter lw m) +-- | The logging effect on IO. +type Logs message = LogsM message IO +-- | A constraint that combines constraints for logging into any +-- log writer monad. +type HasLogWriter message logWriterMonad effects = + ( Member (LogsM message logWriterMonad) effects + , NFData message + , Monad logWriterMonad + , HasLogWriterProxy logWriterMonad + , Lifted logWriterMonad effects ) +-- | A constraint that combines constraints for IO based logging. +type HasLogWriterIO message effects = + ( Member (Logs message) effects + , NFData message + , HasLogWriterProxy IO + , Lifted IO effects ) + +-- | Type constraint alias for the implicit 'LogWriterProxy' +type HasLogWriterProxy m = ( ?logWriterProxy :: LogWriterProxy m ) + +-- | A proxy type to carry the 'LogWriter' effects. +data LogWriterProxy (e :: Type -> Type) = LogWriterProxy + +-- | Set the implicit parameter defined in 'HasLogWriterProxy' for function +-- applications like 'logMsg' +usingLogWriterProxy + :: forall e r a + . LogWriterProxy e + -> (HasLogWriterProxy e => Eff r a) -> Eff r a +usingLogWriterProxy px e = let ?logWriterProxy = px in e + +-- | Return the current 'LogWriterProxy' +logWriterProxy :: HasLogWriterProxy e => LogWriterProxy e +logWriterProxy = ?logWriterProxy + -- | Get the current 'LogWriter' -askLogWriter :: (Member (Logs m) r) => Eff r (LogWriter m) +askLogWriter :: forall m h r . (HasLogWriter m h r) => Eff r (LogWriter m h) askLogWriter = ask +-- | Log a message. +logMsg :: (NFData m, HasLogWriter m h r) => m -> Eff r () +logMsg !(force -> m) = do + lw <- askLogWriter + lift (runLogWriter lw (Identity m)) + +-- | Log a bunch of messages. This might be more efficient than calling 'logMsg' +-- multiple times. +logMsgs :: (NFData (f m), HasLogWriter m h r, Functor f, Traversable f, Foldable f) => f m -> Eff r () +logMsgs !(force -> m) = do + lw <- askLogWriter + lift (runLogWriter lw m) + -- | Change, add or remove log messages and perform arbitrary actions upon -- intercepting a log message. -- @@ -60,45 +168,59 @@ -- sending/receiving, to intercept the messages and execute some code and then -- return a new message. interceptLogging - :: forall r m a - . (Member (Logs m) r, Eff.Lifted IO r) - => (m -> Eff '[Logs m, Eff.Lift IO] ()) - -> Eff r a + :: forall r m h a + . (HasLogWriter m h r) + => (m -> Eff '[LogsM m h, Lift h] ()) + -> (HasLogWriterProxy h => Eff r a) -> Eff r a -interceptLogging interceptor action = do - old <- ask - let new = LogWriter (Eff.runLift . runReader old . interceptor) - local (const new) action +interceptLogging interceptor = + let replaceWriter old = + multiMessageLogWriter + ($ (runLift . runReader old . interceptor)) + in local replaceWriter --- | Intercept logging to change, add or remove log messages. --- --- This is without side effects, hence faster than 'interceptLogging'. -foldLogMessages - :: forall r m a f - . (NFData m, Foldable f, Member (Logs m) r, Eff.Lifted IO r) - => (m -> f m) - -> Eff r a +-- | Map a pure function over log messages. +mapLogMessages + :: forall m h r a + . (HasLogWriter m h r) + => (m -> m) + -> (HasLogWriterProxy h => Eff r a) -> Eff r a -foldLogMessages f = interceptLogging (traverse_ logMsg . f) +mapLogMessages f = + local @(LogWriter m h) + (\(LogWriter old) -> + LogWriter (\xs -> + old (fmap f xs))) --- | Throw away all log messages. -ignoreLogs :: forall message r a . Eff (Logs message ': r) a -> Eff r a -ignoreLogs = runReader (LogWriter (const (pure ()))) +-- | Filter which messages are handled. +filterLogMessages + :: forall r m h a + . (HasLogWriter m h r, HasCallStack) + => (m -> Bool) + -> (HasLogWriterProxy h => Eff r a) + -> Eff r a +filterLogMessages predicate = + local + @(LogWriter m h) + (\(LogWriter old) -> + LogWriter (old . filter predicate . toList)) --- | Apply a function that returns an effect to each log message. -handleLogsWith - :: forall message r a - . (message -> IO ()) - -> Eff (Logs message ': r) a +-- | Define the the 'LogWriter' callback and set the 'HasLogWriterProxy' +-- constraint in the given log writer action. This assumes that the monad used +-- is 'Lift'ed to @r@. +handleLogs + :: forall message writerM r a + . (Lifted writerM r) + => LogWriter message writerM + -> (HasLogWriterProxy writerM => Eff (LogsM message writerM ': r) a) -> Eff r a -handleLogsWith = runReader . LogWriter +handleLogs lw e = + usingLogWriterProxy (LogWriterProxy @writerM) (runReader lw e) --- | Handle the 'Logs' effect using 'Log.LoggingT' 'Log.Handler's. -handleLogsWithLoggingTHandler - :: forall r message a - . (Eff.Lifted IO r) - => (forall b . (Log.Handler IO message -> IO b) -> IO b) - -> Eff (Logs message ': r) a +-- | Throw away all log messages. +ignoreLogs + :: forall message m r a + . (Lifted m r, Applicative m) + => (HasLogWriterProxy m => Eff (LogsM message m ': r) a) -> Eff r a -handleLogsWithLoggingTHandler foldHandler = - handleLogsWith (foldHandler . flip ($!)) +ignoreLogs = handleLogs def diff --git a/src/Control/Eff/Log/Message.hs b/src/Control/Eff/Log/Message.hs --- a/src/Control/Eff/Log/Message.hs +++ b/src/Control/Eff/Log/Message.hs @@ -2,9 +2,14 @@ -- logging them. module Control.Eff.Log.Message ( LogMessage(..) + , HasLogging + , HasLoggingIO , renderRFC5424 , printLogMessage - , relogAsDebugMessages + , ioLogMessageHandler + , ioLogMessageWriter + , traceLogMessageWriter + , renderLogMessage , increaseLogMessageDistance , dropDistantLogMessages , logWithSeverity @@ -85,10 +90,12 @@ import Control.Monad ( (>=>) ) import Control.Monad.IO.Class import Data.Default +import Data.Foldable import Data.Maybe import Data.String import Data.Time.Clock import Data.Time.Format +import Debug.Trace import GHC.Generics import GHC.Stack import System.FilePath.Posix @@ -110,6 +117,13 @@ , _lmDistance :: Int } deriving (Eq, Generic) +-- | A convenient alias for the constraints that enable logging of 'LogMessage's +-- in the monad, which is 'Lift'ed into a given @Eff@ effect list. +type HasLogging writerM effect = (HasLogWriter LogMessage writerM effect) + +-- | Like 'HasLogging' but with 'IO' as effect. +type HasLoggingIO effect = (HasLogWriterIO LogMessage effect) + showLmMessage :: LogMessage -> [String] showLmMessage (LogMessage _f _s _ts _hn _an _pid _mi _sd ti loc msg _dist) = if null msg @@ -127,7 +141,13 @@ ) loc +-- | A 'LogWriter' that applys 'renderLogMessage' to the log message and then +-- traces it using 'traceM'. +traceLogMessageWriter :: Monad m => LogWriter LogMessage m +traceLogMessageWriter = + foldingLogWriter (traverse_ (traceM . renderLogMessage)) + -- | Render a 'LogMessage' human readable. renderLogMessage :: LogMessage -> String renderLogMessage l@(LogMessage _f s ts hn an pid mi sd _ _ _ _) = @@ -237,53 +257,62 @@ printLogMessage :: LogMessage -> IO () printLogMessage = setLogMessageTimestamp >=> putStrLn . renderLogMessage +-- | Set a timestamp (if not set), the thread id (if not set) using IO actions +-- then /write/ the log message using the 'IO' and 'String' based 'LogWriter'. +ioLogMessageWriter + :: HasCallStack => LogWriter String IO -> LogWriter LogMessage IO +ioLogMessageWriter delegatee = foldingLogWriter + ( traverse setLogMessageTimestamp + >=> traverse setLogMessageThreadId + >=> (writeAllLogMessages delegatee . fmap renderLogMessage) + ) + +-- | Use 'ioLogMessageWriter' to /handle/ logging using 'handleLogs'. +ioLogMessageHandler + :: (HasCallStack, Lifted IO e) + => LogWriter String IO + -> (HasLogWriterProxy IO => Eff (Logs LogMessage ': e) a) + -> Eff e a +ioLogMessageHandler delegatee = handleLogs (ioLogMessageWriter delegatee) + -- | An IO action that sets the current UTC time (see 'enableLogMessageTimestamps') -- in 'lmTimestamp'. setLogMessageTimestamp :: MonadIO m => LogMessage -> m LogMessage -setLogMessageTimestamp m = do - now <- liftIO getCurrentTime - return (m & lmTimestamp ?~ now) +setLogMessageTimestamp m = if isNothing (m ^. lmTimestamp) + then do + now <- liftIO getCurrentTime + return (m & lmTimestamp ?~ now) + else return m -- | An IO action appends the the 'ThreadId' of the calling process (see 'myThreadId') -- to 'lmMessage'. setLogMessageThreadId :: MonadIO m => LogMessage -> m LogMessage -setLogMessageThreadId m = do - t <- liftIO myThreadId - return (m & lmThreadId ?~ t) +setLogMessageThreadId m = if isNothing (m ^. lmThreadId) + then do + t <- liftIO myThreadId + return (m & lmThreadId ?~ t) + else return m -- | Increase the /distance/ of log messages by one. -- Logs can be filtered by their distance with 'dropDistantLogMessages' increaseLogMessageDistance - :: (HasCallStack, Member (Logs LogMessage) e, MonadIO (Eff e), Lifted IO e) - => Eff e a - -> Eff e a -increaseLogMessageDistance = interceptLogging (logMsg . over lmDistance (+ 1)) + :: (HasCallStack, HasLogWriter LogMessage h e) => Eff e a -> Eff e a +increaseLogMessageDistance = mapLogMessages (over lmDistance (+ 1)) -- | Drop all log messages with an 'lmDistance' greater than the given -- value. -dropDistantLogMessages - :: (SetMember Lift (Lift IO) r, Member (Logs LogMessage) r) - => Int - -> Eff r a - -> Eff r a -dropDistantLogMessages maxDistance = foldLogMessages - (\lm -> if lm ^. lmDistance > maxDistance then Nothing else Just lm) - --- | Handle a 'Logs' effect for 'String' messages by re-logging the messages --- as 'LogMessage's with 'debugSeverity'. -relogAsDebugMessages - :: (HasCallStack, Member (Logs LogMessage) e, MonadIO (Eff e)) - => Eff (Logs String ': e) a - -> Eff e a -relogAsDebugMessages e = withFrozenCallStack - (do - LogWriter logMsgLogger <- askLogWriter - handleLogsWith (logMsgLogger . debugMessage) e - ) +dropDistantLogMessages :: (HasLogging m r) => Int -> Eff r a -> Eff r a +dropDistantLogMessages maxDistance = + filterLogMessages (\lm -> lm ^. lmDistance <= maxDistance) -- | Log a 'String' as 'LogMessage' with a given 'Severity'. logWithSeverity - :: (HasCallStack, Member (Logs LogMessage) e, MonadIO (Eff e)) + :: ( HasLogWriterProxy h + , HasCallStack + , Monad h + , Member (LogsM LogMessage h) e + , Lifted h e + ) => Severity -> String -> Eff e () @@ -296,56 +325,96 @@ -- | Log a 'String' as 'emergencySeverity'. logEmergency - :: (HasCallStack, Member (Logs LogMessage) e, MonadIO (Eff e)) + :: ( HasCallStack + , HasLogWriterProxy h + , Monad h + , Member (LogsM LogMessage h) e + , Lifted h e + ) => String -> Eff e () logEmergency = withFrozenCallStack (logWithSeverity emergencySeverity) -- | Log a message with 'alertSeverity'. logAlert - :: (HasCallStack, Member (Logs LogMessage) e, MonadIO (Eff e)) + :: ( HasCallStack + , HasLogWriterProxy h + , Monad h + , Member (LogsM LogMessage h) e + , Lifted h e + ) => String -> Eff e () logAlert = withFrozenCallStack (logWithSeverity alertSeverity) -- | Log a 'criticalSeverity' message. logCritical - :: (HasCallStack, Member (Logs LogMessage) e, MonadIO (Eff e)) + :: ( HasCallStack + , HasLogWriterProxy h + , Monad h + , Member (LogsM LogMessage h) e + , Lifted h e + ) => String -> Eff e () logCritical = withFrozenCallStack (logWithSeverity criticalSeverity) -- | Log a 'errorSeverity' message. logError - :: (HasCallStack, Member (Logs LogMessage) e, MonadIO (Eff e)) + :: ( HasCallStack + , HasLogWriterProxy h + , Monad h + , Member (LogsM LogMessage h) e + , Lifted h e + ) => String -> Eff e () logError = withFrozenCallStack (logWithSeverity errorSeverity) -- | Log a 'warningSeverity' message. logWarning - :: (HasCallStack, Member (Logs LogMessage) e, MonadIO (Eff e)) + :: ( HasCallStack + , HasLogWriterProxy h + , Monad h + , Member (LogsM LogMessage h) e + , Lifted h e + ) => String -> Eff e () logWarning = withFrozenCallStack (logWithSeverity warningSeverity) -- | Log a 'noticeSeverity' message. logNotice - :: (HasCallStack, Member (Logs LogMessage) e, MonadIO (Eff e)) + :: ( HasCallStack + , HasLogWriterProxy h + , Monad h + , Member (LogsM LogMessage h) e + , Lifted h e + ) => String -> Eff e () logNotice = withFrozenCallStack (logWithSeverity noticeSeverity) -- | Log a 'informationalSeverity' message. logInfo - :: (HasCallStack, Member (Logs LogMessage) e, MonadIO (Eff e)) + :: ( HasCallStack + , HasLogWriterProxy h + , Monad h + , Member (LogsM LogMessage h) e + , Lifted h e + ) => String -> Eff e () logInfo = withFrozenCallStack (logWithSeverity informationalSeverity) -- | Log a 'debugSeverity' message. logDebug - :: (HasCallStack, Member (Logs LogMessage) e, MonadIO (Eff e)) + :: ( HasLogWriterProxy h + , HasCallStack + , Monad h + , Member (LogsM LogMessage h) e + , Lifted h e + ) => String -> Eff e () logDebug = withFrozenCallStack (logWithSeverity debugSeverity) diff --git a/test/Common.hs b/test/Common.hs --- a/test/Common.hs +++ b/test/Common.hs @@ -6,11 +6,14 @@ import Control.Eff.Extend import Control.Eff.Log import Control.Eff.Lift -import Control.Monad ( void ) +import Control.Monad ( void + , when + ) import Test.Tasty import Test.Tasty.HUnit import Test.Tasty.Runners import GHC.Stack +import Control.Lens setTravisTestOptions :: TestTree -> TestTree setTravisTestOptions = @@ -23,26 +26,20 @@ :: (e -> LogChannel LogMessage -> IO ()) -> (IO (e -> IO ()) -> TestTree) -> TestTree -withTestLogC doSchedule k = withResource - testLogC - testLogJoin - (\logCFactory -> k - (return - (\e -> do - logC <- logCFactory - doSchedule e logC - -- doSchedule e nullLogChannel +withTestLogC doSchedule k = k + (return + (\e -> withAsyncLogChannel + 1000 + (multiMessageLogWriter + (\writeWith -> + writeWith + (\m -> when (view lmSeverity m < noticeSeverity) (printLogMessage m) + ) + ) ) + (doSchedule e) ) ) - -testLogC :: IO (LogChannel LogMessage) -testLogC = - -- filterLogChannel ((< informationalSeverity) . _lmSeverity) <$> - forkLogger 1000 printLogMessage Nothing - -testLogJoin :: LogChannel LogMessage -> IO () -testLogJoin = joinLogChannel untilShutdown :: Member t r => t (ResumeProcess v) -> Eff r () untilShutdown pa = do diff --git a/test/ForkIOScheduler.hs b/test/ForkIOScheduler.hs --- a/test/ForkIOScheduler.hs +++ b/test/ForkIOScheduler.hs @@ -3,9 +3,10 @@ import Control.Exception import Control.Concurrent import Control.Concurrent.STM -import Control.Eff.Loop import Control.Eff.Extend import Control.Eff.Lift +import Control.Eff.Log +import Control.Eff.Loop import Control.Eff.Concurrent.Process import Control.Eff.Concurrent.Process.ForkIOScheduler as Scheduler @@ -14,6 +15,7 @@ import Test.Tasty.HUnit import Data.Dynamic import Common +import Data.Default test_IOExceptionsIsolated :: TestTree test_IOExceptionsIsolated = setTravisTestOptions $ testGroup @@ -26,23 +28,30 @@ ) $ do aVar <- newEmptyTMVarIO - lc <- testLogC - Scheduler.defaultMainWithLogChannel lc $ do - p1 <- spawn $ foreverCheap busyEffect - lift (threadDelay 1000) - void $ spawn $ do - lift (threadDelay 1000) - doExit - lift (threadDelay 100000) - wasStillRunningP1 <- sendShutdownChecked forkIoScheduler - p1 - ExitNormally - lift (atomically (putTMVar aVar wasStillRunningP1)) + withAsyncLogChannel + 1000 + def + (Scheduler.defaultMainWithLogChannel + (do + p1 <- spawn $ foreverCheap busyEffect + lift (threadDelay 1000) + void $ spawn $ do + lift (threadDelay 1000) + doExit + lift (threadDelay 100000) + wasStillRunningP1 <- sendShutdownChecked forkIoScheduler + p1 + ExitNormally + lift (atomically (putTMVar aVar wasStillRunningP1)) + ) + ) wasStillRunningP1 <- atomically (takeTMVar aVar) assertBool "the other process was still running" wasStillRunningP1 | (busyWith , busyEffect) <- - [ ("receiving", void (send (ReceiveMessage @SchedulerIO))) + [ ( "receiving" + , void (send (ReceiveSelectedMessage @SchedulerIO selectAnyMessageLazy)) + ) , ( "sending" , void (send (SendMessage @SchedulerIO 44444 (toDyn "test message"))) ) @@ -52,7 +61,14 @@ , ("selfpid-ing", void (send (SelfPid @SchedulerIO))) , ( "spawn-ing" , void - (send (Spawn @SchedulerIO (void (send (ReceiveMessage @SchedulerIO))))) + (send + (Spawn @SchedulerIO + (void + (send (ReceiveSelectedMessage @SchedulerIO selectAnyMessageLazy) + ) + ) + ) + ) ) ] , (howToExit, doExit ) <- @@ -67,11 +83,12 @@ test_mainProcessSpawnsAChildAndReturns = setTravisTestOptions (testCase "spawn a child and return" - (do - lc <- testLogC - Scheduler.defaultMainWithLogChannel - lc + (withAsyncLogChannel + 1000 + def + (Scheduler.defaultMainWithLogChannel (void (spawn (void (receiveMessage forkIoScheduler)))) + ) ) ) @@ -79,15 +96,16 @@ test_mainProcessSpawnsAChildAndExitsNormally = setTravisTestOptions (testCase "spawn a child and exit normally" - (do - lc <- testLogC - Scheduler.defaultMainWithLogChannel - lc + (withAsyncLogChannel + 1000 + def + (Scheduler.defaultMainWithLogChannel (do void (spawn (void (receiveMessage forkIoScheduler))) void (exitNormally forkIoScheduler) fail "This should not happen!!" ) + ) ) ) @@ -97,10 +115,10 @@ setTravisTestOptions (testCase "spawn a child with a busy send loop and exit normally" - (do - lc <- testLogC - Scheduler.defaultMainWithLogChannel - lc + (withAsyncLogChannel + 1000 + def + (Scheduler.defaultMainWithLogChannel (do void (spawn @@ -111,6 +129,7 @@ void (exitNormally forkIoScheduler) fail "This should not happen!!" ) + ) ) ) @@ -119,15 +138,16 @@ test_mainProcessSpawnsAChildBothReturn = setTravisTestOptions (testCase "spawn a child and let it return and return" - (do - lc <- testLogC - Scheduler.defaultMainWithLogChannel - lc + (withAsyncLogChannel + 1000 + def + (Scheduler.defaultMainWithLogChannel (do child <- spawn (void (receiveMessageAs @String forkIoScheduler)) True <- sendMessageChecked forkIoScheduler child (toDyn "test") return () ) + ) ) ) @@ -135,10 +155,10 @@ test_mainProcessSpawnsAChildBothExitNormally = setTravisTestOptions (testCase "spawn a child and let it exit and exit" - (do - lc <- testLogC - Scheduler.defaultMainWithLogChannel - lc + (withAsyncLogChannel + 1000 + def + (Scheduler.defaultMainWithLogChannel (do child <- spawn (do @@ -150,5 +170,6 @@ void (exitNormally forkIoScheduler) error "This should not happen!!" ) + ) ) ) diff --git a/test/LoggingTests.hs b/test/LoggingTests.hs --- a/test/LoggingTests.hs +++ b/test/LoggingTests.hs @@ -8,11 +8,8 @@ import Common import Control.Concurrent.STM import Control.DeepSeq -import Control.Monad.IO.Class -demo - :: (Member (Logs String) e, Member (Logs OtherLogMsg) e, MonadIO (Eff e)) - => Eff e () +demo :: (HasLogWriterIO String e, HasLogWriterIO OtherLogMsg e) => Eff e () demo = do logMsg "jo" logMsg (OtherLogMsg "test 123") @@ -25,12 +22,15 @@ otherLogsQueue <- newTQueueIO @OtherLogMsg stringLogsQueue <- newTQueueIO @String runLift - (handleLogsWith - (atomically . writeTQueue stringLogsQueue) - (handleLogsWith (atomically . writeTQueue otherLogsQueue) - (interceptLogging reverseStringLogs demo) + (handleLogs + (multiMessageLogWriter ($ (atomically . writeTQueue stringLogsQueue))) + (handleLogs + (multiMessageLogWriter ($ (atomically . writeTQueue otherLogsQueue)) + ) + (mapLogMessages @String reverse demo) ) ) + otherLogs <- atomically (flushTQueue otherLogsQueue) strLogs <- atomically (flushTQueue stringLogsQueue) assertEqual "string logs" ["oj"] strLogs @@ -38,7 +38,3 @@ ] newtype OtherLogMsg = OtherLogMsg String deriving (Eq, NFData, Show) - -reverseStringLogs - :: ('[Logs String, Lift IO] <:: e, MonadIO (Eff e)) => String -> Eff e () -reverseStringLogs = logMsg . reverse diff --git a/test/LoopTests.hs b/test/LoopTests.hs --- a/test/LoopTests.hs +++ b/test/LoopTests.hs @@ -21,27 +21,38 @@ in setTravisTestOptions $ testGroup "Loops without space leaks" - [ testCase "scheduleMonadIOEff with many yields from replicateCheapM_" $ do - res <- - Scheduler.scheduleIOWithLogging - ($! (putStrLn . (">>> " ++))) - $ replicateCheapM_ soMany - $ yieldProcess SP - res @=? Right () - , testCase "replicateCheapM_ of strict Int increments via the state effect" $ do - let - res = run - (execState - (0 :: Int) - (replicateCheapM_ soMany $ modify (force . (+ 1))) - ) - res @=? soMany - , testCase - "'foreverCheap' inside a child process and 'replicateCheapM_' in the main process" + [ testCase + "scheduleMonadIOEff with many yields from replicateCheapM_" $ do res <- Scheduler.scheduleIOWithLogging + (multiMessageLogWriter ($! (putStrLn . (">>> " ++))) + ) + $ replicateCheapM_ soMany + $ yieldProcess SP + res @=? Right () + , testCase + "replicateCheapM_ of strict Int increments via the state effect" + $ do + let + res = run + (execState + (0 :: Int) + ( replicateCheapM_ soMany + $ modify (force . (+ 1)) + ) + ) + res @=? soMany + , testCase + "'foreverCheap' inside a child process and 'replicateCheapM_' in the main process" + $ do + res <- + Scheduler.scheduleIOWithLogging + (multiMessageLogWriter + ($! (putStrLn . (">>> " ++))) + ) + $ do me <- self SP spawn_ @@ -61,35 +72,43 @@ in setTravisTestOptions $ testGroup "Loops WITH space leaks" - [ testCase "scheduleMonadIOEff with many yields from replicateM_" $ do - res <- - Scheduler.scheduleIOWithLogging - ($! (putStrLn . (">>> " ++))) - $ replicateM_ soMany - $ yieldProcess SP - res @=? Right () - , testCase "replicateM_ of strict Int increments via the state effect" $ do - let - res = run - (execState - (0 :: Int) - (replicateM_ soMany $ modify (force . (+ 1))) - ) - res @=? soMany - , testCase - "'forever' inside a child process and 'replicateM_' in the main process" - $ do + [ testCase "scheduleMonadIOEff with many yields from replicateM_" + $ do res <- Scheduler.scheduleIOWithLogging + (multiMessageLogWriter ($! (putStrLn . (">>> " ++))) + ) + $ replicateM_ soMany + $ yieldProcess SP + res @=? Right () + , testCase + "replicateM_ of strict Int increments via the state effect" + $ do + let + res = + run + (execState + (0 :: Int) + ( replicateM_ soMany + $ modify (force . (+ 1)) + ) + ) + res @=? soMany + , testCase + "'forever' inside a child process and 'replicateM_' in the main process" + $ do + res <- + Scheduler.scheduleIOWithLogging + (multiMessageLogWriter + ($! (putStrLn . (">>> " ++))) + ) $ do - me <- self SP - spawn_ - (forever $ sendMessageAs SP me () - ) - replicateM_ - soMany - (void (receiveMessageAs @() SP)) + me <- self SP + spawn_ (forever $ sendMessageAs SP me ()) + replicateM_ + soMany + (void (receiveMessageAs @() SP)) res @=? Right () - ] + ] diff --git a/test/ProcessBehaviourTestCases.hs b/test/ProcessBehaviourTestCases.hs --- a/test/ProcessBehaviourTestCases.hs +++ b/test/ProcessBehaviourTestCases.hs @@ -2,9 +2,7 @@ import Data.List ( sort ) import Data.Foldable ( traverse_ ) -import Data.Dynamic ( fromDynamic - , toDyn - ) +import Data.Dynamic ( toDyn ) import Data.Typeable import Data.Default import Control.Exception @@ -30,23 +28,20 @@ import Debug.Trace test_forkIo :: TestTree -test_forkIo = setTravisTestOptions - (withTestLogC (\c lc -> handleLoggingAndIO_ (ForkIO.schedule c) lc) - (\factory -> testGroup "ForkIOScheduler" [allTests factory]) - ) +test_forkIo = setTravisTestOptions $ withTestLogC + (\c lc -> handleLoggingAndIO_ (ForkIO.schedule c) lc) + (\factory -> testGroup "ForkIOScheduler" [allTests factory]) + test_singleThreaded :: TestTree -test_singleThreaded = setTravisTestOptions - (withTestLogC - (\e logC -> +test_singleThreaded = setTravisTestOptions $ withTestLogC + (\e logC -> -- void (runLift (logToChannel logC (SingleThreaded.schedule (return ()) e))) - let runEff :: Eff '[Logs LogMessage, Lift IO] a -> IO a - runEff = runLift . logToChannel logC - in void $ SingleThreaded.scheduleM runEff yield e - ) - (\factory -> testGroup "SingleThreadedScheduler" [allTests factory]) + let runEff :: Eff '[Logs LogMessage, Lift IO] a -> IO a + runEff = flip handleLoggingAndIO logC + in void $ SingleThreaded.scheduleM runEff yield e ) - + (\factory -> testGroup "SingleThreadedScheduler" [allTests factory]) allTests :: forall r @@ -67,7 +62,6 @@ ] ) - data ReturnToSender deriving Typeable @@ -142,14 +136,8 @@ go :: Int -> Eff (Process r ': r) () go 0 = sendMessageAs SP donePid True go n = do - void $ receiveMessageSuchThat - SP - (MessageSelector - (\m -> do - i <- fromDynamic m - if i == n then Just i else Nothing - ) - ) + void $ receiveSelectedMessage + SP (filterMessage (==n)) go (n - 1) senderLoop receviverPid = @@ -517,7 +505,9 @@ void $ atomically $ takeTMVar schedulerDoneVar | e <- [ThreadKilled, UserInterrupt, HeapOverflow, StackOverflow] , (busyWith, busyEffect) <- - [ ("receiving", void (send (ReceiveMessage @r))) + [ ( "receiving" + , void (send (ReceiveSelectedMessage @r (filterMessage (=="test message")))) + ) , ( "sending" , void (send (SendMessage @r 44444 (toDyn "test message"))) ) @@ -526,7 +516,14 @@ ) , ("selfpid-ing", void (send (SelfPid @r))) , ( "spawn-ing" - , void (send (Spawn @r (void (send (ReceiveMessage @r))))) + , void + (send + (Spawn @r + (void + (send (ReceiveSelectedMessage @r selectAnyMessageLazy)) + ) + ) + ) ) , ("sleeping", lift (threadDelay 100000)) ] @@ -539,7 +536,9 @@ void $ spawn $ foreverCheap busyEffect lift (threadDelay 10000) | (busyWith, busyEffect) <- - [ ("receiving", void (send (ReceiveMessage @r))) + [ ( "receiving" + , void (send (ReceiveSelectedMessage @r (filterMessage (=="test message")))) + ) , ( "sending" , void (send (SendMessage @r 44444 (toDyn "test message"))) ) @@ -548,7 +547,14 @@ ) , ("selfpid-ing", void (send (SelfPid @r))) , ( "spawn-ing" - , void (send (Spawn @r (void (send (ReceiveMessage @r))))) + , void + (send + (Spawn @r + (void + (send (ReceiveSelectedMessage @r selectAnyMessageLazy)) + ) + ) + ) ) ] ] @@ -572,7 +578,9 @@ assertEff "the other process was still running" wasStillRunningP1 | (busyWith , busyEffect) <- - [ ("receiving", void (send (ReceiveMessage @r))) + [ ( "receiving" + , void (send (ReceiveSelectedMessage @r (filterMessage (=="test message")))) + ) , ( "sending" , void (send (SendMessage @r 44444 (toDyn "test message"))) ) @@ -581,7 +589,14 @@ ) , ("selfpid-ing", void (send (SelfPid @r))) , ( "spawn-ing" - , void (send (Spawn @r (void (send (ReceiveMessage @r))))) + , void + (send + (Spawn @r + (void + (send (ReceiveSelectedMessage @r selectAnyMessageLazy)) + ) + ) + ) ) ] , (howToExit, doExit ) <- @@ -604,10 +619,12 @@ . (Member (Logs LogMessage) r, SetMember Lift (Lift IO) r) => IO (Eff (Process r ': r) () -> IO ()) -> TestTree -sendShutdownTests schedulerFactory = - let px :: SchedulerProxy r +sendShutdownTests schedulerFactory + = let + px :: SchedulerProxy r px = SchedulerProxy - in testGroup + in + testGroup "sendShutdown" [ testCase "... self" $ applySchedulerFactory schedulerFactory $ do me <- self px @@ -618,7 +635,8 @@ $ \assertEff -> do me <- self px r <- send (SendShutdown @r me ExitNormally) - traceShowM (show me ++": returned from SendShutdow to self " ++ show r) + traceShowM + (show me ++ ": returned from SendShutdow to self " ++ show r) assertEff "ShutdownRequested must be returned" (case r of @@ -645,7 +663,8 @@ me <- self px other <- spawn (do - untilShutdown (ReceiveMessage @r) + untilShutdown + (ReceiveSelectedMessage @r selectAnyMessageLazy) void (sendMessage px me (toDyn "OK")) ) void (sendShutdown px other ExitNormally)