polysemy-conc 0.4.0.1 → 0.5.0.0

raw patch · 18 files changed

+719/−34 lines, 18 filesdep +data-defaultdep +polysemy-plugindep +torsordep ~polysemydep ~polysemy-timedep ~reludePVP ok

version bump matches the API change (PVP)

Dependencies added: data-default, polysemy-plugin, torsor

Dependency ranges changed: polysemy, polysemy-time, relude, unagi-chan

API changes (from Hackage documentation)

Files

• changelog.md +7/−0
• lib/Polysemy/Conc.hs +82/−2
• lib/Polysemy/Conc/Effect/Events.hs +10/−10
• lib/Polysemy/Conc/Effect/Mask.hs +53/−0
• lib/Polysemy/Conc/Effect/Monitor.hs +55/−0
• lib/Polysemy/Conc/Interpreter/Critical.hs +1/−1
• lib/Polysemy/Conc/Interpreter/Events.hs +6/−6
• lib/Polysemy/Conc/Interpreter/Interrupt.hs +2/−2
• lib/Polysemy/Conc/Interpreter/Mask.hs +63/−0
• lib/Polysemy/Conc/Interpreter/Monitor.hs +76/−0
• lib/Polysemy/Conc/Interpreter/Scoped.hs +50/−3
• lib/Polysemy/Conc/Interpreter/Stack.hs +37/−0
• lib/Polysemy/Conc/Monitor.hs +51/−0
• lib/Polysemy/Conc/Prelude.hs +23/−0
• polysemy-conc.cabal +26/−10
• test/Main.hs +9/−0
• test/Polysemy/Conc/Test/MaskTest.hs +71/−0
• test/Polysemy/Conc/Test/MonitorTest.hs +97/−0

@@ -1,5 +1,12 @@ # Unreleased +# 0.5.0.0++* Add `mask` effects.+* Add `Monitor`, an effect that repeatedly checks a condition and restarts a region when it is met.+* Add interpreter combinators for `Scoped`.+* Add a runner for the default `Conc` stack.+ # 0.4.0.0  * Add `lock`, a combinator for protecting a region with a mutex.

@@ -66,7 +66,7 @@   subscribe,   subscribeWhile,   subscribeLoop,-  EventToken,+  EventResource,   EventChan,   ChanEvents,   EventConsumer,@@ -82,7 +82,48 @@   interpretCritical,   interpretCriticalNull, +  -- * Masking+  Mask,+  UninterruptipleMask,+  mask,+  uninterruptibleMask,+  restore,++  -- * Interpreters+  interpretMaskFinal,+  interpretUninterruptibleMaskFinal,++  -- * Scoped Effects+  Scoped,+  scoped,++  -- ** Interpreters+  runScoped,+  runScopedAs,+  interpretScoped,+  interpretScopedH,+  interpretScopedAs,++  -- * Monitoring+  Monitor,+  monitor,+  withMonitor,+  restart,+  Restart,+  RestartingMonitor,+  MonitorResource (MonitorResource),+  ScopedMonitor,++  -- ** Interpreters+  interpretMonitorRestart,+  interpretMonitorPure,+  monitorClockSkew,+  ClockSkewConfig (ClockSkewConfig),+  clockSkewConfig,+   -- * Other Combinators+  ConcStack,+  runConc,   interpretAtomic,   withAsyncBlock,   withAsync,@@ -101,15 +142,29 @@ import Polysemy.Conc.AtomicState (interpretAtomic) import Polysemy.Conc.Data.QueueResult (QueueResult) import Polysemy.Conc.Effect.Critical (Critical)-import Polysemy.Conc.Effect.Events (EventToken, Events, consume, publish, subscribe)+import Polysemy.Conc.Effect.Events (EventResource, Events, consume, publish, subscribe) import Polysemy.Conc.Effect.Interrupt (Interrupt)+import Polysemy.Conc.Effect.Mask (Mask, UninterruptipleMask, mask, restore, uninterruptibleMask)+import Polysemy.Conc.Effect.Monitor (+  Monitor,+  MonitorResource (MonitorResource),+  Restart,+  RestartingMonitor,+  ScopedMonitor,+  monitor,+  restart,+  withMonitor,+  ) import Polysemy.Conc.Effect.Queue (Queue) import Polysemy.Conc.Effect.Race (Race, race, timeout)+import Polysemy.Conc.Effect.Scoped (Scoped, scoped) import Polysemy.Conc.Effect.Sync (ScopedSync, Sync) import Polysemy.Conc.Events (subscribeLoop, subscribeWhile) import Polysemy.Conc.Interpreter.Critical (interpretCritical, interpretCriticalNull) import Polysemy.Conc.Interpreter.Events (ChanConsumer, ChanEvents, EventChan, EventConsumer, interpretEventsChan) import Polysemy.Conc.Interpreter.Interrupt (interpretInterrupt, interpretInterruptOnce)+import Polysemy.Conc.Interpreter.Mask (interpretMaskFinal, interpretUninterruptibleMaskFinal)+import Polysemy.Conc.Interpreter.Monitor (interpretMonitorPure, interpretMonitorRestart) import Polysemy.Conc.Interpreter.Queue.Pure (   interpretQueueListReadOnlyAtomic,   interpretQueueListReadOnlyAtomicWith,@@ -119,7 +174,10 @@ import Polysemy.Conc.Interpreter.Queue.TB (interpretQueueTB) import Polysemy.Conc.Interpreter.Queue.TBM (interpretQueueTBM) import Polysemy.Conc.Interpreter.Race (interpretRace)+import Polysemy.Conc.Interpreter.Scoped (interpretScoped, interpretScopedAs, interpretScopedH, runScoped, runScopedAs)+import Polysemy.Conc.Interpreter.Stack (ConcStack, runConc) import Polysemy.Conc.Interpreter.Sync (interpretScopedSync, interpretScopedSyncAs, interpretSync, interpretSyncAs)+import Polysemy.Conc.Monitor (ClockSkewConfig (ClockSkewConfig), clockSkewConfig, monitorClockSkew) import Polysemy.Conc.Queue (loop, loopOr) import Polysemy.Conc.Queue.Result (resultToMaybe) import Polysemy.Conc.Race (race_, timeoutAs, timeoutAs_, timeoutMaybe, timeoutU, timeout_)@@ -133,6 +191,7 @@ -- - [MVars](#mvar) -- - [Racing](#race) -- - [Signal handling](#signal)+-- - [Masking](#mask)  -- $queue -- #queue#@@ -156,3 +215,24 @@  -- $signal -- #signal#++-- $mask+-- #mask#+-- The two effects 'Mask' and 'UninterruptipleMask' use the 'Polysemy.Conc.Effect.Scoped' pattern, which allow an effect+-- with resources to be constrained to a region of code.+-- The actual effect is 'Polysemy.Conc.Effect.Mask.RestoreMask', with `mask` and 'uninterruptibleMask' only specializing+-- 'Polysemy.Conc.Effect.scoped' to the appropriate resource type.+--+-- Usage is straightforward:+--+-- @+-- prog :: Member (Mask resource) r+-- prog =+--  mask do+--    doMaskedThing+--    restore do+--      doUnmaskedThing+--    doMaskedThing+-- @+--+-- The @resource@ parameter stays polymorphic; it is used to connect the resource in the interpreter to the callsite.

@@ -5,21 +5,21 @@ import Polysemy (makeSem_) import Polysemy.Conc.Effect.Scoped (Scoped, scoped) --- |Marker for the 'Scoped' token for 'Events'.-newtype EventToken token =-  EventToken { unEventToken :: token }+-- |Marker for the 'Scoped' resource for 'Events'.+newtype EventResource resource =+  EventResource { unEventToken :: resource }   deriving (Eq, Show, Generic)  -- |An event publisher that can be consumed from multiple threads.-data Events (token :: Type) (e :: Type) :: Effect where-  Publish :: e -> Events token e m ()+data Events (resource :: Type) (e :: Type) :: Effect where+  Publish :: e -> Events resource e m ()  makeSem_ ''Events  -- |Publish one event. publish ::-  ∀ e token r .-  Member (Events token e) r =>+  ∀ e resource r .+  Member (Events resource e) r =>   e ->   Sem r () @@ -38,8 +38,8 @@ -- |Create a new scope for 'Events', causing the nested program to get its own copy of the event stream. -- To be used with 'Polysemy.Conc.interpretEventsChan'. subscribe ::-  ∀ e token r .-  Member (Scoped (EventToken token) (Consume e)) r =>+  ∀ e resource r .+  Member (Scoped (EventResource resource) (Consume e)) r =>   InterpreterFor (Consume e) r subscribe =-  scoped @(EventToken token)+  scoped @(EventResource resource)

@@ -0,0 +1,53 @@+{-# options_haddock prune #-}+-- |Description: Mask Effect, Internal+module Polysemy.Conc.Effect.Mask where++import Polysemy (makeSem_)++import Polysemy.Conc.Effect.Scoped (Scoped, scoped)++-- |Part of an effect abstracting 'Control.Exception.mask'.+data RestoreMask :: Effect where+  Restore :: m a -> RestoreMask m a++makeSem_ ''RestoreMask++-- |Restore the previous masking state.+-- Can only be called inside of an action passed to 'mask' or 'uninterruptibleMask'.+restore ::+  ∀ r a .+  Member RestoreMask r =>+  Sem r a ->+  Sem r a++newtype MaskResource resource =+  MaskResource { unMaskResource :: resource }++newtype UninterruptipleMaskResource resource =+  UninterruptipleMaskResource { unUninterruptipleMaskResource :: resource }++-- |The scoped masking effect.+type Mask resource =+  Scoped (MaskResource resource) RestoreMask++-- |The scoped uninterruptible masking effect.+type UninterruptipleMask resource =+  Scoped (UninterruptipleMaskResource resource) RestoreMask++-- |Mark a region as masked.+-- Uses the 'Scoped' pattern.+mask ::+  ∀ resource r .+  Member (Mask resource) r =>+  InterpreterFor RestoreMask r+mask =+  scoped @(MaskResource resource)++-- |Mark a region as uninterruptibly masked.+-- Uses the 'Scoped' pattern.+uninterruptibleMask ::+  ∀ resource r .+  Member (UninterruptipleMask resource) r =>+  InterpreterFor RestoreMask r+uninterruptibleMask =+  scoped @(UninterruptipleMaskResource resource)

@@ -0,0 +1,55 @@+{-# options_haddock prune #-}+-- |Description: Monitor Effect, Internal+module Polysemy.Conc.Effect.Monitor where++import Polysemy.Time (NanoSeconds)++import Polysemy.Conc.Effect.Scoped (Scoped, scoped)++data Restart =+  Restart+  deriving (Eq, Show)++-- |Mark a region as being subject to intervention by a monitoring program.+-- This can mean that a thread is repeatedly checking a condition and cancelling this region when it is unmet.+-- A use case could be checking whether a remote service is available, or whether the system was suspended and resumed.+-- This should be used in a 'Scoped' context, like 'withMonitor'.+data Monitor (action :: Type) :: Effect where+  Monitor :: m a -> Monitor action m a++makeSem ''Monitor++-- |Marker type for a 'Scoped' 'Monitor'.+newtype MonitorResource a =+  MonitorResource { unMonitorResource :: a }++-- |Convenience alias for a 'Scoped' 'Monitor'.+type ScopedMonitor (resource :: Type) (action :: Type) =+  Scoped (MonitorResource resource) (Monitor action)++-- |'Monitor' specialized to the 'Restart' action.+type RestartingMonitor (resource :: Type) =+  ScopedMonitor resource Restart++-- |Resources for a 'Scoped' 'Monitor'.+data MonitorCheck r =+  MonitorCheck {+    interval :: NanoSeconds,+    check :: MVar () -> Sem r ()+  }++-- |Start a region that can contain monitor-intervention regions.+withMonitor ::+  ∀ resource action r .+  Member (ScopedMonitor resource action) r =>+  InterpreterFor (Monitor action) r+withMonitor =+  scoped @(MonitorResource resource)++-- |Variant of 'withMonitor' that uses the 'Restart' strategy.+restart ::+  ∀ resource r .+  Member (ScopedMonitor resource Restart) r =>+  InterpreterFor (Monitor Restart) r+restart =+  withMonitor @resource

@@ -2,7 +2,7 @@ module Polysemy.Conc.Interpreter.Critical where  import qualified Control.Exception as Exception-import Polysemy (interpretH, runT)+import Polysemy (runT) import Polysemy.Final (getInitialStateS, interpretFinal, runS)  import Polysemy.Conc.Effect.Critical (Critical (..))

@@ -9,7 +9,7 @@  import Polysemy.Conc.Async (withAsync_) import qualified Polysemy.Conc.Effect.Events as Events-import Polysemy.Conc.Effect.Events (Consume, EventToken (EventToken), Events)+import Polysemy.Conc.Effect.Events (Consume, EventResource (EventResource), Events) import Polysemy.Conc.Effect.Race (Race) import Polysemy.Conc.Effect.Scoped (Scoped) import Polysemy.Conc.Interpreter.Scoped (runScopedAs)@@ -18,13 +18,13 @@ type ChanEvents e =   Events (OutChan e) e --- |Convenience alias for the default 'EventToken' that uses an 'OutChan'.+-- |Convenience alias for the default 'EventResource' that uses an 'OutChan'. type EventChan e =-  EventToken (OutChan e)+  EventResource (OutChan e)  -- |Convenience alias for the consumer effect. type EventConsumer token e =-  Scoped (EventToken token) (Consume e)+  Scoped (EventResource token) (Consume e)  -- |Convenience alias for the consumer effect using the default implementation. type ChanConsumer e =@@ -37,7 +37,7 @@   Member (Embed IO) r =>   EventChan e ->   InterpreterFor (Consume e) r-interpretConsumeChan (EventToken chan) =+interpretConsumeChan (EventResource chan) =   interpret \case     Events.Consume ->       embed (readChan chan)@@ -77,4 +77,4 @@ interpretEventsChan sem = do   (inChan, outChan) <- embed (newChan @e 64)   withAsync_ (forever (embed (readChan outChan))) do-    runScopedAs (EventToken <$> embed (dupChan inChan)) interpretConsumeChan (interpretEventsInChan inChan sem)+    runScopedAs (EventResource <$> embed (dupChan inChan)) interpretConsumeChan (interpretEventsInChan inChan sem)

@@ -7,12 +7,12 @@ import qualified Data.Map.Strict as Map import qualified Data.Set as Set import qualified Data.Text.IO as Text-import Polysemy (getInspectorT, inspect, interpretH, runT)+import Polysemy (getInspectorT, inspect, runT) import Polysemy.Async (Async, async, await, cancel) import Polysemy.AtomicState (runAtomicStateTVar) import Polysemy.Internal.Tactics (liftT) import Polysemy.Time (Seconds (Seconds))-import System.Posix.Signals (Handler (CatchInfoOnce, CatchOnce, CatchInfo), SignalInfo, installHandler, keyboardSignal)+import System.Posix.Signals (Handler (CatchInfo, CatchInfoOnce, CatchOnce), SignalInfo, installHandler, keyboardSignal)  import qualified Polysemy.Conc.Effect.Critical as Critical import Polysemy.Conc.Effect.Critical (Critical)

@@ -0,0 +1,63 @@+{-# options_haddock prune #-}+-- |Description: Mask Interpreters, Internal+module Polysemy.Conc.Interpreter.Mask where++import qualified Control.Exception as Base+import Polysemy (runTSimple)+import Polysemy.Final (runS, withStrategicToFinal, withWeavingToFinal)++import Polysemy.Conc.Effect.Mask (+  Mask,+  MaskResource (MaskResource),+  RestoreMask (Restore),+  UninterruptipleMask,+  UninterruptipleMaskResource (UninterruptipleMaskResource),+  )+import Polysemy.Conc.Interpreter.Scoped (runScoped)++newtype Restoration =+  Restoration { unRestoration :: ∀ a . IO a -> IO a }++mask ::+  Member (Final IO) r =>+  (MaskResource Restoration -> Sem r a) ->+  Sem r a+mask f =+  withWeavingToFinal @IO \ s lower _ ->+    Base.mask \ restore -> (lower (f (MaskResource (Restoration restore)) <$ s))++uninterruptibleMask ::+  Member (Final IO) r =>+  (UninterruptipleMaskResource Restoration -> Sem r a) ->+  Sem r a+uninterruptibleMask f =+  withWeavingToFinal @IO \ s lower _ ->+    Base.uninterruptibleMask \ restore -> (lower (f (UninterruptipleMaskResource (Restoration restore)) <$ s))++interpretRestoreMask ::+  ∀ r .+  Member (Final IO) r =>+  Restoration ->+  InterpreterFor RestoreMask r+interpretRestoreMask (Restoration restore) =+  interpretH \case+    Restore ma -> do+      let+        restoreSem m =+          withStrategicToFinal do+            (restore <$> runS m)+      restoreSem (runTSimple ma)++-- |Interpret 'Mask' in 'IO'.+interpretMaskFinal ::+  Member (Final IO) r =>+  InterpreterFor (Mask Restoration) r+interpretMaskFinal =+  runScoped mask \ (MaskResource r) -> interpretRestoreMask r++-- |Interpret 'UninterruptipleMask' in 'IO'.+interpretUninterruptibleMaskFinal ::+  Member (Final IO) r =>+  InterpreterFor (UninterruptipleMask Restoration) r+interpretUninterruptibleMaskFinal =+  runScoped uninterruptibleMask \ (UninterruptipleMaskResource r) -> interpretRestoreMask r

@@ -0,0 +1,76 @@+{-# options_haddock prune #-}+-- |Description: Monitor Interpreters, Internal+module Polysemy.Conc.Interpreter.Monitor where++import qualified Control.Exception as Base+import Polysemy (embedFinal, runTSimple)+import Polysemy.Async (Async)+import Polysemy.Error (errorToIOFinal, fromExceptionSem)+import Polysemy.Resource (Resource)+import qualified Polysemy.Time as Time+import Polysemy.Time (Time)++import Polysemy.Conc.Async (withAsync_)+import Polysemy.Conc.Effect.Monitor (+  Monitor (Monitor),+  MonitorCheck (MonitorCheck),+  MonitorResource (MonitorResource),+  RestartingMonitor,+  ScopedMonitor,+  )+import qualified Polysemy.Conc.Effect.Race as Race+import Polysemy.Conc.Effect.Race (Race)+import Polysemy.Conc.Interpreter.Scoped (runScoped, runScopedAs)++newtype CancelResource =+  CancelResource { signal :: MVar () }++data MonitorCancel =+  MonitorCancel+  deriving (Eq, Show, Exception)++interpretMonitorCancel ::+  Members [Race, Async, Final IO] r =>+  MonitorResource CancelResource ->+  InterpreterFor (Monitor action) r+interpretMonitorCancel (MonitorResource CancelResource {..}) =+  interpretH \case+    Monitor ma ->+      leftM (Base.throw MonitorCancel) =<< Race.race (embedFinal @IO (readMVar signal)) (runTSimple ma)++monitorRestart ::+  ∀ t d r a .+  Members [Time t d, Resource, Async, Race, Final IO] r =>+  MonitorCheck r ->+  (MonitorResource CancelResource -> Sem r a) ->+  Sem r a+monitorRestart (MonitorCheck interval check) run = do+  sig <- embedFinal @IO newEmptyMVar+  withAsync_ (Time.loop_ @t @d interval (check sig)) (spin sig)+  where+    spin sig = do+      let res = (MonitorResource (CancelResource sig))+      void (embedFinal @IO (tryTakeMVar sig))+      leftM (spin sig) =<< errorToIOFinal @MonitorCancel (fromExceptionSem @MonitorCancel (raise (run res)))++-- |Interpret @'Scoped' 'Monitor'@ with the 'Restart' strategy.+-- This takes a check action that may put an 'MVar' when the scoped region should be restarted.+-- The check is executed in a loop, with an interval given in 'MonitorCheck'.+interpretMonitorRestart ::+  ∀ t d r .+  Members [Time t d, Resource, Async, Race, Final IO] r =>+  MonitorCheck r ->+  InterpreterFor (RestartingMonitor CancelResource) r+interpretMonitorRestart check =+  runScoped (monitorRestart @t @d check) interpretMonitorCancel++interpretMonitorPure' :: MonitorResource () -> InterpreterFor (Monitor action) r+interpretMonitorPure' _ =+  interpretH \case+    Monitor ma ->+      runTSimple ma++-- |Run 'Monitor' as a no-op.+interpretMonitorPure :: InterpreterFor (ScopedMonitor () action) r+interpretMonitorPure =+  runScopedAs (pure (MonitorResource ())) interpretMonitorPure'

@@ -6,13 +6,15 @@ import Polysemy.Internal.Union (Weaving (Weaving), decomp, hoist, injWeaving)  import Polysemy.Conc.Effect.Scoped (Scoped (InScope, Run))+import Polysemy (Tactical)+import Polysemy.Internal.Tactics (runTactics)  interpretH' ::   ∀ e r .   (∀ x . Weaving e (Sem (e : r)) x -> Sem r x) ->   InterpreterFor e r-interpretH' h sem =-  Sem \ k -> runSem sem $ decomp >>> \case+interpretH' h (Sem m) =+  Sem \ k -> m $ decomp >>> \case     Right wav -> runSem (h wav) k     Left g -> k $ hoist (interpretH' h) g @@ -41,7 +43,7 @@         InScope main ->           ex <$> withResource \ resource -> run (wv (main resource <$ s)) --- |Variant of 'Scoped' in which the resource allocator is a plain action.+-- |Variant of 'runScoped' in which the resource allocator is a plain action. runScopedAs ::   ∀ resource effect r .   Sem r resource ->@@ -49,3 +51,48 @@   InterpreterFor (Scoped resource effect) r runScopedAs resource =   runScoped \ f -> f =<< resource++-- |Variant of 'runScoped' that takes a higher-order handler instead of an interpreter.+interpretScopedH ::+  ∀ resource effect r .+  (∀ x . (resource -> Sem r x) -> Sem r x) ->+  (∀ r0 x . resource -> effect (Sem r0) x -> Tactical effect (Sem r0) r x) ->+  InterpreterFor (Scoped resource effect) r+interpretScopedH withResource scopedHandler =+  run+  where+    run :: InterpreterFor (Scoped resource effect) r+    run =+      interpretH' \ (Weaving effect s wv ex ins) -> case effect of+        Run resource act ->+          ex <$> runTactics s (raise . run . wv) ins (run . wv) (scopedHandler resource act)+        InScope main ->+          ex <$> withResource \ resource -> run (wv (main resource <$ s))++-- |Variant of 'runScoped' that takes a handler instead of an interpreter.+interpretScoped ::+  ∀ resource effect r .+  (∀ x . (resource -> Sem r x) -> Sem r x) ->+  (∀ r0 x . resource -> effect (Sem r0) x -> Sem r x) ->+  InterpreterFor (Scoped resource effect) r+interpretScoped withResource scopedHandler =+  run+  where+    run :: InterpreterFor (Scoped resource effect) r+    run =+      interpretH' \ (Weaving effect s wv ex _) -> case effect of+        Run resource act -> do+          x <- scopedHandler resource act+          pure (ex (x <$ s))+        InScope main ->+          ex <$> withResource \ resource -> run (wv (main resource <$ s))+++-- |Variant of 'interpretScoped' in which the resource allocator is a plain action.+interpretScopedAs ::+  ∀ resource effect r .+  Sem r resource ->+  (∀ r0 x . resource -> effect (Sem r0) x -> Sem r x) ->+  InterpreterFor (Scoped resource effect) r+interpretScopedAs resource =+  interpretScoped \ f -> f =<< resource

@@ -0,0 +1,37 @@+{-# options_haddock prune #-}+-- |Description: Convenience Interpreters for all Conc Effects, Internal+module Polysemy.Conc.Interpreter.Stack where++import Polysemy.Async (Async, asyncToIOFinal)+import Polysemy.Resource (Resource, resourceToIOFinal)++import Polysemy.Conc.Effect.Mask (Mask, UninterruptipleMask)+import Polysemy.Conc.Effect.Race (Race)+import Polysemy.Conc.Interpreter.Mask (Restoration, interpretMaskFinal, interpretUninterruptibleMaskFinal)+import Polysemy.Conc.Interpreter.Race (interpretRace)++-- |A default basic stack with 'Final' for _polysemy-conc_.+type ConcStack =+  [+    UninterruptipleMask Restoration,+    Mask Restoration,+    Race,+    Async,+    Resource,+    Embed IO,+    Final IO+  ]++-- |Interprets 'UninterruptipleMask', 'Mask' and 'Race' in terms of @'Final' 'IO'@ and runs the entire rest of the+-- stack.+runConc ::+  Sem ConcStack a ->+  IO a+runConc =+  runFinal .+  embedToFinal @IO .+  resourceToIOFinal .+  asyncToIOFinal .+  interpretRace .+  interpretMaskFinal .+  interpretUninterruptibleMaskFinal

@@ -0,0 +1,51 @@+module Polysemy.Conc.Monitor where++import qualified Polysemy.Time as Time+import Polysemy.Time (NanoSeconds, Time, TimeUnit, convert, Minutes (Minutes), Seconds (Seconds))+import Torsor (Torsor, difference, minus)++import Polysemy.Conc.Effect.Monitor (MonitorCheck (MonitorCheck))+import Data.Default (Default (def))++-- |Config for 'monitorClockSkew'.+data ClockSkewConfig =+  ClockSkewConfig {+    interval :: NanoSeconds,+    tolerance :: NanoSeconds+  }+  deriving (Eq, Show)++-- |Smart constructor for 'ClockSkewConfig' that takes arbitrary 'TimeUnit's.+clockSkewConfig ::+  TimeUnit u1 =>+  TimeUnit u2 =>+  u1 ->+  u2 ->+  ClockSkewConfig+clockSkewConfig i t =+  ClockSkewConfig (convert i) (convert t)++instance Default ClockSkewConfig where+  def =+    clockSkewConfig (Minutes 1) (Seconds 5)++-- |Check for 'Polysemy.Conc.Effect.Monitor' that checks every @interval@ whether the difference between the current+-- time and the time at the last check is larger than @interval@ + @tolerance@.+-- Can be used to detect that the operating system suspended and resumed.+monitorClockSkew ::+  ∀ t d diff r .+  Torsor t diff =>+  TimeUnit diff =>+  Members [AtomicState (Maybe t), Time t d, Embed IO] r =>+  ClockSkewConfig ->+  MonitorCheck r+monitorClockSkew (ClockSkewConfig interval tolerance) =+  MonitorCheck interval \ signal -> do+    atomicGet >>= \case+      Just prev -> do+        now <- Time.now @t @d+        when (minus (convert (difference now prev)) tolerance > interval) (void (embed @IO (tryPutMVar signal ())))+        atomicPut (Just now)+      Nothing -> do+        now <- Time.now @t @d+        atomicPut (Just now)

@@ -10,6 +10,7 @@   module Relude, ) where +import Control.Exception (try) import Data.Kind (Type) import qualified Data.String.Interpolate as Interpolate import GHC.Err (undefined)@@ -27,6 +28,7 @@   embed,   embedToFinal,   interpret,+  interpretH,   makeSem,   pureT,   raise,@@ -37,6 +39,7 @@   runFinal,   ) import Polysemy.AtomicState (AtomicState, atomicGet, atomicGets, atomicModify', atomicPut)+import Polysemy.Internal.Kind (Append) import Relude hiding (   Reader,   State,@@ -70,3 +73,23 @@ unify =   either id id {-# inline unify #-}++tryAny ::+  Member (Embed IO) r =>+  IO a ->+  Sem r (Either Text a)+tryAny =+  embed @IO . fmap (first show) . try @SomeException+{-# INLINE tryAny #-}++type a ++ b =+  Append a b++leftM ::+  Applicative m =>+  m b ->+  Either a b ->+  m b+leftM f =+  either (const f) pure+{-# inline leftM #-}

@@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack  name:           polysemy-conc-version:        0.4.0.1+version:        0.5.0.0 synopsis:       Polysemy Effects for Concurrency description:    See <https://hackage.haskell.org/package/polysemy-conc/docs/Polysemy-Conc.html> category:       Concurrency@@ -34,6 +34,8 @@       Polysemy.Conc.Effect.Critical       Polysemy.Conc.Effect.Events       Polysemy.Conc.Effect.Interrupt+      Polysemy.Conc.Effect.Mask+      Polysemy.Conc.Effect.Monitor       Polysemy.Conc.Effect.Queue       Polysemy.Conc.Effect.Race       Polysemy.Conc.Effect.Scoped@@ -42,12 +44,16 @@       Polysemy.Conc.Interpreter.Critical       Polysemy.Conc.Interpreter.Events       Polysemy.Conc.Interpreter.Interrupt+      Polysemy.Conc.Interpreter.Mask+      Polysemy.Conc.Interpreter.Monitor       Polysemy.Conc.Interpreter.Queue.Pure       Polysemy.Conc.Interpreter.Queue.TB       Polysemy.Conc.Interpreter.Queue.TBM       Polysemy.Conc.Interpreter.Race       Polysemy.Conc.Interpreter.Scoped+      Polysemy.Conc.Interpreter.Stack       Polysemy.Conc.Interpreter.Sync+      Polysemy.Conc.Monitor       Polysemy.Conc.Prelude       Polysemy.Conc.Queue       Polysemy.Conc.Queue.Result@@ -120,24 +126,28 @@       UndecidableInstances       UnicodeSyntax       ViewPatterns-  ghc-options: -flate-specialise -fspecialise-aggressively -Wall -Wredundant-constraints -Wunused-packages+  ghc-options: -flate-specialise -fspecialise-aggressively -Wall -Wredundant-constraints   build-depends:       async     , base ==4.*     , containers-    , polysemy >=1.5-    , polysemy-time >=0.1.2.1-    , relude >=0.7+    , data-default+    , polysemy >=1.6+    , polysemy-time >=0.1.4+    , relude >=1     , stm     , stm-chans >=2     , string-interpolate >=0.2     , template-haskell     , text     , time+    , torsor     , unagi-chan >=0.4     , unix   mixins:       base hiding (Prelude)+  if impl(ghc >= 8.10)+    ghc-options: -Wunused-packages   default-language: Haskell2010  test-suite polysemy-conc-unit@@ -147,6 +157,8 @@       Polysemy.Conc.Test.EventsTest       Polysemy.Conc.Test.InterruptTest       Polysemy.Conc.Test.LockTest+      Polysemy.Conc.Test.MaskTest+      Polysemy.Conc.Test.MonitorTest       Polysemy.Conc.Test.QueueTest       Polysemy.Conc.Test.SyncTest       Paths_polysemy_conc@@ -210,20 +222,24 @@       UndecidableInstances       UnicodeSyntax       ViewPatterns-  ghc-options: -flate-specialise -fspecialise-aggressively -Wall -Wredundant-constraints -Wunused-packages -threaded -rtsopts -with-rtsopts=-N+  ghc-options: -flate-specialise -fspecialise-aggressively -Wall -Wredundant-constraints -threaded -rtsopts -with-rtsopts=-N   build-depends:-      base ==4.*-    , polysemy >=1.5+      async+    , base ==4.*+    , polysemy     , polysemy-conc+    , polysemy-plugin     , polysemy-test-    , polysemy-time >=0.1.2.1+    , polysemy-time     , stm     , tasty     , time-    , unagi-chan >=0.4+    , unagi-chan     , unix   mixins:       base hiding (Prelude)     , polysemy-conc hiding (Prelude)     , polysemy-conc (Polysemy.Conc.Prelude as Prelude)+  if impl(ghc >= 8.10)+    ghc-options: -Wunused-packages   default-language: Haskell2010

@@ -2,6 +2,8 @@  import Polysemy.Conc.Test.EventsTest (test_events) import Polysemy.Conc.Test.InterruptTest (test_interrupt)+import Polysemy.Conc.Test.MaskTest (test_mask)+import Polysemy.Conc.Test.MonitorTest (test_monitorBasic, test_monitorClockSkew) import Polysemy.Conc.Test.QueueTest (   test_queueBlockTB,   test_queueBlockTBM,@@ -33,6 +35,13 @@     ],     testGroup "interrupt" [       unitTest "interrupt" test_interrupt+    ],+    testGroup "mask" [+      unitTest "mask" test_mask+    ],+    testGroup "monitor" [+      unitTest "basic" test_monitorBasic,+      unitTest "clock skew" test_monitorClockSkew     ]   ] 

@@ -0,0 +1,71 @@+{-# options_ghc -fplugin=Polysemy.Plugin #-}++module Polysemy.Conc.Test.MaskTest where++import qualified Control.Concurrent.Async as Base+import Control.Concurrent.Async (asyncThreadId)+import Control.Exception (throwTo)+import Polysemy (embedFinal)+import Polysemy.Async (Async, async, asyncToIOFinal, await)+import Polysemy.Error (fromExceptionSem, runError)+import Polysemy.Test (Hedgehog, UnitTest, assertEq, evalLeft, runTestAuto)+import System.IO.Error (userError)++import Polysemy.Conc.AtomicState (interpretAtomic)+import Polysemy.Conc.Effect.Mask (restore, uninterruptibleMask)+import qualified Polysemy.Conc.Effect.Sync as Sync+import Polysemy.Conc.Interpreter.Mask (interpretUninterruptibleMaskFinal)+import Polysemy.Conc.Interpreter.Race (interpretRace)+import Polysemy.Conc.Interpreter.Sync (interpretSync)+import qualified Polysemy.Time as Time+import Polysemy.Time (interpretTimeGhc, MilliSeconds (MilliSeconds))++kill ::+  Member (Final IO) r =>+  Base.Async (Maybe ()) ->+  Sem r ()+kill t =+  embedFinal (throwTo (asyncThreadId t) (userError "stop"))++wait ::+  HasCallStack =>+  Members [Hedgehog IO, Async, Final IO] r =>+  Base.Async (Maybe ()) ->+  Sem r ()+wait t =+  withFrozenCallStack do+    void . evalLeft =<< runError @SomeException (fromExceptionSem @SomeException (await t))++test_mask :: UnitTest+test_mask =+  runTestAuto $+  asyncToIOFinal $+  interpretRace $+  interpretTimeGhc $+  interpretUninterruptibleMaskFinal $+  interpretAtomic (0 :: Int) $+  interpretSync @(Proxy 2) $+  interpretSync @(Proxy 1) do+    t1 <- async do+      uninterruptibleMask do+        Sync.putBlock (Proxy @1)+        Sync.takeBlock @(Proxy 2)+        Time.sleep (MilliSeconds 100)+        atomicPut 2+    Sync.takeBlock @(Proxy 1)+    async (kill t1)+    Sync.putBlock (Proxy @2)+    _ <- wait t1+    assertEq 2 =<< atomicGet+    t2 <- async do+      uninterruptibleMask do+        restore do+          Sync.putBlock (Proxy @1)+          void (Sync.takeBlock @(Proxy 2))+          Time.sleep (MilliSeconds 100)+          atomicPut 3+    Sync.takeBlock @(Proxy 1)+    async (kill t2)+    Sync.putBlock (Proxy @2)+    _ <- wait t2+    assertEq 2 =<< atomicGet

@@ -0,0 +1,97 @@+{-# options_ghc -fplugin=Polysemy.Plugin #-}++module Polysemy.Conc.Test.MonitorTest where++import Data.Time (UTCTime)+import Polysemy.Async (async, asyncToIOFinal, await)+import Polysemy.Test (UnitTest, assertEq, assertJust, runTestAuto)+import qualified Polysemy.Time as Time+import Polysemy.Time (+  Hours (Hours),+  MilliSeconds (MilliSeconds),+  Minutes (Minutes),+  NanoSeconds (NanoSeconds),+  Time,+  interpretTimeGhc,+  interpretTimeGhcConstantNow,+  )++import Polysemy.Conc.AtomicState (interpretAtomic)+import qualified Polysemy.Conc.Effect.Monitor as Monitor+import Polysemy.Conc.Effect.Monitor (Monitor, MonitorCheck (MonitorCheck), Restart)+import qualified Polysemy.Conc.Effect.Sync as Sync+import Polysemy.Conc.Effect.Sync (Sync)+import Polysemy.Conc.Interpreter.Monitor (interpretMonitorRestart)+import Polysemy.Conc.Interpreter.Race (interpretRace)+import Polysemy.Conc.Interpreter.Sync (interpretSync)+import Polysemy.Conc.Monitor (clockSkewConfig, monitorClockSkew)++prog ::+  Members [Monitor Restart, Time t d, AtomicState Int, Sync ()] r =>+  Sem r Int+prog = do+  Monitor.monitor do+    atomicModify' (1 +)+    i <- atomicGet+    when (i <= 2) do+      Sync.putBlock ()+      Time.sleep (Hours 1)+    pure i++checker ::+  Members [Time t d, Sync (), Embed IO] r =>+  MVar () ->+  Sem r ()+checker signal =+  Sync.takeBlock *> putMVar signal ()++test_monitorBasic :: UnitTest+test_monitorBasic =+  runTestAuto $+  asyncToIOFinal $+  interpretRace $+  interpretTimeGhc $+  interpretAtomic 0 $+  interpretSync $+  interpretMonitorRestart (MonitorCheck (NanoSeconds 0) checker) do+    assertEq 3 =<< Monitor.restart prog++progSkew ::+  Member (Sync (Proxy "start")) r =>+  Members [Monitor Restart, Time t d, AtomicState Int, Sync (Proxy 1), Sync (Proxy 2), Sync (Proxy 3)] r =>+  Sem r Int+progSkew = do+  Monitor.monitor do+    atomicModify' (1 +)+    Sync.putBlock (Proxy @"start")+    void $ Sync.takeBlock @(Proxy 1)+    Sync.putBlock (Proxy @2)+    void $ Sync.takeBlock @(Proxy 3)+    atomicGet++test_monitorClockSkew :: UnitTest+test_monitorClockSkew =+  runTestAuto $+  asyncToIOFinal $+  interpretRace $+  interpretTimeGhcConstantNow $+  interpretAtomic (0 :: Int) $+  interpretAtomic @(Maybe UTCTime) Nothing $+  interpretSync @(Proxy "start") $+  interpretSync @(Proxy 1) $+  interpretSync @(Proxy 2) $+  interpretSync @(Proxy 3) $+  interpretMonitorRestart (monitorClockSkew (clockSkewConfig (MilliSeconds 1) (Minutes 30))) do+    h <- async (Monitor.restart progSkew)+    _ <- Sync.takeBlock @(Proxy "start")+    Time.adjust (Hours 1)+    _ <- Sync.takeBlock @(Proxy "start")+    Time.adjust (Minutes 10)+    Sync.putBlock (Proxy @1)+    Sync.takeBlock @(Proxy 2)+    Time.adjust (Hours 1)+    _ <- Sync.takeBlock @(Proxy "start")+    Sync.putBlock (Proxy @1)+    Sync.takeBlock @(Proxy 2)+    Sync.putBlock (Proxy @3)+    assertJust 3 =<< await h