diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,5 +1,15 @@ # Changelog for extensible-effects-concurrent +## 0.4.0.0 + +- Switch to `extensible-effects` version `3.0.0.0` +- Improve single threaded scheduler to be more space efficient +- Add some strictness annotations +- Add `Control.Eff.Loop` with (hopefully) constant space `forever` and + `replicateM_` +- Add `Control.Eff.Concurrent`, a module that conveniently re-exports most + library functions. + ## 0.3.0.2 - Improve single threaded scheduler such that the main process can return a value diff --git a/docs/extensible-effects-concurrent-test-Loop-WITH-space-leaks.png b/docs/extensible-effects-concurrent-test-Loop-WITH-space-leaks.png new file mode 100644 Binary files /dev/null and b/docs/extensible-effects-concurrent-test-Loop-WITH-space-leaks.png differ diff --git a/docs/extensible-effects-concurrent-test-Loop-without-space-leaks.png b/docs/extensible-effects-concurrent-test-Loop-without-space-leaks.png new file mode 100644 Binary files /dev/null and b/docs/extensible-effects-concurrent-test-Loop-without-space-leaks.png differ 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.3.0.2 +version: 0.4.0.0 description: Please see the README on GitHub at synopsis: Message passing concurrency as extensible-effect homepage: https://github.com/sheyll/extensible-effects-concurrent#readme @@ -10,9 +10,15 @@ license: BSD3 license-file: LICENSE build-type: Simple -cabal-version: >= 1.10 +cabal-version: >= 1.19 +extra-doc-files: + docs/extensible-effects-concurrent-test-Loop-without-space-leaks.png + docs/extensible-effects-concurrent-test-Loop-WITH-space-leaks.png + extra-source-files: + stack.yaml + stack.nightly-2018-05-29.yaml ChangeLog.md README.md @@ -25,12 +31,13 @@ src build-depends: base >=4.9 && <5, + deepseq, directory, filepath, time, mtl, containers, - QuickCheck <2.11, + QuickCheck <2.12, lens, logging-effect, transformers, @@ -38,12 +45,14 @@ process, monad-control, random, - extensible-effects >= 2.4 && < 2.5, + extensible-effects >= 3, stm >= 2.4.5, tagged exposed-modules: + Control.Eff.Loop, Control.Eff.ExceptionExtra, Control.Eff.Log, + Control.Eff.Concurrent, Control.Eff.Concurrent.Api, Control.Eff.Concurrent.Api.Client, Control.Eff.Concurrent.Api.Server, @@ -53,32 +62,35 @@ Control.Eff.Concurrent.Process.SingleThreadedScheduler, Control.Eff.Concurrent.Api.Observer other-modules: - Control.Eff.InternalExtra, Control.Eff.Concurrent.Api.Internal, Paths_extensible_effects_concurrent, Control.Eff.Concurrent.Examples, Control.Eff.Concurrent.Examples2 default-extensions: + BangPatterns, ConstraintKinds, DeriveFoldable, DeriveFunctor, DeriveFunctor, DeriveGeneric, DeriveTraversable, - GADTs, FlexibleContexts, + FlexibleInstances, + GADTs, GeneralizedNewtypeDeriving, + MultiParamTypeClasses, RankNTypes, ScopedTypeVariables, StandaloneDeriving, TemplateHaskell, TupleSections, TypeApplications, + TypeFamilies, TypeInType, TypeOperators, ViewPatterns default-language: Haskell2010 - ghc-options: -Wall + ghc-options: -Wall -fno-full-laziness test-suite extensible-effects-concurrent-test @@ -88,30 +100,34 @@ other-modules: Paths_extensible_effects_concurrent , ForkIOScheduler + , Interactive , SingleThreadedScheduler + , LoopTests , ProcessBehaviourTestCases , Debug , Common - ghc-options: -threaded -rtsopts -with-rtsopts=-N + ghc-options: -Wall -threaded -fno-full-laziness -rtsopts -with-rtsopts=-N build-depends: base >=4.9 && <5 , extensible-effects-concurrent - , extensible-effects + , extensible-effects >= 3 , tasty , tasty-discover , tasty-hunit , containers - , QuickCheck <2.11 + , deepseq + , QuickCheck <2.12 , lens , HUnit , stm default-language: Haskell2010 - default-extensions: TypeApplications + default-extensions: BangPatterns + , DataKinds + , FlexibleContexts + , GADTs , RankNTypes , ScopedTypeVariables - , TypeOperators - , GADTs , TemplateHaskell - , DataKinds - , FlexibleContexts - , BangPatterns + , TypeApplications + , TypeFamilies + , TypeOperators diff --git a/src/Control/Eff/Concurrent.hs b/src/Control/Eff/Concurrent.hs new file mode 100644 --- /dev/null +++ b/src/Control/Eff/Concurrent.hs @@ -0,0 +1,127 @@ +-- | Erlang style processes with message passing concurrency based on +-- (more) @extensible-effects@. +module Control.Eff.Concurrent + ( + -- * Concurrent Processes with Message Passing Concurrency + module Control.Eff.Concurrent.Process + , + -- * Data Types and Functions for APIs (aka Protocols) + module Control.Eff.Concurrent.Api + , + -- ** /Client/ Functions for Consuming APIs + module Control.Eff.Concurrent.Api.Client + , + -- ** /Server/ Functions for Providing APIs + module Control.Eff.Concurrent.Api.Server + , + -- ** /Observer/ Functions for Events and Event Listener + module Control.Eff.Concurrent.Api.Observer + , + -- * /Scheduler/ Process Effect Handler + -- ** Concurrent Scheduler + module Control.Eff.Concurrent.Process.ForkIOScheduler + , + -- ** Single Threaded Scheduler + module Control.Eff.Concurrent.Process.SingleThreadedScheduler + , + -- * Utilities + -- ** Logging Effect + module Control.Eff.Log + , + -- ** Preventing Space Leaks + module Control.Eff.Loop + ) +where + +import Control.Eff.Concurrent.Process ( ProcessId(..) + , fromProcessId + , Process(..) + , ConsProcess + , ResumeProcess(..) + , SchedulerProxy(..) + , thisSchedulerProxy + , executeAndCatch + , executeAndResume + , yieldProcess + , sendMessage + , sendMessageAs + , sendMessageChecked + , spawn + , spawn_ + , receiveMessage + , receiveMessageAs + , receiveLoop + , self + , sendShutdown + , sendShutdownChecked + , exitWithError + , exitNormally + , raiseError + , catchRaisedError + , ignoreProcessError + ) +import Control.Eff.Concurrent.Api + ( Api + , Synchronicity(..) + , Server(..) + , fromServer + , proxyAsServer + , asServer + ) +import Control.Eff.Concurrent.Api.Client + ( cast + , castChecked + , call + , castRegistered + , callRegistered + , callRegisteredA + , ServesApi + , registerServer + ) +import Control.Eff.Concurrent.Api.Server + ( serve + , ApiHandler(..) + , unhandledCallError + , unhandledCastError + , defaultTermination + , serveBoth + , serve3 + , tryApiHandler + , UnhandledRequest() + , catchUnhandled + , ensureAllHandled + , requestFromDynamic + , exitUnhandled + ) + +import Control.Eff.Concurrent.Api.Observer + ( Observer(..) + , Observable(..) + , notifyObserver + , registerObserver + , forgetObserver + , SomeObserver(..) + , notifySomeObserver + , Observers() + , manageObservers + , addObserver + , removeObserver + , notifyObservers + , CallbackObserver + , spawnCallbackObserver + , spawnLoggingObserver + ) +import Control.Eff.Concurrent.Process.ForkIOScheduler + ( schedule + , defaultMain + , defaultMainWithLogChannel + , SchedulerError(..) + , SchedulerIO + , forkIoScheduler + , HasSchedulerIO + ) + +import Control.Eff.Concurrent.Process.SingleThreadedScheduler + ( schedulePure ) +import Control.Eff.Log +import Control.Eff.Loop 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 @@ -1,18 +1,3 @@ -{-# LANGUAGE GeneralizedNewtypeDeriving #-} -{-# LANGUAGE ConstraintKinds #-} -{-# LANGUAGE FlexibleContexts #-} -{-# LANGUAGE TypeFamilies #-} -{-# LANGUAGE KindSignatures #-} -{-# LANGUAGE DataKinds #-} -{-# LANGUAGE RankNTypes #-} -{-# LANGUAGE TemplateHaskell #-} -{-# LANGUAGE DeriveFunctor #-} -{-# LANGUAGE StandaloneDeriving #-} -{-# LANGUAGE ScopedTypeVariables #-} -{-# LANGUAGE TypeOperators #-} -{-# LANGUAGE TypeApplications #-} -{-# LANGUAGE GADTs #-} - -- | This module contains a mechanism to specify what kind of messages (aka -- /requests/) a 'Server' ('Process') can handle, and if the caller blocks and -- waits for an answer, which the server process provides. @@ -41,7 +26,9 @@ import Data.Kind import Control.Lens -import Data.Typeable (Typeable, typeRep) +import Data.Typeable ( Typeable + , typeRep + ) import Control.Eff.Concurrent.Process -- | This data family defines an API, a communication interface description 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 @@ -1,26 +1,12 @@ -{-# LANGUAGE GeneralizedNewtypeDeriving #-} -{-# LANGUAGE ConstraintKinds #-} -{-# LANGUAGE FlexibleContexts #-} -{-# LANGUAGE TypeFamilies #-} -{-# LANGUAGE KindSignatures #-} -{-# LANGUAGE DataKinds #-} -{-# LANGUAGE RankNTypes #-} -{-# LANGUAGE TemplateHaskell #-} -{-# LANGUAGE DeriveFunctor #-} -{-# LANGUAGE StandaloneDeriving #-} -{-# LANGUAGE ScopedTypeVariables #-} -{-# LANGUAGE TypeOperators #-} -{-# LANGUAGE TypeApplications #-} -{-# LANGUAGE GADTs #-} - -- | Functions for 'Api' clients. -- -- This modules is required to write clients that consume an 'Api'. - module Control.Eff.Concurrent.Api.Client - ( cast + ( -- * Calling APIs directly + cast , castChecked , call + -- * Server Process Registration , castRegistered , callRegistered , callRegisteredA @@ -31,7 +17,7 @@ import Control.Applicative import Control.Eff -import Control.Eff.Reader.Lazy +import Control.Eff.Reader.Strict import Control.Eff.Concurrent.Api import Control.Eff.Concurrent.Api.Internal import Control.Eff.Concurrent.Process @@ -107,7 +93,6 @@ 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 @@ -123,7 +108,7 @@ -- | 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 +registerServer = runReader -- | Like 'call' but take the 'Server' from the reader provided by -- 'registerServer'. diff --git a/src/Control/Eff/Concurrent/Api/Internal.hs b/src/Control/Eff/Concurrent/Api/Internal.hs --- a/src/Control/Eff/Concurrent/Api/Internal.hs +++ b/src/Control/Eff/Concurrent/Api/Internal.hs @@ -1,18 +1,3 @@ -{-# LANGUAGE GeneralizedNewtypeDeriving #-} -{-# LANGUAGE ConstraintKinds #-} -{-# LANGUAGE FlexibleContexts #-} -{-# LANGUAGE TypeFamilies #-} -{-# LANGUAGE KindSignatures #-} -{-# LANGUAGE DataKinds #-} -{-# LANGUAGE RankNTypes #-} -{-# LANGUAGE TemplateHaskell #-} -{-# LANGUAGE DeriveFunctor #-} -{-# LANGUAGE StandaloneDeriving #-} -{-# LANGUAGE ScopedTypeVariables #-} -{-# LANGUAGE TypeOperators #-} -{-# LANGUAGE TypeApplications #-} -{-# LANGUAGE GADTs #-} - -- | Internal request and response type for casts and calls module Control.Eff.Concurrent.Api.Internal 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 @@ -3,29 +3,14 @@ -- 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 #-} -{-# LANGUAGE FlexibleContexts #-} -{-# LANGUAGE FlexibleInstances #-} -{-# LANGUAGE MultiParamTypeClasses #-} -{-# LANGUAGE TypeFamilies #-} -{-# LANGUAGE KindSignatures #-} -{-# LANGUAGE DataKinds #-} -{-# LANGUAGE RankNTypes #-} -{-# LANGUAGE TemplateHaskell #-} -{-# LANGUAGE DeriveFunctor #-} -{-# LANGUAGE StandaloneDeriving #-} -{-# LANGUAGE ScopedTypeVariables #-} -{-# LANGUAGE TypeOperators #-} -{-# LANGUAGE TypeApplications #-} -{-# LANGUAGE GADTs #-} module Control.Eff.Concurrent.Api.Observer - ( Observer(..) + ( -- * Observation API + Observer(..) , Observable(..) , notifyObserver , registerObserver , forgetObserver + -- ** Generalized observation , SomeObserver(..) , notifySomeObserver , Observers() @@ -33,6 +18,7 @@ , addObserver , removeObserver , notifyObservers + -- * Callback 'Observer' , CallbackObserver , spawnCallbackObserver , spawnLoggingObserver @@ -48,12 +34,10 @@ import Control.Eff.Concurrent.Api.Client import Control.Eff.Concurrent.Api.Server import Control.Eff.Log -import Control.Eff.State.Lazy +import Control.Eff.State.Strict import Control.Lens import Control.Monad --- * Observation API - -- | An 'Api' index that support observation of the -- another 'Api' that is 'Observable'. class (Typeable p, Observable o) => Observer p o where @@ -100,8 +84,6 @@ forgetObserver px observer observed = cast px observed (forgetObserverMessage (SomeObserver observer)) --- ** Generalized observation - -- | An existential wrapper around a 'Server' of an 'Observer'. -- Needed to support different types of observers to observe the -- same 'Observable' in a general fashion. @@ -143,7 +125,7 @@ -- | Keep track of registered 'Observer's Observers can be added and removed, -- and an 'Observation' can be sent to all registerd observers at once. manageObservers :: Eff (State (Observers o) ': r) a -> Eff r a -manageObservers = flip evalState (Observers Set.empty) +manageObservers = evalState (Observers Set.empty) -- | Add an 'Observer' to the 'Observers' managed by 'manageObservers'. addObserver @@ -173,8 +155,6 @@ me <- asServer @o <$> self px os <- view observers <$> get mapM_ (notifySomeObserver px me observation) os - --- * Callback 'Observer' -- | An 'Observer' that schedules the observations to an effectful callback. data CallbackObserver o 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 @@ -1,20 +1,4 @@ -{-# LANGUAGE GeneralizedNewtypeDeriving #-} -{-# LANGUAGE ConstraintKinds #-} -{-# LANGUAGE FlexibleContexts #-} -{-# LANGUAGE TypeFamilies #-} -{-# LANGUAGE KindSignatures #-} -{-# LANGUAGE DataKinds #-} -{-# LANGUAGE RankNTypes #-} -{-# LANGUAGE TemplateHaskell #-} -{-# LANGUAGE DeriveFunctor #-} -{-# LANGUAGE StandaloneDeriving #-} -{-# LANGUAGE ScopedTypeVariables #-} -{-# LANGUAGE TypeOperators #-} -{-# LANGUAGE TypeApplications #-} -{-# LANGUAGE GADTs #-} - -- | Functions to implement 'Api' __servers__. - module Control.Eff.Concurrent.Api.Server ( -- * Api Server @@ -37,8 +21,8 @@ where import Control.Eff -import Control.Eff.InternalExtra import qualified Control.Eff.Exception as Exc +import Control.Eff.Extend import Control.Eff.Concurrent.Api import Control.Eff.Concurrent.Api.Internal import Control.Eff.Concurrent.Process @@ -90,7 +74,7 @@ } -> ApiHandler p r -- | Apply either the '_handleCall', '_handleCast' or the '_handleTerminate' --- callback to an incoming 'Request'. Note, it is unlikely that this function must be used. +-- 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, HasCallStack) diff --git a/src/Control/Eff/Concurrent/Examples.hs b/src/Control/Eff/Concurrent/Examples.hs --- a/src/Control/Eff/Concurrent/Examples.hs +++ b/src/Control/Eff/Concurrent/Examples.hs @@ -1,19 +1,4 @@ -{-# LANGUAGE IncoherentInstances #-} -{-# LANGUAGE GeneralizedNewtypeDeriving #-} -{-# LANGUAGE ConstraintKinds #-} -{-# LANGUAGE FlexibleContexts #-} -{-# LANGUAGE FlexibleInstances #-} -{-# LANGUAGE TypeFamilies #-} -{-# LANGUAGE KindSignatures #-} -{-# LANGUAGE DataKinds #-} -{-# LANGUAGE RankNTypes #-} -{-# LANGUAGE TemplateHaskell #-} -{-# LANGUAGE DeriveFunctor #-} -{-# LANGUAGE StandaloneDeriving #-} -{-# LANGUAGE ScopedTypeVariables #-} -{-# LANGUAGE TypeOperators #-} -{-# LANGUAGE TypeApplications #-} -{-# LANGUAGE GADTs #-} +-- | A complete example for the library module Control.Eff.Concurrent.Examples where import GHC.Stack @@ -21,12 +6,7 @@ import Control.Eff.Lift import Control.Monad import Data.Dynamic -import Control.Eff.Concurrent.Api -import Control.Eff.Concurrent.Api.Client -import Control.Eff.Concurrent.Api.Server -import Control.Eff.Concurrent.Process -import Control.Eff.Concurrent.Process.ForkIOScheduler as Scheduler -import Control.Eff.Log +import Control.Eff.Concurrent import qualified Control.Exception as Exc data TestApi diff --git a/src/Control/Eff/Concurrent/Examples2.hs b/src/Control/Eff/Concurrent/Examples2.hs --- a/src/Control/Eff/Concurrent/Examples2.hs +++ b/src/Control/Eff/Concurrent/Examples2.hs @@ -1,32 +1,12 @@ -{-# LANGUAGE GeneralizedNewtypeDeriving #-} -{-# LANGUAGE ConstraintKinds #-} -{-# LANGUAGE FlexibleContexts #-} -{-# LANGUAGE FlexibleInstances #-} -{-# LANGUAGE MultiParamTypeClasses #-} -{-# LANGUAGE TypeFamilies #-} -{-# LANGUAGE KindSignatures #-} -{-# LANGUAGE DataKinds #-} -{-# LANGUAGE RankNTypes #-} -{-# LANGUAGE TemplateHaskell #-} -{-# LANGUAGE DeriveFunctor #-} -{-# LANGUAGE StandaloneDeriving #-} -{-# LANGUAGE ScopedTypeVariables #-} -{-# LANGUAGE TypeOperators #-} -{-# LANGUAGE TypeApplications #-} -{-# LANGUAGE GADTs #-} +-- | Another complete example for the library module Control.Eff.Concurrent.Examples2 where 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.Api -import Control.Eff.Concurrent.Api.Server -import Control.Eff.Concurrent.Api.Client -import Control.Eff.Concurrent.Process -import Control.Eff.Concurrent.Api.Observer -import Control.Eff.Log -import Control.Eff.State.Lazy +import Control.Eff.Concurrent +import Control.Eff.State.Strict import Control.Monad main :: IO () @@ -92,7 +72,6 @@ c <- get logMsg ("Cnt is " ++ show c) reply c - return () -- * Second API @@ -134,11 +113,11 @@ -> Eff r () serverLoop px = do evalState @Integer + 0 (manageObservers @Counter (serveBoth px (counterHandler px) (pocketCalcHandler px))) - 0 -- ** Counter client 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 @@ -10,32 +10,18 @@ -- -- * And a /pure/(rer) coroutine based scheduler in: -- "Control.Eff.Concurrent.Process.SingleThreadedScheduler" -{-# LANGUAGE DeriveTraversable #-} -{-# LANGUAGE DeriveFoldable #-} -{-# LANGUAGE GeneralizedNewtypeDeriving #-} -{-# LANGUAGE ConstraintKinds #-} -{-# LANGUAGE FlexibleContexts #-} -{-# LANGUAGE FlexibleInstances #-} -{-# LANGUAGE TypeFamilies #-} -{-# LANGUAGE KindSignatures #-} -{-# LANGUAGE DataKinds #-} -{-# LANGUAGE RankNTypes #-} -{-# LANGUAGE TemplateHaskell #-} -{-# LANGUAGE DeriveFunctor #-} -{-# LANGUAGE StandaloneDeriving #-} -{-# LANGUAGE ScopedTypeVariables #-} -{-# LANGUAGE TypeOperators #-} -{-# LANGUAGE TypeApplications #-} -{-# LANGUAGE GADTs #-} module Control.Eff.Concurrent.Process - ( ProcessId(..) + ( -- * ProcessId Type + ProcessId(..) , fromProcessId + -- * Process Effects , Process(..) , ConsProcess , ResumeProcess(..) , SchedulerProxy(..) , thisSchedulerProxy , executeAndCatch + , executeAndResume , yieldProcess , sendMessage , sendMessageAs @@ -56,19 +42,19 @@ ) where +import GHC.Generics ( Generic + , Generic1 + ) import GHC.Stack +import Control.DeepSeq import Control.Eff +import Control.Eff.Extend import Control.Lens -import Control.Monad ( forever - , void - ) +import Control.Monad ( void ) import Data.Dynamic import Data.Kind import Text.Printf - --- * Process Effects - -- | The process effect is the basis for message passing concurrency. This -- effect describes an interface for concurrent, communicating isolated -- processes identified uniquely by a process-id. @@ -139,8 +125,13 @@ ResumeWith :: a -> ResumeProcess a -- | This indicates that the action did not complete, and maybe retried RetryLastAction :: ResumeProcess v - deriving (Typeable, Foldable, Functor, Show, Eq, Ord, Traversable) + deriving ( Typeable, Foldable, Functor, Show, Eq, Ord + , Traversable, Generic, Generic1) +instance NFData a => NFData (ResumeProcess a) + +instance NFData1 ResumeProcess + -- | /Cons/ 'Process' onto a list of effects. type ConsProcess r = Process r ': r @@ -154,7 +145,7 @@ executeAndResume processAction = do result <- send processAction case result of - ResumeWith value -> return value + ResumeWith !value -> return value RetryLastAction -> executeAndResume processAction ShutdownRequested -> send (Shutdown @q) OnError e -> send (ExitWithError @q e) @@ -169,7 +160,7 @@ executeAndCatch px processAction = do result <- processAction case result of - ResumeWith value -> return (Right value) + ResumeWith !value -> return (Right value) RetryLastAction -> executeAndCatch px processAction ShutdownRequested -> send (Shutdown @q) OnError e -> return (Left e) @@ -299,13 +290,13 @@ => SchedulerProxy q -> (Either (Maybe String) Dynamic -> Eff r ()) -> Eff r () -receiveLoop _ handlers = forever $ do +receiveLoop px handlers = do mReq <- send (ReceiveMessage @q) case mReq of - RetryLastAction -> return () - ShutdownRequested -> handlers (Left Nothing) - OnError reason -> handlers (Left (Just reason)) - ResumeWith message -> handlers (Right message) + RetryLastAction -> receiveLoop px handlers + ShutdownRequested -> handlers (Left Nothing) >> receiveLoop px handlers + OnError reason -> handlers (Left (Just reason)) >> receiveLoop px handlers + ResumeWith message -> handlers (Right message) >> receiveLoop px handlers -- | Returns the 'ProcessId' of the current process. self 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 @@ -15,21 +15,6 @@ -- -- 'spawn' uses 'forkFinally' and 'STM.TQueue's and tries to catch -- most exceptions. -{-# LANGUAGE ConstraintKinds #-} -{-# LANGUAGE FlexibleContexts #-} -{-# LANGUAGE FlexibleInstances #-} -{-# LANGUAGE MultiParamTypeClasses #-} -{-# LANGUAGE TypeFamilies #-} -{-# LANGUAGE KindSignatures #-} -{-# LANGUAGE DataKinds #-} -{-# LANGUAGE RankNTypes #-} -{-# LANGUAGE TemplateHaskell #-} -{-# LANGUAGE DeriveFunctor #-} -{-# LANGUAGE StandaloneDeriving #-} -{-# LANGUAGE ScopedTypeVariables #-} -{-# LANGUAGE TypeOperators #-} -{-# LANGUAGE TypeApplications #-} -{-# LANGUAGE GADTs #-} module Control.Eff.Concurrent.Process.ForkIOScheduler ( schedule , defaultMain @@ -50,8 +35,8 @@ import Control.Concurrent as Concurrent import Control.Concurrent.STM as STM import Control.Eff +import Control.Eff.Extend import Control.Eff.Concurrent.Process --- import Control.Eff.ExceptionExtra import Control.Eff.Lift import Control.Eff.Log import Control.Eff.Reader.Strict as Reader @@ -80,7 +65,7 @@ instance Show ProcessInfo where show p = "ProcessInfo: " ++ show (p ^. processId) --- | Contains all 'ProcessInfo' elements, as well as the state needed to +-- | Contains all process info'elements, as well as the state needed to -- implement inter process communication. It contains also a 'LogChannel' to -- which the logs of all processes are forwarded to. data Scheduler = @@ -93,7 +78,7 @@ makeLenses ''Scheduler --- | A newtype wrapper around an 'STM.TVar' holding a 'Scheduler' state. +-- | A newtype wrapper around an 'STM.TVar' holding the scheduler state. -- This is needed by 'spawn' and provided by 'runScheduler'. newtype SchedulerVar = SchedulerVar { fromSchedulerVar :: STM.TVar Scheduler } deriving Typeable @@ -101,7 +86,7 @@ -- | A sum-type with errors that can occur when scheduleing messages. data SchedulerError = ProcessNotFound ProcessId - -- ^ No 'ProcessInfo' was found for a 'ProcessId' during internal + -- ^ No process info was found for a 'ProcessId' during internal -- processing. NOTE: This is **ONLY** caused by internal errors, probably by -- an incorrect 'MessagePassing' handler in this module. **Sending a message -- to a process ALWAYS succeeds!** Even if the process does not exist. @@ -238,8 +223,8 @@ (logToChannel l (runReader - (scheduleProcessWithCleanup shutdownAction saveCleanupAndSchedule) schedulerVar + (scheduleProcessWithCleanup shutdownAction saveCleanupAndSchedule) ) ) where @@ -343,14 +328,14 @@ -> (ResumeProcess v -> Eff SchedulerIO a) -> Eff SchedulerIO a -> Eff SchedulerIO a - shutdownOrGo pid k ok = do + shutdownOrGo pid !k !ok = do psVar <- getSchedulerTVar eHasShutdowReq <- lift (do p <- atomically (readTVar psVar) let mPinfo = p ^. processTable . at pid case mPinfo of - Just pinfo -> atomically + Just !pinfo -> atomically (do wasRequested <- readTVar (pinfo ^. shutdownRequested) -- reset the shutdwown request flag, in case the shutdown @@ -432,7 +417,7 @@ lift Concurrent.yield k (ResumeWith pid) - go pid YieldProcess k = shutdownOrGo pid k $ do + go pid YieldProcess !k = shutdownOrGo pid k $ do lift Concurrent.yield k (ResumeWith ()) 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 @@ -1,38 +1,18 @@ -module Control.Eff.Concurrent.Process.Interactive - ( SchedulerSession() - , forkInteractiveScheduler - , killInteractiveScheduler - , submit - , submitCast - , submitCall - ) -where - -import Control.Arrow -import Control.Concurrent -import Control.Concurrent.STM -import Control.Eff -import Control.Eff.Lift -import Control.Eff.Concurrent.Api -import Control.Eff.Concurrent.Api.Client -import Control.Eff.Concurrent.Process -import Control.Monad -import Data.Typeable ( Typeable ) -import System.Timeout - -- | This module provides support for executing 'Process' actions from 'IO'. -- -- One use case is interacting with processes from the REPL, e.g.: -- -- >>> import Control.Eff.Concurrent.Process.SingleThreadedScheduler (defaultMain) -- +-- >>> import Control.Eff.Loop +-- -- >>> import Data.Dynamic -- -- >>> import Data.Maybe -- -- >>> s <- forkInteractiveScheduler Control.Eff.Concurrent.Process.SingleThreadedScheduler.defaultMain -- --- >>> fooPid <- submit s (spawn (forever (receiveMessage SP >>= (logMsg . fromMaybe "Huh!??" . fromDynamic)))) +-- >>> fooPid <- submit s (spawn (foreverCheap (receiveMessage SP >>= (logMsg . fromMaybe "Huh!??" . fromDynamic)))) -- -- >>> fooPid -- <0.1.0> @@ -45,7 +25,28 @@ -- -- -- @since 0.3.0.1 +module Control.Eff.Concurrent.Process.Interactive + ( SchedulerSession() + , forkInteractiveScheduler + , killInteractiveScheduler + , submit + , submitCast + , submitCall + ) +where +import Control.Concurrent +import Control.Concurrent.STM +import Control.Eff +import Control.Eff.Lift +import Control.Eff.Concurrent.Api +import Control.Eff.Concurrent.Api.Client +import Control.Eff.Concurrent.Process +import Control.Monad +import Data.Foldable +import Data.Typeable ( Typeable ) +import System.Timeout + -- | Contains the communication channels to interact with a scheduler running in -- its' own thread. newtype SchedulerSession r = SchedulerSession (TMVar (SchedulerQueue r)) @@ -74,28 +75,26 @@ ) return (SchedulerSession queueVar) where - readEvalPrintLoop = forever . (readAction >>> evalAction >=> printResult) + readEvalPrintLoop queueVar = do + nextActionOrExit <- readAction + case nextActionOrExit of + Left True -> return () + Left False -> readEvalPrintLoop queueVar + Right nextAction -> do + res <- nextAction + traverse_ (lift . putStrLn . (">>> " ++)) res + yieldProcess SP + readEvalPrintLoop queueVar where - readAction queueVar = do - nextActionOrExit <- lift $ atomically - (do - mInQueue <- tryReadTMVar queueVar - case mInQueue of - Nothing -> return (Left True) - Just (SchedulerQueue inQueue) -> do - mnextAction <- tryReadTChan inQueue - case mnextAction of - Nothing -> return (Left False) - Just nextAction -> return (Right nextAction) - ) - case nextActionOrExit of - Left True -> exitNormally SP - Left False -> do - yieldProcess SP - readAction queueVar - Right r -> return r - evalAction = join - printResult = mapM_ (lift . putStrLn) + readAction = lift $ atomically $ do + mInQueue <- tryReadTMVar queueVar + case mInQueue of + Nothing -> return (Left True) + Just (SchedulerQueue inQueue) -> do + mnextAction <- tryReadTChan inQueue + case mnextAction of + Nothing -> return (Left False) + Just nextAction -> return (Right nextAction) -- | Exit the schedulder immediately using an asynchronous exception. killInteractiveScheduler :: SchedulerSession r -> IO () 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 @@ -1,9 +1,10 @@ -- | A coroutine based, single threaded scheduler for 'Process'es. module Control.Eff.Concurrent.Process.SingleThreadedScheduler - ( schedule + ( scheduleM , schedulePure , scheduleIO - , scheduleWithLogging + , scheduleMonadIOEff + , scheduleIOWithLogging , defaultMain , singleThreadedIoScheduler , LoggingAndIo @@ -11,14 +12,17 @@ where import Control.Concurrent ( yield ) +import Control.DeepSeq import Control.Eff -import Control.Eff.Concurrent.Process import Control.Eff.Lift +import Control.Eff.Extend +import Control.Eff.Concurrent.Process import Control.Eff.Log import Control.Lens hiding ( (|>) , Empty ) import Control.Monad ( void ) +import Control.Monad.IO.Class import qualified Data.Sequence as Seq import Data.Sequence ( Seq(..) ) import qualified Data.Map.Strict as Map @@ -27,141 +31,79 @@ import Data.Dynamic import Data.Maybe --- | Invoke 'schedule' with @lift 'yield'@ as yield effect. --- @since 0.3.0.2 -scheduleIO - :: SetMember Lift (Lift IO) r - => Eff (ConsProcess r) a - -> Eff r (Either String a) -scheduleIO = schedule (lift yield) -- | Like 'schedule' but /pure/. The @yield@ effect is just @return ()@. --- @schedulePure == 'run' . 'schedule' (return ())@ +-- @schedulePure == runIdentity . 'scheduleM' (Identity . run) (return ())@ +-- -- @since 0.3.0.2 schedulePure :: Eff (ConsProcess '[]) a -> Either String a -schedulePure = run . schedule (return ()) +schedulePure = runIdentity . scheduleM (Identity . run) (return ()) --- | Like 'schedulePure' but with logging. --- @scheduleWithLogging == 'run' . 'captureLogs' . 'schedule' (return ())@ +-- | Invoke 'schedule' with @lift 'Control.Concurrent.yield'@ as yield effect. +-- @scheduleIO runEff == 'scheduleM' (runLift . runEff) (liftIO 'yield')@ +-- +-- @since 0.4.0.0 +scheduleIO + :: MonadIO m + => (forall b . Eff r b -> Eff '[Lift m] b) + -> Eff (ConsProcess r) a + -> m (Either String a) +scheduleIO runEff = scheduleM (runLift . runEff) (liftIO yield) + +-- | Invoke 'schedule' with @lift 'Control.Concurrent.yield'@ as yield effect. +-- @scheduleMonadIOEff == 'scheduleM' id (liftIO 'yield')@ +-- -- @since 0.3.0.2 -scheduleWithLogging :: Eff (ConsProcess '[Logs m]) a -> (Either String a, Seq m) -scheduleWithLogging = run . captureLogs . schedule (return ()) +scheduleMonadIOEff + :: MonadIO (Eff r) => Eff (ConsProcess r) a -> Eff r (Either String a) +scheduleMonadIOEff = -- schedule (lift yield) + scheduleM id (liftIO yield) --- | Execute a 'Process' and all the other processes 'spawn'ed by it in the +-- | Run processes that have the 'Logs' and the 'Lift' effects. +-- The user must provide a log handler function. +-- +-- Log messages are evaluated strict. +-- +-- @scheduleIOWithLogging == 'run' . 'captureLogs' . 'schedule' (return ())@ +-- +-- @since 0.4.0.0 +scheduleIOWithLogging + :: (NFData l, MonadIO m) + => (l -> m ()) + -> Eff (ConsProcess '[Logs l, Lift m]) a + -> m (Either String a) +scheduleIOWithLogging handleLog e = scheduleIO (handleLogsWith handleLog) e + +-- | Handle the 'Process' effect, as well as all lower effects using an effect handler function. +-- +-- Execute the __main__ 'Process' and all the other processes 'spawn'ed by it in the -- current thread concurrently, using a co-routine based, round-robin -- scheduler. If a process exits with 'exitNormally', 'exitWithError', -- 'raiseError' or is killed by another process @Left ...@ is returned. -- Otherwise, the result will be wrapped in a @Right@. -schedule - :: forall r finalResult - . Eff r () -- ^ 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 (Process r ': r) finalResult - -> Eff r (Either String finalResult) -schedule yieldEff mainProcessAction = do - y <- runAsCoroutine mainProcessAction - go 1 (Map.singleton 0 Seq.empty) (Seq.singleton (y, 0)) - where - go - :: ProcessId - -> Map.Map ProcessId (Seq Dynamic) - -> Seq (OnYield r finalResult, ProcessId) - -> Eff r (Either String finalResult) - go _newPid _msgQs Empty = return (Left "no main process") - - go newPid msgQs allProcs@((processState, pid) :<| rest) - = let - handleExit res = if pid == 0 - then return res - else go newPid (msgQs & at pid .~ Nothing) rest - maybeYield = if pid == 0 then yieldEff else return () - in - case processState of - OnDone r -> handleExit (Right r) - - OnShutdown -> handleExit (Left "process exited normally") - - OnRaiseError errM -> handleExit (Left errM) - - OnExitError errM -> handleExit (Left errM) - - OnSendShutdown targetPid k -> do - let allButTarget = - Seq.filter (\(_, e) -> e /= pid && e /= targetPid) allProcs - targets = Seq.filter (\(_, e) -> e == targetPid) allProcs - suicide = targetPid == pid - targetFound = suicide || not (Seq.null targets) - if suicide - then do - nextK <- k ShutdownRequested - go newPid msgQs (rest :|> (nextK, pid)) - else do - let deliverTheGoodNews (targetState, tPid) = do - nextTargetState <- case targetState of - OnSendShutdown _ tk -> tk ShutdownRequested - OnYield tk -> tk ShutdownRequested - OnSelf tk -> tk ShutdownRequested - OnSend _ _ tk -> tk ShutdownRequested - OnRecv tk -> tk ShutdownRequested - OnSpawn _ tk -> tk ShutdownRequested - OnDone x -> return (OnDone x) - OnShutdown -> return OnShutdown - OnExitError er -> return (OnExitError er) - OnRaiseError er -> return (OnExitError er) -- return (error ("TODO write test "++er)) - return (nextTargetState, tPid) - nextTargets <- traverse deliverTheGoodNews targets - nextK <- k (ResumeWith targetFound) - maybeYield - go newPid - msgQs - (allButTarget Seq.>< (nextTargets :|> (nextK, pid))) - - - OnSelf k -> do - nextK <- k (ResumeWith pid) - maybeYield - go newPid msgQs (rest :|> (nextK, pid)) - - OnYield k -> do - yieldEff - nextK <- k (ResumeWith ()) - go newPid msgQs (rest :|> (nextK, pid)) - - OnSend toPid msg k -> do - nextK <- k (ResumeWith (msgQs ^. at toPid . to isJust)) - maybeYield - go newPid - (msgQs & at toPid . _Just %~ (:|> msg)) - (rest :|> (nextK, pid)) - - recv@(OnRecv k) -> case msgQs ^. at pid of - Nothing -> do - nextK <- k (OnError (show pid ++ " has no message queue!")) - maybeYield - go newPid msgQs (rest :|> (nextK, pid)) - Just Empty -> if Seq.length rest == 0 - then do - nextK <- k (OnError ("Process " ++ show pid ++ " deadlocked!")) - maybeYield - go newPid msgQs (rest :|> (nextK, pid)) - else go newPid msgQs (rest :|> (recv, pid)) - - Just (nextMessage :<| restMessages) -> do - nextK <- k (ResumeWith nextMessage) - maybeYield - go newPid - (msgQs & at pid . _Just .~ restMessages) - (rest :|> (nextK, pid)) - - OnSpawn f k -> do - nextK <- k (ResumeWith newPid) - fk <- runAsCoroutine (f >> exitNormally SP) - maybeYield - go (newPid + 1) - (msgQs & at newPid .~ Just Seq.empty) - (rest :|> (nextK, pid) :|> (fk, newPid)) +-- +-- Every time a process _yields_ the effects are evaluated down to the a value +-- of type @m (Either String a)@. +-- +-- If the evaluator function runs the action down e.g. @IO@ this might improve +-- memory consumption, for long running services, with processes that loop +-- endlessly. +-- +-- @since 0.4.0.0 +scheduleM + :: 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 + -> m (Either String a) +scheduleM runEff yieldEff e = do + y <- runAsCoroutinePure runEff e + handleProcess runEff yieldEff 1 (Map.singleton 0 Seq.empty) (Seq.singleton (y, 0)) +-- | Internal data structure that is part of the coroutine based scheduler +-- implementation. data OnYield r a where OnYield :: (ResumeProcess () -> Eff r (OnYield r a)) -> OnYield r a @@ -170,30 +112,138 @@ OnSpawn :: Eff (Process r ': r) () -> (ResumeProcess ProcessId -> Eff r (OnYield r a)) -> OnYield r a - OnDone :: a -> OnYield r a + OnDone :: !a -> OnYield r a OnShutdown :: OnYield r a - OnExitError :: String -> OnYield r a - OnRaiseError :: String -> OnYield r a - OnSend :: ProcessId -> Dynamic + OnExitError :: !String -> OnYield r a + OnRaiseError :: !String -> OnYield r a + OnSend :: !ProcessId -> !Dynamic -> (ResumeProcess Bool -> Eff r (OnYield r a)) -> OnYield r a OnRecv :: (ResumeProcess Dynamic -> Eff r (OnYield r a)) -> OnYield r a - OnSendShutdown :: ProcessId -> (ResumeProcess Bool -> Eff r (OnYield r a)) -> OnYield r a + OnSendShutdown :: !ProcessId -> (ResumeProcess Bool -> Eff r (OnYield r a)) -> OnYield r a -runAsCoroutine :: forall r v . Eff (Process r ': r) v -> Eff r (OnYield r v) -runAsCoroutine m = handle_relay (return . OnDone) cont m +-- | Internal 'Process' handler function. +handleProcess + :: Monad m + => (forall a . Eff r a -> m a) + -> m () + -> ProcessId + -> Map.Map ProcessId (Seq Dynamic) + -> Seq (OnYield r finalResult, ProcessId) + -> m (Either String finalResult) +handleProcess _runEff _yieldEff _newPid _msgQs Empty = return $ Left "no main process" + +handleProcess runEff yieldEff !newPid !msgQs allProcs@((!processState, !pid) :<| rest) + = let handleExit res = if pid == 0 + then return res + else handleProcess runEff yieldEff newPid (msgQs & at pid .~ Nothing) rest + in + case processState of + OnDone r -> handleExit (Right r) + + OnShutdown -> handleExit (Left "process exited normally") + + OnRaiseError errM -> handleExit (Left errM) + + OnExitError errM -> handleExit (Left errM) + + OnSendShutdown targetPid k -> do + let allButTarget = + Seq.filter (\(_, e) -> e /= pid && e /= targetPid) allProcs + targets = Seq.filter (\(_, e) -> e == targetPid) allProcs + suicide = targetPid == pid + targetFound = suicide || not (Seq.null targets) + if suicide + then do + nextK <- runEff $ k ShutdownRequested + handleProcess runEff yieldEff newPid msgQs (rest :|> (nextK, pid)) + else do + let deliverTheGoodNews (targetState, tPid) = do + nextTargetState <- case targetState of + OnSendShutdown _ tk -> tk ShutdownRequested + OnYield tk -> tk ShutdownRequested + OnSelf tk -> tk ShutdownRequested + OnSend _ _ tk -> tk ShutdownRequested + OnRecv tk -> tk ShutdownRequested + OnSpawn _ tk -> tk ShutdownRequested + OnDone x -> return (OnDone x) + OnShutdown -> return OnShutdown + OnExitError er -> return (OnExitError er) + OnRaiseError er -> return (OnExitError er) + return (nextTargetState, tPid) + nextTargets <- runEff $ traverse deliverTheGoodNews targets + nextK <- runEff $ k (ResumeWith targetFound) + handleProcess runEff + yieldEff + newPid + msgQs + (allButTarget Seq.>< (nextTargets :|> (nextK, pid))) + + + OnSelf k -> do + nextK <- runEff $ k (ResumeWith pid) + handleProcess runEff yieldEff newPid msgQs (rest :|> (nextK, pid)) + + OnYield k -> do + yieldEff + nextK <- runEff $ k (ResumeWith ()) + handleProcess runEff yieldEff newPid msgQs (rest :|> (nextK, pid)) + + OnSend toPid msg k -> do + nextK <- runEff $ k (ResumeWith (msgQs ^. at toPid . to isJust)) + handleProcess runEff + yieldEff + newPid + (msgQs & at toPid . _Just %~ (:|> msg)) + (rest :|> (nextK, pid)) + + recv@(OnRecv k) -> case msgQs ^. at pid of + Nothing -> do + nextK <- runEff $ k (OnError (show pid ++ " has no message queue!")) + handleProcess runEff yieldEff newPid msgQs (rest :|> (nextK, pid)) + Just Empty -> if Seq.length rest == 0 + then do + nextK <- runEff + $ k (OnError ("Process " ++ show pid ++ " deadlocked!")) + handleProcess runEff yieldEff newPid msgQs (rest :|> (nextK, pid)) + else handleProcess runEff yieldEff newPid msgQs (rest :|> (recv, pid)) + + Just (nextMessage :<| restMessages) -> do + nextK <- runEff $ k (ResumeWith nextMessage) + handleProcess runEff + yieldEff + newPid + (msgQs & at pid . _Just .~ restMessages) + (rest :|> (nextK, pid)) + + OnSpawn f k -> do + nextK <- runEff $ k (ResumeWith newPid) + fk <- runAsCoroutinePure runEff (f >> exitNormally SP) + handleProcess runEff + yieldEff + (newPid + 1) + (msgQs & at newPid .~ Just Seq.empty) + (rest :|> (nextK, pid) :|> (fk, newPid)) + +runAsCoroutinePure + :: forall v r m + . Monad m + => (forall a . Eff r a -> m a) + -> Eff (ConsProcess r) v + -> m (OnYield r v) +runAsCoroutinePure runEff = runEff . handle_relay (return . OnDone) cont where cont :: Process r x -> (x -> Eff r (OnYield r v)) -> Eff r (OnYield r v) - cont YieldProcess k = return (OnYield k) - cont SelfPid k = return (OnSelf k) - cont (Spawn e) k = return (OnSpawn e k) - cont Shutdown _k = return OnShutdown - cont (ExitWithError e ) _k = return (OnExitError e) - cont (RaiseError e ) _k = return (OnRaiseError e) - cont (SendMessage tp msg) k = return (OnSend tp msg k) - cont ReceiveMessage k = return (OnRecv k) - cont (SendShutdown pid) k = return (OnSendShutdown pid k) + cont YieldProcess k = return (OnYield k) + cont SelfPid k = return (OnSelf k) + cont (Spawn e) k = return (OnSpawn e k) + cont Shutdown _k = return OnShutdown + cont (ExitWithError !e ) _k = return (OnExitError e) + cont (RaiseError !e ) _k = return (OnRaiseError e) + cont (SendMessage !tp !msg) k = return (OnSend tp msg k) + cont ReceiveMessage k = return (OnRecv k) + cont (SendShutdown !pid) k = return (OnSendShutdown pid k) -- | The concrete list of 'Eff'ects for running this pure scheduler on @IO@ and -- with string logging. @@ -212,4 +262,6 @@ :: HasCallStack => Eff '[Process '[Logs String, Lift IO], Logs String, Lift IO] () -> IO () -defaultMain go = void $ runLift $ handleLogsWith (scheduleIO go) ($! putStrLn) +defaultMain e = void $ runLift $ handleLogsWithLoggingTHandler + (scheduleMonadIOEff e) + ($! putStrLn) diff --git a/src/Control/Eff/InternalExtra.hs b/src/Control/Eff/InternalExtra.hs deleted file mode 100644 --- a/src/Control/Eff/InternalExtra.hs +++ /dev/null @@ -1,17 +0,0 @@ --- | "Control.Eff" utilities, some stolen from future versions of --- @extensible-effects@. In future versions, there might be something like a --- @Control.Eff.Extend@ module that might contain a lot of it, according to --- the discussion here: --- https://github.com/suhailshergill/extensible-effects/issues/98 -module Control.Eff.InternalExtra where - -import Control.Eff - --- | Embeds a less-constrained 'Eff' into a more-constrained one. Analogous to --- MTL's 'lift'. -raise :: Eff r a -> Eff (e ': r) a -raise = loop - where - loop (Val x) = pure x - loop (E u q) = E (weaken u) $ qComps q loop -{-# INLINE raise #-} diff --git a/src/Control/Eff/Log.hs b/src/Control/Eff/Log.hs --- a/src/Control/Eff/Log.hs +++ b/src/Control/Eff/Log.hs @@ -1,15 +1,3 @@ -{-# LANGUAGE UndecidableInstances #-} -{-# LANGUAGE MultiParamTypeClasses #-} -{-# LANGUAGE TypeFamilies #-} -{-# LANGUAGE TypeOperators #-} -{-# LANGUAGE TypeInType #-} -{-# LANGUAGE FlexibleContexts #-} -{-# LANGUAGE FlexibleInstances #-} -{-# LANGUAGE ScopedTypeVariables #-} -{-# LANGUAGE RankNTypes #-} -{-# LANGUAGE GADTs #-} -{-# LANGUAGE KindSignatures #-} -{-# LANGUAGE DataKinds #-} -- | A logging effect based on 'Control.Monad.Log.MonadLog'. module Control.Eff.Log ( @@ -22,6 +10,7 @@ , captureLogs , ignoreLogs , handleLogsWith + , handleLogsWithLoggingTHandler -- * Concurrent Logging , LogChannel() , logToChannel @@ -33,12 +22,17 @@ , closeLogChannelAfter , logChannelBracket , logChannelPutIO + -- ** Internals + , JoinLogChannelException() + , KillLogChannelException() ) where import Control.Concurrent import Control.Concurrent.STM +import Control.DeepSeq import Control.Eff as Eff +import Control.Eff.Extend as Eff import Control.Exception ( bracket ) import qualified Control.Exception as Exc import Control.Monad ( void @@ -51,8 +45,8 @@ 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 Data.Kind ( ) +import Data.Sequence ( Seq() ) import qualified Data.Sequence as Seq import Data.String import Data.Typeable @@ -81,15 +75,14 @@ -- 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 () +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'. @@ -105,56 +98,71 @@ -- 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 () +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) -captureLogs actionThatLogs = - Eff.handle_relay_s - Seq.empty - (\logs result -> return (result, logs)) - handleLogs - actionThatLogs - where - handleLogs :: Seq message - -> Logs message x - -> (Seq message -> Arr r x y) - -> Eff r y - handleLogs logs (LogMsg m) k = - k (logs Seq.:|> m) () +-- | Capture all log messages in a 'Seq' (strict). +captureLogs + :: NFData message => Eff (Logs message ': r) a -> Eff r (a, Seq message) +captureLogs actionThatLogs = Eff.handle_relay_s + Seq.empty + (\logs result -> return (result, logs)) + handleLogs + actionThatLogs + where + handleLogs + :: NFData message + => Seq message + -> Logs message x + -> (Seq message -> Arr r x y) + -> Eff r y + handleLogs !logs (LogMsg !m) k = k (force (logs Seq.:|> m)) () -- | Throw away all log messages. -ignoreLogs :: Eff (Logs message ': r) a - -> Eff r a -ignoreLogs actionThatLogs = - Eff.handle_relay return handleLogs actionThatLogs - where - handleLogs :: Logs m x -> Arr r x y -> Eff r y - handleLogs (LogMsg _) k = k () +ignoreLogs :: forall message r a . Eff (Logs message ': r) a -> Eff r a +ignoreLogs actionThatLogs = Eff.handle_relay return handleLogs actionThatLogs + where + handleLogs :: Logs m x -> Arr r x y -> Eff r y + handleLogs (LogMsg _) k = k () --- | Handle 'Logs' effects using 'Log.LoggingT' 'Log.Handler's. +-- | Handle the 'Logs' effect with a monadic call back function (strict). handleLogsWith :: forall m r message a + . (NFData message, Monad m, SetMember Eff.Lift (Eff.Lift m) r) + => (message -> m ()) + -> Eff (Logs message ': r) a + -> Eff r a +handleLogsWith logMessageHandler = Eff.handle_relay return go + where + go :: Logs message b -> (b -> Eff r c) -> Eff r c + go (LogMsg m) k = do + res <- Eff.lift (logMessageHandler (force m)) + k res + +-- | Handle the 'Logs' effect using 'Log.LoggingT' 'Log.Handler's. +handleLogsWithLoggingTHandler + :: forall m r message a . (Monad m, SetMember Eff.Lift (Eff.Lift m) r) => Eff (Logs message ': r) a -> (forall b . (Log.Handler m message -> m b) -> m b) -> Eff r a -handleLogsWith actionThatLogs foldHandler = Eff.handle_relay return - go - actionThatLogs +handleLogsWithLoggingTHandler actionThatLogs foldHandler = Eff.handle_relay + return + go + actionThatLogs where go :: Logs message b -> (b -> Eff r c) -> Eff r c - go (LogMsg m) k = - Eff.lift (foldHandler (\doLog -> doLog m)) >>= k + go (LogMsg m) k = Eff.lift (foldHandler (\doLog -> doLog m)) >>= k -- | 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 @@ -178,19 +186,17 @@ -> Eff (Logs message ': r) a -> Eff r a logToChannel logChan actionThatLogs = do - handleLogsWith actionThatLogs - (\withHandler -> withHandler (logChannelPutIO logChan)) + handleLogsWithLoggingTHandler + actionThatLogs + (\withHandler -> withHandler (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) +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'. @@ -218,21 +224,18 @@ return (ConcurrentLogChannel msgQ thread) 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 mCloseMsg) -> - do traverse_ handle logMessages - traverse_ handle mCloseMsg - Nothing -> - case Exc.fromException se of - Just (KillLogChannelException mCloseMsg) -> - traverse_ handle mCloseMsg - Nothing -> - mapM_ handle logMessages + 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 logLoop :: TBQueue message -> IO () logLoop tq = do @@ -241,62 +244,66 @@ logLoop tq -- | Filter logs sent to a 'LogChannel' using a predicate. -filterLogChannel :: (message -> Bool) -> LogChannel message -> LogChannel message +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) +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 +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 + 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 + :: (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 + 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 + :: (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 + if f closeLogMessage + then killLogChannel (Just closeLogMessage) lc + else killLogChannel Nothing lc killLogChannel closeLogMessage (ConcurrentLogChannel _tq thread) = throwTo thread (KillLogChannelException closeLogMessage) @@ -335,4 +342,7 @@ logChannelBracket queueLen mWelcome mGoodbye f = control (\runInIO -> do let logHandler = void . runInIO . logMessage - bracket (forkLogger queueLen logHandler mWelcome) (joinLogChannel mGoodbye) f) + bracket (forkLogger queueLen logHandler mWelcome) + (joinLogChannel mGoodbye) + f + ) diff --git a/src/Control/Eff/Loop.hs b/src/Control/Eff/Loop.hs new file mode 100644 --- /dev/null +++ b/src/Control/Eff/Loop.hs @@ -0,0 +1,53 @@ +-- | In rare occasions GHC optimizes innocent looking loops into space-leaking monsters. +-- See the discussion here: for more +-- details, or +-- +-- These functions in this module __/might/__ help, at least in conjunction with +-- the @-fno-full-laziness@ GHC option. +-- +-- There is a unit test in the sources of this module, which can be used to do a +-- comperative heap profiling of these function vs. their counterparts in the +-- @base@ package. +-- +-- Here are the images of the profiling results, the images show that the +-- functions in this module do not leak space, compared to the original functions +-- ('Control.Monad.forever' and 'Control.Monad.replicateM_'): +-- +-- ![Heap profiling of the unit tests called Loop-without-space-leaks shows that less than 200k Bytes were used](docs/extensible-effects-concurrent-test-Loop-without-space-leaks.png) +-- ![Heap profiling of the unit tests called Loop-WITH-space-leaks shows that at least 160M Bytes were used](docs/extensible-effects-concurrent-test-Loop-WITH-space-leaks.png) +-- +-- @since 0.4.0.0 +module Control.Eff.Loop + ( foreverCheap + , replicateCheapM_ + ) +where + +import Control.Monad ( void ) + +-- | A version of 'Control.Monad.forever' that hopefully tricks +-- GHC into __/not/__ creating a space leak. +-- The intuition is, that we want to do something that is /cheap/, and hence +-- should be __recomputed__ instead of shared. +-- +-- @since 0.4.0.0 +{-# NOINLINE foreverCheap #-} +foreverCheap :: Monad m => m a -> m () +foreverCheap x = x >> foreverCheap (x >> x) + +-- | A version of 'Control.Monad.replicateM_' that hopefully tricks +-- GHC into __/not/__ creating a space leak. +-- The intuition is, that we want to do something that is /cheap/, and hence +-- should be __recomputed__ instead of shared. +-- +-- @since 0.4.0.0 +{-# NOINLINE replicateCheapM_ #-} +replicateCheapM_ :: Monad m => Int -> m a -> m () +replicateCheapM_ n e = if n <= 0 + then return () + else if n == 1 + then void e + else + let nL = n `div` 2 + nR = n - nL + in replicateCheapM_ nL e >> replicateCheapM_ nR e diff --git a/stack.nightly-2018-05-29.yaml b/stack.nightly-2018-05-29.yaml new file mode 100644 --- /dev/null +++ b/stack.nightly-2018-05-29.yaml @@ -0,0 +1,68 @@ +# This file was automatically generated by 'stack init' +# +# Some commonly used options have been documented as comments in this file. +# For advanced use and comprehensive documentation of the format, please see: +# https://docs.haskellstack.org/en/stable/yaml_configuration/ + +# Resolver to choose a 'specific' stackage snapshot or a compiler version. +# A snapshot resolver dictates the compiler version and the set of packages +# to be used for project dependencies. For example: +# +# resolver: lts-3.5 +# resolver: nightly-2015-09-21 +# resolver: ghc-7.10.2 +# resolver: ghcjs-0.1.0_ghc-7.10.2 +# resolver: +# name: custom-snapshot +# location: "./custom-snapshot.yaml" +resolver: nightly-2018-05-29 + +# User packages to be built. +# Various formats can be used as shown in the example below. +# +# packages: +# - some-directory +# - https://example.com/foo/bar/baz-0.0.2.tar.gz +# - location: +# git: https://github.com/commercialhaskell/stack.git +# commit: e7b331f14bcffb8367cd58fbfc8b40ec7642100a +# - location: https://github.com/commercialhaskell/stack/commit/e7b331f14bcffb8367cd58fbfc8b40ec7642100a +# extra-dep: true +# subdirs: +# - auto-update +# - wai +# +# A package marked 'extra-dep: true' will only be built if demanded by a +# non-dependency (i.e. a user package), and its test suites and benchmarks +# will not be run. This is useful for tweaking upstream packages. +packages: +- . +# Dependency packages to be pulled from upstream that are not in the resolver +# (e.g., acme-missiles-0.3) +extra-deps: +- brittany-0.10.0.0 +- extensible-effects-3.0.0.0 + +# Override default flag values for local packages and extra-deps +# flags: {} + +# Extra package databases containing global packages +# extra-package-dbs: [] + +# Control whether we use the GHC we find on the path +# system-ghc: true +# +# Require a specific version of stack, using version ranges +# require-stack-version: -any # Default +require-stack-version: ">=1.6" +# +# Override the architecture used by stack, especially useful on Windows +# arch: i386 +# arch: x86_64 +# +# Extra directories used by stack for building +# extra-include-dirs: [/path/to/dir] +# extra-lib-dirs: [/path/to/dir] +# +# Allow a newer minor version of GHC than the snapshot specifies +# compiler-check: newer-minor diff --git a/stack.yaml b/stack.yaml new file mode 100644 --- /dev/null +++ b/stack.yaml @@ -0,0 +1,68 @@ +# This file was automatically generated by 'stack init' +# +# Some commonly used options have been documented as comments in this file. +# For advanced use and comprehensive documentation of the format, please see: +# https://docs.haskellstack.org/en/stable/yaml_configuration/ + +# Resolver to choose a 'specific' stackage snapshot or a compiler version. +# A snapshot resolver dictates the compiler version and the set of packages +# to be used for project dependencies. For example: +# +# resolver: lts-3.5 +# resolver: nightly-2015-09-21 +# resolver: ghc-7.10.2 +# resolver: ghcjs-0.1.0_ghc-7.10.2 +# resolver: +# name: custom-snapshot +# location: "./custom-snapshot.yaml" +resolver: lts-11.9 + +# User packages to be built. +# Various formats can be used as shown in the example below. +# +# packages: +# - some-directory +# - https://example.com/foo/bar/baz-0.0.2.tar.gz +# - location: +# git: https://github.com/commercialhaskell/stack.git +# commit: e7b331f14bcffb8367cd58fbfc8b40ec7642100a +# - location: https://github.com/commercialhaskell/stack/commit/e7b331f14bcffb8367cd58fbfc8b40ec7642100a +# extra-dep: true +# subdirs: +# - auto-update +# - wai +# +# A package marked 'extra-dep: true' will only be built if demanded by a +# non-dependency (i.e. a user package), and its test suites and benchmarks +# will not be run. This is useful for tweaking upstream packages. +packages: +- . +# Dependency packages to be pulled from upstream that are not in the resolver +# (e.g., acme-missiles-0.3) +extra-deps: +- brittany-0.10.0.0 +- extensible-effects-3.0.0.0 + +# Override default flag values for local packages and extra-deps +# flags: {} + +# Extra package databases containing global packages +# extra-package-dbs: [] + +# Control whether we use the GHC we find on the path +# system-ghc: true +# +# Require a specific version of stack, using version ranges +# require-stack-version: -any # Default +require-stack-version: ">=1.6" +# +# Override the architecture used by stack, especially useful on Windows +# arch: i386 +# arch: x86_64 +# +# Extra directories used by stack for building +# extra-include-dirs: [/path/to/dir] +# extra-lib-dirs: [/path/to/dir] +# +# Allow a newer minor version of GHC than the snapshot specifies +# compiler-check: newer-minor diff --git a/test/Common.hs b/test/Common.hs --- a/test/Common.hs +++ b/test/Common.hs @@ -1,38 +1,44 @@ -module Common where +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 +import Control.Concurrent.STM +import Control.Eff.Concurrent.Process +import Control.Eff +import Control.Eff.Extend +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) +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 + :: (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))) + (\logCFactory -> k + (return + (\e -> do + logC <- logCFactory + doSchedule e logC -- noLogger + ) + ) + ) testLogC :: IO (LogChannel String) testLogC = filterLogChannel (\m -> take (length logPrefix) m == logPrefix) - <$> - forkLogger 1 (putStrLn) Nothing + <$> forkLogger 1 (putStrLn) Nothing testLogJoin :: LogChannel String -> IO () testLogJoin = joinLogChannel Nothing @@ -43,29 +49,39 @@ logPrefix :: String logPrefix = "[TEST] " -untilShutdown - :: Member t r => t (ResumeProcess v) -> Eff r () +untilShutdown :: Member t r => t (ResumeProcess v) -> Eff r () untilShutdown pa = do r <- send pa case r of ShutdownRequested -> return () - _ -> untilShutdown pa + _ -> 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 +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 -> 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 +applySchedulerFactory + :: forall r + . (Member (Logs String) r, SetMember Lift (Lift IO) r) + => IO (Eff (Process r ': r) () -> IO ()) + -> Eff (Process r ': r) () + -> IO () +applySchedulerFactory factory procAction = do + scheduler <- factory + scheduler procAction diff --git a/test/ForkIOScheduler.hs b/test/ForkIOScheduler.hs --- a/test/ForkIOScheduler.hs +++ b/test/ForkIOScheduler.hs @@ -4,14 +4,13 @@ import Control.Exception import Control.Concurrent import Control.Concurrent.STM -import Control.Eff +import Control.Eff.Loop +import Control.Eff.Extend import Control.Eff.Lift import Control.Eff.Concurrent.Process import Control.Eff.Concurrent.Process.ForkIOScheduler as Scheduler -import Control.Monad ( void - , forever - ) +import Control.Monad ( void ) import Test.Tasty import Test.Tasty.HUnit import Data.Dynamic @@ -30,7 +29,7 @@ aVar <- newEmptyTMVarIO lc <- testLogC Scheduler.defaultMainWithLogChannel lc $ do - p1 <- spawn $ forever busyEffect + p1 <- spawn $ foreverCheap busyEffect lift (threadDelay 1000) void $ spawn $ do lift (threadDelay 1000) @@ -104,7 +103,7 @@ (do void (spawn - (forever + (foreverCheap (void (sendMessage forkIoScheduler 1000 (toDyn "test"))) ) ) diff --git a/test/Interactive.hs b/test/Interactive.hs new file mode 100644 --- /dev/null +++ b/test/Interactive.hs @@ -0,0 +1,36 @@ +module Interactive +where + +import Control.Concurrent +import Control.Eff +import Control.Eff.Lift +import Control.Eff.Concurrent.Process +import Control.Eff.Concurrent.Process.Interactive +import qualified Control.Eff.Concurrent.Process.SingleThreadedScheduler + as SingleThreaded +import qualified Control.Eff.Concurrent.Process.ForkIOScheduler + as ForkIOScheduler +import Test.Tasty +import Test.Tasty.HUnit +import Common + +test_interactive :: TestTree +test_interactive = setTravisTestOptions $ testGroup + "Interactive" + [ testGroup "SingleThreadedScheduler" $ allTests SingleThreaded.defaultMain + , testGroup "ForkIOScheduler" $ allTests ForkIOScheduler.defaultMain + ] + +allTests :: SetMember Lift (Lift IO) r => (Eff (ConsProcess r) () -> IO ()) -> [TestTree] +allTests scheduler = + [ happyCaseTest scheduler + ] + +happyCaseTest + :: SetMember Lift (Lift IO) r => (Eff (ConsProcess r) () -> IO ()) -> TestTree +happyCaseTest scheduler = + testCase "start, wait and stop interactive scheduler" $ do + s <- forkInteractiveScheduler scheduler + threadDelay 100000 + killInteractiveScheduler s + return () diff --git a/test/LoopTests.hs b/test/LoopTests.hs new file mode 100644 --- /dev/null +++ b/test/LoopTests.hs @@ -0,0 +1,95 @@ +module LoopTests + ( test_loopTests + , test_loopWithLeaksTests + ) +where + +import Control.DeepSeq +import Control.Eff +import Control.Eff.Concurrent +import Control.Eff.State.Strict +import Control.Monad +import Test.Tasty +import Test.Tasty.HUnit +import Common +import Control.Eff.Concurrent.Process.SingleThreadedScheduler + as Scheduler + +test_loopTests :: TestTree +test_loopTests + = let soMany = 1000000 + 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" + $ do + res <- + Scheduler.scheduleIOWithLogging + ($! (putStrLn . (">>> " ++))) + $ do + me <- self SP + spawn_ + (foreverCheap $ sendMessageAs SP me () + ) + replicateCheapM_ + soMany + (void (receiveMessageAs @() SP)) + + res @=? Right () + ] + + +test_loopWithLeaksTests :: TestTree +test_loopWithLeaksTests + = let soMany = 1000000 + 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 + res <- + Scheduler.scheduleIOWithLogging + ($! (putStrLn . (">>> " ++))) + $ do + 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 @@ -13,13 +13,11 @@ import qualified Control.Eff.Concurrent.Process.SingleThreadedScheduler as SingleThreaded import Control.Eff +import Control.Eff.Extend import Control.Eff.Log +import Control.Eff.Loop import Control.Eff.Lift -import Control.Monad ( void - , replicateM - , forever - , when - ) +import Control.Monad import Test.Tasty import Test.Tasty.HUnit import Common @@ -33,10 +31,16 @@ test_singleThreaded :: TestTree test_singleThreaded = setTravisTestOptions (withTestLogC - (const . SingleThreaded.defaultMain) + (\e logC -> + -- void (runLift (logToChannel logC (SingleThreaded.schedule (return ()) e))) + let runEff :: Eff '[Logs String, Lift IO] a -> IO a + runEff = runLift . logToChannel logC + in void $ SingleThreaded.scheduleM runEff yield e + ) (\factory -> testGroup "SingleThreadedScheduler" [allTests factory]) ) + allTests :: forall r . (Member (Logs String) r, SetMember Lift (Lift IO) r) @@ -51,9 +55,45 @@ , concurrencyTests schedulerFactory , exitTests schedulerFactory , pingPongTests schedulerFactory + , yieldLoopTests schedulerFactory ] ) +yieldLoopTests + :: forall r + . (Member (Logs String) r, SetMember Lift (Lift IO) r) + => IO (Eff (Process r ': r) () -> IO ()) + -> TestTree +yieldLoopTests schedulerFactory + = let maxN = 100000 + in + setTravisTestOptions + (testGroup + "yield tests" + [ testCase + "yield many times (replicateM_)" + (applySchedulerFactory schedulerFactory + (replicateM_ maxN (yieldProcess SP)) + ) + , testCase + "yield many times (forM_)" + (applySchedulerFactory + schedulerFactory + (forM_ [1 :: Int .. maxN] (\_ -> yieldProcess SP)) + ) + , testCase + "construct an effect with an exit first, followed by many yields" + (applySchedulerFactory + schedulerFactory + (do + void (exitNormally SP) + replicateM_ 1000000000000 (yieldProcess SP) + ) + ) + ] + ) + + data Ping = Ping ProcessId data Pong = Pong deriving (Eq, Show) @@ -68,7 +108,7 @@ [ testCase "ping pong a message between two processes, both don't yield" $ applySchedulerFactory schedulerFactory $ do - let pongProc = forever $ do + let pongProc = foreverCheap $ do Ping pinger <- receiveMessageAs SP sendMessageAs SP pinger Pong pingProc ponger parent = do @@ -85,7 +125,7 @@ $ applySchedulerFactory schedulerFactory $ do yieldProcess SP - let pongProc = forever $ do + let pongProc = foreverCheap $ do yieldProcess SP Ping pinger <- receiveMessageAs SP yieldProcess SP @@ -117,7 +157,7 @@ $ applySchedulerFactory schedulerFactory $ do pongVar <- lift newEmptyMVar - let pongProc = forever $ do + let pongProc = foreverCheap $ do Pong <- receiveMessageAs SP lift (putMVar pongVar Pong) ponger <- spawn pongProc @@ -152,7 +192,7 @@ $ do void $ raiseError px "test error" error "This should not happen" - , testCase "catch raiseError" + , testCase "catch raiseError 1" $ scheduleAndAssert schedulerFactory $ \assertEff -> catchRaisedError px @@ -192,7 +232,7 @@ void (exitWithError px (show i ++ " died")) assertEff "this should not be reached" False else - forever + foreverCheap (void (sendMessage px 888 (toDyn "test message to 888")) ) ) @@ -252,19 +292,19 @@ m <- receiveMessage px void (sendMessage px me m) ) - child2 <- spawn (forever (void (sendMessage px 888 (toDyn "")))) + child2 <- spawn (foreverCheap (void (sendMessage px 888 (toDyn "")))) True <- sendMessageChecked px child1 (toDyn "test") i <- receiveMessageAs px True <- sendShutdownChecked px child2 assertEff "" (i == "test") - , testCase "most processes send forever" + , testCase "most processes send foreverCheap" $ scheduleAndAssert schedulerFactory $ \assertEff -> do me <- self px traverse_ (\(i :: Int) -> spawn $ do when (i `rem` 5 == 0) $ void $ sendMessage px me (toDyn i) - forever + foreverCheap $ void (sendMessage px 888 (toDyn "test message to 888")) ) [0 .. n] @@ -275,14 +315,14 @@ return j ) assertEff "" (sort oks == [0, 5 .. n]) - , testCase "most processes self forever" + , testCase "most processes self foreverCheap" $ scheduleAndAssert schedulerFactory $ \assertEff -> do me <- self px traverse_ (\(i :: Int) -> spawn $ do when (i `rem` 5 == 0) $ void $ sendMessage px me (toDyn i) - forever $ void (self px) + foreverCheap $ void (self px) ) [0 .. n] oks <- replicateM @@ -292,14 +332,14 @@ return j ) assertEff "" (sort oks == [0, 5 .. n]) - , testCase "most processes sendShutdown forever" + , testCase "most processes sendShutdown foreverCheap" $ scheduleAndAssert schedulerFactory $ \assertEff -> do me <- self px traverse_ (\(i :: Int) -> spawn $ do when (i `rem` 5 == 0) $ void $ sendMessage px me (toDyn i) - forever $ void (sendShutdown px 999) + foreverCheap $ void (sendShutdown px 999) ) [0 .. n] oks <- replicateM @@ -309,7 +349,7 @@ return j ) assertEff "" (sort oks == [0, 5 .. n]) - , testCase "most processes spawn forever" + , testCase "most processes spawn foreverCheap" $ scheduleAndAssert schedulerFactory $ \assertEff -> do me <- self px @@ -317,7 +357,7 @@ (\(i :: Int) -> spawn $ do when (i `rem` 5 == 0) $ void $ sendMessage px me (toDyn i) parent <- self px - forever + foreverCheap $ void (spawn (void @@ -334,14 +374,14 @@ return j ) assertEff "" (sort oks == [0, 5 .. n]) - , testCase "most processes receive forever" + , testCase "most processes receive foreverCheap" $ scheduleAndAssert schedulerFactory $ \assertEff -> do me <- self px traverse_ (\(i :: Int) -> spawn $ do when (i `rem` 5 == 0) $ void $ sendMessage px me (toDyn i) - forever $ void (receiveMessage px) + foreverCheap $ void (receiveMessage px) ) [0 .. n] oks <- replicateM @@ -383,7 +423,7 @@ $ do tid <- lift $ myThreadId lift $ atomically $ putTMVar tidVar tid - forever busyEffect + foreverCheap busyEffect atomically $ putTMVar schedulerDoneVar () tid <- atomically $ takeTMVar tidVar threadDelay 1000 @@ -408,7 +448,7 @@ [ testCase ("a child process, busy with " ++ busyWith) $ applySchedulerFactory schedulerFactory $ do - void $ spawn $ forever busyEffect + void $ spawn $ foreverCheap busyEffect lift (threadDelay 10000) | (busyWith, busyEffect) <- [ ("receiving", void (send (ReceiveMessage @r))) @@ -432,7 +472,7 @@ ) $ scheduleAndAssert schedulerFactory $ \assertEff -> do - p1 <- spawn $ forever busyEffect + p1 <- spawn $ foreverCheap busyEffect lift (threadDelay 1000) void $ spawn $ do lift (threadDelay 1000) diff --git a/test/SingleThreadedScheduler.hs b/test/SingleThreadedScheduler.hs --- a/test/SingleThreadedScheduler.hs +++ b/test/SingleThreadedScheduler.hs @@ -1,15 +1,40 @@ module SingleThreadedScheduler where +import Control.Eff.Loop import Control.Eff.Concurrent.Process import Control.Eff.Concurrent.Process.SingleThreadedScheduler as Scheduler -import Control.Monad ( void - ) +import Control.Monad import Test.Tasty import Test.Tasty.HUnit import Data.Dynamic import Common + +test_pureScheduler :: TestTree +test_pureScheduler = setTravisTestOptions $ testGroup + "Pure Scheduler" + [ testCase "two processes, each calculate and report back to main process" + $ Right (42 :: Int) + @=? Scheduler.schedulePure + (do + adderChild <- spawn $ do + (from, arg1, arg2) <- receiveMessageAs SP + sendMessageAs SP from ((arg1 + arg2) :: Int) + foreverCheap $ void $ receiveMessage SP + + multChild <- spawn $ do + (from, arg1, arg2) <- receiveMessageAs SP + sendMessageAs SP from ((arg1 * arg2) :: Int) + + me <- self SP + sendMessageAs SP adderChild (me, 3 :: Int, 4 :: Int) + x <- receiveMessageAs @Int SP + sendMessageAs SP multChild (me, x, 6 :: Int) + receiveMessageAs @Int SP + ) + ] + test_mainProcessSpawnsAChildAndExitsNormally :: TestTree test_mainProcessSpawnsAChildAndExitsNormally = setTravisTestOptions