diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -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 diff --git a/examples/example-1/Main.hs b/examples/example-1/Main.hs --- a/examples/example-1/Main.hs +++ b/examples/example-1/Main.hs @@ -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 diff --git a/examples/example-2/Main.hs b/examples/example-2/Main.hs --- a/examples/example-2/Main.hs +++ b/examples/example-2/Main.hs @@ -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") diff --git a/extensible-effects-concurrent.cabal b/extensible-effects-concurrent.cabal --- a/extensible-effects-concurrent.cabal +++ b/extensible-effects-concurrent.cabal @@ -1,5 +1,5 @@ name: extensible-effects-concurrent -version: 0.8 +version: 0.9.0 description: Please see the README on GitHub at synopsis: Message passing concurrency as extensible-effect homepage: https://github.com/sheyll/extensible-effects-concurrent#readme @@ -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 diff --git a/src/Control/Eff/Concurrent.hs b/src/Control/Eff/Concurrent.hs --- a/src/Control/Eff/Concurrent.hs +++ b/src/Control/Eff/Concurrent.hs @@ -42,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 diff --git a/src/Control/Eff/Concurrent/Api/Client.hs b/src/Control/Eff/Concurrent/Api/Client.hs --- a/src/Control/Eff/Concurrent/Api/Client.hs +++ b/src/Control/Eff/Concurrent/Api/Client.hs @@ -23,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 diff --git a/src/Control/Eff/Concurrent/Api/Internal.hs b/src/Control/Eff/Concurrent/Api/Internal.hs --- a/src/Control/Eff/Concurrent/Api/Internal.hs +++ b/src/Control/Eff/Concurrent/Api/Internal.hs @@ -1,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 diff --git a/src/Control/Eff/Concurrent/Api/Observer.hs b/src/Control/Eff/Concurrent/Api/Observer.hs --- a/src/Control/Eff/Concurrent/Api/Observer.hs +++ b/src/Control/Eff/Concurrent/Api/Observer.hs @@ -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) diff --git a/src/Control/Eff/Concurrent/Api/Observer/Queue.hs b/src/Control/Eff/Concurrent/Api/Observer/Queue.hs --- a/src/Control/Eff/Concurrent/Api/Observer/Queue.hs +++ b/src/Control/Eff/Concurrent/Api/Observer/Queue.hs @@ -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 = "" + -- | 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 diff --git a/src/Control/Eff/Concurrent/Api/Server.hs b/src/Control/Eff/Concurrent/Api/Server.hs --- a/src/Control/Eff/Concurrent/Api/Server.hs +++ b/src/Control/Eff/Concurrent/Api/Server.hs @@ -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) diff --git a/src/Control/Eff/Concurrent/Process.hs b/src/Control/Eff/Concurrent/Process.hs --- a/src/Control/Eff/Concurrent/Process.hs +++ b/src/Control/Eff/Concurrent/Process.hs @@ -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) diff --git a/src/Control/Eff/Concurrent/Process/ForkIOScheduler.hs b/src/Control/Eff/Concurrent/Process/ForkIOScheduler.hs --- a/src/Control/Eff/Concurrent/Process/ForkIOScheduler.hs +++ b/src/Control/Eff/Concurrent/Process/ForkIOScheduler.hs @@ -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) diff --git a/src/Control/Eff/Concurrent/Process/SingleThreadedScheduler.hs b/src/Control/Eff/Concurrent/Process/SingleThreadedScheduler.hs --- a/src/Control/Eff/Concurrent/Process/SingleThreadedScheduler.hs +++ b/src/Control/Eff/Concurrent/Process/SingleThreadedScheduler.hs @@ -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. diff --git a/src/Control/Eff/Log/Channel.hs b/src/Control/Eff/Log/Channel.hs --- a/src/Control/Eff/Log/Channel.hs +++ b/src/Control/Eff/Log/Channel.hs @@ -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 diff --git a/src/Control/Eff/Log/Handler.hs b/src/Control/Eff/Log/Handler.hs --- a/src/Control/Eff/Log/Handler.hs +++ b/src/Control/Eff/Log/Handler.hs @@ -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 ($!)) diff --git a/src/Control/Eff/Log/Message.hs b/src/Control/Eff/Log/Message.hs --- a/src/Control/Eff/Log/Message.hs +++ b/src/Control/Eff/Log/Message.hs @@ -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' diff --git a/stack.yaml b/stack.yaml --- a/stack.yaml +++ b/stack.yaml @@ -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 diff --git a/test/Common.hs b/test/Common.hs --- a/test/Common.hs +++ b/test/Common.hs @@ -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 diff --git a/test/Debug.hs b/test/Debug.hs --- a/test/Debug.hs +++ b/test/Debug.hs @@ -3,7 +3,6 @@ import ProcessBehaviourTestCases import Test.Tasty -import Test.Tasty.HUnit debugMain :: IO () debugMain = defaultMain test_forkIo diff --git a/test/ForkIOScheduler.hs b/test/ForkIOScheduler.hs --- a/test/ForkIOScheduler.hs +++ b/test/ForkIOScheduler.hs @@ -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")) ] ] diff --git a/test/LoggingTests.hs b/test/LoggingTests.hs --- a/test/LoggingTests.hs +++ b/test/LoggingTests.hs @@ -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 diff --git a/test/ProcessBehaviourTestCases.hs b/test/ProcessBehaviourTestCases.hs --- a/test/ProcessBehaviourTestCases.hs +++ b/test/ProcessBehaviourTestCases.hs @@ -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") ]