packages feed

polysemy-conc (empty) → 0.1.0.0

raw patch · 25 files changed

+1397/−0 lines, 25 filesdep +asyncdep +basedep +containers

Dependencies added: async, base, containers, hedgehog, polysemy, polysemy-conc, polysemy-test, polysemy-time, relude, stm, stm-chans, string-interpolate, tasty, tasty-hedgehog, template-haskell, text, time, unix

Files

+ Changelog.md view
+ LICENSE view
@@ -0,0 +1,34 @@+Copyright (c) 2020 Torsten Schmits++Redistribution and use in source and binary forms, with or without modification, are permitted provided that the+following conditions are met:++  1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following+  disclaimer.+  2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following+  disclaimer in the documentation and/or other materials provided with the distribution.++Subject to the terms and conditions of this license, each copyright holder and contributor hereby grants to those+receiving rights under this license a perpetual, worldwide, non-exclusive, no-charge, royalty-free, irrevocable (except+for failure to satisfy the conditions of this license) patent license to make, have made, use, offer to sell, sell,+import, and otherwise transfer this software, where such license applies only to those patent claims, already acquired+or hereafter acquired, licensable by such copyright holder or contributor that are necessarily infringed by:++  (a) their Contribution(s) (the licensed copyrights of copyright holders and non-copyrightable additions of+  contributors, in source or binary form) alone; or+  (b) combination of their Contribution(s) with the work of authorship to which such Contribution(s) was added by such+  copyright holder or contributor, if, at the time the Contribution is added, such addition causes such combination to+  be necessarily infringed. The patent license shall not apply to any other combinations which include the Contribution.++Except as expressly stated above, no rights or licenses from any copyright holder or contributor is granted under this+license, whether expressly, by implication, estoppel or otherwise.++DISCLAIMER++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES,+INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE+DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDERS OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR+SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,+WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF+THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ lib/Polysemy/Conc.hs view
@@ -0,0 +1,81 @@+-- |Description: Polysemy Effects for Concurrency+module Polysemy.Conc (+  -- * Introduction+  -- $intro++  -- * Queues+  -- $queue+  Queue,+  QueueResult,++  -- ** Interpreters+  interpretQueueTBM,+  interpretQueueTB,++  -- * MVars+  -- $mvar+  Sync,++  -- ** Interpreters+  interpretSync,++  -- * Racing+  -- $race+  Race,+  race,+  race_,+  timeout,+  timeout_,++  -- ** Interpreters+  interpretRace,++  -- * Signal Handling+  -- $signal+  Interrupt,++  -- ** Interpreters+  interpretInterrupt,+) where++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.Data.Sync (Sync)+import Polysemy.Conc.Interrupt (interpretInterrupt)+import Polysemy.Conc.Queue.TB (interpretQueueTB)+import Polysemy.Conc.Queue.TBM (interpretQueueTBM)+import Polysemy.Conc.Race (interpretRace, race_, timeout_)+import Polysemy.Conc.Sync (interpretSync)++-- $intro+-- This library provides an assortment of tools for concurrency-related tasks:+--+-- - [STM queues](#queue)+-- - [MVars](#mvar)+-- - [Racing](#race)+-- - [Signal handling](#signal)++-- $queue+-- #queue#++-- $mvar+-- #mvar#+-- An 'MVar' is abstracted as 'Sync' since it can be used to synchronize threads.++-- $race+-- #race#+-- Racing works like this:+--+-- @+-- prog =+--  Polysemy.Conc.race (httpRequest "hackage.haskell.org") (readFile "\/path\/to\/file") >>= \\case+--    Left _ -> putStrLn "hackage was faster"+--    Right _ -> putStrLn "file was faster"+-- @+--+-- When the first thunk finishes, the other will be killed.++-- $signal+-- #signal#
+ lib/Polysemy/Conc/Critical.hs view
@@ -0,0 +1,32 @@+-- |Description: Critical interpreters+module Polysemy.Conc.Critical where++import qualified Control.Exception as Exception+import Polysemy.Final (getInitialStateS, interpretFinal, runS)++import Polysemy.Conc.Data.Critical (Critical(..))+import Polysemy (interpretH, runT)++-- |Interpret 'Critical' in terms of 'Final' 'IO'.+interpretCritical ::+  Member (Final IO) r =>+  InterpreterFor Critical r+interpretCritical =+  interpretFinal @IO \case+    Catch ma -> do+      s <- getInitialStateS+      o <- runS ma+      pure (run o s)+      where+        run ma' s =+          Exception.catch (fmap Right <$> ma') \ se -> pure (Left se <$ s)+{-# INLINE interpretCritical #-}++-- |Interpret 'Critical' by doing nothing.+interpretCriticalNull ::+  InterpreterFor Critical r+interpretCriticalNull =+  interpretH \case+    Catch ma ->+      fmap (fmap Right) . raise . interpretCriticalNull =<< runT ma+{-# INLINE interpretCriticalNull #-}
+ lib/Polysemy/Conc/Data/Critical.hs view
@@ -0,0 +1,41 @@+{-# options_haddock prune #-}+-- |Description: Critical effect+module Polysemy.Conc.Data.Critical where++-- |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.+data Critical :: Effect where+  -- |Catch all exceptions of type @e@ in this computation.+  Catch :: Exception e => m a -> Critical m (Either e a)++makeSem ''Critical++-- |Catch exceptions of type @e@ and return a fallback value.+catchAs ::+  ∀ e a r .+  Exception e =>+  Member Critical r =>+  a ->+  Sem r a ->+  Sem r a+catchAs a =+  fmap (fromRight a) . catch @_ @e++-- |Convenience overload for 'SomeException'.+run ::+  Member Critical r =>+  Sem r a ->+  Sem r (Either SomeException a)+run =+  catch++-- |Convenience overload for 'SomeException'.+runAs ::+  Member Critical r =>+  a ->+  Sem r a ->+  Sem r a+runAs a =+  fmap (fromRight a) . run
+ lib/Polysemy/Conc/Data/Interrupt.hs view
@@ -0,0 +1,44 @@+{-# options_haddock prune #-}+-- |Description: Interrupt effect+module Polysemy.Conc.Data.Interrupt where++-- |The interrupt handler effect allows three kinds of interaction for interrupt signals:+--+-- - Execute a callback when a signal is received+-- - Block a thread until a signal is received+-- - Kill a thread when a signal is received+--+-- For documentation on the constructors, see the module "Polysemy.Conc.Data.Interrupt".+--+-- @+-- import qualified Polysemy.Conc.Data.Interrupt as Interrupt+--+-- prog = do+--   Interrupt.register "task 1" (putStrLn "interrupted")+--   Interrupt.killOnQuit $ forever do+--    doSomeWork+-- @+data Interrupt :: Effect where+  -- |Add a computation to be executed on interrupt, using the first argument as a key.+  Register :: Text -> IO () -> Interrupt m ()+  -- |Remove the previously added handler with the given key.+  Unregister :: Text -> Interrupt m ()+  -- |Manually trigger the interrupt.+  Quit :: Interrupt m ()+  -- |Block until an interrupt is triggered.+  WaitQuit :: Interrupt m ()+  -- |Indicate whether an interrupt was triggered.+  Interrupted :: Interrupt m Bool+  -- |Execute a computation, waiting for it to finish, killing its thread on interrupt.+  KillOnQuit :: Text -> m a -> Interrupt m (Maybe a)++makeSem ''Interrupt++-- |Variant of 'killOnQuit' that returns @()@.+killOnQuit_ ::+  Member Interrupt r =>+  Text ->+  Sem r a ->+  Sem r ()+killOnQuit_ desc ma =+  void (killOnQuit desc ma)
+ lib/Polysemy/Conc/Data/Queue.hs view
@@ -0,0 +1,46 @@+{-# options_haddock prune #-}+-- |Description: Queue effect+module Polysemy.Conc.Data.Queue where++import Polysemy.Time (TimeUnit)+import Polysemy.Conc.Data.QueueResult (QueueResult)++-- |Abstracts queues like 'Control.Concurrent.STM.TBQueue'.+--+-- For documentation on the constructors, see the module "Polysemy.Conc.Data.Queue".+--+-- @+-- import Polysemy.Conc (Queue, QueueResult)+-- import Polysemy.Conc.Data.Queue as Queue+--+-- prog :: Member (Queue Int) r => Sem r (QueueResult Int)+-- prog = do+--   Queue.write 5+--   Queue.write 10+--   Queue.read >>= \\case+--     QueueResult.Success i -> fmap (i +) \<$> Queue.read+--     r -> pure r+-- @+data Queue d :: Effect where+  -- |Read an element from the queue, blocking until one is available.+  Read :: Queue d m (QueueResult d)+  -- |Read an element from the queue, immediately returning if none is available.+  TryRead :: Queue d m (QueueResult d)+  -- |Read an element from the queue, blocking until one is available or the timeout expires.+  ReadTimeout :: TimeUnit t => t -> Queue d m (QueueResult d)+  -- |Read an element, leaving it in the queue, blocking until one is available.+  Peek :: Queue d m (QueueResult d)+  -- |Read an element, leaving it in the queue, immediately returning if none is available.+  TryPeek :: Queue d m (QueueResult d)+  -- |Write an element to the queue, blocking until a slot is available.+  Write :: d -> Queue d m ()+  -- |Write an element to the queue, immediately returning if no slot is available.+  TryWrite :: d -> Queue d m (QueueResult ())+  -- |Write an element to the queue, blocking until a slot is available or the timeout expires.+  WriteTimeout :: TimeUnit t => t -> d -> Queue d m (QueueResult ())+  -- |Indicate whether the queue is closed.+  Closed :: Queue d m Bool+  -- |Close the queue.+  Close :: Queue d m ()++makeSem ''Queue
+ lib/Polysemy/Conc/Data/QueueResult.hs view
@@ -0,0 +1,30 @@+-- |Description: Queue result ADT+module Polysemy.Conc.Data.QueueResult where++-- |Encodes failure reasons for queues.+--+-- For documentation on the constructors, see the module "Polysemy.Conc.Data.QueueResult".+--+-- @+-- import qualified Polysemy.Conc.Data.QueueResult as QueueResult+-- @+data QueueResult d =+  -- |The operation was successful.+  Success d+  |+  -- |The queue is either full and cannot be added to, or empty and cannot be read from.+  NotAvailable+  |+  -- |The queue was closed by the user.+  Closed+  deriving (Eq, Show, Ord, Functor, Generic)++instance Semigroup d => Semigroup (QueueResult d) where+  Success d1 <> Success d2 = Success (d1 <> d2)+  Closed <> _ = Closed+  _ <> Closed = Closed+  NotAvailable <> _ = NotAvailable+  _ <> NotAvailable = NotAvailable++instance Monoid d => Monoid (QueueResult d) where+  mempty = Success mempty
+ lib/Polysemy/Conc/Data/Race.hs view
@@ -0,0 +1,35 @@+{-# options_haddock prune #-}+-- |Description: Race effect+module Polysemy.Conc.Data.Race where++import Polysemy (makeSem_)++import Polysemy.Time (TimeUnit)++-- |Abstract the concept of running two programs concurrently, aborting the other when one terminates.+-- 'Timeout' is a simpler variant, where one thread just sleeps for a given interval.+data Race :: Effect where+  -- |Run both programs concurrently, returning the result of the faster one.+  Race :: m a -> m b -> Race m (Either a b)+  -- |Return the fallback value if the given program doesn't finish within the specified interval.+  Timeout :: TimeUnit u => a -> u -> m b -> Race m (Either a b)++makeSem_ ''Race++-- |Run both programs concurrently, returning the result of the faster one.+race ::+  ∀ a b r .+  Member Race r =>+  Sem r a ->+  Sem r b ->+  Sem r (Either a b)++-- |Return the fallback value if the given program doesn't finish within the specified interval.+timeout ::+  ∀ a b u r .+  TimeUnit u =>+  Member Race r =>+  a ->+  u ->+  Sem r b ->+  Sem r (Either a b)
+ lib/Polysemy/Conc/Data/Sync.hs view
@@ -0,0 +1,42 @@+{-# options_haddock prune #-}+-- |Description: Sync effect+module Polysemy.Conc.Data.Sync where++import Polysemy.Time (TimeUnit)++-- |Abstracts an 'MVar'.+--+-- For documentation on the constructors, see the module "Polysemy.Conc.Data.Sync".+--+-- @+-- import Polysemy.Conc (Sync)+-- import qualified Polysemy.Conc.Data.Sync as Sync+--+-- prog :: Member (Sync Int) r => Sem r Int+-- prog = do+--   Sync.putTry 5+--   Sync.takeBlock+-- @+data Sync d :: Effect where+  -- |Read the variable, waiting until a value is available.+  Block :: Sync d m d+  -- |Read the variable, waiting until a value is available or the timeout has expired.+  Wait :: TimeUnit u => u -> Sync d m (Maybe d)+  -- |Read the variable, returning 'Nothing' immmediately if no value was available.+  Try :: Sync d m (Maybe d)+  -- |Take the variable, waiting until a value is available.+  TakeBlock :: Sync d m d+  -- |Take the variable, waiting until a value is available or the timeout has expired.+  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)+  -- |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.+  PutWait :: TimeUnit u => u -> d -> Sync d m Bool+  -- |Write the variable, returning 'False' immmediately if no value was available.+  PutTry :: d -> Sync d m Bool+  -- |Indicate whether the variable is empty.+  Empty :: Sync d m Bool++makeSem ''Sync
+ lib/Polysemy/Conc/Interrupt.hs view
@@ -0,0 +1,194 @@+{-# options_haddock prune #-}+-- |Description: Interrupt interpreters+module Polysemy.Conc.Interrupt where++import qualified Control.Concurrent.Async as A+import Control.Concurrent.Async (AsyncCancelled)+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.Async (Async, async, await, cancel)+import Polysemy.Internal.Tactics (liftT)+import Polysemy.Time (Seconds(Seconds))+import System.Posix.Signals (Handler(CatchOnce, CatchInfoOnce), SignalInfo, installHandler, keyboardSignal)++import qualified Polysemy.Conc.Data.Critical as Critical+import Polysemy.Conc.Data.Critical (Critical)+import Polysemy.Conc.Data.Interrupt (Interrupt(..))+import Polysemy.Conc.Data.Race (Race)+import qualified Polysemy.Conc.Data.Sync as Sync+import Polysemy.Conc.Race (race_)+import Polysemy.Conc.Sync (interpretSync)++putErr ::+  Member (Embed IO) r =>+  Text ->+  Sem r ()+putErr =+  embed . Text.hPutStrLn stderr++data InterruptState =+  InterruptState {+    quit :: !(MVar ()),+    finished :: !(MVar ()),+    listeners :: !(Set Text),+    original :: !(SignalInfo -> IO ()),+    handlers :: !(Map Text (IO ()))+  }++modListeners :: (Set Text -> Set Text) -> InterruptState -> InterruptState+modListeners f s@InterruptState {listeners} =+  s {listeners = f listeners}++modHandlers :: (Map Text (IO ()) -> Map Text (IO ())) -> InterruptState -> InterruptState+modHandlers f s@InterruptState {handlers} =+  s {handlers = f handlers}++waitQuit ::+  Members [AtomicState InterruptState, Embed IO] r =>+  Sem r ()+waitQuit = do+  mv <- atomicGets quit+  readMVar mv++checkListeners ::+  Members [AtomicState InterruptState, Embed IO] r =>+  Sem r ()+checkListeners =+  whenM (atomicGets (Set.null . listeners)) do+    fin <- atomicGets finished+    void (tryPutMVar fin ())++onQuit ::+  Members [AtomicState InterruptState, Embed IO] r =>+  Text ->+  Sem r a ->+  Sem r a+onQuit name ma = do+  atomicModify' (modListeners (Set.insert name))+  waitQuit+  a <- ma+  atomicModify' (modListeners (Set.delete name))+  checkListeners+  pure a++processHandler ::+  Member (Embed IO) r =>+  Text ->+  IO () ->+  Sem r ()+processHandler name thunk = do+  putErr [qt|processing interrupt handler: #{name}|]+  embed thunk++execInterrupt ::+  Members [AtomicState InterruptState, Embed IO] r =>+  Sem r (SignalInfo -> Sem r ())+execInterrupt = do+  InterruptState quitSignal finishSignal _ orig _ <- atomicGet+  putMVar quitSignal ()+  traverse_ (uncurry processHandler) . Map.toList =<< atomicGets handlers+  checkListeners+  takeMVar finishSignal+  embed . orig <$ putErr "interrupt handlers finished"++registerHandler ::+  Member (AtomicState InterruptState) r =>+  Text ->+  IO () ->+  Sem r ()+registerHandler name handler =+  atomicModify' (modHandlers (Map.insert name handler))++awaitOrKill ::+  Members [AtomicState InterruptState, Critical, Race, Async, Embed IO] r =>+  Text ->+  A.Async (Maybe a) ->+  Sem r (Maybe a)+awaitOrKill desc handle = do+  interpretSync @() do+    race_ (catchCritical (await handle)) kill+  where+    catchCritical =+      maybe waitKill (pure . Just) <=< Critical.catchAs @AsyncCancelled Nothing+    waitKill =+      Nothing <$ Sync.wait @() (Seconds 1)+    kill = do+      onQuit desc do+        putErr [qt|killing #{desc}|]+        cancel handle+        putErr [qt|killed #{desc}|]+        Sync.putBlock ()+        pure Nothing++interpretInterruptState ::+  Members [AtomicState InterruptState, Critical, Race, Async, Embed IO] r =>+  InterpreterFor Interrupt r+interpretInterruptState =+  interpretH \case+    Register name handler ->+      liftT (registerHandler name handler)+    Unregister name ->+      liftT $ atomicModify' \ s@InterruptState {handlers} -> s {handlers = Map.delete name handlers}+    WaitQuit ->+      liftT waitQuit+    Quit ->+      liftT do+        putErr "manual interrupt"+        void execInterrupt+    Interrupted ->+      liftT . fmap isJust . tryReadMVar =<< atomicGets quit+    KillOnQuit desc ma -> do+      maT <- runT ma+      ins <- getInspectorT+      handle <- raise (interpretInterruptState (async maT))+      result <- liftT (awaitOrKill desc handle)+      pure (join . fmap (inspect ins) <$> result)+{-# INLINE interpretInterruptState #-}++broadcastInterrupt ::+  Members [AtomicState InterruptState, Embed IO] r =>+  SignalInfo ->+  Sem r ()+broadcastInterrupt sig = do+  putErr "caught interrupt signal"+  orig <- execInterrupt+  orig sig++-- The original handler is either the default handler that kills all threads or a handler installed by an environment+-- like ghcid.+-- In the latter case, not calling it results in ghcid misbehaving.+-- To distinguish the two cases, the constructor used by the default is 'Catch', while a custom handler should usually+-- use 'CatchOnce', since you don't want to catch repeated occurences of SIGINT, as it will surely cause problems.+originalHandler :: Handler -> (SignalInfo -> IO ())+originalHandler (CatchOnce thunk) =+  (const thunk)+originalHandler (CatchInfoOnce thunk) =+  thunk+originalHandler _ =+  const pass+{-# INLINE originalHandler #-}++installSignalHandler ::+  TVar InterruptState ->+  IO Handler+installSignalHandler state =+  installHandler keyboardSignal (CatchInfoOnce handler) Nothing+  where+    handler sig =+      runFinal $ embedToFinal @IO $ runAtomicStateTVar state (broadcastInterrupt sig)++-- |Interpret 'Interrupt' by installing a signal handler.+interpretInterrupt ::+  Members [Critical, Race, Async, Embed IO] r =>+  InterpreterFor Interrupt r+interpretInterrupt sem = do+  quitMVar <- newEmptyMVar+  finishMVar <- newEmptyMVar+  state <- newTVarIO (InterruptState quitMVar finishMVar Set.empty (const pass) Map.empty)+  orig <- embed $ installSignalHandler state+  runAtomicStateTVar state do+    atomicModify' \ s -> s {original = originalHandler orig}+    interpretInterruptState $ raiseUnder sem+{-# INLINE interpretInterrupt #-}
+ lib/Polysemy/Conc/Prelude.hs view
@@ -0,0 +1,70 @@+{-# options_haddock hide #-}+{-# LANGUAGE NoImplicitPrelude #-}++module Polysemy.Conc.Prelude (+  module Polysemy.Conc.Prelude,+  module GHC.Err,+  module Polysemy,+  module Polysemy.AtomicState,+  module Relude,+) where++import qualified Data.String.Interpolate as Interpolate+import GHC.Err (undefined)+import Language.Haskell.TH.Quote (QuasiQuoter)+import Polysemy (+  Effect,+  EffectRow,+  Embed,+  Final,+  InterpreterFor,+  Member,+  Members,+  Sem,+  WithTactics,+  embed,+  embedToFinal,+  interpret,+  makeSem,+  pureT,+  raise,+  raiseUnder,+  raiseUnder2,+  raiseUnder3,+  reinterpret,+  runFinal,+  )+import Polysemy.AtomicState (AtomicState, atomicGet, atomicGets, atomicModify', atomicPut, runAtomicStateTVar)+import Relude hiding (+  Reader,+  State,+  Sum,+  Type,+  ask,+  asks,+  evalState,+  filterM,+  get,+  gets,+  hoistEither,+  modify,+  modify',+  put,+  readFile,+  runReader,+  runState,+  state,+  trace,+  traceShow,+  undefined,+  )++qt :: QuasiQuoter+qt =+  Interpolate.i+{-# INLINE qt #-}++unify :: Either a a -> a+unify =+  either id id+{-# INLINE unify #-}
+ lib/Polysemy/Conc/Queue/Result.hs view
@@ -0,0 +1,46 @@+{-# options_haddock hide #-}+module Polysemy.Conc.Queue.Result where++import qualified Polysemy.Conc.Data.QueueResult as QueueResult+import Polysemy.Conc.Data.QueueResult (QueueResult)++closedResult ::+  Maybe d ->+  QueueResult d+closedResult = \case+  Nothing -> QueueResult.Closed+  Just d -> QueueResult.Success d+{-# INLINE closedResult #-}++naResult ::+  Maybe d ->+  QueueResult d+naResult = \case+  Nothing -> QueueResult.NotAvailable+  Just d -> QueueResult.Success d+{-# INLINE naResult #-}++closedNaResult ::+  Maybe (Maybe d) ->+  QueueResult d+closedNaResult = \case+  Nothing -> QueueResult.Closed+  Just Nothing -> QueueResult.NotAvailable+  Just (Just d) -> QueueResult.Success d+{-# INLINE closedNaResult #-}++closedBoolResult ::+  Maybe Bool ->+  QueueResult ()+closedBoolResult = \case+  Nothing -> QueueResult.Closed+  Just False -> QueueResult.NotAvailable+  Just True -> QueueResult.Success ()+{-# INLINE closedBoolResult #-}++resultToMaybe :: QueueResult d -> Maybe d+resultToMaybe = \case+  QueueResult.Success d -> Just d+  QueueResult.NotAvailable -> Nothing+  QueueResult.Closed -> Nothing+{-# INLINE resultToMaybe #-}
+ lib/Polysemy/Conc/Queue/TB.hs view
@@ -0,0 +1,65 @@+-- |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 #-}
+ lib/Polysemy/Conc/Queue/TBM.hs view
@@ -0,0 +1,66 @@+-- |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)++-- |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 [Race, Embed IO] r =>+  -- |Buffer size+  Int ->+  InterpreterFor (Queue d) r+interpretQueueTBM maxQueued sem = do+  queue <- embed (newTBMQueueIO @d maxQueued)+  interpretQueueTBMWith queue sem+{-# INLINE interpretQueueTBM #-}
+ lib/Polysemy/Conc/Queue/Timeout.hs view
@@ -0,0 +1,22 @@+-- |Description: Timeout helper+module Polysemy.Conc.Queue.Timeout where++import Polysemy.Time (TimeUnit)++import qualified Polysemy.Conc.Data.QueueResult as QueueResult+import Polysemy.Conc.Data.QueueResult (QueueResult)+import Polysemy.Conc.Data.Race (Race)+import qualified Polysemy.Conc.Race as Race++-- |Run an 'STM' action atomically with a time limit+withTimeout ::+  TimeUnit t =>+  Members [Race, Embed IO] r =>+  t ->+  STM (Maybe d) ->+  Sem r (QueueResult d)+withTimeout timeout readQ =+  Race.timeout_ QueueResult.NotAvailable timeout reader'+  where+    reader' =+      maybe QueueResult.Closed QueueResult.Success <$> atomically readQ
+ lib/Polysemy/Conc/Race.hs view
@@ -0,0 +1,58 @@+{-# options_haddock prune #-}+-- |Description: Race interpreters+module Polysemy.Conc.Race where++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 qualified System.Timeout as System++import qualified Polysemy.Conc.Data.Race as Race+import Polysemy.Conc.Data.Race (Race)++biseqEither ::+  Functor f =>+  Either (f a) (f b) ->+  f (Either a b)+biseqEither =+  either (fmap Left) (fmap Right)+{-# 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'.+interpretRace ::+  Member (Final IO) r =>+  InterpreterFor Race r+interpretRace =+  interpretFinal @IO \case+    Race.Race left right ->+      fmap (fmap biseqEither) . Async.race <$> runS left <*> runS right+    Race.Timeout err (Time.convert -> MicroSeconds timeout) mb -> do+      mbT <- runS mb+      s <- getInitialStateS+      pure (maybe (Left err <$ s) (fmap Right) <$> System.timeout (fromIntegral timeout) mbT)+{-# INLINE interpretRace #-}++-- |Specialization of 'Race.race' for the case where both thunks return the same type, obviating the need for 'Either'.+race_ ::+  Member Race r =>+  Sem r a ->+  Sem r a ->+  Sem r a+race_ ma mb =+  unify <$> Race.race ma mb+{-# INLINE race_ #-}++-- |Specialization of 'Race.timeout' for the case where the thunk return the same type as the fallback, obviating the+-- need for 'Either'.+timeout_ ::+  TimeUnit u =>+  Member Race r =>+  a ->+  u ->+  Sem r a ->+  Sem r a+timeout_ err interval mb =+  unify <$> Race.timeout err interval mb+{-# INLINE timeout_ #-}
+ lib/Polysemy/Conc/Sync.hs view
@@ -0,0 +1,58 @@+-- |Description: Sync interpreters+module Polysemy.Conc.Sync where++import Control.Concurrent (isEmptyMVar)++import qualified Polysemy.Conc.Data.Race as Race+import Polysemy.Conc.Data.Race (Race)+import qualified Polysemy.Conc.Data.Sync as Sync+import Polysemy.Conc.Data.Sync (Sync)+import qualified Polysemy.Conc.Race as Race++-- |Interpret 'Sync' with the provided 'MVar'.+interpretSyncWith ::+  ∀ d r .+  Members [Race, Embed IO] r =>+  MVar d ->+  InterpreterFor (Sync d) r+interpretSyncWith var =+  interpret \case+    Sync.Block ->+      readMVar var+    Sync.Wait interval ->+      rightToMaybe <$> Race.timeout () interval (readMVar var)+    Sync.Try ->+      tryReadMVar var+    Sync.TakeBlock ->+      takeMVar var+    Sync.TakeWait interval ->+      rightToMaybe <$> Race.timeout () interval (takeMVar var)+    Sync.TakeTry ->+      tryTakeMVar var+    Sync.PutBlock d ->+      putMVar var d+    Sync.PutWait interval d ->+      Race.timeout_ False interval (True <$ putMVar var d)+    Sync.PutTry d ->+      tryPutMVar var d+    Sync.Empty ->+      embed (isEmptyMVar var)++-- |Interpret 'Sync' with an empty 'MVar'.+interpretSync ::+  ∀ d r .+  Members [Race, Embed IO] r =>+  InterpreterFor (Sync d) r+interpretSync sem = do+  var <- newEmptyMVar+  interpretSyncWith var sem++-- |Interpret 'Sync' with an 'MVar' containing the specified value.+interpretSyncAs ::+  ∀ d r .+  Members [Race, Embed IO] r =>+  d ->+  InterpreterFor (Sync d) r+interpretSyncAs d sem = do+  var <- newMVar d+  interpretSyncWith var sem
+ lib/Prelude.hs view
@@ -0,0 +1,5 @@+module Prelude (+  module Polysemy.Conc.Prelude,+) where++import Polysemy.Conc.Prelude
+ polysemy-conc.cabal view
@@ -0,0 +1,222 @@+cabal-version: 2.2++-- This file has been generated from package.yaml by hpack version 0.34.4.+--+-- see: https://github.com/sol/hpack++name:           polysemy-conc+version:        0.1.0.0+synopsis:       Polysemy Effects for Concurrency+description:    See <https://hackage.haskell.org/package/polysemy-conc/docs/Polysemy-Conc.html>+category:       Concurrency+homepage:       https://github.com/tek/polysemy-conc#readme+bug-reports:    https://github.com/tek/polysemy-conc/issues+author:         Torsten Schmits+maintainer:     tek@tryp.io+copyright:      2021 Torsten Schmits+license:        BSD-2-Clause-Patent+license-file:   LICENSE+build-type:     Simple+extra-source-files:+    readme.md+    Changelog.md++source-repository head+  type: git+  location: https://github.com/tek/polysemy-conc++library+  exposed-modules:+      Polysemy.Conc+      Polysemy.Conc.Critical+      Polysemy.Conc.Data.Critical+      Polysemy.Conc.Data.Interrupt+      Polysemy.Conc.Data.Queue+      Polysemy.Conc.Data.QueueResult+      Polysemy.Conc.Data.Race+      Polysemy.Conc.Data.Sync+      Polysemy.Conc.Interrupt+      Polysemy.Conc.Prelude+      Polysemy.Conc.Queue.Result+      Polysemy.Conc.Queue.TB+      Polysemy.Conc.Queue.TBM+      Polysemy.Conc.Queue.Timeout+      Polysemy.Conc.Race+      Polysemy.Conc.Sync+  other-modules:+      Prelude+      Paths_polysemy_conc+  autogen-modules:+      Paths_polysemy_conc+  hs-source-dirs:+      lib+  default-extensions:+      AllowAmbiguousTypes+      ApplicativeDo+      BangPatterns+      BinaryLiterals+      BlockArguments+      ConstraintKinds+      DataKinds+      DefaultSignatures+      DeriveAnyClass+      DeriveDataTypeable+      DeriveFoldable+      DeriveFunctor+      DeriveGeneric+      DeriveTraversable+      DerivingStrategies+      DerivingVia+      DisambiguateRecordFields+      DoAndIfThenElse+      DuplicateRecordFields+      EmptyDataDecls+      ExistentialQuantification+      FlexibleContexts+      FlexibleInstances+      FunctionalDependencies+      GADTs+      GeneralizedNewtypeDeriving+      InstanceSigs+      KindSignatures+      LambdaCase+      LiberalTypeSynonyms+      MultiParamTypeClasses+      MultiWayIf+      NamedFieldPuns+      OverloadedStrings+      OverloadedLists+      PackageImports+      PartialTypeSignatures+      PatternGuards+      PatternSynonyms+      PolyKinds+      QuantifiedConstraints+      QuasiQuotes+      RankNTypes+      RecordWildCards+      RecursiveDo+      ScopedTypeVariables+      StandaloneDeriving+      TemplateHaskell+      TupleSections+      TypeApplications+      TypeFamilies+      TypeFamilyDependencies+      TypeOperators+      TypeSynonymInstances+      UndecidableInstances+      UnicodeSyntax+      ViewPatterns+  ghc-options: -flate-specialise -fspecialise-aggressively -Wall -Wredundant-constraints+  build-depends:+      async+    , base ==4.*+    , containers+    , polysemy >=1.3 && <1.5+    , polysemy-time >=0.1.1.0 && <0.2+    , relude >=0.5 && <0.8+    , stm+    , stm-chans+    , string-interpolate >=0.2.1+    , template-haskell+    , text+    , time+    , unix+  mixins:+      base hiding (Prelude)+  default-language: Haskell2010++test-suite polysemy-conc-unit+  type: exitcode-stdio-1.0+  main-is: Main.hs+  other-modules:+      Polysemy.Conc.Test.InterruptTest+      Polysemy.Conc.Test.QueueTest+      Polysemy.Conc.Test.SyncTest+      Paths_polysemy_conc+  hs-source-dirs:+      test+  default-extensions:+      AllowAmbiguousTypes+      ApplicativeDo+      BangPatterns+      BinaryLiterals+      BlockArguments+      ConstraintKinds+      DataKinds+      DefaultSignatures+      DeriveAnyClass+      DeriveDataTypeable+      DeriveFoldable+      DeriveFunctor+      DeriveGeneric+      DeriveTraversable+      DerivingStrategies+      DerivingVia+      DisambiguateRecordFields+      DoAndIfThenElse+      DuplicateRecordFields+      EmptyDataDecls+      ExistentialQuantification+      FlexibleContexts+      FlexibleInstances+      FunctionalDependencies+      GADTs+      GeneralizedNewtypeDeriving+      InstanceSigs+      KindSignatures+      LambdaCase+      LiberalTypeSynonyms+      MultiParamTypeClasses+      MultiWayIf+      NamedFieldPuns+      OverloadedStrings+      OverloadedLists+      PackageImports+      PartialTypeSignatures+      PatternGuards+      PatternSynonyms+      PolyKinds+      QuantifiedConstraints+      QuasiQuotes+      RankNTypes+      RecordWildCards+      RecursiveDo+      ScopedTypeVariables+      StandaloneDeriving+      TemplateHaskell+      TupleSections+      TypeApplications+      TypeFamilies+      TypeFamilyDependencies+      TypeOperators+      TypeSynonymInstances+      UndecidableInstances+      UnicodeSyntax+      ViewPatterns+  ghc-options: -flate-specialise -fspecialise-aggressively -Wall -Wredundant-constraints -threaded -rtsopts -with-rtsopts=-N+  build-depends:+      async+    , base ==4.*+    , containers+    , hedgehog+    , polysemy >=1.3 && <1.5+    , polysemy-conc+    , polysemy-test+    , polysemy-time >=0.1.1.0 && <0.2+    , relude >=0.5 && <0.8+    , stm+    , stm-chans+    , string-interpolate >=0.2.1+    , tasty+    , tasty-hedgehog+    , template-haskell+    , text+    , time+    , unix+  mixins:+      base hiding (Prelude)+    , polysemy-conc hiding (Prelude)+    , polysemy-conc (Polysemy.Conc.Prelude as Prelude)+  default-language: Haskell2010
+ readme.md view
@@ -0,0 +1,8 @@+# About++This library provides a few convenience [polysemy] effects for using STM queues, MVars, signal handling and racing.++Please visit [hackage] for documentation.++[polysemy]: https://hackage.haskell.org/package/polysemy+[hackage]: https://hackage.haskell.org/package/polysemy-conc
+ test/Main.hs view
@@ -0,0 +1,37 @@+module Main where++import Polysemy.Conc.Test.InterruptTest (test_interrupt)+import Polysemy.Conc.Test.QueueTest (+  test_queueBlockTB,+  test_queueBlockTBM,+  test_queuePeekTBM,+  test_queueTB,+  test_queueTBM,+  test_queueTimeoutTBM,+  )+import Polysemy.Conc.Test.SyncTest (test_sync)+import Polysemy.Test (unitTest)+import Test.Tasty (TestTree, defaultMain, testGroup)++tests :: TestTree+tests =+  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+      ],+    testGroup "sync" [+      unitTest "sync" test_sync+      ],+    testGroup "interrupt" [+      unitTest "interrupt" test_interrupt+      ]+    ]++main :: IO ()+main =+  defaultMain tests
+ test/Polysemy/Conc/Test/InterruptTest.hs view
@@ -0,0 +1,32 @@+module Polysemy.Conc.Test.InterruptTest where++import Polysemy.Async (asyncToIOFinal)+import Polysemy.Test (UnitTest, runTestAuto, assertEq)+import System.Posix (Handler (CatchInfoOnce), SignalInfo, installHandler, keyboardSignal, raiseSignal)++import Polysemy.Conc.Critical (interpretCritical)+import qualified Polysemy.Conc.Data.Interrupt as Interrupt+import Polysemy.Conc.Interrupt (interpretInterrupt)+import Polysemy.Conc.Race (interpretRace)+import Control.Concurrent.STM (modifyTVar)++handler :: MVar () -> TVar Int -> SignalInfo -> IO ()+handler mv tv _ = do+  atomically (modifyTVar tv (5 +))+  putMVar mv ()++test_interrupt :: UnitTest+test_interrupt = do+  runTestAuto do+    tv <- newTVarIO 0+    mv <- newEmptyMVar+    embed (installHandler keyboardSignal (CatchInfoOnce (handler mv tv)) Nothing)+    asyncToIOFinal $ interpretCritical $ interpretRace $ interpretInterrupt do+      Interrupt.register "test 1" do+        atomically (modifyTVar tv (3 +))+      Interrupt.register "test 2" do+        atomically (modifyTVar tv (9 +))+      embed (raiseSignal keyboardSignal)+    takeMVar mv+    result <- readTVarIO tv+    assertEq @_ @IO 17 result
+ test/Polysemy/Conc/Test/QueueTest.hs view
@@ -0,0 +1,85 @@+module Polysemy.Conc.Test.QueueTest where++import Polysemy.Async (asyncToIOFinal, Async)+import Polysemy.Test (UnitTest, assertEq, assertJust, runTestAuto)+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.Race (Race)+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 =>+  Sem r (Maybe Int)+progSuccess = do+  Queue.write @Int 5+  resultToMaybe <$> Queue.read++progTimeoutSuccess ::+  Member (Queue Int) r =>+  Sem r (Maybe Int)+progTimeoutSuccess = do+  Queue.write @Int 5+  resultToMaybe <$> Queue.readTimeout (MilliSeconds 100)++progPeekSuccess ::+  Member (Queue Int) r =>+  Sem r (QueueResult Int, QueueResult Int, Maybe Int)+progPeekSuccess = do+  r1 <- Queue.tryPeek @Int+  Queue.write @Int 5+  r2 <- Queue.peek @Int+  r3 <- resultToMaybe <$> Queue.read @Int+  pure (r1, r2, r3)++run ::+  Members [Embed IO, Final IO] r =>+  Sem (Race : GhcTime : Async : r) a ->+  Sem r a+run =+  asyncToIOFinal .+  interpretTimeGhc .+  interpretRace++test_queueTBM :: UnitTest+test_queueTBM =+  runTestAuto do+    result <- run $ interpretQueueTBM @Int 1 $ progSuccess+    assertJust @_ @IO 5 result++test_queueTimeoutTBM :: UnitTest+test_queueTimeoutTBM =+  runTestAuto do+    result <- run $ interpretQueueTBM @Int 1 $ progTimeoutSuccess+    assertJust @_ @IO 5 result++test_queuePeekTBM :: UnitTest+test_queuePeekTBM =+  runTestAuto do+    result <- run $ interpretQueueTBM @Int 1 $ progPeekSuccess+    assertEq @_ @IO (QueueResult.NotAvailable, QueueResult.Success 5, Just 5) result++test_queueBlockTBM :: UnitTest+test_queueBlockTBM =+  runTestAuto do+    result <- run $ interpretQueueTBM @() 1 $ Queue.readTimeout @() (MilliSeconds 100)+    assertEq @_ @IO QueueResult.NotAvailable result++test_queueTB :: UnitTest+test_queueTB =+  runTestAuto do+    result <- run $ interpretQueueTB @Int 1 $ progSuccess+    assertJust @_ @IO 5 result++test_queueBlockTB :: UnitTest+test_queueBlockTB =+  runTestAuto do+    result <- run $ interpretQueueTB @() 1 $ Queue.readTimeout @() (MilliSeconds 100)+    assertEq @_ @IO QueueResult.NotAvailable result
+ test/Polysemy/Conc/Test/SyncTest.hs view
@@ -0,0 +1,44 @@+module Polysemy.Conc.Test.SyncTest where++import Polysemy.Async (Async, asyncToIOFinal, sequenceConcurrently)+import Polysemy.Test (UnitTest, assertEq, runTestAuto)++import Polysemy.Conc.Data.Race (Race)+import qualified Polysemy.Conc.Data.Sync as Sync+import Polysemy.Conc.Data.Sync (Sync)+import Polysemy.Conc.Race (interpretRace)+import Polysemy.Conc.Sync (interpretSync)++thread1 ::+  Member (Sync Int) r =>+  Sem r Int+thread1 = do+  a <- Sync.takeBlock @Int+  Sync.putBlock (a + 1)+  Sync.takeBlock++thread2 ::+  Member (Sync Int) r =>+  Sem r Int+thread2 =+  Sync.takeTry @Int >>= \case+    Just a -> pure a+    Nothing -> do+      Sync.putBlock @Int 1+      a <- Sync.takeBlock+      Sync.putBlock (a + 1)+      pure a++run ::+  Members [Embed IO, Final IO] r =>+  Sem (Race : Async : r) a ->+  Sem r a+run =+  asyncToIOFinal .+  interpretRace++test_sync :: UnitTest+test_sync =+  runTestAuto do+    result <- run $ interpretSync @Int $ sequenceConcurrently @[] [thread1, thread2]+    assertEq @_ @IO [Just 3, Just 2] result