diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,5 +1,19 @@ # Changelog for extensible-effects-concurrent +## 0.32.0 + +- **Protocol-Server** + - Remove effect parameter from `StartArgument` and `Init` + +- **ForkIO Scheduler** + - Fix monitor reference leak + - Shorten the process detail output, + and return it from `getProcessState` + +- **Async Logging** + - Fix the Asynchronous LogWriter so it does not stop logging after a flood of log messages + + ## 0.31.0 - **Logging** 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 @@ -97,7 +97,7 @@ , Maybe (ReplyTarget SupiCounter (Maybe ())) ) - data instance StartArgument SupiCounter (Processes q) = MkEmptySupiCounter + data instance StartArgument SupiCounter = MkEmptySupiCounter setup _ _ = return (SupiCounterModel (0, emptyObserverRegistry, Nothing), ()) @@ -160,7 +160,7 @@ logCounterObservations = startLink OCCStart instance Member Logs q => Server (Observer CounterChanged) (Processes q) where - data instance StartArgument (Observer CounterChanged) (Processes q) = OCCStart + data instance StartArgument (Observer CounterChanged) = OCCStart newtype instance Model (Observer CounterChanged) = CounterChangedModel () deriving Default update _ _ e = case e of diff --git a/examples/example-embedded-protocols/Main.hs b/examples/example-embedded-protocols/Main.hs --- a/examples/example-embedded-protocols/Main.hs +++ b/examples/example-embedded-protocols/Main.hs @@ -78,7 +78,7 @@ instance Stateful.Server App Effects where newtype instance Model App = MkApp (Maybe SomeBackend) deriving Default - data instance StartArgument App Effects = InitApp + data instance StartArgument App = InitApp update me _x e = case e of OnCall rt (SetBackend b) -> do @@ -203,7 +203,7 @@ instance Stateful.Server Backend1 Effects where type instance Protocol Backend1 = (Backend, ObserverRegistry BackendEvent) newtype instance Model Backend1 = MkBackend1 (Int, ObserverRegistry BackendEvent) - data instance StartArgument Backend1 Effects = InitBackend1 + data instance StartArgument Backend1 = InitBackend1 setup _ _ = pure ( MkBackend1 (0, emptyObserverRegistry), () ) update me _ e = do model <- getModel @Backend1 @@ -259,7 +259,7 @@ instance Effectful.Server Backend2 Effects where type instance ServerEffects Backend2 Effects = State Int ': ObserverRegistryState BackendEvent ': Effects - data instance Init Backend2 Effects = InitBackend2 Int + data instance Init Backend2 = InitBackend2 Int serverTitle (InitBackend2 x) = fromString ("backend-2: " ++ show x) runEffects _me _ e = evalObserverRegistryState (evalState 0 e) onEvent me _ e = do 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.31.0 +version: 0.32.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 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 @@ -107,24 +107,24 @@ makeLenses ''SchedulerState -logSchedulerState :: (HasCallStack, Member Logs e, Lifted IO e) => SchedulerState -> Eff e () -logSchedulerState s = do - (np, pt, pct, pm, nm) <- lift $ atomically $ do +renderSchedulerState :: SchedulerState -> IO ProcessDetails +renderSchedulerState s = do + (np, pt, pct, pm, nm) <- atomically $ do np <- T.pack . show <$> readTVar (s ^. nextPid) - pt <- fmap (T.pack . show) <$> readTVar (s ^. processTable) - pct <- fmap (T.pack . show) . Map.keys <$> readTVar (s ^. processCancellationTable) - pm <- fmap (T.pack . show) . Set.toList <$> readTVar (s ^. processMonitors) + pt <- T.pack . show . Map.size <$> readTVar (s ^. processTable) + pct <- T.pack . show . Map.size <$> readTVar (s ^. processCancellationTable) + pm <- T.pack . show . Set.size <$> readTVar (s ^. processMonitors) nm <- T.pack . show <$> readTVar (s ^. nextMonitorIndex) return (np, pt, pct, pm, nm) - logDebug "ForkIO Scheduler info" - logDebug $ "nextPid: " <> np - logDebug "process table:" - traverse_ logDebug pt - logDebug "process cancellation table:" - traverse_ logDebug pct - logDebug "process monitors:" - traverse_ logDebug pm - logDebug $ "nextMonitorIndex: " <> nm + return + $ MkProcessDetails + $ T.unlines + [ "ForkIO Scheduler nextPid: " <> np + , "ForkIO Scheduler process table entries: " <> pt + , "ForkIO Scheduler process cancellation table entries: " <> pct + , "ForkIO Scheduler process monitors entries: " <> pm + , "ForkIO Scheduler nextMonitorIndex: " <> nm + ] -- | Allocate a new 'MonitorReference' nextMonitorReference :: ProcessId -> SchedulerState -> STM MonitorReference @@ -171,6 +171,11 @@ triggerAndRemoveMonitor :: ProcessId -> Interrupt 'NoRecovery -> SchedulerState -> STM [ProcessId] triggerAndRemoveMonitor downPid reason schedulerState = do + -- remove the monitor entries that the downPid process owned: + modifyTVar' (schedulerState ^. processMonitors) + (Set.filter (\(_, downPid') -> downPid' /= downPid)) + -- now send the process down message and remove the entries that monitor + -- the downPid process: monRefs <- readTVar (schedulerState ^. processMonitors) catMaybes <$> traverse go (toList monRefs) where @@ -252,7 +257,7 @@ void (liftBaseWith (\runS -> timeout - 5000000 + 5_000_000 ( Async.mapConcurrently (\a -> do Async.cancel a @@ -526,14 +531,16 @@ interpretGetProcessState !toPid = do setMyProcessState ProcessBusy schedulerState <- getSchedulerState - when (toPid == 1) $ - logSchedulerState schedulerState let procInfoVar = schedulerState ^. processTable + initPd <- if toPid == 1 + then Just <$> lift (renderSchedulerState schedulerState) + else pure Nothing lift $ atomically $ do procInfoTable <- readTVar procInfoVar traverse (\toProcInfo -> do (pDetails, pState) <- readTVar (toProcInfo ^. processState) - return (toProcInfo ^. processTitle, pDetails, pState)) + let pDetails' = fromMaybe pDetails initPd + return (toProcInfo ^. processTitle, pDetails', pState)) (procInfoTable ^. at toPid) interpretUpdateDetails !td = do setMyProcessState ProcessBusyUpdatingDetails @@ -781,9 +788,9 @@ where exitReasonFromException exc = case Safe.fromException exc of Just Async.AsyncCancelled -> ExitProcessCancelled Nothing - Nothing -> ExitUnhandledError ( "runtime exception: " + Nothing -> ExitUnhandledError ( "runtime exception:\n" <> T.pack (prettyCallStack callStack) - <> " " + <> "\n" <> T.pack (Safe.displayException exc) ) logExitAndTriggerLinksAndMonitors reason pid = do diff --git a/src/Control/Eff/Concurrent/Protocol/Broker.hs b/src/Control/Eff/Concurrent/Protocol/Broker.hs --- a/src/Control/Eff/Concurrent/Protocol/Broker.hs +++ b/src/Control/Eff/Concurrent/Protocol/Broker.hs @@ -112,7 +112,7 @@ , TangibleBroker p , Stateful.Server (Broker p) (Processes e) ) - => Stateful.StartArgument (Broker p) (Processes e) + => Stateful.StartArgument (Broker p) -> Eff (Processes e) (Endpoint (Broker p)) startLink = Stateful.startLink @@ -124,13 +124,13 @@ statefulChild :: forall p e . ( HasCallStack - , IoLogging (Processes e) + , IoLogging e , TangibleBroker (Stateful.Stateful p) - , Stateful.Server (Broker (Stateful.Stateful p)) (Processes e) + , Stateful.Server (Broker (Stateful.Stateful p)) e ) => Timeout - -> (ChildId p -> Stateful.StartArgument p (Processes e)) - -> Stateful.StartArgument (Broker (Stateful.Stateful p)) (Processes e) + -> (ChildId p -> Stateful.StartArgument p) + -> Stateful.StartArgument (Broker (Stateful.Stateful p)) statefulChild t f = MkBrokerConfig t (Stateful.Init . f) -- | Stop the broker and shutdown all processes. @@ -452,10 +452,10 @@ -- * the 'Timeout' after requesting a normal child exit before brutally killing the child. -- -- @since 0.24.0 - data instance StartArgument (Broker p) (Processes q) = MkBrokerConfig + data instance StartArgument (Broker p) = MkBrokerConfig { brokerConfigChildStopTimeout :: Timeout - , brokerConfigStartFun :: ChildId p -> Effectful.Init p (Processes q) + , brokerConfigStartFun :: ChildId p -> Effectful.Init p } data instance Model (Broker p) = 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 @@ -44,7 +44,7 @@ :: forall (tag :: Type) eLoop q e. ( HasCallStack , TangibleCallbacks tag eLoop q - , E.Server (Server tag eLoop) (Processes q) + , E.Server (Server tag eLoop q) (Processes q) , FilteredLogging (Processes q) , HasProcesses e q ) @@ -60,7 +60,7 @@ :: forall (tag :: Type) eLoop q e. ( HasCallStack , TangibleCallbacks tag eLoop q - , E.Server (Server tag eLoop) (Processes q) + , E.Server (Server tag eLoop q) (Processes q) , FilteredLogging (Processes q) , HasProcesses e q ) @@ -71,7 +71,7 @@ -- | Phantom type to indicate a callback based 'E.Server' instance. -- -- @since 0.27.0 -data Server tag eLoop deriving Typeable +data Server tag eLoop e deriving Typeable -- | The constraints for a /tangible/ 'Server' instance. -- @@ -99,10 +99,10 @@ . showSTypeRep (typeOf px) ) -instance (TangibleCallbacks tag eLoop e) => E.Server (Server (tag :: Type) eLoop) (Processes e) where - type ServerPdu (Server tag eLoop) = tag - type ServerEffects (Server tag eLoop) (Processes e) = eLoop - data instance Init (Server tag eLoop) (Processes e) = +instance (TangibleCallbacks tag eLoop e) => E.Server (Server (tag :: Type) eLoop e) (Processes e) where + type ServerPdu (Server tag eLoop e) = tag + type ServerEffects (Server tag eLoop e) (Processes e) = eLoop + data instance Init (Server tag eLoop e) = MkServer { genServerId :: ServerId tag , genServerRunEffects :: forall x . (Endpoint tag -> Eff eLoop x -> Eff (Processes e) x) @@ -111,10 +111,10 @@ runEffects myEp svr = genServerRunEffects svr myEp onEvent myEp svr = genServerOnEvent svr myEp -instance (TangibleCallbacks tag eLoop e) => NFData (E.Init (Server (tag :: Type) eLoop) (Processes e)) where +instance (TangibleCallbacks tag eLoop e) => NFData (E.Init (Server (tag :: Type) eLoop e)) where rnf (MkServer x y z) = rnf x `seq` y `seq` z `seq` () -instance (TangibleCallbacks tag eLoop e) => Show (E.Init (Server (tag :: Type) eLoop) (Processes e)) where +instance (TangibleCallbacks tag eLoop e) => Show (E.Init (Server (tag :: Type) eLoop e)) where showsPrec d svr = showParen (d>=10) ( showsPrec 11 (genServerId svr) @@ -139,7 +139,7 @@ :: forall tag q. ( HasCallStack , TangibleCallbacks tag (Processes q) q - , E.Server (Server tag (Processes q)) (Processes q) + , E.Server (Server tag (Processes q) q) (Processes q) , FilteredLogging q ) => (Endpoint tag -> Event tag -> Eff (Processes q) ()) @@ -154,7 +154,7 @@ :: forall tag q . ( HasCallStack , TangibleCallbacks tag (Processes q) q - , E.Server (Server tag (Processes q)) (Processes q) + , E.Server (Server tag (Processes q) q) (Processes q) , FilteredLogging q ) => (Event tag -> Eff (Processes q) ()) @@ -169,7 +169,7 @@ -- See 'Callbacks'. -- -- @since 0.29.1 -type CallbacksEff tag eLoop e = E.Init (Server tag eLoop) (Processes e) +type CallbacksEff tag eLoop e = E.Init (Server tag eLoop e) -- | A smart constructor for 'CallbacksEff'. -- @@ -178,7 +178,7 @@ :: forall tag eLoop q. ( HasCallStack , TangibleCallbacks tag eLoop q - , E.Server (Server tag eLoop) (Processes q) + , E.Server (Server tag eLoop q) (Processes q) , FilteredLogging q ) => (forall x . Endpoint tag -> Eff eLoop x -> Eff (Processes q) x) @@ -194,7 +194,7 @@ :: ( HasCallStack , TangibleCallbacks tag eLoop q - , E.Server (Server tag eLoop) (Processes q) + , E.Server (Server tag eLoop q) (Processes q) , FilteredLogging q ) => (forall a. Eff eLoop a -> Eff (Processes q) a) 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 @@ -44,7 +44,7 @@ where -- | The value that defines what is required to initiate a 'Server' -- loop. - data Init a e + data Init a -- | The index type of the 'Event's that this server processes. -- This is the first parameter to the 'Request' and therefore of @@ -61,21 +61,21 @@ -- | Return the 'ProcessTitle'. -- -- Usually you should rely on the default implementation - serverTitle :: Init a e -> ProcessTitle + serverTitle :: Init a -> ProcessTitle - default serverTitle :: Typeable a => Init a e -> ProcessTitle + default serverTitle :: Typeable a => Init a -> ProcessTitle serverTitle _ = fromString $ showSTypeable @a "" -- | Process the effects of the implementation - runEffects :: Endpoint (ServerPdu a) -> Init a e -> Eff (ServerEffects a e) x -> Eff e x + runEffects :: Endpoint (ServerPdu a) -> Init a -> Eff (ServerEffects a e) x -> Eff e x - default runEffects :: ServerEffects a e ~ e => Endpoint (ServerPdu a) -> Init a e -> Eff (ServerEffects a e) x -> Eff e x + default runEffects :: ServerEffects a e ~ e => Endpoint (ServerPdu a) -> Init a -> Eff (ServerEffects a e) x -> Eff e x runEffects _ = const id -- | Update the 'Model' based on the 'Event'. - onEvent :: Endpoint (ServerPdu a) -> Init a e -> Event (ServerPdu a) -> Eff (ServerEffects a e) () + onEvent :: Endpoint (ServerPdu a) -> Init a -> Event (ServerPdu a) -> Eff (ServerEffects a e) () - default onEvent :: (Show (Init a e), Member Logs (ServerEffects a e)) => Endpoint (ServerPdu a) -> Init a e -> Event (ServerPdu a) -> Eff (ServerEffects a e) () + default onEvent :: (Show (Init a), Member Logs (ServerEffects a e)) => Endpoint (ServerPdu a) -> Init a -> Event (ServerPdu a) -> Eff (ServerEffects a e) () onEvent _ i e = logInfo ("unhandled: " <> T.pack (show i) <> " " <> T.pack (show e)) @@ -91,9 +91,9 @@ , HasProcesses (ServerEffects a (Processes q)) q , HasProcesses r q , HasCallStack) - => Init a (Processes q) + => Init a -> Eff r (Endpoint (ServerPdu a)) -start a = asEndpoint <$> spawn (serverTitle a) (protocolServerLoop a) +start a = asEndpoint <$> spawn (serverTitle @_ @(Processes q) a) (protocolServerLoop a) -- | Execute the server loop. -- @@ -107,9 +107,9 @@ , HasProcesses (ServerEffects a (Processes q)) q , HasProcesses r q , HasCallStack) - => Init a (Processes q) + => Init a -> Eff r (Endpoint (ServerPdu a)) -startLink a = asEndpoint <$> spawnLink (serverTitle a) (protocolServerLoop a) +startLink a = asEndpoint <$> spawnLink (serverTitle @_ @(Processes q) a) (protocolServerLoop a) -- | Execute the server loop. -- @@ -122,7 +122,7 @@ , Typeable a , Typeable (ServerPdu a) ) - => Init a (Processes q) -> Eff (Processes q) () + => Init a -> Eff (Processes q) () protocolServerLoop a = do myEp <- asEndpoint @(ServerPdu a) <$> self logDebug ("starting") @@ -138,13 +138,13 @@ onRequest :: Request (ServerPdu a) -> Event (ServerPdu a) onRequest (Call o m) = OnCall (replyTarget (MkSerializer toStrictDynamic) o) m onRequest (Cast m) = OnCast m - handleInt myEp i = onEvent myEp a (OnInterrupt i) *> pure Nothing + handleInt myEp i = onEvent @_ @(Processes q) myEp a (OnInterrupt i) *> pure Nothing mainLoop :: (Typeable a) => Endpoint (ServerPdu a) -> Either (Interrupt 'Recoverable) (Event (ServerPdu a)) -> Eff (ServerEffects a (Processes q)) (Maybe ()) mainLoop myEp (Left i) = handleInt myEp i - mainLoop myEp (Right i) = onEvent myEp a i *> pure Nothing + mainLoop myEp (Right i) = onEvent @_ @(Processes q) myEp a i *> pure Nothing -- | This event sum-type is used to communicate incoming messages and other events to the -- instances of 'Server'. 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 @@ -227,7 +227,7 @@ instance (Typeable event, Lifted IO q, Member Logs q) => Server (ObservationQueue event) (Processes q) where type instance Protocol (ObservationQueue event) = Observer event - data instance StartArgument (ObservationQueue event) (Processes q) = + data instance StartArgument (ObservationQueue event) = MkObservationQueue (ObservationQueue event) newtype instance Model (ObservationQueue event) = MkObservationQueueModel () deriving Default 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 @@ -67,7 +67,7 @@ class (Typeable (Protocol a)) => Server (a :: Type) q where -- | The value that defines what is required to initiate a 'Server' -- loop. - data StartArgument a q + data StartArgument a -- | The index type of the 'Event's that this server processes. -- This is the first parameter to the 'Request' and therefore of -- the 'Pdu' family. @@ -87,28 +87,28 @@ -- while it is running. -- -- @since 0.30.0 - title :: StartArgument a q -> ProcessTitle + title :: StartArgument a -> ProcessTitle - default title :: Typeable a => StartArgument a q -> ProcessTitle + default title :: Typeable a => StartArgument a -> ProcessTitle title _ = fromString $ showSTypeable @a "" -- | Return an initial 'Model' and 'Settings' setup :: Endpoint (Protocol a) - -> StartArgument a q + -> StartArgument a -> Eff q (Model a, Settings a) default setup :: (Default (Model a), Default (Settings a)) => Endpoint (Protocol a) - -> StartArgument a q + -> StartArgument a -> Eff q (Model a, Settings a) setup _ _ = pure (def, def) -- | Update the 'Model' based on the 'Event'. update :: Endpoint (Protocol a) - -> StartArgument a q + -> StartArgument a -> Effectful.Event (Protocol a) -> Eff (ModelState a ': SettingsReader a ': q) () @@ -121,7 +121,7 @@ data Stateful a deriving Typeable instance Server a q => Effectful.Server (Stateful a) q where - data Init (Stateful a) q = Init (StartArgument a q) + data Init (Stateful a) = Init (StartArgument a) type ServerPdu (Stateful a) = Protocol a type ServerEffects (Stateful a) q = ModelState a ': SettingsReader a ': q @@ -131,7 +131,7 @@ onEvent selfEndpoint (Init sa) = update selfEndpoint sa - serverTitle (Init startArg) = title startArg + serverTitle (Init startArg) = title @_ @q startArg -- | Execute the server loop. -- @@ -145,7 +145,7 @@ , Server a (Processes q) , HasProcesses r q ) - => StartArgument a (Processes q) -> Eff r (Endpoint (Protocol a)) + => StartArgument a -> Eff r (Endpoint (Protocol a)) startLink = Effectful.startLink . Init -- | Execute the server loop. Please use 'startLink' if you can. @@ -160,7 +160,7 @@ , FilteredLogging (Processes q) , HasProcesses r q ) - => StartArgument a (Processes q) -> Eff r (Endpoint (Protocol a)) + => StartArgument a -> Eff r (Endpoint (Protocol a)) start = Effectful.start . Init -- | The 'Eff'ect type of mutable 'Model' in a 'Server' instance. diff --git a/src/Control/Eff/Concurrent/Protocol/Watchdog.hs b/src/Control/Eff/Concurrent/Protocol/Watchdog.hs --- a/src/Control/Eff/Concurrent/Protocol/Watchdog.hs +++ b/src/Control/Eff/Concurrent/Protocol/Watchdog.hs @@ -214,7 +214,7 @@ , Member Logs e ) => Stateful.Server (Watchdog child) (Processes e) where - data instance StartArgument (Watchdog child) (Processes e) = + data instance StartArgument (Watchdog child) = StartWatchDog { _crashRate :: CrashRate } deriving Typeable @@ -327,7 +327,7 @@ -- ------------------ Start Argument -crashRate :: Lens' (Stateful.StartArgument (Watchdog child) (Processes e)) CrashRate +crashRate :: Lens' (Stateful.StartArgument (Watchdog child)) CrashRate crashRate = lens _crashRate (\m x -> m {_crashRate = x}) -- ----------------- Crash Rate diff --git a/src/Control/Eff/LogWriter/Async.hs b/src/Control/Eff/LogWriter/Async.hs --- a/src/Control/Eff/LogWriter/Async.hs +++ b/src/Control/Eff/LogWriter/Async.hs @@ -5,6 +5,7 @@ ) where +import Control.Concurrent (threadDelay) import Control.Concurrent.Async import Control.Concurrent.STM import Control.DeepSeq @@ -12,7 +13,8 @@ import Control.Eff.Log import Control.Eff.LogWriter.Rich import Control.Exception ( evaluate ) -import Control.Monad ( unless ) +import Control.Lens +import Control.Monad ( unless, when ) import Control.Monad.Trans.Control ( MonadBaseControl , liftBaseOp ) @@ -95,9 +97,9 @@ where logLoop tq = do ms <- atomically $ do - h <- readTBQueue tq - t <- flushTBQueue tq - return (h : t) + isEmpty <- isEmptyTBQueue tq + when isEmpty retry + flushTBQueue tq traverse_ ioWriter ms logLoop tq @@ -106,11 +108,17 @@ where logChannelPutIO (force -> me) = do !m <- evaluate me - atomically - (do - dropMessage <- isFullTBQueue logQ - unless dropMessage (writeTBQueue logQ m) + isFull <- atomically ( + if m^.lmSeverity <= warningSeverity then do + writeTBQueue logQ m + return False + else do + isFull <- isFullTBQueue logQ + unless isFull (writeTBQueue logQ m) + return isFull ) + when isFull $ + threadDelay 1_000 logQ = fromLogChannel lc data LogChannel = ConcurrentLogChannel diff --git a/src/Control/Eff/LogWriter/Rich.hs b/src/Control/Eff/LogWriter/Rich.hs --- a/src/Control/Eff/LogWriter/Rich.hs +++ b/src/Control/Eff/LogWriter/Rich.hs @@ -50,10 +50,12 @@ -> Facility -- ^ The default RFC-5424 facility to put into the 'lmFacility' field. -> LogWriter -- ^ The IO based writer to decorate -> LogWriter -richLogWriter appName facility = mappingLogWriterIO - ( setLogMessageTimestamp - >=> setLogMessageHostname - >=> pure - . set lmFacility facility - . set lmAppName (Just appName) - ) +richLogWriter appName facility = + mappingLogWriterIO + ( setLogMessageTimestamp + >=> setLogMessageHostname + ) + . mappingLogWriter + ( set lmFacility facility + . set lmAppName (Just appName) + ) diff --git a/test/BrokerTests.hs b/test/BrokerTests.hs --- a/test/BrokerTests.hs +++ b/test/BrokerTests.hs @@ -361,7 +361,7 @@ exitBecause (interruptToExit x) _ -> logDebug (pack (show tId) <> ": got some info: " <> pack (show evt)) - data instance StartArgument TestProtocol Effects = TestServerArgs TestProtocolServerMode Int + data instance StartArgument TestProtocol = TestServerArgs TestProtocolServerMode Int type instance ChildId TestProtocol = Int diff --git a/test/GenServerTests.hs b/test/GenServerTests.hs --- a/test/GenServerTests.hs +++ b/test/GenServerTests.hs @@ -33,7 +33,7 @@ -- ---------------------------------------------------------------------------- instance IoLogging e => S.Server Small (Processes e) where - data StartArgument Small (Processes e) = MkSmall + data StartArgument Small = MkSmall newtype instance Model Small = SmallModel String deriving Default update _me MkSmall x = case x of @@ -77,7 +77,7 @@ -- ---------------------------------------------------------------------------- instance IoLogging e => S.Server Big (Processes e) where - data instance StartArgument Big (Processes e) = MkBig + data instance StartArgument Big = MkBig newtype Model Big = BigModel String deriving Default update me MkBig = \case E.OnCall rt req -> diff --git a/test/ObserverTests.hs b/test/ObserverTests.hs --- a/test/ObserverTests.hs +++ b/test/ObserverTests.hs @@ -298,7 +298,7 @@ show (TestObsReg x) = "TestObsReg " ++ show x instance (IoLogging r, HasProcesses r q) => S.Server TestObservable r where - data Init TestObservable r = TestObservableServerInit + data Init TestObservable = TestObservableServerInit type ServerEffects TestObservable r = ObserverRegistryState String ': r runEffects _ _ = evalObserverRegistryState onEvent _ _ = @@ -342,7 +342,7 @@ instance (IoLogging r, HasProcesses r q) => M.Server TestObserver r where - data StartArgument TestObserver r = MkTestObserver + data StartArgument TestObserver = MkTestObserver newtype instance Model TestObserver = TestObserverModel {fromTestObserverModel :: [String]} deriving Default update _ MkTestObserver e = case e of diff --git a/test/ProcessBehaviourTestCases.hs b/test/ProcessBehaviourTestCases.hs --- a/test/ProcessBehaviourTestCases.hs +++ b/test/ProcessBehaviourTestCases.hs @@ -1238,12 +1238,22 @@ (do let expected1 = "test details 1" expected2 = "test details 2" - updateProcessDetails expected1 - (_, actual1, _) <- fromJust <$> (self >>= getProcessState) + parent <- self + testPid <- spawnLink "test" $ do + updateProcessDetails expected1 + sendMessage parent () + () <- receiveMessage + updateProcessDetails expected2 + sendMessage parent () + receiveMessage + () <- receiveMessage + (_, actual1, _) <- fromJust <$> (getProcessState testPid) lift (assertEqual "1" expected1 actual1) - updateProcessDetails expected2 - (_, actual2, _) <- fromJust <$> (self >>= getProcessState) + sendMessage testPid () + () <- receiveMessage + (_, actual2, _) <- fromJust <$> (getProcessState testPid) lift (assertEqual "2" expected2 actual2) + sendMessage testPid () ) ) ] diff --git a/test/WatchdogTests.hs b/test/WatchdogTests.hs --- a/test/WatchdogTests.hs +++ b/test/WatchdogTests.hs @@ -32,7 +32,7 @@ wd <- Watchdog.startLink def unlinkProcess (wd ^. fromEndpoint) do - broker <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id) + broker <- Broker.startLink (Broker.statefulChild @BookShelf @Effects (TimeoutMicros 1_000_000) id) unlinkProcess (broker ^. fromEndpoint) logNotice "started broker" Watchdog.attachTemporary wd broker @@ -41,7 +41,7 @@ logNotice "stopped broker" do - broker <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id) + broker <- Broker.startLink (Broker.statefulChild @BookShelf @Effects (TimeoutMicros 1_000_000) id) unlinkProcess (broker ^. fromEndpoint) logNotice "started new broker" Watchdog.attachTemporary wd broker @@ -61,7 +61,7 @@ unlinkProcess (wd ^. fromEndpoint) logNotice "started watchdog" do - broker <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id) + broker <- Broker.startLink (Broker.statefulChild @BookShelf @Effects (TimeoutMicros 1_000_000) id) unlinkProcess (broker ^. fromEndpoint) logNotice "started broker" Watchdog.attachPermanent wd broker @@ -82,7 +82,7 @@ unlinkProcess (wd ^. fromEndpoint) logNotice "started watchdog" do - broker <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id) + broker <- Broker.startLink (Broker.statefulChild @BookShelf @Effects (TimeoutMicros 1_000_000) id) unlinkProcess (broker ^. fromEndpoint) logNotice "started broker" Watchdog.attachTemporary wd broker @@ -99,7 +99,7 @@ unlinkProcess (wd ^. fromEndpoint) logNotice "started watchdog" do - broker <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id) + broker <- Broker.startLink (Broker.statefulChild @BookShelf @Effects (TimeoutMicros 1_000_000) id) unlinkProcess (broker ^. fromEndpoint) logNotice "started broker" Watchdog.attachPermanent wd broker @@ -114,7 +114,7 @@ assertShutdown (broker ^. fromEndpoint) expected logNotice "crashed broker" do - broker <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id) + broker <- Broker.startLink (Broker.statefulChild @BookShelf @Effects (TimeoutMicros 1_000_000) id) unlinkProcess (broker ^. fromEndpoint) logNotice "started new broker" Watchdog.attachTemporary wd broker @@ -132,7 +132,7 @@ mwd <- monitor (wd ^. fromEndpoint) logNotice "started watchdog" do - broker <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id) + broker <- Broker.startLink (Broker.statefulChild @BookShelf @Effects (TimeoutMicros 1_000_000) id) unlinkProcess (broker ^. fromEndpoint) logNotice "started broker" Watchdog.attachTemporary wd broker @@ -161,18 +161,18 @@ wdMon <- monitor (wd ^. fromEndpoint) logNotice "started watchdog" do - broker1 <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id) + broker1 <- Broker.startLink (Broker.statefulChild @BookShelf @Effects (TimeoutMicros 1_000_000) id) unlinkProcess (broker1^.fromEndpoint) logNotice "started broker 1" Watchdog.attachPermanent wd broker1 logNotice "attached + linked broker 1" - broker2 <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id) + broker2 <- Broker.startLink (Broker.statefulChild @BookShelf @Effects (TimeoutMicros 1_000_000) id) logNotice "started broker 2" Watchdog.attachTemporary wd broker2 logNotice "attached broker 2" - broker3 <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id) + broker3 <- Broker.startLink (Broker.statefulChild @BookShelf @Effects (TimeoutMicros 1_000_000) id) logNotice "started broker 3" Watchdog.attachTemporary wd broker3 logNotice "attached broker 3" @@ -199,7 +199,7 @@ ] , testGroup "restarting children" [ runTestCase "test 7: when a child exits it is restarted" $ do - broker <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id) + broker <- Broker.startLink (Broker.statefulChild @BookShelf @Effects (TimeoutMicros 1_000_000) id) logNotice "started broker" wd <- Watchdog.startLink def logNotice "started watchdog" @@ -211,7 +211,7 @@ , runTestCase "test 8: when the broker emits the shutting down\ \ event the watchdog does not restart children" $ do - broker <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id) + broker <- Broker.startLink (Broker.statefulChild @BookShelf @Effects (TimeoutMicros 1_000_000) id) unlinkProcess (broker ^. fromEndpoint) logNotice "started broker" wd <- Watchdog.startLink def @@ -235,7 +235,7 @@ logNotice "watchdog stopped" , runTestCase "test 9: a child is only restarted if it crashes" $ do - broker <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id) + broker <- Broker.startLink (Broker.statefulChild @BookShelf @Effects (TimeoutMicros 1_000_000) id) logNotice "started broker" wd <- Watchdog.startLink def logNotice "started watchdog" @@ -262,7 +262,7 @@ in [ runTestCase "test 10: if a child crashes 3 times in 300ms, waits 1.1 seconds\ \ and crashes again 3 times, and is restarted three times" $ do - broker <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id) + broker <- Broker.startLink (Broker.statefulChild @BookShelf @Effects (TimeoutMicros 1_000_000) id) logNotice "started broker" wd <- Watchdog.startLink threeTimesASecond logNotice "started watchdog" @@ -289,7 +289,7 @@ , runTestCase "test 11: if a child crashes 4 times within 1s it is not restarted\ \ and the watchdog exits with an error" $ do - broker <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id) + broker <- Broker.startLink (Broker.statefulChild @BookShelf @Effects (TimeoutMicros 1_000_000) id) logNotice "started broker" wd <- Watchdog.startLink threeTimesASecond logNotice "started watchdog" @@ -322,7 +322,7 @@ , runTestCase "test 12: if a child of a linked broker crashes too often,\ \ the watchdog exits with an error and interrupts the broker" $ do - broker <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id) + broker <- Broker.startLink (Broker.statefulChild @BookShelf @Effects (TimeoutMicros 1_000_000) id) logNotice "started broker" unlinkProcess (broker ^. fromEndpoint) mBroker <- monitor (broker ^. fromEndpoint) @@ -371,11 +371,11 @@ unlinkProcess (wd ^. fromEndpoint) logNotice "started watchdog" - brokerT <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id) + brokerT <- Broker.startLink (Broker.statefulChild @BookShelf @Effects (TimeoutMicros 1_000_000) id) unlinkProcess (brokerT^.fromEndpoint) logNotice "started temporary broker" - brokerP <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id) + brokerP <- Broker.startLink (Broker.statefulChild @BookShelf @Effects (TimeoutMicros 1_000_000) id) unlinkProcess (brokerP^.fromEndpoint) logNotice "started permanent broker" @@ -427,7 +427,7 @@ wd <- Watchdog.startLink (1 `Watchdog.crashesPerSeconds` 1) logNotice "started watchdog" - brokerT <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id) + brokerT <- Broker.startLink (Broker.statefulChild @BookShelf @Effects (TimeoutMicros 1_000_000) id) logNotice "started temporary broker" OQ.observe @(Broker.ChildEvent (Stateful BookShelf)) (100 :: Int) brokerT $ do @@ -452,7 +452,7 @@ wd <- Watchdog.startLink (1 `Watchdog.crashesPerSeconds` 10) logNotice "started watchdog" - brokerT <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id) + brokerT <- Broker.startLink (Broker.statefulChild @BookShelf @Effects (TimeoutMicros 1_000_000) id) logNotice "started temporary broker" OQ.observe @(Broker.ChildEvent (Stateful BookShelf)) (100 :: Int) brokerT $ do @@ -492,11 +492,11 @@ setup k = do wd <- Watchdog.startLink (4 `Watchdog.crashesPerSeconds` 30) logNotice "started watchdog" - brokerT <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id) + brokerT <- Broker.startLink (Broker.statefulChild @BookShelf @Effects (TimeoutMicros 1_000_000) id) logNotice "started temporary broker" Watchdog.attachTemporary wd brokerT logNotice "attached temporary broker" - brokerP <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id) + brokerP <- Broker.startLink (Broker.statefulChild @BookShelf @Effects (TimeoutMicros 1_000_000) id) logNotice "started permanent broker" Watchdog.attachTemporary wd brokerP logNotice "attached permanent broker" @@ -586,7 +586,7 @@ bookshelfDemo :: HasCallStack => Eff Effects () bookshelfDemo = do logNotice "Bookshelf Demo Begin" - broker <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id) + broker <- Broker.startLink (Broker.statefulChild @BookShelf @Effects (TimeoutMicros 1_000_000) id) shelf1 <- Broker.spawnOrLookup broker (BookShelfId 1) call shelf1 (AddBook "Solaris") call shelf1 GetBookList >>= logDebug . pack . show @@ -687,7 +687,7 @@ deriving Typeable instance Stateful.Server BookShelf Effects where - newtype StartArgument BookShelf Effects = BookShelfId Int + newtype StartArgument BookShelf = BookShelfId Int deriving (Show, Eq, Ord, Num, NFData, Typeable) newtype instance Model BookShelf = BookShelfModel (Set String) deriving (Eq, Show, Default) @@ -722,7 +722,7 @@ bookShelfModel :: Iso' (Stateful.Model BookShelf) (Set String) bookShelfModel = iso (\(BookShelfModel s) -> s) BookShelfModel -type instance Broker.ChildId BookShelf = Stateful.StartArgument BookShelf Effects +type instance Broker.ChildId BookShelf = Stateful.StartArgument BookShelf instance NFData (Pdu BookShelf r) where rnf GetBookList = ()