polysemy-conc 0.1.1.0 → 0.2.0.0

raw patch · 26 files changed

+640/−304 lines, 26 filesdep −hedgehogdep −tasty-hedgehogPVP ok

version bump matches the API change (PVP)

Dependencies removed: hedgehog, tasty-hedgehog

API changes (from Hackage documentation)

Files

• Changelog.md +0/−0
• changelog.md +10/−0
• lib/Polysemy/Conc.hs +50/−9
• lib/Polysemy/Conc/Async.hs +59/−0
• lib/Polysemy/Conc/AtomicState.hs +14/−0
• lib/Polysemy/Conc/Critical.hs +2/−2
• lib/Polysemy/Conc/Data/Critical.hs +2/−2
• lib/Polysemy/Conc/Effect/Scoped.hs +6/−4
• lib/Polysemy/Conc/Effect/Sync.hs +13/−0
• lib/Polysemy/Conc/Events.hs +27/−0
• lib/Polysemy/Conc/Interpreter/Events.hs +27/−6
• lib/Polysemy/Conc/Interpreter/Queue/Pure.hs +113/−0
• lib/Polysemy/Conc/Interpreter/Queue/TB.hs +65/−0
• lib/Polysemy/Conc/Interpreter/Queue/TBM.hs +76/−0
• lib/Polysemy/Conc/Interpreter/Sync.hs +26/−1
• lib/Polysemy/Conc/Interrupt.hs +4/−3
• lib/Polysemy/Conc/Prelude.hs +3/−3
• lib/Polysemy/Conc/Queue.hs +26/−105
• lib/Polysemy/Conc/Queue/Result.hs +5/−5
• lib/Polysemy/Conc/Queue/TB.hs +0/−65
• lib/Polysemy/Conc/Queue/TBM.hs +0/−67
• lib/Polysemy/Conc/Race.hs +20/−9
• lib/Polysemy/Conc/Retry.hs +63/−0
• lib/Polysemy/Conc/Sync.hs +14/−3
• polysemy-conc.cabal +10/−15
• test/Polysemy/Conc/Test/QueueTest.hs +5/−5

@@ -0,0 +1,10 @@+# Unreleased++# 0.2.0.0+* Add `read*` constructors for `Sync`+* Add `subscribeWhile`, a combinator that consumes events until a condition is met+* Add looping combinators for `Queue`+* Add `retry`, a combinator that runs an action repeatedly until it returns `Right` or a timeout is hit+* Add `Scoped`, an effect for local resource scoping+* Add `withAsync`, a bracketing combinator that runs an async action while the main action runs+* Add `interpretAtomic`, a convenience interpreter for `AtomicState` that runs `runAtomicStateTVar`

@@ -17,12 +17,21 @@   interpretQueueListReadOnlyState,   interpretQueueListReadOnlyStateWith, +  -- ** Combinators+  loop,+  loopOr,+   -- * MVars   -- $mvar   Sync,+  ScopedSync,    -- ** Interpreters   interpretSync,+  interpretSyncAs,+  withSync,+  interpretScopedSync,+  interpretScopedSyncAs,    -- * Racing   -- $race@@ -31,6 +40,9 @@   race_,   timeout,   timeout_,+  timeoutU,+  retrying,+  retryingWithError,    -- ** Interpreters   interpretRace,@@ -46,30 +58,59 @@   Events,   publish,   consume,+  subscribe,+  subscribeWhile,+  subscribeLoop,+  EventToken,+  EventChan,+  ChanEvents,+  EventConsumer,+  ChanConsumer,    -- ** Interpreters   interpretEventsChan,++  -- * Exceptions+  Critical,++  -- ** Interpreters+  interpretCritical,+  interpretCriticalNull,++  -- * Other Combinators+  interpretAtomic,+  withAsyncBlock,+  withAsync,+  withAsync_, ) where +import Polysemy.Conc.Async (withAsync, withAsyncBlock, withAsync_)+import Polysemy.Conc.AtomicState (interpretAtomic)+import Polysemy.Conc.Critical (interpretCritical, interpretCriticalNull)+import Polysemy.Conc.Data.Critical (Critical) import Polysemy.Conc.Data.Interrupt (Interrupt) import Polysemy.Conc.Data.Queue (Queue) import Polysemy.Conc.Data.QueueResult (QueueResult) import Polysemy.Conc.Data.Race (Race, race, timeout)-import Polysemy.Conc.Effect.Events (Events, publish, consume)-import Polysemy.Conc.Effect.Sync (Sync)-import Polysemy.Conc.Interpreter.Events (interpretEventsChan)-import Polysemy.Conc.Interpreter.Sync (interpretSync)-import Polysemy.Conc.Interrupt (interpretInterrupt)-import Polysemy.Conc.Queue (+import Polysemy.Conc.Effect.Events (EventToken, Events, consume, publish, subscribe)+import Polysemy.Conc.Effect.Sync (ScopedSync, Sync)+import Polysemy.Conc.Events (subscribeLoop, subscribeWhile)+import Polysemy.Conc.Interpreter.Events (ChanConsumer, ChanEvents, EventChan, EventConsumer, interpretEventsChan)+import Polysemy.Conc.Interpreter.Queue.Pure (   interpretQueueListReadOnlyAtomic,   interpretQueueListReadOnlyAtomicWith,   interpretQueueListReadOnlyState,   interpretQueueListReadOnlyStateWith,   )+import Polysemy.Conc.Interpreter.Queue.TB (interpretQueueTB)+import Polysemy.Conc.Interpreter.Queue.TBM (interpretQueueTBM)+import Polysemy.Conc.Interpreter.Sync (interpretScopedSync, interpretScopedSyncAs, interpretSync, interpretSyncAs)+import Polysemy.Conc.Interrupt (interpretInterrupt)+import Polysemy.Conc.Queue (loop, loopOr) import Polysemy.Conc.Queue.Result (resultToMaybe)-import Polysemy.Conc.Queue.TB (interpretQueueTB)-import Polysemy.Conc.Queue.TBM (interpretQueueTBM)-import Polysemy.Conc.Race (interpretRace, race_, timeout_)+import Polysemy.Conc.Race (interpretRace, race_, timeoutU, timeout_)+import Polysemy.Conc.Retry (retrying, retryingWithError)+import Polysemy.Conc.Sync (withSync)  -- $intro -- This library provides an assortment of tools for concurrency-related tasks:

@@ -0,0 +1,59 @@+module Polysemy.Conc.Async where++import qualified Control.Concurrent.Async as Base+import Polysemy.Async (Async, async, cancel)+import Polysemy.Resource (Resource, bracket)+import Polysemy.Time (MilliSeconds (MilliSeconds), TimeUnit)++import Polysemy.Conc.Data.Race (Race)+import qualified Polysemy.Conc.Race as Race++-- |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.+--+-- When cancelling, this variant will wait indefinitely for the thread to be gone.+withAsyncBlock ::+  Members [Resource, Async] r =>+  Sem r b ->+  (Base.Async (Maybe b) -> Sem r a) ->+  Sem r a+withAsyncBlock mb =+  bracket (async mb) cancel++-- |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.+--+-- When cancelling, this variant will wait for the specified interval for the thread to be gone.+withAsyncWait ::+  TimeUnit u =>+  Members [Resource, Race, Async] r =>+  u ->+  Sem r b ->+  (Base.Async (Maybe b) -> Sem r a) ->+  Sem r a+withAsyncWait interval mb =+  bracket (async mb) (Race.timeoutU interval . cancel)++-- |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.+--+-- When cancelling, this variant will wait for 500ms for the thread to be gone.+withAsync ::+  Members [Resource, Race, Async] r =>+  Sem r b ->+  (Base.Async (Maybe b) -> Sem r a) ->+  Sem r a+withAsync =+  withAsyncWait (MilliSeconds 500)++-- |Run the first action asynchronously while the second action executes, then cancel the first action.+-- Discards the handle, expecting the async action to either terminate or be cancelled.+--+-- When cancelling, this variant will wait for 500ms for the thread to be gone.+withAsync_ ::+  Members [Resource, Race, Async] r =>+  Sem r b ->+  Sem r a ->+  Sem r a+withAsync_ mb =+  withAsync mb . const

@@ -0,0 +1,14 @@+module Polysemy.Conc.AtomicState where++import Polysemy.AtomicState (runAtomicStateTVar)++-- |Convenience wrapper around 'runAtomicStateTVar' that creates a new 'TVar'.+interpretAtomic ::+  ∀ a r .+  Member (Embed IO) r =>+  a ->+  InterpreterFor (AtomicState a) r+interpretAtomic initial sem = do+  tv <- newTVarIO initial+  runAtomicStateTVar tv sem+{-# inline interpretAtomic #-}

@@ -20,7 +20,7 @@       where         run ma' s =           Exception.catch (fmap Right <$> ma') \ se -> pure (Left se <$ s)-{-# INLINE interpretCritical #-}+{-# inline interpretCritical #-}  -- |Interpret 'Critical' by doing nothing. interpretCriticalNull ::@@ -29,4 +29,4 @@   interpretH \case     Catch ma ->       fmap (fmap Right) . raise . interpretCriticalNull =<< runT ma-{-# INLINE interpretCriticalNull #-}+{-# inline interpretCriticalNull #-}

@@ -4,8 +4,8 @@  -- |An effect that catches exceptions. ----- The point of this is to catch exceptions that haven't been anticipated, as a failsafe.--- Used internally in this library.+-- Provides the exact functionality of `Polysemy.Error.fromExceptionSem`, but pushes the dependency on @Final IO@ to the+-- interpreter, and makes it optional. data Critical :: Effect where   -- |Catch all exceptions of type @e@ in this computation.   Catch :: Exception e => m a -> Critical m (Either e a)

@@ -44,15 +44,17 @@ -- |Interpreter for 'Scoped', taking a @resource@ allocation function and a parameterized interpreter for the plain -- @effect@. ----- @scope@ is a callback function, allowing the user to compute the resource for each program from other effects.+-- @withResource@ is a callback function, allowing the user to acquire the resource for each program from other effects. -- -- @scopedInterpreter@ is a regular interpreter that is called with the @resource@ argument produced by @scope@.+-- /Note/: This function will be called for each action in the program, so if the interpreter allocates any resources,+-- they will be scoped to a single action. Move them to @withResource@ instead. runScoped ::   ∀ resource effect r .   (∀ x . (resource -> Sem r x) -> Sem r x) ->   (resource -> InterpreterFor effect r) ->   InterpreterFor (Scoped resource effect) r-runScoped scope scopedInterpreter =+runScoped withResource scopedInterpreter =   run   where     run :: InterpreterFor (Scoped resource effect) r@@ -60,8 +62,8 @@       interpretH' \ (Weaving effect s wv ex ins) -> case effect of         Run resource act ->           scopedInterpreter resource (liftSem $ injWeaving $ Weaving act s (raise . run . wv) ex ins)-        InScope main -> do-          ex <$> scope \ resource -> run (wv (main resource <$ s))+        InScope main ->+          ex <$> withResource \ resource -> run (wv (main resource <$ s))  -- |Variant of 'Scoped' in which the resource allocator is a plain action. runScopedAs ::

@@ -3,6 +3,7 @@ module Polysemy.Conc.Effect.Sync where  import Polysemy.Time (TimeUnit)+import Polysemy.Conc.Effect.Scoped (Scoped)  -- |Abstracts an 'MVar'. --@@ -30,6 +31,12 @@   TakeWait :: TimeUnit u => u -> Sync d m (Maybe d)   -- |Take the variable, returning 'Nothing' immmediately if no value was available.   TakeTry :: Sync d m (Maybe d)+  -- |Read the variable, waiting until a value is available.+  ReadBlock :: Sync d m d+  -- |Read the variable, waiting until a value is available or the timeout has expired.+  ReadWait :: TimeUnit u => u -> Sync d m (Maybe d)+  -- |Read the variable, returning 'Nothing' immmediately if no value was available.+  ReadTry :: Sync d m (Maybe d)   -- |Write the variable, waiting until it is writable.   PutBlock :: d -> Sync d m ()   -- |Write the variable, waiting until it is writable or the timeout has expired.@@ -40,3 +47,9 @@   Empty :: Sync d m Bool  makeSem ''Sync++newtype SyncResources a =+  SyncResources { unSyncResources :: a }++type ScopedSync res a =+  Scoped (SyncResources res) (Sync a)

@@ -0,0 +1,27 @@+-- |Description: Events Combinators+module Polysemy.Conc.Events where++import Polysemy.Conc.Interpreter.Events (EventConsumer)+import qualified Polysemy.Conc.Effect.Events as Events++-- |Pull repeatedly from the 'Events' channel, passing the event to the supplied callback.+-- Stop when the action returns @False@.+subscribeWhile ::+  ∀ e token r .+  Member (EventConsumer token e) r =>+  (e -> Sem r Bool) ->+  Sem r ()+subscribeWhile action =+  Events.subscribe @e @token spin+  where+    spin =+      whenM (raise . action =<< Events.consume) spin++-- |Pull repeatedly from the 'Events' channel, passing the event to the supplied callback.+subscribeLoop ::+  ∀ e token r .+  Member (EventConsumer token e) r =>+  (e -> Sem r ()) ->+  Sem r ()+subscribeLoop action =+  Events.subscribe @e @token (forever (raise . action =<< Events.consume))

@@ -4,17 +4,37 @@  import Control.Concurrent.Chan.Unagi.Bounded (InChan, OutChan, dupChan, newChan, readChan, tryWriteChan) import Polysemy (InterpretersFor)+import Polysemy.Async (Async)+import Polysemy.Resource (Resource) +import Polysemy.Conc.Async (withAsync_)+import Polysemy.Conc.Data.Race (Race) import qualified Polysemy.Conc.Effect.Events as Events-import Polysemy.Conc.Effect.Events (EventToken (EventToken), Events, Consume)+import Polysemy.Conc.Effect.Events (Consume, EventToken (EventToken), Events) import Polysemy.Conc.Effect.Scoped (Scoped, runScopedAs) +-- |Convenience alias for the default 'Events' that uses an 'OutChan'.+type ChanEvents e =+  Events (OutChan e) e++-- |Convenience alias for the default 'EventToken' that uses an 'OutChan'.+type EventChan e =+  EventToken (OutChan e)++-- |Convenience alias for the consumer effect.+type EventConsumer token e =+  Scoped (EventToken token) (Consume e)++-- |Convenience alias for the consumer effect using the default implementation.+type ChanConsumer e =+  Scoped (EventChan e) (Consume e)+ -- |Interpret 'Consume' by reading from an 'OutChan'. -- Used internally by 'interpretEventsChan', not safe to use directly. interpretConsumeChan ::   ∀ e r .   Member (Embed IO) r =>-  EventToken (OutChan e) ->+  EventChan e ->   InterpreterFor (Consume e) r interpretConsumeChan (EventToken chan) =   interpret \case@@ -51,8 +71,9 @@ -- duplicate to 'interpretConsumeChan'. interpretEventsChan ::   ∀ e r .-  Member (Embed IO) r =>-  InterpretersFor [Events (OutChan e) e, Scoped (EventToken (OutChan e)) (Consume e)] r+  Members [Resource, Race, Async, Embed IO] r =>+  InterpretersFor [Events (OutChan e) e, ChanConsumer e] r interpretEventsChan sem = do-  (inChan, _) <- embed (newChan @e 64)-  runScopedAs (EventToken <$> embed (dupChan inChan)) interpretConsumeChan (interpretEventsInChan inChan sem)+  (inChan, outChan) <- embed (newChan @e 64)+  withAsync_ (forever (embed (readChan outChan))) do+    runScopedAs (EventToken <$> embed (dupChan inChan)) interpretConsumeChan (interpretEventsInChan inChan sem)

@@ -0,0 +1,113 @@+-- |Description: Pure Queue Interpreters+module Polysemy.Conc.Interpreter.Queue.Pure where++import Polysemy.AtomicState (atomicState')+import Polysemy.State (State, evalState, get, gets, put)++import Polysemy.Conc.AtomicState (interpretAtomic)+import qualified Polysemy.Conc.Data.Queue as Queue+import Polysemy.Conc.Data.Queue (Queue)+import qualified Polysemy.Conc.Data.QueueResult as QueueResult+import Polysemy.Conc.Data.QueueResult (QueueResult)++-- |Reinterpret 'Queue' as 'AtomicState' with a list that cannot be written to.+-- Useful for testing.+interpretQueueListReadOnlyAtomicWith ::+  ∀ d r .+  Member (AtomicState [d]) r =>+  InterpreterFor (Queue d) r+interpretQueueListReadOnlyAtomicWith =+  interpret \case+    Queue.Read ->+      read+    Queue.TryRead ->+      read+    Queue.ReadTimeout _ ->+      read+    Queue.Peek ->+      peek+    Queue.TryPeek ->+      peek+    Queue.Write _ ->+      pass+    Queue.TryWrite _ ->+      pure QueueResult.NotAvailable+    Queue.WriteTimeout _ _ ->+      pure QueueResult.NotAvailable+    Queue.Closed ->+      atomicGets @[d] null+    Queue.Close ->+      atomicPut @[d] []+  where+    read :: Sem r (QueueResult d)+    read =+      atomicState' @[d] \case+        [] -> ([], QueueResult.Closed)+        h : t -> (t, QueueResult.Success h)+    peek :: Sem r (QueueResult d)+    peek =+      atomicGets @[d] \case+        [] -> QueueResult.Closed+        h : _ -> QueueResult.Success h+{-# inline interpretQueueListReadOnlyAtomicWith #-}++-- |Variant of 'interpretQueueListReadOnlyAtomicWith' that interprets the 'AtomicState'.+interpretQueueListReadOnlyAtomic ::+  ∀ d r .+  Member (Embed IO) r =>+  [d] ->+  InterpreterFor (Queue d) r+interpretQueueListReadOnlyAtomic ds sem =+  interpretAtomic ds (interpretQueueListReadOnlyAtomicWith (raiseUnder sem))+{-# inline interpretQueueListReadOnlyAtomic #-}++-- |Reinterpret 'Queue' as 'State' with a list that cannot be written to.+-- Useful for testing.+interpretQueueListReadOnlyStateWith ::+  ∀ d r .+  Member (State [d]) r =>+  InterpreterFor (Queue d) r+interpretQueueListReadOnlyStateWith =+  interpret \case+    Queue.Read ->+      read+    Queue.TryRead ->+      read+    Queue.ReadTimeout _ ->+      read+    Queue.Peek ->+      peek+    Queue.TryPeek ->+      peek+    Queue.Write _ ->+      pass+    Queue.TryWrite _ ->+      pure QueueResult.NotAvailable+    Queue.WriteTimeout _ _ ->+      pure QueueResult.NotAvailable+    Queue.Closed ->+      gets @[d] null+    Queue.Close ->+      put @[d] []+  where+    read :: Sem r (QueueResult d)+    read =+      get @[d] >>= \case+        [] -> pure QueueResult.Closed+        h : t -> QueueResult.Success h <$ put t+    peek :: Sem r (QueueResult d)+    peek =+      gets @[d] \case+        [] -> QueueResult.Closed+        h : _ -> QueueResult.Success h+{-# inline interpretQueueListReadOnlyStateWith #-}++-- |Variant of 'interpretQueueListReadOnlyAtomicWith' that interprets the 'State'.+interpretQueueListReadOnlyState ::+  ∀ d r .+  Member (Embed IO) r =>+  [d] ->+  InterpreterFor (Queue d) r+interpretQueueListReadOnlyState ds sem = do+  evalState ds (interpretQueueListReadOnlyStateWith (raiseUnder sem))+{-# inline interpretQueueListReadOnlyState #-}

@@ -0,0 +1,65 @@+-- |Description: Queue Interpreters for 'TBQueue'+module Polysemy.Conc.Interpreter.Queue.TB where++import Control.Concurrent.STM (+  TBQueue,+  isFullTBQueue,+  newTBQueueIO,+  peekTBQueue,+  readTBQueue,+  tryPeekTBQueue,+  tryReadTBQueue,+  writeTBQueue,+  )++import qualified Polysemy.Conc.Data.Queue as Queue+import Polysemy.Conc.Data.Queue (Queue)+import qualified Polysemy.Conc.Data.QueueResult as QueueResult+import Polysemy.Conc.Data.Race (Race)+import Polysemy.Conc.Queue.Result (naResult)+import Polysemy.Conc.Queue.Timeout (withTimeout)++-- |Interpret 'Queue' with a 'TBQueue'.+--+-- This variant expects an allocated queue as an argument.+interpretQueueTBWith ::+  ∀ d r .+  Members [Race, Embed IO] r =>+  TBQueue d ->+  InterpreterFor (Queue d) r+interpretQueueTBWith queue =+  interpret \case+    Queue.Read ->+      atomically (QueueResult.Success <$> readTBQueue queue)+    Queue.TryRead ->+      atomically (naResult <$> tryReadTBQueue queue)+    Queue.ReadTimeout timeout ->+      withTimeout timeout (Just <$> readTBQueue queue)+    Queue.Peek ->+      atomically (QueueResult.Success <$> peekTBQueue queue)+    Queue.TryPeek ->+      atomically (naResult <$> tryPeekTBQueue queue)+    Queue.Write d ->+      atomically (writeTBQueue queue d)+    Queue.TryWrite d ->+      atomically do+        ifM (isFullTBQueue queue) (pure QueueResult.NotAvailable) (QueueResult.Success <$> writeTBQueue queue d)+    Queue.WriteTimeout timeout d ->+      withTimeout timeout (Just <$> writeTBQueue queue d)+    Queue.Closed ->+      pure False+    Queue.Close ->+      pass+{-# inline interpretQueueTBWith #-}++-- |Interpret 'Queue' with a 'TBQueue'.+interpretQueueTB ::+  ∀ d r .+  Members [Race, Embed IO] r =>+  -- |Buffer size+  Natural ->+  InterpreterFor (Queue d) r+interpretQueueTB maxQueued sem = do+  queue <- embed (newTBQueueIO @d maxQueued)+  interpretQueueTBWith queue sem+{-# inline interpretQueueTB #-}

@@ -0,0 +1,76 @@+-- |Description: Queue Interpreters for 'TBMQueue'+module Polysemy.Conc.Interpreter.Queue.TBM where++import Control.Concurrent.STM.TBMQueue (+  TBMQueue,+  closeTBMQueue,+  isClosedTBMQueue,+  newTBMQueueIO,+  peekTBMQueue,+  readTBMQueue,+  tryPeekTBMQueue,+  tryReadTBMQueue,+  tryWriteTBMQueue,+  writeTBMQueue,+  )+import Polysemy.Resource (Resource, bracket)++import qualified Polysemy.Conc.Data.Queue as Queue+import Polysemy.Conc.Data.Queue (Queue)+import Polysemy.Conc.Data.Race (Race)+import Polysemy.Conc.Queue.Result (closedBoolResult, closedNaResult, closedResult)+import Polysemy.Conc.Queue.Timeout (withTimeout)++-- |Interpret 'Queue' with a 'TBMQueue'.+--+-- This variant expects an allocated queue as an argument.+interpretQueueTBMWith ::+  ∀ d r .+  Members [Race, Embed IO] r =>+  TBMQueue d ->+  InterpreterFor (Queue d) r+interpretQueueTBMWith queue =+  interpret \case+    Queue.Read ->+      atomically (closedResult <$> readTBMQueue queue)+    Queue.TryRead ->+      atomically (closedNaResult <$> tryReadTBMQueue queue)+    Queue.ReadTimeout timeout ->+      withTimeout timeout (readTBMQueue queue)+    Queue.Peek ->+      atomically (closedResult <$> peekTBMQueue queue)+    Queue.TryPeek ->+      atomically (closedNaResult <$> tryPeekTBMQueue queue)+    Queue.Write d ->+      atomically (writeTBMQueue queue d)+    Queue.TryWrite d ->+      atomically (closedBoolResult <$> tryWriteTBMQueue queue d)+    Queue.WriteTimeout timeout d ->+      withTimeout timeout do+        ifM (isClosedTBMQueue queue) (pure Nothing) (Just <$> writeTBMQueue queue d)+    Queue.Closed ->+      atomically (isClosedTBMQueue queue)+    Queue.Close ->+      atomically (closeTBMQueue queue)+{-# inline interpretQueueTBMWith #-}++withTBMQueue ::+  ∀ d r a .+  Members [Resource, Embed IO] r =>+  Int ->+  (TBMQueue d -> Sem r a) ->+  Sem r a+withTBMQueue maxQueued =+  bracket (embed (newTBMQueueIO maxQueued)) (atomically . closeTBMQueue)++-- |Interpret 'Queue' with a 'TBMQueue'.+interpretQueueTBM ::+  ∀ d r .+  Members [Resource, Race, Embed IO] r =>+  -- |Buffer size+  Int ->+  InterpreterFor (Queue d) r+interpretQueueTBM maxQueued sem = do+  withTBMQueue maxQueued \ queue ->+    interpretQueueTBMWith queue sem+{-# inline interpretQueueTBM #-}

@@ -5,9 +5,11 @@  import qualified Polysemy.Conc.Data.Race as Race import Polysemy.Conc.Data.Race (Race)+import Polysemy.Conc.Effect.Scoped (Scoped, runScopedAs) import qualified Polysemy.Conc.Effect.Sync as Sync-import Polysemy.Conc.Effect.Sync (Sync)+import Polysemy.Conc.Effect.Sync (Sync, SyncResources (SyncResources), unSyncResources) import qualified Polysemy.Conc.Race as Race+import Polysemy.Resource (Resource)  -- |Interpret 'Sync' with the provided 'MVar'. interpretSyncWith ::@@ -29,6 +31,12 @@       rightToMaybe <$> Race.timeout () interval (takeMVar var)     Sync.TakeTry ->       tryTakeMVar var+    Sync.ReadBlock ->+      readMVar var+    Sync.ReadWait interval ->+      rightToMaybe <$> Race.timeout () interval (readMVar var)+    Sync.ReadTry ->+      tryReadMVar var     Sync.PutBlock d ->       putMVar var d     Sync.PutWait interval d ->@@ -56,3 +64,20 @@ interpretSyncAs d sem = do   var <- newMVar d   interpretSyncWith var sem++-- |Interpret 'Sync' for locally scoped use with an empty 'MVar'.+interpretScopedSync ::+  ∀ d r .+  Members [Resource, Race, Embed IO] r =>+  InterpreterFor (Scoped (SyncResources (MVar d)) (Sync d)) r+interpretScopedSync =+  runScopedAs (SyncResources <$> newEmptyMVar) \ r -> interpretSyncWith (unSyncResources r)++-- |Interpret 'Sync' for locally scoped use with an 'MVar' containing the specified value.+interpretScopedSyncAs ::+  ∀ d r .+  Members [Resource, Race, Embed IO] r =>+  d ->+  InterpreterFor (Scoped (SyncResources (MVar d)) (Sync d)) r+interpretScopedSyncAs d =+  runScopedAs (SyncResources <$> newMVar d) \ r -> interpretSyncWith (unSyncResources r)

@@ -9,6 +9,7 @@ import qualified Data.Text.IO as Text import Polysemy (getInspectorT, inspect, interpretH, 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), SignalInfo, installHandler, keyboardSignal)@@ -145,7 +146,7 @@       handle <- raise (interpretInterruptState (async maT))       result <- liftT (awaitOrKill desc handle)       pure (join . fmap (inspect ins) <$> result)-{-# INLINE interpretInterruptState #-}+{-# inline interpretInterruptState #-}  broadcastInterrupt ::   Members [AtomicState InterruptState, Embed IO] r =>@@ -168,7 +169,7 @@   thunk originalHandler _ =   const pass-{-# INLINE originalHandler #-}+{-# inline originalHandler #-}  installSignalHandler ::   TVar InterruptState ->@@ -191,4 +192,4 @@   runAtomicStateTVar state do     atomicModify' \ s -> s {original = originalHandler orig}     interpretInterruptState $ raiseUnder sem-{-# INLINE interpretInterrupt #-}+{-# inline interpretInterrupt #-}

@@ -36,7 +36,7 @@   reinterpret,   runFinal,   )-import Polysemy.AtomicState (AtomicState, atomicGet, atomicGets, atomicModify', atomicPut, runAtomicStateTVar)+import Polysemy.AtomicState (AtomicState, atomicGet, atomicGets, atomicModify', atomicPut) import Relude hiding (   Reader,   State,@@ -64,9 +64,9 @@ qt :: QuasiQuoter qt =   Interpolate.i-{-# INLINE qt #-}+{-# inline qt #-}  unify :: Either a a -> a unify =   either id id-{-# INLINE unify #-}+{-# inline unify #-}

@@ -1,113 +1,34 @@--- |Description: Pure Queue interpreters+-- |Description: Queue Combinators module Polysemy.Conc.Queue where -import Polysemy.AtomicState (atomicState')-import Polysemy.State (State, evalState, get, gets, put)- import qualified Polysemy.Conc.Data.Queue as Queue import Polysemy.Conc.Data.Queue (Queue) import qualified Polysemy.Conc.Data.QueueResult as QueueResult-import Polysemy.Conc.Data.QueueResult (QueueResult) --- |Reinterpret 'Queue' as 'AtomicState' with a list that cannot be written to.--- Useful for testing.-interpretQueueListReadOnlyAtomicWith ::-  ∀ d r .-  Member (AtomicState [d]) r =>-  InterpreterFor (Queue d) r-interpretQueueListReadOnlyAtomicWith =-  interpret \case-    Queue.Read ->-      read-    Queue.TryRead ->-      read-    Queue.ReadTimeout _ ->-      read-    Queue.Peek ->-      peek-    Queue.TryPeek ->-      peek-    Queue.Write _ ->-      pass-    Queue.TryWrite _ ->-      pure QueueResult.NotAvailable-    Queue.WriteTimeout _ _ ->-      pure QueueResult.NotAvailable-    Queue.Closed ->-      atomicGets @[d] null-    Queue.Close ->-      atomicPut @[d] []-  where-    read :: Sem r (QueueResult d)-    read =-      atomicState' @[d] \case-        [] -> ([], QueueResult.Closed)-        h : t -> (t, QueueResult.Success h)-    peek :: Sem r (QueueResult d)-    peek =-      atomicGets @[d] \case-        [] -> QueueResult.Closed-        h : _ -> QueueResult.Success h-{-# INLINE interpretQueueListReadOnlyAtomicWith #-}---- |Variant of 'interpretQueueListReadOnlyAtomicWith' that interprets the 'AtomicState'.-interpretQueueListReadOnlyAtomic ::-  ∀ d r .-  Member (Embed IO) r =>-  [d] ->-  InterpreterFor (Queue d) r-interpretQueueListReadOnlyAtomic ds sem = do-  tv <- newTVarIO ds-  runAtomicStateTVar tv (interpretQueueListReadOnlyAtomicWith (raiseUnder sem))-{-# INLINE interpretQueueListReadOnlyAtomic #-}---- |Reinterpret 'Queue' as 'State' with a list that cannot be written to.--- Useful for testing.-interpretQueueListReadOnlyStateWith ::-  ∀ d r .-  Member (State [d]) r =>-  InterpreterFor (Queue d) r-interpretQueueListReadOnlyStateWith =-  interpret \case-    Queue.Read ->-      read-    Queue.TryRead ->-      read-    Queue.ReadTimeout _ ->-      read-    Queue.Peek ->-      peek-    Queue.TryPeek ->-      peek-    Queue.Write _ ->-      pass-    Queue.TryWrite _ ->-      pure QueueResult.NotAvailable-    Queue.WriteTimeout _ _ ->-      pure QueueResult.NotAvailable-    Queue.Closed ->-      gets @[d] null-    Queue.Close ->-      put @[d] []+-- |Read from a 'Queue' repeatedly until it is closed.+--+-- When an element is received, call @action@ and recurse if it returns 'True'.+-- When no element is available, evaluate @na@ and recurse if it returns 'True'.+loopOr ::+  Member (Queue d) r =>+  Sem r Bool ->+  (d -> Sem r Bool) ->+  Sem r ()+loopOr na action =+  spin   where-    read :: Sem r (QueueResult d)-    read =-      get @[d] >>= \case-        [] -> pure QueueResult.Closed-        h : t -> QueueResult.Success h <$ put t-    peek :: Sem r (QueueResult d)-    peek =-      gets @[d] \case-        [] -> QueueResult.Closed-        h : _ -> QueueResult.Success h-{-# INLINE interpretQueueListReadOnlyStateWith #-}+    spin =+      Queue.read >>= \case+        QueueResult.Success d -> whenM (action d) spin+        QueueResult.NotAvailable -> whenM na spin+        QueueResult.Closed -> pass --- |Variant of 'interpretQueueListReadOnlyAtomicWith' that interprets the 'State'.-interpretQueueListReadOnlyState ::-  ∀ d r .-  Member (Embed IO) r =>-  [d] ->-  InterpreterFor (Queue d) r-interpretQueueListReadOnlyState ds sem = do-  evalState ds (interpretQueueListReadOnlyStateWith (raiseUnder sem))-{-# INLINE interpretQueueListReadOnlyState #-}+-- |Read from a 'Queue' repeatedly until it is closed.+--+-- When an element is received, call @action@ and recurse.+loop ::+  Member (Queue d) r =>+  (d -> Sem r ()) ->+  Sem r ()+loop action =+  loopOr (pure True) \ d -> True <$ action d

@@ -10,7 +10,7 @@ closedResult = \case   Nothing -> QueueResult.Closed   Just d -> QueueResult.Success d-{-# INLINE closedResult #-}+{-# inline closedResult #-}  naResult ::   Maybe d ->@@ -18,7 +18,7 @@ naResult = \case   Nothing -> QueueResult.NotAvailable   Just d -> QueueResult.Success d-{-# INLINE naResult #-}+{-# inline naResult #-}  closedNaResult ::   Maybe (Maybe d) ->@@ -27,7 +27,7 @@   Nothing -> QueueResult.Closed   Just Nothing -> QueueResult.NotAvailable   Just (Just d) -> QueueResult.Success d-{-# INLINE closedNaResult #-}+{-# inline closedNaResult #-}  closedBoolResult ::   Maybe Bool ->@@ -36,7 +36,7 @@   Nothing -> QueueResult.Closed   Just False -> QueueResult.NotAvailable   Just True -> QueueResult.Success ()-{-# INLINE closedBoolResult #-}+{-# inline closedBoolResult #-}  -- |Turn a 'Success' into 'Just'. resultToMaybe :: QueueResult d -> Maybe d@@ -44,4 +44,4 @@   QueueResult.Success d -> Just d   QueueResult.NotAvailable -> Nothing   QueueResult.Closed -> Nothing-{-# INLINE resultToMaybe #-}+{-# inline resultToMaybe #-}

@@ -1,65 +0,0 @@--- |Description: Queue interpreters for 'TBQueue'-module Polysemy.Conc.Queue.TB where--import Control.Concurrent.STM (-  TBQueue,-  isFullTBQueue,-  newTBQueueIO,-  peekTBQueue,-  readTBQueue,-  tryPeekTBQueue,-  tryReadTBQueue,-  writeTBQueue,-  )--import qualified Polysemy.Conc.Data.Queue as Queue-import Polysemy.Conc.Data.Queue (Queue)-import qualified Polysemy.Conc.Data.QueueResult as QueueResult-import Polysemy.Conc.Data.Race (Race)-import Polysemy.Conc.Queue.Result (naResult)-import Polysemy.Conc.Queue.Timeout (withTimeout)---- |Interpret 'Queue' with a 'TBQueue'.------ This variant expects an allocated queue as an argument.-interpretQueueTBWith ::-  ∀ d r .-  Members [Race, Embed IO] r =>-  TBQueue d ->-  InterpreterFor (Queue d) r-interpretQueueTBWith queue =-  interpret \case-    Queue.Read ->-      atomically (QueueResult.Success <$> readTBQueue queue)-    Queue.TryRead ->-      atomically (naResult <$> tryReadTBQueue queue)-    Queue.ReadTimeout timeout ->-      withTimeout timeout (Just <$> readTBQueue queue)-    Queue.Peek ->-      atomically (QueueResult.Success <$> peekTBQueue queue)-    Queue.TryPeek ->-      atomically (naResult <$> tryPeekTBQueue queue)-    Queue.Write d ->-      atomically (writeTBQueue queue d)-    Queue.TryWrite d ->-      atomically do-        ifM (isFullTBQueue queue) (pure QueueResult.NotAvailable) (QueueResult.Success <$> writeTBQueue queue d)-    Queue.WriteTimeout timeout d ->-      withTimeout timeout (Just <$> writeTBQueue queue d)-    Queue.Closed ->-      pure False-    Queue.Close ->-      pass-{-# INLINE interpretQueueTBWith #-}---- |Interpret 'Queue' with a 'TBQueue'.-interpretQueueTB ::-  ∀ d r .-  Members [Race, Embed IO] r =>-  -- |Buffer size-  Natural ->-  InterpreterFor (Queue d) r-interpretQueueTB maxQueued sem = do-  queue <- embed (newTBQueueIO @d maxQueued)-  interpretQueueTBWith queue sem-{-# INLINE interpretQueueTB #-}

@@ -1,67 +0,0 @@--- |Description: Queue interpreters for 'TBMQueue'-module Polysemy.Conc.Queue.TBM where--import Control.Concurrent.STM.TBMQueue (-  TBMQueue,-  closeTBMQueue,-  isClosedTBMQueue,-  newTBMQueueIO,-  peekTBMQueue,-  readTBMQueue,-  tryPeekTBMQueue,-  tryReadTBMQueue,-  tryWriteTBMQueue,-  writeTBMQueue,-  )--import qualified Polysemy.Conc.Data.Queue as Queue-import Polysemy.Conc.Data.Queue (Queue)-import Polysemy.Conc.Data.Race (Race)-import Polysemy.Conc.Queue.Result (closedBoolResult, closedNaResult, closedResult)-import Polysemy.Conc.Queue.Timeout (withTimeout)-import Polysemy.Resource (bracket, Resource)---- |Interpret 'Queue' with a 'TBMQueue'.------ This variant expects an allocated queue as an argument.-interpretQueueTBMWith ::-  ∀ d r .-  Members [Race, Embed IO] r =>-  TBMQueue d ->-  InterpreterFor (Queue d) r-interpretQueueTBMWith queue =-  interpret \case-    Queue.Read ->-      atomically (closedResult <$> readTBMQueue queue)-    Queue.TryRead ->-      atomically (closedNaResult <$> tryReadTBMQueue queue)-    Queue.ReadTimeout timeout ->-      withTimeout timeout (readTBMQueue queue)-    Queue.Peek ->-      atomically (closedResult <$> peekTBMQueue queue)-    Queue.TryPeek ->-      atomically (closedNaResult <$> tryPeekTBMQueue queue)-    Queue.Write d ->-      atomically (writeTBMQueue queue d)-    Queue.TryWrite d ->-      atomically (closedBoolResult <$> tryWriteTBMQueue queue d)-    Queue.WriteTimeout timeout d ->-      withTimeout timeout do-        ifM (isClosedTBMQueue queue) (pure Nothing) (Just <$> writeTBMQueue queue d)-    Queue.Closed ->-      atomically (isClosedTBMQueue queue)-    Queue.Close ->-      atomically (closeTBMQueue queue)-{-# INLINE interpretQueueTBMWith #-}---- |Interpret 'Queue' with a 'TBMQueue'.-interpretQueueTBM ::-  ∀ d r .-  Members [Resource, Race, Embed IO] r =>-  -- |Buffer size-  Int ->-  InterpreterFor (Queue d) r-interpretQueueTBM maxQueued sem = do-  bracket (embed (newTBMQueueIO @d maxQueued)) (atomically . closeTBMQueue) \ queue ->-    interpretQueueTBMWith queue sem-{-# INLINE interpretQueueTBM #-}

@@ -5,7 +5,7 @@ import qualified Control.Concurrent.Async as Async import Polysemy.Final (getInitialStateS, interpretFinal, runS) import qualified Polysemy.Time as Time-import Polysemy.Time (MicroSeconds(MicroSeconds), TimeUnit)+import Polysemy.Time (MicroSeconds (MicroSeconds), TimeUnit) import qualified System.Timeout as System  import qualified Polysemy.Conc.Data.Race as Race@@ -17,7 +17,7 @@   f (Either a b) biseqEither =   either (fmap Left) (fmap Right)-{-# INLINE biseqEither #-}+{-# inline biseqEither #-}  -- |Interpret 'Race' in terms of 'Async.race' and 'System.timeout'. -- Since this has to pass higher-order thunks as 'IO' arguments, it is interpreted in terms of 'Final IO'.@@ -32,7 +32,7 @@       mbT <- runS mb       s <- getInitialStateS       pure (maybe (Left err <$ s) (fmap Right) <$> System.timeout (fromIntegral timeout) mbT)-{-# INLINE interpretRace #-}+{-# inline interpretRace #-}  -- |Specialization of 'Race.race' for the case where both thunks return the same type, obviating the need for 'Either'. race_ ::@@ -40,9 +40,9 @@   Sem r a ->   Sem r a ->   Sem r a-race_ ma mb =-  unify <$> Race.race ma mb-{-# INLINE race_ #-}+race_ ml mr =+  unify <$> Race.race ml mr+{-# inline race_ #-}  -- |Specialization of 'Race.timeout' for the case where the thunk return the same type as the fallback, obviating the -- need for 'Either'.@@ -53,6 +53,17 @@   u ->   Sem r a ->   Sem r a-timeout_ err interval mb =-  unify <$> Race.timeout err interval mb-{-# INLINE timeout_ #-}+timeout_ err interval ma =+  unify <$> Race.timeout err interval ma+{-# inline timeout_ #-}++-- |Specialization of 'Race.timeout' for unit actions.+timeoutU ::+  TimeUnit u =>+  Member Race r =>+  u ->+  Sem r () ->+  Sem r ()+timeoutU =+  timeout_ ()+{-# inline timeoutU #-}

@@ -0,0 +1,63 @@+-- |Description: Action Retrying+module Polysemy.Conc.Retry where++import qualified Polysemy.Time as Time+import Polysemy.Time (Time, TimeUnit)++import Polysemy.Conc.Data.Race (Race)+import qualified Polysemy.Conc.Effect.Sync as Sync+import Polysemy.Conc.Interpreter.Sync (interpretSync)+import qualified Polysemy.Conc.Race as Race++-- |Run an action repeatedly until it returns 'Right' or the timout has been exceeded.+retrying ::+  ∀ e w u t d r a .+  TimeUnit w =>+  TimeUnit u =>+  Members [Race, Time t d] r =>+  -- |The timeout after which the attempt is abandoned.+  w ->+  -- |The waiting interval between two tries.+  u ->+  Sem r (Either e a) ->+  Sem r (Maybe a)+retrying timeout interval action =+  Race.timeout_ Nothing timeout (Just <$> spin)+  where+    spin =+      action >>= \case+        Right a ->+          pure a+        Left _ -> do+          Time.sleep @t @d interval+          spin++-- |Run an action repeatedly until it returns 'Right' or the timout has been exceeded.+--+-- If the action failed at least once, the last error will be returned in case of timeout.+retryingWithError ::+  ∀ e w u t d r a .+  TimeUnit w =>+  TimeUnit u =>+  Members [Race, Time t d, Embed IO] r =>+  -- |The timeout after which the attempt is abandoned.+  w ->+  -- |The waiting interval between two tries.+  u ->+  Sem r (Either e a) ->+  Sem r (Maybe (Either e a))+retryingWithError timeout interval action =+  interpretSync @e do+    Race.timeout_ Nothing timeout (Just <$> spin) >>= \case+      Just a -> pure (Just (Right a))+      Nothing -> fmap Left <$> Sync.takeTry+  where+    spin =+      raise action >>= \case+        Right a ->+          pure a+        Left e -> do+          void (Sync.takeTry @e)+          Sync.putTry e+          Time.sleep @t @d interval+          spin

@@ -4,9 +4,11 @@ import qualified Polysemy.Time as Time import Polysemy.Time (Time, TimeUnit) +import Polysemy.Conc.Effect.Scoped (Scoped, scoped) import qualified Polysemy.Conc.Effect.Sync as Sync-import Polysemy.Conc.Effect.Sync (Sync)+import Polysemy.Conc.Effect.Sync (Sync, SyncResources) +-- |Run an action repeatedly until the 'Sync' variable is available. whileEmpty ::   ∀ a r .   Member (Sync a) r =>@@ -17,8 +19,9 @@   where     spin = do       action-      whenM (isJust <$> Sync.try @a) spin+      whenM (not <$> Sync.empty @a) spin +-- |Run an action repeatedly until the 'Sync' variable is available, waiting for the specified time between executions. whileEmptyInterval ::   ∀ a u t d r .   TimeUnit u =>@@ -31,4 +34,12 @@   where     spin = do       action-      whenM (isJust <$> Sync.try @a) (Time.sleep @t @d interval *> spin)+      whenM (not <$> Sync.empty @a) (Time.sleep @t @d interval *> spin)++-- |Run an action with a locally scoped 'Sync' variable.+withSync ::+  ∀ d res r .+  Member (Scoped (SyncResources res) (Sync d)) r =>+  InterpreterFor (Sync d) r+withSync =+  scoped @(SyncResources res)

@@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack  name:           polysemy-conc-version:        0.1.1.0+version:        0.2.0.0 synopsis:       Polysemy Effects for Concurrency description:    See <https://hackage.haskell.org/package/polysemy-conc/docs/Polysemy-Conc.html> category:       Concurrency@@ -19,7 +19,7 @@ build-type:     Simple extra-source-files:     readme.md-    Changelog.md+    changelog.md  source-repository head   type: git@@ -28,6 +28,8 @@ library   exposed-modules:       Polysemy.Conc+      Polysemy.Conc.Async+      Polysemy.Conc.AtomicState       Polysemy.Conc.Critical       Polysemy.Conc.Data.Critical       Polysemy.Conc.Data.Interrupt@@ -37,16 +39,19 @@       Polysemy.Conc.Effect.Events       Polysemy.Conc.Effect.Scoped       Polysemy.Conc.Effect.Sync+      Polysemy.Conc.Events       Polysemy.Conc.Interpreter.Events+      Polysemy.Conc.Interpreter.Queue.Pure+      Polysemy.Conc.Interpreter.Queue.TB+      Polysemy.Conc.Interpreter.Queue.TBM       Polysemy.Conc.Interpreter.Sync       Polysemy.Conc.Interrupt       Polysemy.Conc.Prelude       Polysemy.Conc.Queue       Polysemy.Conc.Queue.Result-      Polysemy.Conc.Queue.TB-      Polysemy.Conc.Queue.TBM       Polysemy.Conc.Queue.Timeout       Polysemy.Conc.Race+      Polysemy.Conc.Retry       Polysemy.Conc.Sync   other-modules:       Prelude@@ -204,23 +209,13 @@       ViewPatterns   ghc-options: -flate-specialise -fspecialise-aggressively -Wall -Wredundant-constraints -Wunused-packages -threaded -rtsopts -with-rtsopts=-N   build-depends:-      async-    , base ==4.*-    , containers-    , hedgehog+      base ==4.*     , polysemy >=1.5     , polysemy-conc     , polysemy-test     , polysemy-time >=0.1.2.1-    , relude >=0.7     , stm-    , stm-chans >=2-    , string-interpolate >=0.2     , tasty-    , tasty-hedgehog-    , template-haskell-    , text-    , time     , unagi-chan >=0.4     , unix   mixins:

@@ -1,19 +1,19 @@ module Polysemy.Conc.Test.QueueTest where -import Polysemy.Async (asyncToIOFinal, Async)+import Polysemy.Async (Async, asyncToIOFinal) import Polysemy.Test (UnitTest, assertEq, assertJust, runTestAuto)-import Polysemy.Time (MilliSeconds(MilliSeconds), interpretTimeGhc)+import Polysemy.Time (MilliSeconds (MilliSeconds), interpretTimeGhc) import Polysemy.Time.Ghc (GhcTime)  import qualified Polysemy.Conc.Data.Queue as Queue import Polysemy.Conc.Data.Queue (Queue) import qualified Polysemy.Conc.Data.QueueResult as QueueResult+import Polysemy.Conc.Data.QueueResult (QueueResult) import Polysemy.Conc.Data.Race (Race)+import Polysemy.Conc.Interpreter.Queue.TB (interpretQueueTB)+import Polysemy.Conc.Interpreter.Queue.TBM (interpretQueueTBM) import Polysemy.Conc.Queue.Result (resultToMaybe)-import Polysemy.Conc.Queue.TB (interpretQueueTB)-import Polysemy.Conc.Queue.TBM (interpretQueueTBM) import Polysemy.Conc.Race (interpretRace)-import Polysemy.Conc.Data.QueueResult (QueueResult)  progSuccess ::   Member (Queue Int) r =>