extensible-effects-concurrent 0.11.1 → 0.12.0
raw patch · 25 files changed
+3473/−1466 lines, 25 filesdep +exceptionsdep +safe-exceptionsdep −enclosed-exceptionsdep ~asyncdep ~basedep ~extensible-effects
Dependencies added: exceptions, safe-exceptions
Dependencies removed: enclosed-exceptions
Dependency ranges changed: async, base, extensible-effects
Files
- ChangeLog.md +29/−3
- README.md +3/−3
- examples/example-1/Main.hs +16/−24
- examples/example-2/Main.hs +5/−1
- examples/example-4/Main.hs +18/−6
- extensible-effects-concurrent.cabal +10/−6
- src/Control/Eff/Concurrent.hs +80/−29
- src/Control/Eff/Concurrent/Api/Client.hs +25/−72
- src/Control/Eff/Concurrent/Api/Observer.hs +34/−6
- src/Control/Eff/Concurrent/Api/Observer/Queue.hs +16/−19
- src/Control/Eff/Concurrent/Api/Server.hs +45/−24
- src/Control/Eff/Concurrent/Process.hs +1208/−652
- src/Control/Eff/Concurrent/Process/ForkIOScheduler.hs +385/−156
- src/Control/Eff/Concurrent/Process/Interactive.hs +15/−7
- src/Control/Eff/Concurrent/Process/SingleThreadedScheduler.hs +413/−185
- src/Control/Eff/Concurrent/Process/Timer.hs +184/−0
- src/Control/Eff/ExceptionExtra.hs +220/−9
- src/Control/Eff/Log/Handler.hs +177/−35
- stack.yaml +1/−4
- test/Common.hs +14/−14
- test/ForkIOScheduler.hs +64/−18
- test/Interactive.hs +10/−8
- test/LoopTests.hs +4/−4
- test/ProcessBehaviourTestCases.hs +481/−162
- test/SingleThreadedScheduler.hs +16/−19
ChangeLog.md view
@@ -1,5 +1,31 @@ # Changelog for extensible-effects-concurrent +## 0.12.0++- Add implicit SchedulerProxy+- Add flushMessages+- Add receiving with timeout+- Add process `Link`ing and `Monitoring`.+- Make the distinction between recoverable and non-recoverable exit explicit in+ the type parameter of `ExitReason`, and introduce `interruptXXXX`+ functions in addition to `shutdownXXXX` functions, to throw recoverable exits.+- Merge `ShutdownRequest` and `ExitReason`+- Rename `receiveLoopSuchThat` to `receiveSelectedLoop`+- Pass the exit reason to the callback passed to `receiveSelectedLoop`+- Rename `receiveMessage` to `receiveAnyMessage`+- Rename `receiveAnyLoop` to `receiveAnyLoop`+- Pass the exit reason to the callback passed to `receiveAnyLoop`+- Rename `receiveMessage` to `receiveAnyMessage`+- Rename `receiveMessageAs` to `receiveMessage`+- Rename `receiveLoop` to `receiveLoop`+- Pass the exit reason to the callback passed to `receiveAnyLoop`+- Remove `SchedulerShuttingDown`+- Improve logging for exceptions in `ForkIOScheduler`+- Fix a bug in the logging system that caused all log filters to be forgotten+ when using unliftings such as `MonadBaseControl`, `MonadThrow`, `MonadCatch`+ and `MonadMask`+- Fix the scheduler schutdown to not always run into the cancellation timeout+ ## 0.11.1 - Fix a compilation error@@ -31,7 +57,7 @@ ## 0.9.0 - Make `ForkIOScheduler` faster and more robust-- Add `ProcessExitReason`+- Add `ExitReason` - Add `ProcessState` - Add `ShutdownRequest` type - Rewrite logging to be a `Reader` of a `LogWriter`@@ -136,7 +162,7 @@ - Add support for running and interacting with a scheduler and it's processes from IO, for example from ghci-- Rename `yieldProcess` to `executeAndResume`+- Rename `yieldProcess` to `executeAndResumeOrExit` - Add an actual `yieldProcess`, that behaves like `yield` - Change the return type of function to `()` where applicable to avoid all these `_ <- sendMessage...` or `void $ sendMessage`@@ -170,7 +196,7 @@ - Add initial test suite - Fix shutdown error in `ForkIoScheduler` - Rename `Dispatcher` to `Scheduler`-- Add `receiveLoop` function to `Process`+- Add `receiveAnyLoop` function to `Process` - Change `Api.Server` `serve` to loop instead of handling just one request - Allow combining multiple `ApiHandler` such that one process can handle multiple APIs
README.md view
@@ -41,20 +41,20 @@ person <- spawn (do logInfo "I am waiting for someone to ask me..."- WhoAreYou replyPid <- receiveMessageAs px+ WhoAreYou replyPid <- receiveMessage px sendMessageAs px replyPid "Alice" logInfo (show replyPid ++ " just needed to know it.") ) me <- self px sendMessageAs px person (WhoAreYou me)- personName <- receiveMessageAs px+ personName <- receiveMessage px logInfo ("I just met " ++ personName) ``` **Running** this example causes this output: (_not entirely true because of async logging, but true enough_) -```+```text 2018-11-05T10:50:42 DEBUG scheduler loop entered ForkIOScheduler.hs line 131 2018-11-05T10:50:42 DEBUG !1 [ThreadId 11] enter process ForkIOScheduler.hs line 437 2018-11-05T10:50:42 NOTICE !1 [ThreadId 11] ++++++++ main process started ++++++++ ForkIOScheduler.hs line 394
examples/example-1/Main.hs view
@@ -30,14 +30,16 @@ main = defaultMain (example forkIoScheduler) mainProcessSpawnsAChildAndReturns- :: (HasCallStack, SetMember Process (Process q) r)+ :: (HasCallStack, SetMember Process (Process q) r, Member Interrupts r) => SchedulerProxy q -> Eff r ()-mainProcessSpawnsAChildAndReturns px = void (spawn (void (receiveMessage px)))+mainProcessSpawnsAChildAndReturns px =+ void (spawn (void (receiveAnyMessage px))) example :: ( HasCallStack , SetMember Process (Process q) r+ , Member Interrupts r , HasLogging IO r , HasLogging IO q )@@ -66,33 +68,37 @@ go ('q' : _) -> logInfo "Done." _ -> do- res <- ignoreProcessError px (callRegistered px (SayHello x))+ res <- callRegistered px (SayHello x) logInfo ("Result: " ++ show res) go registerServer server go testServerLoop :: forall r q- . (HasCallStack, SetMember Process (Process q) r, HasLogging IO q)+ . ( HasCallStack+ , SetMember Process (Process q) r+ , HasLogging IO q+ , Member Interrupts r+ ) => SchedulerProxy q -> Eff r (Server TestApi) testServerLoop px = spawnServer px $ apiHandler handleCastTest handleCallTest handleTerminateTest where handleCastTest- :: Api TestApi 'Asynchronous -> Eff (Process q ': q) ApiServerCmd+ :: Api TestApi 'Asynchronous -> Eff (InterruptableProcess q) ApiServerCmd handleCastTest (Shout x) = do me <- self px logInfo (show me ++ " Shouting: " ++ x) return HandleNextRequest handleCallTest :: Api TestApi ( 'Synchronous x)- -> (x -> Eff (Process q ': q) ())- -> Eff (Process q ': q) ApiServerCmd+ -> (x -> Eff (InterruptableProcess q) ())+ -> Eff (InterruptableProcess q) ApiServerCmd handleCallTest (SayHello "e1") _reply = do me <- self px logInfo (show me ++ " raising an error")- raiseError px "No body loves me... :,("+ interrupt (ProcessError "No body loves me... :,(") handleCallTest (SayHello "e2") _reply = do me <- self px logInfo (show me ++ " throwing a MyException ")@@ -107,21 +113,7 @@ me <- self px logInfo (show me ++ " stopping me") void (reply False)- return (StopApiServer Nothing)- handleCallTest (SayHello "xxx") reply = do- me <- self px- logInfo (show me ++ " stopping me with xxx")- void (reply False)- return (StopApiServer (Just "xxx"))- handleCallTest (SayHello "die") reply = do- me <- self px- logInfo (show me ++ " throwing and catching ")- catchRaisedError- px- (\er -> logInfo ("WOW: " ++ show er ++ " - No. This is wrong!"))- (raiseError px "No body loves me... :,(")- void (reply True)- return HandleNextRequest+ return (StopApiServer (ProcessError "test error")) handleCallTest (SayHello x) reply = do me <- self px logInfo (show me ++ " Got Hello: " ++ x)@@ -140,4 +132,4 @@ handleTerminateTest msg = do me <- self px logInfo (show me ++ " is exiting: " ++ show msg)- maybe (exitNormally px) (exitWithError px) msg+ logProcessExit msg
examples/example-2/Main.hs view
@@ -35,6 +35,7 @@ :: ( SetMember Process (Process q) r , Member (Logs LogMessage) q , Member (Logs LogMessage) r+ , Member Interrupts r ) => SchedulerProxy q -> Eff r (Server (CallbackObserver Counter))@@ -53,6 +54,7 @@ , SetMember Process (Process q) r , Member (Logs LogMessage) q , Member (Logs LogMessage) r+ , Member Interrupts r ) => SchedulerProxy q -> ApiHandler Counter r@@ -118,6 +120,7 @@ :: forall r q . ( Member (Logs LogMessage) q , Member (Logs LogMessage) r+ , Member Interrupts r , SetMember Process (Process q) r ) => SchedulerProxy q@@ -132,6 +135,7 @@ :: ( SetMember Process (Process q) r , Member (Logs LogMessage) q , Member (Logs LogMessage) r+ , Member Interrupts r , q <:: r ) => SchedulerProxy q@@ -171,4 +175,4 @@ cnt cntServer1 cast px cntServer2 Inc cnt cntServer2- void $ sendShutdown px pid2 (ExitWithError "test test test")+ void $ sendShutdown px pid2 (NotRecovered (ProcessError "test test test"))
examples/example-4/Main.hs view
@@ -5,6 +5,7 @@ import Control.Eff.Concurrent import Data.Dynamic import Control.Concurrent+import Control.Applicative import Control.DeepSeq main :: IO ()@@ -17,18 +18,29 @@ -- The SchedulerProxy paremeter contains the effects of a specific scheduler -- implementation. -newtype WhoAreYou = WhoAreYou ProcessId deriving (Typeable, NFData)+newtype WhoAreYou = WhoAreYou ProcessId deriving (Typeable, NFData, Show) -firstExample :: (HasLogging IO q) => SchedulerProxy q -> Eff (Process q ': q) ()+firstExample+ :: (HasLogging IO q) => SchedulerProxy q -> Eff (InterruptableProcess q) () firstExample px = do person <- spawn (do logInfo "I am waiting for someone to ask me..."- WhoAreYou replyPid <- receiveMessageAs px- sendMessageAs px replyPid "Alice"+ WhoAreYou replyPid <- receiveMessage px+ sendMessage px replyPid "Alice" logInfo (show replyPid ++ " just needed to know it.") ) me <- self px- sendMessageAs px person (WhoAreYou me)- personName <- receiveMessageAs px+ sendMessage px person (WhoAreYou me)+ personName <- receiveMessage px logInfo ("I just met " ++ personName)+++selectInt :: MessageSelector Int+selectInt = selectMessage++selectString :: MessageSelector String+selectString = selectMessage++selectIntOrString :: MessageSelector (Either Int String)+selectIntOrString = Left <$> selectInt <|> Right <$> selectString
extensible-effects-concurrent.cabal view
@@ -1,5 +1,5 @@ name: extensible-effects-concurrent-version: 0.11.1+version: 0.12.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@@ -35,10 +35,11 @@ src build-depends: async >= 2.2 && <3,- base >=4.7 && <4.12,+ base >=4.7 && <5, data-default >= 0.7 && < 0.8, deepseq >= 1.4 && < 1.5,- enclosed-exceptions >= 1.0 && < 1.1,+ exceptions >= 0.10 && < 0.11,+ safe-exceptions >= 0.1 && < 0.2, filepath >= 1.4 && < 1.5, time >= 1.8 && < 1.9, mtl >= 2.2 && < 2.3,@@ -48,7 +49,7 @@ parallel >= 3.2 && < 3.3, process >= 1.6 && < 1.7, monad-control >= 1.0 && < 1.1,- extensible-effects >= 3.1 && <4,+ extensible-effects >= 3.1.0.2 && <4, stm >= 2.4.5 && <2.6, transformers-base >= 0.4 && < 0.5 autogen-modules: Paths_extensible_effects_concurrent@@ -64,6 +65,7 @@ Control.Eff.Concurrent.Api.Client, Control.Eff.Concurrent.Api.Server, Control.Eff.Concurrent.Process,+ Control.Eff.Concurrent.Process.Timer, Control.Eff.Concurrent.Process.ForkIOScheduler, Control.Eff.Concurrent.Process.Interactive, Control.Eff.Concurrent.Process.SingleThreadedScheduler,@@ -97,6 +99,7 @@ TypeInType, TypeOperators, ViewPatterns+ other-extensions: ImplicitParams default-language: Haskell2010 ghc-options: -Wall -fno-full-laziness @@ -220,10 +223,11 @@ , SingleThreadedScheduler ghc-options: -Wall -threaded -fno-full-laziness -rtsopts -with-rtsopts=-N build-depends:- base >=4.7 && <4.12+ async+ , base >=4.7 && <4.12 , data-default , extensible-effects-concurrent- , extensible-effects >= 3+ , extensible-effects >= 3.1.0.2 && < 3.2 , tasty , tasty-discover , tasty-hunit
src/Control/Eff/Concurrent.hs view
@@ -5,6 +5,9 @@ -- * Concurrent Processes with Message Passing Concurrency module Control.Eff.Concurrent.Process ,+ -- * Timers and Timeouts+ module Control.Eff.Concurrent.Process.Timer+ , -- * Data Types and Functions for APIs (aka Protocols) module Control.Eff.Concurrent.Api ,@@ -36,12 +39,30 @@ ) where -import Control.Eff.Concurrent.Process ( ProcessId(..)+import Control.Eff.Concurrent.Process ( Process(..)+ , ProcessId(..) , fromProcessId- , Process(..) , ConsProcess , ResumeProcess(..) , SchedulerProxy(..)+ , HasScheduler+ , getSchedulerProxy+ , withSchedulerProxy+ , thisSchedulerProxy+ , ProcessState(..)+ , yieldProcess+ , sendMessage+ , sendAnyMessage+ , sendShutdown+ , sendInterrupt+ , makeReference+ , receiveAnyMessage+ , receiveMessage+ , receiveSelectedMessage+ , flushMessages+ , receiveAnyLoop+ , receiveLoop+ , receiveSelectedLoop , MessageSelector ( runMessageSelector )@@ -56,35 +77,63 @@ , selectDynamicMessage , selectDynamicMessageLazy , selectAnyMessageLazy- , ProcessState(..)- , ProcessExitReason(..)- , ShutdownRequest(..)- , thisSchedulerProxy- , executeAndCatch- , executeAndResume- , yieldProcess- , sendMessage- , sendMessageAs- , sendMessageChecked+ , self+ , isProcessAlive , spawn , spawn_- , receiveMessage- , receiveMessageAs- , receiveSelectedMessage- , receiveLoop- , receiveLoopAs- , receiveLoopSuchThat- , self- , sendShutdown- , sendShutdownChecked- , makeReference- , exitWithError+ , spawnLink+ , spawnRaw+ , spawnRaw_+ , exitBecause , exitNormally- , raiseError- , catchRaisedError- , ignoreProcessError+ , exitWithError+ , linkProcess+ , unlinkProcess+ , monitor+ , demonitor+ , ProcessDown(..)+ , selectProcessDown+ , becauseProcessIsDown+ , MonitorReference(..)+ , withMonitor+ , receiveWithMonitor+ , provideInterruptsShutdown+ , handleInterrupts+ , exitOnInterrupt+ , logInterrupts+ , provideInterrupts+ , mergeEitherInterruptAndExitReason+ , interrupt+ , executeAndResume+ , executeAndResumeOrExit+ , executeAndResumeOrThrow+ , ExitReason(..)+ , ExitRecovery(..)+ , InterruptReason+ , Interrupts+ , InterruptableProcess+ , ExitSeverity(..)+ , SomeExitReason(SomeExitReason)+ , toExitRecovery+ , isRecoverable+ , toExitSeverity+ , isBecauseDown+ , isCrash+ , toCrashReason+ , fromSomeExitReason , logProcessExit )+import Control.Eff.Concurrent.Process.Timer+ ( Timeout(fromTimeoutMicros)+ , TimerReference()+ , TimerElapsed(fromTimerElapsed)+ , sendAfter+ , startTimer+ , selectTimerElapsed+ , receiveAfter+ , receiveSelectedAfter+ )+ import Control.Eff.Concurrent.Api ( Api , Synchronicity(..) , Server(..)@@ -94,11 +143,9 @@ ) import Control.Eff.Concurrent.Api.Client ( cast- , castChecked , call , castRegistered , callRegistered- , callRegisteredA , ServesApi , registerServer , whereIsServer@@ -161,12 +208,16 @@ , defaultMain , defaultMainWithLogChannel , ProcEff+ , InterruptableProcEff , SchedulerIO , HasSchedulerIO , forkIoScheduler ) import Control.Eff.Concurrent.Process.SingleThreadedScheduler- ( schedulePure )+ ( schedulePure+ , defaultMainSingleThreaded+ , singleThreadedIoScheduler+ ) import Control.Eff.Log import Control.Eff.Loop
src/Control/Eff/Concurrent/Api/Client.hs view
@@ -4,12 +4,10 @@ module Control.Eff.Concurrent.Api.Client ( -- * Calling APIs directly cast- , castChecked , call -- * Server Process Registration , castRegistered , callRegistered- , callRegisteredA , ServesApi , ServerReader , whereIsServer@@ -17,45 +15,24 @@ ) where -import Control.Applicative import Control.Eff import Control.Eff.Reader.Strict import Control.Eff.Concurrent.Api import Control.Eff.Concurrent.Api.Internal import Control.Eff.Concurrent.Process-import Data.Dynamic-import Data.Typeable ( Typeable- , typeRep- )+import Data.Typeable ( Typeable ) import Control.DeepSeq import GHC.Stack -- | Send an 'Api' request that has no return value and return as fast as -- possible. The type signature enforces that the corresponding 'Api' clause is--- 'Asynchronous'. Return @True@ if the message was sent to the process. Note--- that this is totally not the same as that the request was successfully--- handled. If that is important, use 'call' instead.-castChecked- :: forall r q o- . ( HasCallStack- , SetMember Process (Process q) r- , Typeable o- , Typeable (Api o 'Asynchronous)- )- => SchedulerProxy q- -> Server o- -> Api o 'Asynchronous- -> Eff r Bool-castChecked px (Server pid) castMsg =- sendMessageChecked px pid (toDyn $! (Cast $! castMsg))---- | Send an 'Api' request that has no return value and return as fast as--- possible. The type signature enforces that the corresponding 'Api' clause is--- 'Asynchronous'.+-- 'Asynchronous'. The operation never fails, if it is important to know if the+-- message was delivered, use 'call' instead. cast :: forall r q o . ( HasCallStack , SetMember Process (Process q) r+ , Member Interrupts r , Typeable o , Typeable (Api o 'Asynchronous) )@@ -63,9 +40,7 @@ -> Server o -> Api o 'Asynchronous -> Eff r ()-cast px toServer apiRequest = do- _ <- castChecked px toServer apiRequest- return ()+cast px (Server pid) castMsg = sendMessage px pid (Cast $! castMsg) -- | Send an 'Api' request and wait for the server to return a result value. --@@ -74,40 +49,31 @@ call :: forall result api r q . ( SetMember Process (Process q) r+ , Member Interrupts r , Typeable api , Typeable (Api api ( 'Synchronous result)) , Typeable result , HasCallStack , NFData result+ , Show result ) => SchedulerProxy q -> Server api -> Api api ( 'Synchronous result) -> Eff r result-call px (Server pidInt) req = do+call px (Server pidInternal) req = do fromPid <- self px callRef <- makeReference px let requestMessage = Call callRef fromPid $! req- wasSent <- sendMessageChecked px pidInt (toDyn $! requestMessage)- if wasSent- then- let selectResult :: MessageSelector result- selectResult =- let extractResult :: Response api result -> Maybe result- extractResult (Response _pxResult callRefMsg result) =- if callRefMsg == callRef then Just result else Nothing- in selectMessageWith extractResult- in receiveSelectedMessage px selectResult- else raiseError- px- ( "failed to send a call for: '"- ++ show (typeRep requestMessage)- ++ "' to: "- ++ show pidInt- ++ " with call-ref: "- ++ show callRef- )-+ sendMessage px pidInternal requestMessage+ let selectResult :: MessageSelector result+ selectResult =+ let extractResult :: Response api result -> Maybe result+ extractResult (Response _pxResult callRefMsg result) =+ if callRefMsg == callRef then Just result else Nothing+ in selectMessageWith extractResult+ rres <- receiveWithMonitor px pidInternal selectResult+ either (interrupt . becauseProcessIsDown) return rres -- | 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@@ -136,7 +102,13 @@ -- | Like 'call' but take the 'Server' from the reader provided by -- 'registerServer'. callRegistered- :: (Typeable reply, ServesApi o r q, HasCallStack, NFData reply)+ :: ( Typeable reply+ , ServesApi o r q+ , HasCallStack+ , NFData reply+ , Show reply+ , Member Interrupts r+ ) => SchedulerProxy q -> Api o ( 'Synchronous reply) -> Eff r reply@@ -144,29 +116,10 @@ serverPid <- whereIsServer call px serverPid method --- | Like 'callRegistered' but also catch errors raised if e.g. the server--- crashed. By allowing 'Alternative' instances to contain the reply,--- application level errors can be combined with errors rising from inter--- process communication.-callRegisteredA- :: forall r q o f reply- . ( Alternative f- , Typeable f- , Typeable reply- , ServesApi o r q- , HasCallStack- , NFData (f reply)- )- => SchedulerProxy q- -> Api o ( 'Synchronous (f reply))- -> Eff r (f reply)-callRegisteredA px method =- catchRaisedError px (const (return (empty @f))) (callRegistered px method)- -- | Like 'cast' but take the 'Server' from the reader provided by -- 'registerServer'. castRegistered- :: (Typeable o, ServesApi o r q, HasCallStack)+ :: (Typeable o, ServesApi o r q, HasCallStack, Member Interrupts r) => SchedulerProxy q -> Api o 'Asynchronous -> Eff r ()
src/Control/Eff/Concurrent/Api/Observer.hs view
@@ -59,7 +59,12 @@ -- | Send an 'Observation' to an 'Observer' notifyObserver- :: (SetMember Process (Process q) r, Observable o, Observer p o, HasCallStack)+ :: ( SetMember Process (Process q) r+ , Observable o+ , Observer p o+ , HasCallStack+ , Member Interrupts r+ ) => SchedulerProxy q -> Server p -> Server o@@ -70,7 +75,12 @@ -- | Send the 'registerObserverMessage' registerObserver- :: (SetMember Process (Process q) r, Observable o, Observer p o, HasCallStack)+ :: ( SetMember Process (Process q) r+ , Observable o+ , Observer p o+ , HasCallStack+ , Member Interrupts r+ ) => SchedulerProxy q -> Server p -> Server o@@ -80,7 +90,11 @@ -- | Send the 'forgetObserverMessage' forgetObserver- :: (SetMember Process (Process q) r, Observable o, Observer p o)+ :: ( SetMember Process (Process q) r+ , Observable o+ , Observer p o+ , Member Interrupts r+ ) => SchedulerProxy q -> Server p -> Server o@@ -106,7 +120,11 @@ -- | Send an 'Observation' to 'SomeObserver'. notifySomeObserver- :: (SetMember Process (Process q) r, Observable o, HasCallStack)+ :: ( SetMember Process (Process q) r+ , Observable o+ , HasCallStack+ , Member Interrupts r+ ) => SchedulerProxy q -> Server o -> Observation o@@ -141,7 +159,11 @@ -- | Delete an 'Observer' from the 'Observers' managed by 'manageObservers'. removeObserver- :: (SetMember Process (Process q) r, Member (ObserverState o) r, Observable o)+ :: ( SetMember Process (Process q) r+ , Member (ObserverState o) r+ , Observable o+ , Member Interrupts r+ ) => SomeObserver o -> Eff r () removeObserver = modify . over observers . Set.delete@@ -153,6 +175,7 @@ . ( Observable o , SetMember Process (Process q) r , Member (ObserverState o) r+ , Member Interrupts r ) => SchedulerProxy q -> Observation o@@ -185,10 +208,14 @@ , Show (Observation o) , Observable o , Member (Logs LogMessage) q+ , Member Interrupts r , HasCallStack ) => SchedulerProxy q- -> (Server o -> Observation o -> Eff (Process q ': q) ApiServerCmd)+ -> ( Server o+ -> Observation o+ -> Eff (InterruptableProcess q) ApiServerCmd+ ) -> Eff r (Server (CallbackObserver o)) spawnCallbackObserver px onObserve = spawnServerWithEffects px@@ -208,6 +235,7 @@ , Observable o , Member (Logs LogMessage) q , Member (Logs LogMessage) r+ , Member Interrupts r , HasCallStack ) => SchedulerProxy q
src/Control/Eff/Concurrent/Api/Observer/Queue.hs view
@@ -12,7 +12,7 @@ import Control.Concurrent.STM import Control.Eff-import Control.Eff.ExceptionExtra+import Control.Eff.ExceptionExtra ( ) import Control.Eff.Lift import Control.Eff.Concurrent.Process import Control.Eff.Log@@ -21,7 +21,7 @@ import Control.Eff.Concurrent.Api.Observer import Control.Eff.Concurrent.Api.Server import Control.Eff.Reader.Strict-import Control.Exception+import Control.Exception.Safe as Safe import Control.Monad.IO.Class import Control.Monad ( unless ) import Data.Typeable@@ -53,7 +53,6 @@ => Eff r (Observation o) readObservationQueue = do ObservationQueue q <- ask @(ObservationQueue o)- logDebug ((logPrefix (Proxy @o)) ++ " reading") liftIO (atomically (readTBQueue q)) -- | Read queued observations captured by observing a 'Server' that implements@@ -70,7 +69,6 @@ ) => Eff r (Maybe (Observation o)) tryReadObservationQueue = do- logDebug ((logPrefix (Proxy @o)) ++ " try reading") ObservationQueue q <- ask @(ObservationQueue o) liftIO (atomically (tryReadTBQueue q)) @@ -87,7 +85,6 @@ ) => Eff r [Observation o] flushObservationQueue = do- logDebug ((logPrefix (Proxy @o)) ++ " flush") ObservationQueue q <- ask @(ObservationQueue o) liftIO (atomically (flushTBQueue q)) @@ -102,8 +99,11 @@ , Observable o , HasLogging IO q , HasLogging IO r+ , Member Interrupts q+ , Member Interrupts r , Lifted IO r , HasCallStack+ , MonadCatch (Eff r) ) => SchedulerProxy q -> Int@@ -127,8 +127,11 @@ , Observable o , HasLogging IO r , HasLogging IO q+ , Member Interrupts r+ , Member Interrupts q , Lifted IO q , HasCallStack+ , MonadCatch (Eff r) ) => SchedulerProxy q -> Server o@@ -147,13 +150,7 @@ ) cbo <- spawnCallbackObserver px- (\from observation -> do- logDebug- (printf "%s enqueue observation %s from %s"- (logPrefix (Proxy @o))- (show observation)- (show from)- )+ (\_from observation -> do liftIO (atomically (writeTBQueue q observation)) return HandleNextRequest )@@ -163,13 +160,8 @@ (show cbo) ) registerObserver SchedulerProxy cbo oSvr- logDebug- (printf "%s registered at: %s" (logPrefix (Proxy @o)) (show oSvr))- logDebug (printf "%s running" (logPrefix (Proxy @o))) res <- k- logDebug (printf "%s finished" (logPrefix (Proxy @o))) forgetObserver SchedulerProxy cbo oSvr- logDebug (printf "%s unregistered" (logPrefix (Proxy @o))) sendShutdown px (_fromServer cbo) ExitNormally logDebug (printf "%s stopped observer process" (logPrefix (Proxy @o))) return res@@ -177,13 +169,18 @@ withQueue :: forall a b e- . (HasCallStack, Typeable a, Show (Observation a), HasLogging IO e)+ . ( HasCallStack+ , Typeable a+ , Show (Observation a)+ , HasLogging IO e+ , MonadCatch (Eff e)+ ) => Int -> Eff (ObservationQueueReader a ': e) b -> Eff e b withQueue queueLimit e = do q <- liftIO (newTBQueueIO queueLimit)- res <- liftTry @SomeException (runReader (ObservationQueue q) e)+ res <- Safe.tryAny (runReader (ObservationQueue q) e) rest <- liftIO (atomically (flushTBQueue q)) unless (null rest)
src/Control/Eff/Concurrent/Api/Server.hs view
@@ -34,6 +34,8 @@ import Control.Eff.Concurrent.Api import Control.Eff.Concurrent.Api.Internal import Control.Eff.Concurrent.Process+import Control.Eff.Exception+import Control.Eff.Log import Control.Lens import Data.Proxy import Data.Typeable ( Typeable@@ -74,7 +76,7 @@ -- -- The default behavior is defined in 'defaultTermination'. , _terminateCallback- :: Maybe (Maybe String -> Eff eff ())+ :: Maybe (ExitReason 'Recoverable -> Eff eff ()) } -> ApiHandler api eff @@ -93,7 +95,7 @@ -> (r -> Eff e ()) -> Eff e ApiServerCmd )- -> (Maybe String -> Eff e ())+ -> (ExitReason 'Recoverable -> Eff e ()) -> ApiHandler api e apiHandler c d e = ApiHandler { _castCallback = Just c@@ -108,7 +110,7 @@ apiHandlerForever :: (Api api 'Asynchronous -> Eff e ()) -> (forall r . Api api ( 'Synchronous r) -> (r -> Eff e ()) -> Eff e ())- -> (Maybe String -> Eff e ())+ -> (ExitReason 'Recoverable -> Eff e ()) -> ApiHandler api e apiHandlerForever c d = apiHandler (\someCast -> c someCast >> return HandleNextRequest)@@ -174,7 +176,7 @@ -- | Tell the server to exit, this will make 'serve' stop handling requests without -- exitting the process. '_terminateCallback' will be invoked with the given -- optional reason.- StopApiServer :: Maybe String -> ApiServerCmd+ StopApiServer :: ExitReason 'Recoverable -> ApiServerCmd -- SendReply :: reply -> ApiServerCmd () -> ApiServerCmd (reply -> Eff eff ()) deriving (Show, Typeable, Generic) @@ -198,7 +200,7 @@ -- data ServerCallback eff = ServerCallback { _requestHandlerSelector :: MessageSelector (Eff eff ApiServerCmd)- , _terminationHandler :: Maybe String -> Eff eff ()+ , _terminationHandler :: ExitReason 'Recoverable -> Eff eff () } makeLenses ''ServerCallback@@ -209,9 +211,9 @@ runMessageSelector (view requestHandlerSelector l) x <|> runMessageSelector (view requestHandlerSelector r) x) & terminationHandler .~- (\errorMessage ->- do (l^.terminationHandler) errorMessage- (r^.terminationHandler) errorMessage)+ (\reason ->+ do (l^.terminationHandler) reason+ (r^.terminationHandler) reason) instance Monoid (ServerCallback eff) where mappend = (<>)@@ -232,7 +234,8 @@ -- in a type safe way. toServerPids :: proxy a -> ProcessId -> ServerPids a -- | Convert the value to a 'ServerCallback'- toServerCallback :: (SetMember Process (Process effScheduler) (ServerEff a))+ toServerCallback+ :: (Member Interrupts (ServerEff a), SetMember Process (Process effScheduler) (ServerEff a)) => SchedulerProxy effScheduler -> a -> ServerCallback (ServerEff a) instance Servable (ServerCallback eff) where@@ -261,6 +264,7 @@ :: forall a effScheduler . ( Servable a , SetMember Process (Process effScheduler) (ServerEff a)+ , Member Interrupts (ServerEff a) , HasCallStack ) => SchedulerProxy effScheduler@@ -271,7 +275,7 @@ stopServer reason = do (serverCb ^. terminationHandler) reason return (Just ())- in receiveLoopSuchThat px (serverCb ^. requestHandlerSelector) $ \case+ in receiveSelectedLoop px (serverCb ^. requestHandlerSelector) $ \case Left reason -> stopServer reason Right handleIt -> handleIt >>= \case HandleNextRequest -> return Nothing@@ -282,8 +286,9 @@ spawnServer :: forall a effScheduler eff . ( Servable a- , ServerEff a ~ (Process effScheduler ': effScheduler)+ , ServerEff a ~ (InterruptableProcess effScheduler) , SetMember Process (Process effScheduler) eff+ , Member Interrupts eff , HasCallStack ) => SchedulerProxy effScheduler@@ -298,12 +303,14 @@ . ( Servable a , SetMember Process (Process effScheduler) (ServerEff a) , SetMember Process (Process effScheduler) eff+ , Member Interrupts eff+ , Member Interrupts (ServerEff a) , HasCallStack ) => SchedulerProxy effScheduler -> a -> ( Eff (ServerEff a) ()- -> Eff (Process effScheduler ': effScheduler) ()+ -> Eff (InterruptableProcess effScheduler) () ) -> Eff eff (ServerPids a) spawnServerWithEffects px a handleEff = do@@ -316,6 +323,7 @@ . ( HasCallStack , Typeable api , SetMember Process (Process effScheduler) eff+ , Member Interrupts eff ) => SchedulerProxy effScheduler -> ApiHandler api eff@@ -333,12 +341,13 @@ . ( HasCallStack , Typeable api , SetMember Process (Process effScheduler) eff+ , Member (Exc (ExitReason 'Recoverable)) eff ) => SchedulerProxy effScheduler -> ApiHandler api eff -> MessageSelector (Eff eff ApiServerCmd)-selectHandlerMethod px handlers =- selectDynamicMessageLazy (fmap (applyHandlerMethod px handlers) . fromDynamic)+selectHandlerMethod px handlers = selectDynamicMessageLazy+ (fmap (applyHandlerMethod px handlers) . fromDynamic) -- | Apply either the '_callCallback', '_castCallback' or the '_terminateCallback' -- callback to an incoming request.@@ -346,6 +355,7 @@ :: forall eff effScheduler api . ( Typeable api , SetMember Process (Process effScheduler) eff+ , Member (Exc (ExitReason 'Recoverable)) eff , HasCallStack ) => SchedulerProxy effScheduler@@ -362,37 +372,48 @@ where sendReply :: Typeable reply => reply -> Eff eff () sendReply reply =- sendMessage px fromPid (toDyn $! (Response (Proxy @api) callRef $! reply))+ sendMessage px fromPid (Response (Proxy @api) callRef $! reply) -- | A default handler to use in '_callCallback' in 'ApiHandler'. It will call -- 'raiseError' with a nice error message. unhandledCallError :: forall p x r q- . (Typeable p, HasCallStack, SetMember Process (Process q) r)+ . ( Typeable p+ , HasCallStack+ , SetMember Process (Process q) r+ , Member (Exc (ExitReason 'Recoverable)) r+ ) => SchedulerProxy q -> Api p ( 'Synchronous x) -> (x -> Eff r ()) -> Eff r ApiServerCmd-unhandledCallError px _api _ =- raiseError px ("unhandled call on api: " ++ show (typeRep (Proxy @p)))+unhandledCallError _px _api _ = throwError+ (ProcessError ("unhandled call on api: " ++ show (typeRep (Proxy @p)))) -- | A default handler to use in '_castCallback' in 'ApiHandler'. It will call -- 'raiseError' with a nice error message. unhandledCastError :: forall p r q- . (Typeable p, HasCallStack, SetMember Process (Process q) r)+ . ( Typeable p+ , HasCallStack+ , SetMember Process (Process q) r+ , Member (Exc (ExitReason 'Recoverable)) r+ ) => SchedulerProxy q -> Api p 'Asynchronous -> Eff r ApiServerCmd-unhandledCastError px _api =- raiseError px ("unhandled cast on api: " ++ show (typeRep (Proxy @p)))+unhandledCastError _px _api = throwError+ (ProcessError ("unhandled cast on api: " ++ show (typeRep (Proxy @p)))) -- | Either do nothing, if the error message is @Nothing@, -- or call 'exitWithError' with the error message. defaultTermination :: forall q r- . (HasCallStack, SetMember Process (Process q) r)+ . ( HasCallStack+ , SetMember Process (Process q) r+ , Member (Logs LogMessage) r+ ) => SchedulerProxy q- -> Maybe String+ -> ExitReason 'Recoverable -> Eff r ()-defaultTermination px = maybe (return ()) (exitWithError px)+defaultTermination _px r = logNotice ("server process terminating " ++ show r)
src/Control/Eff/Concurrent/Process.hs view
@@ -1,652 +1,1208 @@--- | The message passing effect.------ This module describes an abstract message passing effect, and a process--- effect, mimicking Erlang's process and message semantics.------ Two __scheduler__ implementations for the 'Process' effect are provided:------ * A scheduler using @forkIO@, i.e. relying on the multi threaded GHC runtime:--- "Control.Eff.Concurrent.Process.ForkIOScheduler"------ * And a /pure/(rer) coroutine based scheduler in:--- "Control.Eff.Concurrent.Process.SingleThreadedScheduler"-module Control.Eff.Concurrent.Process- ( -- * ProcessId Type- ProcessId(..)- , fromProcessId- -- * Process Effects- , Process(..)- , ConsProcess- , ResumeProcess(..)- , SchedulerProxy(..)- , MessageSelector(runMessageSelector)- , selectMessage- , selectMessageLazy- , selectMessageProxy- , selectMessageProxyLazy- , filterMessage- , filterMessageLazy- , selectMessageWith- , selectMessageWithLazy- , selectDynamicMessage- , selectDynamicMessageLazy- , selectAnyMessageLazy- , ProcessState(..)- , ProcessExitReason(..)- , ShutdownRequest(..)- , thisSchedulerProxy- , executeAndCatch- , executeAndResume- , yieldProcess- , sendMessage- , sendMessageAs- , sendMessageChecked- , spawn- , spawn_- , receiveMessage- , receiveMessageAs- , receiveSelectedMessage- , receiveLoop- , receiveLoopAs- , receiveLoopSuchThat- , self- , sendShutdown- , sendShutdownChecked- , makeReference- , exitWithError- , exitNormally- , raiseError- , catchRaisedError- , ignoreProcessError- , logProcessExit- )-where--import GHC.Generics ( Generic- , Generic1- )-import Control.DeepSeq-import Control.Eff-import Control.Eff.Extend-import Control.Eff.Log.Handler-import Control.Eff.Log.Message-import Control.Lens-import Control.Monad ( void- , (>=>)- )-import Data.Default-import Data.Dynamic-import Data.Kind-import GHC.Stack-import qualified Control.Exception as Exc---- | 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.------ Processes can raise exceptions that can be caught, exit gracefully or with an--- error, or be killed by other processes, with the option of ignoring the--- shutdown request.------ Process Scheduling is implemented in different modules. All scheduler--- implementations should follow some basic rules:------ * fair scheduling------ * sending a message does not block------ * receiving a message does block------ * spawning a child blocks only a very moment------ * a newly spawned process shall be scheduled before the parent process after--- * the spawn------ * when the first process exists, all process should be killed immediately-data Process (r :: [Type -> Type]) b where- -- | In cooperative schedulers, this will give processing time to the- -- scheduler. Every other operation implicitly serves the same purpose.- YieldProcess :: Process r (ResumeProcess ())- -- | Return the current 'ProcessId'- SelfPid :: Process r (ResumeProcess ProcessId)- -- | Start a new process, the new process will execute an effect, the function- -- will return immediately with a 'ProcessId'.- Spawn :: Eff (Process r ': r) () -> Process r (ResumeProcess ProcessId)- -- | Process exit, this is the same as if the function that was applied to a- -- spawn function returned.- Shutdown :: ShutdownRequest -> Process r a- -- | Raise an error, that can be handled.- RaiseError :: String -> Process r b- -- | Request that another a process exits. The targeted process is interrupted- -- and gets a 'ShutdownRequested', the target process may decide to ignore the shutdown- -- requests.- SendShutdown :: ProcessId -> ShutdownRequest -> Process r (ResumeProcess Bool)- -- LinkProcesses :: ProcessId -> ProcessId -> Process ()- -- | Send a message to a process addressed by the 'ProcessId'. Sending a- -- message should **always succeed** and return **immediately**, even if the- -- destination process does not exist, or does not accept messages of the- -- given type.- SendMessage :: ProcessId -> Dynamic -> Process r (ResumeProcess Bool)- -- | Receive a message that matches a criterium.- -- This should block until an a message was received. The message is returned- -- as a 'ProcessMessage' value. The function should also return if an exception- -- was caught or a shutdown was requested.- ReceiveSelectedMessage :: forall r a . MessageSelector a -> Process r (ResumeProcess a)- -- | Generate a unique 'Int' for the current process.- MakeReference :: Process r (ResumeProcess Int)--instance Show (Process r b) where- showsPrec d = \case- YieldProcess -> showString "YieldProcess"- SelfPid -> showString "SelfPid"- Spawn _ -> showString "Spawn"- Shutdown sr ->- showParen (d >= 10) (showString "Shutdown " . showsPrec 10 sr)- RaiseError sr ->- showParen (d >= 10) (showString "RaiseError " . showsPrec 10 sr)- SendShutdown toPid sr -> showParen- (d >= 10)- ( showString "SendShutdown "- . showsPrec 10 toPid- . showChar ' '- . showsPrec 10 sr- )- SendMessage toPid sr -> showParen- (d >= 10)- ( showString "SendMessage "- . showsPrec 10 toPid- . showChar ' '- . showsPrec 10 sr- )- ReceiveSelectedMessage _ -> showString "ReceiveSelectedMessage"- MakeReference -> showString "MakeReference"---- | Every 'Process' action returns it's actual result wrapped in this type. It--- will allow to signal errors as well as pass on normal results such as--- incoming messages.-data ResumeProcess v where- -- | The process received a 'ShutdownRequest'.- ShutdownRequested :: ShutdownRequest -> ResumeProcess v- -- | The process is required to exit from an error condition, that cannot be- -- recovered from.- OnError :: String -> ResumeProcess v- -- | The process may resume to do work, using the given result.- 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, Generic, Generic1)--instance NFData a => NFData (ResumeProcess a)--instance NFData1 ResumeProcess---- | A function that deciced if the next message will be received by--- 'ReceiveSelectedMessage'. It conveniently is an instance of 'Monoid'--- with first come, first serve bais.-newtype MessageSelector a =- MessageSelector {runMessageSelector :: Dynamic -> Maybe a }- deriving (Semigroup, Monoid, Functor)----- | Create a message selector for a value that can be obtained by 'fromDynamic'.--- It will also 'force' the result.------ @since 0.9.1-selectMessage :: (NFData t, Typeable t) => MessageSelector t-selectMessage = selectDynamicMessage fromDynamic---- | Create a message selector for a value that can be obtained by 'fromDynamic'.--- It will also 'force' the result.------ @since 0.9.1-selectMessageLazy :: Typeable t => MessageSelector t-selectMessageLazy = selectDynamicMessageLazy fromDynamic---- | Create a message selector from a predicate. It will 'force' the result.------ @since 0.9.1-filterMessage :: (Typeable a, NFData a) => (a -> Bool) -> MessageSelector a-filterMessage predicate = selectDynamicMessage- (\d -> case fromDynamic d of- Just a | predicate a -> Just a- _ -> Nothing- )---- | Create a message selector from a predicate. It will 'force' the result.------ @since 0.9.1-filterMessageLazy :: Typeable a => (a -> Bool) -> MessageSelector a-filterMessageLazy predicate = selectDynamicMessageLazy- (\d -> case fromDynamic d of- Just a | predicate a -> Just a- _ -> Nothing- )---- | Select a message of type @a@ and apply the given function to it.--- If the function returns 'Just' The 'ReceiveSelectedMessage' function will--- return the result (sans @Maybe@). It will 'force' the result.------ @since 0.9.1-selectMessageWith- :: (Typeable a, NFData b) => (a -> Maybe b) -> MessageSelector b-selectMessageWith f = selectDynamicMessage (fromDynamic >=> f)---- | Select a message of type @a@ and apply the given function to it.--- If the function returns 'Just' The 'ReceiveSelectedMessage' function will--- return the result (sans @Maybe@). It will 'force' the result.------ @since 0.9.1-selectMessageWithLazy :: Typeable a => (a -> Maybe b) -> MessageSelector b-selectMessageWithLazy f = selectDynamicMessageLazy (fromDynamic >=> f)---- | Create a message selector. It will 'force' the result.------ @since 0.9.1-selectDynamicMessage :: NFData a => (Dynamic -> Maybe a) -> MessageSelector a-selectDynamicMessage = MessageSelector . (force .)---- | Create a message selector.------ @since 0.9.1-selectDynamicMessageLazy :: (Dynamic -> Maybe a) -> MessageSelector a-selectDynamicMessageLazy = MessageSelector---- | Create a message selector that will match every message. This is /lazy/--- because the result is not 'force'ed.------ @since 0.9.1-selectAnyMessageLazy :: MessageSelector Dynamic-selectAnyMessageLazy = MessageSelector Just---- | Create a message selector for a value that can be obtained by 'fromDynamic'--- with a proxy argument. It will also 'force' the result.------ @since 0.9.1-selectMessageProxy- :: forall proxy t . (NFData t, Typeable t) => proxy t -> MessageSelector t-selectMessageProxy _ = selectDynamicMessage fromDynamic---- | Create a message selector for a value that can be obtained by 'fromDynamic'--- with a proxy argument. It will also 'force' the result.------ @since 0.9.1-selectMessageProxyLazy- :: forall proxy t . (Typeable t) => proxy t -> MessageSelector t-selectMessageProxyLazy _ = selectDynamicMessageLazy fromDynamic---- | Every function for 'Process' things needs such a proxy value--- for the low-level effect list, i.e. the effects identified by--- @__r__@ in @'Process' r : r@, this might be dependent on the--- scheduler implementation.-data SchedulerProxy :: [Type -> Type] -> Type where- -- | Tell the typechecker what effects we have below 'Process'- SchedulerProxy :: SchedulerProxy q- -- | Like 'SchedulerProxy' but shorter- SP :: SchedulerProxy q- -- | Like 'SP' but different- Scheduler :: SchedulerProxy q---- | /Cons/ 'Process' onto a list of effects.-type ConsProcess r = Process r ': r---- | Return a 'SchedulerProxy' for a 'Process' effect.-thisSchedulerProxy :: Eff (Process r ': r) (SchedulerProxy r)-thisSchedulerProxy = return SchedulerProxy---- | The state that a 'Process' is currently in.-data ProcessState =- ProcessBooting -- ^ The process has just been started but not- -- called 'handleProcess' yet.- | ProcessIdle -- ^ The process yielded it's timeslice- | ProcessBusy -- ^ The process is busy with non-blocking- | ProcessBusySending -- ^ The process is busy with sending a message- | ProcessBusySendingShutdown -- ^ The process is busy with killing- | ProcessBusyReceiving -- ^ The process blocked by a 'receiveMessage'- | ProcessShuttingDown -- ^ The process was shutdown or crashed- deriving (Read, Show, Ord, Eq, Enum, Generic)--instance NFData ProcessState--instance Default ProcessState where def = ProcessBusy---- | A sum-type with reasons for why a process exists the scheduling loop,--- this includes errors, that can occur when scheduleing messages.-data ProcessExitReason =- ProcessRaisedError String- -- ^ A process called 'raiseError'.- | ProcessCaughtIOException String String- -- ^ A process called 'exitWithError'.- | ProcessShutDown ShutdownRequest- -- ^ A process exits.- | ProcessReturned- -- ^ A process function returned.- | SchedulerShuttingDown- -- ^ An action was not performed while the scheduler was exiting.- deriving (Typeable, Show, Generic)--instance NFData ProcessExitReason--instance Semigroup ProcessExitReason where- SchedulerShuttingDown <> _ = SchedulerShuttingDown- _ <> SchedulerShuttingDown = SchedulerShuttingDown- ProcessReturned <> x = x- x <> ProcessReturned = x- ProcessShutDown ExitNormally <> x = x- x <> ProcessShutDown ExitNormally = x- (ProcessRaisedError e1) <> (ProcessRaisedError e2) =- ProcessRaisedError (e1 ++ " and " ++ e2 )- e1 <> e2 =- ProcessShutDown (ExitWithError (show e1 ++ " and " ++ show e2 ))--instance Monoid ProcessExitReason where- mempty = ProcessShutDown ExitNormally- mappend = (<>)--instance Exc.Exception ProcessExitReason---- | When a process sends a process shutdown request to another process,--- it can specify the reason for the shutdown using this data type.-data ShutdownRequest =- ExitNormally- | ExitWithError String- deriving (Typeable, Show, Generic, Eq, Ord)--instance NFData ShutdownRequest---- | Execute a 'Process' action and resume the process, retry the action or exit--- the process when a shutdown was requested.-executeAndResume- :: forall r q v- . (SetMember Process (Process q) r, HasCallStack)- => Process q (ResumeProcess v)- -> Eff r v-executeAndResume processAction = do- result <- send processAction- case result of- ResumeWith !value -> return value- RetryLastAction -> executeAndResume processAction- ShutdownRequested r -> send (Shutdown @q r)- OnError e -> send (Shutdown @q (ExitWithError e))---- | Execute a and action and resume the process, retry the action, shutdown the process or return an error.-executeAndCatch- :: forall q r v- . (SetMember Process (Process q) r, HasCallStack)- => SchedulerProxy q- -> Eff r (ResumeProcess v)- -> Eff r (Either String v)-executeAndCatch px processAction = do- result <- processAction- case result of- ResumeWith !value -> return (Right value)- RetryLastAction -> executeAndCatch px processAction- ShutdownRequested r -> send (Shutdown @q r)- OnError e -> return (Left e)---- | Use 'executeAndResume' to execute 'YieldProcess'. Refer to 'YieldProcess'--- for more information.-yieldProcess- :: forall r q- . (HasCallStack, SetMember Process (Process q) r)- => SchedulerProxy q- -> Eff r ()-yieldProcess _ = executeAndResume YieldProcess---- | Send a message to a process addressed by the 'ProcessId'.--- See 'SendMessage'.-sendMessage- :: forall r q- . (HasCallStack, SetMember Process (Process q) r)- => SchedulerProxy q- -> ProcessId- -> Dynamic- -> Eff r ()-sendMessage px pid message = void (sendMessageChecked px pid message)---- | Send a message to a process addressed by the 'ProcessId'.--- See 'SendMessage'. Return @True@ if the process existed.--- I you don't care, just 'sendMessage' instead.-sendMessageChecked- :: forall r q- . (HasCallStack, SetMember Process (Process q) r)- => SchedulerProxy q- -> ProcessId- -> Dynamic- -> Eff r Bool-sendMessageChecked _ pid message =- executeAndResume (SendMessage pid $! message)---- | Send a message to a process addressed by the 'ProcessId'.--- See 'SendMessage'.-sendMessageAs- :: forall o r q- . (HasCallStack, SetMember Process (Process q) r, Typeable o)- => SchedulerProxy q- -> ProcessId- -> o- -> Eff r ()-sendMessageAs px pid = sendMessage px pid . toDyn---- | Exit a process addressed by the 'ProcessId'.--- See 'SendShutdown'.-sendShutdown- :: forall r q- . (HasCallStack, SetMember Process (Process q) r)- => SchedulerProxy q- -> ProcessId- -> ShutdownRequest- -> Eff r ()-sendShutdown px pid s = void (sendShutdownChecked px pid s)---- | Like 'sendShutdown', but also return @True@ iff the process to exit exists.-sendShutdownChecked- :: forall r q- . (HasCallStack, SetMember Process (Process q) r)- => SchedulerProxy q- -> ProcessId- -> ShutdownRequest- -> Eff r Bool-sendShutdownChecked _ pid s = executeAndResume (SendShutdown pid s)---- | Start a new process, the new process will execute an effect, the function--- will return immediately with a 'ProcessId'.-spawn- :: forall r q- . (HasCallStack, SetMember Process (Process q) r)- => Eff (Process q ': q) ()- -> Eff r ProcessId-spawn child = executeAndResume (Spawn @q child)---- | Like 'spawn' but return @()@.-spawn_- :: forall r q- . (HasCallStack, SetMember Process (Process q) r)- => Eff (Process q ': q) ()- -> Eff r ()-spawn_ = void . spawn---- | Block until a message was received.--- See 'ReceiveMessage' for more documentation.-receiveMessage- :: forall r q- . (HasCallStack, SetMember Process (Process q) r)- => SchedulerProxy q- -> Eff r Dynamic-receiveMessage _ =- executeAndResume (ReceiveSelectedMessage selectAnyMessageLazy)---- | Block until a message was received, that is not 'Nothing' after applying--- a callback to it.--- See 'ReceiveSelectedMessage' for more documentation.-receiveSelectedMessage- :: forall r q a- . (HasCallStack, Typeable a, SetMember Process (Process q) r)- => SchedulerProxy q- -> MessageSelector a- -> Eff r a-receiveSelectedMessage _ f = executeAndResume (ReceiveSelectedMessage f)---- | Receive and cast the message to some 'Typeable' instance.--- See 'ReceiveSelectedMessage' for more documentation.--- This will wait for a message of the return type using 'receiveSelectedMessage'-receiveMessageAs- :: forall a r q- . (HasCallStack, Typeable a, SetMember Process (Process q) r)- => SchedulerProxy q- -> Eff r a-receiveMessageAs px = receiveSelectedMessage px (MessageSelector fromDynamic)---- | Enter a loop to receive messages and pass them to a callback, until the--- function returns 'Just' a result.--- See 'selectAnyMessageLazy' or 'ReceiveSelectedMessage' for more documentation.-receiveLoop- :: forall r q endOfLoopResult- . (SetMember Process (Process q) r, HasCallStack)- => SchedulerProxy q- -> (Either (Maybe String) Dynamic -> Eff r (Maybe endOfLoopResult))- -> Eff r endOfLoopResult-receiveLoop px handlers = do- mReq <- send (ReceiveSelectedMessage @q selectAnyMessageLazy)- mRes <- case mReq of- RetryLastAction -> return Nothing- ShutdownRequested ExitNormally -> handlers (Left Nothing)- ShutdownRequested (ExitWithError reason) -> handlers (Left (Just reason))- OnError reason -> handlers (Left (Just reason))- ResumeWith message -> handlers (Right message)- maybe (receiveLoop px handlers) return mRes---- | Run receive message of a certain type until the handler--- function returns 'Just' a result.--- Like 'receiveLoopSuchThat' applied to 'fromDynamic'-receiveLoopAs- :: forall r q a endOfLoopResult- . (SetMember Process (Process q) r, HasCallStack, Typeable a)- => SchedulerProxy q- -> (Either (Maybe String) a -> Eff r (Maybe endOfLoopResult))- -> Eff r endOfLoopResult-receiveLoopAs px handlers = do- mReq <- send (ReceiveSelectedMessage @q @a (MessageSelector fromDynamic))- mRes <- case mReq of- RetryLastAction -> return Nothing- ShutdownRequested ExitNormally -> handlers (Left Nothing)- ShutdownRequested (ExitWithError reason) -> handlers (Left (Just reason))- OnError reason -> handlers (Left (Just reason))- ResumeWith message -> handlers (Right message)- maybe (receiveLoopAs px handlers) return mRes----- | Like 'receiveLoop' but /selective/: Only the messages of the given type will be--- received.-receiveLoopSuchThat- :: forall r q a endOfLoopResult- . (SetMember Process (Process q) r, HasCallStack)- => SchedulerProxy q- -> MessageSelector a- -> (Either (Maybe String) a -> Eff r (Maybe endOfLoopResult))- -> Eff r endOfLoopResult-receiveLoopSuchThat px selectMesage handlers = do- mReq <- send (ReceiveSelectedMessage @q @a selectMesage)- mRes <- case mReq of- RetryLastAction -> return Nothing- ShutdownRequested ExitNormally -> handlers (Left Nothing)- ShutdownRequested (ExitWithError reason) -> handlers (Left (Just reason))- OnError reason -> handlers (Left (Just reason))- ResumeWith message -> handlers (Right message)- maybe (receiveLoopSuchThat px selectMesage handlers) return mRes---- | Returns the 'ProcessId' of the current process.-self- :: (HasCallStack, SetMember Process (Process q) r)- => SchedulerProxy q- -> Eff r ProcessId-self _px = executeAndResume SelfPid---- | Generate a unique 'Int' for the current process.-makeReference- :: (HasCallStack, SetMember Process (Process q) r)- => SchedulerProxy q- -> Eff r Int-makeReference _px = executeAndResume MakeReference---- | Exit the process.-exitNormally- :: forall r q a- . (HasCallStack, SetMember Process (Process q) r)- => SchedulerProxy q- -> Eff r a-exitNormally _ = send (Shutdown @q ExitNormally)---- | Exit the process with an error.-exitWithError- :: forall r q a- . (HasCallStack, SetMember Process (Process q) r)- => SchedulerProxy q- -> String- -> Eff r a-exitWithError _ = send . (Shutdown @q . (ExitWithError $!))---- | Thrown an error, can be caught by 'catchRaisedError'.-raiseError- :: forall r q b- . (HasCallStack, SetMember Process (Process q) r)- => SchedulerProxy q- -> String- -> Eff r b-raiseError _ = send . (RaiseError @q $!)---- | Catch and handle an error raised by 'raiseError'. Works independent of the--- handler implementation.-catchRaisedError- :: forall r q w- . (HasCallStack, SetMember Process (Process q) r)- => SchedulerProxy q- -> (String -> Eff r w)- -> Eff r w- -> Eff r w-catchRaisedError _ onErr = interpose return go- where- go :: forall b . Process q b -> (b -> Eff r w) -> Eff r w- go (RaiseError emsg) _k = onErr emsg- go s k = send s >>= k---- | Like 'catchRaisedError' it catches 'raiseError', but instead of invoking a--- user provided handler, the result is wrapped into an 'Either'.-ignoreProcessError- :: (HasCallStack, SetMember Process (Process q) r)- => SchedulerProxy q- -> Eff r a- -> Eff r (Either String a)-ignoreProcessError px = catchRaisedError px (return . Left) . fmap Right---- | Each process is identified by a single process id, that stays constant--- throughout the life cycle of a process. Also, message sending relies on these--- values to address messages to processes.-newtype ProcessId = ProcessId { _fromProcessId :: Int }- deriving (Eq,Ord,Typeable,Bounded,Num, Enum, Integral, Real, NFData)--instance Read ProcessId where- readsPrec _ ('!':rest1) =- case reads rest1 of- [(c, rest2)] -> [(ProcessId c, rest2)]- _ -> []- readsPrec _ _ = []--instance Show ProcessId where- showsPrec _ (ProcessId !c) = showChar '!' . shows c--makeLenses ''ProcessId---- | Log the 'ProcessExitReaons'-logProcessExit- :: (HasCallStack, HasLogWriter LogMessage h e)- => ProcessExitReason- -> Eff e ()-logProcessExit ex = withFrozenCallStack $ case ex of- ProcessReturned -> logDebug "returned"- ProcessShutDown ExitNormally -> logDebug "shutdown"- ProcessShutDown (ExitWithError m) -> logError ("exit with error: " ++ show m)- ProcessCaughtIOException w m ->- logError ("runtime exception: " ++ m ++ " caught here: " ++ w)- ProcessRaisedError m -> logError ("unhandled process exception: " ++ show m)- SchedulerShuttingDown -> logInfo "scheduler schutting down"+{-# LANGUAGE ImplicitParams #-}+-- | The message passing effect.+--+-- This module describes an abstract message passing effect, and a process+-- effect, mimicking Erlang's process and message semantics.+--+-- Two __scheduler__ implementations for the 'Process' effect are provided:+--+-- * A scheduler using @forkIO@, i.e. relying on the multi threaded GHC runtime:+-- "Control.Eff.Concurrent.Process.ForkIOScheduler"+--+-- * And a /pure/(rer) coroutine based scheduler in:+-- "Control.Eff.Concurrent.Process.SingleThreadedScheduler"+module Control.Eff.Concurrent.Process+ ( -- * Process Effect+ -- ** Effect Type Handling+ Process(..)+ -- ** ProcessId Type+ , ProcessId(..)+ , fromProcessId+ , ConsProcess+ , ResumeProcess(..)+ -- ** Scheduler Effect Identification+ , SchedulerProxy(..)+ , HasScheduler+ , getSchedulerProxy+ , withSchedulerProxy+ , thisSchedulerProxy+ -- ** Process State+ , ProcessState(..)+ -- ** Yielding+ , yieldProcess+ -- ** Sending Messages+ , sendMessage+ , sendAnyMessage+ , sendShutdown+ , sendInterrupt+ -- ** Utilities+ , makeReference+ -- ** Receiving Messages+ , receiveMessage+ , receiveSelectedMessage+ , flushMessages+ , receiveAnyMessage+ , receiveLoop+ , receiveSelectedLoop+ , receiveAnyLoop+ -- ** Selecting Messages to Receive+ , MessageSelector(runMessageSelector)+ , selectMessage+ , selectMessageLazy+ , selectMessageProxy+ , selectMessageProxyLazy+ , filterMessage+ , filterMessageLazy+ , selectMessageWith+ , selectMessageWithLazy+ , selectDynamicMessage+ , selectDynamicMessageLazy+ , selectAnyMessageLazy+ -- ** Process Life Cycle Management+ , self+ , isProcessAlive+ -- ** Spawning+ , spawn+ , spawn_+ , spawnLink+ , spawnRaw+ , spawnRaw_+ -- ** Process Exit or Interrupt+ , exitBecause+ , exitNormally+ , exitWithError+ -- ** Links+ , linkProcess+ , unlinkProcess+ -- ** Monitors+ , monitor+ , demonitor+ , ProcessDown(..)+ , selectProcessDown+ , becauseProcessIsDown+ , MonitorReference(..)+ , withMonitor+ , receiveWithMonitor+ -- ** Process Interrupt Handling+ , provideInterruptsShutdown+ , handleInterrupts+ , exitOnInterrupt+ , logInterrupts+ , provideInterrupts+ , mergeEitherInterruptAndExitReason+ , interrupt+ -- ** Process Operation Execution+ , executeAndResume+ , executeAndResumeOrExit+ , executeAndResumeOrThrow+ -- ** Exit Or Interrupt Reasons+ , ExitReason(..)+ , ExitRecovery(..)+ , InterruptReason+ , Interrupts+ , InterruptableProcess+ , ExitSeverity(..)+ , SomeExitReason(SomeExitReason)+ , toExitRecovery+ , isRecoverable+ , toExitSeverity+ , isBecauseDown+ , isCrash+ , toCrashReason+ , fromSomeExitReason+ , logProcessExit+ )+where++import GHC.Generics ( Generic+ , Generic1+ )+import Control.DeepSeq+import Control.Eff+import Control.Eff.Exception+import Control.Eff.Extend+import Control.Eff.Log.Handler+import Control.Eff.Log.Message+import Control.Lens+import Control.Monad ( void+ , (>=>)+ )+import Data.Default+import Data.Dynamic+import Data.Kind+import GHC.Stack+import Data.Function+import Control.Applicative+import Data.Maybe+import qualified Control.Exception as Exc++-- | 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.+--+-- Processes can raise exceptions that can be caught, exit gracefully or with an+-- error, or be killed by other processes, with the option of ignoring the+-- shutdown request.+--+-- Process Scheduling is implemented in different modules. All scheduler+-- implementations should follow some basic rules:+--+-- * fair scheduling+--+-- * sending a message does not block+--+-- * receiving a message does block+--+-- * spawning a child blocks only a very moment+--+-- * a newly spawned process shall be scheduled before the parent process after+-- * the spawnRaw+--+-- * when the first process exists, all process should be killed immediately+data Process (r :: [Type -> Type]) b where+ -- | Remove all messages from the process' message queue+ FlushMessages :: Process r (ResumeProcess [Dynamic])+ -- | In cooperative schedulers, this will give processing time to the+ -- scheduler. Every other operation implicitly serves the same purpose.+ --+ -- @since 0.12.0+ YieldProcess :: Process r (ResumeProcess ())+ -- | Return the current 'ProcessId'+ SelfPid :: Process r (ResumeProcess ProcessId)+ -- | Start a new process, the new process will execute an effect, the function+ -- will return immediately with a 'ProcessId'.+ Spawn :: Eff (Process r ': r) () -> Process r (ResumeProcess ProcessId)+ -- | Start a new process, and 'Link' to it .+ --+ -- @since 0.12.0+ SpawnLink :: Eff (Process r ': r) () -> Process r (ResumeProcess ProcessId)+ -- | Get the process state (or 'Nothing' if the process is dead)+ GetProcessState :: ProcessId -> Process r (ResumeProcess (Maybe ProcessState))+ -- | Shutdown the process; irregardles of the exit reason, this function never+ -- returns,+ Shutdown :: ExitReason 'NoRecovery -> Process r a+ -- | Raise an error, that can be handled.+ SendShutdown :: ProcessId -> ExitReason 'NoRecovery -> Process r (ResumeProcess ())+ -- | Request that another a process interrupts. The targeted process is interrupted+ -- and gets an 'Interrupted', the target process may decide to ignore the+ -- interrupt and continue as if nothing happened.+ SendInterrupt :: ProcessId -> InterruptReason -> Process r (ResumeProcess ())+ -- | Send a message to a process addressed by the 'ProcessId'. Sending a+ -- message should **always succeed** and return **immediately**, even if the+ -- destination process does not exist, or does not accept messages of the+ -- given type.+ SendMessage :: ProcessId -> Dynamic -> Process r (ResumeProcess ())+ -- | Receive a message that matches a criterium.+ -- This should block until an a message was received. The message is returned+ -- as a 'ProcessMessage' value. The function should also return if an exception+ -- was caught or a shutdown was requested.+ ReceiveSelectedMessage :: forall r a . MessageSelector a -> Process r (ResumeProcess a)+ -- | Generate a unique 'Int' for the current process.+ MakeReference :: Process r (ResumeProcess Int)+ -- | Monitor another process. When the monitored process exits a+ -- 'ProcessDown' is sent to the calling process.+ -- The return value is a unique identifier for that monitor.+ -- There can be multiple monitors on the same process,+ -- and a message for each will be sent.+ -- If the process is already dead, the 'ProcessDown' message+ -- will be sent immediately, w.thout exit reason+ --+ -- @since 0.12.0+ Monitor :: ProcessId -> Process r (ResumeProcess MonitorReference)+ -- | Remove a monitor.+ --+ -- @since 0.12.0+ Demonitor :: MonitorReference -> Process r (ResumeProcess ())+ -- | Connect the calling process to another process, such that+ -- if one of the processes crashes (i.e. 'isCrash' returns 'True'), the other+ -- is shutdown with the 'ProcessExitReaon' 'LinkedProcessCrashed'.+ --+ -- @since 0.12.0+ Link :: ProcessId -> Process r (ResumeProcess ())+ -- | Unlink the calling proccess from the other process.+ --+ -- @since 0.12.0+ Unlink :: ProcessId -> Process r (ResumeProcess ())++instance Show (Process r b) where+ showsPrec d = \case+ FlushMessages -> showString "flush messages"+ YieldProcess -> showString "yield process"+ SelfPid -> showString "lookup the current process id"+ Spawn _ -> showString "spawn a new process"+ SpawnLink _ -> showString "spawn a new process and link to it"+ Shutdown sr -> showParen (d >= 10) (showString "shutdown " . showsPrec 10 sr)+ SendShutdown toPid sr -> showParen+ (d >= 10)+ ( showString "shutting down "+ . showsPrec 10 toPid+ . showChar ' '+ . showsPrec 10 sr+ )+ SendInterrupt toPid sr -> showParen+ (d >= 10)+ ( showString "interrupting "+ . showsPrec 10 toPid+ . showChar ' '+ . showsPrec 10 sr+ )+ SendMessage toPid sr -> showParen+ (d >= 10)+ ( showString "sending to "+ . showsPrec 10 toPid+ . showChar ' '+ . showsPrec 10 sr+ )+ ReceiveSelectedMessage _ -> showString "receive a message"+ GetProcessState pid -> showString "get the process state of " . shows pid+ MakeReference -> showString "generate a unique reference"+ Monitor pid -> showString "monitor " . shows pid+ Demonitor i -> showString "demonitor " . shows i+ Link l -> showString "link " . shows l+ Unlink l -> showString "unlink " . shows l++-- | Every 'Process' action returns it's actual result wrapped in this type. It+-- will allow to signal errors as well as pass on normal results such as+-- incoming messages.+data ResumeProcess v where+ -- | The current operation of the process was interrupted with a+ -- 'ExitReason'. If 'isRecoverable' holds for the given reason,+ -- the process may choose to continue.+ Interrupted :: InterruptReason -> ResumeProcess v+ -- | The process may resume to do work, using the given result.+ ResumeWith :: a -> ResumeProcess a+ deriving ( Typeable, Generic, Generic1, Show )++instance NFData a => NFData (ResumeProcess a)++instance NFData1 ResumeProcess++-- | A function that deciced if the next message will be received by+-- 'ReceiveSelectedMessage'. It conveniently is an instance of+-- 'Alternative' so the message selector can be combined:+-- >+-- > selectInt :: MessageSelector Int+-- > selectInt = selectMessage+-- >+-- > selectString :: MessageSelector String+-- > selectString = selectMessage+-- >+-- > selectIntOrString :: MessageSelector (Either Int String)+-- > selectIntOrString =+-- > Left <$> selectTimeout<|> Right <$> selectString+--+newtype MessageSelector a =+ MessageSelector {runMessageSelector :: Dynamic -> Maybe a }+ deriving (Semigroup, Monoid, Functor)++instance Applicative MessageSelector where+ pure = MessageSelector . pure . pure+ (MessageSelector f) <*> (MessageSelector x) =+ MessageSelector (\dyn -> f dyn <*> x dyn)++instance Alternative MessageSelector where+ empty = MessageSelector (const empty)+ (MessageSelector l) <|> (MessageSelector r) =+ MessageSelector (\dyn -> l dyn <|> r dyn)++-- | Create a message selector for a value that can be obtained by 'fromDynamic'.+-- It will also 'force' the result.+--+-- @since 0.9.1+selectMessage :: (NFData t, Typeable t) => MessageSelector t+selectMessage = selectDynamicMessage fromDynamic++-- | Create a message selector for a value that can be obtained by 'fromDynamic'.+-- It will also 'force' the result.+--+-- @since 0.9.1+selectMessageLazy :: Typeable t => MessageSelector t+selectMessageLazy = selectDynamicMessageLazy fromDynamic++-- | Create a message selector from a predicate. It will 'force' the result.+--+-- @since 0.9.1+filterMessage :: (Typeable a, NFData a) => (a -> Bool) -> MessageSelector a+filterMessage predicate = selectDynamicMessage+ (\d -> case fromDynamic d of+ Just a | predicate a -> Just a+ _ -> Nothing+ )++-- | Create a message selector from a predicate. It will 'force' the result.+--+-- @since 0.9.1+filterMessageLazy :: Typeable a => (a -> Bool) -> MessageSelector a+filterMessageLazy predicate = selectDynamicMessageLazy+ (\d -> case fromDynamic d of+ Just a | predicate a -> Just a+ _ -> Nothing+ )++-- | Select a message of type @a@ and apply the given function to it.+-- If the function returns 'Just' The 'ReceiveSelectedMessage' function will+-- return the result (sans @Maybe@). It will 'force' the result.+--+-- @since 0.9.1+selectMessageWith+ :: (Typeable a, NFData b) => (a -> Maybe b) -> MessageSelector b+selectMessageWith f = selectDynamicMessage (fromDynamic >=> f)++-- | Select a message of type @a@ and apply the given function to it.+-- If the function returns 'Just' The 'ReceiveSelectedMessage' function will+-- return the result (sans @Maybe@). It will 'force' the result.+--+-- @since 0.9.1+selectMessageWithLazy :: Typeable a => (a -> Maybe b) -> MessageSelector b+selectMessageWithLazy f = selectDynamicMessageLazy (fromDynamic >=> f)++-- | Create a message selector. It will 'force' the result.+--+-- @since 0.9.1+selectDynamicMessage :: NFData a => (Dynamic -> Maybe a) -> MessageSelector a+selectDynamicMessage = MessageSelector . (force .)++-- | Create a message selector.+--+-- @since 0.9.1+selectDynamicMessageLazy :: (Dynamic -> Maybe a) -> MessageSelector a+selectDynamicMessageLazy = MessageSelector++-- | Create a message selector that will match every message. This is /lazy/+-- because the result is not 'force'ed.+--+-- @since 0.9.1+selectAnyMessageLazy :: MessageSelector Dynamic+selectAnyMessageLazy = MessageSelector Just++-- | Create a message selector for a value that can be obtained by 'fromDynamic'+-- with a proxy argument. It will also 'force' the result.+--+-- @since 0.9.1+selectMessageProxy+ :: forall proxy t . (NFData t, Typeable t) => proxy t -> MessageSelector t+selectMessageProxy _ = selectDynamicMessage fromDynamic++-- | Create a message selector for a value that can be obtained by 'fromDynamic'+-- with a proxy argument. It will also 'force' the result.+--+-- @since 0.9.1+selectMessageProxyLazy+ :: forall proxy t . (Typeable t) => proxy t -> MessageSelector t+selectMessageProxyLazy _ = selectDynamicMessageLazy fromDynamic++-- | Every function for 'Process' things needs such a proxy value+-- for the low-level effect list, i.e. the effects identified by+-- @__r__@ in @'Process' r : r@, this might be dependent on the+-- scheduler implementation.+data SchedulerProxy :: [Type -> Type] -> Type where+ -- | Tell the typechecker what effects we have below 'Process'+ SchedulerProxy :: SchedulerProxy q+ -- | Like 'SchedulerProxy' but shorter+ SP :: SchedulerProxy q+ -- | Like 'SP' but different+ Scheduler :: SchedulerProxy q++-- | A constraint for the implicit 'SchedulerProxy' parameter.+-- Use 'getSchedulerProxy' to query it. _EXPERIMENTAL_+--+-- @since 0.12.0+type HasScheduler q = (?_schedulerProxy :: SchedulerProxy q)++-- | Get access to the 'SchedulerProxy' for the current scheduler effects.+-- _EXPERIMENTAL_+--+-- @since 0.12.0+getSchedulerProxy :: HasScheduler q => SchedulerProxy q+getSchedulerProxy = ?_schedulerProxy++-- | Set the 'SchedulerProxy' to use, this satisfies 'HasScheduler' .+-- _EXPERIMENTAL_+--+-- @since 0.12.0+withSchedulerProxy :: SchedulerProxy q -> (HasScheduler q => a) -> a+withSchedulerProxy px x = let ?_schedulerProxy = px in x+++-- | /Cons/ 'Process' onto a list of effects.+type ConsProcess r = Process r ': r++-- | Return a 'SchedulerProxy' for a 'Process' effect.+thisSchedulerProxy :: Eff (Process r ': r) (SchedulerProxy r)+thisSchedulerProxy = return SchedulerProxy++-- | The state that a 'Process' is currently in.+data ProcessState =+ ProcessBooting -- ^ The process has just been started but not+ -- called 'handleProcess' yet.+ | ProcessIdle -- ^ The process yielded it's timeslice+ | ProcessBusy -- ^ The process is busy with non-blocking+ | ProcessBusySending -- ^ The process is busy with sending a message+ | ProcessBusySendingShutdown -- ^ The process is busy with killing+ | ProcessBusySendingInterrupt -- ^ The process is busy with killing+ | ProcessBusyReceiving -- ^ The process blocked by a 'receiveAnyMessage'+ | ProcessBusyLinking -- ^ The process blocked by a 'linkProcess'+ | ProcessBusyUnlinking -- ^ The process blocked by a 'unlinkProcess'+ | ProcessBusyMonitoring -- ^ The process blocked by a 'monitor'+ | ProcessBusyDemonitoring -- ^ The process blocked by a 'demonitor'+ | ProcessInterrupted -- ^ The process was interrupted+ | ProcessShuttingDown -- ^ The process was shutdown or crashed+ deriving (Read, Show, Ord, Eq, Enum, Generic)++instance NFData ProcessState++instance Default ProcessState where def = ProcessBooting++-- | This kind is used to indicate if a 'ExitReason' can be treated like+-- a short interrupt which can be handled or ignored.+data ExitRecovery = Recoverable | NoRecovery+ deriving (Typeable, Ord, Eq, Generic)++instance NFData ExitRecovery++instance Show ExitRecovery where+ showsPrec d =+ showParen (d>=10) .+ (\case+ Recoverable -> showString "recoverable"+ NoRecovery -> showString "not recoverable")++-- | Get the 'ExitRecover'y+toExitRecovery :: ExitReason r -> ExitRecovery+toExitRecovery = \case+ (ProcessNotRunning _) -> Recoverable+ (LinkedProcessCrashed _) -> Recoverable+ (ProcessError _) -> Recoverable+ ExitNormally -> NoRecovery+ (NotRecovered _ ) -> NoRecovery+ (UnexpectedException _ _) -> NoRecovery+ Killed -> NoRecovery++-- | This value indicates wether a process exited in way consistent with+-- the planned behaviour or not.+data ExitSeverity = NormalExit | Crash+ deriving (Typeable, Ord, Eq, Generic)++instance Show ExitSeverity where+ showsPrec d =+ showParen (d>=10) .+ (\case+ NormalExit -> showString "exit success"+ Crash -> showString "crash")++instance NFData ExitSeverity++-- | Get the 'ExitSeverity' of a 'ExitReason'.+toExitSeverity :: ExitReason e -> ExitSeverity+toExitSeverity = \case+ ExitNormally -> NormalExit+ _ -> Crash++-- | A sum-type with reasons for why a process exists the scheduling loop,+-- this includes errors, that can occur when scheduleing messages.+data ExitReason (t :: ExitRecovery) where+ -- | A process that should be running was not running.+ ProcessNotRunning+ :: ProcessId -> ExitReason 'Recoverable+ -- | A linked process is down+ LinkedProcessCrashed+ :: ProcessId -> ExitReason 'Recoverable+ -- | An exit reason that has an error message but isn't 'Recoverable'.+ ProcessError+ :: String -> ExitReason 'Recoverable+ -- | A process function returned or exited without any error.+ ExitNormally+ :: ExitReason 'NoRecovery+ -- | An unhandled 'Recoverable' allows 'NoRecovery'.+ NotRecovered+ :: (ExitReason 'Recoverable) -> ExitReason 'NoRecovery+ -- | An unexpected runtime exception was thrown, i.e. an exception+ -- derived from 'Control.Exception.Safe.SomeException'+ UnexpectedException+ :: String -> String -> ExitReason 'NoRecovery+ -- | A process was cancelled (e.g. killed, in 'Async.cancel')+ Killed+ :: ExitReason 'NoRecovery+ deriving Typeable++instance Show (ExitReason x) where+ showsPrec d =+ showParen (d>=10) .+ (\case+ ProcessNotRunning p -> showString "process not running: " . shows p+ LinkedProcessCrashed m -> showString "linked process "+ . shows m . showString " crashed"+ ProcessError reason -> showString "error: " . showString reason+ ExitNormally -> showString "exit normally"+ NotRecovered e -> showString "not recovered from: " . shows e+ UnexpectedException w m -> showString "unhandled runtime exception: "+ . showString m+ . showString " caught here: "+ . showString w+ Killed -> showString "killed"+ )++instance Exc.Exception (ExitReason 'Recoverable)+instance Exc.Exception (ExitReason 'NoRecovery )++instance NFData (ExitReason x) where+ rnf (ProcessNotRunning !l) = rnf l+ rnf (LinkedProcessCrashed !l) = rnf l+ rnf (ProcessError !l) = rnf l+ rnf ExitNormally = rnf ()+ rnf (NotRecovered !l) = rnf l+ rnf (UnexpectedException !l1 !l2) = rnf l1 `seq` rnf l2 `seq` ()+ rnf Killed = rnf ()++instance Ord (ExitReason x) where+ compare (ProcessNotRunning l) (ProcessNotRunning r) = compare l r+ compare (ProcessNotRunning _) _ = LT+ compare _ (ProcessNotRunning _) = GT+ compare (LinkedProcessCrashed l) (LinkedProcessCrashed r) = compare l r+ compare (LinkedProcessCrashed _) _ = LT+ compare _ (LinkedProcessCrashed _) = GT+ compare (ProcessError l) (ProcessError r) = compare l r+ compare ExitNormally ExitNormally = EQ+ compare ExitNormally _ = LT+ compare _ ExitNormally = GT+ compare (NotRecovered l) (NotRecovered r) = compare l r+ compare (NotRecovered _) _ = LT+ compare _ (NotRecovered _) = GT+ compare (UnexpectedException l1 l2) (UnexpectedException r1 r2) =+ compare l1 r1 <> compare l2 r2+ compare (UnexpectedException _ _) _ = LT+ compare _ (UnexpectedException _ _) = GT+ compare Killed Killed = EQ++instance Eq (ExitReason x) where+ (==) (ProcessNotRunning l) (ProcessNotRunning r) = (==) l r+ (==) ExitNormally ExitNormally = True+ (==) (LinkedProcessCrashed l) (LinkedProcessCrashed r) = l == r+ (==) (ProcessError l) (ProcessError r) = (==) l r+ (==) (NotRecovered l) (NotRecovered r) = (==) l r+ (==) (UnexpectedException l1 l2) (UnexpectedException r1 r2) =+ (==) l1 r1 && (==) l2 r2+ (==) Killed Killed = True+ (==) _ _ = False++-- | A predicate for linked process __crashes__.+isBecauseDown :: Maybe ProcessId -> ExitReason r -> Bool+isBecauseDown mp = \case+ ProcessNotRunning _ -> False+ LinkedProcessCrashed p -> maybe True (== p) mp+ ProcessError _ -> False+ ExitNormally -> False+ NotRecovered e -> isBecauseDown mp e+ UnexpectedException _ _ -> False+ Killed -> False++-- | 'ExitReason's which are recoverable are interrupts.+type InterruptReason = ExitReason 'Recoverable++-- | 'Exc'eptions containing 'InterruptReason's.+-- See 'handleInterrupts', 'exitOnInterrupt' or 'provideInterrupts'+type Interrupts = Exc InterruptReason++-- | This adds a layer of the 'Interrupts' effect ontop of 'ConsProcess'+type InterruptableProcess e = Interrupts ': ConsProcess e++-- | Handle all 'InterruptReason's of an 'InterruptableProcess' by+-- wrapping them up in 'NotRecovered' and then do a process 'Shutdown'.+provideInterruptsShutdown+ :: forall e a . Eff (InterruptableProcess e) a -> Eff (ConsProcess e) a+provideInterruptsShutdown e = do+ res <- provideInterrupts e+ case res of+ Left ex -> send (Shutdown @e (NotRecovered ex))+ Right a -> return a++-- | Handle 'InterruptReason's arising during process operations, e.g.+-- when a linked process crashes while we wait in a 'receiveSelectedMessage'+-- via a call to 'interrupt'.+handleInterrupts+ :: (HasCallStack, Member Interrupts r)+ => (InterruptReason -> Eff r a)+ -> Eff r a+ -> Eff r a+handleInterrupts = flip catchError++-- | Handle interrupts by logging them with `logProcessExit` and otherwise+-- ignoring them.+logInterrupts+ :: (HasCallStack, '[Interrupts, Logs LogMessage] <:: r)+ => Eff r ()+ -> Eff r ()+logInterrupts = handleInterrupts logProcessExit++-- | Handle 'InterruptReason's arising during process operations, e.g.+-- when a linked process crashes while we wait in a 'receiveSelectedMessage'+-- via a call to 'interrupt'.+exitOnInterrupt+ :: (HasCallStack, Member Interrupts r, SetMember Process (Process q) r)+ => SchedulerProxy q+ -> Eff r a+ -> Eff r a+exitOnInterrupt px = handleInterrupts (exitBecause px . NotRecovered)++-- | Handle 'InterruptReason's arising during process operations, e.g.+-- when a linked process crashes while we wait in a 'receiveSelectedMessage'+-- via a call to 'interrupt'.+provideInterrupts+ :: HasCallStack => Eff (Interrupts ': r) a -> Eff r (Either InterruptReason a)+provideInterrupts = runError+++-- | Wrap all (left) 'InterruptReason's into 'NotRecovered' and+-- return the (right) 'NoRecovery' 'ExitReason's as is.+mergeEitherInterruptAndExitReason+ :: Either InterruptReason (ExitReason 'NoRecovery) -> ExitReason 'NoRecovery+mergeEitherInterruptAndExitReason = either NotRecovered id++-- | Throw an 'InterruptReason', can be handled by 'recoverFromInterrupt' or+-- 'exitOnInterrupt' or 'provideInterrupts'.+interrupt :: (HasCallStack, Member Interrupts r) => InterruptReason -> Eff r a+interrupt = throwError++-- | A predicate for crashes. A /crash/ happens when a process exits+-- with an 'ExitReason' other than 'ExitNormally'+isCrash :: ExitReason x -> Bool+isCrash (NotRecovered !x) = isCrash x+isCrash ExitNormally = False+isCrash _ = True++-- | A predicate for recoverable exit reasons. This predicate defines the+-- exit reasonson which functions such as 'executeAndResume'+isRecoverable :: ExitReason x -> Bool+isRecoverable (toExitRecovery -> Recoverable) = True+isRecoverable _ = False++-- | An existential wrapper around 'ExitReason'+data SomeExitReason where+ SomeExitReason :: ExitReason x -> SomeExitReason++instance Ord SomeExitReason where+ compare = compare `on` fromSomeExitReason++instance Eq SomeExitReason where+ (==) = (==) `on` fromSomeExitReason++instance Show SomeExitReason where+ show = show . fromSomeExitReason++instance NFData SomeExitReason where+ rnf = rnf . fromSomeExitReason++-- | Partition a 'SomeExitReason' back into either a 'NoRecovery'+-- or a 'Recoverable' 'ExitReason'+fromSomeExitReason+ :: SomeExitReason -> Either (ExitReason 'NoRecovery) InterruptReason+fromSomeExitReason (SomeExitReason e) = case e of+ recoverable@(ProcessNotRunning _) -> Right recoverable+ recoverable@(LinkedProcessCrashed _) -> Right recoverable+ recoverable@(ProcessError _) -> Right recoverable+ noRecovery@ExitNormally -> Left noRecovery+ noRecovery@(NotRecovered _ ) -> Left noRecovery+ noRecovery@(UnexpectedException _ _) -> Left noRecovery+ noRecovery@Killed -> Left noRecovery++-- | Print a 'ExitReason' to 'Just' a formatted 'String' when 'isCrash'+-- is 'True'.+-- This can be useful in combination with view patterns, e.g.:+--+-- > logCrash :: ExitReason -> Eff e ()+-- > logCrash (toCrashReason -> Just reason) = logError reason+-- > logCrash _ = return ()+--+-- Though this can be improved to:+--+-- > logCrash = traverse_ logError . toCrashReason+--+toCrashReason :: ExitReason x -> Maybe String+toCrashReason e | isCrash e = Just (show e)+ | otherwise = Nothing++-- | Log the 'ProcessExitReaons'+logProcessExit+ :: (HasCallStack, Member (Logs LogMessage) e) => ExitReason x -> Eff e ()+logProcessExit (toCrashReason -> Just ex) = withFrozenCallStack (logError ex)+logProcessExit ex = withFrozenCallStack (logDebug (show ex))+++-- | Execute a and action and return the result;+-- if the process is interrupted by an error or exception, or an explicit+-- shutdown from another process, or through a crash of a linked process, i.e.+-- whenever the exit reason satisfies 'isRecoverable', return the exit reason.+executeAndResume+ :: forall q r v+ . (SetMember Process (Process q) r, HasCallStack)+ => Process q (ResumeProcess v)+ -> Eff r (Either (ExitReason 'Recoverable) v)+executeAndResume processAction = do+ result <- send processAction+ case result of+ ResumeWith !value -> return (Right value)+ Interrupted r -> return (Left r)++-- | Execute a 'Process' action and resume the process, exit+-- the process when an 'Interrupts' was raised. Use 'executeAndResume' to catch+-- interrupts.+executeAndResumeOrExit+ :: forall r q v+ . (SetMember Process (Process q) r, HasCallStack)+ => Process q (ResumeProcess v)+ -> Eff r v+executeAndResumeOrExit processAction = do+ result <- send processAction+ case result of+ ResumeWith !value -> return value+ Interrupted r -> send (Shutdown @q (NotRecovered r))++-- | Execute a 'Process' action and resume the process, exit+-- the process when an 'Interrupts' was raised. Use 'executeAndResume' to catch+-- interrupts.+executeAndResumeOrThrow+ :: forall q r v+ . (SetMember Process (Process q) r, HasCallStack, Member Interrupts r)+ => Process q (ResumeProcess v)+ -> Eff r v+executeAndResumeOrThrow processAction = do+ result <- send processAction+ case result of+ ResumeWith !value -> return value+ Interrupted r -> interrupt r++-- * Process Effects++-- | Use 'executeAndResumeOrExit' to execute 'YieldProcess'. Refer to 'YieldProcess'+-- for more information.+yieldProcess+ :: forall r q+ . (SetMember Process (Process q) r, HasCallStack, Member Interrupts r)+ => SchedulerProxy q+ -> Eff r ()+yieldProcess _ = executeAndResumeOrThrow YieldProcess++-- | Send a message to a process addressed by the 'ProcessId'.+-- See 'SendMessage'.+--+sendMessage+ :: forall r q o+ . ( SetMember Process (Process q) r+ , HasCallStack+ , Member Interrupts r+ , Typeable o+ )+ => SchedulerProxy q+ -> ProcessId+ -> o+ -> Eff r ()+sendMessage _ pid message =+ executeAndResumeOrThrow (SendMessage pid $! toDyn $! message)++-- | Send a 'Dynamic' value to a process addressed by the 'ProcessId'.+-- See 'SendMessage'.+sendAnyMessage+ :: forall r q+ . (SetMember Process (Process q) r, HasCallStack, Member Interrupts r)+ => SchedulerProxy q+ -> ProcessId+ -> Dynamic+ -> Eff r ()+sendAnyMessage _ pid message =+ rnf pid `seq` executeAndResumeOrThrow (SendMessage pid $! message)++-- | Exit a process addressed by the 'ProcessId'. The process will exit,+-- it might do some cleanup, but is ultimately unrecoverable.+-- See 'SendShutdown'.+sendShutdown+ :: forall r q+ . (SetMember Process (Process q) r, HasCallStack, Member Interrupts r)+ => SchedulerProxy q+ -> ProcessId+ -> ExitReason 'NoRecovery+ -> Eff r ()+sendShutdown _ pid s =+ pid `deepseq` s `deepseq` executeAndResumeOrThrow (SendShutdown pid s)++-- | Interrupts a process addressed by the 'ProcessId'. The process might exit,+-- or it may continue.+-- | Like 'sendInterrupt', but also return @True@ iff the process to exit exists.+sendInterrupt+ :: forall r q+ . (SetMember Process (Process q) r, HasCallStack, Member Interrupts r)+ => SchedulerProxy q+ -> ProcessId+ -> InterruptReason+ -> Eff r ()+sendInterrupt _ pid s =+ pid `deepseq` s `deepseq` executeAndResumeOrThrow (SendInterrupt pid s)++-- | Start a new process, the new process will execute an effect, the function+-- will return immediately with a 'ProcessId'. If the new process is+-- interrupted, the process will 'Shutdown' with the 'InterruptReason'+-- wrapped in 'NotCovered'. For specific use cases it might be better to use+-- 'spawnRaw'.+spawn+ :: forall r q+ . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)+ => Eff (InterruptableProcess q) ()+ -> Eff r ProcessId+spawn child =+ executeAndResumeOrThrow (Spawn @q (provideInterruptsShutdown child))++-- | Like 'spawn' but return @()@.+spawn_+ :: forall r q+ . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)+ => Eff (InterruptableProcess q) ()+ -> Eff r ()+spawn_ child = void (spawn child)++-- | Start a new process, and immediately link to it.+--+-- @since 0.12.0+spawnLink+ :: forall r q+ . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)+ => Eff (InterruptableProcess q) ()+ -> Eff r ProcessId+spawnLink child =+ executeAndResumeOrThrow (SpawnLink @q (provideInterruptsShutdown child))++-- | Start a new process, the new process will execute an effect, the function+-- will return immediately with a 'ProcessId'. The spawned process has only the+-- /raw/ 'ConsProcess' effects. For non-library code 'spawn' might be better+-- suited.+spawnRaw+ :: forall r q+ . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)+ => Eff (ConsProcess q) ()+ -> Eff r ProcessId+spawnRaw child = executeAndResumeOrThrow (Spawn @q child)++-- | Like 'spawnRaw' but return @()@.+spawnRaw_+ :: forall r q+ . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)+ => Eff (ConsProcess q) ()+ -> Eff r ()+spawnRaw_ = void . spawnRaw++-- | Return 'True' if the process is alive.+--+-- @since 0.12.0+isProcessAlive+ :: forall r q+ . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)+ => SchedulerProxy q+ -> ProcessId+ -> Eff r Bool+isProcessAlive _px pid =+ isJust <$> executeAndResumeOrThrow (GetProcessState pid)++-- | Block until a message was received.+-- See 'ReceiveSelectedMessage' for more documentation.+receiveAnyMessage+ :: forall r q+ . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)+ => SchedulerProxy q+ -> Eff r Dynamic+receiveAnyMessage _ =+ executeAndResumeOrThrow (ReceiveSelectedMessage selectAnyMessageLazy)++-- | Block until a message was received, that is not 'Nothing' after applying+-- a callback to it.+-- See 'ReceiveSelectedMessage' for more documentation.+receiveSelectedMessage+ :: forall r q a+ . ( HasCallStack+ , Show a+ , SetMember Process (Process q) r+ , Member Interrupts r+ )+ => SchedulerProxy q+ -> MessageSelector a+ -> Eff r a+receiveSelectedMessage _ f = executeAndResumeOrThrow (ReceiveSelectedMessage f)++-- | Receive and cast the message to some 'Typeable' instance.+-- See 'ReceiveSelectedMessage' for more documentation.+-- This will wait for a message of the return type using 'receiveSelectedMessage'+receiveMessage+ :: forall a r q+ . ( HasCallStack+ , Typeable a+ , Show a+ , SetMember Process (Process q) r+ , Member Interrupts r+ )+ => SchedulerProxy q+ -> Eff r a+receiveMessage px = receiveSelectedMessage px (MessageSelector fromDynamic)++-- | Remove and return all messages currently enqueued in the process message+-- queue.+--+-- @since 0.12.0+flushMessages+ :: forall r q+ . ( HasCallStack+ , SetMember Process (Process q) r+ , Member Interrupts r+ , HasScheduler q+ )+ => Eff r [Dynamic]+flushMessages =+ executeAndResumeOrThrow @q FlushMessages++-- | Enter a loop to receive messages and pass them to a callback, until the+-- function returns 'Just' a result.+-- Only the messages of the given type will be received.+-- If the process is interrupted by an exception of by a 'SendShutdown' from+-- another process, with an exit reason that satisfies 'isRecoverable', then+-- the callback will be invoked with @'Left' 'ProcessExitReaon'@, otherwise the+-- process will be exited with the same reason using 'exitBecause'.+-- See also 'ReceiveSelectedMessage' for more documentation.+receiveSelectedLoop+ :: forall r q a endOfLoopResult+ . (SetMember Process (Process q) r, HasCallStack)+ => SchedulerProxy q+ -> MessageSelector a+ -> (Either InterruptReason a -> Eff r (Maybe endOfLoopResult))+ -> Eff r endOfLoopResult+receiveSelectedLoop px selectMesage handlers = do+ mReq <- send (ReceiveSelectedMessage @q @a selectMesage)+ mRes <- case mReq of+ Interrupted reason -> handlers (Left reason)+ ResumeWith message -> handlers (Right message)+ maybe (receiveSelectedLoop px selectMesage handlers) return mRes++-- | Like 'receiveSelectedLoop' but /not selective/.+-- See also 'selectAnyMessageLazy', 'receiveSelectedLoop'.+receiveAnyLoop+ :: forall r q endOfLoopResult+ . (SetMember Process (Process q) r, HasCallStack)+ => SchedulerProxy q+ -> (Either InterruptReason Dynamic -> Eff r (Maybe endOfLoopResult))+ -> Eff r endOfLoopResult+receiveAnyLoop px = receiveSelectedLoop px selectAnyMessageLazy++-- | Like 'receiveSelectedLoop' but refined to casting to a specific 'Typeable'+-- using 'selectMessageLazy'.+receiveLoop+ :: forall r q a endOfLoopResult+ . (SetMember Process (Process q) r, HasCallStack, Typeable a)+ => SchedulerProxy q+ -> (Either InterruptReason a -> Eff r (Maybe endOfLoopResult))+ -> Eff r endOfLoopResult+receiveLoop px = receiveSelectedLoop px selectMessageLazy++-- | Returns the 'ProcessId' of the current process.+self+ :: (HasCallStack, SetMember Process (Process q) r)+ => SchedulerProxy q+ -> Eff r ProcessId+self _px = executeAndResumeOrExit SelfPid++-- | Generate a unique 'Int' for the current process.+makeReference+ :: (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)+ => SchedulerProxy q+ -> Eff r Int+makeReference _px = executeAndResumeOrThrow MakeReference++-- | A value that contains a unique reference of a process+-- monitoring.+--+-- @since 0.12.0+data MonitorReference =+ MonitorReference { monitorIndex :: Int+ , monitoredProcess :: ProcessId+ }+ deriving (Read, Eq, Ord, Generic, Typeable)++instance NFData MonitorReference++instance Show MonitorReference where+ showsPrec d m =+ showParen (d>=10)+ ( showString "monitor: "+ . shows (monitorIndex m)+ . showChar ' '+ . shows (monitoredProcess m))++-- | Monitor another process. When the monitored process exits a+-- 'ProcessDown' is sent to the calling process.+-- The return value is a unique identifier for that monitor.+-- There can be multiple monitors on the same process,+-- and a message for each will be sent.+-- If the process is already dead, the 'ProcessDown' message+-- will be sent immediately, w.thout exit reason+--+-- @since 0.12.0+monitor+ :: forall r q+ . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)+ => SchedulerProxy q+ -> ProcessId+ -> Eff r MonitorReference+monitor _px = executeAndResumeOrThrow . Monitor . force++-- | Remove a monitor created with 'monitor'.+--+-- @since 0.12.0+demonitor+ :: forall r q+ . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)+ => SchedulerProxy q+ -> MonitorReference+ -> Eff r ()+demonitor _px = executeAndResumeOrThrow . Demonitor . force++-- | 'monitor' another process before while performing an action+-- and 'demonitor' afterwards.+--+-- @since 0.12.0+withMonitor+ :: ( HasCallStack+ , Member Interrupts r+ , SetMember Process (Process q) r+ , Member Interrupts r+ )+ => SchedulerProxy q+ -> ProcessId+ -> (MonitorReference -> Eff r a)+ -> Eff r a+withMonitor px pid eff = monitor px pid >>= \ref -> eff ref <* demonitor px ref++-- | A 'MessageSelector' for receiving either a monitor of the+-- given process or another message.+--+-- @since 0.12.0+receiveWithMonitor+ :: ( HasCallStack+ , Member Interrupts r+ , SetMember Process (Process q) r+ , Member Interrupts r+ , Typeable a+ , Show a+ )+ => SchedulerProxy q+ -> ProcessId+ -> MessageSelector a+ -> Eff r (Either ProcessDown a)+receiveWithMonitor px pid sel = withMonitor+ px+ pid+ (\ref -> receiveSelectedMessage+ px+ (Left <$> selectProcessDown ref <|> Right <$> sel)+ )++-- | A monitored process exited.+-- This message is sent to a process by the scheduler, when+-- a process that was monitored via a 'SchedulerCommand' died.+--+-- @since 0.12.0+data ProcessDown =+ ProcessDown+ { downReference :: !MonitorReference+ , downReason :: !SomeExitReason+ }+ deriving (Typeable, Generic, Eq, Ord)++-- | Trigger an 'Interrupt' for a 'ProcessDown' message.+-- The reason will be 'ProcessNotRunning'+--+-- @since 0.12.0+becauseProcessIsDown :: ProcessDown -> InterruptReason+becauseProcessIsDown = ProcessNotRunning . monitoredProcess . downReference++instance NFData ProcessDown++instance Show ProcessDown where+ showsPrec d =+ showParen+ (d>=10)+ . (\case+ ProcessDown ref reason ->+ showString "monitored process down "+ . showsPrec 11 ref . showChar ' '+ . showsPrec 11 reason+ )++-- | A 'MesssageSelector' for the 'ProcessDown' message of a specific+-- process.+--+-- @since 0.12.0+selectProcessDown :: MonitorReference -> MessageSelector ProcessDown+selectProcessDown ref0 =+ filterMessageLazy (\(ProcessDown ref _reason) -> ref0 == ref)++-- | Connect the calling process to another process, such that+-- if one of the processes crashes (i.e. 'isCrash' returns 'True'), the other+-- is shutdown with the 'ProcessExitReaon' 'LinkedProcessCrashed'.+--+-- @since 0.12.0+linkProcess+ :: forall r q+ . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)+ => SchedulerProxy q+ -> ProcessId+ -> Eff r ()+linkProcess _px = executeAndResumeOrThrow . Link . force++-- | Unlink the calling proccess from the other process.+--+-- @since 0.12.0+unlinkProcess+ :: forall r q+ . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)+ => SchedulerProxy q+ -> ProcessId+ -> Eff r ()+unlinkProcess _px = executeAndResumeOrThrow . Unlink . force++-- | Exit the process with a 'ProcessExitReaon'.+exitBecause+ :: forall r q a+ . (HasCallStack, SetMember Process (Process q) r)+ => SchedulerProxy q+ -> ExitReason 'NoRecovery+ -> Eff r a+exitBecause _ = send . Shutdown @q . force++-- | Exit the process.+exitNormally+ :: forall r q a+ . (HasCallStack, SetMember Process (Process q) r)+ => SchedulerProxy q+ -> Eff r a+exitNormally px = exitBecause px ExitNormally++-- | Exit the process with an error.+exitWithError+ :: forall r q a+ . (HasCallStack, SetMember Process (Process q) r)+ => SchedulerProxy q+ -> String+ -> Eff r a+exitWithError px = exitBecause px . NotRecovered . ProcessError++-- | Each process is identified by a single process id, that stays constant+-- throughout the life cycle of a process. Also, message sending relies on these+-- values to address messages to processes.+newtype ProcessId = ProcessId { _fromProcessId :: Int }+ deriving (Eq,Ord,Typeable,Bounded,Num, Enum, Integral, Real, NFData)++instance Read ProcessId where+ readsPrec _ ('!':rest1) =+ case reads rest1 of+ [(c, rest2)] -> [(ProcessId c, rest2)]+ _ -> []+ readsPrec _ _ = []++instance Show ProcessId where+ showsPrec _ (ProcessId !c) = showChar '!' . shows c++makeLenses ''ProcessId
src/Control/Eff/Concurrent/Process/ForkIOScheduler.hs view
@@ -18,18 +18,17 @@ , defaultMain , defaultMainWithLogChannel , ProcEff+ , InterruptableProcEff , SchedulerIO , HasSchedulerIO , forkIoScheduler ) where -import GHC.Stack-import Data.Kind ( )-import qualified Control.Exception as Exc-import qualified Control.Eff.ExceptionExtra as ExcExtra+import Control.Exception.Safe as Safe+import qualified Control.Eff.ExceptionExtra as ExcExtra () import Control.Concurrent.STM as STM-import Control.Concurrent (threadDelay, yield)+import Control.Concurrent (yield) import qualified Control.Concurrent.Async as Async import Control.Concurrent.Async (Async(..)) import Control.Eff@@ -44,24 +43,27 @@ , (>=>) ) import Control.Monad.Trans.Control (MonadBaseControl(..))-import qualified Control.Exception.Enclosed as Exceptions+import Data.Default import Data.Dynamic+import Data.Foldable+import Data.Kind ( ) import Data.Map ( Map )-import Data.Default+import qualified Data.Map as Map+import Data.Set ( Set )+import qualified Data.Set as Set+import Data.Maybe import Data.Sequence ( Seq(..) ) import qualified Data.Sequence as Seq-import Control.DeepSeq-import Control.Monad.IO.Class-import Text.Printf-import Data.Maybe-+import GHC.Stack+import Text.Printf+import System.Timeout -- * Process Types -- | A message queue of a process, contains the actual queue and maybe an -- exit reason. data MessageQ = MessageQ { _incomingMessages :: Seq Dynamic- , _shutdownRequests :: Maybe ShutdownRequest+ , _shutdownRequests :: Maybe SomeExitReason } instance Default MessageQ where def = MessageQ def def@@ -70,7 +72,8 @@ -- | Return any '_shutdownRequests' from a 'MessageQ' in a 'TVar' and -- reset the '_shutdownRequests' field to 'Nothing' in the 'TVar'.-tryTakeNextShutdownRequestSTM :: TVar MessageQ -> STM (Maybe ShutdownRequest)+tryTakeNextShutdownRequestSTM+ :: TVar MessageQ -> STM (Maybe SomeExitReason) tryTakeNextShutdownRequestSTM mqVar = do mq <- readTVar mqVar when (isJust (mq^.shutdownRequests))@@ -85,6 +88,7 @@ ProcessInfo { _processId :: ProcessId , _processState :: TVar ProcessState , _messageQ :: TVar MessageQ+ , _processLinks :: TVar (Set ProcessId) } makeLenses ''ProcessInfo@@ -97,27 +101,72 @@ data SchedulerState = SchedulerState { _nextPid :: TVar ProcessId , _processTable :: TVar (Map ProcessId ProcessInfo)- , _processCancellationTable :: TVar (Map ProcessId (Async ProcessExitReason))+ , _processCancellationTable+ :: TVar (Map ProcessId+ (Async (ExitReason 'NoRecovery)))+ , _processMonitors :: TVar (Set (MonitorReference, ProcessId))+ , _nextMonitorIndex :: TVar Int+ -- Set of monitors and monitor owners } makeLenses ''SchedulerState +-- | Add monitor: If the process is dead, enqueue a ProcessDown message into the+-- owners message queue+addMonitoring :: ProcessId -> ProcessId -> SchedulerState -> STM MonitorReference+addMonitoring target owner schedulerState = do+ mi <- readTVar (schedulerState ^. nextMonitorIndex)+ modifyTVar' (schedulerState ^. nextMonitorIndex) (+1)+ let mref = MonitorReference mi target+ when (target /= owner) $ do+ pt <- readTVar (schedulerState ^. processTable)+ if Map.member target pt then+ modifyTVar' (schedulerState ^. processMonitors) (Set.insert (mref, owner))+ else+ let pdown = (ProcessDown mref (SomeExitReason (ProcessNotRunning target)))+ in enqueueMessageOtherProcess owner (toDyn pdown) schedulerState+ return mref +removeMonitoring :: MonitorReference -> SchedulerState -> STM ()+removeMonitoring mref schedulerState =+ modifyTVar' (schedulerState ^. processMonitors)+ (Set.filter (\(ref,_) -> ref /= mref))++triggerAndRemoveMonitor :: ProcessId -> SomeExitReason -> SchedulerState -> STM ()+triggerAndRemoveMonitor downPid reason schedulerState = do+ monRefs <- readTVar (schedulerState ^. processMonitors)+ traverse_ go monRefs+ where+ go (mr, owner) =+ when (monitoredProcess mr == downPid) (do+ let pdown = ProcessDown mr reason+ enqueueMessageOtherProcess owner (toDyn pdown) schedulerState+ removeMonitoring mr schedulerState)+ -- * Process Implementation instance Show ProcessInfo where show p = "process info: " ++ show (p ^. processId) - -- | Create a new 'ProcessInfo' newProcessInfo :: HasCallStack => ProcessId -> STM ProcessInfo-newProcessInfo a = ProcessInfo a <$> newTVar ProcessBooting <*> newTVar def+newProcessInfo a =+ ProcessInfo a+ <$> newTVar ProcessBooting+ <*> newTVar def+ <*> newTVar def -- * Scheduler Implementation -- | Create a new 'SchedulerState' newSchedulerState :: HasCallStack => STM SchedulerState-newSchedulerState = SchedulerState <$> newTVar 1 <*> newTVar def <*> newTVar def+newSchedulerState =+ SchedulerState+ <$> newTVar 1+ <*> newTVar def+ <*> newTVar def+ <*> newTVar def+ <*> newTVar def -- | Create a new 'SchedulerState' run an IO action, catching all exceptions, -- and when the actions returns, clean up and kill all processes.@@ -126,22 +175,22 @@ => Eff SchedulerIO () -> Eff LoggingAndIO () withNewSchedulerState mainProcessAction =- Exceptions.tryAny (lift (atomically newSchedulerState))- >>= either- (logError . ("scheduler setup crashed " ++) . Exc.displayException)- (flip runReader- ( (logDebug "scheduler loop entered"- *> ExcExtra.liftTry @Exc.SomeException mainProcessAction- <* logDebug "scheduler loop returned")- >>= either- (logError . ("scheduler loop crashed " ++) . Exc.displayException)- (\result ->- do logDebug "scheduler cleanup begin"- Exceptions.tryAny tearDownScheduler- >>= either- (logError . ("scheduler cleanup crashed " ++) . Exc.displayException)- (const (logDebug "scheduler cleanup done"))- return result)))+ Safe.bracketWithError+ (lift (atomically newSchedulerState))+ (\exceptions schedulerState -> do+ traverse_+ (logError . ("scheduler setup crashed with: " ++) . Safe.displayException)+ exceptions+ logDebug "scheduler cleanup begin"+ runReader schedulerState tearDownScheduler+ )+ (\schedulerState -> do+ logDebug "scheduler loop entered"+ x <- runReader schedulerState mainProcessAction+ logDebug "scheduler loop returned"+ return x+ )+ where tearDownScheduler = do schedulerState <- getSchedulerState@@ -151,19 +200,21 @@ logNotice ("cancelling processes: " ++ show (toListOf (ifolded.asIndex) allProcesses)) void (liftBaseWith- (\runS -> Async.race+ (\runS -> timeout 5000000 (Async.mapConcurrently (\a -> do Async.cancel a runS (logNotice ("process cancelled: "++ show (asyncThreadId a))) )- allProcesses)- (threadDelay 5000000)))+ allProcesses+ >> runS (logNotice "all processes cancelled")))) -- | The concrete list of 'Eff'ects of processes compatible with this scheduler. -- This builds upon 'SchedulerIO'. type ProcEff = ConsProcess SchedulerIO +-- | The concrete list of the effects, that the 'Process' uses+type InterruptableProcEff = InterruptableProcess SchedulerIO -- | Type class constraint to indicate that an effect union contains the -- effects required by every process and the scheduler implementation itself.@@ -184,7 +235,7 @@ -- | Start the message passing concurrency system then execute a 'Process' on -- top of 'SchedulerIO' effect. All logging is sent to standard output.-defaultMain :: HasCallStack => Eff ProcEff () -> IO ()+defaultMain :: HasCallStack => Eff InterruptableProcEff () -> IO () defaultMain c = withAsyncLogChannel 1024 (multiMessageLogWriter ($ printLogMessage))@@ -193,48 +244,37 @@ -- | Start the message passing concurrency system then execute a 'Process' on -- top of 'SchedulerIO' effect. All logging is sent to standard output. defaultMainWithLogChannel- :: HasCallStack => Eff ProcEff () -> LogChannel LogMessage -> IO ()+ :: HasCallStack => Eff InterruptableProcEff () -> LogChannel LogMessage -> IO () defaultMainWithLogChannel = handleLoggingAndIO_ . schedule -- | A 'SchedulerProxy' for 'SchedulerIO' forkIoScheduler :: SchedulerProxy SchedulerIO forkIoScheduler = SchedulerProxy --- | Run a process effect that returns a 'ProcessExitReason' catching all--- exceptions. If a runtime exception (e.g. 'SomeException') is caught, it--- will be '<>'-tryReplaceReason- :: (MonadIO (Eff e), MonadBaseControl IO (Eff e), Lifted IO e, HasCallStack)- => Eff e ProcessExitReason- -> Eff e ProcessExitReason-tryReplaceReason e = withFrozenCallStack $ do- let here = prettyCallStack callStack- res <- ExcExtra.liftTry @Exc.SomeException (ExcExtra.liftTry @Async.AsyncCancelled e) -- Exceptions.tryAny e- case res of- Left s ->- liftIO (Exc.evaluate (force (ProcessCaughtIOException here (Exc.displayException s))))- Right (Left _) ->- return SchedulerShuttingDown- Right (Right res1) -> return res1 -- ** MessagePassing execution handleProcess :: (HasLogging IO SchedulerIO, HasCallStack) => ProcessInfo- -> Eff ProcEff ProcessExitReason- -> Eff SchedulerIO ProcessExitReason+ -> Eff ProcEff (ExitReason 'NoRecovery)+ -> Eff SchedulerIO (ExitReason 'NoRecovery) handleProcess myProcessInfo =- handle_relay_s- 0- (const return)- (\ !nextRef !request k ->- stepProcessInterpreter nextRef request k return- )+ handle_relay_s+ 0+ (const return)+ (\ !nextRef !request k ->+ stepProcessInterpreter nextRef request k return+ ) where myPid = myProcessInfo ^. processId myProcessStateVar = myProcessInfo ^. processState setMyProcessState = lift . atomically . setMyProcessStateSTM+ -- DEBUG variant:+ -- setMyProcessState st = do+ -- oldSt <- lift (atomically (readTVar myProcessStateVar <* setMyProcessStateSTM st))+ -- logDebug ("state change: "++ show oldSt ++ " -> " ++ show st)+ setMyProcessStateSTM = writeTVar myProcessStateVar myMessageQVar = myProcessInfo ^. messageQ @@ -252,8 +292,8 @@ diskontinueWith :: forall a . HasCallStack- => Arr SchedulerIO ProcessExitReason a- -> Arr SchedulerIO ProcessExitReason a+ => Arr SchedulerIO (ExitReason 'NoRecovery) a+ -> Arr SchedulerIO (ExitReason 'NoRecovery) a diskontinueWith diskontinue !reason = do setMyProcessState ProcessShuttingDown diskontinue reason@@ -264,30 +304,40 @@ => Int -> Process SchedulerIO v -> (Int -> Arr SchedulerIO v a)- -> Arr SchedulerIO ProcessExitReason a+ -> Arr SchedulerIO (ExitReason 'NoRecovery) a -> Eff SchedulerIO a stepProcessInterpreter !nextRef !request kontinue diskontinue = -- handle process shutdown requests: -- 1. take process exit reason -- 2. set process state to ProcessShuttingDown- -- 3. apply kontinue to (Right ShutdownRequested)+ -- 3. apply kontinue to (Right Interrupted) -- tryTakeNextShutdownRequest- >>= maybe noShutdownRequested onShutdownRequested+ >>= maybe noShutdownRequested+ (either onShutdownRequested onInterruptRequested+ . fromSomeExitReason) where tryTakeNextShutdownRequest = lift (atomically (tryTakeNextShutdownRequestSTM myMessageQVar)) onShutdownRequested shutdownRequest = do- logDebug "shutdownRequested"+ logDebug ("shutdown requested: " ++ show shutdownRequest) setMyProcessState ProcessShuttingDown interpretRequestAfterShutdownRequest- (kontinueWith kontinue nextRef) (diskontinueWith diskontinue) shutdownRequest request + onInterruptRequested interruptRequest = do+ logDebug ("interrupt requested: " ++ show interruptRequest)+ setMyProcessState ProcessShuttingDown+ interpretRequestAfterInterruptRequest+ (kontinueWith kontinue nextRef)+ (diskontinueWith diskontinue)+ interruptRequest+ request+ noShutdownRequested = do setMyProcessState ProcessBusy interpretRequest@@ -303,59 +353,174 @@ interpretRequestAfterShutdownRequest :: forall v a . HasCallStack+ => Arr SchedulerIO (ExitReason 'NoRecovery) a+ -> (ExitReason 'NoRecovery)+ -> Process SchedulerIO v+ -> Eff SchedulerIO a+ interpretRequestAfterShutdownRequest diskontinue shutdownRequest =+ \case+ SendMessage _ _ -> diskontinue shutdownRequest+ SendInterrupt _ _ -> diskontinue shutdownRequest+ SendShutdown toPid r ->+ if toPid == myPid+ then diskontinue r+ else diskontinue shutdownRequest+ Spawn _ -> diskontinue shutdownRequest+ SpawnLink _ -> diskontinue shutdownRequest+ ReceiveSelectedMessage _ -> diskontinue shutdownRequest+ FlushMessages -> diskontinue shutdownRequest+ SelfPid -> diskontinue shutdownRequest+ MakeReference -> diskontinue shutdownRequest+ YieldProcess -> diskontinue shutdownRequest+ Shutdown r -> diskontinue r+ GetProcessState _ -> diskontinue shutdownRequest+ Monitor _ -> diskontinue shutdownRequest+ Demonitor _ -> diskontinue shutdownRequest+ Link _ -> diskontinue shutdownRequest+ Unlink _ -> diskontinue shutdownRequest++ interpretRequestAfterInterruptRequest+ :: forall v a+ . HasCallStack => Arr SchedulerIO v a- -> Arr SchedulerIO ProcessExitReason a- -> ShutdownRequest+ -> Arr SchedulerIO (ExitReason 'NoRecovery) a+ -> (ExitReason 'Recoverable) -> Process SchedulerIO v -> Eff SchedulerIO a- interpretRequestAfterShutdownRequest kontinue diskontinue shutdownRequest =+ interpretRequestAfterInterruptRequest kontinue diskontinue interruptRequest = \case- SendMessage _ _ -> kontinue (ShutdownRequested shutdownRequest)+ SendMessage _ _ -> kontinue (Interrupted interruptRequest)+ SendInterrupt _ _ -> kontinue (Interrupted interruptRequest) SendShutdown toPid r -> if toPid == myPid- then diskontinue (ProcessShutDown r)- else kontinue (ShutdownRequested shutdownRequest)- Spawn _ -> kontinue (ShutdownRequested shutdownRequest)- ReceiveSelectedMessage _ -> kontinue (ShutdownRequested shutdownRequest)- SelfPid -> kontinue (ShutdownRequested shutdownRequest)- MakeReference -> kontinue (ShutdownRequested shutdownRequest)- YieldProcess -> kontinue (ShutdownRequested shutdownRequest)- Shutdown r -> diskontinue (ProcessShutDown r)- RaiseError e -> diskontinue (ProcessRaisedError e)+ then diskontinue r+ else kontinue (Interrupted interruptRequest)+ Spawn _ -> kontinue (Interrupted interruptRequest)+ SpawnLink _ -> kontinue (Interrupted interruptRequest)+ ReceiveSelectedMessage _ -> kontinue (Interrupted interruptRequest)+ FlushMessages -> kontinue (Interrupted interruptRequest)+ SelfPid -> kontinue (Interrupted interruptRequest)+ MakeReference -> kontinue (Interrupted interruptRequest)+ YieldProcess -> kontinue (Interrupted interruptRequest)+ Shutdown r -> diskontinue r+ GetProcessState _ -> kontinue (Interrupted interruptRequest)+ Monitor _ -> kontinue (Interrupted interruptRequest)+ Demonitor _ -> kontinue (Interrupted interruptRequest)+ Link _ -> kontinue (Interrupted interruptRequest)+ Unlink _ -> kontinue (Interrupted interruptRequest) interpretRequest :: forall v a . HasCallStack => (Int -> Arr SchedulerIO v a)- -> Arr SchedulerIO ProcessExitReason a+ -> Arr SchedulerIO (ExitReason 'NoRecovery) a -> Int -> Process SchedulerIO v -> Eff SchedulerIO a interpretRequest kontinue diskontinue nextRef = \case- SendMessage toPid msg -> interpretSend toPid msg >>= kontinue nextRef . ResumeWith+ SendMessage toPid msg -> interpretSend toPid msg >>= kontinue nextRef . ResumeWith+ SendInterrupt toPid msg ->+ if toPid == myPid+ then kontinue nextRef (Interrupted msg)+ else interpretSendShutdownOrInterrupt toPid (SomeExitReason msg)+ >>= kontinue nextRef . ResumeWith SendShutdown toPid msg -> if toPid == myPid- then kontinue nextRef (ShutdownRequested msg)- else interpretSendShutdown toPid msg >>= kontinue nextRef . ResumeWith- Spawn child -> spawnNewProcess child >>= kontinue nextRef . ResumeWith . fst- ReceiveSelectedMessage f -> interpretReceive f >>= kontinue nextRef+ then diskontinue msg+ else interpretSendShutdownOrInterrupt toPid (SomeExitReason msg)+ >>= kontinue nextRef . ResumeWith+ Spawn child -> spawnNewProcess Nothing child+ >>= kontinue nextRef . ResumeWith . fst+ SpawnLink child -> spawnNewProcess (Just myProcessInfo) child+ >>= kontinue nextRef . ResumeWith . fst+ ReceiveSelectedMessage f -> interpretReceive f >>= either diskontinue (kontinue nextRef)+ FlushMessages -> interpretFlush >>= kontinue nextRef SelfPid -> kontinue nextRef (ResumeWith myPid) MakeReference -> kontinue (nextRef + 1) (ResumeWith nextRef) YieldProcess -> kontinue nextRef (ResumeWith ())- Shutdown r -> diskontinue (ProcessShutDown r)- RaiseError e -> diskontinue (ProcessRaisedError e)+ Shutdown r -> diskontinue r+ GetProcessState toPid -> interpretGetProcessState toPid >>= kontinue nextRef . ResumeWith+ Monitor target -> interpretMonitor target >>= kontinue nextRef . ResumeWith+ Demonitor ref -> interpretDemonitor ref >>= kontinue nextRef . ResumeWith+ Link toPid -> interpretLink toPid+ >>= kontinue nextRef . either Interrupted ResumeWith+ Unlink toPid -> interpretUnlink toPid >>= kontinue nextRef . ResumeWith where+ interpretMonitor !target = do+ setMyProcessState ProcessBusyMonitoring+ schedulerState <- getSchedulerState+ lift (atomically (addMonitoring target myPid schedulerState)) - interpretSend !toPid !msg =+ interpretDemonitor !ref = do+ setMyProcessState ProcessBusyMonitoring+ schedulerState <- getSchedulerState+ lift (atomically (removeMonitoring ref schedulerState))++ interpretUnlink !toPid = do+ setMyProcessState ProcessBusyUnlinking+ schedulerState <- getSchedulerState+ let procInfosVar = schedulerState^.processTable+ lift $ atomically $ do+ procInfos <- readTVar procInfosVar+ traverse_+ (\toProcInfo ->+ modifyTVar' (toProcInfo ^. processLinks) (Set.delete myPid))+ (procInfos ^. at toPid)+ modifyTVar' (myProcessInfo ^. processLinks) (Set.delete toPid)++ interpretGetProcessState !toPid = do+ setMyProcessState ProcessBusy+ schedulerState <- getSchedulerState+ let procInfosVar = schedulerState^.processTable+ lift $ atomically $ do+ procInfos <- readTVar procInfosVar+ traverse (\toProcInfo -> readTVar (toProcInfo ^. processState))+ (procInfos ^. at toPid)++ interpretLink !toPid = do+ setMyProcessState ProcessBusyLinking+ schedulerState <- getSchedulerState+ let procInfosVar = schedulerState^.processTable+ lift $ atomically $ do+ procInfos <- readTVar procInfosVar+ case procInfos ^. at toPid of+ Just toProcInfo -> do+ modifyTVar' (toProcInfo ^. processLinks) (Set.insert myPid)+ modifyTVar' (myProcessInfo ^. processLinks) (Set.insert toPid)+ return (Right ())+ Nothing ->+ return (Left (LinkedProcessCrashed toPid))++ interpretSend !toPid msg = setMyProcessState ProcessBusySending *>- getSchedulerState >>= lift . atomically . enqueueMessageOtherProcess toPid msg+ getSchedulerState+ >>= lift+ . atomically+ . enqueueMessageOtherProcess toPid msg - interpretSendShutdown !toPid !msg =- setMyProcessState ProcessBusySendingShutdown *>- getSchedulerState >>= lift . atomically . enqueueShutdownRequest toPid msg+ interpretSendShutdownOrInterrupt !toPid !msg =+ setMyProcessState+ (either+ (const ProcessBusySendingShutdown)+ (const ProcessBusySendingInterrupt)+ (fromSomeExitReason msg))+ *> getSchedulerState+ >>= lift+ . atomically+ . enqueueShutdownRequest toPid msg - interpretReceive :: MessageSelector b -> Eff SchedulerIO (ResumeProcess b)+ interpretFlush :: Eff SchedulerIO (ResumeProcess [Dynamic])+ interpretFlush = do+ setMyProcessState ProcessBusyReceiving+ lift $ atomically $ do+ myMessageQ <- readTVar myMessageQVar+ modifyTVar' myMessageQVar (incomingMessages .~ Seq.Empty)+ return (ResumeWith (toList (myMessageQ ^. incomingMessages)))++ interpretReceive+ :: MessageSelector b+ -> Eff SchedulerIO (Either (ExitReason 'NoRecovery) (ResumeProcess b)) interpretReceive f = do setMyProcessState ProcessBusyReceiving lift $ atomically $ do@@ -366,10 +531,14 @@ Nothing -> retry Just (selectedMessage, otherMessages) -> do modifyTVar' myMessageQVar (incomingMessages .~ otherMessages)- return (ResumeWith selectedMessage)+ return (Right (ResumeWith selectedMessage)) Just shutdownRequest -> do modifyTVar' myMessageQVar (shutdownRequests .~ Nothing)- return (ShutdownRequested shutdownRequest)+ case fromSomeExitReason shutdownRequest of+ Left sr ->+ return (Left sr)+ Right ir ->+ return (Right (Interrupted ir)) where partitionMessages Seq.Empty _acc = Nothing@@ -383,14 +552,14 @@ -- effect and a 'LogChannel' for concurrent logging. schedule :: (HasLogging IO SchedulerIO, HasCallStack)- => Eff ProcEff ()+ => Eff InterruptableProcEff () -> Eff LoggingAndIO () schedule procEff = liftBaseWith (\runS -> Async.withAsync (runS $ withNewSchedulerState $ do- (_, mainProcAsync) <- spawnNewProcess $ do+ (_, mainProcAsync) <- spawnNewProcess Nothing $ do logNotice "++++++++ main process started ++++++++"- procEff+ provideInterruptsShutdown procEff logNotice "++++++++ main process returned ++++++++" lift (void (Async.wait mainProcAsync)) )@@ -402,78 +571,138 @@ spawnNewProcess :: (HasLogging IO SchedulerIO, HasCallStack)- => Eff ProcEff ()- -> Eff SchedulerIO (ProcessId, Async ProcessExitReason)-spawnNewProcess mfa = do+ => Maybe ProcessInfo+ -> Eff ProcEff ()+ -> Eff SchedulerIO (ProcessId, Async (ExitReason 'NoRecovery))+spawnNewProcess mlinkedParent mfa = do schedulerState <- getSchedulerState- liftBaseWith- (\inScheduler -> do+ procInfo <- allocateProcInfo schedulerState+ traverse_ (linkToParent procInfo) mlinkedParent+ procAsync <- doForkProc procInfo schedulerState+ return (procInfo ^. processId, procAsync)+ where+ linkToParent toProcInfo parent = do+ let toPid = toProcInfo ^. processId+ parentPid = parent ^. processId+ lift $ atomically $ do+ modifyTVar' (toProcInfo ^. processLinks) (Set.insert parentPid)+ modifyTVar' (parent ^. processLinks) (Set.insert toPid) - let nextPidVar = schedulerState ^. nextPid- processInfosVar = schedulerState ^. processTable- cancellationsVar = schedulerState ^. processCancellationTable+ allocateProcInfo schedulerState =+ lift (atomically (do+ let nextPidVar = schedulerState ^. nextPid+ processInfosVar = schedulerState ^. processTable+ pid <- readTVar nextPidVar+ modifyTVar' nextPidVar (+1)+ procInfo <- newProcessInfo pid+ modifyTVar' processInfosVar (at pid ?~ procInfo)+ return procInfo+ )) - procInfo <- atomically (do- pid <- readTVar nextPidVar- modifyTVar' nextPidVar (+1)- procInfo <- newProcessInfo pid- modifyTVar' processInfosVar (at pid ?~ procInfo)- return procInfo)+ logAppendProcInfo pid =+ let addProcessId = over lmProcessId+ (maybe (Just (printf "% 9s" (show pid))) Just)+ in traverseLogMessages+ (traverse setLogMessageThreadId+ >=> traverse (return . addProcessId)) - let pid = procInfo ^. processId- logAppendProcInfo =- traverseLogMessages- ( lift . traverse setLogMessageThreadId- >=> traverse (return . addProcessId))- where- addProcessId =- over lmProcessId (maybe (Just (printf "% 9s" (show pid))) Just)+ triggerProcessLinksAndMonitors !pid !reason !linkSetVar = do+ schedulerState <- getSchedulerState+ lift $ atomically $+ triggerAndRemoveMonitor pid (SomeExitReason reason) schedulerState+ let exitSeverity = toExitSeverity reason+ sendIt !linkedPid = do+ let msg = SomeExitReason (LinkedProcessCrashed pid)+ lift $ atomically $ do+ procInfos <- readTVar (schedulerState ^. processTable)+ let mLinkedProcInfo = procInfos ^? at linkedPid . _Just+ case mLinkedProcInfo of+ Nothing ->+ return (Left linkedPid)+ Just linkedProcInfo ->+ let linkedMsgQVar = linkedProcInfo ^. messageQ+ linkedLinkSetVar = linkedProcInfo ^. processLinks+ in do linkedLinkSet <- readTVar linkedLinkSetVar+ if Set.member pid linkedLinkSet+ then do+ writeTVar linkedLinkSetVar+ (Set.delete pid linkedLinkSet)+ when (exitSeverity == Crash)+ (modifyTVar' linkedMsgQVar+ (shutdownRequests ?~ msg))+ return (Right linkedPid)+ else+ return (Left linkedPid)+ linkedPids <- lift (atomically (do linkSet <- readTVar linkSetVar+ writeTVar linkSetVar def+ return linkSet))+ res <- traverse sendIt (toList linkedPids)+ traverse_ (logDebug . either (("linked process no found: " ++) . show)+ (("sent shutdown to linked process: " ++) . show))+ res - procAsync <- Async.async- (inScheduler- (logAppendProcInfo- (do- logDebug "enter process"- e <- tryReplaceReason- (handleProcess procInfo (mfa *> return ProcessReturned))- logProcessExit e- lastProcessState <- lift (readTVarIO (procInfo ^. processState))- when ( lastProcessState /= ProcessShuttingDown- && lastProcessState /= ProcessIdle )- (logNotice- ("process was not in shutting down state when it exitted: "- ++ show lastProcessState))- return e- )- )- <* atomically- (do modifyTVar' processInfosVar (at pid .~ Nothing)- modifyTVar' cancellationsVar (at pid .~ Nothing)- )- )- atomically (modifyTVar' cancellationsVar (at pid ?~ procAsync))- return (pid, procAsync))- >>= restoreM+ doForkProc :: ProcessInfo -> SchedulerState -> Eff SchedulerIO (Async (ExitReason 'NoRecovery))+ doForkProc procInfo schedulerState =+ restoreM =<< liftBaseWith+ (\inScheduler -> do+ let cancellationsVar = schedulerState ^. processCancellationTable+ processInfosVar = schedulerState ^. processTable+ pid = procInfo ^. processId+ procAsync <- Async.async (+ inScheduler (logAppendProcInfo pid+ (Safe.bracketWithError+ (logDebug "enter process")+ (\mExc () -> do+ lift (atomically+ (do modifyTVar' processInfosVar (at pid .~ Nothing)+ modifyTVar' cancellationsVar (at pid .~ Nothing)))+ traverse_+ (\exc -> logExitAndTriggerLinksAndMonitors+ (exitReasonFromException exc)+ pid)+ mExc+ )+ (const+ (do res <- handleProcess procInfo (mfa >> return ExitNormally)+ logExitAndTriggerLinksAndMonitors res pid))+ )))+ atomically (modifyTVar' cancellationsVar (at pid ?~ procAsync))+ return procAsync)+ where+ exitReasonFromException exc =+ case Safe.fromException exc of+ Just Async.AsyncCancelled ->+ Killed --- Scheduler Accessor+ Nothing ->+ UnexpectedException+ (prettyCallStack callStack)+ (Safe.displayException exc) + logExitAndTriggerLinksAndMonitors reason pid =+ do triggerProcessLinksAndMonitors pid reason (procInfo ^. processLinks)+ logProcessExit reason+ return reason++-- * Scheduler Accessor+ getSchedulerState :: HasSchedulerIO r => Eff r SchedulerState getSchedulerState = ask enqueueMessageOtherProcess ::- HasCallStack => ProcessId -> Dynamic -> SchedulerState -> STM Bool+ HasCallStack => ProcessId -> Dynamic -> SchedulerState -> STM () enqueueMessageOtherProcess toPid msg schedulerState = view (at toPid) <$> readTVar (schedulerState ^. processTable)- >>= maybe (return False)+ >>= maybe (return ()) (\toProcessTable -> do modifyTVar' (toProcessTable ^. messageQ ) (incomingMessages %~ (:|> msg))- return True)+ return ()) enqueueShutdownRequest ::- HasCallStack => ProcessId -> ShutdownRequest -> SchedulerState -> STM Bool+ HasCallStack => ProcessId -> SomeExitReason -> SchedulerState -> STM () enqueueShutdownRequest toPid msg schedulerState = view (at toPid) <$> readTVar (schedulerState ^. processTable)- >>= maybe (return False)+ >>= maybe (return ()) (\toProcessTable -> do modifyTVar' (toProcessTable ^. messageQ ) (shutdownRequests ?~ msg)- return True)+ return ())
src/Control/Eff/Concurrent/Process/Interactive.hs view
@@ -12,7 +12,7 @@ -- -- >>> s <- forkInteractiveScheduler Control.Eff.Concurrent.Process.SingleThreadedScheduler.defaultMain ----- >>> fooPid <- submit s (spawn (foreverCheap (receiveMessage SP >>= (logMsg . fromMaybe "Huh!??" . fromDynamic))))+-- >>> fooPid <- submit s (spawn (foreverCheap (receiveAnyMessage SP >>= (logMsg . fromMaybe "Huh!??" . fromDynamic)))) -- -- >>> fooPid -- <0.1.0>@@ -45,7 +45,7 @@ import Control.Monad import Data.Foldable import Data.Typeable ( Typeable )-import Control.DeepSeq+import Control.DeepSeq import System.Timeout -- | Contains the communication channels to interact with a scheduler running in@@ -53,14 +53,14 @@ newtype SchedulerSession r = SchedulerSession (TMVar (SchedulerQueue r)) newtype SchedulerQueue r =- SchedulerQueue (TChan (Eff (Process r ': r) (Maybe String)))+ SchedulerQueue (TChan (Eff (InterruptableProcess r) (Maybe String))) -- | Fork a scheduler with a process that communicates with it via 'MVar', -- which is also the reason for the @Lift IO@ constraint. forkInteractiveScheduler :: forall r . (SetMember Lift (Lift IO) r)- => (Eff (Process r ': r) () -> IO ())+ => (Eff (InterruptableProcess r) () -> IO ()) -> IO (SchedulerSession r) forkInteractiveScheduler ioScheduler = do inQueue <- newTChanIO@@ -76,6 +76,8 @@ ) return (SchedulerSession queueVar) where+ readEvalPrintLoop+ :: TMVar (SchedulerQueue r) -> Eff (InterruptableProcess r) () readEvalPrintLoop queueVar = do nextActionOrExit <- readAction case nextActionOrExit of@@ -108,7 +110,7 @@ :: forall r a . (SetMember Lift (Lift IO) r) => SchedulerSession r- -> Eff (Process r ': r) a+ -> Eff (InterruptableProcess r) a -> IO a submit (SchedulerSession qVar) theAction = do mResVar <- timeout 5000000 $ atomically@@ -131,7 +133,7 @@ -- | Combination of 'submit' and 'cast'. submitCast :: forall o r- . (SetMember Lift (Lift IO) r, Typeable o)+ . (SetMember Lift (Lift IO) r, Typeable o, Member Interrupts r) => SchedulerSession r -> Server o -> Api o 'Asynchronous@@ -141,7 +143,13 @@ -- | Combination of 'submit' and 'cast'. submitCall :: forall o q r- . (SetMember Lift (Lift IO) r, Typeable o, Typeable q, NFData q)+ . ( SetMember Lift (Lift IO) r+ , Typeable o+ , Typeable q+ , NFData q+ , Show q+ , Member Interrupts r+ ) => SchedulerSession r -> Server o -> Api o ( 'Synchronous q)
src/Control/Eff/Concurrent/Process/SingleThreadedScheduler.hs view
@@ -5,9 +5,8 @@ , scheduleIO , scheduleMonadIOEff , scheduleIOWithLogging- , defaultMain+ , defaultMainSingleThreaded , singleThreadedIoScheduler- , LoggingAndIo ) where @@ -21,22 +20,169 @@ import Control.Lens hiding ( (|>) , Empty )-import Control.Monad ( void )+import Control.Monad ( void+ , when+ ) import Control.Monad.IO.Class import qualified Data.Sequence as Seq import Data.Sequence ( Seq(..) ) import qualified Data.Map.Strict as Map+import qualified Data.Set as Set import GHC.Stack import Data.Kind ( ) import Data.Dynamic-import Data.Maybe+import Data.Foldable+import Data.Monoid+import qualified Control.Monad.State.Strict as State +-- -----------------------------------------------------------------------------+-- STS+-- ----------------------------------------------------------------------------- +data STS r m = STS+ { _nextPid :: !ProcessId+ , _nextRef :: !Int+ , _msgQs :: !(Map.Map ProcessId (Seq Dynamic))+ , _monitors :: !(Set.Set (MonitorReference, ProcessId))+ , _processLinks :: !(Set.Set (ProcessId, ProcessId))+ , _runEff :: (forall a . Eff r a -> m a)+ , _yieldEff :: m ()+ }++initStsMainProcess :: (forall a . Eff r a -> m a) -> m () -> STS r m+initStsMainProcess = STS 1 0 (Map.singleton 0 Seq.empty) Set.empty Set.empty++makeLenses ''STS++instance Show (STS r m) where+ showsPrec d sts = showParen+ (d >= 10)+ ( showString "STS "+ . showString "nextRef: "+ . shows (_nextRef sts)+ . showString " msgQs: "+ . appEndo+ (foldMap+ (\(pid, msgs) -> Endo (showString " " . shows pid . showString ": ")+ <> foldMap (\m -> Endo (shows (dynTypeRep m))) (toList msgs)+ )+ (sts ^.. msgQs . itraversed . withIndex )+ )+ )++dropMsgQ :: ProcessId -> STS r m -> STS r m+dropMsgQ pid = msgQs . at pid .~ Nothing++getProcessState :: ProcessId -> STS r m -> Maybe ProcessState+getProcessState pid sts = toPS <$> sts ^. msgQs . at pid+ where+ toPS Empty = ProcessIdle+ toPS (_ :<| _) = ProcessBusy -- TODO get more detailed state++incRef :: STS r m -> (Int, STS r m)+incRef sts = (sts ^. nextRef, sts & nextRef %~ (+ 1))++enqueueMsg :: ProcessId -> Dynamic -> STS r m -> STS r m+enqueueMsg toPid msg = msgQs . at toPid . _Just %~ (:|> msg)++newProcessQ :: Maybe ProcessId -> STS r m -> (ProcessId, STS r m)+newProcessQ parentLink sts =+ ( sts ^. nextPid+ , let stsQ =+ sts & nextPid %~ (+ 1) & msgQs . at (sts ^. nextPid) ?~ Seq.empty+ in case parentLink of+ Nothing -> stsQ+ Just pid ->+ let (Nothing, stsQL) = addLink pid (sts ^. nextPid) stsQ in stsQL+ )++flushMsgs :: ProcessId -> STS m r -> ([Dynamic], STS m r)+flushMsgs pid = State.runState $ do+ msgs <- msgQs . at pid . _Just <<.= Empty+ return (toList msgs)++receiveMsg+ :: ProcessId -> MessageSelector a -> STS m r -> Maybe (Maybe (a, STS m r))+receiveMsg pid messageSelector sts = case sts ^. msgQs . at pid of+ Nothing -> Nothing+ Just msgQ -> Just $ case partitionMessages msgQ Empty of+ Nothing -> Nothing+ Just (result, otherMessages) ->+ Just (result, sts & msgQs . at pid . _Just .~ otherMessages)+ where+ partitionMessages Empty _acc = Nothing+ partitionMessages (m :<| msgRest) acc = maybe+ (partitionMessages msgRest (acc :|> m))+ (\res -> Just (res, acc Seq.>< msgRest))+ (runMessageSelector messageSelector m)++-- | Add monitor: If the process is dead, enqueue a ProcessDown message into the+-- owners message queue+addMonitoring+ :: ProcessId -> ProcessId -> STS m r -> (MonitorReference, STS m r)+addMonitoring owner target = State.runState $ do+ mi <- State.state incRef+ let mref = MonitorReference mi target+ when (target /= owner) $ do+ pt <- use msgQs+ if Map.member target pt+ then monitors %= Set.insert (mref, owner)+ else+ let pdown =+ (ProcessDown mref (SomeExitReason (ProcessNotRunning target)))+ in State.modify' (enqueueMsg owner (toDyn pdown))+ return mref++removeMonitoring :: MonitorReference -> STS m r -> STS m r+removeMonitoring mref = monitors %~ Set.filter (\(ref, _) -> ref /= mref)++triggerAndRemoveMonitor :: ProcessId -> SomeExitReason -> STS m r -> STS m r+triggerAndRemoveMonitor downPid reason = State.execState $ do+ monRefs <- use monitors+ traverse_ go monRefs+ where+ go (mr, owner) = when+ (monitoredProcess mr == downPid)+ (let pdown = ProcessDown mr reason+ in State.modify' (enqueueMsg owner (toDyn pdown) . removeMonitoring mr)+ )++addLink :: ProcessId -> ProcessId -> STS m r -> (Maybe InterruptReason, STS m r)+addLink fromPid toPid = State.runState $ do+ hasToPid <- use (msgQs . to (Map.member toPid))+ if hasToPid+ then do+ let (a, b) =+ if fromPid <= toPid then (fromPid, toPid) else (toPid, fromPid)+ processLinks %= Set.insert (a, b)+ return Nothing+ else return (Just (LinkedProcessCrashed toPid))++removeLinksTo :: ProcessId -> STS m r -> ([ProcessId], STS m r)+removeLinksTo pid sts = flip State.runState sts $ do+ pl <- use processLinks+ let aPids = pl ^.. folded . filtered (\(_, b) -> b == pid) . _1+ let bPids = pl ^.. folded . filtered (\(a, _) -> a == pid) . _2+ processLinks %= Set.filter (\(a, b) -> a /= pid && b /= pid)+ return (aPids ++ bPids)++kontinue :: STS r m -> (ResumeProcess a -> Eff r a1) -> a -> m a1+kontinue sts k x = (sts ^. runEff) (k (ResumeWith x))++diskontinue :: STS r m -> (ResumeProcess v -> Eff r a) -> InterruptReason -> m a+diskontinue sts k e = (sts ^. runEff) (k (Interrupted e))++-- -----------------------------------------------------------------------------+-- Meat Of The Thing+-- -----------------------------------------------------------------------------+ -- | Like 'schedule' but /pure/. The @yield@ effect is just @return ()@. -- @schedulePure == runIdentity . 'scheduleM' (Identity . run) (return ())@ -- -- @since 0.3.0.2-schedulePure :: Eff (ConsProcess '[Logs LogMessage]) a -> Either String a+schedulePure+ :: Eff (InterruptableProcess '[Logs LogMessage]) a+ -> Either (ExitReason 'NoRecovery) a schedulePure e = run (scheduleM ignoreLogs (return ()) e) -- | Invoke 'schedule' with @lift 'Control.Concurrent.yield'@ as yield effect.@@ -46,16 +192,18 @@ 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)+ -> Eff (InterruptableProcess r) a+ -> m (Either (ExitReason 'NoRecovery) a)+scheduleIO r = scheduleM (runLift . r) (liftIO yield) -- | Invoke 'schedule' with @lift 'Control.Concurrent.yield'@ as yield effect. -- @scheduleMonadIOEff == 'scheduleM' id (liftIO 'yield')@ -- -- @since 0.3.0.2 scheduleMonadIOEff- :: MonadIO (Eff r) => Eff (ConsProcess r) a -> Eff r (Either String a)+ :: MonadIO (Eff r)+ => Eff (InterruptableProcess r) a+ -> Eff r (Either (ExitReason 'NoRecovery) a) scheduleMonadIOEff = -- schedule (lift yield) scheduleM id (liftIO yield) @@ -70,8 +218,8 @@ scheduleIOWithLogging :: (NFData l) => LogWriter l IO- -> Eff (ConsProcess '[Logs l, LogWriterReader l IO, Lift IO]) a- -> IO (Either String a)+ -> Eff (InterruptableProcess '[Logs l, LogWriterReader l IO, Lift IO]) a+ -> IO (Either (ExitReason 'NoRecovery) a) scheduleIOWithLogging h = scheduleIO (writeLogs h) -- | Handle the 'Process' effect, as well as all lower effects using an effect handler function.@@ -97,223 +245,303 @@ -> 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- 0- (Map.singleton 0 Seq.empty)- (Seq.singleton (y, 0))+ -> Eff (InterruptableProcess r) a+ -> m (Either (ExitReason 'NoRecovery) a)+scheduleM r y e = do+ c <- runAsCoroutinePure r (provideInterruptsShutdown e)+ handleProcess (initStsMainProcess r y) (Seq.singleton (c, 0)) -- | Internal data structure that is part of the coroutine based scheduler -- implementation. data OnYield r a where+ OnFlushMessages :: (ResumeProcess [Dynamic] -> Eff r (OnYield r a))+ -> OnYield r a OnYield :: (ResumeProcess () -> Eff r (OnYield r a)) -> OnYield r a OnSelf :: (ResumeProcess ProcessId -> Eff r (OnYield r a)) -> OnYield r a- OnSpawn :: Eff (Process r ': r) ()+ OnSpawn :: Bool+ -> Eff (Process r ': r) () -> (ResumeProcess ProcessId -> Eff r (OnYield r a)) -> OnYield r a OnDone :: !a -> OnYield r a- OnShutdown :: ShutdownRequest -> OnYield r a- OnRaiseError :: !String -> OnYield r a+ OnShutdown :: ExitReason 'NoRecovery -> OnYield r a+ OnInterrupt :: ExitReason 'Recoverable+ -> (ResumeProcess b -> Eff r (OnYield r a))+ -> OnYield r a OnSend :: !ProcessId -> !Dynamic- -> (ResumeProcess Bool -> Eff r (OnYield r a))+ -> (ResumeProcess () -> Eff r (OnYield r a)) -> OnYield r a OnRecv :: MessageSelector b -> (ResumeProcess b -> Eff r (OnYield r a)) -> OnYield r a- OnSendShutdown :: !ProcessId -> !ShutdownRequest -> (ResumeProcess Bool -> Eff r (OnYield r a)) -> OnYield r a+ OnGetProcessState+ :: ProcessId+ -> (ResumeProcess (Maybe ProcessState) -> Eff r (OnYield r a))+ -> OnYield r a+ OnSendShutdown :: !ProcessId -> ExitReason 'NoRecovery+ -> (ResumeProcess () -> Eff r (OnYield r a)) -> OnYield r a+ OnSendInterrupt :: !ProcessId -> ExitReason 'Recoverable+ -> (ResumeProcess () -> Eff r (OnYield r a)) -> OnYield r a OnMakeReference :: (ResumeProcess Int -> Eff r (OnYield r a)) -> OnYield r a+ OnMonitor+ :: ProcessId+ -> (ResumeProcess MonitorReference -> Eff r (OnYield r a))+ -> OnYield r a+ OnDemonitor+ :: MonitorReference+ -> (ResumeProcess () -> Eff r (OnYield r a))+ -> OnYield r a+ OnLink+ :: ProcessId+ -> (ResumeProcess () -> Eff r (OnYield r a))+ -> OnYield r a+ OnUnlink+ :: ProcessId+ -> (ResumeProcess () -> Eff r (OnYield r a))+ -> OnYield r a +instance Show (OnYield r a) where+ show =+ \case+ OnFlushMessages _ -> "OnFlushMessages"+ OnYield _ -> "OnYield"+ OnSelf _ -> "OnSelf"+ OnSpawn False _ _ -> "OnSpawn"+ OnSpawn True _ _ -> "OnSpawn (link)"+ OnDone _ -> "OnDone"+ OnShutdown e -> "OnShutdown " ++ show e+ OnInterrupt e _ -> "OnInterrupt "++ show e+ OnSend toP _ _ -> "OnSend " ++ show toP+ OnRecv _ _ -> "OnRecv"+ OnGetProcessState p _ -> "OnGetProcessState " ++ show p+ OnSendShutdown p e _ -> "OnSendShutdow " ++ show p ++ " " ++ show e+ OnSendInterrupt p e _ -> "OnSendInterrupt " ++ show p ++ " " ++ show e+ OnMakeReference _ -> "OnMakeReference"+ OnMonitor p _ -> "OnMonitor " ++ show p+ OnDemonitor p _ -> "OnDemonitor " ++ show p+ OnLink p _ -> "OnLink " ++ show p+ OnUnlink p _ -> "OnUnlink " ++ show p++runAsCoroutinePure+ :: forall v r m+ . Monad m+ => (forall a . Eff r a -> m a)+ -> Eff (ConsProcess r) v+ -> m (OnYield r v)+runAsCoroutinePure r = r . handle_relay (return . OnDone) cont+ where+ cont :: Process r x -> (x -> Eff r (OnYield r v)) -> Eff r (OnYield r v)+ cont FlushMessages k = return (OnFlushMessages k)+ cont YieldProcess k = return (OnYield k)+ cont SelfPid k = return (OnSelf k)+ cont (Spawn e ) k = return (OnSpawn False e k)+ cont (SpawnLink e ) k = return (OnSpawn True e k)+ cont (Shutdown !sr ) _k = return (OnShutdown sr)+ cont (SendMessage !tp !msg ) k = return (OnSend tp msg k)+ cont (ReceiveSelectedMessage f ) k = return (OnRecv f k)+ cont (GetProcessState !tp) k = return (OnGetProcessState tp k)+ cont (SendInterrupt !tp !er ) k = return (OnSendInterrupt tp er k)+ cont (SendShutdown !pid !sr ) k = return (OnSendShutdown pid sr k)+ cont MakeReference k = return (OnMakeReference k)+ cont (Monitor !pid) k = return (OnMonitor pid k)+ cont (Demonitor !ref) k = return (OnDemonitor ref k)+ cont (Link !pid) k = return (OnLink pid k)+ cont (Unlink !pid) k = return (OnUnlink pid k)+ -- | Internal 'Process' handler function. handleProcess :: Monad m- => (forall a . Eff r a -> m a)- -> m ()- -> ProcessId- -> Int- -> Map.Map ProcessId (Seq Dynamic)+ => STS r m -> Seq (OnYield r finalResult, ProcessId)- -> m (Either String finalResult)-handleProcess _runEff _yieldEff _newPid _nextRef _msgQs Empty =- return $ Left "no main process"+ -> m (Either (ExitReason 'NoRecovery) finalResult)+handleProcess _sts Empty =+ return $ Left (NotRecovered (ProcessError "no main process")) -handleProcess runEff yieldEff !newPid !nextRef !msgQs allProcs@((!processState, !pid) :<| rest)- = let handleExit res = if pid == 0- then return res- else handleProcess runEff- yieldEff- newPid- nextRef- (msgQs & at pid .~ Nothing)- rest+handleProcess sts allProcs@((!processState, !pid) :<| rest)+ = let+ handleExit res = if pid == 0+ then return res+ else do+ let (downPids, stsNew) = removeLinksTo pid sts+ linkedPids = filter (/= pid) downPids+ reason = LinkedProcessCrashed pid+ unlinkLoop [] ps = return ps+ unlinkLoop (dPid : dPidRest) ps = do+ ps' <- sendInterruptToOtherPid dPid reason ps+ unlinkLoop dPidRest ps'+ let allProcsWithoutPid = Seq.filter (\(_, p) -> p /= pid) rest+ nextTargets <- unlinkLoop linkedPids allProcsWithoutPid+ handleProcess+ (dropMsgQ+ pid+ (triggerAndRemoveMonitor+ pid+ (either SomeExitReason (const (SomeExitReason ExitNormally)) res+ )+ stsNew+ )+ )+ nextTargets in case processState of- OnDone r -> handleExit (Right r)+ OnDone r -> handleExit (Right r) - OnShutdown ExitNormally -> handleExit (Left "process exited normally")- OnShutdown (ExitWithError e) -> handleExit (Left e)+ OnShutdown e -> handleExit (Left e) - OnRaiseError errM -> handleExit (Left errM)+ OnInterrupt e k -> do+ nextK <- diskontinue sts k e+ handleProcess sts (rest :|> (nextK, pid)) - OnSendShutdown targetPid sr k -> do+ OnSendInterrupt targetPid sr k -> doSendInterrupt targetPid sr k++ OnSendShutdown targetPid sr 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)+ targets = Seq.filter (\(_, e) -> e == targetPid) allProcs+ suicide = targetPid == pid if suicide- then do- nextK <- runEff $ k (ShutdownRequested sr)- handleProcess runEff- yieldEff- newPid- nextRef- msgQs- (rest :|> (nextK, pid))+ then handleExit (Left sr) else do- let- deliverTheGoodNews (targetState, tPid) = do- nextTargetState <- case targetState of- OnSendShutdown _ _ tk -> tk (ShutdownRequested sr)- OnYield tk -> tk (ShutdownRequested sr)- OnSelf tk -> tk (ShutdownRequested sr)- OnSend _ _ tk -> tk (ShutdownRequested sr)- OnRecv _ tk -> tk (ShutdownRequested sr)- OnSpawn _ tk -> tk (ShutdownRequested sr)- OnDone x -> return (OnDone x)- OnShutdown _ -> return (OnShutdown sr)- OnRaiseError er -> return (OnShutdown (ExitWithError er))- OnMakeReference tk -> tk (ShutdownRequested sr)- return (nextTargetState, tPid)- nextTargets <- runEff $ traverse deliverTheGoodNews targets- nextK <- runEff $ k (ResumeWith targetFound)+ let deliverTheGoodNews (targetState, tPid) = do+ nextTargetState <- case targetState of+ OnSendInterrupt _ _ _tk -> return (OnShutdown sr)+ OnSendShutdown _ _ _tk -> return (OnShutdown sr)+ OnFlushMessages _tk -> return (OnShutdown sr)+ OnYield _tk -> return (OnShutdown sr)+ OnSelf _tk -> return (OnShutdown sr)+ OnSend _ _ _tk -> return (OnShutdown sr)+ OnRecv _ _tk -> return (OnShutdown sr)+ OnSpawn _ _ _tk -> return (OnShutdown sr)+ OnDone x -> return (OnDone x)+ OnGetProcessState _ _tk -> return (OnShutdown sr)+ OnShutdown sr' -> return (OnShutdown sr')+ OnInterrupt _er _tk -> return (OnShutdown sr)+ OnMakeReference _tk -> return (OnShutdown sr)+ OnMonitor _ _tk -> return (OnShutdown sr)+ OnDemonitor _ _tk -> return (OnShutdown sr)+ OnLink _ _tk -> return (OnShutdown sr)+ OnUnlink _ _tk -> return (OnShutdown sr)+ return (nextTargetState, tPid)+ nextTargets <- _runEff sts $ traverse deliverTheGoodNews targets+ nextK <- kontinue sts k () handleProcess- runEff- yieldEff- newPid- nextRef- msgQs+ sts (allButTarget Seq.>< (nextTargets :|> (nextK, pid))) - OnSelf k -> do- nextK <- runEff $ k (ResumeWith pid)- handleProcess runEff- yieldEff- newPid- nextRef- msgQs- (rest :|> (nextK, pid))+ nextK <- kontinue sts k pid+ handleProcess sts (rest :|> (nextK, pid)) OnMakeReference k -> do- nextK <- runEff $ k (ResumeWith nextRef)- handleProcess runEff- yieldEff- newPid- (nextRef + 1)- msgQs- (rest :|> (nextK, pid))+ let (ref, stsNext) = incRef sts+ nextK <- kontinue sts k ref+ handleProcess stsNext (rest :|> (nextK, pid)) OnYield k -> do- yieldEff- nextK <- runEff $ k (ResumeWith ())- handleProcess runEff- yieldEff- newPid- nextRef- msgQs- (rest :|> (nextK, pid))+ sts ^. yieldEff+ nextK <- kontinue sts k ()+ handleProcess sts (rest :|> (nextK, pid)) OnSend toPid msg k -> do- nextK <- runEff $ k (ResumeWith (msgQs ^. at toPid . to isJust))- handleProcess runEff- yieldEff- newPid- nextRef- (msgQs & at toPid . _Just %~ (:|> msg))- (rest :|> (nextK, pid))+ nextK <- kontinue sts k ()+ handleProcess (enqueueMsg toPid msg sts) (rest :|> (nextK, pid)) - OnSpawn f k -> do- nextK <- runEff $ k (ResumeWith newPid)- fk <- runAsCoroutinePure runEff (f >> exitNormally SP)- handleProcess runEff- yieldEff- (newPid + 1)- 0- (msgQs & at newPid ?~ Seq.empty)- (rest :|> (nextK, pid) :|> (fk, newPid))+ OnGetProcessState toPid k -> do+ nextK <- kontinue sts k (getProcessState toPid sts)+ handleProcess sts (rest :|> (nextK, pid)) - recv@(OnRecv messageSelector k) -> case msgQs ^. at pid of- Nothing -> do- nextK <- runEff $ k (OnError (show pid ++ " has no message queue!"))- handleProcess runEff- yieldEff- newPid- nextRef- 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- nextRef- msgQs- (rest :|> (nextK, pid))- else handleProcess runEff- yieldEff- newPid- nextRef- msgQs- (rest :|> (recv, pid))- Just messages ->- let partitionMessages Empty _acc = Nothing- partitionMessages (m :<| msgRest) acc = maybe- (partitionMessages msgRest (acc :|> m))- (\res -> Just (res, acc Seq.>< msgRest))- (runMessageSelector messageSelector m)- in case partitionMessages messages Empty of- Nothing -> handleProcess runEff- yieldEff- newPid- nextRef- msgQs- (rest :|> (recv, pid))- Just (result, otherMessages) -> do- nextK <- runEff $ k (ResumeWith result)- handleProcess runEff- yieldEff- newPid- nextRef- (msgQs & at pid . _Just .~ otherMessages)- (rest :|> (nextK, pid))+ OnSpawn link f k -> do+ let (newPid, newSts) =+ newProcessQ (if link then Just pid else Nothing) sts+ fk <- runAsCoroutinePure (newSts ^. runEff) (f >> exitNormally SP)+ nextK <- kontinue newSts k newPid+ handleProcess newSts (rest :|> (fk, newPid) :|> (nextK, pid)) -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+ OnFlushMessages k -> do+ let (msgs, newSts) = flushMsgs pid sts+ nextK <- kontinue newSts k msgs+ handleProcess newSts (rest :|> (nextK, pid))++ recv@(OnRecv messageSelector k) ->+ case receiveMsg pid messageSelector sts of+ Nothing -> do+ nextK <- diskontinue+ sts+ k+ (ProcessError (show pid ++ " has no message queue"))++ handleProcess sts (rest :|> (nextK, pid))+ Just Nothing -> if Seq.length rest == 0+ then do+ nextK <- diskontinue+ sts+ k+ (ProcessError (show pid ++ " deadlocked"))+ handleProcess sts (rest :|> (nextK, pid))+ else handleProcess sts (rest :|> (recv, pid))+ Just (Just (result, newSts)) -> do+ nextK <- kontinue newSts k result+ handleProcess newSts (rest :|> (nextK, pid))++ OnMonitor toPid k -> do+ let (ref, stsNew) = addMonitoring pid toPid sts+ nextK <- kontinue stsNew k ref+ handleProcess stsNew (rest :|> (nextK, pid))++ OnDemonitor monRef k -> do+ let stsNew = removeMonitoring monRef sts+ nextK <- kontinue stsNew k ()+ handleProcess stsNew (rest :|> (nextK, pid))++ OnLink toPid k -> do+ let (downInterrupts, stsNew) = addLink pid toPid sts+ nextK <- case downInterrupts of+ Nothing -> kontinue stsNew k ()+ Just i -> diskontinue stsNew k i+ handleProcess stsNew (rest :|> (nextK, pid))++ OnUnlink toPid k -> do+ let (_, stsNew) = removeLinksTo toPid sts+ nextK <- kontinue stsNew k ()+ handleProcess stsNew (rest :|> (nextK, pid)) 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 !sr ) _k = return (OnShutdown sr)- cont (RaiseError !e ) _k = return (OnRaiseError e)- cont (SendMessage !tp !msg ) k = return (OnSend tp msg k)- cont (ReceiveSelectedMessage f) k = return (OnRecv f k)- cont (SendShutdown !pid !sr ) k = return (OnSendShutdown pid sr k)- cont MakeReference k = return (OnMakeReference k)+ doSendInterrupt targetPid sr k = do+ let suicide = targetPid == pid+ if suicide+ then do+ nextK <- diskontinue sts k sr+ handleProcess sts (rest :|> (nextK, pid))+ else do+ nextTargets <- sendInterruptToOtherPid targetPid sr rest+ nextK <- kontinue sts k ()+ handleProcess sts (nextTargets :|> (nextK, pid)) + sendInterruptToOtherPid targetPid sr procs = do+ let allButTarget = Seq.filter (\(_, e) -> e /= targetPid) procs+ targets = Seq.filter (\(_, e) -> e == targetPid) procs+ deliverTheGoodNews (targetState, tPid) = do+ nextTargetState <- case targetState of+ OnSendInterrupt _ _ tk -> tk (Interrupted sr)+ OnSendShutdown _ _ tk -> tk (Interrupted sr)+ OnFlushMessages tk -> tk (Interrupted sr)+ OnYield tk -> tk (Interrupted sr)+ OnSelf tk -> tk (Interrupted sr)+ OnSend _ _ tk -> tk (Interrupted sr)+ OnRecv _ tk -> tk (Interrupted sr)+ OnSpawn _ _ tk -> tk (Interrupted sr)+ OnDone x -> return (OnDone x)+ OnGetProcessState _ tk -> tk (Interrupted sr)+ OnShutdown sr' -> return (OnShutdown sr')+ OnInterrupt er tk -> tk (Interrupted er)+ OnMakeReference tk -> tk (Interrupted sr)+ OnMonitor _ tk -> tk (Interrupted sr)+ OnDemonitor _ tk -> tk (Interrupted sr)+ OnLink _ tk -> tk (Interrupted sr)+ OnUnlink _ tk -> tk (Interrupted sr)+ return (nextTargetState, tPid)+ nextTargets <- _runEff sts $ traverse deliverTheGoodNews targets+ return (nextTargets Seq.>< allButTarget)+ -- | The concrete list of 'Eff'ects for running this pure scheduler on @IO@ and -- with string logging. type LoggingAndIo =@@ -328,15 +556,15 @@ -- | Execute a 'Process' using 'schedule' on top of 'Lift' @IO@ and 'Logs' -- @String@ effects.-defaultMain+defaultMainSingleThreaded :: HasCallStack => Eff- ( ConsProcess+ ( InterruptableProcess '[Logs LogMessage, LogWriterReader LogMessage IO, Lift IO] ) () -> IO ()-defaultMain =+defaultMainSingleThreaded = void . runLift . writeLogs (multiMessageLogWriter ($ printLogMessage))
+ src/Control/Eff/Concurrent/Process/Timer.hs view
@@ -0,0 +1,184 @@+-- | Functions for timeouts when receiving messages.+--+-- NOTE: If you use a single threaded scheduler, these functions will not work+-- as expected. (This is an open TODO)+--+-- @since 0.12.0+module Control.Eff.Concurrent.Process.Timer+ ( Timeout(fromTimeoutMicros)+ , TimerReference()+ , TimerElapsed(fromTimerElapsed)+ , sendAfter+ , startTimer+ , cancelTimer+ , selectTimerElapsed+ , receiveAfter+ , receiveSelectedAfter+ )+ -- , receiveSelectedAfter, receiveAnyAfter, sendMessageAfter)+where++import Control.Eff.Concurrent.Process+import Control.Concurrent+import Control.Eff.Lift+import Control.Eff+import Control.DeepSeq+import Control.Monad.IO.Class+import Data.Typeable+import Control.Applicative+import GHC.Stack+++-- | Wait for a message of the given type for the given time. When no message+-- arrives in time, return 'Nothing'. This is based on+-- 'receiveSelectedAfter'.+--+-- @since 0.12.0+receiveAfter+ :: forall a r q+ . ( Lifted IO q+ , HasCallStack+ , SetMember Process (Process q) r+ , Member Interrupts r+ , Typeable a+ , NFData a+ , Show a+ )+ => SchedulerProxy q+ -> Timeout+ -> Eff r (Maybe a)+receiveAfter px t =+ either (const Nothing) Just <$> receiveSelectedAfter px (selectMessage @a) t++-- | Wait for a message of the given type for the given time. When no message+-- arrives in time, return 'Left' 'TimerElapsed'. This is based on+-- 'selectTimerElapsed' and 'startTimer'.+--+-- @since 0.12.0+receiveSelectedAfter+ :: forall a r q+ . ( Lifted IO q+ , HasCallStack+ , SetMember Process (Process q) r+ , Member Interrupts r+ , Show a+ )+ => SchedulerProxy q+ -> MessageSelector a+ -> Timeout+ -> Eff r (Either TimerElapsed a)+receiveSelectedAfter px sel t = do+ timerRef <- startTimer px t+ res <- receiveSelectedMessage+ px+ (Left <$> selectTimerElapsed timerRef <|> Right <$> sel)+ cancelTimer px timerRef+ return res++-- | A 'MessageSelector' matching 'TimerElapsed' messages created by+-- 'startTimer'.+--+-- @since 0.12.0+selectTimerElapsed :: TimerReference -> MessageSelector TimerElapsed+selectTimerElapsed timerRef =+ filterMessage (\(TimerElapsed timerRefIn) -> timerRef == timerRefIn)+++-- | A number of micro seconds.+--+-- @since 0.12.0+newtype Timeout = TimeoutMicros {fromTimeoutMicros :: Int}+ deriving (NFData, Ord,Eq, Num, Integral, Real, Enum, Typeable)++instance Show Timeout where+ showsPrec d (TimeoutMicros t) =+ showParen (d>=10) (showString "timeout: " . shows t . showString " µs")++-- | The reference to a timer started by 'startTimer', required to stop+-- a timer via 'cancelTimer'.+--+-- @since 0.12.0+newtype TimerReference = TimerReference ProcessId+ deriving (NFData, Ord,Eq, Num, Integral, Real, Enum, Typeable)++instance Show TimerReference where+ showsPrec d (TimerReference t) =+ showParen (d>=10) (showString "timer: " . shows t)++-- | A value to be sent when timer started with 'startTimer' has elapsed.+--+-- @since 0.12.0+newtype TimerElapsed = TimerElapsed {fromTimerElapsed :: TimerReference}+ deriving (NFData, Ord,Eq, Typeable)++instance Show TimerElapsed where+ showsPrec d (TimerElapsed t) =+ showParen (d>=10) (shows t . showString " elapsed")--+-- @since 0.12.0+++-- | Send a message to a given process after waiting. The message is created by+-- applying the function parameter to the 'TimerReference', such that the+-- message can directly refer to the timer.+--+-- @since 0.12.0+sendAfter+ :: forall r q message+ . ( Lifted IO q+ , HasCallStack+ , SetMember Process (Process q) r+ , Member Interrupts r+ , Typeable message+ , NFData message+ )+ => SchedulerProxy q+ -> ProcessId+ -> Timeout+ -> (TimerReference -> message)+ -> Eff r TimerReference+sendAfter px pid (TimeoutMicros 0) mkMsg = TimerReference <$> spawn+ ( yieldProcess px+ >> self px+ >>= (sendMessage SP pid . force . mkMsg . TimerReference)+ )+sendAfter px pid (TimeoutMicros t) mkMsg = TimerReference <$> spawn+ ( liftIO (threadDelay t)+ >> self px+ >>= (sendMessage SP pid . force . mkMsg . TimerReference)+ )++-- | Start a new timer, after the time has elapsed, 'TimerElapsed' is sent to+-- calling process. The message also contains the 'TimerReference' returned by+-- this function. Use 'cancelTimer' to cancel the timer. Use+-- 'selectTimerElapsed' to receive the message using 'receiveSelectedMessage'.+-- To receive messages with guarded with a timeout see 'receiveAfter'.+--+-- @since 0.12.0+startTimer+ :: forall r q+ . ( Lifted IO q+ , HasCallStack+ , SetMember Process (Process q) r+ , Member Interrupts r+ )+ => SchedulerProxy q+ -> Timeout+ -> Eff r TimerReference+startTimer px t = do+ p <- self px+ sendAfter px p t TimerElapsed++-- | Cancel a timer started with 'startTimer'.+--+-- @since 0.12.0+cancelTimer+ :: forall r q+ . ( Lifted IO q+ , HasCallStack+ , SetMember Process (Process q) r+ , Member Interrupts r+ )+ => SchedulerProxy q+ -> TimerReference+ -> Eff r ()+cancelTimer px (TimerReference tr) = sendShutdown px tr ExitNormally
src/Control/Eff/ExceptionExtra.hs view
@@ -3,30 +3,241 @@ {-# LANGUAGE MonoLocalBinds #-} {-# LANGUAGE DataKinds #-} {-# LANGUAGE TypeOperators #-}+{-# OPTIONS_GHC -Wno-orphans #-} -- | Add-ons to 'Control.Eff.Exception' and 'Control.Excepion' module Control.Eff.ExceptionExtra ( liftTry , maybeThrow- , module X+ , module Eff ) where -import qualified Control.Exception as Exc+import qualified Control.Exception.Safe as Safe+import qualified Control.Monad.Catch as Catch import Control.Eff+import Control.Eff.Extend import Control.Eff.Lift-import Control.Eff.Exception as X+import Control.Eff.Exception as Eff import GHC.Stack+import Control.Eff.Reader.Lazy as Lazy+import Control.Eff.Reader.Strict as Strict --- | Catch 'Exc.Exception' thrown by an effect.+-- | Catch 'Safe.Exception' thrown by an effect. liftTry :: forall e r a- . (HasCallStack, Exc.Exception e, SetMember Lift (Lift IO) r)+ . (HasCallStack, Safe.Exception e, Lifted IO r) => Eff r a -> Eff r (Either e a) liftTry m = (Right <$> m) `catchDynE` (return . Left) --- | Very similar to 'X.liftEither' but for 'Maybe's. Unlike 'X.liftMaybe' this--- will throw the given value (instead of using 'X.Fail').-maybeThrow :: Member (X.Exc x) e => x -> Maybe a -> Eff e a-maybeThrow x = X.liftEither . maybe (Left x) Right+-- | Very similar to 'Eff.liftEither' but for 'Maybe's. Unlike 'Eff.liftMaybe' this+-- will throw the given value (instead of using 'Eff.Fail').+maybeThrow :: Member (Eff.Exc x) e => x -> Maybe a -> Eff e a+maybeThrow x = Eff.liftEither . maybe (Left x) Right++-- * Orphan 'MonadThrow', 'MonadCatch' and 'MonadMask' instances++-- ** Lazy Reader++instance Catch.MonadThrow (Eff e) => Catch.MonadThrow (Eff (Lazy.Reader x ': e)) where+ throwM exception = raise (Catch.throwM exception)++instance Catch.MonadCatch (Eff e) => Catch.MonadCatch (Eff (Lazy.Reader x ': e)) where+ catch effect handler = do+ readerValue <- Lazy.ask @x+ let nestedEffects =+ Lazy.runReader readerValue effect+ nestedHandler exception =+ Lazy.runReader readerValue (handler exception)+ raise (Catch.catch nestedEffects nestedHandler)++instance Catch.MonadMask (Eff e) => Catch.MonadMask (Eff (Lazy.Reader x ': e)) where+ mask maskedEffect = do+ readerValue <- Lazy.ask @x+ raise+ (Catch.mask+ (\nestedUnmask -> Lazy.runReader+ readerValue+ (maskedEffect+ (\unmasked ->+ raise (nestedUnmask (Lazy.runReader readerValue unmasked))+ )+ )+ )+ )+ uninterruptibleMask maskedEffect = do+ readerValue <- Lazy.ask @x+ raise+ (Catch.uninterruptibleMask+ (\nestedUnmask -> Lazy.runReader+ readerValue+ (maskedEffect+ (\unmasked ->+ raise (nestedUnmask (Lazy.runReader readerValue unmasked))+ )+ )+ )+ )+ generalBracket acquire release use = do+ readerValue <- Lazy.ask @x+ let+ lower :: Eff (Lazy.Reader x ': e) a -> Eff e a+ lower = Lazy.runReader readerValue+ raise+ (Catch.generalBracket+ (lower acquire)+ (((.).(.)) lower release)+ (lower . use))++-- ** Strict Reader++instance Catch.MonadThrow (Eff e) => Catch.MonadThrow (Eff (Strict.Reader x ': e)) where+ throwM exception = raise (Catch.throwM exception)++instance Catch.MonadCatch (Eff e) => Catch.MonadCatch (Eff (Strict.Reader x ': e)) where+ catch effect handler = do+ readerValue <- Strict.ask @x+ let nestedEffects =+ Strict.runReader readerValue effect+ nestedHandler exception =+ Strict.runReader readerValue (handler exception)+ raise (Catch.catch nestedEffects nestedHandler)++instance Catch.MonadMask (Eff e) => Catch.MonadMask (Eff (Strict.Reader x ': e)) where+ mask maskedEffect = do+ readerValue <- Strict.ask @x+ raise+ (Catch.mask+ (\nestedUnmask -> Strict.runReader+ readerValue+ (maskedEffect+ (\unmasked ->+ raise (nestedUnmask (Strict.runReader readerValue unmasked))+ )+ )+ )+ )+ uninterruptibleMask maskedEffect = do+ readerValue <- Strict.ask @x+ raise+ (Catch.uninterruptibleMask+ (\nestedUnmask -> Strict.runReader+ readerValue+ (maskedEffect+ (\unmasked ->+ raise (nestedUnmask (Strict.runReader readerValue unmasked))+ )+ )+ )+ )+ generalBracket acquire release use = do+ readerValue <- Strict.ask @x+ let+ lower :: Eff (Strict.Reader x ': e) a -> Eff e a+ lower = Strict.runReader readerValue+ raise+ (Catch.generalBracket+ (lower acquire)+ (((.).(.)) lower release)+ (lower . use))++-- ** Lifted IO++instance Catch.MonadThrow m => Catch.MonadThrow (Eff '[Lift m]) where+ throwM exception = lift (Catch.throwM exception)++instance Catch.MonadCatch m => Catch.MonadCatch (Eff '[Lift m]) where+ catch effect handler = do+ let nestedEffects = runLift effect+ nestedHandler exception = runLift (handler exception)+ lift (Catch.catch nestedEffects nestedHandler)++instance Catch.MonadMask m => Catch.MonadMask (Eff '[Lift m]) where+ mask maskedEffect =+ lift+ (Catch.mask+ (\nestedUnmask -> runLift+ (maskedEffect+ (\unmasked -> lift (nestedUnmask (runLift unmasked))+ )+ )+ )+ )+ uninterruptibleMask maskedEffect =+ lift+ (Catch.uninterruptibleMask+ (\nestedUnmask -> runLift+ (maskedEffect+ (\unmasked ->+ lift (nestedUnmask (runLift unmasked))+ )+ )+ )+ )+ generalBracket acquire release use =+ lift+ (Catch.generalBracket+ (runLift acquire)+ (((.).(.)) runLift release)+ (runLift . use))+++instance Catch.MonadThrow (Eff e) => Catch.MonadThrow (Eff (Exc x ': e)) where+ throwM exception = raise (Catch.throwM exception)++instance Catch.MonadCatch (Eff e) => Catch.MonadCatch (Eff (Exc x ': e)) where+ catch effect handler = do+ let nestedEffects =+ runError effect+ nestedHandler exception =+ runError (handler exception)+ errorOrResult <- raise (Catch.catch nestedEffects nestedHandler)+ liftEither errorOrResult++instance Catch.MonadMask (Eff e) => Catch.MonadMask (Eff (Exc x ': e)) where+ mask maskedEffect = do+ errorOrResult <- raise+ (Catch.mask+ (\nestedUnmask -> runError+ (maskedEffect+ (\unmasked -> do+ errorOrResult <- raise (nestedUnmask (runError unmasked))+ liftEither errorOrResult+ )+ )+ )+ )+ liftEither errorOrResult+ uninterruptibleMask maskedEffect = do+ errorOrResult <- raise+ (Catch.uninterruptibleMask+ (\nestedUnmask -> runError+ (maskedEffect+ (\unmasked -> do+ errorOrResult <- raise (nestedUnmask (runError unmasked))+ liftEither errorOrResult+ )+ )+ )+ )+ liftEither errorOrResult+ generalBracket acquire release use = do+ (useResultOrError, releaseResultOrError) <- raise+ (Catch.generalBracket+ (runError acquire)+ (\resourceRight exitCase ->+ case resourceRight of+ Left e -> return (Left e) -- acquire failed+ Right resource ->+ case exitCase of+ Catch.ExitCaseSuccess (Right b) -> runError (release resource (Catch.ExitCaseSuccess b))+ Catch.ExitCaseException e -> runError (release resource (Catch.ExitCaseException e))+ _ -> runError (release resource Catch.ExitCaseAbort)+ )+ (\case+ Left e -> return (Left e)+ Right resource -> runError (use resource)))+ c <- liftEither releaseResultOrError -- if both results are bad, fire the release error+ b <- liftEither useResultOrError+ return (b, c)
src/Control/Eff/Log/Handler.hs view
@@ -27,9 +27,11 @@ import Control.Eff.Extend import Control.Eff.Reader.Strict import Control.Eff.Lift as Eff+import qualified Control.Exception.Safe as Safe import Data.Foldable ( traverse_ ) import Data.Default import Control.Monad+import qualified Control.Monad.Catch as Catch import Control.Monad.Trans.Control ( MonadBaseControl ( restoreM , liftBaseWith@@ -49,29 +51,70 @@ -- multiple times. -- The messages are reduced to normal form (strict). logMsgs- :: (Traversable f, MonadPlus f, NFData1 f, NFData m, Member (Logs m) e)+ :: ( Traversable f+ , MonadPlus f+ , NFData1 f+ , NFData (f m)+ , NFData m+ , Member (Logs m) e+ ) => f m -> Eff e ()-logMsgs !msgs = rnf1 msgs `seq` send (LogMsgs msgs)+logMsgs !msgs = rnf1 msgs `seq` do+ f <- send AskLogFilter+ send+ (LogMsgs+ (do+ m <- msgs+ maybe mzero (return . force) (f m)+ )+ ) -- ** Filter and Transform Log Messages -- | Map a pure function over log messages. mapLogMessages- :: forall m r b . (Member (Logs m) r) => (m -> m) -> Eff r b -> Eff r b-mapLogMessages f = interpose return go+ :: forall m r b+ . (NFData m, Member (Logs m) r)+ => (m -> m)+ -> Eff r b+ -> Eff r b+mapLogMessages f eff = do+ old <- send AskLogFilter+ interpose return (go (fmap f . old)) eff where- go :: Logs m a -> Arr r a b -> Eff r b- go (LogMsgs ms) k = logMsgs (fmap f ms) >>= k+ go :: (m -> Maybe m) -> Logs m a -> Arr r a b -> Eff r b+ go t AskLogFilter k = k t+ go _ (LogMsgs ms) k = logMsgs ms >>= k --- | Keep only those messages, for which the predicate holds.+-- | Keep only those messages, for which a predicate holds.+--+-- E.g. to keep only messages which begin with @"OMG"@:+--+-- > filterLogMessages (\msg -> case msg of+-- > 'O':'M':'G':_ -> True+-- > _ -> False)+-- > (do logMsg "this message will not be logged"+-- > logMsg "OMG logged") filterLogMessages- :: forall m r b . (Member (Logs m) r) => (m -> Bool) -> Eff r b -> Eff r b-filterLogMessages predicate = interpose return go+ :: forall m r b+ . (NFData m, Member (Logs m) r)+ => (m -> Bool)+ -> Eff r b+ -> Eff r b+filterLogMessages predicate eff = do+ old <- send AskLogFilter+ interpose return (go (\m -> if predicate m then old m else Nothing)) eff where- go :: Logs m a -> Arr r a b -> Eff r b- go (LogMsgs ms) k = logMsgs (mfilter predicate ms) >>= k+ go :: (m -> Maybe m) -> Logs m a -> Arr r a b -> Eff r b+ go t AskLogFilter k = k t+ go _ (LogMsgs ms) k = logMsgs ms >>= k +-- interpose return go+-- where+-- go :: Logs m a -> Arr r a b -> Eff r b+ -- go (LogMsgs ms) k = logMsgs (mfilter predicate ms) >>= k+ -- ** Filter and Transform Log Messages effectfully -- | Map an 'Eff'ectful function over every bunch of log messages.@@ -84,22 +127,24 @@ -- > => Eff e a -- > -> Eff e a -- > appendTimestamp = traverseLogMessages $ \ms -> do--- > now <- lift getCurrentTime+-- > now <- getCurrentTime -- > return (fmap (show now ++) ms) traverseLogMessages- :: forall m r b- . (Member (Logs m) r)- => ( forall f- . (MonadPlus f, Traversable f, NFData1 f)- => f m- -> Eff r (f m)+ :: forall m r h b+ . ( Member (Logs m) r+ , Monad h+ , Lifted h r+ , Member (Reader (LogWriter m h)) r )+ => (forall f . (MonadPlus f, Traversable f, NFData1 f) => f m -> h (f m)) -> Eff r b -> Eff r b-traverseLogMessages f = interpose return go- where- go :: Logs m a -> Arr r a b -> Eff r b- go (LogMsgs ms) k = f ms >>= logMsgs >>= k+traverseLogMessages f = changeLogWriter+ (\msgs -> do+ lw <- ask+ msgs' <- lift (f msgs)+ lift (runLogWriter lw msgs')+ ) -- ** Change the Log Writer @@ -130,7 +175,8 @@ ignoreLogs = handle_relay return go where go :: Logs message v -> Arr r v a -> Eff r a- go (LogMsgs _) k = k ()+ go (LogMsgs _) k = k ()+ go AskLogFilter k = k (const Nothing) -- | Trace all log messages using 'traceM'. The message value is -- converted to 'String' using the given function.@@ -143,6 +189,7 @@ where go :: Logs message v -> Arr r v a -> Eff r a go (LogMsgs ms) k = traverse_ (traceM . toString) ms >> k ()+ go AskLogFilter k = k pure -- ** Log Message Writer Creation @@ -233,26 +280,38 @@ -- Log messages are consumed by 'LogWriter's installed via -- 'runLogs' or more high level functions like 'writeLogs'. data Logs m v where+ AskLogFilter+ :: (NFData m) => Logs m (m -> Maybe m) LogMsgs- :: (Traversable f, MonadPlus f, NFData1 f, NFData m)- => f m- -> Logs m ()+ :: (Traversable f, MonadPlus f, NFData1 f, NFData m, NFData (f m))+ => f m -> Logs m () -- | Install 'Logs' handler that 'ask's for a 'LogWriter' for the -- message type and applies the log writer to the messages.-runLogs+runLogsFiltered :: forall m h e b- . (Applicative h, Lifted h e, Member (LogWriterReader m h) e)- => Eff (Logs m ': e) b+ . (NFData m, Applicative h, Lifted h e, Member (LogWriterReader m h) e)+ => (m -> Maybe m)+ -> Eff (Logs m ': e) b -> Eff e b-runLogs = handle_relay return go+runLogsFiltered f = handle_relay return (go f) where- go :: Logs m a -> Arr e a b -> Eff e b- go (LogMsgs ms) k = do+ go :: (m -> Maybe m) -> Logs m a -> Arr e a c -> Eff e c+ go lt AskLogFilter k = k lt+ go _lt (LogMsgs ms) k = do logWrtr <- ask- lift (writeAllLogMessages logWrtr ms)+ lift (writeAllLogMessages logWrtr (force ms)) k () +-- | Install 'Logs' handler that 'ask's for a 'LogWriter' for the+-- message type and applies the log writer to the messages.+runLogs+ :: forall m h e b+ . (Applicative h, Lifted h e, Member (LogWriterReader m h) e, NFData m)+ => Eff (Logs m ': e) b+ -> Eff e b+runLogs = runLogsFiltered pure+ -- | Handle log message effects by a monadic action, e.g. an IO action -- to send logs to the console output or a log-server. -- The monadic log writer action is wrapped in a newtype called@@ -263,17 +322,35 @@ -- 'mulitMessageLogWriter'. writeLogs :: forall message writerM r a- . (Applicative writerM, Lifted writerM r)+ . (Applicative writerM, Lifted writerM r, NFData message) => LogWriter message writerM -> Eff (Logs message ': Reader (LogWriter message writerM) ': r) a -> Eff r a writeLogs w = runReader w . runLogs +-- | Handle log message effects by a monadic action, e.g. an IO action+-- to send logs to the console output or a log-server.+-- The monadic log writer action is wrapped in a newtype called+-- 'LogWriter'.+--+-- Use the smart constructors below to create them, e.g.+-- 'foldingLogWriter', 'singleMessageLogWriter' or+-- 'mulitMessageLogWriter'.+writeLogsFiltered+ :: forall message writerM r a+ . (Applicative writerM, Lifted writerM r, NFData message)+ => (message -> Maybe message)+ -> LogWriter message writerM+ -> Eff (Logs message ': Reader (LogWriter message writerM) ': r) a+ -> Eff r a+writeLogsFiltered f w = runReader w . runLogsFiltered f+ -- | This instance allows liftings of the 'Logs' effect, but only, if there is -- a 'LogWriter' in effect. instance ( MonadBase m m , Lifted m r+ , NFData l , MonadBaseControl m (Eff r) ) => MonadBaseControl m (Eff (Logs l ': LogWriterReader l m ': r)) where@@ -283,6 +360,71 @@ liftBaseWith f = do l <- askLogWriter- raise (raise (liftBaseWith (\runInBase -> f (runInBase . writeLogs l))))+ lf <- send AskLogFilter+ raise (raise (liftBaseWith (\runInBase -> f (runInBase . writeLogsFiltered lf l)))) restoreM = raise . raise . restoreM+++instance (NFData l, Lifted m e, Catch.MonadThrow (Eff e))+ => Catch.MonadThrow (Eff (Logs l ': LogWriterReader l m ': e)) where+ throwM exception = raise (raise (Catch.throwM exception))++instance (NFData l, Applicative m, Lifted m e, Catch.MonadCatch (Eff e))+ => Catch.MonadCatch (Eff (Logs l ': LogWriterReader l m ': e)) where+ catch effect handler = do+ logWriter <- ask @(LogWriter l m)+ logFilter <- send AskLogFilter+ let lower = writeLogsFiltered logFilter logWriter+ nestedEffects = lower effect+ nestedHandler exception = lower (handler exception)+ raise (raise (Catch.catch nestedEffects nestedHandler))++instance (NFData l, Applicative m, Lifted m e, Catch.MonadMask (Eff e))+ => Catch.MonadMask (Eff (Logs l ': LogWriterReader l m ': e)) where+ mask maskedEffect = do+ logWriter <- ask @(LogWriter l m)+ logFilter <- send AskLogFilter+ let+ lower :: Eff (Logs l ': LogWriterReader l m ': e) a -> Eff e a+ lower = writeLogsFiltered logFilter logWriter+ raise+ (raise+ (Catch.mask+ (\nestedUnmask -> lower+ (maskedEffect+ ( raise . raise . nestedUnmask . lower )+ )+ )+ )+ )+ uninterruptibleMask maskedEffect = do+ logWriter <- ask @(LogWriter l m)+ logFilter <- send AskLogFilter+ let+ lower :: Eff (Logs l ': LogWriterReader l m ': e) a -> Eff e a+ lower = writeLogsFiltered logFilter logWriter+ raise+ (raise+ (Catch.uninterruptibleMask+ (\nestedUnmask -> lower+ (maskedEffect+ ( raise . raise . nestedUnmask . lower )+ )+ )+ )+ )+ generalBracket acquire release use = do+ logWriter <- ask @(LogWriter l m)+ logFilter <- send AskLogFilter+ let+ lower :: Eff (Logs l ': LogWriterReader l m ': e) a -> Eff e a+ lower = writeLogsFiltered logFilter logWriter+ raise+ (raise+ (Catch.generalBracket+ (lower acquire)+ (((.).(.)) lower release)+ (lower . use)+ )+ )
stack.yaml view
@@ -40,14 +40,11 @@ # Dependency packages to be pulled from upstream that are not in the resolver # (e.g., acme-missiles-0.3) extra-deps:- - extensible-effects-3.1.0.1-+ - extensible-effects-3.1.0.2 # 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 #
test/Common.hs view
@@ -1,11 +1,13 @@ module Common where +import Control.Concurrent 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.Lens import Control.Monad ( void , when )@@ -13,11 +15,10 @@ import Test.Tasty.HUnit import Test.Tasty.Runners import GHC.Stack-import Control.Lens setTravisTestOptions :: TestTree -> TestTree setTravisTestOptions =- localOption (timeoutSeconds 300) . localOption (NumThreads 1)+ localOption (timeoutSeconds 60) . localOption (NumThreads 1) timeoutSeconds :: Integer -> Timeout timeoutSeconds seconds = Timeout (seconds * 1000000) (show seconds ++ "s")@@ -33,27 +34,26 @@ (multiMessageLogWriter (\writeWith -> writeWith- (\m -> when (view lmSeverity m < noticeSeverity) (printLogMessage m)- )+ (\m -> when (view lmSeverity m < errorSeverity) $ printLogMessage m) ) ) (doSchedule e) ) ) -untilShutdown :: Member t r => t (ResumeProcess v) -> Eff r ()-untilShutdown pa = do+untilInterrupted :: Member t r => t (ResumeProcess v) -> Eff r ()+untilInterrupted pa = do r <- send pa case r of- ShutdownRequested _ -> return ()- _ -> untilShutdown pa+ Interrupted _ -> return ()+ _ -> untilInterrupted pa scheduleAndAssert :: forall r . (SetMember Lift (Lift IO) r, Member (Logs LogMessage) r)- => IO (Eff (Process r ': r) () -> IO ())- -> ( (String -> Bool -> Eff (Process r ': r) ())- -> Eff (Process r ': r) ()+ => IO (Eff (InterruptableProcess r) () -> IO ())+ -> ( (String -> Bool -> Eff (InterruptableProcess r) ())+ -> Eff (InterruptableProcess r) () ) -> IO () scheduleAndAssert schedulerFactory testCaseAction = withFrozenCallStack $ do@@ -71,9 +71,9 @@ applySchedulerFactory :: forall r . (Member (Logs LogMessage) r, SetMember Lift (Lift IO) r)- => IO (Eff (Process r ': r) () -> IO ())- -> Eff (Process r ': r) ()+ => IO (Eff (InterruptableProcess r) () -> IO ())+ -> Eff (InterruptableProcess r) () -> IO () applySchedulerFactory factory procAction = do scheduler <- factory- scheduler procAction+ scheduler (procAction >> lift (threadDelay 20000))
test/ForkIOScheduler.hs view
@@ -2,15 +2,17 @@ import Control.Exception import Control.Concurrent+import Control.Concurrent.Async import Control.Concurrent.STM import Control.Eff.Extend import Control.Eff.Lift import Control.Eff.Log import Control.Eff.Loop import Control.Eff.Concurrent.Process+import Control.Eff.Concurrent.Process.Timer import Control.Eff.Concurrent.Process.ForkIOScheduler as Scheduler-import Control.Monad ( void )+import Control.Monad ( void, replicateM_ ) import Test.Tasty import Test.Tasty.HUnit import Data.Dynamic@@ -30,7 +32,7 @@ aVar <- newEmptyTMVarIO withAsyncLogChannel 1000- def+ def -- (singleMessageLogWriter (putStrLn . renderLogMessage)) (Scheduler.defaultMainWithLogChannel (do p1 <- spawn $ foreverCheap busyEffect@@ -39,10 +41,23 @@ lift (threadDelay 1000) doExit lift (threadDelay 100000)- wasStillRunningP1 <- sendShutdownChecked forkIoScheduler- p1- ExitNormally- lift (atomically (putTMVar aVar wasStillRunningP1))++ me <- self forkIoScheduler+ spawn_+ (lift (threadDelay 10000) >> sendMessage forkIoScheduler me ()+ )+ eres <- receiveWithMonitor forkIoScheduler+ p1+ (selectMessage @())+ case eres of+ Left _down -> lift (atomically (putTMVar aVar False))+ Right () -> withMonitor forkIoScheduler p1 $ \ref -> do+ sendShutdown forkIoScheduler+ p1+ (NotRecovered (ProcessError "test 123"))+ _down <- receiveSelectedMessage forkIoScheduler+ (selectProcessDown ref)+ lift (atomically (putTMVar aVar True)) ) ) @@ -64,8 +79,7 @@ (send (Spawn @SchedulerIO (void- (send (ReceiveSelectedMessage @SchedulerIO selectAnyMessageLazy)- )+ (send (ReceiveSelectedMessage @SchedulerIO selectAnyMessageLazy)) ) ) )@@ -73,6 +87,7 @@ ] , (howToExit, doExit ) <- [ ("throw async exception", void (lift (throw UserInterrupt)))+ , ("cancel process" , void (lift (throw AsyncCancelled))) , ("division by zero" , void ((lift . print) ((123 :: Int) `div` 0))) , ("call 'fail'" , void (fail "test fail")) , ("call 'error'" , void (error "test error"))@@ -87,7 +102,7 @@ 1000 def (Scheduler.defaultMainWithLogChannel- (void (spawn (void (receiveMessage forkIoScheduler))))+ (void (spawn (void (receiveAnyMessage forkIoScheduler)))) ) ) )@@ -101,7 +116,7 @@ def (Scheduler.defaultMainWithLogChannel (do- void (spawn (void (receiveMessage forkIoScheduler)))+ void (spawn (void (receiveAnyMessage forkIoScheduler))) void (exitNormally forkIoScheduler) fail "This should not happen!!" )@@ -122,9 +137,7 @@ (do void (spawn- (foreverCheap- (void (sendMessage forkIoScheduler 1000 (toDyn "test")))- )+ (foreverCheap (void (sendMessage forkIoScheduler 1000 "test"))) ) void (exitNormally forkIoScheduler) fail "This should not happen!!"@@ -143,8 +156,8 @@ def (Scheduler.defaultMainWithLogChannel (do- child <- spawn (void (receiveMessageAs @String forkIoScheduler))- True <- sendMessageChecked forkIoScheduler child (toDyn "test")+ child <- spawn (void (receiveMessage @String forkIoScheduler))+ sendMessage forkIoScheduler child "test" return () ) )@@ -160,16 +173,49 @@ def (Scheduler.defaultMainWithLogChannel (do- child <- spawn+ child <- spawn $ void $ provideInterrupts $ exitOnInterrupt+ forkIoScheduler (do- void (receiveMessageAs @String forkIoScheduler)+ void (receiveMessage @String forkIoScheduler) void (exitNormally forkIoScheduler) error "This should not happen (child)!!" )- True <- sendMessageChecked forkIoScheduler child (toDyn "test")+ sendMessage forkIoScheduler child "test" void (exitNormally forkIoScheduler) error "This should not happen!!" ) ) ) )++test_timer :: TestTree+test_timer =+ setTravisTestOptions+ $ testCase "flush via timer"+ $ withAsyncLogChannel 1000 def+ $ Scheduler.defaultMainWithLogChannel+ $ do+ let n = 100+ testMsg :: Float+ testMsg = 123+ flushMsgs px = do+ res <- receiveSelectedAfter px (selectDynamicMessageLazy Just) 0+ case res of+ Left _to -> return ()+ Right _ -> flushMsgs px+ me <- self SP+ spawn_+ (do+ replicateM_ n $ sendMessage SP me "bad message"+ replicateM_ n $ sendMessage SP me (3123 :: Integer)+ sendMessage SP me testMsg+ )+ do+ res <- receiveAfter @Float SP 1000000+ lift (res @?= Just testMsg)+ flushMsgs SP+ res <- receiveSelectedAfter SP (selectDynamicMessageLazy Just) 10000+ case res of+ Left _ -> return ()+ Right x -> lift (False @? "unexpected message in queue " ++ show x)+ lift (threadDelay 100)
test/Interactive.hs view
@@ -1,5 +1,4 @@-module Interactive-where+module Interactive where import Control.Concurrent import Control.Eff@@ -17,17 +16,20 @@ test_interactive :: TestTree test_interactive = setTravisTestOptions $ testGroup "Interactive"- [ testGroup "SingleThreadedScheduler" $ allTests SingleThreaded.defaultMain+ [ testGroup "SingleThreadedScheduler" $ allTests SingleThreaded.defaultMainSingleThreaded , testGroup "ForkIOScheduler" $ allTests ForkIOScheduler.defaultMain ] -allTests :: SetMember Lift (Lift IO) r => (Eff (ConsProcess r) () -> IO ()) -> [TestTree]-allTests scheduler =- [ happyCaseTest scheduler- ]+allTests+ :: SetMember Lift (Lift IO) r+ => (Eff (InterruptableProcess r) () -> IO ())+ -> [TestTree]+allTests scheduler = [happyCaseTest scheduler] happyCaseTest- :: SetMember Lift (Lift IO) r => (Eff (ConsProcess r) () -> IO ()) -> TestTree+ :: SetMember Lift (Lift IO) r+ => (Eff (InterruptableProcess r) () -> IO ())+ -> TestTree happyCaseTest scheduler = testCase "start, wait and stop interactive scheduler" $ do s <- forkInteractiveScheduler scheduler
test/LoopTests.hs view
@@ -56,11 +56,11 @@ $ do me <- self SP spawn_- (foreverCheap $ sendMessageAs SP me ()+ (foreverCheap $ sendMessage SP me () ) replicateCheapM_ soMany- (void (receiveMessageAs @() SP))+ (void (receiveMessage @() SP)) res @=? Right () ]@@ -105,10 +105,10 @@ ) $ do me <- self SP- spawn_ (forever $ sendMessageAs SP me ())+ spawn_ (forever $ sendMessage SP me ()) replicateM_ soMany- (void (receiveMessageAs @() SP))+ (void (receiveMessage @() SP)) res @=? Right () ]
test/ProcessBehaviourTestCases.hs view
@@ -2,13 +2,14 @@ import Data.List ( sort ) import Data.Foldable ( traverse_ )-import Data.Dynamic ( toDyn )+import qualified Data.Dynamic as Dynamic import Data.Typeable import Data.Default import Control.Exception import Control.Concurrent import Control.Concurrent.STM import Control.Eff.Concurrent.Process+import Control.Eff.Concurrent.Process.Timer import Control.Eff.Concurrent.Api import Control.Eff.Concurrent.Api.Client import Control.Eff.Concurrent.Api.Server@@ -25,8 +26,12 @@ import Test.Tasty import Test.Tasty.HUnit import Common-import Debug.Trace+import Control.Applicative+import Data.Void +testInterruptReason :: InterruptReason+testInterruptReason = ProcessError "test interrupt"+ test_forkIo :: TestTree test_forkIo = setTravisTestOptions $ withTestLogC (\c lc -> handleLoggingAndIO_ (ForkIO.schedule c) lc)@@ -37,7 +42,9 @@ test_singleThreaded = setTravisTestOptions $ withTestLogC (\e logC -> -- void (runLift (logToChannel logC (SingleThreaded.schedule (return ()) e)))- let runEff :: Eff '[Logs LogMessage, LogWriterReader LogMessage IO, Lift IO] a -> IO a+ let runEff+ :: Eff '[Logs LogMessage, LogWriterReader LogMessage IO, Lift IO] a+ -> IO a runEff = flip handleLoggingAndIO logC in void $ SingleThreaded.scheduleM runEff yield e )@@ -46,7 +53,7 @@ allTests :: forall r . (Member (Logs LogMessage) r, SetMember Lift (Lift IO) r)- => IO (Eff (Process r ': r) () -> IO ())+ => IO (Eff (InterruptableProcess r) () -> IO ()) -> TestTree allTests schedulerFactory = localOption (timeoutSeconds 300)@@ -59,6 +66,9 @@ , pingPongTests schedulerFactory , yieldLoopTests schedulerFactory , selectiveReceiveTests schedulerFactory+ , linkingTests schedulerFactory+ , monitoringTests schedulerFactory+ , timerTests schedulerFactory ] ) @@ -75,7 +85,7 @@ returnToSender :: forall q r- . (HasCallStack, SetMember Process (Process q) r)+ . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r) => SchedulerProxy q -> Server ReturnToSender -> String@@ -83,12 +93,12 @@ returnToSender px toP msg = do me <- self px _ <- call px toP (ReturnToSender me msg)- msgEcho <- receiveMessageAs @String px+ msgEcho <- receiveMessage @String px return (msgEcho == msg) stopReturnToSender :: forall q r- . (HasCallStack, SetMember Process (Process q) r)+ . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r) => SchedulerProxy q -> Server ReturnToSender -> Eff r ()@@ -99,6 +109,7 @@ . ( HasCallStack , SetMember Process (Process q) r , Member (Logs LogMessage) q+ , Member Interrupts r ) => SchedulerProxy q -> Eff r (Server ReturnToSender)@@ -108,11 +119,11 @@ (\m k -> case m of StopReturnToSender -> do k ()- return (StopApiServer Nothing)+ return (StopApiServer testInterruptReason) ReturnToSender fromP echoMsg -> do- ok <- sendMessageChecked px fromP (toDyn echoMsg)+ sendMessage px fromP echoMsg yieldProcess px- k ok+ k True return HandleNextRequest ) }@@ -121,7 +132,7 @@ selectiveReceiveTests :: forall r . (Member (Logs LogMessage) r, SetMember Lift (Lift IO) r)- => IO (Eff (Process r ': r) () -> IO ())+ => IO (Eff (InterruptableProcess r) () -> IO ()) -> TestTree selectiveReceiveTests schedulerFactory = setTravisTestOptions (testGroup@@ -133,20 +144,19 @@ nMax = 10 receiverLoop donePid = go nMax where- go :: Int -> Eff (Process r ': r) ()- go 0 = sendMessageAs SP donePid True+ go :: Int -> Eff (InterruptableProcess r) ()+ go 0 = sendMessage SP donePid True go n = do- void $ receiveSelectedMessage- SP (filterMessage (==n))+ void $ receiveSelectedMessage SP (filterMessage (== n)) go (n - 1) senderLoop receviverPid =- traverse_ (sendMessageAs SP receviverPid) [1 .. nMax]+ traverse_ (sendMessage SP receviverPid) [1 .. nMax] me <- self SP receiverPid <- spawn (receiverLoop me) spawn_ (senderLoop receiverPid)- ok <- receiveMessageAs @Bool SP+ ok <- receiveMessage @Bool SP lift (ok @? "selective receive failed") , testCase "receive a message while waiting for a call reply" $ applySchedulerFactory schedulerFactory@@ -155,6 +165,25 @@ ok <- returnToSender SP srv "test" () <- stopReturnToSender SP srv lift (ok @? "selective receive failed")+ , testCase "flush messages"+ $ applySchedulerFactory schedulerFactory+ $ withSchedulerProxy SP+ $ do+ let px = SP+ me <- self px+ spawn_+ $ replicateM_ 10 (sendMessage px me True) >> sendMessage px me ()+ spawn_ $ replicateM_+ 10+ (sendMessage px me (123.23411 :: Float)) >> sendMessage px me ()+ spawn_+ $ replicateM_ 10 (sendMessage px me "123") >> sendMessage px me ()+ () <- receiveMessage px+ () <- receiveMessage px+ () <- receiveMessage px+ -- replicateCheapM_ 40 (yieldProcess px)+ msgs <- flushMessages+ lift (length msgs @?= 30) ] ) @@ -162,7 +191,7 @@ yieldLoopTests :: forall r . (Member (Logs LogMessage) r, SetMember Lift (Lift IO) r)- => IO (Eff (Process r ': r) () -> IO ())+ => IO (Eff (InterruptableProcess r) () -> IO ()) -> TestTree yieldLoopTests schedulerFactory = let maxN = 100000@@ -195,13 +224,15 @@ data Ping = Ping ProcessId+ deriving (Eq, Show)+ data Pong = Pong deriving (Eq, Show) pingPongTests :: forall r . (Member (Logs LogMessage) r, SetMember Lift (Lift IO) r)- => IO (Eff (Process r ': r) () -> IO ())+ => IO (Eff (InterruptableProcess r) () -> IO ()) -> TestTree pingPongTests schedulerFactory = testGroup "Yield Tests"@@ -209,17 +240,17 @@ $ applySchedulerFactory schedulerFactory $ do let pongProc = foreverCheap $ do- Ping pinger <- receiveMessageAs SP- sendMessageAs SP pinger Pong+ Ping pinger <- receiveMessage SP+ sendMessage SP pinger Pong pingProc ponger parent = do me <- self SP- sendMessageAs SP ponger (Ping me)- Pong <- receiveMessageAs SP- sendMessageAs SP parent True+ sendMessage SP ponger (Ping me)+ Pong <- receiveMessage SP+ sendMessage SP parent True pongPid <- spawn pongProc me <- self SP spawn_ (pingProc pongPid me)- ok <- receiveMessageAs @Bool SP+ ok <- receiveMessage @Bool SP lift (ok @? "ping pong failed") , testCase "ping pong a message between two processes, with massive yielding" $ applySchedulerFactory schedulerFactory@@ -227,19 +258,19 @@ yieldProcess SP let pongProc = foreverCheap $ do yieldProcess SP- Ping pinger <- receiveMessageAs SP+ Ping pinger <- receiveMessage SP yieldProcess SP- sendMessageAs SP pinger Pong+ sendMessage SP pinger Pong yieldProcess SP pingProc ponger parent = do yieldProcess SP me <- self SP yieldProcess SP- sendMessageAs SP ponger (Ping me)+ sendMessage SP ponger (Ping me) yieldProcess SP- Pong <- receiveMessageAs SP+ Pong <- receiveMessage SP yieldProcess SP- sendMessageAs SP parent True+ sendMessage SP parent True yieldProcess SP yieldProcess SP pongPid <- spawn pongProc@@ -248,7 +279,7 @@ yieldProcess SP spawn_ (pingProc pongPid me) yieldProcess SP- ok <- receiveMessageAs @Bool SP+ ok <- receiveMessage @Bool SP yieldProcess SP lift (ok @? "ping pong failed") yieldProcess SP@@ -258,10 +289,10 @@ $ do pongVar <- lift newEmptyMVar let pongProc = foreverCheap $ do- Pong <- receiveMessageAs SP+ Pong <- receiveMessage SP lift (putMVar pongVar Pong) ponger <- spawn pongProc- sendMessageAs SP ponger Pong+ sendMessage SP ponger Pong let waitLoop = do p <- lift (tryTakeMVar pongVar) case p of@@ -276,7 +307,7 @@ errorTests :: forall r . (Member (Logs LogMessage) r, SetMember Lift (Lift IO) r)- => IO (Eff (Process r ': r) () -> IO ())+ => IO (Eff (InterruptableProcess r) () -> IO ()) -> TestTree errorTests schedulerFactory = let@@ -286,66 +317,47 @@ testGroup "causing and handling errors" [ testGroup- "raiseError"- [ testCase "unhandled raiseError"- $ applySchedulerFactory schedulerFactory- $ do- void $ raiseError px "test error"- error "This should not happen"- , testCase "catch raiseError 1"- $ scheduleAndAssert schedulerFactory- $ \assertEff -> catchRaisedError- px- (assertEff "error must be caught" . (== "test error 2"))- (void (raiseError px "test error 2"))- , testCase "catch raiseError from a long sub block"- $ scheduleAndAssert schedulerFactory- $ \assertEff -> catchRaisedError- px- (assertEff "error must be caught" . (== "test error 3"))- (do- replicateM_ 100000 (void (self px))- void (raiseError px "test error 3")- )- ]- , testGroup- "exitWithError"- [ testCase "unhandled exitWithError"- $ applySchedulerFactory schedulerFactory- $ do- void $ exitWithError px "test error"- error "This should not happen"- , testCase "cannot catch exitWithError"- $ applySchedulerFactory schedulerFactory- $ do- void $ ignoreProcessError px $ exitWithError px "test error 4"- error "This should not happen"- , testCase "multi process exitWithError"- $ scheduleAndAssert schedulerFactory- $ \assertEff -> do- me <- self px- let n = 15- traverse_- (\(i :: Int) -> spawn $ if i `rem` 5 == 0- then do- void $ sendMessage px me (toDyn i)- void (exitWithError px (show i ++ " died"))- assertEff "this should not be reached" False- else- foreverCheap- (void (sendMessage px 888 (toDyn "test message to 888"))+ "exitWithError"+ [ testCase "unhandled exitWithError"+ $ applySchedulerFactory schedulerFactory+ $ do+ void $ exitWithError px "test error"+ error "This should not happen"+ , testCase "cannot catch exitWithError"+ $ applySchedulerFactory schedulerFactory+ $ do+ void $ exitWithError px "test error 4"+ error "This should not happen"+ , testCase "multi process exitWithError"+ $ scheduleAndAssert schedulerFactory+ $ \assertEff -> do+ me <- self px+ let n = 15+ traverse_+ (\(i :: Int) -> spawn $ foreverCheap+ (void+ ( sendMessage px (888888 + fromIntegral i) "test message"+ >> yieldProcess px )- )- [0, 5 .. n]- oks <- replicateM (length [0, 5 .. n]) (receiveMessageAs px)- assertEff "" (sort oks == [0, 5 .. n])- ]+ )+ )+ [0 .. n]+ traverse_+ (\(i :: Int) -> spawn $ do+ void $ sendMessage px me i+ void (exitWithError px (show i ++ " died"))+ error "this should not be reached"+ )+ [0 .. n]+ oks <- replicateM (length [0 .. n]) (receiveMessage px)+ assertEff "" (sort oks == [0 .. n])+ ] ] concurrencyTests :: forall r . (Member (Logs LogMessage) r, SetMember Lift (Lift IO) r)- => IO (Eff (Process r ': r) () -> IO ())+ => IO (Eff (InterruptableProcess r) () -> IO ()) -> TestTree concurrencyTests schedulerFactory = let@@ -364,7 +376,7 @@ (const (spawn (do- m <- receiveMessage px+ m <- receiveAnyMessage px void (sendMessage px me m) ) )@@ -384,17 +396,13 @@ me <- self px child1 <- spawn (do- m <- receiveMessage px- void (sendMessage px me m)+ m <- receiveAnyMessage px+ void (sendAnyMessage px me m) )- child2 <- spawn- (foreverCheap (void (sendMessage px 888 (toDyn ""))))- True <- sendMessageChecked px child1 (toDyn "test")- traceM "now receiveMessageAs"- i <- receiveMessageAs px- traceM ("receiveMessageAs returned " ++ show i)- True <- sendShutdownChecked px child2 ExitNormally- traceM "sendShutdownChecked"+ child2 <- spawn (foreverCheap (void (sendMessage px 888888 "")))+ sendMessage px child1 "test"+ i <- receiveMessage px+ sendInterrupt px child2 testInterruptReason assertEff "" (i == "test") , testCase "most processes send foreverCheap" $ scheduleAndAssert schedulerFactory@@ -402,12 +410,11 @@ me <- self px traverse_ (\(i :: Int) -> spawn $ do- when (i `rem` 5 == 0) $ void $ sendMessage px me (toDyn i)- foreverCheap- $ void (sendMessage px 888 (toDyn "test message to 888"))+ when (i `rem` 5 == 0) $ void $ sendMessage px me i+ foreverCheap $ void (sendMessage px 888 "test message to 888") ) [0 .. n]- oks <- replicateM (length [0, 5 .. n]) (receiveMessageAs px)+ oks <- replicateM (length [0, 5 .. n]) (receiveMessage px) assertEff "" (sort oks == [0, 5 .. n]) , testCase "most processes self foreverCheap" $ scheduleAndAssert schedulerFactory@@ -415,11 +422,11 @@ me <- self px traverse_ (\(i :: Int) -> spawn $ do- when (i `rem` 5 == 0) $ void $ sendMessage px me (toDyn i)+ when (i `rem` 5 == 0) $ void $ sendMessage px me i foreverCheap $ void (self px) ) [0 .. n]- oks <- replicateM (length [0, 5 .. n]) (receiveMessageAs px)+ oks <- replicateM (length [0, 5 .. n]) (receiveMessage px) assertEff "" (sort oks == [0, 5 .. n]) , testCase "most processes sendShutdown foreverCheap" $ scheduleAndAssert schedulerFactory@@ -427,11 +434,11 @@ me <- self px traverse_ (\(i :: Int) -> spawn $ do- when (i `rem` 5 == 0) $ void $ sendMessage px me (toDyn i)+ when (i `rem` 5 == 0) $ void $ sendMessage px me i foreverCheap $ void (sendShutdown px 999 ExitNormally) ) [0 .. n]- oks <- replicateM (length [0, 5 .. n]) (receiveMessageAs px)+ oks <- replicateM (length [0, 5 .. n]) (receiveMessage px) assertEff "" (sort oks == [0, 5 .. n]) , testCase "most processes spawn foreverCheap" $ scheduleAndAssert schedulerFactory@@ -439,19 +446,16 @@ me <- self px traverse_ (\(i :: Int) -> spawn $ do- when (i `rem` 5 == 0) $ void $ sendMessage px me (toDyn i)+ when (i `rem` 5 == 0) $ void $ sendMessage px me i parent <- self px foreverCheap $ void (spawn- (void- (sendMessage px parent (toDyn "test msg from child")- )- )+ (void (sendMessage px parent "test msg from child")) ) ) [0 .. n]- oks <- replicateM (length [0, 5 .. n]) (receiveMessageAs px)+ oks <- replicateM (length [0, 5 .. n]) (receiveMessage px) assertEff "" (sort oks == [0, 5 .. n]) , testCase "most processes receive foreverCheap" $ scheduleAndAssert schedulerFactory@@ -459,18 +463,18 @@ me <- self px traverse_ (\(i :: Int) -> spawn $ do- when (i `rem` 5 == 0) $ void $ sendMessage px me (toDyn i)- foreverCheap $ void (receiveMessage px)+ when (i `rem` 5 == 0) $ void $ sendMessage px me i+ foreverCheap $ void (receiveAnyMessage px) ) [0 .. n]- oks <- replicateM (length [0, 5 .. n]) (receiveMessageAs px)+ oks <- replicateM (length [0, 5 .. n]) (receiveMessage px) assertEff "" (sort oks == [0, 5 .. n]) ] exitTests :: forall r . (Member (Logs LogMessage) r, SetMember Lift (Lift IO) r)- => IO (Eff (Process r ': r) () -> IO ())+ => IO (Eff (InterruptableProcess r) () -> IO ()) -> TestTree exitTests schedulerFactory = let@@ -506,10 +510,15 @@ | e <- [ThreadKilled, UserInterrupt, HeapOverflow, StackOverflow] , (busyWith, busyEffect) <- [ ( "receiving"- , void (send (ReceiveSelectedMessage @r (filterMessage (=="test message"))))+ , void+ (send+ (ReceiveSelectedMessage @r (filterMessage (== "test message"))+ )+ ) ) , ( "sending"- , void (send (SendMessage @r 44444 (toDyn "test message")))+ , void+ (send (SendMessage @r 44444 (Dynamic.toDyn "test message"))) ) , ( "sending shutdown" , void (send (SendShutdown @r 44444 ExitNormally))@@ -537,10 +546,15 @@ lift (threadDelay 10000) | (busyWith, busyEffect) <- [ ( "receiving"- , void (send (ReceiveSelectedMessage @r (filterMessage (=="test message"))))+ , void+ (send+ (ReceiveSelectedMessage @r (filterMessage (== "test message"))+ )+ ) ) , ( "sending"- , void (send (SendMessage @r 44444 (toDyn "test message")))+ , void+ (send (SendMessage @r 44444 (Dynamic.toDyn "test message"))) ) , ( "sending shutdown" , void (send (SendShutdown @r 44444 ExitNormally))@@ -574,15 +588,23 @@ lift (threadDelay 1000) doExit lift (threadDelay 100000)- wasStillRunningP1 <- sendShutdownChecked px p1 ExitNormally- assertEff "the other process was still running"- wasStillRunningP1+ wasRunningP1 <- isProcessAlive SP p1+ sendShutdown px p1 ExitNormally+ lift (threadDelay 100000)+ stillRunningP1 <- isProcessAlive SP p1+ assertEff "the other process did not die still running"+ (not stillRunningP1 && wasRunningP1) | (busyWith , busyEffect) <- [ ( "receiving"- , void (send (ReceiveSelectedMessage @r (filterMessage (=="test message"))))+ , void+ (send+ (ReceiveSelectedMessage @r (filterMessage (== "test message"))+ )+ ) ) , ( "sending"- , void (send (SendMessage @r 44444 (toDyn "test message")))+ , void+ (send (SendMessage @r 44444 (Dynamic.toDyn "test message"))) ) , ( "sending shutdown" , void (send (SendShutdown @r 44444 ExitNormally))@@ -602,7 +624,9 @@ , (howToExit, doExit ) <- [ ("normally" , void (exitNormally px)) , ("simply returning", return ())- , ("raiseError", void (raiseError px "test error raised"))+ , ( "raiseError"+ , void (interrupt (ProcessError "test error raised"))+ ) , ("exitWithError", void (exitWithError px "test error exit")) , ( "sendShutdown to self" , do@@ -617,31 +641,27 @@ sendShutdownTests :: forall r . (Member (Logs LogMessage) r, SetMember Lift (Lift IO) r)- => IO (Eff (Process r ': r) () -> IO ())+ => IO (Eff (InterruptableProcess r) () -> IO ()) -> TestTree-sendShutdownTests schedulerFactory- = let- px :: SchedulerProxy r+sendShutdownTests schedulerFactory =+ let px :: SchedulerProxy r px = SchedulerProxy- in- testGroup+ in testGroup "sendShutdown" [ testCase "... self" $ applySchedulerFactory schedulerFactory $ do me <- self px- void $ sendShutdown px me ExitNormally- raiseError px "sendShutdown must not return"- , testCase "... self low-level"+ void $ send (SendShutdown @r me ExitNormally)+ interrupt (ProcessError "sendShutdown must not return")+ , testCase "sendInterrupt to self" $ scheduleAndAssert schedulerFactory $ \assertEff -> do me <- self px- r <- send (SendShutdown @r me ExitNormally)- traceShowM- (show me ++ ": returned from SendShutdow to self " ++ show r)+ r <- send (SendInterrupt @r me (ProcessError "123")) assertEff- "ShutdownRequested must be returned"+ "Interrupted must be returned" (case r of- ShutdownRequested ExitNormally -> True- _ -> False+ Interrupted (ProcessError "123") -> True+ _ -> False ) , testGroup "... other process"@@ -651,11 +671,12 @@ me <- self px other <- spawn (do- untilShutdown (SendMessage @r 666 (toDyn "test"))- void (sendMessage px me (toDyn "OK"))+ untilInterrupted+ (SendMessage @r 666666 (Dynamic.toDyn "test"))+ void (sendMessage px me "OK") )- void (sendShutdown px other ExitNormally)- a <- receiveMessageAs px+ void (sendInterrupt px other testInterruptReason)+ a <- receiveMessage px assertEff "" (a == "OK") , testCase "while it is receiving" $ scheduleAndAssert schedulerFactory@@ -663,12 +684,12 @@ me <- self px other <- spawn (do- untilShutdown+ untilInterrupted (ReceiveSelectedMessage @r selectAnyMessageLazy)- void (sendMessage px me (toDyn "OK"))+ void (sendMessage px me "OK") )- void (sendShutdown px other ExitNormally)- a <- receiveMessageAs px+ void (sendInterrupt px other testInterruptReason)+ a <- receiveMessage px assertEff "" (a == "OK") , testCase "while it is self'ing" $ scheduleAndAssert schedulerFactory@@ -676,11 +697,11 @@ me <- self px other <- spawn (do- untilShutdown (SelfPid @r)- void (sendMessage px me (toDyn "OK"))+ untilInterrupted (SelfPid @r)+ void (sendMessage px me "OK") )- void (sendShutdown px other (ExitWithError "testError"))- a <- receiveMessageAs px+ void (sendInterrupt px other (ProcessError "testError"))+ a <- receiveMessage px assertEff "" (a == "OK") , testCase "while it is spawning" $ scheduleAndAssert schedulerFactory@@ -688,11 +709,11 @@ me <- self px other <- spawn (do- untilShutdown (Spawn @r (return ()))- void (sendMessage px me (toDyn "OK"))+ untilInterrupted (Spawn @r (return ()))+ void (sendMessage px me "OK") )- void (sendShutdown px other ExitNormally)- a <- receiveMessageAs px+ void (sendInterrupt px other testInterruptReason)+ a <- receiveMessage px assertEff "" (a == "OK") , testCase "while it is sending shutdown messages" $ scheduleAndAssert schedulerFactory@@ -700,11 +721,309 @@ me <- self px other <- spawn (do- untilShutdown (SendShutdown @r 777 ExitNormally)- void (sendMessage px me (toDyn "OK"))+ untilInterrupted (SendShutdown @r 777 ExitNormally)+ void (sendMessage px me "OK") )- void (sendShutdown px other ExitNormally)- a <- receiveMessageAs px+ void (sendInterrupt px other testInterruptReason)+ a <- receiveMessage px assertEff "" (a == "OK")+ , testCase "handleInterrupt handles my own interrupts"+ $ scheduleAndAssert schedulerFactory+ $ \assertEff ->+ handleInterrupts+ (\e -> return (ProcessError "test" == e))+ (interrupt (ProcessError "test") >> return False)+ >>= assertEff "exception handler not invoked" ] ]++linkingTests+ :: forall r+ . (Member (Logs LogMessage) r, SetMember Lift (Lift IO) r)+ => IO (Eff (InterruptableProcess r) () -> IO ())+ -> TestTree+linkingTests schedulerFactory = setTravisTestOptions+ (testGroup+ "process linking tests"+ [ testCase "link process with it self"+ $ applySchedulerFactory schedulerFactory+ $ do+ me <- self SP+ handleInterrupts+ (\er -> lift (False @? ("unexpected interrupt: " ++ show er)))+ (do+ linkProcess SP me+ lift (threadDelay 10000)+ )+ , testCase "link with not running process"+ $ applySchedulerFactory schedulerFactory+ $ do+ let testPid = 234234234+ handleInterrupts+ (lift . (@?= LinkedProcessCrashed testPid))+ (do+ linkProcess SP testPid+ void (receiveMessage @Void SP)+ )+ , testCase "linked process exit message is NormalExit"+ $ applySchedulerFactory schedulerFactory+ $ do+ foo <- spawn (void (receiveMessage @Void SP))+ handleInterrupts+ (lift . (\e -> e /= LinkedProcessCrashed foo @? show e))+ (do+ linkProcess SP foo+ sendShutdown SP foo ExitNormally+ lift (threadDelay 1000)+ )+ , testCase "linked process exit message is Crash"+ $ applySchedulerFactory schedulerFactory+ $ do+ foo <- spawn (void (receiveMessage @Void SP))+ handleInterrupts+ (lift . (@?= LinkedProcessCrashed foo))+ (do+ linkProcess SP foo+ sendShutdown SP foo Killed+ void (receiveMessage @Void SP)+ )+ , testCase "link multiple times"+ $ applySchedulerFactory schedulerFactory+ $ do+ foo <- spawn (void (receiveMessage @Void SP))+ handleInterrupts+ (lift . (@?= LinkedProcessCrashed foo))+ (do+ linkProcess SP foo+ linkProcess SP foo+ linkProcess SP foo+ linkProcess SP foo+ linkProcess SP foo+ linkProcess SP foo+ linkProcess SP foo+ sendShutdown SP foo Killed+ void (receiveMessage @Void SP)+ )+ , testCase "unlink multiple times"+ $ applySchedulerFactory schedulerFactory+ $ do+ foo <- spawn (void (receiveMessage @Void SP))+ handleInterrupts+ (lift . (False @?) . show)+ (do+ spawn_ (void (receiveAnyMessage SP))+ linkProcess SP foo+ linkProcess SP foo+ linkProcess SP foo+ unlinkProcess SP foo+ unlinkProcess SP foo+ unlinkProcess SP foo+ unlinkProcess SP foo+ withMonitor SP foo $ \ref -> do+ sendShutdown SP foo Killed+ void (receiveSelectedMessage SP (selectProcessDown ref))+ )+ , testCase "spawnLink" $ applySchedulerFactory schedulerFactory $ do+ let foo = void (receiveMessage @Void SP)+ handleInterrupts (\er -> lift (isBecauseDown Nothing er @? show er)) $ do+ x <- spawnLink foo+ sendShutdown SP x Killed+ void (receiveMessage @Void SP)+ , testCase "ignore normal exit"+ $ applySchedulerFactory schedulerFactory+ $ do+ mainProc <- self SP+ let linkingServer = void $ exitOnInterrupt SP $ do+ logNotice "linker"+ foreverCheap $ do+ x <- receiveMessage SP+ linkProcess SP x+ sendMessage SP mainProc True+ linker <- spawnLink linkingServer+ logNotice "mainProc"+ do+ x <- spawnLink (logNotice "x 1" >> void (receiveMessage @Void SP))+ withMonitor SP x $ \xRef -> do+ sendMessage SP linker x+ void $ receiveSelectedMessage SP (filterMessage id)+ sendShutdown SP x ExitNormally+ void (receiveSelectedMessage SP (selectProcessDown xRef))+ do+ x <- spawnLink (logNotice "x 2" >> void (receiveMessage @Void SP))+ withMonitor SP x $ \xRef -> do+ sendMessage SP linker x+ void $ receiveSelectedMessage SP (filterMessage id)+ sendShutdown SP x ExitNormally+ void (receiveSelectedMessage SP (selectProcessDown xRef))+ handleInterrupts (lift . (LinkedProcessCrashed linker @=?)) $ do+ sendShutdown SP linker Killed+ void (receiveMessage @Void SP)+ , testCase "unlink" $ applySchedulerFactory schedulerFactory $ do+ let+ foo1 = void (receiveAnyMessage SP)+ foo2 foo1Pid = do+ linkProcess SP foo1Pid+ ("unlink foo1", barPid) <- receiveMessage SP+ unlinkProcess SP foo1Pid+ sendMessage SP barPid ("unlinked foo1", foo1Pid)+ "the end" <- receiveMessage SP+ exitWithError SP "foo two"+ bar foo2Pid parentPid = do+ linkProcess SP foo2Pid+ me <- self SP+ sendMessage SP foo2Pid ("unlink foo1", me)+ ("unlinked foo1", foo1Pid) <- receiveMessage SP+ handleInterrupts+ (const (return ()))+ (do+ linkProcess SP foo1Pid+ sendShutdown SP foo1Pid Killed+ void (receiveMessage @Void SP)+ )+ handleInterrupts+ (\er ->+ void+ (sendMessage SP parentPid (LinkedProcessCrashed foo2Pid == er))+ )+ (do+ sendMessage SP foo2Pid "the end"+ void (receiveAnyMessage SP)+ )+ foo1Pid <- spawn foo1+ foo2Pid <- spawn (foo2 foo1Pid)+ me <- self SP+ barPid <- spawn (bar foo2Pid me)+ handleInterrupts+ (\er -> lift (LinkedProcessCrashed barPid @?= er))+ (do+ res <- receiveMessage @Bool SP+ lift (threadDelay 100000)+ lift (res @?= True)+ )+ ]+ )++monitoringTests+ :: forall r+ . (Member (Logs LogMessage) r, SetMember Lift (Lift IO) r)+ => IO (Eff (InterruptableProcess r) () -> IO ())+ -> TestTree+monitoringTests schedulerFactory = setTravisTestOptions+ (testGroup+ "process monitoring tests"+ [ testCase "monitored process not running"+ $ applySchedulerFactory schedulerFactory+ $ do+ let badPid = 132123+ ref <- monitor SP badPid+ pd <- receiveSelectedMessage SP (selectProcessDown ref)+ lift (downReason pd @?= SomeExitReason (ProcessNotRunning badPid))+ lift (threadDelay 10000)+ , testCase "monitored process exit normally"+ $ applySchedulerFactory schedulerFactory+ $ do+ target <- spawn (receiveMessage SP >>= exitBecause SP)+ ref <- monitor SP target+ sendMessage SP target ExitNormally+ pd <- receiveSelectedMessage SP (selectProcessDown ref)+ lift (downReason pd @?= SomeExitReason ExitNormally)+ lift (threadDelay 10000)+ , testCase "multiple monitors some demonitored"+ $ applySchedulerFactory schedulerFactory+ $ do+ target <- spawn (receiveMessage SP >>= exitBecause SP)+ ref1 <- monitor SP target+ ref2 <- monitor SP target+ ref3 <- monitor SP target+ ref4 <- monitor SP target+ ref5 <- monitor SP target+ demonitor SP ref3+ demonitor SP ref5+ sendMessage SP target ExitNormally+ pd1 <- receiveSelectedMessage SP (selectProcessDown ref1)+ lift (downReason pd1 @?= SomeExitReason ExitNormally)+ pd2 <- receiveSelectedMessage SP (selectProcessDown ref2)+ lift (downReason pd2 @?= SomeExitReason ExitNormally)+ pd4 <- receiveSelectedMessage SP (selectProcessDown ref4)+ lift (downReason pd4 @?= SomeExitReason ExitNormally)+ lift (threadDelay 10000)+ , testCase "monitored process killed"+ $ applySchedulerFactory schedulerFactory+ $ do+ target <- spawn (receiveMessage SP >>= exitBecause SP)+ ref <- monitor SP target+ sendMessage SP target Killed+ pd <- receiveSelectedMessage SP (selectProcessDown ref)+ lift (downReason pd @?= SomeExitReason Killed)+ lift (threadDelay 10000)+ , testCase "demonitored process killed"+ $ applySchedulerFactory schedulerFactory+ $ do+ target <- spawn (receiveMessage SP >>= exitBecause SP)+ ref <- monitor SP target+ demonitor SP ref+ sendMessage SP target Killed+ me <- self SP+ spawn_ (lift (threadDelay 10000) >> sendMessage SP me ())+ pd <- receiveSelectedMessage+ SP+ (Right <$> selectProcessDown ref <|> Left <$> selectMessage @())+ lift (pd @?= Left ())+ lift (threadDelay 10000)+ ]+ )++timerTests+ :: forall r+ . (Member (Logs LogMessage) r, SetMember Lift (Lift IO) r)+ => IO (Eff (InterruptableProcess r) () -> IO ())+ -> TestTree+timerTests schedulerFactory = setTravisTestOptions+ (testGroup+ "process timer tests"+ [ testCase "receiveAfter into timeout"+ $ applySchedulerFactory schedulerFactory+ $ do+ pd <- receiveAfter @Void SP 1000+ lift (pd @?= Nothing)+ lift (threadDelay 10000)+ , testCase "receiveAfter no timeout"+ $ applySchedulerFactory schedulerFactory+ $ do+ me <- self SP+ other <- spawn+ (do+ () <- receiveMessage @() SP+ sendMessage SP me (123 :: Int)+ )+ pd1 <- receiveAfter @() SP 10000+ lift (pd1 @?= Nothing)+ sendMessage SP other ()+ pd2 <- receiveAfter @Int SP 10000+ lift (pd2 @?= Just 123)+ lift (threadDelay 10000)+ , testCase "many receiveAfters"+ $ applySchedulerFactory schedulerFactory+ $ do+ let n = 5+ testMsg :: Float+ testMsg = 123+ me <- self SP+ other <- spawn+ (do+ replicateM_ n $ sendMessage SP me "bad message"+ () <- receiveMessage @() SP+ replicateM_ n $ sendMessage SP me testMsg+ )+ receiveAfter @Float SP 100 >>= lift . (@?= Nothing)+ sendMessage SP other ()+ replicateM_+ n+ (do+ res <- receiveAfter @Float SP 10000+ lift (res @?= Just testMsg)+ )++ lift (threadDelay 100)+ ]+ )
test/SingleThreadedScheduler.hs view
@@ -1,5 +1,4 @@-module SingleThreadedScheduler-where+module SingleThreadedScheduler where import Control.Eff.Loop import Control.Eff.Concurrent.Process@@ -19,19 +18,19 @@ @=? Scheduler.schedulePure (do adderChild <- spawn $ do- (from, arg1, arg2) <- receiveMessageAs SP- sendMessageAs SP from ((arg1 + arg2) :: Int)- foreverCheap $ void $ receiveMessage SP+ (from, arg1, arg2) <- receiveMessage SP+ sendMessage SP from ((arg1 + arg2) :: Int)+ foreverCheap $ void $ receiveAnyMessage SP multChild <- spawn $ do- (from, arg1, arg2) <- receiveMessageAs SP- sendMessageAs SP from ((arg1 * arg2) :: Int)+ (from, arg1, arg2) <- receiveMessage SP+ sendMessage 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+ sendMessage SP adderChild (me, 3 :: Int, 4 :: Int)+ x <- receiveMessage @Int SP+ sendMessage SP multChild (me, x, 6 :: Int)+ receiveMessage @Int SP ) ] @@ -40,9 +39,10 @@ test_mainProcessSpawnsAChildAndExitsNormally = setTravisTestOptions (testCase "spawn a child and exit normally"- (Scheduler.defaultMain+ (Scheduler.defaultMainSingleThreaded (do- void (spawn (void (receiveMessage singleThreadedIoScheduler)))+ void+ (spawn (void (receiveAnyMessage singleThreadedIoScheduler))) void (exitNormally singleThreadedIoScheduler) fail "This should not happen!!" )@@ -54,18 +54,15 @@ test_mainProcessSpawnsAChildBothExitNormally = setTravisTestOptions (testCase "spawn a child and let it exit and exit"- (Scheduler.defaultMain+ (Scheduler.defaultMainSingleThreaded (do child <- spawn (do- void- (receiveMessageAs @String singleThreadedIoScheduler)+ void (receiveMessage @String singleThreadedIoScheduler) void (exitNormally singleThreadedIoScheduler) error "This should not happen (child)!!" )- True <- sendMessageChecked singleThreadedIoScheduler- child- (toDyn "test")+ sendMessage singleThreadedIoScheduler child (toDyn "test") void (exitNormally singleThreadedIoScheduler) error "This should not happen!!" )