polysemy-conc 0.14.0.0 → 0.14.1.0

raw patch · 8 files changed

+182/−69 lines, 8 filesPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

API changes (from Hackage documentation)

Files

• lib/Polysemy/Conc/Effect/Monitor.hs +2/−2
• lib/Polysemy/Conc/Interpreter/Monitor.hs +21/−11
• lib/Polysemy/Conc/Monitor.hs +5/−9
• polysemy-conc.cabal +3/−1
• test/Main.hs +21/−18
• test/Polysemy/Conc/Test/LockTest.hs +4/−3
• test/Polysemy/Conc/Test/MonitorTest.hs +112/−25
• test/Polysemy/Conc/Test/Run.hs +14/−0

@@ -38,7 +38,7 @@ data MonitorCheck r =   MonitorCheck {     interval :: NanoSeconds,-    check :: MVar () -> Sem r ()+    check :: Sem r Bool   }  -- | Transform the stack of the check in a 'MonitorCheck'.@@ -47,7 +47,7 @@   MonitorCheck r ->   MonitorCheck r' hoistMonitorCheck f MonitorCheck {..} =-  MonitorCheck {check = f . check, ..}+  MonitorCheck {check = f check, ..}  -- | Start a region that can contain monitor-intervention regions. withMonitor ::

@@ -18,8 +18,8 @@ import qualified Polysemy.Conc.Effect.Race as Race import Polysemy.Conc.Effect.Race (Race) -newtype CancelResource =-  CancelResource { signal :: MVar () }+data CancelResource =+  CancelResource { kill :: MVar () }  data MonitorCancel =   MonitorCancel@@ -33,14 +33,23 @@   (CancelResource -> Sem r a) ->   Sem r a monitorRestart (MonitorCheck interval check) use = do-  sig <- embedFinal @IO newEmptyMVar-  withAsync_ (Time.loop_ @t @d interval (check sig)) (spin sig)-  where-    spin sig = do-      let res = (CancelResource sig)-      void (embedFinal @IO (tryTakeMVar sig))-      either (const (spin sig)) pure =<< errorToIOFinal @MonitorCancel (fromExceptionSem @MonitorCancel (raise (use res)))+  kill <- embedFinal @IO newEmptyMVar+  start <- embedFinal @IO newEmptyMVar+  let +    runCheck = do+      embedFinal (readMVar start)+      whenM check $ embedFinal @IO do+        takeMVar start+        void (tryPutMVar kill ())++    spin = do+      let res = CancelResource {..}+      embedFinal (putMVar start ())+      leftA (const spin) =<< errorToIOFinal @MonitorCancel (fromExceptionSem @MonitorCancel (raise (use res)))++  withAsync_ (Time.loop_ @t @d interval runCheck) spin+ -- | Interpret @'Polysemy.Conc.Scoped' 'Monitor'@ with the 'Polysemy.Conc.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'.@@ -51,8 +60,9 @@   InterpreterFor RestartingMonitor r interpretMonitorRestart check =   interpretScopedH (const (monitorRestart @t @d (hoistMonitorCheck raise check))) \ CancelResource {..} -> \case-    Monitor ma ->-      either (const (Base.throw MonitorCancel)) pure =<< Race.race (embedFinal @IO (readMVar signal)) (runTSimple ma)+    Monitor ma -> do+      void (embedFinal @IO (tryTakeMVar kill))+      leftA (const (Base.throw MonitorCancel)) =<< Race.race (embedFinal @IO (readMVar kill)) (runTSimple ma)  interpretMonitorPure' :: () -> InterpreterFor (Monitor action) r interpretMonitorPure' _ =

@@ -42,12 +42,8 @@   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)+  MonitorCheck interval do+    now <- Time.now @t @d+    atomicState \ s -> (Just now, maybe False (skewed now) s)+  where+    skewed now prev = minus (convert (difference now prev)) tolerance > interval

@@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack  name:           polysemy-conc-version:        0.14.0.0+version:        0.14.1.0 synopsis:       Polysemy effects for concurrency description:    See https://hackage.haskell.org/package/polysemy-conc/docs/Polysemy-Conc.html category:       Concurrency@@ -132,6 +132,7 @@       Polysemy.Conc.Test.MaskTest       Polysemy.Conc.Test.MonitorTest       Polysemy.Conc.Test.QueueTest+      Polysemy.Conc.Test.Run       Polysemy.Conc.Test.SyncTest   hs-source-dirs:       test@@ -185,6 +186,7 @@     , tasty >=1.4.2 && <1.5     , tasty-hedgehog >=1.3.0.0 && <1.5     , time >=1.12.2 && <1.13+    , torsor ==0.1.*   mixins:       base hiding (Prelude)     , incipit-core (IncipitCore as Prelude)

@@ -1,6 +1,5 @@ module Main where -import Hedgehog (property, test, withTests) import Polysemy.Conc.Test.EventsTest (test_events) import Polysemy.Conc.Test.LockTest (test_lock) import Polysemy.Conc.Test.MaskTest (test_mask)@@ -13,39 +12,43 @@   test_queueTBM,   test_queueTimeoutTBM,   )+import Polysemy.Conc.Test.Run (unitTestTimes) import Polysemy.Conc.Test.SyncTest (test_sync, test_syncLock)-import Polysemy.Test (unitTest)-import Test.Tasty (TestTree, defaultMain, testGroup)-import Test.Tasty.Hedgehog (testProperty)+import Test.Tasty (TestTree, Timeout (NoTimeout), adjustOption, defaultMain, mkTimeout, testGroup) +defaultTimeout :: Timeout -> Timeout+defaultTimeout = \case+  NoTimeout -> mkTimeout 60_000_000+  t -> t+ tests :: TestTree tests =+  adjustOption defaultTimeout $   testGroup "main" [     testGroup "queue" [-      unitTest "TBM success" test_queueTBM,-      unitTest "TBM timeout" test_queueTimeoutTBM,-      unitTest "TBM peek" test_queuePeekTBM,-      unitTest "TBM block" test_queueBlockTBM,-      unitTest "TB success" test_queueTB,-      unitTest "TB block" test_queueBlockTB+      unitTestTimes 100 "TBM success" test_queueTBM,+      unitTestTimes 100 "TBM timeout" test_queueTimeoutTBM,+      unitTestTimes 100 "TBM peek" test_queuePeekTBM,+      unitTestTimes 10 "TBM block" test_queueBlockTBM,+      unitTestTimes 100 "TB success" test_queueTB,+      unitTestTimes 10 "TB block" test_queueBlockTB     ],     testGroup "events" [-      testProperty "events" (withTests 100 (property (test test_events)))+      unitTestTimes 100 "events" test_events     ],     testGroup "sync" [-      unitTest "sync" test_sync,-      unitTest "lock" test_syncLock+      unitTestTimes 100 "sync" test_sync,+      unitTestTimes 10 "lock" test_syncLock     ],     test_lock,     testGroup "mask" [-      unitTest "mask" test_mask+      unitTestTimes 10 "mask" test_mask     ],     testGroup "monitor" [-      unitTest "basic" test_monitorBasic,-      unitTest "clock skew" test_monitorClockSkew+      unitTestTimes 100 "basic" test_monitorBasic,+      unitTestTimes 1000 "clock skew" test_monitorClockSkew     ]   ]  main :: IO ()-main =-  defaultMain tests+main = defaultMain tests

@@ -2,7 +2,7 @@  module Polysemy.Conc.Test.LockTest where -import Polysemy.Test (UnitTest, assert, assertJust, runTestAuto, unitTest)+import Polysemy.Test (UnitTest, assert, assertJust, runTestAuto) import Polysemy.Time (GhcTime, MilliSeconds (MilliSeconds), Seconds (Seconds), interpretTimeGhc) import Test.Tasty (TestTree, testGroup) @@ -16,6 +16,7 @@ import Polysemy.Conc.Interpreter.Race (interpretRace) import Polysemy.Conc.Interpreter.Sync (interpretSync) import Polysemy.Conc.Race (timeout_)+import Polysemy.Conc.Test.Run (unitTestTimes)  interpretLockTest ::   Members [Resource, Embed IO, Final IO] r =>@@ -52,6 +53,6 @@ test_lock :: TestTree test_lock =   testGroup "lock" [-    unitTest "basic" test_lockBasic,-    unitTest "reentry" test_lockReentry+    unitTestTimes 10 "basic" test_lockBasic,+    unitTestTimes 10 "reentry" test_lockReentry   ]

@@ -2,18 +2,25 @@  module Polysemy.Conc.Test.MonitorTest where +import Control.Concurrent (threadDelay)+import qualified Data.Text as Text import Data.Time (UTCTime)-import Polysemy.Test (UnitTest, assertEq, assertJust, runTestAuto)+import Polysemy.Test (TestError (TestError), UnitTest, assertEq, assertJust, runTestAuto) import qualified Polysemy.Time as Time import Polysemy.Time (+  GhcTime,   Hours (Hours),   MilliSeconds (MilliSeconds),   Minutes (Minutes),   NanoSeconds (NanoSeconds),+  Seconds (Seconds),   Time,+  TimeUnit,+  convert,   interpretTimeGhc,   interpretTimeGhcConstantNow,   )+import Torsor (difference)  import Polysemy.Conc.AtomicState (interpretAtomic) import qualified Polysemy.Conc.Effect.Monitor as Monitor@@ -39,10 +46,9 @@  checker ::   Members [Time t d, Sync (), Embed IO] r =>-  MVar () ->-  Sem r ()-checker signal =-  Sync.takeBlock *> embed (putMVar signal ())+  Sem r Bool+checker =+  True <$ Sync.takeBlock  test_monitorBasic :: UnitTest test_monitorBasic =@@ -55,42 +61,123 @@   interpretMonitorRestart (MonitorCheck (NanoSeconds 0) checker) do     assertEq 3 =<< Monitor.restart prog +locked ::+  Members [AtomicState [Text], Error TestError, Embed IO] r =>+  Sem r a+locked = do+  embed (threadDelay 1_000_000)+  msgs <- atomicGet+  throw (TestError (Text.unlines ("Locked after:" : reverse msgs)))++takeBlock ::+  ∀ label r .+  HasCallStack =>+  Members [AtomicState [Text], Sync (Proxy label), Error TestError, Embed IO] r =>+  Text ->+  Sem r ()+takeBlock desc =+  withFrozenCallStack do+    atomicModify' (("taking '" <> desc <> "'") :)+    whenM (isNothing <$> Sync.takeWait (Seconds 1)) locked++putBlock ::+  ∀ label r .+  HasCallStack =>+  Members [AtomicState [Text], Sync (Proxy label), Error TestError, Embed IO] r =>+  Text ->+  Sem r ()+putBlock desc =+  withFrozenCallStack do+    atomicModify' (("putting '" <> desc <> "'") :)+    unlessM (Sync.putWait (Seconds 1) Proxy) locked+ progSkew ::-  Member (Sync (Proxy "start")) r =>-  Members [Monitor Restart, Time t d, AtomicState Int, Sync (Proxy 1), Sync (Proxy 2), Sync (Proxy 3)] r =>+  Members [Sync (Proxy "start"), Sync (Proxy "1"), Sync (Proxy "2"), Sync (Proxy "3"), Embed IO] r =>+  Members [Monitor Restart, AtomicState [Text], AtomicState Int, Error TestError, Time t d] r =>   Sem r Int-progSkew = do+progSkew =   Monitor.monitor do     atomicModify' (1 +)-    Sync.putBlock (Proxy @"start")-    void $ Sync.takeBlock @(Proxy 1)-    Sync.putBlock (Proxy @2)-    void $ Sync.takeBlock @(Proxy 3)+    putBlock @"start" "start"+    takeBlock @"1" "1"+    putBlock @"2" "2"+    takeBlock @"3" "3"     atomicGet +sleepPoll ::+  Members [GhcTime, Embed IO] r =>+  TimeUnit u =>+  u ->+  UTCTime ->+  Sem r ()+sleepPoll duration start =+  spin+  where+    spin = do+      embed (threadDelay 10_000)+      unlessM (later <$> Time.now) spin+    later now =+      difference now start >= diff+    diff =+      convert duration++interceptTime ::+  ∀ r a .+  Members [AtomicState [Text], GhcTime, Sync (Proxy "sleep"), Error TestError, Embed IO] r =>+  Sem r a ->+  Sem r a+interceptTime =+  intercept \case+    Time.Now -> do+      atomicModify' ("now" :)+      Time.now+    Time.Today ->+      Time.today+    Time.Sleep t -> do+      atomicModify' ("sleep" :)+      now <- Time.now+      putBlock @"sleep" "sleep"+      sleepPoll t now+      atomicModify' ("slept" :)+    Time.SetTime now ->+      Time.setTime now+    Time.Adjust (diff :: u) -> do+      atomicModify' ("adjust" :)+      send (Time.Adjust diff)+    Time.SetDate startAt ->+      Time.setDate startAt+ test_monitorClockSkew :: UnitTest test_monitorClockSkew =   runTestAuto $   asyncToIOFinal $   interpretRace $-  interpretTimeGhcConstantNow $   interpretAtomic (0 :: Int) $+  interpretAtomic ([] :: [Text]) $   interpretAtomic @(Maybe UTCTime) Nothing $   interpretSync @(Proxy "start") $-  interpretSync @(Proxy 1) $-  interpretSync @(Proxy 2) $-  interpretSync @(Proxy 3) $-  interpretMonitorRestart (monitorClockSkew (clockSkewConfig (MilliSeconds 1) (Minutes 30))) do+  interpretSync @(Proxy "sleep") $+  interpretSync @(Proxy "1") $+  interpretSync @(Proxy "2") $+  interpretSync @(Proxy "3") $+  interpretTimeGhcConstantNow $+  interceptTime $+  interpretMonitorRestart monitor do     h <- async (Monitor.restart progSkew)-    _ <- Sync.takeBlock @(Proxy "start")+    takeBlock @"start" "start 1"+    takeBlock @"sleep" "sleep"     Time.adjust (Hours 1)-    _ <- Sync.takeBlock @(Proxy "start")+    _ <- takeBlock @"start" "start 2"+    takeBlock @"sleep" "sleep"     Time.adjust (Minutes 10)-    Sync.putBlock (Proxy @1)-    Sync.takeBlock @(Proxy 2)+    putBlock @"1" "1 1"+    takeBlock @"2" "2 1"+    takeBlock @"sleep" "sleep"     Time.adjust (Hours 1)-    _ <- Sync.takeBlock @(Proxy "start")-    Sync.putBlock (Proxy @1)-    Sync.takeBlock @(Proxy 2)-    Sync.putBlock (Proxy @3)+    _ <- takeBlock @"start" "start 3"+    putBlock @"1" "1 2"+    takeBlock @"2" "2 2"+    putBlock @"3" "3"     assertJust 3 =<< await h+  where+    monitor = monitorClockSkew (clockSkewConfig (MilliSeconds 1) (Minutes 30))

@@ -0,0 +1,14 @@+module Polysemy.Conc.Test.Run where++import Hedgehog (TestLimit, property, test, withTests)+import Polysemy.Test (UnitTest)+import Test.Tasty (TestName, TestTree)+import Test.Tasty.Hedgehog (testProperty)++unitTestTimes ::+  TestLimit ->+  TestName ->+  UnitTest ->+  TestTree+unitTestTimes n desc =+  testProperty desc . withTests n . property . test