polysemy-conc 0.3.0.0 → 0.4.0.0

raw patch · 7 files changed

+143/−13 lines, 7 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• changelog.md +7/−0
• lib/Polysemy/Conc.hs +13/−3
• lib/Polysemy/Conc/Async.hs +47/−0
• lib/Polysemy/Conc/Interpreter/Interrupt.hs +28/−7
• lib/Polysemy/Conc/Sync.hs +18/−2
• polysemy-conc.cabal +3/−1
• test/Polysemy/Conc/Test/LockTest.hs +27/−0

@@ -1,5 +1,12 @@ # Unreleased +# 0.4.0.0++* Add `lock`, a combinator for protecting a region with a mutex.+* Add `scheduleAsync`, a combinator for running an action async that allows the handle to be used before the thread+  starts+* Change the default signal handler for `Interrupt` to `CatchInfo`, catching repeated signals.+ # 0.3.0.0  * Change `Race.timeout` to take a `Sem` for the fallback instead of a pure value.

@@ -30,6 +30,7 @@   interpretSync,   interpretSyncAs,   withSync,+  lock,   interpretScopedSync,   interpretScopedSyncAs, @@ -56,6 +57,7 @@    -- ** Interpreters   interpretInterrupt,+  interpretInterruptOnce,    -- * Event Channels   Events,@@ -85,9 +87,17 @@   withAsyncBlock,   withAsync,   withAsync_,+  scheduleAsync,+  scheduleAsyncIO, ) where -import Polysemy.Conc.Async (withAsync, withAsyncBlock, withAsync_)+import Polysemy.Conc.Async (+  scheduleAsync,+  scheduleAsyncIO,+  withAsync,+  withAsyncBlock,+  withAsync_,+  ) import Polysemy.Conc.AtomicState (interpretAtomic) import Polysemy.Conc.Data.QueueResult (QueueResult) import Polysemy.Conc.Effect.Critical (Critical)@@ -99,7 +109,7 @@ 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)+import Polysemy.Conc.Interpreter.Interrupt (interpretInterrupt, interpretInterruptOnce) import Polysemy.Conc.Interpreter.Queue.Pure (   interpretQueueListReadOnlyAtomic,   interpretQueueListReadOnlyAtomicWith,@@ -114,7 +124,7 @@ import Polysemy.Conc.Queue.Result (resultToMaybe) import Polysemy.Conc.Race (race_, timeoutAs, timeoutAs_, timeoutMaybe, timeoutU, timeout_) import Polysemy.Conc.Retry (retrying, retryingWithError)-import Polysemy.Conc.Sync (withSync)+import Polysemy.Conc.Sync (lock, withSync)  -- $intro -- This library provides an assortment of tools for concurrency-related tasks:

@@ -6,7 +6,11 @@ import Polysemy.Time (MilliSeconds (MilliSeconds), TimeUnit)  import Polysemy.Conc.Effect.Race (Race)+import qualified Polysemy.Conc.Effect.Sync as Sync import qualified Polysemy.Conc.Race as Race+import Polysemy.Conc.Interpreter.Sync (interpretSync)+import Polysemy.Conc.Effect.Sync (Sync, ScopedSync)+import Polysemy.Conc.Sync (withSync)  -- |Run the first action asynchronously while the second action executes, then cancel the first action. -- Passes the handle into the action to allow it to await its result.@@ -57,3 +61,46 @@   Sem r a withAsync_ mb =   withAsync mb . const++-- |Run an action with 'async', but don't start it right away, so the thread handle can be processed before the action+-- executes.+--+-- Takes a callback function that is invoked after spawning the thread.+-- The callback receives the 'Base.Async' handle and a unit action that starts the computation.+--+-- This is helpful if the 'Base.Async' has to be stored in state and the same state is written when the action finishes.+-- In that case, the race condition causes the handle to be written over the finished state.+--+-- @+-- makeRequest = put Nothing+--+-- main = scheduleAsync makeRequest \ handle start -> do+--   put (Just handle)+--   start -- now makeRequest is executed+-- @+scheduleAsync ::+  ∀ res b r a .+  Members [ScopedSync res (), Async, Race] r =>+  Sem r b ->+  (Base.Async (Maybe b) -> Sem (Sync () : r) () -> Sem (Sync () : r) a) ->+  Sem r a+scheduleAsync mb f =+  withSync @() @res do+    h <- async do+      Sync.readBlock @()+      raise mb+    f h (Sync.putBlock ())++-- |Variant of 'scheduleAsync' that directly interprets the 'MVar' used for signalling.+scheduleAsyncIO ::+  ∀ b r a .+  Members [Resource, Async, Race, Embed IO] r =>+  Sem r b ->+  (Base.Async (Maybe b) -> Sem (Sync () : r) () -> Sem (Sync () : r) a) ->+  Sem r a+scheduleAsyncIO mb f =+  interpretSync @() do+    h <- async do+      Sync.readBlock @()+      raise mb+    f h (Sync.putBlock ())

@@ -12,7 +12,7 @@ import Polysemy.AtomicState (runAtomicStateTVar) import Polysemy.Internal.Tactics (liftT) import Polysemy.Time (Seconds (Seconds))-import System.Posix.Signals (Handler (CatchInfoOnce, CatchOnce), SignalInfo, installHandler, keyboardSignal)+import System.Posix.Signals (Handler (CatchInfoOnce, CatchOnce, CatchInfo), SignalInfo, installHandler, keyboardSignal)  import qualified Polysemy.Conc.Effect.Critical as Critical import Polysemy.Conc.Effect.Critical (Critical)@@ -173,23 +173,44 @@  installSignalHandler ::   TVar InterruptState ->+  ((SignalInfo -> IO ()) -> Handler) ->   IO Handler-installSignalHandler state =-  installHandler keyboardSignal (CatchInfoOnce handler) Nothing+installSignalHandler state consHandler =+  installHandler keyboardSignal (consHandler handler) Nothing   where     handler sig =       runFinal $ embedToFinal @IO $ runAtomicStateTVar state (broadcastInterrupt sig)  -- |Interpret 'Interrupt' by installing a signal handler.-interpretInterrupt ::+--+-- Takes a constructor for 'Handler'.+interpretInterruptWith ::   Members [Critical, Race, Async, Embed IO] r =>+  ((SignalInfo -> IO ()) -> Handler) ->   InterpreterFor Interrupt r-interpretInterrupt sem = do+interpretInterruptWith consHandler sem = do   quitMVar <- newEmptyMVar   finishMVar <- newEmptyMVar   state <- newTVarIO (InterruptState quitMVar finishMVar Set.empty (const pass) Map.empty)-  orig <- embed $ installSignalHandler state+  orig <- embed $ installSignalHandler state consHandler   runAtomicStateTVar state do     atomicModify' \ s -> s {original = originalHandler orig}     interpretInterruptState $ raiseUnder sem-{-# inline interpretInterrupt #-}++-- |Interpret 'Interrupt' by installing a signal handler.+--+-- Catches repeat invocations of SIGINT.+interpretInterrupt ::+  Members [Critical, Race, Async, Embed IO] r =>+  InterpreterFor Interrupt r+interpretInterrupt =+  interpretInterruptWith CatchInfo++-- |Interpret 'Interrupt' by installing a signal handler.+--+-- Catches only the first invocation of SIGINT.+interpretInterruptOnce ::+  Members [Critical, Race, Async, Embed IO] r =>+  InterpreterFor Interrupt r+interpretInterruptOnce =+  interpretInterruptWith CatchInfoOnce

@@ -7,7 +7,7 @@ import qualified Polysemy.Time as Time import Polysemy.Time (Time, TimeUnit) -import Polysemy.Conc.Effect.Scoped (Scoped, scoped)+import Polysemy.Conc.Effect.Scoped (scoped) import qualified Polysemy.Conc.Effect.Sync as Sync import Polysemy.Conc.Effect.Sync (   ScopedSync,@@ -27,6 +27,7 @@   try,   wait,   )+import Polysemy.Resource (finally, Resource)  -- |Run an action repeatedly until the 'Sync' variable is available. whileEmpty ::@@ -59,7 +60,22 @@ -- |Run an action with a locally scoped 'Sync' variable. withSync ::   ∀ d res r .-  Member (Scoped (SyncResources res) (Sync d)) r =>+  Member (ScopedSync res d) r =>   InterpreterFor (Sync d) r withSync =   scoped @(SyncResources res)++-- |Run the action @ma@ with an exclusive lock (mutex).+-- When multiple threads call the action concurrently, only one is allowed to execute it at a time.+-- The value @l@ is used to disambiguate the 'Sync' from other uses of the combinator.+-- You can pass in something like @Proxy @"db-write"@.+--+-- /Note:/ The 'Sync' must be interpreted with an initially full @MVar@, e.g. using 'Polysemy.Conc.interpretSyncAs'.+lock ::+  ∀ l r a .+  Members [Sync l, Resource] r =>+  l ->+  Sem r a ->+  Sem r a+lock l ma =+  finally (takeBlock @l *> ma) (putTry l)

@@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack  name:           polysemy-conc-version:        0.3.0.0+version:        0.4.0.0 synopsis:       Polysemy Effects for Concurrency description:    See <https://hackage.haskell.org/package/polysemy-conc/docs/Polysemy-Conc.html> category:       Concurrency@@ -146,6 +146,7 @@   other-modules:       Polysemy.Conc.Test.EventsTest       Polysemy.Conc.Test.InterruptTest+      Polysemy.Conc.Test.LockTest       Polysemy.Conc.Test.QueueTest       Polysemy.Conc.Test.SyncTest       Paths_polysemy_conc@@ -218,6 +219,7 @@     , polysemy-time >=0.1.2.1     , stm     , tasty+    , time     , unagi-chan >=0.4     , unix   mixins:

@@ -0,0 +1,27 @@+module Polysemy.Conc.Test.LockTest where++import Data.Time (Day, UTCTime)+import Polysemy.Async (asyncToIOFinal, sequenceConcurrently)+import Polysemy.Test (UnitTest, assertEq, runTestAuto)+import qualified Polysemy.Time as Time+import Polysemy.Time (MicroSeconds (MicroSeconds), interpretTimeGhc)++import Polysemy.Conc.AtomicState (interpretAtomic)+import Polysemy.Conc.Interpreter.Race (interpretRace)+import Polysemy.Conc.Interpreter.Sync (interpretSyncAs)+import Polysemy.Conc.Sync (lock)++test_lock :: UnitTest+test_lock =+  runTestAuto $+  interpretTimeGhc $+  asyncToIOFinal $+  interpretRace $+  interpretSyncAs () $+  interpretAtomic (0 :: Int) do+    void $ sequenceConcurrently @[] $ [(1 :: Int)..100] $> do+      lock () do+        cur <- atomicGet @Int+        Time.sleep @UTCTime @Day (MicroSeconds 100)+        atomicPut (cur + 1)+    assertEq @Int @IO 100 =<< atomicGet