extensible-effects-concurrent 0.8 → 0.9.0
raw patch · 22 files changed
+1334/−981 lines, 22 filesdep +asyncdep +enclosed-exceptions
Dependencies added: async, enclosed-exceptions
Files
- ChangeLog.md +26/−4
- examples/example-1/Main.hs +48/−28
- examples/example-2/Main.hs +34/−21
- extensible-effects-concurrent.cabal +8/−1
- src/Control/Eff/Concurrent.hs +24/−2
- src/Control/Eff/Concurrent/Api/Client.hs +25/−13
- src/Control/Eff/Concurrent/Api/Internal.hs +5/−6
- src/Control/Eff/Concurrent/Api/Observer.hs +17/−24
- src/Control/Eff/Concurrent/Api/Observer/Queue.hs +66/−15
- src/Control/Eff/Concurrent/Api/Server.hs +185/−54
- src/Control/Eff/Concurrent/Process.hs +192/−76
- src/Control/Eff/Concurrent/Process/ForkIOScheduler.hs +361/−461
- src/Control/Eff/Concurrent/Process/SingleThreadedScheduler.hs +88/−45
- src/Control/Eff/Log/Channel.hs +24/−6
- src/Control/Eff/Log/Handler.hs +40/−73
- src/Control/Eff/Log/Message.hs +76/−25
- stack.yaml +13/−13
- test/Common.hs +8/−12
- test/Debug.hs +0/−1
- test/ForkIOScheduler.hs +11/−10
- test/LoggingTests.hs +23/−17
- test/ProcessBehaviourTestCases.hs +60/−74
ChangeLog.md view
@@ -1,5 +1,31 @@ # Changelog for extensible-effects-concurrent +## 0.9.0++- Make `ForkIOScheduler` faster and more robust+- Add `ProcessExitReason`+- Add `ProcessState`+- Add `ShutdownRequest` type+- Rewrite logging to be a `Reader` of a `LogWriter`+- Remove pure logging, the `Logs...` constraint must be+ accompanied by `Lifted IO` (or `MonadIO`) in many log functions+ most prominently `logMsg`+- Add a `lmDistance` field in `LogMessage`+- Add `increaseLogMessageDistance` and `dropDistantLogMessages`+ using the new `lmDistance` field+- Add a newtype for the argument to selective receives: `MessageSelector`+- Add a `makeReference` function to `Process` which will return process local+ unique `Int`s+- Rename `spawnServer` to `spawnServerWithEffects` and add a simpler version of+ `spawnServerWithEffects` called `spawnServer`+- Make all `ApiHandler` handler callbacks optional (by changing the type to `Maybe ...`)+- `ApiHandler` must now return an `ApiServerCmd`.+- Add `ApiServerCmd` which allows handler functions to leave to server loop without+ exitting the process+- Fix `Observer.Queue`+- Rename fields in `ApiHandler`+- Add smart constructors for `ApiHandler`+ ## 0.8.0 - Add selective receive@@ -120,10 +146,6 @@ - Change `Api.Server` `serve` to loop instead of handling just one request - Allow combining multiple `ApiHandler` such that one process can handle multiple APIs--### TODO--- Add `Kill` action to `Process` ## 0.1.3.0
examples/example-1/Main.hs view
@@ -48,22 +48,23 @@ example px = do me <- self px logInfo ("I am " ++ show me)- server <- asServer @TestApi <$> spawn testServerLoop+ server <- testServerLoop px logInfo ("Started server " ++ show server) let go = do+ lift (putStr "Enter something: ") x <- lift getLine case x of- ('k' : rest) -> do+ ('K' : rest) -> do callRegistered px (TerminateError rest) go- ('s' : _) -> do+ ('S' : _) -> do callRegistered px Terminate go- ('c' : _) -> do+ ('C' : _) -> do castRegistered px (Shout x) go- ('r' : rest) -> do- void (replicateM (read rest) (castRegistered px (Shout x)))+ ('R' : rest) -> do+ replicateM_ (read rest) (castRegistered px (Shout x)) go ('q' : _) -> logInfo "Done." _ -> do@@ -73,57 +74,76 @@ registerServer server go testServerLoop- :: forall r- . (HasCallStack, Member (Logs LogMessage) r, SetMember Lift (Lift IO) r)- => Eff (Process r ': r) ()-testServerLoop = serve px $ ApiHandler handleCast handleCall handleTerminate+ :: forall r q+ . ( HasCallStack+ , Member (Logs LogMessage) q+ , SetMember Lift (Lift IO) q+ , SetMember Process (Process q) r+ )+ => SchedulerProxy q+ -> Eff r (Server TestApi)+testServerLoop px = spawnServer px+ $ ApiHandler (Just handleCastTest) (Just handleCallTest) Nothing -- (Just handleTerminateTest) where- px :: SchedulerProxy r- px = SchedulerProxy- handleCast :: Api TestApi 'Asynchronous -> Eff (Process r ': r) ()- handleCast (Shout x) = do+ handleCastTest+ :: Api TestApi 'Asynchronous -> Eff (Process q ': q) ApiServerCmd+ handleCastTest (Shout x) = do me <- self px logInfo (show me ++ " Shouting: " ++ x)- handleCall+ return HandleNextRequest+ handleCallTest :: Api TestApi ( 'Synchronous x)- -> (x -> Eff (Process r ': r) ())- -> Eff (Process r ': r) ()- handleCall (SayHello "e1") _reply = do+ -> (x -> Eff (Process q ': q) ())+ -> Eff (Process q ': q) ApiServerCmd+ handleCallTest (SayHello "e1") _reply = do me <- self px logInfo (show me ++ " raising an error") raiseError px "No body loves me... :,("- handleCall (SayHello "e2") _reply = do+ handleCallTest (SayHello "e2") _reply = do me <- self px logInfo (show me ++ " throwing a MyException ") lift (Exc.throw MyException)- handleCall (SayHello "self") reply = do+ handleCallTest (SayHello "self") reply = do me <- self px logInfo (show me ++ " casting to self") cast px (asServer @TestApi me) (Shout "from me") void (reply False)- handleCall (SayHello "die") reply = do+ return HandleNextRequest+ handleCallTest (SayHello "stop") reply = do 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)- handleCall (SayHello x) reply = do+ return HandleNextRequest+ handleCallTest (SayHello x) reply = do me <- self px logInfo (show me ++ " Got Hello: " ++ x) void (reply (length x > 3))- handleCall Terminate reply = do+ return HandleNextRequest+ handleCallTest Terminate reply = do me <- self px logInfo (show me ++ " exiting") void (reply ()) exitNormally px- handleCall (TerminateError msg) reply = do+ handleCallTest (TerminateError msg) reply = do me <- self px logInfo (show me ++ " exiting with error: " ++ msg) void (reply ()) exitWithError px msg- handleTerminate msg = do- me <- self px- logInfo (show me ++ " is exiting: " ++ show msg)- maybe (exitNormally px) (exitWithError px) msg+ -- handleTerminateTest msg = do+ -- me <- self px+ -- logInfo (show me ++ " is exiting: " ++ show msg)+ -- maybe (exitNormally px) (exitWithError px) msg
examples/example-2/Main.hs view
@@ -9,6 +9,7 @@ as Pure import Control.Eff.Concurrent import Control.Eff.State.Strict+import Control.Eff.Lift import Control.Monad main :: IO ()@@ -36,7 +37,7 @@ forgetObserverMessage = UnobserveCounter logCounterObservations- :: (SetMember Process (Process q) r, Member (Logs LogMessage) q)+ :: (SetMember Process (Process q) r, Member (Logs LogMessage) q, Lifted IO q) => SchedulerProxy q -> Eff r (Server (CallbackObserver Counter)) logCounterObservations px = spawnCallbackObserver@@ -53,26 +54,30 @@ , Member (State Integer) r , Member (Logs LogMessage) r , SetMember Process (Process q) r+ , Lifted IO r ) => SchedulerProxy q -> ApiHandler Counter r-counterHandler px = ApiHandler @Counter handleCast- handleCall- (defaultTermination px)+counterHandler px = castAndCallHandler handleCastCounter handleCallCounter where- handleCast :: Api Counter 'Asynchronous -> Eff r ()- handleCast (ObserveCounter o) = addObserver o- handleCast (UnobserveCounter o) = removeObserver o- handleCast Inc = do+ handleCastCounter :: Api Counter 'Asynchronous -> Eff r ApiServerCmd+ handleCastCounter (ObserveCounter o) =+ addObserver o >> return HandleNextRequest+ handleCastCounter (UnobserveCounter o) =+ removeObserver o >> return HandleNextRequest+ handleCastCounter Inc = do logInfo "Inc" modify (+ (1 :: Integer)) currentCount <- get notifyObservers px (CountChanged currentCount)- handleCall :: Api Counter ( 'Synchronous x) -> (x -> Eff r ()) -> Eff r ()- handleCall Cnt reply = do+ return HandleNextRequest+ handleCallCounter+ :: Api Counter ( 'Synchronous x) -> (x -> Eff r ()) -> Eff r ApiServerCmd+ handleCallCounter Cnt reply = do c <- get logInfo ("Cnt is " ++ show c) reply c+ return HandleNextRequest -- * Second API @@ -89,30 +94,35 @@ . ( Member (State Integer) r , Member (Logs LogMessage) r , SetMember Process (Process q) r+ , Lifted IO r ) => SchedulerProxy q -> ApiHandler PocketCalc r-pocketCalcHandler px = ApiHandler @PocketCalc handleCast- handleCall- (defaultTermination px)+pocketCalcHandler _ = castAndCallHandler handleCastCalc handleCallCalc where- handleCast :: Api PocketCalc 'Asynchronous -> Eff r ()- handleCast (AAdd x) = do+ handleCastCalc :: Api PocketCalc 'Asynchronous -> Eff r ApiServerCmd+ handleCastCalc (AAdd x) = do logInfo ("AsyncAdd " ++ show x) modify (+ x) c <- get @Integer logInfo ("Accumulator is " ++ show c)- handleCall :: Api PocketCalc ( 'Synchronous x) -> (x -> Eff r ()) -> Eff r ()- handleCall (Add x) reply = do+ return HandleNextRequest+ handleCallCalc+ :: Api PocketCalc ( 'Synchronous x) -> (x -> Eff r ()) -> Eff r ApiServerCmd+ handleCallCalc (Add x) reply = do logInfo ("Add " ++ show x) modify (+ x) c <- get logInfo ("Accumulator is " ++ show c) reply c+ return HandleNextRequest serverLoop :: forall r q- . (Member (Logs LogMessage) r, SetMember Process (Process q) r)+ . ( Member (Logs LogMessage) r+ , SetMember Process (Process q) r+ , Lifted IO r+ ) => SchedulerProxy q -> Eff r () serverLoop px = evalState @Integer@@ -123,9 +133,12 @@ -- ** Counter client counterExample- :: ( Member (Logs LogMessage) r+ :: ( SetMember Process (Process q) r , Member (Logs LogMessage) q- , SetMember Process (Process q) r+ , Member (Logs LogMessage) r+ , Lifted IO q+ , Lifted IO r+ , q <:: r ) => SchedulerProxy q -> Eff r ()@@ -163,4 +176,4 @@ cnt cntServer1 cast px cntServer2 Inc cnt cntServer2- void $ sendShutdown px pid2+ void $ sendShutdown px pid2 (ExitWithError "test test test")
extensible-effects-concurrent.cabal view
@@ -1,5 +1,5 @@ name: extensible-effects-concurrent-version: 0.8+version: 0.9.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@@ -33,10 +33,12 @@ hs-source-dirs: src build-depends:+ async >= 2.2 && <3, base >=4.9 && <5, data-default, deepseq, directory,+ enclosed-exceptions >= 1.0 && < 1.1, filepath, time, mtl,@@ -106,8 +108,10 @@ hs-source-dirs: examples/example-1 build-depends: base >=4.9 && <5+ , data-default , extensible-effects-concurrent , extensible-effects >= 3+ , lens ghc-options: -Wall -threaded -fno-full-laziness -rtsopts -with-rtsopts=-N default-language: Haskell2010 default-extensions: BangPatterns@@ -130,8 +134,10 @@ hs-source-dirs: examples/example-2 build-depends: base >=4.9 && <5+ , data-default , extensible-effects-concurrent , extensible-effects >= 3+ , lens ghc-options: -Wall -threaded -fno-full-laziness -rtsopts -with-rtsopts=-N default-language: Haskell2010 default-extensions: BangPatterns@@ -165,6 +171,7 @@ ghc-options: -Wall -threaded -fno-full-laziness -rtsopts -with-rtsopts=-N build-depends: base >=4.9 && <5+ , data-default , extensible-effects-concurrent , extensible-effects >= 3 , tasty
src/Control/Eff/Concurrent.hs view
@@ -42,6 +42,10 @@ , ConsProcess , ResumeProcess(..) , SchedulerProxy(..)+ , MessageSelector(..)+ , ProcessState(..)+ , ProcessExitReason(..)+ , ShutdownRequest(..) , thisSchedulerProxy , executeAndCatch , executeAndResume@@ -53,15 +57,20 @@ , spawn_ , receiveMessage , receiveMessageAs+ , receiveMessageSuchThat , receiveLoop+ , receiveLoopAs+ , receiveLoopSuchThat , self , sendShutdown , sendShutdownChecked+ , makeReference , exitWithError , exitNormally , raiseError , catchRaisedError , ignoreProcessError+ , logProcessExit ) import Control.Eff.Concurrent.Api ( Api , Synchronicity(..)@@ -85,7 +94,20 @@ import Control.Eff.Concurrent.Api.Server ( serve , spawnServer+ , spawnServerWithEffects , ApiHandler(..)+ , castCallback+ , callCallback+ , terminateCallback+ , apiHandler+ , apiHandlerForever+ , castHandler+ , castHandlerForever+ , callHandler+ , callHandlerForever+ , castAndCallHandler+ , castAndCallHandlerForever+ , ApiServerCmd(..) , unhandledCallError , unhandledCastError , defaultTermination@@ -125,10 +147,10 @@ ( schedule , defaultMain , defaultMainWithLogChannel- , SchedulerError(..)+ , ProcEff , SchedulerIO- , forkIoScheduler , HasSchedulerIO+ , forkIoScheduler ) import Control.Eff.Concurrent.Process.SingleThreadedScheduler
src/Control/Eff/Concurrent/Api/Client.hs view
@@ -23,6 +23,7 @@ import Control.Eff.Concurrent.Api import Control.Eff.Concurrent.Api.Internal import Control.Eff.Concurrent.Process+import Control.Monad ( (>=>) ) import Data.Dynamic import Data.Typeable ( Typeable , typeRep@@ -46,7 +47,7 @@ -> Api o 'Asynchronous -> Eff r Bool castChecked px (Server pid) castMsg =- withFrozenCallStack $ sendMessageChecked px pid (toDyn $! (Cast $! 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@@ -62,7 +63,7 @@ -> Server o -> Api o 'Asynchronous -> Eff r ()-cast px toServer apiRequest = withFrozenCallStack $ do+cast px toServer apiRequest = do _ <- castChecked px toServer apiRequest return () @@ -82,18 +83,29 @@ -> Server api -> Api api ( 'Synchronous result) -> Eff r result-call px (Server pidInt) req = withFrozenCallStack $ do+call px (Server pidInt) req = do fromPid <- self px- let requestMessage = Call fromPid $! req+ callRef <- makeReference px+ let requestMessage = Call callRef fromPid $! req wasSent <- sendMessageChecked px pidInt (toDyn $! requestMessage) if wasSent then- let extractResult :: Response api result -> result- extractResult (Response _pxResult result) = result- in extractResult <$> receiveMessageAs px+ 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 MessageSelector (fromDynamic >=> extractResult)+ in receiveMessageSuchThat px selectResult else raiseError px- ("Could not send request message " ++ show (typeRep requestMessage))+ ( "failed to send a call for: '"+ ++ show (typeRep requestMessage)+ ++ "' to: "+ ++ show pidInt+ ++ " with call-ref: "+ ++ show callRef+ ) -- | Instead of passing around a 'Server' value and passing to functions like@@ -114,7 +126,7 @@ -- 'Api' instance. registerServer :: HasCallStack => Server o -> Eff (ServerReader o ': r) a -> Eff r a-registerServer = withFrozenCallStack runReader+registerServer = runReader -- | Get the 'Server' registered with 'registerServer'. whereIsServer :: Member (ServerReader o) e => Eff e (Server o)@@ -127,7 +139,7 @@ => SchedulerProxy q -> Api o ( 'Synchronous reply) -> Eff r reply-callRegistered px method = withFrozenCallStack $ do+callRegistered px method = do serverPid <- whereIsServer call px serverPid method @@ -146,8 +158,8 @@ => SchedulerProxy q -> Api o ( 'Synchronous (f reply)) -> Eff r (f reply)-callRegisteredA px method = withFrozenCallStack- $ catchRaisedError px (const (return (empty @f))) (callRegistered px method)+callRegisteredA px method =+ catchRaisedError px (const (return (empty @f))) (callRegistered px method) -- | Like 'cast' but take the 'Server' from the reader provided by -- 'registerServer'.@@ -156,6 +168,6 @@ => SchedulerProxy q -> Api o 'Asynchronous -> Eff r ()-castRegistered px method = withFrozenCallStack $ do+castRegistered px method = do serverPid <- whereIsServer cast px serverPid method
src/Control/Eff/Concurrent/Api/Internal.hs view
@@ -1,24 +1,23 @@ -- | Internal request and response type for casts and calls module Control.Eff.Concurrent.Api.Internal- ( Request (..)- , Response (..)+ ( Request(..)+ , Response(..) ) where -import Data.Typeable (Typeable)+import Data.Typeable ( Typeable ) import Data.Proxy import Control.Eff.Concurrent.Api import Control.Eff.Concurrent.Process data Request api where Call :: forall api apiCallReplyType . (Typeable api, Typeable apiCallReplyType, Typeable (Api api ('Synchronous apiCallReplyType)))- => ProcessId -> Api api ('Synchronous apiCallReplyType) -> Request api+ => Int -> ProcessId -> Api api ('Synchronous apiCallReplyType) -> Request api Cast :: forall api . (Typeable api, Typeable (Api api 'Asynchronous)) => Api api 'Asynchronous -> Request api- Terminate :: Maybe String -> Request api deriving Typeable data Response api apiCallReplyType where- Response :: (Typeable api, Typeable apiCallReplyType) => Proxy api -> apiCallReplyType -> Response api apiCallReplyType+ Response :: (Typeable api, Typeable apiCallReplyType) => Proxy api -> Int -> apiCallReplyType -> Response api apiCallReplyType deriving Typeable
src/Control/Eff/Concurrent/Api/Observer.hs view
@@ -35,12 +35,10 @@ import Control.Eff.Concurrent.Api import Control.Eff.Concurrent.Api.Client import Control.Eff.Concurrent.Api.Server-import Control.Eff.Exception-import Control.Eff.Extend import Control.Eff.Log+import Control.Eff.Lift import Control.Eff.State.Strict import Control.Lens-import Control.Monad -- | An 'Api' index that support observation of the -- another 'Api' that is 'Observable'.@@ -68,8 +66,8 @@ -> Server o -> Observation o -> Eff r ()-notifyObserver px observer observed observation = withFrozenCallStack- $ cast px observer (observationMessage observed observation)+notifyObserver px observer observed observation =+ cast px observer (observationMessage observed observation) -- | Send the 'registerObserverMessage' registerObserver@@ -78,8 +76,8 @@ -> Server p -> Server o -> Eff r ()-registerObserver px observer observed = withFrozenCallStack- $ cast px observed (registerObserverMessage (SomeObserver observer))+registerObserver px observer observed =+ cast px observed (registerObserverMessage (SomeObserver observer)) -- | Send the 'forgetObserverMessage' forgetObserver@@ -88,8 +86,8 @@ -> Server p -> Server o -> Eff r ()-forgetObserver px observer observed = withFrozenCallStack- $ cast px observed (forgetObserverMessage (SomeObserver observer))+forgetObserver px observer observed =+ cast px observed (forgetObserverMessage (SomeObserver observer)) -- | An existential wrapper around a 'Server' of an 'Observer'. -- Needed to support different types of observers to observe the@@ -116,7 +114,7 @@ -> SomeObserver o -> Eff r () notifySomeObserver px observed observation (SomeObserver observer) =- withFrozenCallStack $ notifyObserver px observer observed observation+ notifyObserver px observer observed observation -- ** Manage 'Observers's @@ -193,22 +191,16 @@ => SchedulerProxy q -> (Server o -> Observation o -> Eff (Process q ': q) Bool) -> Eff r (Server (CallbackObserver o))-spawnCallbackObserver px onObserve = withFrozenCallStack $ spawnServer+spawnCallbackObserver px onObserve = spawnServerWithEffects px- (ApiHandler @(CallbackObserver o) handleCast- (unhandledCallError px)- (defaultTermination px)- )- (runError @ExitCallbackObserver >=> return . const ())+ (castHandler handleCastCbo)+ id where- handleCast (CbObserved fromSvr v) =- raise (onObserve fromSvr v) >>= flip when (throwError ExitCallbackObserver)--data ExitCallbackObserver = ExitCallbackObserver-+ handleCastCbo (CbObserved fromSvr v) = do+ continueObservation <- onObserve fromSvr v+ return+ (if continueObservation then HandleNextRequest else StopApiServer Nothing) --- | Use 'spawnCallbackObserver' to create a universal logging observer,--- using the 'Show' instance of the 'Observation'. -- | Start a new process for an 'Observer' that schedules -- all observations to an effectful callback. --@@ -220,10 +212,11 @@ , Show (Observation o) , Observable o , Member (Logs LogMessage) q+ , Lifted IO q , HasCallStack ) => SchedulerProxy q -> Eff r (Server (CallbackObserver o))-spawnLoggingObserver px = withFrozenCallStack $ spawnCallbackObserver+spawnLoggingObserver px = spawnCallbackObserver px (\s o -> logDebug (show s ++ " OBSERVED: " ++ show o) >> return True)
src/Control/Eff/Concurrent/Api/Observer/Queue.hs view
@@ -22,7 +22,9 @@ import Control.Eff.Reader.Strict import Control.Exception import Control.Monad.IO.Class+import Control.Monad ( unless ) import Data.Typeable+import Text.Printf import GHC.Stack -- | Contains a 'TBQueue' capturing observations received by 'enqueueObservationsRegistered'@@ -32,16 +34,25 @@ -- | A 'Reader' for an 'ObservationQueue'. type ObservationQueueReader a = Reader (ObservationQueue a) +logPrefix :: forall o proxy . (HasCallStack, Typeable o) => proxy o -> String+logPrefix px = "<observation queue " ++ show (typeRep px) ++ ">"+ -- | Read queued observations captured by observing a 'Server' that implements -- an 'Observable' 'Api' using 'enqueueObservationsRegistered' or 'enqueueObservations'. -- This blocks until the next 'Observation' received. For a non-blocking -- variant use 'tryReadObservationQueue' or 'flushObservationQueue'. readObservationQueue :: forall o r- . (Member (ObservationQueueReader o) r, HasCallStack, MonadIO (Eff r))+ . ( Member (ObservationQueueReader o) r+ , HasCallStack+ , MonadIO (Eff r)+ , Typeable o+ , Member (Logs LogMessage) r+ ) => Eff r (Observation o)-readObservationQueue = withFrozenCallStack $ do+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@@ -50,9 +61,15 @@ -- Use 'readObservationQueue' to block until an observation is observed. tryReadObservationQueue :: forall o r- . (Member (ObservationQueueReader o) r, HasCallStack, MonadIO (Eff r))+ . ( Member (ObservationQueueReader o) r+ , HasCallStack+ , MonadIO (Eff r)+ , Typeable o+ , Member (Logs LogMessage) r+ ) => Eff r (Maybe (Observation o))-tryReadObservationQueue = withFrozenCallStack $ do+tryReadObservationQueue = do+ logDebug ((logPrefix (Proxy @o)) ++ " try reading") ObservationQueue q <- ask @(ObservationQueue o) liftIO (atomically (tryReadTBQueue q)) @@ -61,9 +78,15 @@ -- variant use 'readObservationQueue'. flushObservationQueue :: forall o r- . (Member (ObservationQueueReader o) r, HasCallStack, MonadIO (Eff r))+ . ( Member (ObservationQueueReader o) r+ , HasCallStack+ , MonadIO (Eff r)+ , Typeable o+ , Member (Logs LogMessage) r+ ) => Eff r [Observation o]-flushObservationQueue = withFrozenCallStack $ do+flushObservationQueue = do+ logDebug ((logPrefix (Proxy @o)) ++ " flush") ObservationQueue q <- ask @(ObservationQueue o) liftIO (atomically (flushTBQueue q)) @@ -87,7 +110,7 @@ -> Int -> Eff (ObservationQueueReader o ': r) a -> Eff r a-enqueueObservationsRegistered px queueLimit k = withFrozenCallStack $ do+enqueueObservationsRegistered px queueLimit k = do oSvr <- whereIsServer @o enqueueObservations px oSvr queueLimit k @@ -115,35 +138,63 @@ -> Int -> Eff (ObservationQueueReader o ': r) a -> Eff r a-enqueueObservations px oSvr queueLimit k = withFrozenCallStack $ withQueue+enqueueObservations px oSvr queueLimit k = withQueue queueLimit (do ObservationQueue q <- ask @(ObservationQueue o) do+ logDebug+ (printf "%s starting with queue limit: %d"+ (logPrefix (Proxy @o))+ queueLimit+ ) cbo <- spawnCallbackObserver px- (\_from observeration -> do- liftIO (atomically (writeTBQueue q observeration))+ (\from observation -> do+ logDebug+ (printf "%s enqueue observation %s from %s"+ (logPrefix (Proxy @o))+ (show observation)+ (show from)+ )+ liftIO (atomically (writeTBQueue q observation)) return True )+ logDebug+ (printf "%s started observer process %s"+ (logPrefix (Proxy @o))+ (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- sendShutdown px (_fromServer cbo)+ logDebug (printf "%s unregistered" (logPrefix (Proxy @o)))+ sendShutdown px (_fromServer cbo) ExitNormally+ logDebug (printf "%s stopped observer process" (logPrefix (Proxy @o))) return res ) withQueue- :: ( HasCallStack+ :: forall a b e+ . ( HasCallStack , MonadIO (Eff e) , Member (Logs LogMessage) e+ , Typeable a+ , Show (Observation a) , SetMember Lift (Lift IO) 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)- _ <- liftIO (atomically (flushTBQueue q))+ q <- liftIO (newTBQueueIO queueLimit)+ res <- liftTry @SomeException (runReader (ObservationQueue q) e)+ rest <- liftIO (atomically (flushTBQueue q))+ unless+ (null rest)+ (logNotice (logPrefix (Proxy @a) ++ " unread observations: " ++ show rest)) either (\em -> logError (show em) >> liftIO (throwIO em)) return res
src/Control/Eff/Concurrent/Api/Server.hs view
@@ -4,8 +4,21 @@ -- * Api Server serve , spawnServer+ , spawnServerWithEffects -- * Api Callbacks , ApiHandler(..)+ , castCallback+ , callCallback+ , terminateCallback+ , apiHandler+ , apiHandlerForever+ , castHandler+ , castHandlerForever+ , callHandler+ , callHandlerForever+ , castAndCallHandler+ , castAndCallHandlerForever+ , ApiServerCmd(..) , unhandledCallError , unhandledCastError , defaultTermination@@ -30,7 +43,8 @@ import Control.Applicative import Data.Kind import GHC.Stack-+import Data.Maybe+import Data.Default -- | A record of callbacks, handling requests sent to a /server/ 'Process', all -- belonging to a specific 'Api' family instance.@@ -39,25 +53,132 @@ data ApiHandler api eff where ApiHandler :: { -- | A cast will not return a result directly. This is used for async- -- methods.- _handleCast- :: HasCallStack- => Api api 'Asynchronous -> Eff eff ()+ -- methods. This returns an 'ApiServerCmd' to the server loop.+ _castCallback+ :: Maybe (Api api 'Asynchronous -> Eff eff ApiServerCmd) -- | A call is a blocking operation, the caller is blocked until this -- handler calls the reply continuation.- , _handleCall- :: forall reply . HasCallStack- => Api api ('Synchronous reply) -> (reply -> Eff eff ()) -> Eff eff ()- -- | This callback is called with @Nothing@ if the process exits- -- peacefully, or @Just "error message..."@ if the process exits with an- -- error. This function is responsible to exit the process if necessary.+ -- This returns an 'ApiServerCmd' to the server loop.+ , _callCallback+ :: forall reply . Maybe (Api api ('Synchronous reply) -> (reply -> Eff eff ()) -> Eff eff ApiServerCmd)+ -- | This callback is called with @Nothing@ if one of these things happen:+ --+ -- * the process exits+ -- * '_callCallback' or '_castCallback' return 'StopApiServer'+ --+ -- If the process exist peacefully the parameter is 'Nothing',+ -- otherwise @Just "error message..."@ if the process exits with an+ -- error.+ -- -- The default behavior is defined in 'defaultTermination'.- , _handleTerminate- :: HasCallStack- => Maybe String -> Eff eff ()+ , _terminateCallback+ :: Maybe (Maybe String -> Eff eff ()) } -> ApiHandler api eff +instance Default (ApiHandler api eff) where+ def = ApiHandler { _castCallback = def+ , _callCallback = def+ , _terminateCallback = def+ }++-- | Create an 'ApiHandler' with a '_castCallback', a '_callCallback' and+-- a '_terminateCallback' implementation.+apiHandler+ :: (Api api 'Asynchronous -> Eff e ApiServerCmd)+ -> ( forall r+ . Api api ( 'Synchronous r)+ -> (r -> Eff e ())+ -> Eff e ApiServerCmd+ )+ -> (Maybe String -> Eff e ())+ -> ApiHandler api e+apiHandler c d e = ApiHandler+ { _castCallback = Just c+ , _callCallback = Just d+ , _terminateCallback = Just e+ }++-- | Like 'apiHandler' but the server will loop until an error is raised or+-- the process exits.+-- The callback actions won't decide wether to stop the+-- server or not, instead the 'ApiServerCmd' 'HandleNextRequest' is used.+apiHandlerForever+ :: (Api api 'Asynchronous -> Eff e ())+ -> (forall r . Api api ( 'Synchronous r) -> (r -> Eff e ()) -> Eff e ())+ -> (Maybe String -> Eff e ())+ -> ApiHandler api e+apiHandlerForever c d = apiHandler+ (\someCast -> c someCast >> return HandleNextRequest)+ (\someCall k -> d someCall k >> return HandleNextRequest)++-- | Create an 'ApiHandler' with only a '_castCallback' implementation.+castHandler+ :: (Api api 'Asynchronous -> Eff eff ApiServerCmd) -> ApiHandler api eff+castHandler c = def { _castCallback = Just c }++-- | Like 'castHandler' but the server will loop until an error is raised or+-- the process exits. See 'apiHandlerForver'.+castHandlerForever+ :: (Api api 'Asynchronous -> Eff eff ()) -> ApiHandler api eff+castHandlerForever c =+ castHandler (\someCast -> c someCast >> return HandleNextRequest)++-- | Create an 'ApiHandler' with only a '_callCallback' implementation.+callHandler+ :: ( forall r+ . Api api ( 'Synchronous r)+ -> (r -> Eff e ())+ -> Eff e ApiServerCmd+ )+ -> ApiHandler api e+callHandler c = def { _callCallback = Just c }++-- | Like 'callHandler' but the server will loop until an error is raised or+-- the process exits. See 'apiHandlerForver'.+callHandlerForever+ :: (forall r . Api api ( 'Synchronous r) -> (r -> Eff e ()) -> Eff e ())+ -> ApiHandler api e+callHandlerForever d =+ callHandler (\someCall k -> d someCall k >> return HandleNextRequest)++-- | Create an 'ApiHandler' with only a '_castCallback' and '_callCallback' implementation.+castAndCallHandler+ :: (Api api 'Asynchronous -> Eff e ApiServerCmd)+ -> ( forall r+ . Api api ( 'Synchronous r)+ -> (r -> Eff e ())+ -> Eff e ApiServerCmd+ )+ -> ApiHandler api e+castAndCallHandler c d = def { _castCallback = Just c, _callCallback = Just d }++-- | Like 'castAndCallHandler' but the server will loop until an error is raised or+-- the process exits. See 'apiHandlerForver'.+castAndCallHandlerForever+ :: (Api api 'Asynchronous -> Eff e ())+ -> (forall r . Api api ( 'Synchronous r) -> (r -> Eff e ()) -> Eff e ())+ -> ApiHandler api e+castAndCallHandlerForever c d = castAndCallHandler+ (\someCast -> c someCast >> return HandleNextRequest)+ (\someCall k -> d someCall k >> return HandleNextRequest)++-- | A command to the server loop started e.g. by 'server' or 'spawnServerWithEffects'.+-- Typically returned by an 'ApiHandler' member to indicate if the server+-- should continue or stop.+data ApiServerCmd where+ -- | Tell the server to keep the server loop running+ HandleNextRequest :: ApiServerCmd+ -- | 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+ -- SendReply :: reply -> ApiServerCmd () -> ApiServerCmd (reply -> Eff eff ())+ deriving (Show, Typeable)+++makeLenses ''ApiHandler+ -- | Building block for composition of 'ApiHandler'. -- A wrapper for 'ApiHandler'. Use this to combine 'ApiHandler', allowing a -- process to implement several 'Api' instances. The termination will be evenly@@ -73,7 +194,7 @@ -- @@@ -- data ServerCallback eff =- ServerCallback { _requestHandlerSelector :: MessageSelector (Eff eff ())+ ServerCallback { _requestHandlerSelector :: MessageSelector (Eff eff ApiServerCmd) , _terminationHandler :: Maybe String -> Eff eff () } @@ -144,15 +265,34 @@ -> Eff (ServerEff a) () serve px a = let serverCb = toServerCallback px a+ stopServer reason = do+ (serverCb ^. terminationHandler) reason+ return (Just ()) in receiveLoopSuchThat px (serverCb ^. requestHandlerSelector) $ \case- Left reason -> (serverCb ^. terminationHandler) reason- Right handleIt -> handleIt+ Left reason -> stopServer reason+ Right handleIt -> handleIt >>= \case+ HandleNextRequest -> return Nothing+ StopApiServer reason -> stopServer reason -- | Spawn a new process, that will receive and process incoming requests--- until the process exits. Also handle all internal effects.+-- until the process exits. spawnServer :: forall a effScheduler eff . ( Servable a+ , ServerEff a ~ (Process effScheduler ': effScheduler)+ , SetMember Process (Process effScheduler) eff+ , HasCallStack+ )+ => SchedulerProxy effScheduler+ -> a+ -> Eff eff (ServerPids a)+spawnServer px a = spawnServerWithEffects px a id++-- | Spawn a new process, that will receive and process incoming requests+-- until the process exits. Also handle all internal effects.+spawnServerWithEffects+ :: forall a effScheduler eff+ . ( Servable a , SetMember Process (Process effScheduler) (ServerEff a) , SetMember Process (Process effScheduler) eff , HasCallStack@@ -163,7 +303,7 @@ -> Eff (Process effScheduler ': effScheduler) () ) -> Eff eff (ServerPids a)-spawnServer px a handleEff = do+spawnServerWithEffects px a handleEff = do pid <- spawn (handleEff (serve px a)) return (toServerPids (Proxy @a) pid) @@ -179,7 +319,8 @@ -> ServerCallback eff apiHandlerServerCallback px handlers = mempty { _requestHandlerSelector = selectHandlerMethod px handlers- , _terminationHandler = _handleTerminate handlers+ , _terminationHandler = fromMaybe (const (return ()))+ (_terminateCallback handlers) } -- | Try to parse an incoming message to an API request, and apply either@@ -192,11 +333,11 @@ ) => SchedulerProxy effScheduler -> ApiHandler api eff- -> MessageSelector (Eff eff ())+ -> MessageSelector (Eff eff ApiServerCmd) selectHandlerMethod px handlers = MessageSelector (fmap (applyHandlerMethod px handlers) . fromDynamic) --- | Apply either the '_handleCall', '_handleCast' or the '_handleTerminate'+-- | Apply either the '_callCallback', '_castCallback' or the '_terminateCallback' -- callback to an incoming request. applyHandlerMethod :: forall eff effScheduler api@@ -207,54 +348,44 @@ => SchedulerProxy effScheduler -> ApiHandler api eff -> Request api- -> Eff eff ()-applyHandlerMethod _px handlers (Terminate e) = _handleTerminate handlers e-applyHandlerMethod _ handlers (Cast request) = _handleCast handlers request-applyHandlerMethod px handlers (Call fromPid request) = _handleCall handlers- request- sendReply+ -> Eff eff ApiServerCmd+applyHandlerMethod px handlers (Cast request) =+ fromMaybe (unhandledCastError px) (_castCallback handlers) request+applyHandlerMethod px handlers (Call callRef fromPid request) = fromMaybe+ (unhandledCallError px)+ (_callCallback handlers)+ request+ sendReply where sendReply :: Typeable reply => reply -> Eff eff () sendReply reply =- sendMessage px fromPid (toDyn $! (Response (Proxy @api) $! reply))+ sendMessage px fromPid (toDyn $! (Response (Proxy @api) callRef $! reply)) --- | A default handler to use in '_handleCall' in 'ApiHandler'. It will call+-- | A default handler to use in '_callCallback' in 'ApiHandler'. It will call -- 'raiseError' with a nice error message. unhandledCallError :: forall p x r q- . ( Show (Api p ( 'Synchronous x))- , Typeable p- , HasCallStack- , SetMember Process (Process q) r- )+ . (Typeable p, HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> Api p ( 'Synchronous x) -> (x -> Eff r ())- -> Eff r ()-unhandledCallError px api _ = withFrozenCallStack $ raiseError- px- ("Unhandled call: (" ++ show api ++ " :: " ++ show (typeRep (Proxy @p)) ++ ")"- )+ -> Eff r ApiServerCmd+unhandledCallError px _api _ =+ raiseError px ("unhandled call on api: " ++ show (typeRep (Proxy @p))) --- | A default handler to use in '_handleCast' in 'ApiHandler'. It will call+-- | A default handler to use in '_castCallback' in 'ApiHandler'. It will call -- 'raiseError' with a nice error message. unhandledCastError :: forall p r q- . ( Show (Api p 'Asynchronous)- , Typeable p- , HasCallStack- , SetMember Process (Process q) r- )+ . (Typeable p, HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> Api p 'Asynchronous- -> Eff r ()-unhandledCastError px api = withFrozenCallStack $ raiseError- px- ("Unhandled cast: (" ++ show api ++ " :: " ++ show (typeRep (Proxy @p)) ++ ")"- )+ -> Eff r ApiServerCmd+unhandledCastError px _api =+ raiseError px ("unhandled cast on api: " ++ show (typeRep (Proxy @p))) --- | Exit the process either normally of the error message is @Nothing@--- or with 'exitWithError' otherwise.+-- | 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)@@ -262,4 +393,4 @@ -> Maybe String -> Eff r () defaultTermination px =- withFrozenCallStack $ maybe (exitNormally px) (exitWithError px)+ maybe (return ()) (exitWithError px)
src/Control/Eff/Concurrent/Process.hs view
@@ -20,6 +20,9 @@ , ResumeProcess(..) , SchedulerProxy(..) , MessageSelector(..)+ , ProcessState(..)+ , ProcessExitReason(..)+ , ShutdownRequest(..) , thisSchedulerProxy , executeAndCatch , executeAndResume@@ -38,26 +41,32 @@ , self , sendShutdown , sendShutdownChecked+ , makeReference , exitWithError , exitNormally , raiseError , catchRaisedError , ignoreProcessError+ , logProcessExit ) where import GHC.Generics ( Generic , Generic1 )-import GHC.Stack 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 Control.Monad.IO.Class+import Data.Default import Data.Dynamic import Data.Kind-import Text.Printf+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@@ -93,15 +102,13 @@ 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 :: Process r a- -- | Exit the process due to an error, this cannot be caught.- ExitWithError :: String -> Process r b+ 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 -> Process r (ResumeProcess Bool)+ 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@@ -114,13 +121,42 @@ ReceiveMessage :: Process r (ResumeProcess Dynamic) -- | Wait for the next message that matches a criterium. Similar to 'ReceiveMessage'. ReceiveMessageSuchThat :: 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+ )+ ReceiveMessage -> showString "ReceiveMessage"+ ReceiveMessageSuchThat _ -> showString "ReceiveMessageSuchThat"+ 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 is required to exit.- ShutdownRequested :: ResumeProcess v+ -- | 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@@ -161,6 +197,66 @@ 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@@ -168,13 +264,13 @@ . (SetMember Process (Process q) r, HasCallStack) => Process q (ResumeProcess v) -> Eff r v-executeAndResume processAction = withFrozenCallStack $ do+executeAndResume processAction = do result <- send processAction case result of- ResumeWith !value -> return value- RetryLastAction -> executeAndResume processAction- ShutdownRequested -> send (Shutdown @q)- OnError e -> send (ExitWithError @q e)+ 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@@ -183,13 +279,13 @@ => SchedulerProxy q -> Eff r (ResumeProcess v) -> Eff r (Either String v)-executeAndCatch px processAction = withFrozenCallStack $ do+executeAndCatch px processAction = do result <- processAction case result of- ResumeWith !value -> return (Right value)- RetryLastAction -> executeAndCatch px processAction- ShutdownRequested -> send (Shutdown @q)- OnError e -> return (Left e)+ 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.@@ -198,7 +294,7 @@ . (HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> Eff r ()-yieldProcess _ = withFrozenCallStack $ executeAndResume YieldProcess+yieldProcess _ = executeAndResume YieldProcess -- | Send a message to a process addressed by the 'ProcessId'. -- See 'SendMessage'.@@ -209,8 +305,7 @@ -> ProcessId -> Dynamic -> Eff r ()-sendMessage px pid message =- withFrozenCallStack $ void (sendMessageChecked px pid message)+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.@@ -223,7 +318,7 @@ -> Dynamic -> Eff r Bool sendMessageChecked _ pid message =- withFrozenCallStack $ executeAndResume (SendMessage pid $! message)+ executeAndResume (SendMessage pid $! message) -- | Send a message to a process addressed by the 'ProcessId'. -- See 'SendMessage'.@@ -234,7 +329,7 @@ -> ProcessId -> o -> Eff r ()-sendMessageAs px pid = withFrozenCallStack (sendMessage px pid . toDyn)+sendMessageAs px pid = sendMessage px pid . toDyn -- | Exit a process addressed by the 'ProcessId'. -- See 'SendShutdown'.@@ -243,8 +338,9 @@ . (HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> ProcessId+ -> ShutdownRequest -> Eff r ()-sendShutdown px pid = withFrozenCallStack (void (sendShutdownChecked px pid))+sendShutdown px pid s = void (sendShutdownChecked px pid s) -- | Like 'sendShutdown', but also return @True@ iff the process to exit exists. sendShutdownChecked@@ -252,9 +348,9 @@ . (HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> ProcessId+ -> ShutdownRequest -> Eff r Bool-sendShutdownChecked _ pid =- withFrozenCallStack (executeAndResume (SendShutdown pid))+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'.@@ -263,7 +359,7 @@ . (HasCallStack, SetMember Process (Process q) r) => Eff (Process q ': q) () -> Eff r ProcessId-spawn child = withFrozenCallStack (executeAndResume (Spawn @q child))+spawn child = executeAndResume (Spawn @q child) -- | Like 'spawn' but return @()@. spawn_@@ -271,7 +367,7 @@ . (HasCallStack, SetMember Process (Process q) r) => Eff (Process q ': q) () -> Eff r ()-spawn_ = withFrozenCallStack (void . spawn)+spawn_ = void . spawn -- | Block until a message was received. -- See 'ReceiveMessage' for more documentation.@@ -280,7 +376,7 @@ . (HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> Eff r Dynamic-receiveMessage _ = withFrozenCallStack executeAndResume ReceiveMessage+receiveMessage _ = executeAndResume ReceiveMessage -- | Block until a message was received, that is not 'Nothing' after applying -- a callback to it.@@ -291,8 +387,7 @@ => SchedulerProxy q -> MessageSelector a -> Eff r a-receiveMessageSuchThat _ f =- withFrozenCallStack (executeAndResume (ReceiveMessageSuchThat f))+receiveMessageSuchThat _ f = executeAndResume (ReceiveMessageSuchThat f) -- | Receive and cast the message to some 'Typeable' instance. -- See 'ReceiveMessageSuchThat' for more documentation.@@ -302,78 +397,87 @@ . (HasCallStack, Typeable a, SetMember Process (Process q) r) => SchedulerProxy q -> Eff r a-receiveMessageAs px =- withFrozenCallStack (receiveMessageSuchThat px (MessageSelector fromDynamic))+receiveMessageAs px = receiveMessageSuchThat px (MessageSelector fromDynamic) -- | Enter a loop to receive messages and pass them to a callback, until the--- function returns 'False'.+-- function returns 'Just' a result. -- See 'receiveMesage' or 'ReceiveMessage' for more documentation. receiveLoop- :: forall r q+ :: forall r q endOfLoopResult . (SetMember Process (Process q) r, HasCallStack) => SchedulerProxy q- -> (Either (Maybe String) Dynamic -> Eff r ())- -> Eff r ()+ -> (Either (Maybe String) Dynamic -> Eff r (Maybe endOfLoopResult))+ -> Eff r endOfLoopResult receiveLoop px handlers = do mReq <- send (ReceiveMessage @q)- case mReq of- RetryLastAction -> receiveLoop px handlers- ShutdownRequested -> handlers (Left Nothing) >> receiveLoop px handlers- OnError reason -> handlers (Left (Just reason)) >> receiveLoop px handlers- ResumeWith message -> handlers (Right message) >> receiveLoop px handlers+ 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 in an endless loop.+-- | 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+ :: forall r q a endOfLoopResult . (SetMember Process (Process q) r, HasCallStack, Typeable a) => SchedulerProxy q- -> (Either (Maybe String) a -> Eff r ())- -> Eff r ()+ -> (Either (Maybe String) a -> Eff r (Maybe endOfLoopResult))+ -> Eff r endOfLoopResult receiveLoopAs px handlers = do mReq <- send (ReceiveMessageSuchThat @a @q (MessageSelector fromDynamic))- case mReq of- RetryLastAction -> receiveLoopAs px handlers- ShutdownRequested -> handlers (Left Nothing) >> receiveLoopAs px handlers- OnError reason ->- handlers (Left (Just reason)) >> receiveLoopAs px handlers- ResumeWith message -> handlers (Right message) >> receiveLoopAs px handlers+ 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+ :: forall r q a endOfLoopResult . (SetMember Process (Process q) r, HasCallStack) => SchedulerProxy q -> MessageSelector a- -> (Either (Maybe String) a -> Eff r ())- -> Eff r ()+ -> (Either (Maybe String) a -> Eff r (Maybe endOfLoopResult))+ -> Eff r endOfLoopResult receiveLoopSuchThat px selectMesage handlers = do mReq <- send (ReceiveMessageSuchThat @a @q selectMesage)- case mReq of- RetryLastAction -> receiveLoopSuchThat px selectMesage handlers- ShutdownRequested ->- handlers (Left Nothing) >> receiveLoopSuchThat px selectMesage handlers- OnError reason ->- handlers (Left (Just reason))- >> receiveLoopSuchThat px selectMesage handlers- ResumeWith message ->- handlers (Right message) >> receiveLoopSuchThat px selectMesage handlers+ 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 = withFrozenCallStack $ executeAndResume SelfPid+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 _ = withFrozenCallStack $ send (Shutdown @q)+exitNormally _ = send (Shutdown @q ExitNormally) -- | Exit the process with an error. exitWithError@@ -382,7 +486,7 @@ => SchedulerProxy q -> String -> Eff r a-exitWithError _ = withFrozenCallStack $ send . (ExitWithError @q $!)+exitWithError _ = send . (Shutdown @q . (ExitWithError $!)) -- | Thrown an error, can be caught by 'catchRaisedError'. raiseError@@ -391,7 +495,7 @@ => SchedulerProxy q -> String -> Eff r b-raiseError _ = withFrozenCallStack $ send . (RaiseError @q $!)+raiseError _ = send . (RaiseError @q $!) -- | Catch and handle an error raised by 'raiseError'. Works independent of the -- handler implementation.@@ -402,7 +506,7 @@ -> (String -> Eff r w) -> Eff r w -> Eff r w-catchRaisedError _ onErr = withFrozenCallStack $ interpose return go+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@@ -415,24 +519,36 @@ => SchedulerProxy q -> Eff r a -> Eff r (Either String a)-ignoreProcessError px =- withFrozenCallStack $ catchRaisedError px (return . Left) . fmap Right+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)+ deriving (Eq,Ord,Typeable,Bounded,Num, Enum, Integral, Real, NFData) instance Read ProcessId where- readsPrec _ ('<':'0':'.':rest1) =+ readsPrec _ ('!':rest1) = case reads rest1 of- [(c, '.':'0':'>':rest2)] -> [(ProcessId c, rest2)]+ [(c, rest2)] -> [(ProcessId c, rest2)] _ -> [] readsPrec _ _ = [] instance Show ProcessId where- show (ProcessId c) =- printf "<0.%d.0>" c+ showsPrec _ (ProcessId !c) = showChar '!' . shows c makeLenses ''ProcessId++-- | Log the 'ProcessExitReaons'+logProcessExit+ :: ('[Logs LogMessage] <:: e, MonadIO (Eff e), HasCallStack)+ => 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)+ _ -> logError ("scheduler error: " ++ show ex)
src/Control/Eff/Concurrent/Process/ForkIOScheduler.hs view
@@ -13,27 +13,25 @@ -- The 'Eff' handler for 'Process' and 'MessagePassing' use -- are implemented and available through 'spawn'. ----- 'spawn' uses 'forkFinally' and 'STM.TQueue's and tries to catch--- most exceptions. module Control.Eff.Concurrent.Process.ForkIOScheduler ( schedule , defaultMain , defaultMainWithLogChannel- , SchedulerError(..)+ , ProcEff , SchedulerIO- , forkIoScheduler , HasSchedulerIO+ , forkIoScheduler ) where -import Data.Foldable import GHC.Stack-import Data.Bifunctor-import Data.Maybe import Data.Kind ( ) import qualified Control.Exception as Exc-import Control.Concurrent as Concurrent+import qualified Control.Eff.ExceptionExtra as ExcExtra import Control.Concurrent.STM as STM+import Control.Concurrent (threadDelay, yield)+import qualified Control.Concurrent.Async as Async+import Control.Concurrent.Async (Async(..)) import Control.Eff import Control.Eff.Extend import Control.Eff.Concurrent.Process@@ -44,532 +42,434 @@ import Control.Lens import Control.Monad ( when , void- , join , (>=>) )-import qualified Control.Monad.State as Mtl+import Control.Monad.Trans.Control (MonadBaseControl(..))+import qualified Control.Exception.Enclosed as Exceptions import Data.Dynamic import Data.Map ( Map )-import qualified Data.Map as Map-import Text.Printf+import Data.Default+import Data.Sequence ( Seq(..) )+import qualified Data.Sequence as Seq+import Control.DeepSeq+import Control.Monad.IO.Class+import Text.Printf+import Data.Maybe ++-- * 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+ }++instance Default MessageQ where def = MessageQ def def++makeLenses ''MessageQ++-- | 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 mqVar = do+ mq <- readTVar mqVar+ when (isJust (mq^.shutdownRequests))+ (writeTVar mqVar (mq & shutdownRequests .~ Nothing))+ return (mq^.shutdownRequests)++ -- | Information about a process, needed to implement 'MessagePassing' and -- 'Process' handlers. The message queue is backed by a 'STM.TQueue' and contains -- 'MessageQEntry' values. data ProcessInfo = ProcessInfo { _processId :: ProcessId- , _messageQ :: STM.TQueue (Maybe Dynamic)- , _shutdownRequested :: STM.TVar Bool+ , _processState :: TVar ProcessState+ , _messageQ :: TVar MessageQ } makeLenses ''ProcessInfo -instance Show ProcessInfo where- show p = "ProcessInfo: " ++ show (p ^. processId)+-- * Scheduler Types -- | Contains all process info'elements, as well as the state needed to -- implement inter process communication. It contains also a 'LogChannel' to -- which the logs of all processes are forwarded to. data SchedulerState =- SchedulerState { _nextPid :: ProcessId- , _processTable :: Map ProcessId ProcessInfo- , _threadIdTable :: Map ProcessId ThreadId- , _schedulerShuttingDown :: Bool- , _logChannel :: LogChannel LogMessage+ SchedulerState { _nextPid :: TVar ProcessId+ , _processTable :: TVar (Map ProcessId ProcessInfo)+ , _processCancellationTable :: TVar (Map ProcessId (Async ProcessExitReason)) } makeLenses ''SchedulerState --- | A newtype wrapper around an 'STM.TVar' holding the scheduler state.--- This is needed by 'spawn' and provided by 'runScheduler'.-newtype SchedulerVar = SchedulerVar { fromSchedulerVar :: STM.TVar SchedulerState }- deriving Typeable --- | A sum-type with errors that can occur when scheduleing messages.-data SchedulerError =- ProcessNotFound ProcessId- -- ^ No process info was found for a 'ProcessId' during internal- -- processing. NOTE: This is **ONLY** caused by internal errors, probably by- -- an incorrect 'MessagePassing' handler in this module. **Sending a message- -- to a process ALWAYS succeeds!** Even if the process does not exist.- | ProcessRaisedError String- -- ^ A process called 'raiseError'.- | ProcessExitError String- -- ^ A process called 'exitWithError'.- | ProcessShuttingDown- -- ^ A process exits.- | SchedulerShuttingDown- -- ^ An action was not performed while the scheduler was exiting.- deriving (Typeable, Show)+-- * Process Implementation -instance Exc.Exception SchedulerError+instance Show ProcessInfo where+ show p = "process info: " ++ show (p ^. processId) --- | An alias for the constraints for the effects essential to this scheduler--- implementation, i.e. these effects allow 'spawn'ing new 'Process'es.--- See SchedulerIO-type HasSchedulerIO r = ( HasCallStack- , SetMember Lift (Lift IO) r- , Member (Logs LogMessage) r- , Member (Reader SchedulerVar) r) --- | The concrete list of 'Eff'ects for this scheduler implementation.--- See HasSchedulerIO-type SchedulerIO =- '[ Reader SchedulerVar- , Logs LogMessage- , Lift IO- ]+-- | Create a new 'ProcessInfo'+newProcessInfo :: HasCallStack => ProcessId -> STM ProcessInfo+newProcessInfo a = ProcessInfo a <$> newTVar ProcessBooting <*> newTVar def --- | A 'SchedulerProxy' for 'SchedulerIO'-forkIoScheduler :: SchedulerProxy SchedulerIO-forkIoScheduler = SchedulerProxy+-- * Scheduler Implementation --- | This is the main entry point to running a message passing concurrency--- application. This function takes a 'Process' on top of the 'SchedulerIO'--- effect and a 'LogChannel' for concurrent logging.-schedule- :: HasCallStack- => Eff (ConsProcess SchedulerIO) ()- -> LogChannel LogMessage- -> IO ()-schedule e logC =- withFrozenCallStack $ void $ withNewSchedulerState $ \schedulerStateVar -> do- pidVar <- newEmptyTMVarIO- scheduleProcessWithShutdownAction schedulerStateVar pidVar- $ interceptLogging (setLogMessageThreadId >=> logMsg)- $ do- logNotice "++++++++ main process started ++++++++"- e- logNotice "++++++++ main process returned ++++++++"+-- | Create a new 'SchedulerState'+newSchedulerState :: HasCallStack => STM SchedulerState+newSchedulerState = SchedulerState <$> newTVar 1 <*> 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.+withNewSchedulerState+ :: HasCallStack => 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))) where- withNewSchedulerState :: HasCallStack => (SchedulerVar -> IO a) -> IO a- withNewSchedulerState mainProcessAction = do- myTId <- myThreadId- Exc.bracket- (newTVarIO (SchedulerState myPid Map.empty Map.empty False logC))- (tearDownScheduler myTId)- (mainProcessAction . SchedulerVar)- where- myPid = 1- tearDownScheduler myTId v = do- logChannelPutIO logC =<< debugMessageIO "begin scheduler tear down"- sch <-- (atomically- (do- sch <- readTVar v- let sch' = sch & schedulerShuttingDown .~ True- writeTVar v sch'- return sch'- )- )- logChannelPutIO logC =<< debugMessageIO- ( "killing "- ++ let ts = (sch ^.. threadIdTable . traversed)- in if length ts > 100- then show (length ts) ++ " threads"- else show ts- )+ tearDownScheduler = do+ schedulerState <- getSchedulerState+ let cancelTableVar = schedulerState ^. processCancellationTable+ -- cancel all processes+ allProcesses <- lift (atomically (readTVar cancelTableVar <* writeTVar cancelTableVar def))+ logNotice ("cancelling processes: " ++ show (toListOf (ifolded.asIndex) allProcesses))+ void+ (liftBaseWith+ (\runS -> Async.race+ (Async.mapConcurrently+ (\a -> do+ Async.cancel a+ runS (logNotice ("process cancelled: "++ show (asyncThreadId a)))+ )+ allProcesses)+ (threadDelay 5000000))) - imapM_ (killProcThread myTId) (sch ^. threadIdTable)- Concurrent.yield- atomically- (do- scheduler <- readTVar v- let allThreadsDead =- scheduler- ^. threadIdTable- . to Map.null- && scheduler- ^. processTable- . to Map.null- STM.check allThreadsDead- )- logChannelPutIO logC =<< infoMessageIO "all threads dead"+-- | The concrete list of 'Eff'ects of processes compatible with this scheduler.+-- This builds upon 'SchedulerIO'.+type ProcEff = ConsProcess SchedulerIO - killProcThread myTId _pid tid = when (myTId /= tid) (killThread tid) +-- | Type class constraint to indicate that an effect union contains the+-- effects required by every process and the scheduler implementation itself.+type HasSchedulerIO r = ( HasCallStack+ , Lifted IO r+ , SchedulerIO <:: r+ )++-- | The concrete list of 'Eff'ects for this scheduler implementation.+type SchedulerIO = ( Reader SchedulerState : LoggingAndIO)++-- | Basic effects: 'Logs' 'LogMessage' and 'Lift' IO+type LoggingAndIO =+ '[ Logs LogMessage+ , Lift IO+ ]+ -- | 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 (ConsProcess SchedulerIO) () -> IO ()-defaultMain c = withFrozenCallStack $ runLoggingT+defaultMain :: HasCallStack => Eff ProcEff () -> IO ()+defaultMain c = runLoggingT (logChannelBracket 128- (Just (infoMessage "main process started"))- (schedule c)+ Nothing+ (handleLoggingAndIO_ (schedule c)) ) (printLogMessage :: LogMessage -> IO ()) -- | Start the message passing concurrency system then execute a 'Process' on -- top of 'SchedulerIO' effect. All logging is sent to standard output. defaultMainWithLogChannel- :: HasCallStack- => LogChannel LogMessage- -> Eff (ConsProcess SchedulerIO) ()- -> IO ()+ :: HasCallStack => LogChannel LogMessage -> Eff ProcEff () -> IO () defaultMainWithLogChannel logC c =- withFrozenCallStack $ closeLogChannelAfter logC (schedule c logC)+ closeLogChannelAfter logC (handleLoggingAndIO_ (schedule c) logC) -scheduleProcessWithShutdownAction- :: HasCallStack- => SchedulerVar- -> STM.TMVar ProcessId- -> Eff (ConsProcess SchedulerIO) ()- -> IO (Either Exc.SomeException ())-scheduleProcessWithShutdownAction schedulerVar pidVar procAction = do- cleanupVar <- newEmptyTMVarIO- logC <- getLogChannelIO schedulerVar- eeres <- Exc.try (runProcEffects cleanupVar logC)- let eres = join eeres- getAndExecCleanup cleanupVar eres logC- logChannelPutIO logC =<< debugMessageIO "process cleanup finished"- return eres- where+-- | A 'SchedulerProxy' for 'SchedulerIO'+forkIoScheduler :: SchedulerProxy SchedulerIO+forkIoScheduler = SchedulerProxy - runProcEffects cleanupVar l =- Data.Bifunctor.first Exc.SomeException <$> runLift- (logToChannel- l- (runReader- schedulerVar- (scheduleProcessWithCleanup shutdownAction saveCleanupAndSchedule)- )- )- where- shutdownAction = ShutdownAction- (\eres -> do- let ex = either id (const ProcessShuttingDown) eres- Exc.throw ex- )- saveCleanupAndSchedule cleanUpAction pid = do- lift- (atomically- (do- STM.putTMVar cleanupVar cleanUpAction- STM.putTMVar pidVar pid- )- )- interceptLogging- (setLogMessageThreadId >=> logMsg . over- lmMessage- (printf "% 9s %s" (show pid))- )- $ do- logDebug "begin process"- procAction- getAndExecCleanup cleanupVar eres lc = do- mcleanup <- atomically (STM.tryTakeTMVar cleanupVar)- traverse_ execCleanup mcleanup- where- execCleanup ca = do- runCleanUpAction ca- logChannelPutIO lc =<< case eres of- Left se -> case Exc.fromException se of- Nothing -> errorMessageIO- ("process caught exception: " ++ Exc.displayException se)- Just schedulerErr- -> let exceptionMsg =- " - full exception message: " ++ Exc.displayException se- in- case schedulerErr of- ProcessShuttingDown -> debugMessageIO "process shutdown"- ProcessExitError m -> errorMessageIO- ("process exited with error: " ++ show m ++ exceptionMsg)- ProcessRaisedError m -> errorMessageIO- ("process raised error: " ++ show m ++ exceptionMsg)- _ ->- errorMessageIO- ("scheduler error: " ++ show schedulerErr ++ exceptionMsg- )- Right _ -> debugMessageIO "process function returned"+-- | 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 -getLogChannel :: HasSchedulerIO r => Eff r (LogChannel LogMessage)-getLogChannel = do- s <- getSchedulerVar- lift (getLogChannelIO s)+handleProcess+ :: HasCallStack+ => ProcessInfo+ -> Eff ProcEff ProcessExitReason+ -> Eff SchedulerIO ProcessExitReason+handleProcess myProcessInfo =+ 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+ setMyProcessStateSTM = writeTVar myProcessStateVar+ myMessageQVar = myProcessInfo ^. messageQ -getLogChannelIO :: SchedulerVar -> IO (LogChannel LogMessage)-getLogChannelIO s = view logChannel <$> readTVarIO (fromSchedulerVar s)+ kontinueWith+ :: forall s v a+ . HasCallStack+ => (s -> Arr SchedulerIO v a)+ -> s+ -> Arr SchedulerIO v a+ kontinueWith kontinue !nextRef !result = do+ setMyProcessState ProcessIdle+ lift yield+ kontinue nextRef result -overProcessInfo- :: HasSchedulerIO r- => ProcessId- -> Mtl.StateT ProcessInfo STM.STM a- -> Eff r (Either SchedulerError a)-overProcessInfo pid stAction = overScheduler- (do- res <- use (processTable . at pid)- case res of- Nothing -> return (Left (ProcessNotFound pid))- Just pinfo -> do- (x, pinfoOut) <- Mtl.lift (Mtl.runStateT stAction pinfo)- processTable . at pid . _Just .= pinfoOut- return (Right x)- )+ diskontinueWith+ :: forall a+ . HasCallStack+ => Arr SchedulerIO ProcessExitReason a+ -> Arr SchedulerIO ProcessExitReason a+ diskontinueWith diskontinue !reason = do+ setMyProcessState ProcessShuttingDown+ diskontinue reason --- ** MessagePassing execution+ stepProcessInterpreter+ :: forall v a+ . HasCallStack+ => Int+ -> Process SchedulerIO v+ -> (Int -> Arr SchedulerIO v a)+ -> Arr SchedulerIO ProcessExitReason a+ -> Eff SchedulerIO a+ stepProcessInterpreter !nextRef !request kontinue diskontinue = -spawnImpl- :: HasCallStack- => Eff (ConsProcess SchedulerIO) ()- -> Eff SchedulerIO ProcessId-spawnImpl mfa = do- schedulerVar <- ask- pidVar <- lift STM.newEmptyTMVarIO- void $ lift $ Concurrent.forkIO $ void $ scheduleProcessWithShutdownAction- schedulerVar- pidVar- mfa- lift Concurrent.yield -- this is important, removing this causes test failures- lift (atomically (STM.readTMVar pidVar))+ -- handle process shutdown requests:+ -- 1. take process exit reason+ -- 2. set process state to ProcessShuttingDown+ -- 3. apply kontinue to (Right ShutdownRequested)+ --+ tryTakeNextShutdownRequest+ >>= maybe noShutdownRequested onShutdownRequested+ where+ tryTakeNextShutdownRequest =+ lift (atomically (tryTakeNextShutdownRequestSTM myMessageQVar)) -newtype CleanUpAction = CleanUpAction { runCleanUpAction :: IO () }+ onShutdownRequested shutdownRequest = do+ logDebug "shutdownRequested"+ setMyProcessState ProcessShuttingDown+ interpretRequestAfterShutdownRequest+ (kontinueWith kontinue nextRef)+ (diskontinueWith diskontinue)+ shutdownRequest+ request -scheduleProcessWithCleanup- :: HasCallStack- => ShutdownAction- -> (CleanUpAction -> ProcessId -> Eff (ConsProcess SchedulerIO) ())- -> Eff SchedulerIO (Either SchedulerError ())-scheduleProcessWithCleanup shutdownAction processAction = withMessageQueue- (\cleanUpAction pinfo -> handle_relay- return- (go (pinfo ^. processId))- (processAction cleanUpAction (pinfo ^. processId))- )- where- shutdownOrGo+ noShutdownRequested = do+ setMyProcessState ProcessBusy+ interpretRequest+ (kontinueWith kontinue)+ (diskontinueWith diskontinue)+ nextRef+ request++ -- This gets no nextRef and may not pass a Left value to the continuation.+ -- This forces the caller to defer the process exit to the next request+ -- and hence ensures that the scheduler code cannot forget to allow the+ -- client code to react to a shutdown request.+ interpretRequestAfterShutdownRequest :: forall v a . HasCallStack- => ProcessId- -> (ResumeProcess v -> Eff SchedulerIO a)- -> Eff SchedulerIO a+ => Arr SchedulerIO v a+ -> Arr SchedulerIO ProcessExitReason a+ -> ShutdownRequest+ -> Process SchedulerIO v -> Eff SchedulerIO a- shutdownOrGo pid !k !ok = do- psVar <- getSchedulerTVar- eHasShutdowReq <- lift- (do- p <- readTVarIO psVar- let mPinfo = p ^. processTable . at pid- case mPinfo of- Just !pinfo -> atomically- (do- wasRequested <- readTVar (pinfo ^. shutdownRequested)- -- reset the shutdwown request flag, in case the shutdown- -- is prevented by the process- writeTVar (pinfo ^. shutdownRequested) False- return (Right wasRequested)- )- Nothing -> return (Left (ProcessNotFound pid))- )- case eHasShutdowReq of- Right True -> k ShutdownRequested- Right False -> ok- Left e -> k (OnError (show e))+ interpretRequestAfterShutdownRequest kontinue diskontinue shutdownRequest =+ \case+ SendMessage _ _ -> kontinue (ShutdownRequested shutdownRequest)+ SendShutdown toPid r ->+ if toPid == myPid+ then diskontinue (ProcessShutDown r)+ else kontinue (ShutdownRequested shutdownRequest)+ Spawn _ -> kontinue (ShutdownRequested shutdownRequest)+ ReceiveMessage -> kontinue (ShutdownRequested shutdownRequest)+ ReceiveMessageSuchThat _ -> 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) - go+ interpretRequest :: forall v a . HasCallStack- => ProcessId+ => (Int -> Arr SchedulerIO v a)+ -> Arr SchedulerIO ProcessExitReason a+ -> Int -> Process SchedulerIO v- -> (v -> Eff SchedulerIO a) -> Eff SchedulerIO a- go pid (SendMessage toPid reqIn) k = shutdownOrGo pid k $ do- eres <- do- psVar <- getSchedulerTVar- lift- ( Right- <$> (do- p <- readTVarIO psVar- let mto = p ^. processTable . at toPid- case mto of- Just toProc -> do- atomically (writeTQueue (toProc ^. messageQ) (Just reqIn))- return True- Nothing -> return False- )- )- lift Concurrent.yield- let kArg = either (OnError . show @SchedulerError) ResumeWith eres- k kArg-- go pid (SendShutdown toPid) k = shutdownOrGo pid k $ do- eres <- do- psVar <- getSchedulerTVar- lift- ( Right- <$> (do- p <- readTVarIO psVar- let mto = p ^. processTable . at toPid- case mto of- Just toProc -> atomically $ do- writeTVar (toProc ^. shutdownRequested) True- writeTQueue (toProc ^. messageQ) Nothing -- this wakes up a receiver- return True- Nothing -> return False- )- )- let kArg = either (OnError . show @SchedulerError) resume eres- resume = if toPid == pid then const ShutdownRequested else ResumeWith- lift Concurrent.yield- k kArg-- go pid (Spawn child) k = shutdownOrGo pid k $ do- res <- spawnImpl child- k (ResumeWith res)-- go pid ReceiveMessage k = shutdownOrGo pid k $ do- emq <- overProcessInfo pid (use messageQ)- case emq of- Left e -> k (OnError (show @SchedulerError e))- Right mq -> do- emdynMsg <- lift (Right <$> atomically (readTQueue mq))- k- (either (OnError . show @SchedulerError)- (maybe RetryLastAction ResumeWith)- emdynMsg- )+ interpretRequest kontinue diskontinue nextRef =+ \case+ SendMessage toPid msg -> interpretSend toPid 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+ ReceiveMessage -> interpretReceive (MessageSelector Just) >>= kontinue nextRef+ ReceiveMessageSuchThat f -> interpretReceive f >>= 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)+ where - go pid (ReceiveMessageSuchThat selectMsg) k = shutdownOrGo pid k $ do- emq <- overProcessInfo pid (use messageQ)- case emq of- Left e -> k (OnError (show @SchedulerError e))- Right mq -> do- let readTQueueUntilMatches =- let enqueueAgain = traverse_ (unGetTQueue mq)- untilMessageMatches unmatched = do- msg <- readTQueue mq- maybe- (do- -- msg is 'Nothing', this means the receive call- -- must be interrupted (e.g. because the process was killed, etc)- enqueueAgain unmatched- return Nothing- )- ( maybe- (untilMessageMatches (msg : unmatched))- (\result -> do- enqueueAgain unmatched- return (Just result)- )- . runMessageSelector selectMsg- )- msg- in untilMessageMatches []+ interpretSend !toPid !msg =+ setMyProcessState ProcessBusySending *>+ getSchedulerState >>= lift . atomically . enqueueMessageOtherProcess toPid msg - emdynMsg <- lift (Right <$> atomically readTQueueUntilMatches)- k- (either (OnError . show @SchedulerError)- (maybe RetryLastAction ResumeWith)- emdynMsg- )+ interpretSendShutdown !toPid !msg =+ setMyProcessState ProcessBusySendingShutdown *>+ getSchedulerState >>= lift . atomically . enqueueShutdownRequest toPid msg - go pid SelfPid k = shutdownOrGo pid k $ do- lift Concurrent.yield- k (ResumeWith pid)+ interpretReceive :: MessageSelector b -> Eff SchedulerIO (ResumeProcess b)+ interpretReceive f = do+ setMyProcessState ProcessBusyReceiving+ lift $ atomically $ do+ myMessageQ <- readTVar myMessageQVar+ case myMessageQ ^. shutdownRequests of+ Nothing ->+ case partitionMessages (myMessageQ ^. incomingMessages) Seq.Empty of+ Nothing -> retry+ Just (selectedMessage, otherMessages) -> do+ modifyTVar' myMessageQVar (incomingMessages .~ otherMessages)+ return (ResumeWith selectedMessage)+ Just shutdownRequest -> do+ modifyTVar' myMessageQVar (shutdownRequests .~ Nothing)+ return (ShutdownRequested shutdownRequest) - go pid YieldProcess !k = shutdownOrGo pid k $ do- lift Concurrent.yield- k (ResumeWith ())+ where+ partitionMessages Seq.Empty _acc = Nothing+ partitionMessages (m :<| msgRest) acc = maybe+ (partitionMessages msgRest (acc :|> m))+ (\res -> Just (res, acc Seq.>< msgRest))+ (runMessageSelector f m) - go _pid Shutdown _k = invokeShutdownAction shutdownAction (Right ())+-- | This is the main entry point to running a message passing concurrency+-- application. This function takes a 'Process' on top of the 'SchedulerIO'+-- effect and a 'LogChannel' for concurrent logging.+schedule :: HasCallStack => Eff ProcEff () -> Eff LoggingAndIO ()+schedule procEff =+ liftBaseWith (\runS ->+ Async.withAsync (runS $ withNewSchedulerState $ do+ (_, mainProcAsync) <- spawnNewProcess $ do+ logNotice "++++++++ main process started ++++++++"+ procEff+ logNotice "++++++++ main process returned ++++++++"+ lift (void (Async.wait mainProcAsync))+ )+ (\ast -> runS $ do+ a <- restoreM ast+ void $ lift (Async.wait a)+ )+ ) >>= restoreM - go _pid (ExitWithError msg) _k =- invokeShutdownAction shutdownAction (Left (ProcessExitError msg))+spawnNewProcess :: Eff ProcEff () -> Eff SchedulerIO (ProcessId, Async ProcessExitReason)+spawnNewProcess mfa = do+ schedulerState <- getSchedulerState+ liftBaseWith+ (\inScheduler -> do - go _pid (RaiseError msg) _k =- invokeShutdownAction shutdownAction (Left (ProcessExitError msg))+ let nextPidVar = schedulerState ^. nextPid+ processInfosVar = schedulerState ^. processTable+ cancellationsVar = schedulerState ^. processCancellationTable -data ShutdownAction =- ShutdownAction (forall a . Either SchedulerError () -> IO a)+ procInfo <- atomically (do+ pid <- readTVar nextPidVar+ modifyTVar' nextPidVar (+1)+ procInfo <- newProcessInfo pid+ modifyTVar' processInfosVar (at pid ?~ procInfo)+ return procInfo) -invokeShutdownAction- :: (HasCallStack, SetMember Lift (Lift IO) r, Member (Logs LogMessage) r)- => ShutdownAction- -> Either SchedulerError ()- -> Eff r a-invokeShutdownAction (ShutdownAction a) res = lift (a res)+ let pid = procInfo ^. processId+ logAppendProcInfo =+ interceptLogging+ (lift . setLogMessageThreadId >=> logMsg . addProcessId)+ where+ addProcessId =+ over lmProcessId (maybe (Just (printf "% 9s" (show pid))) Just) -withMessageQueue- :: HasSchedulerIO r- => (CleanUpAction -> ProcessInfo -> Eff r a)- -> Eff r (Either SchedulerError a)-withMessageQueue m = do- mpinfo <- createQueue- lc <- getLogChannel- case mpinfo of- Right pinfo -> do- cleanUpAction <-- getSchedulerTVar >>= return . CleanUpAction . destroyQueue- lc- (pinfo ^. processId)- Right <$> m cleanUpAction pinfo- Left e -> return $ Left e- where- createQueue = do- myTId <- lift myThreadId- overScheduler- (do- abortNow <- use schedulerShuttingDown- if abortNow- then return (Left SchedulerShuttingDown)- else do- pid <- nextPid <<+= 1- channel <- Mtl.lift newTQueue- shutdownIndicator <- Mtl.lift (newTVar False)- let pinfo = ProcessInfo pid channel shutdownIndicator- threadIdTable . at pid .= Just myTId- processTable . at pid .= Just pinfo- return (Right pinfo)- )- destroyQueue lc pid psVar = do- didWork <- Exc.try- (overSchedulerIO- psVar- (do- os <- processTable . at pid <<.= Nothing- ot <- threadIdTable . at pid <<.= Nothing- return (os, isJust os || isJust ot)+ 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)+ ) )- )- let getCause =- Exc.try @Exc.SomeException- (overSchedulerIO psVar (preuse (processTable . at pid)))- >>= either- (return . (show pid ++) . show)- (return . (maybe (show pid) show))-- case didWork of- Right (_pinfo, True ) -> return ()- Right (pinfo , False) -> logChannelPutIO lc- =<< errorMessageIO ("queue already destroyed: " ++ show pinfo)- Left (e :: Exc.SomeException) ->- getCause- >>= ( ( errorMessageIO- . (("failed to destroy queue: " ++ show e ++ " ") ++)- )- >=> logChannelPutIO lc- )-+ atomically (modifyTVar' cancellationsVar (at pid ?~ procAsync))+ return (pid, procAsync))+ >>= restoreM -overScheduler- :: HasSchedulerIO r- => Mtl.StateT SchedulerState STM.STM (Either SchedulerError a)- -> Eff r (Either SchedulerError a)-overScheduler stAction = do- psVar <- getSchedulerTVar- lift (overSchedulerIO psVar stAction)+-- Scheduler Accessor -overSchedulerIO- :: STM.TVar SchedulerState -> Mtl.StateT SchedulerState STM.STM a -> IO a-overSchedulerIO psVar stAction = STM.atomically- (do- ps <- STM.readTVar psVar- (result, psModified) <- Mtl.runStateT stAction ps- STM.writeTVar psVar psModified- return result- )+getSchedulerState :: HasSchedulerIO r => Eff r SchedulerState+getSchedulerState = ask -getSchedulerTVar :: HasSchedulerIO r => Eff r (TVar SchedulerState)-getSchedulerTVar = fromSchedulerVar <$> ask+enqueueMessageOtherProcess ::+ HasCallStack => ProcessId -> Dynamic -> SchedulerState -> STM Bool+enqueueMessageOtherProcess toPid msg schedulerState =+ view (at toPid) <$> readTVar (schedulerState ^. processTable)+ >>= maybe (return False)+ (\toProcessTable -> do+ modifyTVar' (toProcessTable ^. messageQ ) (incomingMessages %~ (:|> msg))+ return True) -getSchedulerVar :: HasSchedulerIO r => Eff r SchedulerVar-getSchedulerVar = ask+enqueueShutdownRequest ::+ HasCallStack => ProcessId -> ShutdownRequest -> SchedulerState -> STM Bool+enqueueShutdownRequest toPid msg schedulerState =+ view (at toPid) <$> readTVar (schedulerState ^. processTable)+ >>= maybe (return False)+ (\toProcessTable -> do+ modifyTVar' (toProcessTable ^. messageQ ) (shutdownRequests ?~ msg)+ return True)
src/Control/Eff/Concurrent/Process/SingleThreadedScheduler.hs view
@@ -68,11 +68,11 @@ -- -- @since 0.4.0.0 scheduleIOWithLogging- :: (NFData l, MonadIO m)- => (l -> m ())- -> Eff (ConsProcess '[Logs l, Lift m]) a- -> m (Either String a)-scheduleIOWithLogging handleLog = scheduleIO (handleLogsLifted handleLog)+ :: (NFData l)+ => (l -> IO ())+ -> Eff (ConsProcess '[Logs l, Lift IO]) a+ -> IO (Either String a)+scheduleIOWithLogging handleLog = scheduleIO (handleLogsWith handleLog) -- | Handle the 'Process' effect, as well as all lower effects using an effect handler function. --@@ -103,6 +103,7 @@ handleProcess runEff yieldEff 1+ 0 (Map.singleton 0 Seq.empty) (Seq.singleton (y, 0)) @@ -117,15 +118,15 @@ -> (ResumeProcess ProcessId -> Eff r (OnYield r a)) -> OnYield r a OnDone :: !a -> OnYield r a- OnShutdown :: OnYield r a- OnExitError :: !String -> OnYield r a+ OnShutdown :: ShutdownRequest -> OnYield r a OnRaiseError :: !String -> OnYield r a OnSend :: !ProcessId -> !Dynamic -> (ResumeProcess Bool -> Eff r (OnYield r a)) -> OnYield r a OnRecv :: MessageSelector b -> (ResumeProcess b -> Eff r (OnYield r a)) -> OnYield r a- OnSendShutdown :: !ProcessId -> (ResumeProcess Bool -> Eff r (OnYield r a)) -> OnYield r a+ OnSendShutdown :: !ProcessId -> !ShutdownRequest -> (ResumeProcess Bool -> Eff r (OnYield r a)) -> OnYield r a+ OnMakeReference :: (ResumeProcess Int -> Eff r (OnYield r a)) -> OnYield r a -- | Internal 'Process' handler function. handleProcess@@ -133,31 +134,32 @@ => (forall a . Eff r a -> m a) -> m () -> ProcessId+ -> Int -> Map.Map ProcessId (Seq Dynamic) -> Seq (OnYield r finalResult, ProcessId) -> m (Either String finalResult)-handleProcess _runEff _yieldEff _newPid _msgQs Empty =+handleProcess _runEff _yieldEff _newPid _nextRef _msgQs Empty = return $ Left "no main process" -handleProcess runEff yieldEff !newPid !msgQs allProcs@((!processState, !pid) :<| rest)+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 in case processState of- OnDone r -> handleExit (Right r)-- OnShutdown -> handleExit (Left "process exited normally")+ OnDone r -> handleExit (Right r) - OnRaiseError errM -> handleExit (Left errM)+ OnShutdown ExitNormally -> handleExit (Left "process exited normally")+ OnShutdown (ExitWithError e) -> handleExit (Left e) - OnExitError errM -> handleExit (Left errM)+ OnRaiseError errM -> handleExit (Left errM) - OnSendShutdown targetPid k -> do+ OnSendShutdown targetPid sr k -> do let allButTarget = Seq.filter (\(_, e) -> e /= pid && e /= targetPid) allProcs targets = Seq.filter (\(_, e) -> e == targetPid) allProcs@@ -165,46 +167,73 @@ targetFound = suicide || not (Seq.null targets) if suicide then do- nextK <- runEff $ k ShutdownRequested- handleProcess runEff yieldEff newPid msgQs (rest :|> (nextK, pid))+ nextK <- runEff $ k (ShutdownRequested sr)+ handleProcess runEff+ yieldEff+ newPid+ nextRef+ msgQs+ (rest :|> (nextK, pid)) else do- let deliverTheGoodNews (targetState, tPid) = do- nextTargetState <- case targetState of- OnSendShutdown _ tk -> tk ShutdownRequested- OnYield tk -> tk ShutdownRequested- OnSelf tk -> tk ShutdownRequested- OnSend _ _ tk -> tk ShutdownRequested- OnRecv _ tk -> tk ShutdownRequested- OnSpawn _ tk -> tk ShutdownRequested- OnDone x -> return (OnDone x)- OnShutdown -> return OnShutdown- OnExitError er -> return (OnExitError er)- OnRaiseError er -> return (OnExitError er)- return (nextTargetState, tPid)+ 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) handleProcess runEff yieldEff newPid+ nextRef msgQs (allButTarget Seq.>< (nextTargets :|> (nextK, pid))) OnSelf k -> do nextK <- runEff $ k (ResumeWith pid)- handleProcess runEff yieldEff newPid msgQs (rest :|> (nextK, pid))+ handleProcess runEff+ yieldEff+ newPid+ nextRef+ msgQs+ (rest :|> (nextK, pid)) + OnMakeReference k -> do+ nextK <- runEff $ k (ResumeWith nextRef)+ handleProcess runEff+ yieldEff+ newPid+ (nextRef + 1)+ msgQs+ (rest :|> (nextK, pid))+ OnYield k -> do yieldEff nextK <- runEff $ k (ResumeWith ())- handleProcess runEff yieldEff newPid msgQs (rest :|> (nextK, pid))+ handleProcess runEff+ yieldEff+ newPid+ nextRef+ msgQs+ (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)) @@ -214,21 +243,33 @@ handleProcess runEff yieldEff (newPid + 1)+ 0 (msgQs & at newPid ?~ Seq.empty) (rest :|> (nextK, pid) :|> (fk, newPid)) recv@(OnRecv selectMessage k) -> case msgQs ^. at pid of Nothing -> do nextK <- runEff $ k (OnError (show pid ++ " has no message queue!"))- handleProcess runEff yieldEff newPid msgQs (rest :|> (nextK, pid))+ 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 msgQs (rest :|> (nextK, pid))+ handleProcess runEff+ yieldEff+ newPid+ nextRef+ msgQs+ (rest :|> (nextK, pid)) else handleProcess runEff yieldEff newPid+ nextRef msgQs (rest :|> (recv, pid)) Just messages ->@@ -241,6 +282,7 @@ Nothing -> handleProcess runEff yieldEff newPid+ nextRef msgQs (rest :|> (recv, pid)) Just (result, otherMessages) -> do@@ -248,6 +290,7 @@ handleProcess runEff yieldEff newPid+ nextRef (msgQs & at pid . _Just .~ otherMessages) (rest :|> (nextK, pid)) @@ -260,16 +303,16 @@ runAsCoroutinePure runEff = runEff . handle_relay (return . OnDone) cont where cont :: Process r x -> (x -> Eff r (OnYield r v)) -> Eff r (OnYield r v)- cont YieldProcess k = return (OnYield k)- cont SelfPid k = return (OnSelf k)- cont (Spawn e) k = return (OnSpawn e k)- cont Shutdown _k = return OnShutdown- cont (ExitWithError !e ) _k = return (OnExitError e)- cont (RaiseError !e ) _k = return (OnRaiseError e)- cont (SendMessage !tp !msg) k = return (OnSend tp msg k)- cont ReceiveMessage k = return (OnRecv (MessageSelector Just) k)- cont (ReceiveMessageSuchThat f ) k = return (OnRecv f k)- cont (SendShutdown !pid) k = return (OnSendShutdown pid k)+ cont YieldProcess k = return (OnYield k)+ cont SelfPid k = return (OnSelf k)+ cont (Spawn e ) k = return (OnSpawn e k)+ cont (Shutdown !sr ) _k = return (OnShutdown sr)+ cont (RaiseError !e ) _k = return (OnRaiseError e)+ cont (SendMessage !tp !msg) k = return (OnSend tp msg k)+ cont ReceiveMessage k = return (OnRecv (MessageSelector Just) k)+ cont (ReceiveMessageSuchThat f) k = return (OnRecv f k)+ cont (SendShutdown !pid !sr ) k = return (OnSendShutdown pid sr k)+ cont MakeReference k = return (OnMakeReference k) -- | The concrete list of 'Eff'ects for running this pure scheduler on @IO@ and -- with string logging.
src/Control/Eff/Log/Channel.hs view
@@ -2,7 +2,7 @@ module Control.Eff.Log.Channel ( LogChannel() , logToChannel- , noLogger+ , nullLogChannel , forkLogger , filterLogChannel , joinLogChannel@@ -12,12 +12,15 @@ , logChannelPutIO , JoinLogChannelException() , KillLogChannelException()+ , handleLoggingAndIO+ , handleLoggingAndIO_ ) where import Control.Concurrent import Control.Concurrent.STM import Control.Eff as Eff+import Control.Eff.Lift import Control.Exception ( bracket ) import qualified Control.Exception as Exc import Control.Monad ( void@@ -33,6 +36,7 @@ import Data.Kind ( ) import Data.String import Data.Typeable+import GHC.Stack -- | A log channel processes logs from the 'Logs' effect by en-queuing them in a -- shared queue read from a seperate processes. A channel can contain log@@ -68,8 +72,8 @@ -- | Create a 'LogChannel' that will discard all messages sent -- via 'forwardLogstochannel' or 'logChannelPutIO'.-noLogger :: LogChannel message-noLogger = DiscardLogs+nullLogChannel :: LogChannel message+nullLogChannel = DiscardLogs -- | Fork a new process, that applies a monadic action to all log messages sent -- via 'logToChannel' or 'logChannelPutIO'.@@ -104,8 +108,11 @@ logLoop :: TBQueue message -> IO () logLoop tq = do- m <- atomically $ readTBQueue tq- handle m+ m <- atomically $ do+ h <- readTBQueue tq+ t <- flushTBQueue tq+ return (h : t)+ traverse_ handle m logLoop tq -- | Filter logs sent to a 'LogChannel' using a predicate.@@ -113,7 +120,7 @@ :: (message -> Bool) -> LogChannel message -> LogChannel message filterLogChannel = FilteredLogChannel --- | Run an action and close a 'LogChannel' created by 'noLogger', 'forkLogger'+-- | Run an action and close a 'LogChannel' created by 'nullLogChannel', 'forkLogger' -- or 'filterLogChannel' afterwards using 'joinLogChannel'. If a -- 'Exc.SomeException' was thrown, the log channel is killed with -- 'killLogChannel', and the exception is re-thrown.@@ -187,3 +194,14 @@ let logHandler = void . runInIO . logMessage bracket (forkLogger queueLen logHandler mWelcome) joinLogChannel f )++-- | Handle the 'LoggingAndIO' effects, using a 'LogChannel' for the 'Logs'+-- effect.+handleLoggingAndIO+ :: HasCallStack => Eff '[Logs m, Lift IO] a -> LogChannel m -> IO a+handleLoggingAndIO e lc = runLift (logToChannel lc e)++-- | Like 'handleLoggingAndIO' but return @()@.+handleLoggingAndIO_+ :: HasCallStack => Eff '[Logs m, Lift IO] a -> LogChannel m -> IO ()+handleLoggingAndIO_ = ((.) . (.)) void handleLoggingAndIO
src/Control/Eff/Log/Handler.hs view
@@ -2,37 +2,46 @@ module Control.Eff.Log.Handler ( -- * Logging Effect- Logs(..)+ Logs+ , LogWriter(..)+ , askLogWriter , logMsg , interceptLogging , foldLogMessages- , relogAsString- , captureLogs , ignoreLogs , handleLogsWith- , handleLogsLifted , handleLogsWithLoggingTHandler ) where import Control.DeepSeq import Control.Eff as Eff-import Control.Eff.Extend as Eff+import Control.Eff.Reader.Strict import qualified Control.Eff.Lift as Eff import qualified Control.Monad.Log as Log+import Control.Monad.IO.Class import Data.Foldable ( traverse_ ) import Data.Kind ( )-import Data.Sequence ( Seq() )-import qualified Data.Sequence as Seq --- | Logging effect type, parameterized by a log message type.-data Logs message a where- LogMsg :: message -> Logs message ()+-- | A function that takes a log message and returns an effect that+-- /logs/ the message.+newtype LogWriter message = LogWriter { runLogWriter :: message -> IO () } +-- | Logging effect type, parameterized by a log message type. This is a reader+-- for a 'LogWriter'.+type Logs message = Reader (LogWriter message)+ -- | Log a message.-logMsg :: Member (Logs m) r => m -> Eff r ()-logMsg = send . LogMsg+logMsg :: (NFData m, MonadIO (Eff r), Member (Logs m) r) => m -> Eff r ()+logMsg !(force -> m) = do+ lw <- ask+ liftIO (runLogWriter lw m) ++-- | Get the current 'LogWriter'+askLogWriter :: (Member (Logs m) r) => Eff r (LogWriter m)+askLogWriter = ask+ -- | Change, add or remove log messages and perform arbitrary actions upon -- intercepting a log message. --@@ -51,87 +60,45 @@ -- sending/receiving, to intercept the messages and execute some code and then -- return a new message. interceptLogging- :: forall r m a . Member (Logs m) r => (m -> Eff r ()) -> Eff r a -> Eff r a-interceptLogging interceptor = interpose return go- where- go :: Member (Logs m) r => Logs m x -> (Arr r x y) -> Eff r y- go (LogMsg m) k = do- interceptor m- k ()+ :: forall r m a+ . (Member (Logs m) r, Eff.Lifted IO r)+ => (m -> Eff '[Logs m, Eff.Lift IO] ())+ -> Eff r a+ -> Eff r a+interceptLogging interceptor action = do+ old <- ask+ let new = LogWriter (Eff.runLift . runReader old . interceptor)+ local (const new) action -- | Intercept logging to change, add or remove log messages. -- -- This is without side effects, hence faster than 'interceptLogging'. foldLogMessages :: forall r m a f- . (Foldable f, Member (Logs m) r)+ . (NFData m, Foldable f, Member (Logs m) r, Eff.Lifted IO r) => (m -> f m) -> Eff r a -> Eff r a-foldLogMessages interceptor = interpose return go- where- go :: Member (Logs m) r => Logs m x -> (Arr r x y) -> Eff r y- go (LogMsg m) k = do- traverse_ logMsg (interceptor m)- k ()---- | Capture all log messages in a 'Seq' (strict).-captureLogs- :: NFData message => Eff (Logs message ': r) a -> Eff r (a, Seq message)-captureLogs = Eff.handle_relay_s Seq.empty- (\logs result -> return (result, logs))- handleLogs- where- handleLogs- :: NFData message- => Seq message- -> Logs message x- -> (Seq message -> Arr r x y)- -> Eff r y- handleLogs !logs (LogMsg !m) k = k (force (logs Seq.:|> m)) ()+foldLogMessages f = interceptLogging (traverse_ logMsg . f) -- | Throw away all log messages. ignoreLogs :: forall message r a . Eff (Logs message ': r) a -> Eff r a-ignoreLogs = Eff.handle_relay return handleLogs- where- handleLogs :: Logs m x -> Arr r x y -> Eff r y- handleLogs (LogMsg _) k = k ()---- | Handle a 'Logs' effect with a message that has a 'Show' instance by--- **re-logging** each message applied to 'show'.-relogAsString- :: forall m e a- . (Show m, Member (Logs String) e)- => Eff (Logs m ': e) a- -> Eff e a-relogAsString = handleLogsWith (logMsg . show)+ignoreLogs = runReader (LogWriter (const (pure ()))) -- | Apply a function that returns an effect to each log message. handleLogsWith- :: forall message e a- . (message -> Eff e ())- -> Eff (Logs message ': e) a- -> Eff e a-handleLogsWith h = handle_relay return $ \(LogMsg m) k -> h m >>= k---- | Handle the 'Logs' effect with a monadic call back function (strict).-handleLogsLifted- :: forall m r message a- . (NFData message, Monad m, SetMember Eff.Lift (Eff.Lift m) r)- => (message -> m ())+ :: forall message r a+ . (message -> IO ()) -> Eff (Logs message ': r) a -> Eff r a-handleLogsLifted logMessageHandler = handleLogsWith go- where- go :: message -> Eff r ()- go m = Eff.lift (logMessageHandler (force m))+handleLogsWith = runReader . LogWriter -- | Handle the 'Logs' effect using 'Log.LoggingT' 'Log.Handler's. handleLogsWithLoggingTHandler- :: forall m r message a- . (Monad m, SetMember Eff.Lift (Eff.Lift m) r)- => (forall b . (Log.Handler m message -> m b) -> m b)+ :: forall r message a+ . (Eff.Lifted IO r)+ => (forall b . (Log.Handler IO message -> IO b) -> IO b) -> Eff (Logs message ': r) a -> Eff r a handleLogsWithLoggingTHandler foldHandler =- handleLogsWith (Eff.lift . foldHandler . flip ($))+ handleLogsWith (foldHandler . flip ($!))
src/Control/Eff/Log/Message.hs view
@@ -5,6 +5,8 @@ , renderRFC5424 , printLogMessage , relogAsDebugMessages+ , increaseLogMessageDistance+ , dropDistantLogMessages , logWithSeverity , logEmergency , logAlert@@ -64,6 +66,7 @@ , lmSrcLoc , lmThreadId , lmMessage+ , lmDistance , setCallStack , setLogMessageTimestamp , setLogMessageThreadId@@ -73,22 +76,23 @@ ) where -import Data.Time.Clock-import Data.Time.Format-import Control.Lens+import Control.Concurrent+import Control.DeepSeq import Control.Eff+import Control.Eff.Lift import Control.Eff.Log.Handler-import GHC.Stack-import Data.Default-import Control.DeepSeq+import Control.Lens+import Control.Monad ( (>=>) ) import Control.Monad.IO.Class+import Data.Default import Data.Maybe import Data.String-import Control.Concurrent+import Data.Time.Clock+import Data.Time.Format import GHC.Generics-import Text.Printf+import GHC.Stack import System.FilePath.Posix-import Control.Monad ( (>=>) )+import Text.Printf -- | A message data type inspired by the RFC-5424 Syslog Protocol data LogMessage =@@ -102,11 +106,12 @@ , _lmStructuredData :: [StructuredDataElement] , _lmThreadId :: Maybe ThreadId , _lmSrcLoc :: Maybe SrcLoc- , _lmMessage :: String}+ , _lmMessage :: String+ , _lmDistance :: Int } deriving (Eq, Generic) showLmMessage :: LogMessage -> [String]-showLmMessage (LogMessage _f _s _ts _hn _an _pid _mi _sd ti loc msg) =+showLmMessage (LogMessage _f _s _ts _hn _an _pid _mi _sd ti loc msg _dist) = if null msg then [] else@@ -125,7 +130,7 @@ -- | Render a 'LogMessage' human readable. renderLogMessage :: LogMessage -> String-renderLogMessage l@(LogMessage _f s ts hn an pid mi sd _ _ _) =+renderLogMessage l@(LogMessage _f s ts hn an pid mi sd _ _ _ _) = unwords $ filter (not . null) ( maybe@@ -144,7 +149,7 @@ -- | Render a 'LogMessage' according to the rules in the given RFC, except for -- the rules concerning unicode and ascii renderRFC5424 :: LogMessage -> String-renderRFC5424 l@(LogMessage f s ts hn an pid mi sd _ _ _) = unwords+renderRFC5424 l@(LogMessage f s ts hn an pid mi sd _ _ _ _) = unwords ( ("<" ++ show (fromSeverity s + fromFacility f * 8) ++ ">" ++ "1") : maybe "-"@@ -223,7 +228,7 @@ (_, srcLoc) : _ -> m & lmSrcLoc ?~ srcLoc instance Default LogMessage where- def = LogMessage def def def def def def def def def def ""+ def = LogMessage def def def def def def def def def def "" 0 instance IsString LogMessage where fromString = infoMessage@@ -246,17 +251,39 @@ t <- liftIO myThreadId return (m & lmThreadId ?~ t) +-- | Increase the /distance/ of log messages by one.+-- Logs can be filtered by their distance with 'dropDistantLogMessages'+increaseLogMessageDistance+ :: (HasCallStack, Member (Logs LogMessage) e, MonadIO (Eff e), Lifted IO e)+ => Eff e a+ -> Eff e a+increaseLogMessageDistance = interceptLogging (logMsg . over lmDistance (+ 1))++-- | Drop all log messages with an 'lmDistance' greater than the given+-- value.+dropDistantLogMessages+ :: (SetMember Lift (Lift IO) r, Member (Logs LogMessage) r)+ => Int+ -> Eff r a+ -> Eff r a+dropDistantLogMessages maxDistance = foldLogMessages+ (\lm -> if lm ^. lmDistance > maxDistance then Nothing else Just lm)+ -- | Handle a 'Logs' effect for 'String' messages by re-logging the messages -- as 'LogMessage's with 'debugSeverity'. relogAsDebugMessages- :: (HasCallStack, Member (Logs LogMessage) e)+ :: (HasCallStack, Member (Logs LogMessage) e, MonadIO (Eff e)) => Eff (Logs String ': e) a -> Eff e a-relogAsDebugMessages = withFrozenCallStack (handleLogsWith logDebug)+relogAsDebugMessages e = withFrozenCallStack+ (do+ LogWriter logMsgLogger <- askLogWriter+ handleLogsWith (logMsgLogger . debugMessage) e+ ) -- | Log a 'String' as 'LogMessage' with a given 'Severity'. logWithSeverity- :: (HasCallStack, Member (Logs LogMessage) e)+ :: (HasCallStack, Member (Logs LogMessage) e, MonadIO (Eff e)) => Severity -> String -> Eff e ()@@ -268,35 +295,59 @@ . flip (set lmMessage) def -- | Log a 'String' as 'emergencySeverity'.-logEmergency :: (HasCallStack, Member (Logs LogMessage) e) => String -> Eff e ()+logEmergency+ :: (HasCallStack, Member (Logs LogMessage) e, MonadIO (Eff e))+ => String+ -> Eff e () logEmergency = withFrozenCallStack (logWithSeverity emergencySeverity) -- | Log a message with 'alertSeverity'.-logAlert :: (HasCallStack, Member (Logs LogMessage) e) => String -> Eff e ()+logAlert+ :: (HasCallStack, Member (Logs LogMessage) e, MonadIO (Eff e))+ => String+ -> Eff e () logAlert = withFrozenCallStack (logWithSeverity alertSeverity) -- | Log a 'criticalSeverity' message.-logCritical :: (HasCallStack, Member (Logs LogMessage) e) => String -> Eff e ()+logCritical+ :: (HasCallStack, Member (Logs LogMessage) e, MonadIO (Eff e))+ => String+ -> Eff e () logCritical = withFrozenCallStack (logWithSeverity criticalSeverity) -- | Log a 'errorSeverity' message.-logError :: HasCallStack => Member (Logs LogMessage) e => String -> Eff e ()+logError+ :: (HasCallStack, Member (Logs LogMessage) e, MonadIO (Eff e))+ => String+ -> Eff e () logError = withFrozenCallStack (logWithSeverity errorSeverity) -- | Log a 'warningSeverity' message.-logWarning :: (HasCallStack, Member (Logs LogMessage) e) => String -> Eff e ()+logWarning+ :: (HasCallStack, Member (Logs LogMessage) e, MonadIO (Eff e))+ => String+ -> Eff e () logWarning = withFrozenCallStack (logWithSeverity warningSeverity) -- | Log a 'noticeSeverity' message.-logNotice :: (HasCallStack, Member (Logs LogMessage) e) => String -> Eff e ()+logNotice+ :: (HasCallStack, Member (Logs LogMessage) e, MonadIO (Eff e))+ => String+ -> Eff e () logNotice = withFrozenCallStack (logWithSeverity noticeSeverity) -- | Log a 'informationalSeverity' message.-logInfo :: (HasCallStack, Member (Logs LogMessage) e) => String -> Eff e ()+logInfo+ :: (HasCallStack, Member (Logs LogMessage) e, MonadIO (Eff e))+ => String+ -> Eff e () logInfo = withFrozenCallStack (logWithSeverity informationalSeverity) -- | Log a 'debugSeverity' message.-logDebug :: (HasCallStack, Member (Logs LogMessage) e) => String -> Eff e ()+logDebug+ :: (HasCallStack, Member (Logs LogMessage) e, MonadIO (Eff e))+ => String+ -> Eff e () logDebug = withFrozenCallStack (logWithSeverity debugSeverity) -- | Construct a 'LogMessage' with 'errorSeverity'
stack.yaml view
@@ -36,23 +36,23 @@ # non-dependency (i.e. a user package), and its test suites and benchmarks # will not be run. This is useful for tweaking upstream packages. packages:-- .-# Dependency packages to be pulled from upstream that are not in the resolver-# (e.g., acme-missiles-0.3)+ - .+ # 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.1 -# Override default flag values for local packages and extra-deps-# flags: {}+ # Override default flag values for local packages and extra-deps+ # flags: {} -# Extra package databases containing global packages-# extra-package-dbs: []+ # Extra package databases containing global packages+ # extra-package-dbs: [] -# Control whether we use the GHC we find on the path-# system-ghc: true-#-# Require a specific version of stack, using version ranges-# require-stack-version: -any # Default+ # Control whether we use the GHC we find on the path+ # system-ghc: true+ #+ # Require a specific version of stack, using version ranges+ # require-stack-version: -any # Default require-stack-version: ">=1.7" # # Override the architecture used by stack, especially useful on Windows
test/Common.hs view
@@ -10,6 +10,7 @@ import Test.Tasty import Test.Tasty.HUnit import Test.Tasty.Runners+import GHC.Stack setTravisTestOptions :: TestTree -> TestTree setTravisTestOptions =@@ -29,31 +30,26 @@ (return (\e -> do logC <- logCFactory- doSchedule e logC -- noLogger+ doSchedule e logC+ -- doSchedule e nullLogChannel ) ) ) testLogC :: IO (LogChannel LogMessage) testLogC =- filterLogChannel (\m -> take (length logPrefix) (_lmMessage m) == logPrefix)- <$> forkLogger 1 printLogMessage Nothing+ -- filterLogChannel ((< informationalSeverity) . _lmSeverity) <$>+ forkLogger 1000 printLogMessage Nothing testLogJoin :: LogChannel LogMessage -> IO () testLogJoin = joinLogChannel -tlog :: Member (Logs LogMessage) r => String -> Eff r ()-tlog = logInfo . (logPrefix ++)--logPrefix :: String-logPrefix = "[TEST] "- untilShutdown :: Member t r => t (ResumeProcess v) -> Eff r () untilShutdown pa = do r <- send pa case r of- ShutdownRequested -> return ()- _ -> untilShutdown pa+ ShutdownRequested _ -> return ()+ _ -> untilShutdown pa scheduleAndAssert :: forall r@@ -63,7 +59,7 @@ -> Eff (Process r ': r) () ) -> IO ()-scheduleAndAssert schedulerFactory testCaseAction = do+scheduleAndAssert schedulerFactory testCaseAction = withFrozenCallStack $ do resultVar <- newEmptyTMVarIO void (applySchedulerFactory
test/Debug.hs view
@@ -3,7 +3,6 @@ import ProcessBehaviourTestCases import Test.Tasty-import Test.Tasty.HUnit debugMain :: IO () debugMain = defaultMain test_forkIo
test/ForkIOScheduler.hs view
@@ -1,5 +1,4 @@-module ForkIOScheduler-where+module ForkIOScheduler where import Control.Exception import Control.Concurrent@@ -35,7 +34,9 @@ lift (threadDelay 1000) doExit lift (threadDelay 100000)- wasStillRunningP1 <- sendShutdownChecked forkIoScheduler p1+ wasStillRunningP1 <- sendShutdownChecked forkIoScheduler+ p1+ ExitNormally lift (atomically (putTMVar aVar wasStillRunningP1)) wasStillRunningP1 <- atomically (takeTMVar aVar)@@ -45,8 +46,10 @@ , ( "sending" , void (send (SendMessage @SchedulerIO 44444 (toDyn "test message"))) )- , ("sending shutdown", void (send (SendShutdown @SchedulerIO 44444)))- , ("selfpid-ing" , void (send (SelfPid @SchedulerIO)))+ , ( "sending shutdown"+ , void (send (SendShutdown @SchedulerIO 44444 ExitNormally))+ )+ , ("selfpid-ing", void (send (SelfPid @SchedulerIO))) , ( "spawn-ing" , void (send (Spawn @SchedulerIO (void (send (ReceiveMessage @SchedulerIO)))))@@ -54,11 +57,9 @@ ] , (howToExit, doExit ) <- [ ("throw async exception", void (lift (throw UserInterrupt)))- , ( "division by zero"- , void (return ((123 :: Int) `div` 0) >>= lift . putStrLn . show)- )- , ("call 'fail'" , void (fail "test fail"))- , ("call 'error'", void (error "test error"))+ , ("division by zero" , void ((lift . print) ((123 :: Int) `div` 0)))+ , ("call 'fail'" , void (fail "test fail"))+ , ("call 'error'" , void (error "test error")) ] ]
test/LoggingTests.hs view
@@ -1,38 +1,44 @@ module LoggingTests where import Control.Eff+import Control.Eff.Lift import Control.Eff.Log import Test.Tasty import Test.Tasty.HUnit import Common+import Control.Concurrent.STM import Control.DeepSeq-import qualified Data.Sequence as Seq-+import Control.Monad.IO.Class -demo :: Member (Logs String) e => Eff e ()+demo+ :: (Member (Logs String) e, Member (Logs OtherLogMsg) e, MonadIO (Eff e))+ => Eff e () demo = do logMsg "jo"- logMsg "test 123"+ logMsg (OtherLogMsg "test 123") test_loggingInterception :: TestTree test_loggingInterception = setTravisTestOptions $ testGroup "Intercept Logging" [ testCase "Convert Log Message Type" $ do- let (((), strLogs), otherLogs) = run- (captureLogs @OtherLogMsg- (captureLogs @String (interceptLogging toOtherLogMsg demo))- )- assertEqual "string logs must be empty" Seq.empty strLogs- assertEqual "other logs should contain all logging"- (Seq.fromList [OtherLogMsg "jo", OtherLogMsg "test 123"])- otherLogs+ otherLogsQueue <- newTQueueIO @OtherLogMsg+ stringLogsQueue <- newTQueueIO @String+ runLift+ (handleLogsWith+ (atomically . writeTQueue stringLogsQueue)+ (handleLogsWith (atomically . writeTQueue otherLogsQueue)+ (interceptLogging reverseStringLogs demo)+ )+ )+ otherLogs <- atomically (flushTQueue otherLogsQueue)+ strLogs <- atomically (flushTQueue stringLogsQueue)+ assertEqual "string logs" ["oj"] strLogs+ assertEqual "other logs" [OtherLogMsg "test 123"] otherLogs ] newtype OtherLogMsg = OtherLogMsg String deriving (Eq, NFData, Show) -toOtherLogMsg :: ('[Logs String, Logs OtherLogMsg] <:: e) => String -> Eff e ()-toOtherLogMsg = logMsg . OtherLogMsg--demo2 :: Member (Logs LogMessage) e => Eff e ()-demo2 = relogAsDebugMessages demo+reverseStringLogs+ :: ('[Logs String, Lift IO] <:: e, MonadIO (Eff e)) => String -> Eff e ()+reverseStringLogs = logMsg . reverse
test/ProcessBehaviourTestCases.hs view
@@ -1,9 +1,12 @@ module ProcessBehaviourTestCases where import Data.List ( sort )-import Data.Dynamic import Data.Foldable ( traverse_ )-import Data.Dynamic ( fromDynamic )+import Data.Dynamic ( fromDynamic+ , toDyn+ )+import Data.Typeable+import Data.Default import Control.Exception import Control.Concurrent import Control.Concurrent.STM@@ -24,10 +27,11 @@ import Test.Tasty import Test.Tasty.HUnit import Common+import Debug.Trace test_forkIo :: TestTree test_forkIo = setTravisTestOptions- (withTestLogC ForkIO.schedule+ (withTestLogC (\c lc -> handleLoggingAndIO_ (ForkIO.schedule c) lc) (\factory -> testGroup "ForkIOScheduler" [allTests factory]) ) @@ -83,8 +87,8 @@ -> String -> Eff r Bool returnToSender px toP msg = do- me <- self px- call px toP (ReturnToSender me msg)+ me <- self px+ _ <- call px toP (ReturnToSender me msg) msgEcho <- receiveMessageAs @String px return (msgEcho == msg) @@ -94,9 +98,7 @@ => SchedulerProxy q -> Server ReturnToSender -> Eff r ()-stopReturnToSender px toP = do- me <- self px- call px toP StopReturnToSender+stopReturnToSender px toP = call px toP StopReturnToSender returnToSenderServer :: forall q r@@ -107,14 +109,18 @@ => SchedulerProxy q -> Eff r (Server ReturnToSender) returnToSenderServer px = asServer <$> spawn- (serve px $ ApiHandler- { _handleCall = \m k -> case m of- StopReturnToSender -> k () >> exitNormally px- ReturnToSender fromP echoMsg ->- sendMessageChecked px fromP (toDyn echoMsg)- >>= (\res -> yieldProcess px >> k res)- , _handleCast = logWarning . show- , _handleTerminate = logWarning . show+ (serve px $ def+ { _callCallback = Just+ (\m k -> case m of+ StopReturnToSender -> do+ k ()+ return (StopApiServer Nothing)+ ReturnToSender fromP echoMsg -> do+ ok <- sendMessageChecked px fromP (toDyn echoMsg)+ yieldProcess px+ k ok+ return HandleNextRequest+ ) } ) @@ -310,7 +316,7 @@ px (assertEff "error must be caught" . (== "test error 3")) (do- void (replicateM 100000 (void (self px)))+ replicateM_ 100000 (void (self px)) void (raiseError px "test error 3") ) ]@@ -343,12 +349,7 @@ ) ) [0, 5 .. n]- oks <- replicateM- (length [0, 5 .. n])- (do- j <- receiveMessageAs px- return j- )+ oks <- replicateM (length [0, 5 .. n]) (receiveMessageAs px) assertEff "" (sort oks == [0, 5 .. n]) ] ]@@ -401,8 +402,11 @@ child2 <- spawn (foreverCheap (void (sendMessage px 888 (toDyn "")))) True <- sendMessageChecked px child1 (toDyn "test")- i <- receiveMessageAs px- True <- sendShutdownChecked px child2+ traceM "now receiveMessageAs"+ i <- receiveMessageAs px+ traceM ("receiveMessageAs returned " ++ show i)+ True <- sendShutdownChecked px child2 ExitNormally+ traceM "sendShutdownChecked" assertEff "" (i == "test") , testCase "most processes send foreverCheap" $ scheduleAndAssert schedulerFactory@@ -415,12 +419,7 @@ $ void (sendMessage px 888 (toDyn "test message to 888")) ) [0 .. n]- oks <- replicateM- (length [0, 5 .. n])- (do- j <- receiveMessageAs px- return j- )+ oks <- replicateM (length [0, 5 .. n]) (receiveMessageAs px) assertEff "" (sort oks == [0, 5 .. n]) , testCase "most processes self foreverCheap" $ scheduleAndAssert schedulerFactory@@ -432,12 +431,7 @@ foreverCheap $ void (self px) ) [0 .. n]- oks <- replicateM- (length [0, 5 .. n])- (do- j <- receiveMessageAs px- return j- )+ oks <- replicateM (length [0, 5 .. n]) (receiveMessageAs px) assertEff "" (sort oks == [0, 5 .. n]) , testCase "most processes sendShutdown foreverCheap" $ scheduleAndAssert schedulerFactory@@ -446,15 +440,10 @@ traverse_ (\(i :: Int) -> spawn $ do when (i `rem` 5 == 0) $ void $ sendMessage px me (toDyn i)- foreverCheap $ void (sendShutdown px 999)+ foreverCheap $ void (sendShutdown px 999 ExitNormally) ) [0 .. n]- oks <- replicateM- (length [0, 5 .. n])- (do- j <- receiveMessageAs px- return j- )+ oks <- replicateM (length [0, 5 .. n]) (receiveMessageAs px) assertEff "" (sort oks == [0, 5 .. n]) , testCase "most processes spawn foreverCheap" $ scheduleAndAssert schedulerFactory@@ -474,12 +463,7 @@ ) ) [0 .. n]- oks <- replicateM- (length [0, 5 .. n])- (do- j <- receiveMessageAs px- return j- )+ oks <- replicateM (length [0, 5 .. n]) (receiveMessageAs px) assertEff "" (sort oks == [0, 5 .. n]) , testCase "most processes receive foreverCheap" $ scheduleAndAssert schedulerFactory@@ -491,12 +475,7 @@ foreverCheap $ void (receiveMessage px) ) [0 .. n]- oks <- replicateM- (length [0, 5 .. n])- (do- j <- receiveMessageAs px- return j- )+ oks <- replicateM (length [0, 5 .. n]) (receiveMessageAs px) assertEff "" (sort oks == [0, 5 .. n]) ] @@ -542,8 +521,10 @@ , ( "sending" , void (send (SendMessage @r 44444 (toDyn "test message"))) )- , ("sending shutdown", void (send (SendShutdown @r 44444)))- , ("selfpid-ing" , void (send (SelfPid @r)))+ , ( "sending shutdown"+ , void (send (SendShutdown @r 44444 ExitNormally))+ )+ , ("selfpid-ing", void (send (SelfPid @r))) , ( "spawn-ing" , void (send (Spawn @r (void (send (ReceiveMessage @r))))) )@@ -562,8 +543,10 @@ , ( "sending" , void (send (SendMessage @r 44444 (toDyn "test message"))) )- , ("sending shutdown", void (send (SendShutdown @r 44444)))- , ("selfpid-ing" , void (send (SelfPid @r)))+ , ( "sending shutdown"+ , void (send (SendShutdown @r 44444 ExitNormally))+ )+ , ("selfpid-ing", void (send (SelfPid @r))) , ( "spawn-ing" , void (send (Spawn @r (void (send (ReceiveMessage @r))))) )@@ -585,7 +568,7 @@ lift (threadDelay 1000) doExit lift (threadDelay 100000)- wasStillRunningP1 <- sendShutdownChecked px p1+ wasStillRunningP1 <- sendShutdownChecked px p1 ExitNormally assertEff "the other process was still running" wasStillRunningP1 | (busyWith , busyEffect) <-@@ -593,8 +576,10 @@ , ( "sending" , void (send (SendMessage @r 44444 (toDyn "test message"))) )- , ("sending shutdown", void (send (SendShutdown @r 44444)))- , ("selfpid-ing" , void (send (SelfPid @r)))+ , ( "sending shutdown"+ , void (send (SendShutdown @r 44444 ExitNormally))+ )+ , ("selfpid-ing", void (send (SelfPid @r))) , ( "spawn-ing" , void (send (Spawn @r (void (send (ReceiveMessage @r))))) )@@ -607,7 +592,7 @@ , ( "sendShutdown to self" , do me <- self px- void (sendShutdown px me)+ void (sendShutdown px me ExitNormally) ) ] ]@@ -626,18 +611,19 @@ "sendShutdown" [ testCase "... self" $ applySchedulerFactory schedulerFactory $ do me <- self px- void $ sendShutdown px me+ void $ sendShutdown px me ExitNormally raiseError px "sendShutdown must not return" , testCase "... self low-level" $ scheduleAndAssert schedulerFactory $ \assertEff -> do me <- self px- r <- send (SendShutdown @r me)+ r <- send (SendShutdown @r me ExitNormally)+ traceShowM (show me ++": returned from SendShutdow to self " ++ show r) assertEff "ShutdownRequested must be returned" (case r of- ShutdownRequested -> True- _ -> False+ ShutdownRequested ExitNormally -> True+ _ -> False ) , testGroup "... other process"@@ -650,7 +636,7 @@ untilShutdown (SendMessage @r 666 (toDyn "test")) void (sendMessage px me (toDyn "OK")) )- void (sendShutdown px other)+ void (sendShutdown px other ExitNormally) a <- receiveMessageAs px assertEff "" (a == "OK") , testCase "while it is receiving"@@ -662,7 +648,7 @@ untilShutdown (ReceiveMessage @r) void (sendMessage px me (toDyn "OK")) )- void (sendShutdown px other)+ void (sendShutdown px other ExitNormally) a <- receiveMessageAs px assertEff "" (a == "OK") , testCase "while it is self'ing"@@ -674,7 +660,7 @@ untilShutdown (SelfPid @r) void (sendMessage px me (toDyn "OK")) )- void (sendShutdown px other)+ void (sendShutdown px other (ExitWithError "testError")) a <- receiveMessageAs px assertEff "" (a == "OK") , testCase "while it is spawning"@@ -686,7 +672,7 @@ untilShutdown (Spawn @r (return ())) void (sendMessage px me (toDyn "OK")) )- void (sendShutdown px other)+ void (sendShutdown px other ExitNormally) a <- receiveMessageAs px assertEff "" (a == "OK") , testCase "while it is sending shutdown messages"@@ -695,10 +681,10 @@ me <- self px other <- spawn (do- untilShutdown (SendShutdown @r 777)+ untilShutdown (SendShutdown @r 777 ExitNormally) void (sendMessage px me (toDyn "OK")) )- void (sendShutdown px other)+ void (sendShutdown px other ExitNormally) a <- receiveMessageAs px assertEff "" (a == "OK") ]