diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,5 +1,13 @@ # Changelog for extensible-effects-concurrent +## 0.28.0 +- Simplify `Protocol.Observer` registration API +- Rewrite `Protocol.Observer.Queue` API +- Add the `ProcessId` to the `ProcessDown` message + +## 0.27.1 +- Introduce `HasProcesses` and `HasSafeProcesses` everywhere + ## 0.27.0 - Improve/fix `EmbedProtocol` type class - Add a _this_ like parameter to the methods of `EffectfulServer` 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 @@ -41,7 +41,7 @@ lift (threadDelay 500000) o <- logCounterObservations lift (threadDelay 500000) - registerObserver o c + registerObserver @CounterChanged c o lift (threadDelay 500000) cast c Inc lift (threadDelay 500000) @@ -92,13 +92,13 @@ type instance Model SupiCounter = ( Integer - , Observers CounterChanged + , ObserverRegistry CounterChanged , Maybe (ReplyTarget SupiCounter (Maybe ())) ) data instance StartArgument SupiCounter (Processes q) = MkEmptySupiCounter - setup _ _ = return ((0, emptyObservers, Nothing), ()) + setup _ _ = return ((0, emptyObserverRegistry, Nothing), ()) update _ _ = \case OnCall rt callReq -> @@ -113,14 +113,20 @@ case castReq of ToPdu1 Inc -> do val' <- view _1 <$> modifyAndGetModel @SupiCounter (_1 %~ (+ 1)) - zoomModel @SupiCounter _2 (observed (CounterChanged val')) + zoomModel @SupiCounter _2 (observerRegistryNotify (CounterChanged val')) when (val' > 5) $ getAndModifyModel @SupiCounter (_3 .~ Nothing) >>= traverse_ (\rt' -> sendReply rt' (Just ())) . view _3 ToPdu2 x -> - zoomModel @SupiCounter _2 (handleObserverRegistration x) + zoomModel @SupiCounter _2 (observerRegistryHandlePdu x) ToPdu3 _ -> error "unreachable" + OnDown pd -> do + wasRemoved <- zoomModel @SupiCounter _2 (observerRegistryRemoveProcess @CounterChanged (downProcess pd)) + if wasRemoved + then logDebug ("removed: " <> T.pack (show pd)) + else logError ("unexpected: " <> T.pack (show pd)) + other -> logWarning (T.pack (show other)) spawnCounter :: (LogIo q) => Eff (Processes q) ( Endpoint SupiCounter ) @@ -131,16 +137,12 @@ logCounterObservations :: (LogIo q, Typeable q) - => Eff (Processes q) (Observer CounterChanged) -logCounterObservations = do - svr <- start OCCStart - pure (toObserver svr) + => Eff (Processes q) (Endpoint (Observer CounterChanged)) +logCounterObservations = start OCCStart instance Member Logs q => Server (Observer CounterChanged) (Processes q) where data StartArgument (Observer CounterChanged) (Processes q) = OCCStart - type Model (Observer CounterChanged) = Observers CounterChanged - setup _ _ = pure (emptyObservers, ()) - update _ _ = - \case - OnCast r -> handleObservations (\msg -> logInfo' ("observed: " ++ show msg)) r - wtf -> logNotice (T.pack (show wtf)) + update _ _ e = + case e of + OnCast (Observed msg) -> logInfo' ("observerRegistryNotify: " ++ show msg) + _ -> logNotice (T.pack (show e)) diff --git a/extensible-effects-concurrent.cabal b/extensible-effects-concurrent.cabal --- a/extensible-effects-concurrent.cabal +++ b/extensible-effects-concurrent.cabal @@ -1,6 +1,6 @@ cabal-version: 2.0 name: extensible-effects-concurrent -version: 0.27.0 +version: 0.28.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 @@ -99,6 +99,7 @@ Control.Eff.Concurrent.Protocol.Observer Control.Eff.Concurrent.Protocol.Observer.Queue other-modules: + Control.Eff.Concurrent.Misc, Control.Eff.Concurrent.Protocol.Supervisor.InternalState, Paths_extensible_effects_concurrent default-extensions: @@ -276,6 +277,7 @@ , LoggingTests , LogMessageIdeaTest , LoopTests + , ObserverTests , ProcessBehaviourTestCases , SupervisorTests , SingleThreadedScheduler 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 @@ -37,9 +37,6 @@ -- ** /Observer/ Functions for Events and Event Listener module Control.Eff.Concurrent.Protocol.Observer , - -- *** Capture /Observation/ in a FIFO Queue - module Control.Eff.Concurrent.Protocol.Observer.Queue - , -- * Utilities -- ** Logging Effect module Control.Eff.Log @@ -74,7 +71,12 @@ ) where -import Control.Eff.Concurrent.Process ( Process(..) +import Control.Eff.Concurrent.Process + ( Process(..) + , SafeProcesses + , Processes + , HasProcesses + , HasSafeProcesses , ProcessTitle(..) , fromProcessTitle , ProcessDetails(..) @@ -86,73 +88,58 @@ , Serializer(..) , ProcessId(..) , fromProcessId - , SafeProcesses - , ResumeProcess(..) - , ProcessState(..) , yieldProcess , sendMessage , sendAnyMessage - , sendShutdown - , sendInterrupt , makeReference - , getProcessState - , updateProcessDetails - , receiveAnyMessage , receiveMessage , receiveSelectedMessage , flushMessages - , receiveAnyLoop + , receiveAnyMessage , receiveLoop , receiveSelectedLoop - , MessageSelector - ( runMessageSelector - ) - , selectMessage + , receiveAnyLoop + , MessageSelector(runMessageSelector) , selectMessage , filterMessage - , filterMessage , selectMessageWith - , selectMessageWith , selectDynamicMessage - , selectDynamicMessage , selectAnyMessage , self , isProcessAlive + , getProcessState + , updateProcessDetails , spawn , spawn_ , spawnLink , spawnRaw , spawnRaw_ + , ResumeProcess(..) + , executeAndResume + , executeAndResumeOrExit + , executeAndResumeOrThrow + , interrupt + , sendInterrupt , exitBecause , exitNormally , exitWithError + , sendShutdown -- TODO rename to 'sendExit' , linkProcess , unlinkProcess , monitor , demonitor , ProcessDown(..) , selectProcessDown + , selectProcessDownByProcessId , becauseProcessIsDown , MonitorReference(..) , withMonitor , receiveWithMonitor - , provideInterruptsShutdown - , handleInterrupts - , tryUninterrupted - , exitOnInterrupt - , logInterrupts - , provideInterrupts - , mergeEitherInterruptAndExitReason - , interrupt - , executeAndResume - , executeAndResumeOrExit - , executeAndResumeOrThrow , Interrupt(..) + , Interrupts , interruptToExit , ExitRecovery(..) , RecoverableInterrupt - , Interrupts - , Processes , ExitSeverity(..) , SomeExitReason(SomeExitReason) , toExitRecovery @@ -163,6 +150,13 @@ , toCrashReason , fromSomeExitReason , logProcessExit + , provideInterruptsShutdown + , handleInterrupts + , tryUninterrupted + , exitOnInterrupt + , logInterrupts + , provideInterrupts + , mergeEitherInterruptAndExitReason ) import Control.Eff.Concurrent.Process.Timer ( Timeout(fromTimeoutMicros) @@ -207,32 +201,20 @@ ) import Control.Eff.Concurrent.Protocol.Observer ( Observer(..) - , Pdu - ( RegisterObserver - , ForgetObserver - , Observed - ) + , ObservationSink + , IsObservable + , CanObserve + , Pdu(RegisterObserver, ForgetObserver, Observed) , registerObserver , forgetObserver - , handleObservations - , toObserver - , toObserverFor - , ObserverRegistry - , ObserverState - , Observers() - , emptyObservers - , handleObserverRegistration - , manageObservers - , observed - ) -import Control.Eff.Concurrent.Protocol.Observer.Queue - ( ObservationQueue() - , ObservationQueueReader - , readObservationQueue - , tryReadObservationQueue - , flushObservationQueue - , withObservationQueue - , spawnLinkObservationQueueWriter + , forgetObserverUnsafe + , ObserverRegistry(..) + , ObserverRegistryState + , evalObserverRegistryState + , emptyObserverRegistry + , observerRegistryHandlePdu + , observerRegistryRemoveProcess + , observerRegistryNotify ) import Control.Eff.Concurrent.Protocol.Wrapper ( Request(..) diff --git a/src/Control/Eff/Concurrent/Misc.hs b/src/Control/Eff/Concurrent/Misc.hs new file mode 100644 --- /dev/null +++ b/src/Control/Eff/Concurrent/Misc.hs @@ -0,0 +1,49 @@ +-- | Internal module containing internal helpers that didn't +-- make it into their own library. +module Control.Eff.Concurrent.Misc + ( showSTypeRepPrec + , showSTypeRep + , showSTypeable + , showSPrecTypeable + ) + where + +import Data.Dynamic +import Data.Typeable () +import Type.Reflection + +-- | Render a 'Typeable' to a 'ShowS'. +-- +-- @since 0.28.0 +showSTypeable :: forall message . Typeable message => ShowS +showSTypeable = showSTypeRep (SomeTypeRep (typeRep @message)) + +-- | Render a 'Typeable' to a 'ShowS' with a precedence parameter. +-- +-- @since 0.28.0 +showSPrecTypeable :: forall message . Typeable message => Int -> ShowS +showSPrecTypeable d = showSTypeRepPrec d (SomeTypeRep (typeRep @message)) + +-- | This is equivalent to @'showSTypeRepPrec' 0@ +-- +-- @since 0.24.0 +showSTypeRep :: SomeTypeRep -> ShowS +showSTypeRep = showSTypeRepPrec 0 + +-- | An internal utility to print 'Typeable' without the kinds. +-- This is like 'showsPrec' in that it accepts a /precedence/ parameter, +-- and the result is in parentheses when the precedence is higher than 9. +-- +-- @since 0.24.0 +showSTypeRepPrec :: Int -> SomeTypeRep -> ShowS +showSTypeRepPrec d (SomeTypeRep tr) sIn = + let (con, conArgs) = splitApps tr + in case conArgs of + [] -> showString (tyConName con) sIn + _ -> + showParen + (d >= 10) + (showString (tyConName con) . showChar ':' . + foldr1 (\f acc -> showChar '-' . f . acc) + (showSTypeRepPrec 10 <$> conArgs)) + sIn 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 @@ -13,9 +13,14 @@ -- "Control.Eff.Concurrent.Process.SingleThreadedScheduler" module Control.Eff.Concurrent.Process ( -- * Process Effect - -- ** Effect Type Handling Process(..) - -- ** Process Info + -- ** Process Effect Aliases + , SafeProcesses + , Processes + -- ** Process Effect Contraints + , HasProcesses + , HasSafeProcesses + -- ** Process Info , ProcessTitle(..) , fromProcessTitle , ProcessDetails(..) @@ -27,12 +32,9 @@ , unwrapStrictDynamic , Serializer(..) - -- ** ProcessId Type , ProcessId(..) , fromProcessId - , SafeProcesses - , ResumeProcess(..) -- ** Process State , ProcessState(..) -- ** Yielding @@ -40,8 +42,6 @@ -- ** Sending Messages , sendMessage , sendAnyMessage - , sendShutdown - , sendInterrupt -- ** Utilities , makeReference -- ** Receiving Messages @@ -59,7 +59,7 @@ , selectMessageWith , selectDynamicMessage , selectAnyMessage - -- ** Process Life Cycle Management + -- ** Process State Reflection , self , isProcessAlive , getProcessState @@ -70,42 +70,39 @@ , spawnLink , spawnRaw , spawnRaw_ - -- ** Process Exit or Interrupt Recoverable + -- ** Process Operation Execution + , ResumeProcess(..) + , executeAndResume + , executeAndResumeOrExit + , executeAndResumeOrThrow + -- ** Exits and Interrupts + -- *** Interrupting Processes + , interrupt + , sendInterrupt + -- *** Exitting Processes , exitBecause , exitNormally , exitWithError - -- ** Links + , sendShutdown -- TODO rename to 'sendExit' + -- *** Linking Processes , linkProcess , unlinkProcess - -- ** Monitors + -- *** Monitor Processes , monitor , demonitor , ProcessDown(..) , selectProcessDown + , selectProcessDownByProcessId , becauseProcessIsDown , MonitorReference(..) , withMonitor , receiveWithMonitor - -- ** Process Interrupt Recoverable Handling - , provideInterruptsShutdown - , handleInterrupts - , tryUninterrupted - , exitOnInterrupt - , logInterrupts - , provideInterrupts - , mergeEitherInterruptAndExitReason - , interrupt - -- ** Process Operation Execution - , executeAndResume - , executeAndResumeOrExit - , executeAndResumeOrThrow - -- ** Exit Or Interrupt Recoverable Reasons + -- *** Exit and Interrupt Reasons , Interrupt(..) + , Interrupts , interruptToExit , ExitRecovery(..) , RecoverableInterrupt - , Interrupts - , Processes , ExitSeverity(..) , SomeExitReason(SomeExitReason) , toExitRecovery @@ -116,10 +113,20 @@ , toCrashReason , fromSomeExitReason , logProcessExit + -- ** Control Flow + -- *** Process Interrupt Recoverable Handling + , provideInterruptsShutdown + , handleInterrupts + , tryUninterrupted + , exitOnInterrupt + , logInterrupts + , provideInterrupts + , mergeEitherInterruptAndExitReason ) where import Control.Applicative +import Control.Eff.Concurrent.Misc import Control.DeepSeq import Control.Eff import Control.Eff.Exception @@ -341,6 +348,12 @@ { runSerializer :: message -> StrictDynamic } deriving (Typeable) +instance NFData (Serializer message) where + rnf (MkSerializer !s) = s `seq` () + +instance Typeable message => Show (Serializer message) where + showsPrec d _x = showParen (d >= 10) (showSTypeable @message . showString "-serializer") + instance Contravariant Serializer where contramap f (MkSerializer b) = MkSerializer (b . f) @@ -646,10 +659,32 @@ -- | This adds a layer of the 'Interrupts' effect on top of 'Processes' type Processes e = Interrupts ': SafeProcesses e +-- | A constraint for an effect set that requires the presence of 'Processes'. +-- +-- This constrains the effect list to look like this: +-- @[e1 ... eN, 'Interrupts', 'Process' [e(N+1) .. e(N+k)], e(N+1) .. e(N+k)]@ +-- +-- It contrains @e@ beyond 'HasSafeProcesses' to encompass 'Interrupts'. +-- +-- @since 0.27.1 +type HasProcesses e inner = (HasSafeProcesses e inner, Member Interrupts e) + -- | /Cons/ 'Process' onto a list of effects. This is called @SafeProcesses@ because -- the the actions cannot be interrupted in. type SafeProcesses r = Process r ': r +-- | A constraint for an effect set that requires the presence of 'SafeProcesses'. +-- +-- This constrains the effect list to look like this: +-- @[e1 ... eN, 'Process' [e(N+1) .. e(N+k)], e(N+1) .. e(N+k)]@ +-- +-- It contrains @e@ to support the (only) 'Process' effect. +-- +-- This is more relaxed that 'HasProcesses' since it does not require 'Interrupts'. +-- +-- @since 0.27.1 +type HasSafeProcesses e inner = (SetMember Process (Process inner) e) + -- | 'Exc'eptions containing 'Interrupt's. -- See 'handleInterrupts', 'exitOnInterrupt' or 'provideInterrupts' type Interrupts = Exc (Interrupt 'Recoverable) @@ -697,7 +732,7 @@ -- when a linked process crashes while we wait in a 'receiveSelectedMessage' -- via a call to 'interrupt'. exitOnInterrupt - :: (HasCallStack, Member Interrupts r, SetMember Process (Process q) r) + :: (HasCallStack, HasProcesses r q) => Eff r a -> Eff r a exitOnInterrupt = handleInterrupts (exitBecause . interruptToExit) @@ -793,7 +828,7 @@ -- whenever the exit reason satisfies 'isRecoverable', return the exit reason. executeAndResume :: forall q r v - . (SetMember Process (Process q) r, HasCallStack) + . (HasSafeProcesses r q, HasCallStack) => Process q (ResumeProcess v) -> Eff r (Either (Interrupt 'Recoverable) v) executeAndResume processAction = do @@ -807,7 +842,7 @@ -- interrupts. executeAndResumeOrExit :: forall r q v - . (SetMember Process (Process q) r, HasCallStack) + . (HasSafeProcesses r q, HasCallStack) => Process q (ResumeProcess v) -> Eff r v executeAndResumeOrExit processAction = do @@ -821,7 +856,7 @@ -- interrupts. executeAndResumeOrThrow :: forall q r v - . (SetMember Process (Process q) r, HasCallStack, Member Interrupts r) + . (HasProcesses r q, HasCallStack) => Process q (ResumeProcess v) -> Eff r v executeAndResumeOrThrow processAction = do @@ -836,7 +871,7 @@ -- for more information. yieldProcess :: forall r q - . (SetMember Process (Process q) r, HasCallStack, Member Interrupts r) + . (HasProcesses r q, HasCallStack) => Eff r () yieldProcess = executeAndResumeOrThrow YieldProcess @@ -846,9 +881,8 @@ -- The message will be reduced to normal form ('rnf') by/in the caller process. sendMessage :: forall r q o - . ( SetMember Process (Process q) r + . ( HasProcesses r q , HasCallStack - , Member Interrupts r , Typeable o , NFData o ) @@ -863,7 +897,7 @@ -- See 'SendMessage'. sendAnyMessage :: forall r q - . (SetMember Process (Process q) r, HasCallStack, Member Interrupts r) + . (HasCallStack, HasProcesses r q) => ProcessId -> StrictDynamic -> Eff r () @@ -875,7 +909,7 @@ -- See 'SendShutdown'. sendShutdown :: forall r q - . (SetMember Process (Process q) r, HasCallStack, Member Interrupts r) + . (HasCallStack, HasProcesses r q) => ProcessId -> Interrupt 'NoRecovery -> Eff r () @@ -887,7 +921,7 @@ -- | Like 'sendInterrupt', but also return @True@ iff the process to exit exists. sendInterrupt :: forall r q - . (SetMember Process (Process q) r, HasCallStack, Member Interrupts r) + . (HasCallStack, HasProcesses r q) => ProcessId -> Interrupt 'Recoverable -> Eff r () @@ -901,7 +935,7 @@ -- 'spawnRaw'. spawn :: forall r q - . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r) + . (HasCallStack, HasProcesses r q) => ProcessTitle -> Eff (Processes q) () -> Eff r ProcessId @@ -911,7 +945,7 @@ -- | Like 'spawn' but return @()@. spawn_ :: forall r q - . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r) + . (HasCallStack, HasProcesses r q) => ProcessTitle -> Eff (Processes q) () -> Eff r () @@ -922,7 +956,7 @@ -- @since 0.12.0 spawnLink :: forall r q - . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r) + . (HasCallStack, HasProcesses r q) => ProcessTitle -> Eff (Processes q) () -> Eff r ProcessId @@ -935,7 +969,7 @@ -- suited. spawnRaw :: forall r q - . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r) + . (HasCallStack, HasProcesses r q) => ProcessTitle -> Eff (SafeProcesses q) () -> Eff r ProcessId @@ -944,7 +978,7 @@ -- | Like 'spawnRaw' but return @()@. spawnRaw_ :: forall r q - . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r) + . (HasCallStack, HasProcesses r q) => ProcessTitle -> Eff (SafeProcesses q) () -> Eff r () @@ -955,7 +989,7 @@ -- @since 0.12.0 isProcessAlive :: forall r q - . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r) + . (HasCallStack, HasProcesses r q) => ProcessId -> Eff r Bool isProcessAlive pid = isJust <$> executeAndResumeOrThrow (GetProcessState pid) @@ -967,7 +1001,7 @@ -- @since 0.24.1 getProcessState :: forall r q - . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r) + . (HasCallStack, HasProcesses r q) => ProcessId -> Eff r (Maybe (ProcessTitle, ProcessDetails, ProcessState)) getProcessState pid = executeAndResumeOrThrow (GetProcessState pid) @@ -977,7 +1011,7 @@ -- @since 0.24.1 updateProcessDetails :: forall r q - . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r) + . (HasCallStack, HasProcesses r q) => ProcessDetails -> Eff r () updateProcessDetails pd = executeAndResumeOrThrow (UpdateProcessDetails pd) @@ -986,7 +1020,7 @@ -- See 'ReceiveSelectedMessage' for more documentation. receiveAnyMessage :: forall r q - . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r) + . (HasCallStack, HasProcesses r q) => Eff r StrictDynamic receiveAnyMessage = executeAndResumeOrThrow (ReceiveSelectedMessage selectAnyMessage) @@ -998,8 +1032,7 @@ :: forall r q a . ( HasCallStack , Show a - , SetMember Process (Process q) r - , Member Interrupts r + , HasProcesses r q ) => MessageSelector a -> Eff r a @@ -1014,8 +1047,7 @@ , Typeable a , NFData a , Show a - , SetMember Process (Process q) r - , Member Interrupts r + , HasProcesses r q ) => Eff r a receiveMessage = receiveSelectedMessage (MessageSelector fromStrictDynamic) @@ -1026,7 +1058,7 @@ -- @since 0.12.0 flushMessages :: forall r q - . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r) + . (HasCallStack, HasProcesses r q) => Eff r [StrictDynamic] flushMessages = executeAndResumeOrThrow @q FlushMessages @@ -1040,7 +1072,7 @@ -- See also 'ReceiveSelectedMessage' for more documentation. receiveSelectedLoop :: forall r q a endOfLoopResult - . (SetMember Process (Process q) r, HasCallStack) + . (HasSafeProcesses r q, HasCallStack) => MessageSelector a -> (Either (Interrupt 'Recoverable) a -> Eff r (Maybe endOfLoopResult)) -> Eff r endOfLoopResult @@ -1055,7 +1087,7 @@ -- See also 'selectAnyMessage', 'receiveSelectedLoop'. receiveAnyLoop :: forall r q endOfLoopResult - . (SetMember Process (Process q) r, HasCallStack) + . (HasSafeProcesses r q, HasCallStack) => (Either (Interrupt 'Recoverable) StrictDynamic -> Eff r (Maybe endOfLoopResult)) -> Eff r endOfLoopResult receiveAnyLoop = receiveSelectedLoop selectAnyMessage @@ -1064,18 +1096,18 @@ -- using 'selectMessage'. receiveLoop :: forall r q a endOfLoopResult - . (SetMember Process (Process q) r, HasCallStack, NFData a, Typeable a) + . (HasSafeProcesses r q, HasCallStack, NFData a, Typeable a) => (Either (Interrupt 'Recoverable) a -> Eff r (Maybe endOfLoopResult)) -> Eff r endOfLoopResult receiveLoop = receiveSelectedLoop selectMessage -- | Returns the 'ProcessId' of the current process. -self :: (HasCallStack, SetMember Process (Process q) r) => Eff r ProcessId +self :: (HasCallStack, HasSafeProcesses r q) => Eff r ProcessId self = executeAndResumeOrExit SelfPid -- | Generate a unique 'Int' for the current process. makeReference - :: (HasCallStack, SetMember Process (Process q) r, Member Interrupts r) + :: (HasCallStack, HasProcesses r q) => Eff r Int makeReference = executeAndResumeOrThrow MakeReference @@ -1109,7 +1141,7 @@ -- @since 0.12.0 monitor :: forall r q - . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r) + . (HasCallStack, HasProcesses r q) => ProcessId -> Eff r MonitorReference monitor = executeAndResumeOrThrow . Monitor . force @@ -1119,7 +1151,7 @@ -- @since 0.12.0 demonitor :: forall r q - . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r) + . (HasCallStack, HasProcesses r q) => MonitorReference -> Eff r () demonitor = executeAndResumeOrThrow . Demonitor . force @@ -1129,11 +1161,7 @@ -- -- @since 0.12.0 withMonitor - :: ( HasCallStack - , Member Interrupts r - , SetMember Process (Process q) r - , Member Interrupts r - ) + :: (HasCallStack, HasProcesses r q) => ProcessId -> (MonitorReference -> Eff r a) -> Eff r a @@ -1145,9 +1173,7 @@ -- @since 0.12.0 receiveWithMonitor :: ( HasCallStack - , Member Interrupts r - , SetMember Process (Process q) r - , Member Interrupts r + , HasProcesses r q , Typeable a , Show a ) @@ -1169,6 +1195,7 @@ ProcessDown { downReference :: !MonitorReference , downReason :: !SomeExitReason + , downProcess :: !ProcessId } deriving (Typeable, Generic, Eq, Ord) @@ -1186,21 +1213,35 @@ showsPrec d = showParen (d >= 10) . (\case - ProcessDown ref reason -> - showString "monitored process down " - . showsPrec 11 ref + ProcessDown ref reason pid -> + showString "down: " + . shows pid . showChar ' ' + . shows ref + . showChar ' ' . showsPrec 11 reason ) -- | A 'MessageSelector' for the 'ProcessDown' message of a specific -- process. -- +-- The parameter is the value obtained by 'monitor'. +-- -- @since 0.12.0 selectProcessDown :: MonitorReference -> MessageSelector ProcessDown selectProcessDown ref0 = - filterMessage (\(ProcessDown ref _reason) -> ref0 == ref) + filterMessage (\(ProcessDown ref _reason _pid) -> ref0 == ref) +-- | A 'MessageSelector' for the 'ProcessDown' message. of a specific +-- process. +-- +-- In contrast to 'selectProcessDown' this function matches the 'ProcessId'. +-- +-- @since 0.28.0 +selectProcessDownByProcessId :: ProcessId -> MessageSelector ProcessDown +selectProcessDownByProcessId pid0 = + filterMessage (\(ProcessDown _ref _reason pid) -> pid0 == pid) + -- | Connect the calling process to another process, such that -- if one of the processes crashes (i.e. 'isCrash' returns 'True'), the other -- is shutdown with the 'Interrupt' 'LinkedProcessCrashed'. @@ -1208,7 +1249,7 @@ -- @since 0.12.0 linkProcess :: forall r q - . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r) + . (HasCallStack, HasProcesses r q) => ProcessId -> Eff r () linkProcess = executeAndResumeOrThrow . Link . force @@ -1218,7 +1259,7 @@ -- @since 0.12.0 unlinkProcess :: forall r q - . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r) + . (HasCallStack, HasProcesses r q) => ProcessId -> Eff r () unlinkProcess = executeAndResumeOrThrow . Unlink . force @@ -1226,20 +1267,20 @@ -- | Exit the process with a 'Interrupt'. exitBecause :: forall r q a - . (HasCallStack, SetMember Process (Process q) r) + . (HasCallStack, HasSafeProcesses r q) => Interrupt 'NoRecovery -> Eff r a exitBecause = send . Shutdown @q . force -- | Exit the process. exitNormally - :: forall r q a . (HasCallStack, SetMember Process (Process q) r) => Eff r a + :: forall r q a . (HasCallStack, HasSafeProcesses r q) => Eff r a exitNormally = exitBecause ExitNormally -- | Exit the process with an error. exitWithError :: forall r q a - . (HasCallStack, SetMember Process (Process q) r) + . (HasCallStack,HasSafeProcesses r q) => String -> Eff r a exitWithError = exitBecause . interruptToExit . ErrorInterrupt 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 @@ -131,7 +131,7 @@ -- | Add monitor: If the process is dead, enqueue a 'ProcessDown' message into the -- owners message queue addMonitoring - :: ProcessId -> ProcessId -> SchedulerState -> STM MonitorReference + :: HasCallStack => ProcessId -> ProcessId -> SchedulerState -> STM MonitorReference addMonitoring target owner schedulerState = do aNewMonitorIndex <- readTVar (schedulerState ^. nextMonitorIndex) modifyTVar' (schedulerState ^. nextMonitorIndex) (+ 1) @@ -143,13 +143,13 @@ (Set.insert (monitorRef, owner)) else let processDownMessage = - ProcessDown monitorRef (SomeExitReason (OtherProcessNotRunning target)) + ProcessDown monitorRef (SomeExitReason (OtherProcessNotRunning target)) target in enqueueMessageOtherProcess owner (toStrictDynamic processDownMessage) schedulerState return monitorRef -removeMonitoring :: MonitorReference -> SchedulerState -> STM () +removeMonitoring :: HasCallStack => MonitorReference -> SchedulerState -> STM () removeMonitoring monitorRef schedulerState = modifyTVar' (schedulerState ^. processMonitors) (Set.filter (\(ref, _) -> ref /= monitorRef)) @@ -163,7 +163,7 @@ go (mr, owner) = when (monitoredProcess mr == downPid) (do - let processDownMessage = ProcessDown mr reason + let processDownMessage = ProcessDown mr reason downPid enqueueMessageOtherProcess owner (toStrictDynamic processDownMessage) schedulerState removeMonitoring mr schedulerState ) @@ -191,8 +191,7 @@ -- | 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 BaseEffects () -> Eff LoggingAndIo () +withNewSchedulerState :: HasCallStack => Eff BaseEffects () -> Eff LoggingAndIo () withNewSchedulerState mainProcessAction = Safe.bracketWithError (lift (atomically newSchedulerState)) (\exceptions schedulerState -> do 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 @@ -154,7 +154,7 @@ (\res -> Just (res, acc Seq.>< msgRest)) (runMessageSelector messageSelector m) --- | Add monitor: If the process is dead, enqueue a ProcessDown message into the +-- | Add monitor: If the process is dead, enqueue a 'ProcessDown' message into the -- owners message queue addMonitoring :: ProcessId -> ProcessId -> STS m r -> (MonitorReference, STS m r) @@ -166,7 +166,7 @@ pt <- use msgQs if Map.member target pt then monitors %= Set.insert (mref, owner) - else let pdown = ProcessDown mref (SomeExitReason (OtherProcessNotRunning target)) + else let pdown = ProcessDown mref (SomeExitReason (OtherProcessNotRunning target)) target in State.modify' (enqueueMsg owner (toStrictDynamic pdown)) return mref @@ -180,7 +180,7 @@ where go (mr, owner) = when (monitoredProcess mr == downPid) - (let pdown = ProcessDown mr reason + (let pdown = ProcessDown mr reason downPid in State.modify' (enqueueMsg owner (toStrictDynamic pdown) . removeMonitoring mr) ) diff --git a/src/Control/Eff/Concurrent/Process/Timer.hs b/src/Control/Eff/Concurrent/Process/Timer.hs --- a/src/Control/Eff/Concurrent/Process/Timer.hs +++ b/src/Control/Eff/Concurrent/Process/Timer.hs @@ -39,8 +39,7 @@ :: forall a r q . ( Lifted IO q , HasCallStack - , SetMember Process (Process q) r - , Member Interrupts r + , HasProcesses r q , Typeable a , NFData a , Show a @@ -59,8 +58,7 @@ :: forall a r q . ( Lifted IO q , HasCallStack - , SetMember Process (Process q) r - , Member Interrupts r + , HasProcesses r q , Show a ) => MessageSelector a @@ -80,8 +78,7 @@ :: forall a r q . ( Lifted IO q , HasCallStack - , SetMember Process (Process q) r - , Member Interrupts r + , HasProcesses r q , Show a ) => ProcessId @@ -150,8 +147,7 @@ :: forall r q message . ( Lifted IO q , HasCallStack - , SetMember Process (Process q) r - , Member Interrupts r + , HasProcesses r q , Typeable message , NFData message ) @@ -179,8 +175,7 @@ :: forall r q . ( Lifted IO q , HasCallStack - , SetMember Process (Process q) r - , Member Interrupts r + , HasProcesses r q ) => Timeout -> Eff r TimerReference @@ -195,8 +190,7 @@ :: forall r q . ( Lifted IO q , HasCallStack - , SetMember Process (Process q) r - , Member Interrupts r + , HasProcesses r q ) => TimerReference -> Eff r () diff --git a/src/Control/Eff/Concurrent/Protocol.hs b/src/Control/Eff/Concurrent/Protocol.hs --- a/src/Control/Eff/Concurrent/Protocol.hs +++ b/src/Control/Eff/Concurrent/Protocol.hs @@ -47,11 +47,10 @@ , EmbedProtocol(..) , toEmbeddedEndpoint , fromEmbeddedEndpoint - , prettyTypeableShows - , prettyTypeableShowsPrec ) where +import Control.Eff.Concurrent.Misc import Control.DeepSeq import Control.Eff.Concurrent.Process import Control.Lens @@ -63,8 +62,28 @@ import Type.Reflection --- | This data family defines the __protocol data units__ (PDU) of a /protocol/. +-- | A server process for /protocol/. -- +-- Protocols are represented by phantom types, which are used in different places to +-- index type families and type class instances. +-- +-- A 'Process' can send and receive any messages. An 'Endpoint' +-- wraps around a 'ProcessId' and carries a phantom type to indicate +-- the kinds of messages accepted by the process. +-- +-- As a metaphor, communication between processes can be thought of waiting for +-- and sending __protocol data units__ belonging to some protocol. +newtype Endpoint protocol = Endpoint { _fromEndpoint :: ProcessId } + deriving (Eq,Ord,Typeable, NFData) + +instance Typeable protocol => Show (Endpoint protocol) where + showsPrec d (Endpoint c) = + showParen (d>=10) + (showSTypeRep (SomeTypeRep (typeRep @protocol)) . showsPrec 10 c) + +-- | This type class and the associated data family defines the +-- __protocol data units__ (PDU) of a /protocol/. +-- -- A Protocol in the sense of a communication interface description -- between processes. -- @@ -80,7 +99,7 @@ -- > -- > data BookShop deriving Typeable -- > --- > instance HasPdu BookShop r where +-- > instance Typeable r => HasPdu BookShop r where -- > data instance Pdu BookShop r where -- > RentBook :: BookId -> Pdu BookShop ('Synchronous (Either RentalError RentalId)) -- > BringBack :: RentalId -> Pdu BookShop 'Asynchronous @@ -92,7 +111,7 @@ -- > -- -- @since 0.25.1 -class (NFData (Pdu protocol reply), Show (Pdu protocol reply), Typeable protocol, Typeable reply) => HasPdu (protocol :: Type) (reply :: Synchronicity) where +class (Tangible (Pdu protocol reply), Typeable protocol, Typeable reply) => HasPdu (protocol :: Type) (reply :: Synchronicity) where -- | The __protocol data unit__ type for the given protocol. data family Pdu protocol reply @@ -134,6 +153,7 @@ ( Typeable p , Typeable r , Tangible (Pdu p r) + , HasPdu p r ) -- | The (promoted) constructors of this type specify (at the type level) the @@ -154,45 +174,11 @@ ProtocolReply ('Synchronous t) = t ProtocolReply 'Asynchronous = () --- | This is a tag-type that wraps around a 'ProcessId' and holds an 'Pdu' index --- type. -newtype Endpoint protocol = Endpoint { _fromEndpoint :: ProcessId } - deriving (Eq,Ord,Typeable, NFData) - -instance Typeable protocol => Show (Endpoint protocol) where - showsPrec d (Endpoint c) = - showParen (d>=10) - (prettyTypeableShows (SomeTypeRep (typeRep @protocol)) . showsPrec 10 c) - --- | This is equivalent to @'prettyTypeableShowsPrec' 0@ --- --- @since 0.24.0 -prettyTypeableShows :: SomeTypeRep -> ShowS -prettyTypeableShows = prettyTypeableShowsPrec 0 - --- | An internal utility to print 'Typeable' without the kinds. --- This is like 'showsPrec' in that it accepts a /precedence/ parameter, --- and the result is in parentheses when the precedence is higher than 9. --- --- @since 0.24.0 -prettyTypeableShowsPrec :: Int -> SomeTypeRep -> ShowS -prettyTypeableShowsPrec d (SomeTypeRep tr) sIn = - let (con, conArgs) = splitApps tr - in case conArgs of - [] -> showString (tyConName con) sIn - _ -> - showParen - (d >= 10) - (showString (tyConName con) . showChar ':' . - foldr1 (\f acc -> showChar '-' . f . acc) - (prettyTypeableShowsPrec 10 <$> conArgs)) - sIn - type instance ToPretty (Endpoint a) = ToPretty a <+> PutStr "endpoint" makeLenses ''Endpoint -instance (HasPdu a1 r, HasPdu a2 r) => HasPdu (a1, a2) r where +instance (Typeable r, HasPdu a1 r, HasPdu a2 r) => HasPdu (a1, a2) r where data instance Pdu (a1, a2) r where ToPduLeft :: Pdu a1 r -> Pdu (a1, a2) r ToPduRight :: Pdu a2 r -> Pdu (a1, a2) r @@ -208,7 +194,7 @@ Nothing -> Nothing -instance (HasPdu a1 r, HasPdu a2 r, HasPdu a3 r) => HasPdu (a1, a2, a3) r where +instance (Typeable r, HasPdu a1 r, HasPdu a2 r, HasPdu a3 r) => HasPdu (a1, a2, a3) r where data instance Pdu (a1, a2, a3) r where ToPdu1 :: Pdu a1 r -> Pdu (a1, a2, a3) r ToPdu2 :: Pdu a2 r -> Pdu (a1, a2, a3) r @@ -229,7 +215,7 @@ Nothing -> Nothing -instance (HasPdu a1 r, HasPdu a2 r, HasPdu a3 r, HasPdu a4 r) => HasPdu (a1, a2, a3, a4) r where +instance (Typeable r, HasPdu a1 r, HasPdu a2 r, HasPdu a3 r, HasPdu a4 r) => HasPdu (a1, a2, a3, a4) r where data instance Pdu (a1, a2, a3, a4) r where ToPdu1Of4 :: Pdu a1 r -> Pdu (a1, a2, a3, a4) r ToPdu2Of4 :: Pdu a2 r -> Pdu (a1, a2, a3, a4) r @@ -255,7 +241,7 @@ Nothing -> Nothing -instance (HasPdu a1 r, HasPdu a2 r, HasPdu a3 r, HasPdu a4 r, HasPdu a5 r) => HasPdu (a1, a2, a3, a4, a5) r where +instance (Typeable r, HasPdu a1 r, HasPdu a2 r, HasPdu a3 r, HasPdu a4 r, HasPdu a5 r) => HasPdu (a1, a2, a3, a4, a5) r where data instance Pdu (a1, a2, a3, a4, a5) r where ToPdu1Of5 :: Pdu a1 r -> Pdu (a1, a2, a3, a4, a5) r ToPdu2Of5 :: Pdu a2 r -> Pdu (a1, a2, a3, a4, a5) r @@ -302,10 +288,16 @@ -- Laws: @embeddedPdu = prism' embedPdu fromPdu@ -- -- @since 0.24.0 -class EmbedProtocol protocol embeddedProtocol (result :: Synchronicity) where +class + ( HasPdu protocol result + , HasPdu embeddedProtocol result + ) + => EmbedProtocol protocol embeddedProtocol (result :: Synchronicity) where + -- | A 'Prism' for the embedded 'Pdu's. embeddedPdu :: Prism' (Pdu protocol result) (Pdu embeddedProtocol result) embeddedPdu = prism' embedPdu fromPdu + -- | Embed the 'Pdu' value of an embedded protocol into the corresponding -- 'Pdu' value. embedPdu :: Pdu embeddedProtocol result -> Pdu protocol result @@ -332,7 +324,7 @@ fromEmbeddedEndpoint :: forall outer inner r . EmbedProtocol outer inner r => Endpoint inner -> Endpoint outer fromEmbeddedEndpoint = coerce -instance EmbedProtocol a a r where +instance HasPdu a r => EmbedProtocol a a r where embeddedPdu = prism' id Just embedPdu = id fromPdu = Just @@ -345,12 +337,12 @@ showsPrec d (ToPduLeft x) = showsPrec d x showsPrec d (ToPduRight y) = showsPrec d y -instance EmbedProtocol (a1, a2) a1 r where +instance (Typeable r, HasPdu a1 r, HasPdu a2 r) => EmbedProtocol (a1, a2) a1 r where embedPdu = ToPduLeft fromPdu (ToPduLeft l) = Just l fromPdu _ = Nothing -instance EmbedProtocol (a1, a2) a2 r where +instance (Typeable r, HasPdu a1 r, HasPdu a2 r) => EmbedProtocol (a1, a2) a2 r where embeddedPdu = prism' ToPduRight $ \case ToPduRight r -> Just r @@ -366,17 +358,17 @@ showsPrec d (ToPdu2 y) = showsPrec d y showsPrec d (ToPdu3 z) = showsPrec d z -instance EmbedProtocol (a1, a2, a3) a1 r where +instance (Typeable r, HasPdu a1 r, HasPdu a2 r, HasPdu a3 r) => EmbedProtocol (a1, a2, a3) a1 r where embedPdu = ToPdu1 fromPdu (ToPdu1 l) = Just l fromPdu _ = Nothing -instance EmbedProtocol (a1, a2, a3) a2 r where +instance (Typeable r, HasPdu a1 r, HasPdu a2 r, HasPdu a3 r) => EmbedProtocol (a1, a2, a3) a2 r where embedPdu = ToPdu2 fromPdu (ToPdu2 l) = Just l fromPdu _ = Nothing -instance EmbedProtocol (a1, a2, a3) a3 r where +instance (Typeable r, HasPdu a1 r, HasPdu a2 r, HasPdu a3 r) => EmbedProtocol (a1, a2, a3) a3 r where embedPdu = ToPdu3 fromPdu (ToPdu3 l) = Just l fromPdu _ = Nothing @@ -393,27 +385,27 @@ showsPrec d (ToPdu3Of4 z) = showsPrec d z showsPrec d (ToPdu4Of4 w) = showsPrec d w -instance EmbedProtocol (a1, a2, a3, a4) a1 r where +instance (Typeable r, HasPdu a1 r, HasPdu a2 r, HasPdu a3 r, HasPdu a4 r) => EmbedProtocol (a1, a2, a3, a4) a1 r where embedPdu = ToPdu1Of4 fromPdu (ToPdu1Of4 l) = Just l fromPdu _ = Nothing -instance EmbedProtocol (a1, a2, a3, a4) a2 r where +instance (Typeable r, HasPdu a1 r, HasPdu a2 r, HasPdu a3 r, HasPdu a4 r) => EmbedProtocol (a1, a2, a3, a4) a2 r where embedPdu = ToPdu2Of4 fromPdu (ToPdu2Of4 l) = Just l fromPdu _ = Nothing -instance EmbedProtocol (a1, a2, a3, a4) a3 r where +instance (Typeable r, HasPdu a1 r, HasPdu a2 r, HasPdu a3 r, HasPdu a4 r) => EmbedProtocol (a1, a2, a3, a4) a3 r where embedPdu = ToPdu3Of4 fromPdu (ToPdu3Of4 l) = Just l fromPdu _ = Nothing -instance EmbedProtocol (a1, a2, a3, a4) a4 r where +instance (Typeable r, HasPdu a1 r, HasPdu a2 r, HasPdu a3 r, HasPdu a4 r) => EmbedProtocol (a1, a2, a3, a4) a4 r where embedPdu = ToPdu4Of4 fromPdu (ToPdu4Of4 l) = Just l fromPdu _ = Nothing -instance (NFData (Pdu a1 r), NFData (Pdu a2 r), NFData (Pdu a3 r), NFData (Pdu a4 r), NFData (Pdu a5 r)) => NFData (Pdu (a1, a2, a3, a4, a5) r) where +instance (Typeable r, NFData (Pdu a1 r), NFData (Pdu a2 r), NFData (Pdu a3 r), NFData (Pdu a4 r), NFData (Pdu a5 r)) => NFData (Pdu (a1, a2, a3, a4, a5) r) where rnf (ToPdu1Of5 x) = rnf x rnf (ToPdu2Of5 y) = rnf y rnf (ToPdu3Of5 z) = rnf z @@ -427,27 +419,27 @@ showsPrec d (ToPdu4Of5 w) = showsPrec d w showsPrec d (ToPdu5Of5 v) = showsPrec d v -instance EmbedProtocol (a1, a2, a3, a4, a5) a1 r where +instance (Typeable r, HasPdu a1 r, HasPdu a2 r, HasPdu a3 r, HasPdu a4 r, HasPdu a5 r) => EmbedProtocol (a1, a2, a3, a4, a5) a1 r where embedPdu = ToPdu1Of5 fromPdu (ToPdu1Of5 l) = Just l fromPdu _ = Nothing -instance EmbedProtocol (a1, a2, a3, a4, a5) a2 r where +instance (Typeable r, HasPdu a1 r, HasPdu a2 r, HasPdu a3 r, HasPdu a4 r, HasPdu a5 r) => EmbedProtocol (a1, a2, a3, a4, a5) a2 r where embedPdu = ToPdu2Of5 fromPdu (ToPdu2Of5 l) = Just l fromPdu _ = Nothing -instance EmbedProtocol (a1, a2, a3, a4, a5) a3 r where +instance (Typeable r, HasPdu a1 r, HasPdu a2 r, HasPdu a3 r, HasPdu a4 r, HasPdu a5 r) => EmbedProtocol (a1, a2, a3, a4, a5) a3 r where embedPdu = ToPdu3Of5 fromPdu (ToPdu3Of5 l) = Just l fromPdu _ = Nothing -instance EmbedProtocol (a1, a2, a3, a4, a5) a4 r where +instance (Typeable r, HasPdu a1 r, HasPdu a2 r, HasPdu a3 r, HasPdu a4 r, HasPdu a5 r) => EmbedProtocol (a1, a2, a3, a4, a5) a4 r where embedPdu = ToPdu4Of5 fromPdu (ToPdu4Of5 l) = Just l fromPdu _ = Nothing -instance EmbedProtocol (a1, a2, a3, a4, a5) a5 r where +instance (Typeable r, HasPdu a1 r, HasPdu a2 r, HasPdu a3 r, HasPdu a4 r, HasPdu a5 r) => EmbedProtocol (a1, a2, a3, a4, a5) a5 r where embedPdu = ToPdu5Of5 fromPdu (ToPdu5Of5 l) = Just l fromPdu _ = Nothing diff --git a/src/Control/Eff/Concurrent/Protocol/CallbackServer.hs b/src/Control/Eff/Concurrent/Protocol/CallbackServer.hs --- a/src/Control/Eff/Concurrent/Protocol/CallbackServer.hs +++ b/src/Control/Eff/Concurrent/Protocol/CallbackServer.hs @@ -16,11 +16,12 @@ import Control.DeepSeq import Control.Eff -import Control.Eff.Extend () +import Control.Eff.Concurrent.Misc import Control.Eff.Concurrent.Process import Control.Eff.Concurrent.Protocol import qualified Control.Eff.Concurrent.Protocol.EffectfulServer as E import Control.Eff.Concurrent.Protocol.EffectfulServer (Event(..)) +import Control.Eff.Extend () import Control.Eff.Log import Data.Kind import Data.String @@ -71,8 +72,7 @@ -- @since 0.27.0 type TangibleCallbacks tag eLoop e = ( LogIo e - , SetMember Process (Process e) eLoop - , Member Interrupts eLoop + , HasProcesses eLoop e , Typeable e , Typeable eLoop , Typeable tag @@ -91,7 +91,7 @@ (d >= 10) (showString (T.unpack x) . showString " :: " - . prettyTypeableShows (typeOf px) + . showSTypeRep (typeOf px) ) instance (TangibleCallbacks tag eLoop e) => E.Server (Server (tag :: Type) eLoop) (Processes e) where @@ -113,7 +113,7 @@ showsPrec d svr = showParen (d>=10) ( showsPrec 11 (genServerId svr) - . showChar ' ' . prettyTypeableShows (typeRep (Proxy @tag)) + . showChar ' ' . showSTypeRep (typeRep (Proxy @tag)) . showString " callback-server" ) diff --git a/src/Control/Eff/Concurrent/Protocol/Client.hs b/src/Control/Eff/Concurrent/Protocol/Client.hs --- a/src/Control/Eff/Concurrent/Protocol/Client.hs +++ b/src/Control/Eff/Concurrent/Protocol/Client.hs @@ -36,18 +36,17 @@ -- -- The message will be reduced to normal form ('rnf') in the caller process. cast - :: forall o' o r q + :: forall destination protocol r q . ( HasCallStack - , SetMember Process (Process q) r - , Member Interrupts r - , HasPdu o' 'Asynchronous - , HasPdu o 'Asynchronous - , EmbedProtocol o' o 'Asynchronous + , HasProcesses r q + , HasPdu destination 'Asynchronous + , HasPdu protocol 'Asynchronous + , EmbedProtocol destination protocol 'Asynchronous ) - => Endpoint o' - -> Pdu o 'Asynchronous + => Endpoint destination + -> Pdu protocol 'Asynchronous -> Eff r () -cast (Endpoint pid) castMsg = sendMessage pid (Cast (embedPdu @o' castMsg)) +cast (Endpoint pid) castMsg = sendMessage pid (Cast (embedPdu @destination castMsg)) -- | Send a request 'Pdu' and wait for the server to return a result value. -- @@ -56,28 +55,27 @@ -- -- __Always prefer 'callWithTimeout' over 'call'__ call - :: forall result protocol' protocol r q - . ( SetMember Process (Process q) r - , Member Interrupts r - , TangiblePdu protocol' ( 'Synchronous result) + :: forall result destination protocol r q + . ( HasProcesses r q + , TangiblePdu destination ( 'Synchronous result) , TangiblePdu protocol ( 'Synchronous result) - , EmbedProtocol protocol' protocol ( 'Synchronous result) + , EmbedProtocol destination protocol ( 'Synchronous result) , Tangible result , HasCallStack ) - => Endpoint protocol' + => Endpoint destination -> Pdu protocol ( 'Synchronous result) -> Eff r result call (Endpoint pidInternal) req = do callRef <- makeReference fromPid <- self - let requestMessage = Call origin $! (embedPdu @protocol' req) - origin = RequestOrigin @protocol' @result fromPid callRef + let requestMessage = Call origin $! (embedPdu @destination req) + origin = RequestOrigin @destination @result fromPid callRef sendMessage pidInternal requestMessage let selectResult :: MessageSelector result selectResult = let extractResult - :: Reply protocol' result -> Maybe result + :: Reply destination result -> Maybe result extractResult (Reply origin' result) = if origin == origin' then Just result else Nothing in selectMessageWith extractResult @@ -99,31 +97,30 @@ -- -- @since 0.22.0 callWithTimeout - :: forall result protocol' protocol r q - . ( SetMember Process (Process q) r - , Member Interrupts r - , TangiblePdu protocol' ( 'Synchronous result) + :: forall result destination protocol r q + . ( HasProcesses r q + , TangiblePdu destination ( 'Synchronous result) , TangiblePdu protocol ( 'Synchronous result) - , EmbedProtocol protocol' protocol ( 'Synchronous result) + , EmbedProtocol destination protocol ( 'Synchronous result) , Tangible result , Member Logs r , Lifted IO q , Lifted IO r , HasCallStack ) - => Endpoint protocol' + => Endpoint destination -> Pdu protocol ( 'Synchronous result) -> Timeout -> Eff r result callWithTimeout serverP@(Endpoint pidInternal) req timeOut = do fromPid <- self callRef <- makeReference - let requestMessage = Call origin $! embedPdu @protocol' req - origin = RequestOrigin @protocol' @result fromPid callRef + let requestMessage = Call origin $! embedPdu @destination req + origin = RequestOrigin @destination @result fromPid callRef sendMessage pidInternal requestMessage let selectResult = let extractResult - :: Reply protocol' result -> Maybe result + :: Reply destination result -> Maybe result extractResult (Reply origin' result) = if origin == origin' then Just result else Nothing in selectMessageWith extractResult @@ -146,9 +143,9 @@ -- only a __single server__ for a given 'Pdu' instance. This type alias is -- convenience to express that an effect has 'Process' and a reader for a -- 'Endpoint'. -type ServesProtocol o r q = +type ServesProtocol o r q = -- TODO remove! ( Typeable o - , SetMember Process (Process q) r + , HasSafeProcesses r q , Member (EndpointReader o) r ) @@ -176,6 +173,7 @@ , Tangible reply , TangiblePdu o ( 'Synchronous reply) , Member Interrupts r + ) => Pdu o ( 'Synchronous reply) -> Eff r reply @@ -205,15 +203,14 @@ -- This function makes use of AmbigousTypes and TypeApplications. -- -- When not working with an embedded 'Pdu' use 'callEndpointReader'. --- +-- -- @since 0.25.1 callSingleton :: forall outer inner reply q e . ( HasCallStack , EmbedProtocol outer inner ( 'Synchronous reply) , Member (EndpointReader outer) e - , SetMember Process (Process q) e - , Member Interrupts e + , HasProcesses e q , TangiblePdu outer ('Synchronous reply) , TangiblePdu inner ('Synchronous reply) , Tangible reply @@ -234,8 +231,7 @@ . ( HasCallStack , EmbedProtocol outer inner 'Asynchronous , Member (EndpointReader outer) e - , SetMember Process (Process q) e - , Member Interrupts e + , HasProcesses e q , HasPdu outer 'Asynchronous , HasPdu inner 'Asynchronous ) diff --git a/src/Control/Eff/Concurrent/Protocol/EffectfulServer.hs b/src/Control/Eff/Concurrent/Protocol/EffectfulServer.hs --- a/src/Control/Eff/Concurrent/Protocol/EffectfulServer.hs +++ b/src/Control/Eff/Concurrent/Protocol/EffectfulServer.hs @@ -13,6 +13,7 @@ import Control.Applicative import Control.DeepSeq import Control.Eff +import Control.Eff.Concurrent.Misc import Control.Eff.Extend () import Control.Eff.Concurrent.Process import Control.Eff.Concurrent.Process.Timer @@ -63,7 +64,7 @@ serverTitle :: Init a e -> ProcessTitle default serverTitle :: Typeable (ServerPdu a) => Init a e -> ProcessTitle - serverTitle _ = fromString $ prettyTypeableShows (typeRep (Proxy @(ServerPdu a))) "-server" + serverTitle _ = fromString $ showSTypeable @(ServerPdu a) "-server" -- | Process the effects of the implementation runEffects :: Endpoint (ServerPdu a) -> Init a e -> Eff (ServerEffects a e) x -> Eff e x @@ -82,32 +83,32 @@ -- -- @since 0.24.0 start - :: forall a q h + :: forall a r q h . ( Server a (Processes q) , Typeable a , Typeable (ServerPdu a) , LogsTo h (Processes q) - , SetMember Process (Process q) (ServerEffects a (Processes q)) - , Member Interrupts (ServerEffects a (Processes q)) + , HasProcesses (ServerEffects a (Processes q)) q + , HasProcesses r q , HasCallStack) => Init a (Processes q) - -> Eff (Processes q) (Endpoint (ServerPdu a)) + -> Eff r (Endpoint (ServerPdu a)) start a = asEndpoint <$> spawn (serverTitle a) (protocolServerLoop a) -- | Execute the server loop. -- -- @since 0.24.0 startLink - :: forall a q h + :: forall a r q h . ( Typeable a , Typeable (ServerPdu a) , Server a (Processes q) , LogsTo h (Processes q) - , SetMember Process (Process q) (ServerEffects a (Processes q)) - , Member Interrupts (ServerEffects a (Processes q)) + , HasProcesses (ServerEffects a (Processes q)) q + , HasProcesses r q , HasCallStack) => Init a (Processes q) - -> Eff (Processes q) (Endpoint (ServerPdu a)) + -> Eff r (Endpoint (ServerPdu a)) startLink a = asEndpoint <$> spawnLink (serverTitle a) (protocolServerLoop a) -- | Execute the server loop. @@ -117,8 +118,7 @@ :: forall q h a . ( Server a (Processes q) , LogsTo h (Processes q) - , SetMember Process (Process q) (ServerEffects a (Processes q)) - , Member Interrupts (ServerEffects a (Processes q)) + , HasProcesses (ServerEffects a (Processes q)) q , Typeable a , Typeable (ServerPdu a) ) @@ -133,10 +133,10 @@ where lmAddEp myEp = lmProcessId ?~ myEp sel :: MessageSelector (Event (ServerPdu a)) - sel = onRequest <$> selectMessage @(Request (ServerPdu a)) - <|> OnDown <$> selectMessage @ProcessDown - <|> OnTimeOut <$> selectMessage @TimerElapsed - <|> OnMessage <$> selectAnyMessage + sel = onRequest <$> selectMessage @(Request (ServerPdu a)) + <|> OnDown <$> selectMessage @ProcessDown + <|> OnTimeOut <$> selectMessage @TimerElapsed + <|> OnMessage <$> selectAnyMessage where onRequest :: Request (ServerPdu a) -> Event (ServerPdu a) onRequest (Call o m) = OnCall (replyTarget (MkSerializer toStrictDynamic) o) m diff --git a/src/Control/Eff/Concurrent/Protocol/Observer.hs b/src/Control/Eff/Concurrent/Protocol/Observer.hs --- a/src/Control/Eff/Concurrent/Protocol/Observer.hs +++ b/src/Control/Eff/Concurrent/Protocol/Observer.hs @@ -1,296 +1,330 @@ -- | Observer Effects -- --- This module supports the implementation of observers and observables. Expected use +-- This module supports the implementation of observerRegistry and observables. Expected use -- case is event propagation. -- +-- The observable event sources and the observers are usually server processes for a +-- protocol that embeds the 'ObserverRegistry' and 'Observer' 'Pdu's respectively. +-- +-- A generic FIFO queue based observer can be found in "Control.Eff.Concurrent.Protocol.Observer.Queue". +-- -- @since 0.16.0 module Control.Eff.Concurrent.Protocol.Observer ( Observer(..) - , TangibleObserver + , ObservationSink() + , IsObservable + , CanObserve , Pdu(RegisterObserver, ForgetObserver, Observed) , registerObserver , forgetObserver - , handleObservations - , toObserver - , toObserverFor - , ObserverRegistry - , ObserverState - , Observers() - , emptyObservers - , handleObserverRegistration - , manageObservers - , observed + , forgetObserverUnsafe + , ObserverRegistry(..) + , ObserverRegistryState + , observerRegistryNotify + , evalObserverRegistryState + , emptyObserverRegistry + , observerRegistryHandlePdu + , observerRegistryRemoveProcess ) where -import Control.DeepSeq (NFData(rnf)) +import Control.DeepSeq ( NFData(rnf) ) import Control.Eff +import Control.Eff.Concurrent.Misc import Control.Eff.Concurrent.Process import Control.Eff.Concurrent.Protocol import Control.Eff.Concurrent.Protocol.Client +import Control.Eff.Concurrent.Protocol.Wrapper (Request(Cast)) import Control.Eff.State.Strict import Control.Eff.Log import Control.Lens -import Data.Data (typeOf) +import Control.Monad import Data.Dynamic -import Data.Foldable import Data.Kind -import Data.Proxy -import Data.Set ( Set ) -import qualified Data.Set as Set +import Data.Semigroup +import Data.Map ( Map ) +import qualified Data.Map as Map import Data.Text ( pack ) -import Data.Typeable ( typeRep ) import Data.Type.Pretty +import GHC.Generics import GHC.Stack -- * Observers --- | Describes a process that observes another via 'Asynchronous' 'Pdu' messages. +-- ** Observables + +-- | A /protocol/ to communicate 'Observed' events from a sources to many sinks. -- --- An observer consists of a filter and a process id. The filter converts an observation to --- a message understood by the observer process, and the 'ProcessId' is used to send the message. +-- A sink is any process that serves a protocol with a 'Pdu' instance that embeds +-- the 'Observer' Pdu via an 'EmbedProtocol' instance. -- --- @since 0.16.0 -data Observer o where - Observer - :: ( Tangible o - , HasPdu p 'Asynchronous - , Tangible (Endpoint p) - , Typeable p - ) - => (o -> Maybe (Pdu p 'Asynchronous)) -> Endpoint p -> Observer o +-- This type has /dual use/, for one it serves as type-index for 'Pdu', i.e. +-- 'HasPdu' respectively, and secondly it contains an 'ObservationSink' and +-- a 'MonitorReference'. +-- +-- The 'ObservationSink' is used to serialize and send the 'Observed' events, +-- while the 'ProcessId' serves as key for internal maps. +-- +-- @since 0.28.0 +newtype Observer event = + MkObserver (Arg ProcessId (ObservationSink event)) + deriving (Eq, Ord, Typeable) +instance NFData (Observer event) where + rnf (MkObserver (Arg x y)) = rnf x `seq` rnf y --- | The constraints on the type parameters to an 'Observer' --- --- @since 0.24.0 -type TangibleObserver o = - ( Tangible o, HasPdu (Observer o) 'Asynchronous) +instance Typeable event => Show (Observer event) where + showsPrec d (MkObserver (Arg x (MkObservationSink _ m))) = + showParen (d>=10) (showSTypeable @event . showString "-observer: " . shows x . showChar ' ' . shows m ) -type instance ToPretty (Observer o) = - PrettyParens ("observing" <:> ToPretty o) +type instance ToPretty (Observer event) = + PrettyParens ("observing" <:> ToPretty event) -instance (NFData o) => NFData (Observer o) where - rnf (Observer k s) = rnf k `seq` rnf s +instance (Typeable r, Tangible event) => HasPdu (Observer event) (r :: Synchronicity) where + data Pdu (Observer event) r where + Observed :: event -> Pdu (Observer event) 'Asynchronous + deriving Typeable -instance Show (Observer o) where - showsPrec d (Observer _ p) = showParen - (d >= 10) - (shows (typeRep (Proxy :: Proxy o)) . showString " observer: " . shows p) +instance NFData event => NFData (Pdu (Observer event) r) where + rnf (Observed event) = rnf event -instance Ord (Observer o) where - compare (Observer _ s1) (Observer _ s2) = - compare (s1 ^. fromEndpoint) (s2 ^. fromEndpoint) +instance Show event => Show (Pdu (Observer event) r) where + showsPrec d (Observed event) = + showParen (d >= 10) (showString "observered: " . shows event) -instance Eq (Observer o) where - (==) (Observer _ s1) (Observer _ s2) = - (==) (s1 ^. fromEndpoint) (s2 ^. fromEndpoint) +-- | The Information necessary to wrap an 'Observed' event to a process specific +-- message, e.g. the embedded 'Observer' 'Pdu' instance, and the 'MonitorReference' of +-- the destination process. +-- +-- @since 0.28.0 +data ObservationSink event = + MkObservationSink + { _observerSerializer :: Serializer (Pdu (Observer event) 'Asynchronous) + , _observerMonitorReference :: MonitorReference + } + deriving (Generic, Typeable) --- | And an 'Observer' to the set of recipients for all observations reported by 'observed'. --- Note that the observers are keyed by the observing process, i.e. a previous entry for the process +instance NFData (ObservationSink event) where + rnf (MkObservationSink s p) = s `seq` rnf p + +-- | Convenience type alias. +-- +-- @since 0.28.0 +type IsObservable eventSource event = + ( Tangible event + , EmbedProtocol eventSource (ObserverRegistry event) 'Asynchronous + ) + +-- | Convenience type alias. +-- +-- @since 0.28.0 +type CanObserve eventSink event = + ( Tangible event + , EmbedProtocol eventSink (Observer event) 'Asynchronous + ) + +-- | And an 'Observer' to the set of recipients for all observations reported by 'observerRegistryNotify'. +-- Note that the observerRegistry are keyed by the observing process, i.e. a previous entry for the process -- contained in the 'Observer' is overwritten. If you want multiple entries for a single process, just -- combine several filter functions. -- -- @since 0.16.0 registerObserver - :: ( SetMember Process (Process q) r - , HasCallStack - , Member Interrupts r - , TangibleObserver o - , EmbedProtocol x (ObserverRegistry o) 'Asynchronous - , HasPdu x 'Asynchronous + :: forall event eventSink eventSource r q . + ( HasCallStack + , HasProcesses r q + , IsObservable eventSource event + , CanObserve eventSink event ) - => Observer o - -> Endpoint x + => Endpoint eventSource + -> Endpoint eventSink -> Eff r () -registerObserver observer observerRegistry = - cast observerRegistry (RegisterObserver observer) +registerObserver eventSource eventSink = + cast eventSource (RegisterObserver serializer (eventSink ^. fromEndpoint)) + where + serializer = + MkSerializer + ( toStrictDynamic + . Cast + . embedPdu @eventSink @(Observer event) @( 'Asynchronous ) + ) + -- | Send the 'ForgetObserver' message -- -- @since 0.16.0 forgetObserver - :: ( SetMember Process (Process q) r + :: forall event eventSink eventSource r q . + ( HasProcesses r q , HasCallStack - , Member Interrupts r - , Typeable o - , NFData o - , EmbedProtocol x (ObserverRegistry o) 'Asynchronous - , HasPdu x 'Asynchronous + , IsObservable eventSource event + , CanObserve eventSink event ) - => Observer o - -> Endpoint x + => Endpoint eventSource + -> Endpoint eventSink -> Eff r () -forgetObserver observer observerRegistry = - cast observerRegistry (ForgetObserver observer) - --- ** Observer Support Functions - --- | A minimal Protocol for handling observations. --- This is one simple way of receiving observations - of course users can use --- any other 'Asynchronous' 'Pdu' message type for receiving observations. --- --- @since 0.16.0 -instance (NFData o, Show o, Typeable o, Typeable r) => HasPdu (Observer o) r where - data instance Pdu (Observer o) r where - -- | This message denotes that the given value was 'observed'. - -- - -- @since 0.16.1 - Observed :: o -> Pdu (Observer o) 'Asynchronous - deriving Typeable - -instance NFData o => NFData (Pdu (Observer o) r) where - rnf (Observed o) = rnf o - -instance Show o => Show (Pdu (Observer o) r) where - showsPrec d (Observed o) = showParen (d>=10) (showString "observered: " . shows o) - --- | Based on the 'Pdu' instance for 'Observer' this simplified writing --- a callback handler for observations. In order to register to --- and 'ObserverRegistry' use 'toObserver'. --- --- @since 0.16.0 -handleObservations - :: (HasCallStack, Typeable o, SetMember Process (Process q) r, NFData (Observer o)) - => (o -> Eff r ()) - -> Pdu (Observer o) 'Asynchronous -> Eff r () -handleObservations k (Observed r) = k r +forgetObserver eventSource eventSink = + forgetObserverUnsafe @event @eventSource eventSource (eventSink ^. fromEndpoint) --- | Use a 'Endpoint' as an 'Observer' for 'handleObservations'. +-- | Send the 'ForgetObserver' message, use a raw 'ProcessId' as parameter. -- --- @since 0.16.0 -toObserver - :: forall o p - . ( HasPdu p 'Asynchronous - , EmbedProtocol p (Observer o) 'Asynchronous - , TangibleObserver o - ) - => Endpoint p - -> Observer o -toObserver = toObserverFor (embedPdu @p . Observed) +-- @since 0.28.0 +forgetObserverUnsafe + :: forall event eventSource r q . + ( HasProcesses r q + , HasCallStack + , IsObservable eventSource event + ) + => Endpoint eventSource + -> ProcessId + -> Eff r () +forgetObserverUnsafe eventSource eventSink = + cast eventSource (ForgetObserver @event eventSink) --- | Create an 'Observer' that conditionally accepts all observations of the --- given type and applies the given function to them; the function takes an observation and returns an 'Pdu' --- cast that the observer server is compatible to. --- --- @since 0.16.0 -toObserverFor - :: (TangibleObserver o, Typeable a, HasPdu a 'Asynchronous) - => (o -> Pdu a 'Asynchronous) - -> Endpoint a - -> Observer o -toObserverFor wrapper = Observer (Just . wrapper) +-- ** Observer Support Functions -- * Managing Observers -- | A protocol for managing 'Observer's, encompassing registration and de-registration of -- 'Observer's. -- --- @since 0.16.0 -data ObserverRegistry (o :: Type) +-- @since 0.28.0 +data ObserverRegistry (event :: Type) = MkObserverRegistry + { _observerRegistry :: Map ProcessId (ObservationSink event) } deriving Typeable -type instance ToPretty (ObserverRegistry o) = - PrettyParens ("observer registry" <:> ToPretty o) +type instance ToPretty (ObserverRegistry event) = + PrettyParens ("observer registry" <:> ToPretty event) -instance (Typeable o, Typeable r) => HasPdu (ObserverRegistry o) r where +instance (Tangible event, Typeable r) => HasPdu (ObserverRegistry event) r where -- | Protocol for managing observers. This can be added to any server for any number of different observation types. - -- The functions 'manageObservers' and 'handleObserverRegistration' are used to include observer handling; + -- The functions 'evalObserverRegistryState' and 'observerRegistryHandlePdu' are used to include observer handling; -- -- @since 0.16.0 - data instance Pdu (ObserverRegistry o) r where + data instance Pdu (ObserverRegistry event) r where -- | This message denotes that the given 'Observer' should receive observations until 'ForgetObserver' is -- received. -- - -- @since 0.16.1 - RegisterObserver :: NFData o => Observer o -> Pdu (ObserverRegistry o) 'Asynchronous + -- @since 0.28.0 + RegisterObserver :: Serializer (Pdu (Observer event) 'Asynchronous) -> ProcessId -> Pdu (ObserverRegistry event) 'Asynchronous -- | This message denotes that the given 'Observer' should not receive observations anymore. -- -- @since 0.16.1 - ForgetObserver :: NFData o => Observer o -> Pdu (ObserverRegistry o) 'Asynchronous + ForgetObserver :: ProcessId -> Pdu (ObserverRegistry event) 'Asynchronous +-- -- | This message denotes that a monitored process died +-- -- +-- -- @since 0.28.0 +-- ObserverMightBeDown :: MonitorReference -> Pdu (ObserverRegistry event) ( 'Synchronous Bool) deriving Typeable -instance NFData (Pdu (ObserverRegistry o) r) where - rnf (RegisterObserver o) = rnf o - rnf (ForgetObserver o) = rnf o +instance NFData (Pdu (ObserverRegistry event) r) where + rnf (RegisterObserver ser pid) = rnf ser `seq` rnf pid + rnf (ForgetObserver pid) = rnf pid -instance Show (Pdu (ObserverRegistry o) r) where - showsPrec d (RegisterObserver o) = showParen (d >= 10) (showString "register observer: " . showsPrec 11 o) - showsPrec d (ForgetObserver o) = showParen (d >= 10) (showString "forget observer: " . showsPrec 11 o) +instance Typeable event => Show (Pdu (ObserverRegistry event) r) where + showsPrec d (RegisterObserver ser pid) = showParen (d >= 10) (showString "register observer: " . shows ser . showChar ' ' . shows pid) + showsPrec d (ForgetObserver p) = showParen (d >= 10) (showString "forget observer: " . shows p) -- ** Protocol for integrating 'ObserverRegistry' into processes. -- | Provide the implementation for the 'ObserverRegistry' Protocol, this handled 'RegisterObserver' and 'ForgetObserver' --- messages. It also adds the 'ObserverState' constraint to the effect list. +-- messages. It also adds the 'ObserverRegistryState' constraint to the effect list. -- --- @since 0.16.0 -handleObserverRegistration - :: forall o q r +-- @since 0.28.0 +observerRegistryHandlePdu + :: forall event q r . ( HasCallStack - , Typeable o - , SetMember Process (Process q) r - , Member (ObserverState o) r + , Typeable event + , HasProcesses r q + , Member (ObserverRegistryState event) r , Member Logs r ) - => Pdu (ObserverRegistry o) 'Asynchronous -> Eff r () -handleObserverRegistration = \case - RegisterObserver ob -> do - os <- get @(Observers o) - logDebug ("registering " - <> pack (show (typeOf ob)) - <> " current number of observers: " - <> pack (show (Set.size (view observers os)))) - put (over observers (Set.insert ob) os) + => Pdu (ObserverRegistry event) 'Asynchronous -> Eff r () +observerRegistryHandlePdu = \case + RegisterObserver ser pid -> do + monRef <- monitor pid + let sink = MkObservationSink ser monRef + observer = MkObserver (Arg pid sink) + modify @(ObserverRegistry event) (over observerRegistry (Map.insert pid sink)) + os <- get @(ObserverRegistry event) + logDebug ( "registered " + <> pack (show observer) + <> " current number of observers: " -- TODO put this info into the process details + <> pack (show (Map.size (view observerRegistry os)))) ForgetObserver ob -> do - os <- get @(Observers o) - logDebug ("forgetting " - <> pack (show (typeOf ob)) - <> " current number of observers: " - <> pack (show (Set.size (view observers os)))) - put (over observers (Set.delete ob) os) + wasRemoved <- observerRegistryRemoveProcess @event ob + unless wasRemoved $ + logDebug (pack (show ("unknown observer " ++ show ob))) + +-- | Remove the entry in the 'ObserverRegistry' for the 'ProcessId' +-- and return 'True' if there was an entry, 'False' otherwise. +-- +-- @since 0.28.0 +observerRegistryRemoveProcess + :: forall event q r + . ( HasCallStack + , Typeable event + , HasProcesses r q + , Member (ObserverRegistryState event) r + , Member Logs r + ) + => ProcessId -> Eff r Bool +observerRegistryRemoveProcess ob = do + mSink <- view (observerRegistry . at ob) <$> get @(ObserverRegistry event) + modify @(ObserverRegistry event) (observerRegistry . at ob .~ Nothing) + os <- get @(ObserverRegistry event) + maybe + (pure False) + (foundIt os) + mSink + where + foundIt os sink@(MkObservationSink _ monRef) = do + demonitor monRef + logDebug ( "removed: " + <> (pack $ show $ MkObserver $ Arg ob sink) + <> " current number of observers: " + <> pack (show (Map.size (view observerRegistry os)))) + pure True + -- | Keep track of registered 'Observer's. -- --- Handle the 'ObserverState' introduced by 'handleObserverRegistration'. +-- Handle the 'ObserverRegistryState' effect, i.e. run 'evalState' on an 'emptyObserverRegistry'. -- --- @since 0.16.0 -manageObservers :: Eff (ObserverState o ': r) a -> Eff r a -manageObservers = evalState (Observers Set.empty) +-- @since 0.28.0 +evalObserverRegistryState :: HasCallStack => Eff (ObserverRegistryState event ': r) a -> Eff r a +evalObserverRegistryState = evalState emptyObserverRegistry --- | The empty 'ObserverState' +-- | The empty 'ObserverRegistryState' -- --- @since 0.24.0 -emptyObservers :: Observers o -emptyObservers = Observers Set.empty - --- | Internal state for 'manageObservers' -newtype Observers o = - Observers { _observers :: Set (Observer o) } - deriving (Eq, Ord, Typeable, Show, NFData) +-- @since 0.28.0 +emptyObserverRegistry :: ObserverRegistry event +emptyObserverRegistry = MkObserverRegistry Map.empty --- | Alias for the effect that contains the observers managed by 'manageObservers' -type ObserverState o = State (Observers o) +-- | Alias for the effect that contains the observers managed by 'evalObserverRegistryState' +type ObserverRegistryState event = State (ObserverRegistry event) -observers :: Iso' (Observers o) (Set (Observer o)) -observers = iso _observers Observers +-- | An 'Iso' for the 'Map' used internally. +observerRegistry :: Iso' (ObserverRegistry event) (Map ProcessId (ObservationSink event)) +observerRegistry = iso _observerRegistry MkObserverRegistry -- | Report an observation to all observers. --- The process needs to 'manageObservers' and to 'handleObserverRegistration'. +-- The process needs to 'evalObserverRegistryState' and to 'observerRegistryHandlePdu'. -- --- @since 0.16.0 -observed - :: forall o r q - . ( SetMember Process (Process q) r - , Member (ObserverState o) r - , Member Interrupts r - , TangibleObserver o +-- @since 0.28.0 +observerRegistryNotify + :: forall event r q + . ( HasProcesses r q + , Member (ObserverRegistryState event) r + , Tangible event + , HasCallStack ) - => o + => event -> Eff r () -observed observation = do - os <- view observers <$> get - mapM_ notifySomeObserver os +observerRegistryNotify observation = do + os <- view observerRegistry <$> get + mapM_ notifySomeObserver (Map.assocs os) where - notifySomeObserver (Observer messageFilter receiver) = - traverse_ (cast receiver) (messageFilter observation) + notifySomeObserver (destination, (MkObservationSink serializer _)) = + sendAnyMessage destination (runSerializer serializer (Observed observation)) diff --git a/src/Control/Eff/Concurrent/Protocol/Observer/Queue.hs b/src/Control/Eff/Concurrent/Protocol/Observer/Queue.hs --- a/src/Control/Eff/Concurrent/Protocol/Observer/Queue.hs +++ b/src/Control/Eff/Concurrent/Protocol/Observer/Queue.hs @@ -3,11 +3,10 @@ module Control.Eff.Concurrent.Protocol.Observer.Queue ( ObservationQueue(..) , ObservationQueueReader - , readObservationQueue - , tryReadObservationQueue - , flushObservationQueue - , withObservationQueue - , spawnLinkObservationQueueWriter + , observe + , await + , tryRead + , flush ) where @@ -21,100 +20,138 @@ import Control.Eff.Log import Control.Eff.Reader.Strict import Control.Exception.Safe as Safe +import Control.Lens import Control.Monad.IO.Class -import Control.Monad ( unless ) +import Control.Monad ( unless, when ) import qualified Data.Text as T import Data.Typeable import GHC.Stack -- | Contains a 'TBQueue' capturing observations. --- See 'spawnLinkObservationQueueWriter', 'readObservationQueue'. +-- See 'observe'. newtype ObservationQueue a = ObservationQueue (TBQueue a) -- | A 'Reader' for an 'ObservationQueue'. type ObservationQueueReader a = Reader (ObservationQueue a) -logPrefix :: forall o proxy . (HasCallStack, Typeable o) => proxy o -> T.Text +logPrefix :: forall event proxy . (HasCallStack, Typeable event) => proxy event -> T.Text logPrefix px = "observation queue: " <> T.pack (show (typeRep px)) --- | Read queued observations captured and enqueued in the shared 'TBQueue' by 'spawnLinkObservationQueueWriter'. +-- | Read queued observations captured and enqueued in the shared 'ObservationQueue' by 'observe'. +-- -- This blocks until something was captured or an interrupt or exceptions was thrown. For a non-blocking --- variant use 'tryReadObservationQueue' or 'flushObservationQueue'. -readObservationQueue - :: forall o r - . ( Member (ObservationQueueReader o) r +-- variant use 'tryRead' or 'flush'. +-- +-- @since 0.28.0 +await + :: forall event r + . ( Member (ObservationQueueReader event) r , HasCallStack , MonadIO (Eff r) - , Typeable o + , Typeable event , Member Logs r ) - => Eff r o -readObservationQueue = do - ObservationQueue q <- ask @(ObservationQueue o) + => Eff r event +await = do + ObservationQueue q <- ask @(ObservationQueue event) liftIO (atomically (readTBQueue q)) --- | Read queued observations captured and enqueued in the shared 'TBQueue' by 'spawnLinkObservationQueueWriter'. +-- | Read queued observations captured and enqueued in the shared 'ObservationQueue' by 'observe'. +-- -- Return the oldest enqueued 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 +-- Use 'await' to block until an observation is observerRegistryNotify. +-- +-- @since 0.28.0 +tryRead + :: forall event r + . ( Member (ObservationQueueReader event) r , HasCallStack , MonadIO (Eff r) - , Typeable o + , Typeable event , Member Logs r ) - => Eff r (Maybe o) -tryReadObservationQueue = do - ObservationQueue q <- ask @(ObservationQueue o) + => Eff r (Maybe event) +tryRead = do + ObservationQueue q <- ask @(ObservationQueue event) liftIO (atomically (tryReadTBQueue q)) -- | Read at once all currently queued observations captured and enqueued --- in the shared 'TBQueue' by 'spawnLinkObservationQueueWriter'. +-- in the shared 'ObservationQueue' by 'observe'. +-- -- This returns immediately all currently enqueued observations. --- For a blocking variant use 'readObservationQueue'. -flushObservationQueue - :: forall o r - . ( Member (ObservationQueueReader o) r +-- For a blocking variant use 'await'. +-- +-- @since 0.28.0 +flush + :: forall event r + . ( Member (ObservationQueueReader event) r , HasCallStack , MonadIO (Eff r) - , Typeable o + , Typeable event , Member Logs r ) - => Eff r [o] -flushObservationQueue = do - ObservationQueue q <- ask @(ObservationQueue o) + => Eff r [event] +flush = do + ObservationQueue q <- ask @(ObservationQueue event) liftIO (atomically (flushTBQueue q)) --- | Create a mutable queue for observations. Use 'spawnLinkObservationQueueWriter' for a simple way to get --- a process that enqueues all observations. +-- | Listen to, and capture observations in an 'ObservationQueue'. -- +-- Fork an 'Observer' process that runs only while the body expression is executed. +-- Register the observer to the observable process passed to this function. +-- +-- The captured observations can be obtained by 'await', +-- 'tryRead' and 'flush'. +-- +-- The queue size is limited to the given number. +-- -- ==== __Example__ -- -- @ --- withObservationQueue 100 $ do --- q <- ask \@(ObservationQueue TestEvent) --- wq <- spawnLinkObservationQueueWriter q --- registerObserver wq testServer --- ... --- cast testServer DoSomething --- evt <- readObservationQueue \@TestEvent --- ... +-- +-- import qualified Control.Eff.Concurrent.Observer.Queue as OQ +-- +-- foo = +-- do +-- observed <- start SomeObservable +-- OQ.observe 100 observed $ do +-- ... +-- cast observed DoSomething +-- evt <- OQ.await \@TestEvent +-- ... -- @ -- --- @since 0.18.0 +-- @since 0.28.0 +observe + :: forall event eventSource e q len b + . ( HasCallStack + , HasProcesses e q + , LogIo q + , LogIo e + , IsObservable eventSource event + , Integral len + , Server (ObservationQueue event) (Processes q) + ) + => len + -> Endpoint eventSource + -> Eff (ObservationQueueReader event ': e) b + -> Eff e b +observe queueLimit eventSource e = + withObservationQueue queueLimit (withWriter @event eventSource e) + + withObservationQueue - :: forall o b e len + :: forall event b e len . ( HasCallStack - , Typeable o - , Show o + , Typeable event + , Show event , Member Logs e , Lifted IO e , Integral len , Member Interrupts e ) => len - -> Eff (ObservationQueueReader o ': e) b + -> Eff (ObservationQueueReader event ': e) b -> Eff e b withObservationQueue queueLimit e = do q <- lift (newTBQueueIO (fromIntegral queueLimit)) @@ -123,39 +160,75 @@ rest <- lift (atomically (flushTBQueue q)) unless (null rest) - (logNotice (logPrefix (Proxy @o) <> " unread observations: " <> T.pack (show rest))) + (logNotice (logPrefix (Proxy @event) <> " unread observations: " <> T.pack (show rest))) either (\em -> logError (T.pack (show em)) >> lift (throwIO em)) return res -- | Spawn a process that can be used as an 'Observer' that enqueues the observations into an -- 'ObservationQueue'. See 'withObservationQueue' for an example. -- --- The observations can be obtained by 'readObservationQueue'. All observations are captured up to +-- The observations can be obtained by 'await'. All observations are captured up to -- the queue size limit, such that the first message received will be first message --- returned by 'readObservationQueue'. +-- returned by 'await'. -- --- @since 0.18.0 -spawnLinkObservationQueueWriter - :: forall o q h - . ( TangibleObserver o - , HasPdu (Observer o) 'Asynchronous - , Member Logs q +-- @since 0.28.0 +spawnWriter + :: forall event r q h + . ( Member Logs q , Lifted IO q , LogsTo h (Processes q) - , HasCallStack) - => ObservationQueue o - -> Eff (Processes q) (Observer o) -spawnLinkObservationQueueWriter q = do - cbo <- startLink (MkObservationQueue q) - pure (toObserver cbo) + , HasProcesses r q + , Typeable event + , HasCallStack + , Server (ObservationQueue event) (Processes q) + ) + => ObservationQueue event + -> Eff r (Endpoint (Observer event)) +spawnWriter q = + start @_ @r @q @h (MkObservationQueue q) -instance (TangibleObserver o, HasPdu (Observer o) 'Asynchronous, Lifted IO q, Member Logs q) => Server (ObservationQueue o) (Processes q) where - type Protocol (ObservationQueue o) = Observer o +-- | Spawn a process that can be used as an 'Observer' that enqueues the observations into an +-- 'ObservationQueue'. See 'withObservationQueue'. +-- +-- The observations can be obtained by 'await'. All observations are captured up to +-- the queue size limit, such that the first message received will be first message +-- returned by 'await'. +-- +-- @since 0.28.0 +withWriter + :: forall event eventSource e q b + . ( HasCallStack + , HasProcesses e q + , LogIo q + , IsObservable eventSource event + , Member (ObservationQueueReader event) e + ) + => Endpoint eventSource + -> Eff e b + -> Eff e b +withWriter eventSource e = do + q <- ask @(ObservationQueue event) + w <- spawnWriter @event q + registerObserver @event eventSource w + res <- e + forgetObserver @event eventSource w + sendShutdown (w^.fromEndpoint) ExitNormally + pure res - data instance StartArgument (ObservationQueue o) (Processes q) = - MkObservationQueue (ObservationQueue o) +instance (Typeable event, Lifted IO q, Member Logs q) => Server (ObservationQueue event) (Processes q) where + type Protocol (ObservationQueue event) = Observer event + + data instance StartArgument (ObservationQueue event) (Processes q) = + MkObservationQueue (ObservationQueue event) + update _ (MkObservationQueue (ObservationQueue q)) = \case - OnCast r -> - handleObservations (lift . atomically . writeTBQueue q) r - otherMsg -> logError ("unexpected: " <> T.pack (show otherMsg)) + OnCast (Observed !event) -> do + isFull <- lift . atomically $ do + isFull <- isFullTBQueue q + unless isFull (writeTBQueue q event) + pure isFull + when isFull $ + logWarning "queue full" + otherMsg -> + logError ("unexpected: " <> T.pack (show otherMsg)) diff --git a/src/Control/Eff/Concurrent/Protocol/StatefulServer.hs b/src/Control/Eff/Concurrent/Protocol/StatefulServer.hs --- a/src/Control/Eff/Concurrent/Protocol/StatefulServer.hs +++ b/src/Control/Eff/Concurrent/Protocol/StatefulServer.hs @@ -116,28 +116,30 @@ -- -- @since 0.24.0 start - :: forall a q h + :: forall a r q h . ( HasCallStack , Typeable a , LogsTo h (Processes q) , Effectful.Server (Stateful a) (Processes q) , Server a (Processes q) + , HasProcesses r q ) - => StartArgument a (Processes q) -> Eff (Processes q) (Endpoint (Protocol a)) + => StartArgument a (Processes q) -> Eff r (Endpoint (Protocol a)) start = Effectful.start . Init -- | Execute the server loop. -- -- @since 0.24.0 startLink - :: forall a q h + :: forall a r q h . ( HasCallStack , Typeable a , LogsTo h (Processes q) , Effectful.Server (Stateful a) (Processes q) , Server a (Processes q) + , HasProcesses r q ) - => StartArgument a (Processes q) -> Eff (Processes q) (Endpoint (Protocol a)) + => StartArgument a (Processes q) -> Eff r (Endpoint (Protocol a)) startLink = Effectful.startLink . Init diff --git a/src/Control/Eff/Concurrent/Protocol/Supervisor.hs b/src/Control/Eff/Concurrent/Protocol/Supervisor.hs --- a/src/Control/Eff/Concurrent/Protocol/Supervisor.hs +++ b/src/Control/Eff/Concurrent/Protocol/Supervisor.hs @@ -199,20 +199,12 @@ Just existingChild -> sendReply rt (Left (AlreadyStarted i (existingChild ^. childEndpoint))) - update _ _supConfig (OnDown (ProcessDown mrChild reason)) = do - oldEntry <- lookupAndRemoveChildByMonitor @(ChildId p) @p mrChild + update _ _supConfig (OnDown pd) = do + oldEntry <- lookupAndRemoveChildByMonitor @(ChildId p) @p (downReference pd) case oldEntry of - Nothing -> - logWarning ("unexpected process down: " - <> pack (show mrChild) - <> " reason: " - <> pack (show reason) - ) + Nothing -> logWarning ("unexpected: " <> pack (show pd)) Just (i, c) -> do - logInfo ( "process down: " - <> pack (show mrChild) - <> " reason: " - <> pack (show reason) + logInfo ( pack (show pd) <> " for child " <> pack (show i) <> " => " @@ -270,8 +262,7 @@ -- @since 0.23.0 stopSupervisor :: ( HasCallStack - , Member Interrupts e - , SetMember Process (Process q0) e + , HasProcesses e q , Member Logs e , Lifted IO e , TangibleSup p @@ -291,10 +282,9 @@ isSupervisorAlive :: forall p q0 e . ( HasCallStack - , Member Interrupts e , Member Logs e , Typeable p - , SetMember Process (Process q0) e + , HasProcesses e q0 ) => Endpoint (Sup p) -> Eff e Bool @@ -306,9 +296,8 @@ monitorSupervisor :: forall p q0 e . ( HasCallStack - , Member Interrupts e , Member Logs e - , SetMember Process (Process q0) e + , HasProcesses e q0 , TangibleSup p ) => Endpoint (Sup p) @@ -322,9 +311,8 @@ spawnChild :: forall p q0 e . ( HasCallStack - , Member Interrupts e , Member Logs e - , SetMember Process (Process q0) e + , HasProcesses e q0 , TangibleSup p , Typeable (Protocol p) ) @@ -340,9 +328,8 @@ lookupChild :: forall p e q0 . ( HasCallStack - , Member Interrupts e , Member Logs e - , SetMember Process (Process q0) e + , HasProcesses e q0 , TangibleSup p , Typeable (Protocol p) ) @@ -360,9 +347,8 @@ stopChild :: forall p e q0 . ( HasCallStack - , Member Interrupts e , Member Logs e - , SetMember Process (Process q0) e + , HasProcesses e q0 , TangibleSup p ) => Endpoint (Sup p) @@ -376,8 +362,7 @@ getDiagnosticInfo :: forall p e q0 . ( HasCallStack - , Member Interrupts e - , SetMember Process (Process q0) e + , HasProcesses e q0 , TangibleSup p ) => Endpoint (Sup p) @@ -389,8 +374,7 @@ stopOrKillChild :: forall p e q0 . ( HasCallStack - , SetMember Process (Process q0) e - , Member Interrupts e + , HasProcesses e q0 , Lifted IO e , Lifted IO q0 , Member Logs e @@ -424,7 +408,7 @@ stopAllChildren :: forall p e q0 . ( HasCallStack - , SetMember Process (Process q0) e + , HasProcesses e q0 , Lifted IO e , Lifted IO q0 , Member Logs e diff --git a/src/Control/Eff/Concurrent/Protocol/Wrapper.hs b/src/Control/Eff/Concurrent/Protocol/Wrapper.hs --- a/src/Control/Eff/Concurrent/Protocol/Wrapper.hs +++ b/src/Control/Eff/Concurrent/Protocol/Wrapper.hs @@ -90,12 +90,7 @@ -- | Create a new, unique 'RequestOrigin' value for the current process. -- -- @since 0.24.0 -makeRequestOrigin - :: ( Typeable r - , NFData r - , SetMember Process (Process q0) e - , '[Interrupts] <:: e) - => Eff e (RequestOrigin p r) +makeRequestOrigin :: (Typeable r, NFData r, HasProcesses e q0) => Eff e (RequestOrigin p r) makeRequestOrigin = RequestOrigin <$> self <*> makeReference instance NFData (RequestOrigin p r) @@ -154,8 +149,7 @@ -- -- @since 0.25.1 sendReply - :: ( SetMember Process (Process q) eff - , Member Interrupts eff + :: ( HasProcesses eff q , Tangible reply , Typeable protocol ) diff --git a/test/Common.hs b/test/Common.hs --- a/test/Common.hs +++ b/test/Common.hs @@ -8,7 +8,7 @@ import Control.Eff.Extend import Control.Monad (void) import GHC.Stack -import Test.Tasty +import Test.Tasty import Test.Tasty.HUnit import Test.Tasty.Runners @@ -58,3 +58,9 @@ applySchedulerFactory factory procAction = do scheduler <- factory scheduler (procAction >> lift (threadDelay 20000)) + +awaitProcessDown :: (HasCallStack, HasProcesses r q) => ProcessId -> Eff r ProcessDown +awaitProcessDown p = do + m <- monitor p + receiveSelectedMessage (selectProcessDown m) + diff --git a/test/GenServerTests.hs b/test/GenServerTests.hs --- a/test/GenServerTests.hs +++ b/test/GenServerTests.hs @@ -75,7 +75,7 @@ showsPrec d (BigCast x) = showParen (d > 10) (showString "SmallCast " . showString x) showsPrec d (BigSmall x) = showParen (d > 10) (showString "BigSmall " . showsPrec 11 x) -instance EmbedProtocol Big Small r where +instance Typeable r => EmbedProtocol Big Small r where embeddedPdu = prism' BigSmall @@ -94,11 +94,11 @@ BigCall o -> do logNotice ("BigCall " <> pack (show o)) sendReply rt o - BigSmall x -> S.update (toEmbeddedEndpoint me) MkSmall (S.OnCall (toEmbeddedReplyTarget rt) x) + BigSmall x -> S.update (toEmbeddedEndpoint @_ @_ @('Synchronous Bool) me) MkSmall (S.OnCall (toEmbeddedReplyTarget rt) x) E.OnCast req -> case req of BigCast o -> S.putModel @Big o - BigSmall x -> S.update (toEmbeddedEndpoint me) MkSmall (S.OnCast x) + BigSmall x -> S.update (toEmbeddedEndpoint @_ @_ @('Asynchronous) me) MkSmall (S.OnCast x) other -> interrupt (ErrorInterrupt (show other)) -- ---------------------------------------------------------------------------- diff --git a/test/ObserverTests.hs b/test/ObserverTests.hs new file mode 100644 --- /dev/null +++ b/test/ObserverTests.hs @@ -0,0 +1,343 @@ +{-# LANGUAGE UndecidableInstances #-} +module ObserverTests + ( test_observer + ) where + +import Common +import Control.DeepSeq +import Control.Eff +import Control.Eff.Concurrent +import qualified Control.Eff.Concurrent.Protocol.EffectfulServer as S +import qualified Control.Eff.Concurrent.Protocol.Observer.Queue as OQ +import qualified Control.Eff.Concurrent.Protocol.StatefulServer as M +import Control.Lens +import Control.Monad +import Data.Text as T +import Data.Typeable (Typeable) +import Test.Tasty +import Test.Tasty.HUnit + + +test_observer :: HasCallStack => TestTree +test_observer = testGroup "observer" (basicTests ++ [observerQueueTests]) + + +basicTests :: HasCallStack => [TestTree] +basicTests = + [runTestCase "when no observer is present, nothing crashes" + $ do + testObservable <- S.start TestObservableServerInit + call testObservable (SendTestEvent "1") + cast testObservable StopTestObservable + + void $ awaitProcessDown (testObservable ^. fromEndpoint) + + , runTestCase "observers receive only messages sent after registration" + $ do + testObservable <- S.start TestObservableServerInit + call testObservable (SendTestEvent "1") + call testObservable (SendTestEvent "2") + call testObservable (SendTestEvent "3") + testObserver <- M.start MkTestObserver + registerObserver @String testObservable testObserver + call testObservable (SendTestEvent "4") + call testObservable (SendTestEvent "5") + es <- call testObserver GetCapturedEvents + lift (["4", "5"] @=? es) + cast testObservable StopTestObservable + void $ awaitProcessDown (testObservable ^. fromEndpoint) + , runTestCase "observers receive only messages sent before de-registration" + $ do + testObservable <- S.start TestObservableServerInit + call testObservable (SendTestEvent "1") + call testObservable (SendTestEvent "2") + call testObservable (SendTestEvent "3") + testObserver <- M.start MkTestObserver + registerObserver @String testObservable testObserver + call testObservable (SendTestEvent "4") + call testObservable (SendTestEvent "5") + forgetObserver @String testObservable testObserver + call testObservable (SendTestEvent "6") + call testObservable (SendTestEvent "7") + es <- call testObserver GetCapturedEvents + lift (["4", "5"] @=? es) + cast testObservable StopTestObservable + void $ awaitProcessDown (testObservable ^. fromEndpoint) + , runTestCase "observers receive only messages sent between registration and deregistration" + $ do + testObservable <- S.start TestObservableServerInit + call testObservable (SendTestEvent "1") + call testObservable (SendTestEvent "2") + call testObservable (SendTestEvent "3") + testObserver <- M.start MkTestObserver + registerObserver @String testObservable testObserver + call testObservable (SendTestEvent "4") + call testObservable (SendTestEvent "5") + forgetObserver @String testObservable testObserver + call testObservable (SendTestEvent "6") + registerObserver @String testObservable testObserver + call testObservable (SendTestEvent "7") + es <- call testObserver GetCapturedEvents + lift (["4", "5", "7"] @=? es) + cast testObservable StopTestObservable + void $ awaitProcessDown (testObservable ^. fromEndpoint) + , runTestCase "all observers receive all messages sent between registration and deregistration" + $ do + testObservable <- S.start TestObservableServerInit + call testObservable (SendTestEvent "1") + call testObservable (SendTestEvent "2") + call testObservable (SendTestEvent "3") + testObserver1 <- M.start MkTestObserver + testObserver2 <- M.start MkTestObserver + testObserver3 <- M.start MkTestObserver + registerObserver @String testObservable testObserver1 + call testObservable (SendTestEvent "4") + registerObserver @String testObservable testObserver2 + call testObservable (SendTestEvent "5") + registerObserver @String testObservable testObserver3 + call testObservable (SendTestEvent "6") + forgetObserver @String testObservable testObserver1 + call testObservable (SendTestEvent "7") + forgetObserver @String testObservable testObserver2 + call testObservable (SendTestEvent "8") + forgetObserver @String testObservable testObserver3 + call testObservable (SendTestEvent "9") + es1 <- call testObserver1 GetCapturedEvents + lift (["4", "5", "6"] @=? es1) + es2 <- call testObserver2 GetCapturedEvents + lift (["5", "6", "7"] @=? es2) + es3 <- call testObserver3 GetCapturedEvents + lift (["6", "7", "8"] @=? es3) + cast testObservable StopTestObservable + void $ awaitProcessDown (testObservable ^. fromEndpoint) + , runTestCase "when an observer exits, the messages are still deliviered to the others" + $ do + testObservable <- S.start TestObservableServerInit + call testObservable (SendTestEvent "1") + call testObservable (SendTestEvent "2") + call testObservable (SendTestEvent "3") + testObserver1 <- M.start MkTestObserver + testObserver2 <- M.start MkTestObserver + registerObserver @String testObservable testObserver1 + registerObserver @String testObservable testObserver2 + call testObservable (SendTestEvent "4") + call testObservable (SendTestEvent "5") + sendShutdown (testObserver2^.fromEndpoint) ExitNormally + void $ awaitProcessDown (testObserver2^.fromEndpoint) + call testObservable (SendTestEvent "6") + es1 <- call testObserver1 GetCapturedEvents + lift (["4", "5", "6"] @=? es1) + cast testObservable StopTestObservable + void $ awaitProcessDown (testObservable ^. fromEndpoint) + , runTestCase "when an observer registers multiple times, it still gets the messages only once" + $ do + testObservable <- S.start TestObservableServerInit + call testObservable (SendTestEvent "1") + call testObservable (SendTestEvent "2") + call testObservable (SendTestEvent "3") + testObserver1 <- M.start MkTestObserver + registerObserver @String testObservable testObserver1 + registerObserver @String testObservable testObserver1 + call testObservable (SendTestEvent "4") + call testObservable (SendTestEvent "5") + call testObservable (SendTestEvent "6") + es1 <- call testObserver1 GetCapturedEvents + lift (["4", "5", "6"] @=? es1) + cast testObservable StopTestObservable + void $ awaitProcessDown (testObservable ^. fromEndpoint) + , runTestCase "when an observer is forgotton multiple times, nothing bad happens" + $ do + testObservable <- S.start TestObservableServerInit + call testObservable (SendTestEvent "1") + call testObservable (SendTestEvent "2") + call testObservable (SendTestEvent "3") + testObserver1 <- M.start MkTestObserver + registerObserver @String testObservable testObserver1 + call testObservable (SendTestEvent "4") + call testObservable (SendTestEvent "5") + forgetObserver @String testObservable testObserver1 + forgetObserver @String testObservable testObserver1 + forgetObserver @String testObservable testObserver1 + call testObservable (SendTestEvent "6") + es1 <- call testObserver1 GetCapturedEvents + lift (["4", "5"] @=? es1) + cast testObservable StopTestObservable + void $ awaitProcessDown (testObservable ^. fromEndpoint) + ] + +observerQueueTests :: TestTree +observerQueueTests = + testGroup "observer-queue" + [ runTestCase "tryRead" + $ do + testObservable <- S.start TestObservableServerInit + let len :: Int + len = 1 + OQ.observe @String len testObservable $ do + OQ.tryRead @String >>= lift . (Nothing @=?) + call testObservable (SendTestEvent "1") + OQ.tryRead @String >>= lift . (Just "1" @=?) + + cast testObservable StopTestObservable + void $ awaitProcessDown (testObservable ^. fromEndpoint) + , runTestCase "observe then read" + $ do + testObservable <- S.start TestObservableServerInit + let len :: Int + len = 1 + OQ.observe @String len testObservable $ do + call testObservable (SendTestEvent "1") + OQ.await @String >>= lift . ("1" @=?) + call testObservable (SendTestEvent "2") + OQ.await @String >>= lift . ("2" @=?) + call testObservable (SendTestEvent "3") + OQ.await @String >>= lift . ("3" @=?) + + cast testObservable StopTestObservable + void $ awaitProcessDown (testObservable ^. fromEndpoint) + , runTestCase "FIFO" + $ do + testObservable <- S.start TestObservableServerInit + let len :: Int + len = 3 + OQ.observe @String len testObservable $ do + call testObservable (SendTestEvent "1") + call testObservable (SendTestEvent "2") + call testObservable (SendTestEvent "3") + OQ.await @String >>= lift . ("1" @=?) + OQ.await @String >>= lift . ("2" @=?) + OQ.await @String >>= lift . ("3" @=?) + + cast testObservable StopTestObservable + void $ awaitProcessDown (testObservable ^. fromEndpoint) + , runTestCase "flush" + $ do + testObservable <- S.start TestObservableServerInit + let len :: Int + len = 3 + OQ.observe @String len testObservable $ do + call testObservable (SendTestEvent "1") + call testObservable (SendTestEvent "2") + call testObservable (SendTestEvent "3") + OQ.flush @String >>= lift . (["1", "2", "3"] @=?) + OQ.tryRead @String >>= lift . (Nothing @=?) + + cast testObservable StopTestObservable + void $ awaitProcessDown (testObservable ^. fromEndpoint) + , runTestCase "when the queue is full, new observations are dropped" + $ do + testObservable <- S.start TestObservableServerInit + let len :: Int + len = 2 + OQ.observe @String len testObservable $ do + call testObservable (SendTestEvent "1") + call testObservable (SendTestEvent "2") + call testObservable (SendTestEvent "3") + call testObservable (SendTestEvent "4") + OQ.await @String >>= lift . ("1" @=?) + OQ.await @String >>= lift . ("2" @=?) + OQ.tryRead @String >>= lift . (Nothing @=?) + + cast testObservable StopTestObservable + void $ awaitProcessDown (testObservable ^. fromEndpoint) + , runTestCase "flush after queue full" + $ do + testObservable <- S.start TestObservableServerInit + let len :: Int + len = 3 + OQ.observe @String len testObservable $ do + call testObservable (SendTestEvent "1") + call testObservable (SendTestEvent "2") + call testObservable (SendTestEvent "3") + call testObservable (SendTestEvent "4") + call testObservable (SendTestEvent "5") + OQ.flush @String >>= lift . (["1", "2", "3"] @=?) + OQ.flush @String >>= lift . ([] @=?) + OQ.tryRead @String >>= lift . (Nothing @=?) + call testObservable (SendTestEvent "6") + OQ.await @String >>= lift . ("6" @=?) + + cast testObservable StopTestObservable + void $ awaitProcessDown (testObservable ^. fromEndpoint) + ] + + +data TestObservable deriving Typeable + +instance Typeable r => HasPdu TestObservable r where + data Pdu TestObservable r where + SendTestEvent :: String -> Pdu TestObservable ('Synchronous ()) + StopTestObservable :: Pdu TestObservable 'Asynchronous + TestObsReg :: Pdu (ObserverRegistry String) 'Asynchronous -> Pdu TestObservable 'Asynchronous + deriving (Typeable) + +instance EmbedProtocol TestObservable (ObserverRegistry String) 'Asynchronous where + embeddedPdu = prism' TestObsReg $ + \case + TestObsReg e -> Just e + _ -> Nothing + +instance Typeable r => NFData (Pdu TestObservable r) where + rnf (SendTestEvent s) = rnf s + rnf StopTestObservable = () + rnf (TestObsReg x) = rnf x + +instance Typeable r => Show (Pdu TestObservable r) where + show (SendTestEvent x) = "SendTestEvent " ++ show x + show StopTestObservable = "StopTestObservable" + show (TestObsReg x) = "TestObsReg " ++ show x + +instance (LogIo r, HasProcesses r q) => S.Server TestObservable r where + data Init TestObservable r = TestObservableServerInit + type ServerEffects TestObservable r = ObserverRegistryState String ': r + runEffects _ _ = evalObserverRegistryState + onEvent _ _ = + \case + S.OnCall rt e -> + case e of + SendTestEvent x -> observerRegistryNotify x >> sendReply rt () + S.OnCast (TestObsReg x) -> observerRegistryHandlePdu x + S.OnCast StopTestObservable -> exitNormally + S.OnDown pd -> do + logDebug ("inspecting: " <> pack (show pd)) + wasHandled <- observerRegistryRemoveProcess @String (downProcess pd) + unless wasHandled $ + logError ("the process down message was not handled: " <> pack (show pd)) + other -> + logError ("unexpected: " <> pack (show other)) + + +data TestObserver deriving Typeable + +instance Typeable r => HasPdu TestObserver r where + data Pdu TestObserver r where + GetCapturedEvents :: Pdu TestObserver ('Synchronous [String]) + OnTestEvent :: Pdu (Observer String) 'Asynchronous -> Pdu TestObserver 'Asynchronous + deriving Typeable + +instance NFData (Pdu TestObserver r) where + rnf GetCapturedEvents = () + rnf (OnTestEvent e) = rnf e + +instance Show (Pdu TestObserver r) where + showsPrec _ GetCapturedEvents = showString "GetCapturedEvents" + showsPrec d (OnTestEvent e) = showParen (d>=10) (showString "OnTestEvent " . shows e) + +instance EmbedProtocol TestObserver (Observer String) 'Asynchronous where + embeddedPdu = prism' OnTestEvent $ + \case + OnTestEvent e -> Just e + + +instance (LogIo r, HasProcesses r q) => M.Server TestObserver r where + data StartArgument TestObserver r = MkTestObserver + type Model TestObserver = [String] + setup _ MkTestObserver = pure ([], ()) + update _ MkTestObserver e = + case e of + M.OnCall rt GetCapturedEvents -> + M.getAndPutModel @TestObserver [] >>= sendReply rt + M.OnCast (OnTestEvent (Observed x)) -> + M.modifyModel @TestObserver (++ [x]) + _ -> + logError ("unexpected: " <> pack (show e)) diff --git a/test/ProcessBehaviourTestCases.hs b/test/ProcessBehaviourTestCases.hs --- a/test/ProcessBehaviourTestCases.hs +++ b/test/ProcessBehaviourTestCases.hs @@ -107,7 +107,7 @@ returnToSender :: forall q r - . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r) + . (HasCallStack, HasProcesses r q) => Endpoint ReturnToSender -> String -> Eff r Bool @@ -119,7 +119,7 @@ stopReturnToSender :: forall q r - . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r) + . (HasCallStack, HasProcesses r q) => Endpoint ReturnToSender -> Eff r () stopReturnToSender toP = call toP StopReturnToSender @@ -821,12 +821,12 @@ l <- receiveMessage _ <- monitor l sendMessage l () - pL@(ProcessDown _ _) <- receiveMessage + pL@(ProcessDown _ _ _) <- receiveMessage logCritical' ("linked process down: " <> show pL) _ <- monitor u mpU <- receiveAfter (TimeoutMicros 1000) case mpU of - Just (pU@(ProcessDown _ _)) -> + Just (pU@(ProcessDown _ _ _)) -> error ("unlinked process down: " <> show pU) Nothing -> logInfo "passed" @@ -845,12 +845,12 @@ l <- receiveMessage _ <- monitor l sendMessage l () - pL@(ProcessDown _ _) <- receiveMessage + pL@(ProcessDown _ _ _) <- receiveMessage logCritical' ("linked process down: " <> show pL) _ <- monitor u mpU <- receiveAfter (TimeoutMicros 1000) case mpU of - Just (pU@(ProcessDown _ _)) -> + Just (pU@(ProcessDown _ _ _)) -> logInfo' ("unlinked process down: " <> show pU) Nothing -> error "linked process not exited!" diff --git a/test/SupervisorTests.hs b/test/SupervisorTests.hs --- a/test/SupervisorTests.hs +++ b/test/SupervisorTests.hs @@ -47,11 +47,11 @@ logInfo ("still alive 2: " <> pack (show supAliveAfter2)) lift (supAliveAfter2 @=? True) sendMessage testWorker () - _ <- monitor testWorker - d1@(ProcessDown _ _) <- receiveMessage + testWorkerMonitorRef <- monitor testWorker + d1 <- receiveSelectedMessage (selectProcessDown testWorkerMonitorRef) logInfo ("got test worker down: " <> pack (show d1)) - _ <- monitorSupervisor sup - d2@(ProcessDown _ _) <- receiveMessage + testSupervisorMonitorRef <- monitorSupervisor sup + d2 <- receiveSelectedMessage (selectProcessDown testSupervisorMonitorRef) logInfo ("got supervisor down: " <> pack (show d2)) supAliveAfterOwnerExited <- isSupervisorAlive sup logInfo ("still alive after owner exited: " <> pack (show supAliveAfterOwnerExited)) @@ -60,19 +60,19 @@ "Diagnostics" [ runTestCase "When only time passes the diagnostics do not change" $ do sup <- startTestSup - diag1 <- Sup.getDiagnosticInfo sup + info1 <- Sup.getDiagnosticInfo sup lift (threadDelay 10000) - diag2 <- Sup.getDiagnosticInfo sup - lift (assertEqual "diagnostics should not differ: " diag1 diag2) + info2 <- Sup.getDiagnosticInfo sup + lift (assertEqual "diagnostics should not differ: " info1 info2) , runTestCase "When a child is started the diagnostics change" $ do sup <- startTestSup - diag1 <- Sup.getDiagnosticInfo sup - logInfo ("got diagnostics: " <> diag1) + info1 <- Sup.getDiagnosticInfo sup + logInfo ("got diagnostics: " <> info1) let childId = 1 _child <- fromRight (error "failed to spawn child") <$> Sup.spawnChild sup childId - diag2 <- Sup.getDiagnosticInfo sup - logInfo ("got diagnostics: " <> diag2) - lift $ assertBool ("diagnostics should differ: " ++ show (diag1, diag2)) (diag1 /= diag2) + info2 <- Sup.getDiagnosticInfo sup + logInfo ("got diagnostics: " <> info2) + lift $ assertBool ("diagnostics should differ: " ++ show (info1, info2)) (info1 /= info2) ] , let childId = 1 in testGroup @@ -87,10 +87,10 @@ isSupervisorAlive sup >>= lift . assertBool "supervisor process not running" call child (TestGetStringLength "123") >>= lift . assertEqual "child not working" 3 stopSupervisor sup - d1@(ProcessDown mon1 er1) <- + d1@(ProcessDown mon1 er1 _) <- fromMaybe (error "receive timeout 1") <$> receiveAfter (TimeoutMicros 1000000) logInfo ("got process down: " <> pack (show d1)) - d2@(ProcessDown mon2 er2) <- + d2@(ProcessDown mon2 er2 _) <- fromMaybe (error "receive timeout 2") <$> receiveAfter (TimeoutMicros 1000000) logInfo ("got process down: " <> pack (show d2)) case if mon1 == supMon && mon2 == childMon @@ -119,10 +119,10 @@ isSupervisorAlive sup >>= lift . assertBool "supervisor process not running" call child (TestGetStringLength "123") >>= lift . assertEqual "child not working" 3 stopSupervisor sup - d1@(ProcessDown mon1 er1) <- + d1@(ProcessDown mon1 er1 _) <- fromMaybe (error "receive timeout 1") <$> receiveAfter (TimeoutMicros 1000000) logInfo ("got process down: " <> pack (show d1)) - d2@(ProcessDown mon2 er2) <- + d2@(ProcessDown mon2 er2 _) <- fromMaybe (error "receive timeout 2") <$> receiveAfter (TimeoutMicros 1000000) logInfo ("got process down: " <> pack (show d2)) case if mon1 == supMon && mon2 == childMon @@ -180,7 +180,7 @@ [ runTestCase "When a child is started it can be stopped" $ do (sup, cm) <- startTestSupAndChild Sup.stopChild sup i >>= lift . assertBool "child not found" - (ProcessDown _ r) <- receiveSelectedMessage (selectProcessDown cm) + (ProcessDown _ r _) <- receiveSelectedMessage (selectProcessDown cm) lift (assertEqual "bad exit reason" (SomeExitReason ExitNormally) r) , runTestCase "When a child is stopped but doesn't exit voluntarily, it is kill after some time" $ do @@ -188,7 +188,7 @@ c <- spawnTestChild sup i cm <- monitor (_fromEndpoint c) Sup.stopChild sup i >>= lift . assertBool "child not found" - (ProcessDown _ r) <- receiveSelectedMessage (selectProcessDown cm) + (ProcessDown _ r _) <- receiveSelectedMessage (selectProcessDown cm) case r of SomeExitReason (ExitUnhandledError _) -> return () _ -> lift (assertFailure ("bad exit reason: " ++ show r)) @@ -215,26 +215,26 @@ [ runTestCase "When a child exits normally, lookupChild will not find it" $ do (sup, c, cm) <- startTestSupAndChild cast c (TestInterruptWith NormalExitRequested) - (ProcessDown _ r) <- receiveSelectedMessage (selectProcessDown cm) + (ProcessDown _ r _) <- receiveSelectedMessage (selectProcessDown cm) lift (assertEqual "bad exit reason" (SomeExitReason ExitNormally) r) x <- Sup.lookupChild sup i lift (assertEqual "lookup should not find a child" Nothing x) , runTestCase "When a child exits with an error, lookupChild will not find it" $ do (sup, c, cm) <- startTestSupAndChild cast c (TestInterruptWith (ErrorInterrupt "test error reason")) - (ProcessDown _ _) <- receiveSelectedMessage (selectProcessDown cm) + (ProcessDown _ _ _) <- receiveSelectedMessage (selectProcessDown cm) x <- Sup.lookupChild sup i lift (assertEqual "lookup should not find a child" Nothing x) , runTestCase "When a child is interrupted from another process and dies, lookupChild will not find it" $ do (sup, c, cm) <- startTestSupAndChild sendInterrupt (_fromEndpoint c) NormalExitRequested - (ProcessDown _ _) <- receiveSelectedMessage (selectProcessDown cm) + (ProcessDown _ _ _) <- receiveSelectedMessage (selectProcessDown cm) x <- Sup.lookupChild sup i lift (assertEqual "lookup should not find a child" Nothing x) , runTestCase "When a child is shutdown from another process and dies, lookupChild will not find it" $ do (sup, c, cm) <- startTestSupAndChild sendShutdown (_fromEndpoint c) ExitProcessCancelled - (ProcessDown _ _) <- receiveSelectedMessage (selectProcessDown cm) + (ProcessDown _ _ _) <- receiveSelectedMessage (selectProcessDown cm) x <- Sup.lookupChild sup i lift (assertEqual "lookup should not find a child" Nothing x) ]