polysemy-conc 0.1.0.3 → 0.1.1.0

raw patch · 13 files changed

+366/−116 lines, 13 filesdep +unagi-chandep ~polysemyPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependencies added: unagi-chan

Dependency ranges changed: polysemy

API changes (from Hackage documentation)

Files

• lib/Polysemy/Conc.hs +12/−2
• lib/Polysemy/Conc/Data/Sync.hs +0/−42
• lib/Polysemy/Conc/Effect/Events.hs +31/−0
• lib/Polysemy/Conc/Effect/Scoped.hs +73/−0
• lib/Polysemy/Conc/Effect/Sync.hs +42/−0
• lib/Polysemy/Conc/Interpreter/Events.hs +58/−0
• lib/Polysemy/Conc/Interpreter/Sync.hs +58/−0
• lib/Polysemy/Conc/Interrupt.hs +9/−9
• lib/Polysemy/Conc/Prelude.hs +3/−1
• lib/Polysemy/Conc/Sync.hs +29/−53
• polysemy-conc.cabal +13/−6
• test/Polysemy/Conc/Test/EventsTest.hs +35/−0
• test/Polysemy/Conc/Test/SyncTest.hs +3/−3

@@ -41,13 +41,24 @@    -- ** Interpreters   interpretInterrupt,++  -- * Event Channels+  Events,+  publish,+  consume,++  -- ** Interpreters+  interpretEventsChan, ) 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.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 (   interpretQueueListReadOnlyAtomic,@@ -59,7 +70,6 @@ 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:

@@ -1,42 +0,0 @@-{-# 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

@@ -0,0 +1,31 @@+{-# options_haddock prune #-}+-- |Description: Events/Consume Effects, Internal+module Polysemy.Conc.Effect.Events where++import Polysemy.Conc.Effect.Scoped (Scoped, scoped)++-- |Consume events emitted by 'Events'.+data Consume (e :: Type) :: Effect where+  Consume :: Consume e m e++makeSem ''Consume++-- |Marker for the 'Scoped' token for 'Events'.+newtype EventToken token =+  EventToken { unEventToken :: token }+  deriving (Eq, Show, Generic)++-- |An event publisher that can be consumed from multiple threads.+data Events (token :: Type) (e :: Type) :: Effect where+  Publish :: e -> Events token e m ()++makeSem ''Events++-- |Create a new scope for 'Events', causing the nested program to get its own copy of the event stream.+-- To be used with 'Polysemy.Conc.interpretEventsChan'.+subscribe ::+  ∀ e token r .+  Member (Scoped (EventToken token) (Consume e)) r =>+  InterpreterFor (Consume e) r+subscribe =+  scoped @(EventToken token)

@@ -0,0 +1,73 @@+{-# options_haddock prune #-}+-- |Description: Scoped Effect, Internal+module Polysemy.Conc.Effect.Scoped where++import Polysemy (transform)+import Polysemy.Internal (Sem (Sem, runSem), liftSem, send)+import Polysemy.Internal.Union (Weaving (Weaving), decomp, hoist, injWeaving)++-- |@Scoped@ transforms a program so that @effect@ is associated with a @resource@ within that program.+-- This requires the interpreter for @effect@ to be parameterized by @resource@ and constructed for every program using+-- @Scoped@ separately.+--+-- An application for this is 'Polysemy.Conc.Events', in which each program using the effect 'Consume' is interpreted+-- with its own copy of the event channel; or a database transaction, in which a transaction handle is created for the+-- wrapped program and passed to the interpreter for the database effect.+--+-- Resource creation is performed by the function passed to 'runScoped'.+--+-- The constructors are not intended to be used directly; the smart constructor 'scoped' is used like a local+-- interpreter for @effect@.+data Scoped (resource :: Type) (effect :: Effect) :: Effect where+  Run :: ∀ resource effect m a . resource -> effect m a -> Scoped resource effect m a+  InScope :: ∀ resource effect m a . (resource -> m a) -> Scoped resource effect m a++interpretH' ::+  ∀ e r .+  (∀ x . Weaving e (Sem (e : r)) x -> Sem r x) ->+  InterpreterFor e r+interpretH' h sem =+  Sem \ k -> runSem sem $ decomp >>> \case+    Right wav -> runSem (h wav) k+    Left g -> k $ hoist (interpretH' h) g++-- |Constructor for 'Scoped', taking a nested program and transforming all instances of @effect@ to+-- @Scoped resource effect@.+scoped ::+  ∀ resource effect r .+  Member (Scoped resource effect) r =>+  InterpreterFor effect r+scoped main =+  send $ InScope @resource @effect \ resource ->+    transform @effect (Run resource) main++-- |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.+--+-- @scopedInterpreter@ is a regular interpreter that is called with the @resource@ argument produced by @scope@.+runScoped ::+  ∀ resource effect r .+  (∀ x . (resource -> Sem r x) -> Sem r x) ->+  (resource -> InterpreterFor effect r) ->+  InterpreterFor (Scoped resource effect) r+runScoped scope scopedInterpreter =+  run+  where+    run :: InterpreterFor (Scoped resource effect) r+    run =+      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))++-- |Variant of 'Scoped' in which the resource allocator is a plain action.+runScopedAs ::+  ∀ resource effect r .+  Sem r resource ->+  (resource -> InterpreterFor effect r) ->+  InterpreterFor (Scoped resource effect) r+runScopedAs resource =+  runScoped \ f -> f =<< resource

@@ -0,0 +1,42 @@+{-# options_haddock prune #-}+-- |Description: Sync effect+module Polysemy.Conc.Effect.Sync where++import Polysemy.Time (TimeUnit)++-- |Abstracts an 'MVar'.+--+-- For documentation on the constructors, see the module "Polysemy.Conc.Effect.Sync".+--+-- @+-- import Polysemy.Conc (Sync)+-- import qualified Polysemy.Conc.Effect.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

@@ -0,0 +1,58 @@+{-# options_haddock prune #-}+-- |Description: Events/Consume Interpreters, Internal+module Polysemy.Conc.Interpreter.Events where++import Control.Concurrent.Chan.Unagi.Bounded (InChan, OutChan, dupChan, newChan, readChan, tryWriteChan)+import Polysemy (InterpretersFor)++import qualified Polysemy.Conc.Effect.Events as Events+import Polysemy.Conc.Effect.Events (EventToken (EventToken), Events, Consume)+import Polysemy.Conc.Effect.Scoped (Scoped, runScopedAs)++-- |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) ->+  InterpreterFor (Consume e) r+interpretConsumeChan (EventToken chan) =+  interpret \case+    Events.Consume ->+      embed (readChan chan)++-- |Interpret 'Events' by writing to an 'InChan'.+-- Used internally by 'interpretEventsChan', not safe to use directly.+-- When the channel queue is full, this silently discards events.+interpretEventsInChan ::+  ∀ e r .+  Member (Embed IO) r =>+  InChan e ->+  InterpreterFor (Events (OutChan e) e) r+interpretEventsInChan inChan =+  interpret \case+    Events.Publish e ->+      void (embed (tryWriteChan inChan e))++-- |Interpret 'Events' and 'Consume' together by connecting them to the two ends of an unagi channel.+-- 'Consume' is only interpreted in a 'Scoped' manner, ensuring that a new duplicate of the channel is created so that+-- all consumers see all events (from the moment they are connected).+--+-- This should be used in conjunction with 'Polysemy.Conc.subscribe':+--+-- @+-- interpretEventsChan do+--   async $ subscribe do+--     putStrLn =<< consume+--   publish "hello"+-- @+--+-- Whenever 'Polysemy.Conc.subscribe' creates a new scope, this interpreter calls 'dupChan' and passes the+-- duplicate to 'interpretConsumeChan'.+interpretEventsChan ::+  ∀ e r .+  Member (Embed IO) r =>+  InterpretersFor [Events (OutChan e) e, Scoped (EventToken (OutChan e)) (Consume e)] r+interpretEventsChan sem = do+  (inChan, _) <- embed (newChan @e 64)+  runScopedAs (EventToken <$> embed (dupChan inChan)) interpretConsumeChan (interpretEventsInChan inChan sem)

@@ -0,0 +1,58 @@+-- |Description: Sync interpreters+module Polysemy.Conc.Interpreter.Sync where++import Control.Concurrent (isEmptyMVar)++import qualified Polysemy.Conc.Data.Race as Race+import Polysemy.Conc.Data.Race (Race)+import qualified Polysemy.Conc.Effect.Sync as Sync+import Polysemy.Conc.Effect.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

@@ -10,16 +10,16 @@ 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 Polysemy.Time (Seconds (Seconds))+import System.Posix.Signals (Handler (CatchInfoOnce, CatchOnce), 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.Interrupt (Interrupt (..)) import Polysemy.Conc.Data.Race (Race)-import qualified Polysemy.Conc.Data.Sync as Sync+import qualified Polysemy.Conc.Effect.Sync as Sync+import Polysemy.Conc.Interpreter.Sync (interpretSync) import Polysemy.Conc.Race (race_)-import Polysemy.Conc.Sync (interpretSync)  putErr ::   Member (Embed IO) r =>@@ -87,10 +87,10 @@   Sem r (SignalInfo -> Sem r ()) execInterrupt = do   InterruptState quitSignal finishSignal _ orig _ <- atomicGet-  putMVar quitSignal ()-  traverse_ (uncurry processHandler) . Map.toList =<< atomicGets handlers-  checkListeners-  takeMVar finishSignal+  whenM (tryPutMVar quitSignal ()) do+    traverse_ (uncurry processHandler) . Map.toList =<< atomicGets handlers+    checkListeners+    takeMVar finishSignal   embed . orig <$ putErr "interrupt handlers finished"  registerHandler ::

@@ -1,7 +1,8 @@ {-# options_haddock hide #-}-{-# LANGUAGE NoImplicitPrelude #-}+{-# language NoImplicitPrelude #-}  module Polysemy.Conc.Prelude (+  module Data.Kind,   module Polysemy.Conc.Prelude,   module GHC.Err,   module Polysemy,@@ -9,6 +10,7 @@   module Relude, ) where +import Data.Kind (Type) import qualified Data.String.Interpolate as Interpolate import GHC.Err (undefined) import Language.Haskell.TH.Quote (QuasiQuoter)

@@ -1,58 +1,34 @@--- |Description: Sync interpreters+-- |Description: Sync Combinators 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+import qualified Polysemy.Time as Time+import Polysemy.Time (Time, TimeUnit) --- |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)+import qualified Polysemy.Conc.Effect.Sync as Sync+import Polysemy.Conc.Effect.Sync (Sync) --- |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+whileEmpty ::+  ∀ a r .+  Member (Sync a) r =>+  Sem r () ->+  Sem r ()+whileEmpty action =+  spin+  where+    spin = do+      action+      whenM (isJust <$> Sync.try @a) spin --- |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+whileEmptyInterval ::+  ∀ a u t d r .+  TimeUnit u =>+  Members [Time t d, Sync a] r =>+  u ->+  Sem r () ->+  Sem r ()+whileEmptyInterval interval action =+  spin+  where+    spin = do+      action+      whenM (isJust <$> Sync.try @a) (Time.sleep @t @d interval *> spin)

@@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack  name:           polysemy-conc-version:        0.1.0.3+version:        0.1.1.0 synopsis:       Polysemy Effects for Concurrency description:    See <https://hackage.haskell.org/package/polysemy-conc/docs/Polysemy-Conc.html> category:       Concurrency@@ -34,7 +34,11 @@       Polysemy.Conc.Data.Queue       Polysemy.Conc.Data.QueueResult       Polysemy.Conc.Data.Race-      Polysemy.Conc.Data.Sync+      Polysemy.Conc.Effect.Events+      Polysemy.Conc.Effect.Scoped+      Polysemy.Conc.Effect.Sync+      Polysemy.Conc.Interpreter.Events+      Polysemy.Conc.Interpreter.Sync       Polysemy.Conc.Interrupt       Polysemy.Conc.Prelude       Polysemy.Conc.Queue@@ -109,12 +113,12 @@       UndecidableInstances       UnicodeSyntax       ViewPatterns-  ghc-options: -flate-specialise -fspecialise-aggressively -Wall -Wredundant-constraints+  ghc-options: -flate-specialise -fspecialise-aggressively -Wall -Wredundant-constraints -Wunused-packages   build-depends:       async     , base ==4.*     , containers-    , polysemy >=1.3+    , polysemy >=1.5     , polysemy-time >=0.1.2.1     , relude >=0.7     , stm@@ -123,6 +127,7 @@     , template-haskell     , text     , time+    , unagi-chan >=0.4     , unix   mixins:       base hiding (Prelude)@@ -132,6 +137,7 @@   type: exitcode-stdio-1.0   main-is: Main.hs   other-modules:+      Polysemy.Conc.Test.EventsTest       Polysemy.Conc.Test.InterruptTest       Polysemy.Conc.Test.QueueTest       Polysemy.Conc.Test.SyncTest@@ -196,13 +202,13 @@       UndecidableInstances       UnicodeSyntax       ViewPatterns-  ghc-options: -flate-specialise -fspecialise-aggressively -Wall -Wredundant-constraints -threaded -rtsopts -with-rtsopts=-N+  ghc-options: -flate-specialise -fspecialise-aggressively -Wall -Wredundant-constraints -Wunused-packages -threaded -rtsopts -with-rtsopts=-N   build-depends:       async     , base ==4.*     , containers     , hedgehog-    , polysemy >=1.3+    , polysemy >=1.5     , polysemy-conc     , polysemy-test     , polysemy-time >=0.1.2.1@@ -215,6 +221,7 @@     , template-haskell     , text     , time+    , unagi-chan >=0.4     , unix   mixins:       base hiding (Prelude)

@@ -0,0 +1,35 @@+module Polysemy.Conc.Test.EventsTest where++import Control.Concurrent.Chan.Unagi.Bounded (OutChan)+import Polysemy.Async (async, asyncToIOFinal, await)+import Polysemy.Test (UnitTest, assertJust, runTestAuto)++import qualified Polysemy.Conc.Effect.Events as Events+import qualified Polysemy.Conc.Effect.Sync as Sync+import Polysemy.Conc.Interpreter.Events (interpretEventsChan)+import Polysemy.Conc.Interpreter.Sync (interpretSync)+import Polysemy.Conc.Race (interpretRace)++test_events :: UnitTest+test_events =+  (runTestAuto . asyncToIOFinal) do+    interpretRace $ interpretSync @() $ interpretEventsChan @Int $ interpretEventsChan @Text do+      thread1 <- async do+        Events.subscribe @Int @(OutChan Int) do+          Events.consume @Int+      thread2 <- async do+        Events.subscribe @Int @(OutChan Int) do+          Events.consume @Int+      thread3 <- async do+        Events.subscribe @Text @(OutChan Text) do+          Sync.putBlock ()+          Events.consume @Text+      Sync.takeBlock @()+      Events.publish @(OutChan Text) ("test" :: Text)+      Events.publish @(OutChan Int) (1 :: Int)+      num1 <- await thread1+      num2 <- await thread2+      text1 <- await thread3+      assertJust @_ @IO 1 num1+      assertJust @_ @IO 1 num2+      assertJust @_ @IO "test" text1

@@ -4,10 +4,10 @@ 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 qualified Polysemy.Conc.Effect.Sync as Sync+import Polysemy.Conc.Effect.Sync (Sync)+import Polysemy.Conc.Interpreter.Sync (interpretSync) import Polysemy.Conc.Race (interpretRace)-import Polysemy.Conc.Sync (interpretSync)  data Thread1 = Thread1 data Thread2 = Thread2