extensible-effects-concurrent 0.7.1 → 0.7.2
raw patch · 10 files changed
+276/−82 lines, 10 files
Files
- ChangeLog.md +9/−0
- extensible-effects-concurrent.cabal +2/−1
- src/Control/Eff/Concurrent.hs +16/−2
- src/Control/Eff/Concurrent/Api/Client.hs +18/−12
- src/Control/Eff/Concurrent/Api/Observer.hs +30/−30
- src/Control/Eff/Concurrent/Api/Observer/Queue.hs +149/−0
- src/Control/Eff/Concurrent/Api/Server.hs +11/−8
- src/Control/Eff/Concurrent/Process.hs +21/−18
- src/Control/Eff/Concurrent/Process/ForkIOScheduler.hs +12/−10
- src/Control/Eff/ExceptionExtra.hs +8/−1
ChangeLog.md view
@@ -1,5 +1,14 @@ # Changelog for extensible-effects-concurrent +## 0.7.2++- Add `withFrozenCallStack` to exposed functions+- Add `ObserverState` type alias+- Add `ObservationQueue` utility+- Fix missing re-exports from+ `Control.Eff.Concurrent.Api.Client`+ in `Control.Eff.Concurrent` introduced in recent versions+ ## 0.7.1 - Improve call-stack support in log messages
extensible-effects-concurrent.cabal view
@@ -1,5 +1,5 @@ name: extensible-effects-concurrent-version: 0.7.1+version: 0.7.2 description: Please see the README on GitHub at <https://github.com/sheyll/extensible-effects-concurrent#readme> synopsis: Message passing concurrency as extensible-effect homepage: https://github.com/sheyll/extensible-effects-concurrent#readme@@ -69,6 +69,7 @@ Control.Eff.Concurrent.Process.Interactive, Control.Eff.Concurrent.Process.SingleThreadedScheduler, Control.Eff.Concurrent.Api.Observer+ Control.Eff.Concurrent.Api.Observer.Queue other-modules: Control.Eff.Concurrent.Api.Internal, Paths_extensible_effects_concurrent,
src/Control/Eff/Concurrent.hs view
@@ -17,6 +17,9 @@ -- ** /Observer/ Functions for Events and Event Listener module Control.Eff.Concurrent.Api.Observer ,+ -- *** Capture /Observation/ in a FIFO Queue+ module Control.Eff.Concurrent.Api.Observer.Queue+ , -- * /Scheduler/ Process Effect Handler -- ** Concurrent Scheduler module Control.Eff.Concurrent.Process.ForkIOScheduler@@ -60,8 +63,7 @@ , catchRaisedError , ignoreProcessError )-import Control.Eff.Concurrent.Api- ( Api+import Control.Eff.Concurrent.Api ( Api , Synchronicity(..) , Server(..) , fromServer@@ -77,6 +79,8 @@ , callRegisteredA , ServesApi , registerServer+ , whereIsServer+ , ServerReader ) import Control.Eff.Concurrent.Api.Server ( serve@@ -97,6 +101,7 @@ import Control.Eff.Concurrent.Api.Observer ( Observer(..) , Observable(..)+ , ObserverState , notifyObserver , registerObserver , forgetObserver@@ -110,6 +115,15 @@ , CallbackObserver , spawnCallbackObserver , spawnLoggingObserver+ )+import Control.Eff.Concurrent.Api.Observer.Queue+ ( ObservationQueue()+ , ObservationQueueReader+ , readObservationQueue+ , tryReadObservationQueue+ , flushObservationQueue+ , enqueueObservationsRegistered+ , enqueueObservations ) import Control.Eff.Concurrent.Process.ForkIOScheduler ( schedule
src/Control/Eff/Concurrent/Api/Client.hs view
@@ -46,7 +46,7 @@ -> Api o 'Asynchronous -> Eff r Bool castChecked px (Server pid) castMsg =- sendMessageChecked px pid (toDyn $! (Cast $! castMsg))+ withFrozenCallStack $ 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 +62,7 @@ -> Server o -> Api o 'Asynchronous -> Eff r ()-cast px toServer apiRequest = do+cast px toServer apiRequest = withFrozenCallStack $ do _ <- castChecked px toServer apiRequest return () @@ -82,7 +82,7 @@ -> Server api -> Api api ( 'Synchronous result) -> Eff r result-call px (Server pidInt) req = do+call px (Server pidInt) req = withFrozenCallStack $ do fromPid <- self px let requestMessage = Call fromPid $! req wasSent <- sendMessageChecked px pidInt (toDyn $! requestMessage)@@ -112,8 +112,9 @@ -- | Run a reader effect that contains __the one__ server handling a specific -- 'Api' instance.-registerServer :: Server o -> Eff (ServerReader o ': r) a -> Eff r a-registerServer = runReader+registerServer+ :: HasCallStack => Server o -> Eff (ServerReader o ': r) a -> Eff r a+registerServer = withFrozenCallStack runReader -- | Get the 'Server' registered with 'registerServer'. whereIsServer :: Member (ServerReader o) e => Eff e (Server o)@@ -122,11 +123,11 @@ -- | Like 'call' but take the 'Server' from the reader provided by -- 'registerServer'. callRegistered- :: (Typeable reply, ServesApi o r q)+ :: (Typeable reply, ServesApi o r q, HasCallStack) => SchedulerProxy q -> Api o ( 'Synchronous reply) -> Eff r reply-callRegistered px method = do+callRegistered px method = withFrozenCallStack $ do serverPid <- whereIsServer call px serverPid method @@ -136,20 +137,25 @@ -- process communication. callRegisteredA :: forall r q o f reply- . (Alternative f, Typeable f, Typeable reply, ServesApi o r q)+ . ( Alternative f+ , Typeable f+ , Typeable reply+ , ServesApi o r q+ , HasCallStack+ ) => SchedulerProxy q -> Api o ( 'Synchronous (f reply)) -> Eff r (f reply)-callRegisteredA px method =- catchRaisedError px (const (return (empty @f))) (callRegistered px method)+callRegisteredA px method = withFrozenCallStack+ $ catchRaisedError px (const (return (empty @f))) (callRegistered px method) -- | Like 'cast' but take the 'Server' from the reader provided by -- 'registerServer'. castRegistered- :: (Typeable o, ServesApi o r q)+ :: (Typeable o, ServesApi o r q, HasCallStack) => SchedulerProxy q -> Api o 'Asynchronous -> Eff r ()-castRegistered px method = do+castRegistered px method = withFrozenCallStack $ do serverPid <- whereIsServer cast px serverPid method
src/Control/Eff/Concurrent/Api/Observer.hs view
@@ -14,6 +14,7 @@ , SomeObserver(..) , notifySomeObserver , Observers()+ , ObserverState , manageObservers , addObserver , removeObserver@@ -43,7 +44,7 @@ -- another 'Api' that is 'Observable'. class (Typeable p, Observable o) => Observer p o where -- | Wrap the 'Observation' and the 'ProcessId' (i.e. the 'Server')- -- that caused the observation into a 'Api' value that the+ -- that caused the observation into an 'Api' value that the -- 'Observable' understands. observationMessage :: Server o -> Observation o -> Api p 'Asynchronous @@ -65,8 +66,8 @@ -> Server o -> Observation o -> Eff r ()-notifyObserver px observer observed observation =- cast px observer (observationMessage observed observation)+notifyObserver px observer observed observation = withFrozenCallStack+ $ cast px observer (observationMessage observed observation) -- | Send the 'registerObserverMessage' registerObserver@@ -75,8 +76,8 @@ -> Server p -> Server o -> Eff r ()-registerObserver px observer observed =- cast px observed (registerObserverMessage (SomeObserver observer))+registerObserver px observer observed = withFrozenCallStack+ $ cast px observed (registerObserverMessage (SomeObserver observer)) -- | Send the 'forgetObserverMessage' forgetObserver@@ -85,8 +86,8 @@ -> Server p -> Server o -> Eff r ()-forgetObserver px observer observed =- cast px observed (forgetObserverMessage (SomeObserver observer))+forgetObserver px observer observed = withFrozenCallStack+ $ 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@@ -113,38 +114,35 @@ -> SomeObserver o -> Eff r () notifySomeObserver px observed observation (SomeObserver observer) =- notifyObserver px observer observed observation+ withFrozenCallStack $ notifyObserver px observer observed observation -- ** Manage 'Observers's --- | Internal state for 'manageobservers'+-- | Internal state for 'manageObservers' data Observers o = Observers { _observers :: Set (SomeObserver o) } +-- | Alias for the effect that contains the observers managed by 'manageObservers'+type ObserverState o = State (Observers o)+ observers :: Iso' (Observers o) (Set (SomeObserver o)) observers = iso _observers Observers -- | Keep track of registered 'Observer's Observers can be added and removed, -- and an 'Observation' can be sent to all registerd observers at once.-manageObservers :: Eff (State (Observers o) ': r) a -> Eff r a+manageObservers :: Eff (ObserverState o ': r) a -> Eff r a manageObservers = evalState (Observers Set.empty) -- | Add an 'Observer' to the 'Observers' managed by 'manageObservers'. addObserver- :: ( SetMember Process (Process q) r- , Member (State (Observers o)) r- , Observable o- )+ :: (SetMember Process (Process q) r, Member (ObserverState o) r, Observable o) => SomeObserver o -> Eff r () addObserver = modify . over observers . Set.insert -- | Delete an 'Observer' from the 'Observers' managed by 'manageObservers'. removeObserver- :: ( SetMember Process (Process q) r- , Member (State (Observers o)) r- , Observable o- )+ :: (SetMember Process (Process q) r, Member (ObserverState o) r, Observable o) => SomeObserver o -> Eff r () removeObserver = modify . over observers . Set.delete@@ -155,7 +153,7 @@ :: forall o r q . ( Observable o , SetMember Process (Process q) r- , Member (State (Observers o)) r+ , Member (ObserverState o) r ) => SchedulerProxy q -> Observation o@@ -188,21 +186,22 @@ , Show (Observation o) , Observable o , Member (Logs LogMessage) q+ , HasCallStack ) => SchedulerProxy q -> (Server o -> Observation o -> Eff (Process q ': q) Bool) -> Eff r (Server (CallbackObserver o)) spawnCallbackObserver px onObserve =- asServer @(CallbackObserver o)- <$> (spawn @r @q $ do- let loopUntil = serve- px- (ApiHandler @(CallbackObserver o) (handleCast loopUntil)- (unhandledCallError px)- (defaultTermination px)- )- loopUntil- )+ withFrozenCallStack $ asServer @(CallbackObserver o) <$> spawn @r @q+ (do+ let loopUntil = serve+ px+ (ApiHandler @(CallbackObserver o) (handleCast loopUntil)+ (unhandledCallError px)+ (defaultTermination px)+ )+ loopUntil+ ) where handleCast k (CbObserved fromSvr v) = onObserve fromSvr v >>= flip when k @@ -219,9 +218,10 @@ , Show (Observation o) , Observable o , Member (Logs LogMessage) q+ , HasCallStack ) => SchedulerProxy q -> Eff r (Server (CallbackObserver o))-spawnLoggingObserver px = spawnCallbackObserver+spawnLoggingObserver px = withFrozenCallStack $ spawnCallbackObserver px (\s o -> logDebug (show s ++ " OBSERVED: " ++ show o) >> return True)
+ src/Control/Eff/Concurrent/Api/Observer/Queue.hs view
@@ -0,0 +1,149 @@+-- | Capture 'Observation's and enqueue then into an STM 'TBQeueu'.+module Control.Eff.Concurrent.Api.Observer.Queue+ ( ObservationQueue()+ , ObservationQueueReader+ , readObservationQueue+ , tryReadObservationQueue+ , flushObservationQueue+ , enqueueObservationsRegistered+ , enqueueObservations+ )+where++import Control.Concurrent.STM+import Control.Eff+import Control.Eff.ExceptionExtra+import Control.Eff.Lift+import Control.Eff.Concurrent.Process+import Control.Eff.Log+import Control.Eff.Concurrent.Api+import Control.Eff.Concurrent.Api.Client+import Control.Eff.Concurrent.Api.Observer+import Control.Eff.Reader.Strict+import Control.Exception+import Control.Monad.IO.Class+import Data.Typeable+import GHC.Stack++-- | Contains a 'TBQueue' capturing observations received by 'enqueueObservationsRegistered'+-- or 'enqueueObservations'.+newtype ObservationQueue a = ObservationQueue (TBQueue (Observation a))++-- | A 'Reader' for an 'ObservationQueue'.+type ObservationQueueReader a = Reader (ObservationQueue a)++-- | 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))+ => Eff r (Observation o)+readObservationQueue = withFrozenCallStack $ do+ ObservationQueue q <- ask @(ObservationQueue o)+ liftIO (atomically (readTBQueue q))++-- | Read queued observations captured by observing a 'Server' that implements+-- an 'Observable' 'Api' using 'enqueueObservationsRegistered' or 'enqueueObservations'.+-- Return the next 'Observation' immediately or 'Nothing' if the queue is empty.+-- Use 'readObservationQueue' to block until an observation is observed.+tryReadObservationQueue+ :: forall o r+ . (Member (ObservationQueueReader o) r, HasCallStack, MonadIO (Eff r))+ => Eff r (Maybe (Observation o))+tryReadObservationQueue = withFrozenCallStack $ do+ ObservationQueue q <- ask @(ObservationQueue o)+ liftIO (atomically (tryReadTBQueue q))++-- | Read all currently queued 'Observation's captured by 'enqueueObservations'.+-- This returns immediately all currently enqueued 'Observation's. For a blocking+-- variant use 'readObservationQueue'.+flushObservationQueue+ :: forall o r+ . (Member (ObservationQueueReader o) r, HasCallStack, MonadIO (Eff r))+ => Eff r [Observation o]+flushObservationQueue = withFrozenCallStack $ do+ ObservationQueue q <- ask @(ObservationQueue o)+ liftIO (atomically (flushTBQueue q))++-- | Observe a(the) registered 'Server' that implements an 'Observable' 'Api'.+-- Based on 'enqueueObservations'.+enqueueObservationsRegistered+ :: forall o r q a+ . ( ServesApi o r q+ , SetMember Process (Process q) r+ , Typeable o+ , Show (Observation o)+ , Observable o+ , Member (Logs LogMessage) q+ , MonadIO (Eff (Process q ': q))+ , MonadIO (Eff r)+ , SetMember Lift (Lift IO) r+ , Member (Logs LogMessage) r+ , HasCallStack+ )+ => SchedulerProxy q+ -> Int+ -> Eff (ObservationQueueReader o ': r) a+ -> Eff r a+enqueueObservationsRegistered px queueLimit k = withFrozenCallStack $ do+ oSvr <- whereIsServer @o+ enqueueObservations px oSvr queueLimit k++-- | Observe a 'Server' that implements an 'Observable' 'Api', the 'Observation's+-- can be obtained by 'readObservationQueue'. All observations are captured up to+-- the queue size limit, such that the first message received will be first message+-- returned by 'readObservationQueue'.+--+-- This function captures runtime exceptions and cleans up accordingly.+enqueueObservations+ :: forall o r q a+ . ( SetMember Process (Process q) r+ , Typeable o+ , Show (Observation o)+ , Observable o+ , Member (Logs LogMessage) q+ , MonadIO (Eff (Process q ': q))+ , MonadIO (Eff r)+ , SetMember Lift (Lift IO) r+ , Member (Logs LogMessage) r+ , HasCallStack+ )+ => SchedulerProxy q+ -> Server o+ -> Int+ -> Eff (ObservationQueueReader o ': r) a+ -> Eff r a+enqueueObservations px oSvr queueLimit k = withFrozenCallStack $ withQueue+ queueLimit+ (do+ ObservationQueue q <- ask @(ObservationQueue o)+ do+ cbo <- spawnCallbackObserver+ px+ (\_from observeration -> do+ liftIO (atomically (writeTBQueue q observeration))+ return True+ )+ registerObserver SchedulerProxy cbo oSvr+ res <- k+ forgetObserver SchedulerProxy cbo oSvr+ sendShutdown px (_fromServer cbo)+ return res+ )++withQueue+ :: ( HasCallStack+ , MonadIO (Eff e)+ , Member (Logs LogMessage) e+ , 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))+ either (\em -> logError (show em) >> liftIO (throwIO em)) return res
src/Control/Eff/Concurrent/Api/Server.hs view
@@ -105,7 +105,7 @@ -> Api p ( 'Synchronous x) -> (x -> Eff r ()) -> Eff r ()-unhandledCallError px api _ = raiseError+unhandledCallError px api _ = withFrozenCallStack $ raiseError px ("Unhandled call: (" ++ show api ++ " :: " ++ show (typeRep (Proxy @p)) ++ ")" )@@ -122,7 +122,7 @@ => SchedulerProxy q -> Api p 'Asynchronous -> Eff r ()-unhandledCastError px api = raiseError+unhandledCastError px api = withFrozenCallStack $ raiseError px ("Unhandled cast: (" ++ show api ++ " :: " ++ show (typeRep (Proxy @p)) ++ ")" )@@ -135,7 +135,8 @@ => SchedulerProxy q -> Maybe String -> Eff r ()-defaultTermination px = maybe (exitNormally px) (exitWithError px)+defaultTermination px =+ withFrozenCallStack $ maybe (exitNormally px) (exitWithError px) -- | 'serve' two 'ApiHandler's at once. The first handler is used for@@ -224,7 +225,7 @@ -> (Dynamic -> Eff r a) -> Eff r a catchUnhandled effect handler =- effect `Exc.catchError` (handler . fromUnhandledRequest)+ withFrozenCallStack $ effect `Exc.catchError` (handler . fromUnhandledRequest) -- | Catch 'UnhandledRequest's and terminate the process with an error, if necessary. ensureAllHandled@@ -233,7 +234,7 @@ => SchedulerProxy q -> Eff (Exc.Exc UnhandledRequest ': r) () -> Eff r ()-ensureAllHandled px effect = do+ensureAllHandled px effect = withFrozenCallStack $ do result <- Exc.runError effect either (exitUnhandled px . fromUnhandledRequest) return result @@ -244,8 +245,10 @@ . (HasCallStack, Typeable a, Member (Exc.Exc UnhandledRequest) r) => Dynamic -> Eff r a-requestFromDynamic message =- maybe (Exc.throwError (UnhandledRequest message)) return (fromDynamic message)+requestFromDynamic message = withFrozenCallStack $ maybe+ (Exc.throwError (UnhandledRequest message))+ return+ (fromDynamic message) -- | If an incoming message could not be casted to a 'Request' corresponding to -- an 'ApiHandler' (e.g. with 'requestFromDynamic') one should use this function to@@ -256,6 +259,6 @@ => SchedulerProxy q -> Dynamic -> Eff r ()-exitUnhandled px message = do+exitUnhandled px message = withFrozenCallStack $ do let reason = "unhandled message: " ++ show message exitWithError px reason
src/Control/Eff/Concurrent/Process.hs view
@@ -142,7 +142,7 @@ . (SetMember Process (Process q) r, HasCallStack) => Process q (ResumeProcess v) -> Eff r v-executeAndResume processAction = do+executeAndResume processAction = withFrozenCallStack $ do result <- send processAction case result of ResumeWith !value -> return value@@ -157,7 +157,7 @@ => SchedulerProxy q -> Eff r (ResumeProcess v) -> Eff r (Either String v)-executeAndCatch px processAction = do+executeAndCatch px processAction = withFrozenCallStack $ do result <- processAction case result of ResumeWith !value -> return (Right value)@@ -186,7 +186,7 @@ . (HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> Eff r ()-yieldProcess _ = executeAndResume YieldProcess+yieldProcess _ = withFrozenCallStack $ executeAndResume YieldProcess -- | Send a message to a process addressed by the 'ProcessId'. -- See 'SendMessage'.@@ -197,7 +197,8 @@ -> ProcessId -> Dynamic -> Eff r ()-sendMessage px pid message = void (sendMessageChecked px pid message)+sendMessage px pid message =+ withFrozenCallStack $ void (sendMessageChecked px pid message) -- | Send a message to a process addressed by the 'ProcessId'. -- See 'SendMessage'. Return @True@ if the process existed.@@ -210,7 +211,7 @@ -> Dynamic -> Eff r Bool sendMessageChecked _ pid message =- executeAndResume (SendMessage pid $! message)+ withFrozenCallStack $ executeAndResume (SendMessage pid $! message) -- | Send a message to a process addressed by the 'ProcessId'. -- See 'SendMessage'.@@ -221,7 +222,7 @@ -> ProcessId -> o -> Eff r ()-sendMessageAs px pid = sendMessage px pid . toDyn+sendMessageAs px pid = withFrozenCallStack $ sendMessage px pid . toDyn -- | Exit a process addressed by the 'ProcessId'. -- See 'SendShutdown'.@@ -231,7 +232,7 @@ => SchedulerProxy q -> ProcessId -> Eff r ()-sendShutdown px pid = void (sendShutdownChecked px pid)+sendShutdown px pid = withFrozenCallStack $ void (sendShutdownChecked px pid) -- | Like 'sendShutdown', but also return @True@ iff the process to exit exists. sendShutdownChecked@@ -240,7 +241,8 @@ => SchedulerProxy q -> ProcessId -> Eff r Bool-sendShutdownChecked _ pid = executeAndResume (SendShutdown pid)+sendShutdownChecked _ pid =+ withFrozenCallStack $ executeAndResume (SendShutdown pid) -- | Start a new process, the new process will execute an effect, the function -- will return immediately with a 'ProcessId'.@@ -249,7 +251,7 @@ . (HasCallStack, SetMember Process (Process q) r) => Eff (Process q ': q) () -> Eff r ProcessId-spawn child = executeAndResume (Spawn @q child)+spawn child = withFrozenCallStack $ executeAndResume (Spawn @q child) -- | Like 'spawn' but return @()@. spawn_@@ -257,7 +259,7 @@ . (HasCallStack, SetMember Process (Process q) r) => Eff (Process q ': q) () -> Eff r ()-spawn_ = void . spawn+spawn_ = withFrozenCallStack $ void . spawn -- | Block until a message was received. receiveMessage@@ -265,7 +267,7 @@ . (HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> Eff r Dynamic-receiveMessage _ = executeAndResume ReceiveMessage+receiveMessage _ = withFrozenCallStack $ executeAndResume ReceiveMessage -- | Receive and cast the message to some 'Typeable' instance. receiveMessageAs@@ -273,7 +275,7 @@ . (HasCallStack, Typeable a, SetMember Process (Process q) r) => SchedulerProxy q -> Eff r a-receiveMessageAs px = do+receiveMessageAs px = withFrozenCallStack $ do messageDynamic <- receiveMessage px let castAndCheck dm = case fromDynamic dm of Nothing -> Left ("Invalid message type received: " ++ show dm)@@ -302,7 +304,7 @@ :: (HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> Eff r ProcessId-self _px = executeAndResume SelfPid+self _px = withFrozenCallStack $ executeAndResume SelfPid -- | Exit the process. exitNormally@@ -310,7 +312,7 @@ . (HasCallStack, SetMember Process (Process q) r) => SchedulerProxy q -> Eff r a-exitNormally _ = send (Shutdown @q)+exitNormally _ = withFrozenCallStack $ send (Shutdown @q) -- | Exit the process with an error. exitWithError@@ -319,7 +321,7 @@ => SchedulerProxy q -> String -> Eff r a-exitWithError _ = send . (ExitWithError @q $!)+exitWithError _ = withFrozenCallStack $ send . (ExitWithError @q $!) -- | Thrown an error, can be caught by 'catchRaisedError'. raiseError@@ -328,7 +330,7 @@ => SchedulerProxy q -> String -> Eff r b-raiseError _ = send . (RaiseError @q $!)+raiseError _ = withFrozenCallStack $ send . (RaiseError @q $!) -- | Catch and handle an error raised by 'raiseError'. Works independent of the -- handler implementation.@@ -339,7 +341,7 @@ -> (String -> Eff r w) -> Eff r w -> Eff r w-catchRaisedError _ onErr = interpose return go+catchRaisedError _ onErr = withFrozenCallStack $ interpose return go where go :: forall b . Process q b -> (b -> Eff r w) -> Eff r w go (RaiseError emsg) _k = onErr emsg@@ -352,7 +354,8 @@ => SchedulerProxy q -> Eff r a -> Eff r (Either String a)-ignoreProcessError px = catchRaisedError px (return . Left) . fmap Right+ignoreProcessError px =+ withFrozenCallStack $ 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
src/Control/Eff/Concurrent/Process/ForkIOScheduler.hs view
@@ -132,14 +132,15 @@ => Eff (ConsProcess SchedulerIO) () -> LogChannel LogMessage -> IO ()-schedule e logC = void $ withNewSchedulerState $ \schedulerStateVar -> do- pidVar <- newEmptyTMVarIO- scheduleProcessWithShutdownAction schedulerStateVar pidVar- $ interceptLogging (setLogMessageThreadId >=> logMsg)- $ do- logNotice "++++++++ main process started ++++++++"- e- logNotice "++++++++ main process returned ++++++++"+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 ++++++++" where withNewSchedulerState :: HasCallStack => (SchedulerVar -> IO a) -> IO a withNewSchedulerState mainProcessAction = do@@ -189,7 +190,7 @@ -- | Start the message passing concurrency system then execute a 'Process' on -- top of 'SchedulerIO' effect. All logging is sent to standard output. defaultMain :: HasCallStack => Eff (ConsProcess SchedulerIO) () -> IO ()-defaultMain c = runLoggingT+defaultMain c = withFrozenCallStack $ runLoggingT (logChannelBracket 128 (Just (infoMessage "main process started")) (schedule c)@@ -203,7 +204,8 @@ => LogChannel LogMessage -> Eff (ConsProcess SchedulerIO) () -> IO ()-defaultMainWithLogChannel logC c = closeLogChannelAfter logC (schedule c logC)+defaultMainWithLogChannel logC c =+ withFrozenCallStack $ closeLogChannelAfter logC (schedule c logC) scheduleProcessWithShutdownAction :: HasCallStack
src/Control/Eff/ExceptionExtra.hs view
@@ -3,9 +3,10 @@ {-# LANGUAGE MonoLocalBinds #-} {-# LANGUAGE DataKinds #-} {-# LANGUAGE TypeOperators #-}--- | Add-ons to 'Control.Eff.Exception'+-- | Add-ons to 'Control.Eff.Exception' and 'Control.Excepion' module Control.Eff.ExceptionExtra ( liftTry+ , maybeThrow , module X ) where@@ -23,3 +24,9 @@ => Eff r a -> Eff r (Either e a) liftTry m = (Right <$> m) `catchDynE` (return . Left)+++-- | Very similar to 'X.liftEither' but for 'Maybe's. Unlike 'X.liftMaybe' this+-- will throw the given value (instead of using 'X.Fail').+maybeThrow :: Member (X.Exc x) e => x -> Maybe a -> Eff e a+maybeThrow x = X.liftEither . maybe (Left x) Right