extensible-effects-concurrent 0.3.0.2 → 0.4.0.0

raw patch · 27 files changed

+1059/−606 lines, 27 filesdep +deepseqdep ~QuickCheckdep ~extensible-effectsbinary-addedPVP ok

version bump matches the API change (PVP)

Dependencies added: deepseq

Dependency ranges changed: QuickCheck, extensible-effects

API changes (from Hackage documentation)

Files

• ChangeLog.md +10/−0
• docs/extensible-effects-concurrent-test-Loop-WITH-space-leaks.png binary
• docs/extensible-effects-concurrent-test-Loop-without-space-leaks.png binary
• extensible-effects-concurrent.cabal +32/−16
• src/Control/Eff/Concurrent.hs +127/−0
• src/Control/Eff/Concurrent/Api.hs +3/−16
• src/Control/Eff/Concurrent/Api/Client.hs +5/−20
• src/Control/Eff/Concurrent/Api/Internal.hs +0/−15
• src/Control/Eff/Concurrent/Api/Observer.hs +6/−26
• src/Control/Eff/Concurrent/Api/Server.hs +2/−18
• src/Control/Eff/Concurrent/Examples.hs +2/−22
• src/Control/Eff/Concurrent/Examples2.hs +4/−25
• src/Control/Eff/Concurrent/Process.hs +23/−32
• src/Control/Eff/Concurrent/Process/ForkIOScheduler.hs +8/−23
• src/Control/Eff/Concurrent/Process/Interactive.hs +43/−44
• src/Control/Eff/Concurrent/Process/SingleThreadedScheduler.hs +195/−143
• src/Control/Eff/InternalExtra.hs +0/−17
• src/Control/Eff/Log.hs +125/−115
• src/Control/Eff/Loop.hs +53/−0
• stack.nightly-2018-05-29.yaml +68/−0
• stack.yaml +68/−0
• test/Common.hs +57/−41
• test/ForkIOScheduler.hs +5/−6
• test/Interactive.hs +36/−0
• test/LoopTests.hs +95/−0
• test/ProcessBehaviourTestCases.hs +65/−25
• test/SingleThreadedScheduler.hs +27/−2

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

binary file changed (absent → 83907 bytes)

binary file changed (absent → 87117 bytes)

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

@@ -0,0 +1,53 @@+-- | In rare occasions GHC optimizes innocent looking loops into space-leaking monsters.+-- See the discussion here: <https://ghc.haskell.org/trac/ghc/ticket/13080 GHC issue 13080> for more+-- details, or <https://ro-che.info/articles/2017-01-10-nested-loop-space-leak this blog post about space leaks in nested loops>+--+-- 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

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

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

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

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

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

@@ -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 ()+            ]

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

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