git.iko.soy/effectful
on master help sign in
Add support for unlift strategies (#17)
8b82309Andrzej Rybczak
bench/Main.hs+3−1
compare with master
unifiedsplit
@@ -12,10 +12,12 @@ import Test.Tasty.Bench
12
12
13import Countdown
13import Countdown
14import FileSizes
14import FileSizes
15+import Unlift
15
16
16main :: IO ()
17main :: IO ()
17main = defaultMain
18main = defaultMain
19+ [ unliftBenchmark
18− [ bgroup "countdown" $ map countdown [1000, 10000]
20+ , bgroup "countdown" $ map countdown [1000, 10000]
19 , bgroup "filesize" $ map filesize [100, 1000]
21 , bgroup "filesize" $ map filesize [100, 1000]
20 ]
22 ]
21
23
bench/Unlift.hsnew file+105
@@ -0,0 +1,105 @@
1+{-# LANGUAGE CPP #-}
2+module Unlift (unliftBenchmark) where
3+
4+#ifdef VERSION_criterion
5+import Criterion
6+import Criterion.Main
7+#endif
8+
9+#ifdef VERSION_tasty_bench
10+import Test.Tasty.Bench
11+#endif
12+
13+import UnliftIO
14+
15+import Effectful
16+import Effectful.State.Dynamic
17+
18+unliftBenchmark :: Benchmark
19+unliftBenchmark = bgroup "unlift"
20+ [ bgroup "reference"
21+ [ bench "dummy" $ nfIO benchDummy
22+ , bench "async" $ nfIO benchAsync
23+ , bench "concurrently" $ nfIO benchConcurrently
24+ ]
25+ , bgroup "shallow"
26+ [ bgroup "dummy"
27+ [ bench "seq" $ nfAppIO (runShallow SeqUnlift) benchDummy
28+ , shallowConc "conc" 1 benchDummy
29+ ]
30+ , shallowConc "async" 1 benchAsync
31+ , shallowConc "concurrently" 2 benchConcurrently
32+ ]
33+ , bgroup "deep"
34+ [ bgroup "dummy"
35+ [ bench "seq" $ nfAppIO (runDeep SeqUnlift) benchDummy
36+ , deepConc "conc" 1 benchDummy
37+ ]
38+ , deepConc "async" 1 benchAsync
39+ , deepConc "concurrently" 2 benchConcurrently
40+ ]
41+ ]
42+
43+----------------------------------------
44+
45+runShallow :: UnliftStrategy -> Eff '[IOE] a -> IO a
46+runShallow s = runIOE . withUnliftStrategy s
47+
48+runDeep
49+ :: UnliftStrategy
50+ -> Eff '[ State (), State (), State (), State (), State ()
51+ , State (), State (), State (), State (), State ()
52+ , IOE
53+ ] a
54+ -> IO a
55+runDeep s = runIOE
56+ . evalState () . evalState () . evalState () . evalState () . evalState ()
57+ . evalState () . evalState () . evalState () . evalState () . evalState ()
58+ . withUnliftStrategy s
59+
60+----------------------------------------
61+
62+shallowConc
63+ :: String
64+ -> Int
65+ -> Eff '[IOE] ()
66+ -> Benchmark
67+shallowConc name n f = bgroup name
68+ [ bench "ephemeral/limited" $ nfAppIO
69+ (runShallow (ConcUnlift Ephemeral $ Limited n)) f
70+ , bench "ephemeral/unlimited" $ nfAppIO
71+ (runShallow (ConcUnlift Ephemeral Unlimited)) f
72+ , bench "persistent/limited" $ nfAppIO
73+ (runShallow (ConcUnlift Persistent $ Limited n)) f
74+ , bench "persistent/unlimited" $ nfAppIO
75+ (runShallow (ConcUnlift Persistent Unlimited)) f
76+ ]
77+
78+deepConc
79+ :: String
80+ -> Int
81+ -> Eff '[ State (), State (), State (), State (), State ()
82+ , State (), State (), State (), State (), State ()
83+ , IOE] ()
84+ -> Benchmark
85+deepConc name n f = bgroup name
86+ [ bench "ephemeral/limited" $ nfAppIO
87+ (runDeep (ConcUnlift Ephemeral $ Limited n)) f
88+ , bench "ephemeral/unlimited" $ nfAppIO
89+ (runDeep (ConcUnlift Ephemeral Unlimited)) f
90+ , bench "persistent/limited" $ nfAppIO
91+ (runDeep (ConcUnlift Persistent $ Limited n)) f
92+ , bench "persistent/unlimited" $ nfAppIO
93+ (runDeep (ConcUnlift Persistent Unlimited)) f
94+ ]
95+
96+----------------------------------------
97+
98+benchDummy :: MonadUnliftIO m => m ()
99+benchDummy = withRunInIO $ \runIO -> runIO $ pure ()
100+
101+benchAsync :: MonadUnliftIO m => m ()
102+benchAsync = withAsync (pure ()) wait
103+
104+benchConcurrently :: MonadUnliftIO m => m ()
105+benchConcurrently = concurrently_ (pure ()) (pure ())
effectful.cabal+17−1
@@ -47,6 +47,16 @@ common language
47 TypeFamilies
47 TypeFamilies
48 TypeOperators
48 TypeOperators
49
49
50+library effectful-internal-utils
51+ import: language
52+
53+ build-depends: base
54+
55+ hs-source-dirs: utils
56+ c-sources: utils/utils.c
57+
58+ exposed-modules: Effectful.Internal.Utils
59+
50library
60library
51 import: language
61 import: language
52
62
@@ -55,8 +65,8 @@ library
55 build-depends: base >= 4.13 && <5
65 build-depends: base >= 4.13 && <5
56 , async
66 , async
57 , containers
67 , containers
68+ , effectful-internal-utils
58 , exceptions
69 , exceptions
59− , ghc-prim
60 , monad-control
70 , monad-control
61 , primitive
71 , primitive
62 , resourcet
72 , resourcet
@@ -77,6 +87,7 @@ library
77 Effectful.Internal.Effect
87 Effectful.Internal.Effect
78 Effectful.Internal.Env
88 Effectful.Internal.Env
79 Effectful.Internal.Monad
89 Effectful.Internal.Monad
90+ Effectful.Internal.Unlift
80 Effectful.Interpreter
91 Effectful.Interpreter
81 Effectful.Monad
92 Effectful.Monad
82 Effectful.Reader
93 Effectful.Reader
@@ -89,6 +100,8 @@ library
89 Effectful.Writer.Dynamic
100 Effectful.Writer.Dynamic
90 Effectful.Writer.MVar
101 Effectful.Writer.MVar
91
102
103+ reexported-modules: Effectful.Internal.Utils
104+
92test-suite test
105test-suite test
93 import: language
106 import: language
94
107
@@ -109,6 +122,7 @@ test-suite test
109 main-is: Main.hs
122 main-is: Main.hs
110
123
111 other-modules: AsyncTests
124 other-modules: AsyncTests
125+ ConcurrencyTests
112 ErrorTests
126 ErrorTests
113 StateTests
127 StateTests
114 Utils
128 Utils
@@ -133,6 +147,7 @@ benchmark bench
133 , tasty-bench
147 , tasty-bench
134 , transformers
148 , transformers
135 , unix
149 , unix
150+ , unliftio
136
151
137 hs-source-dirs: bench
152 hs-source-dirs: bench
138
153
@@ -141,3 +156,4 @@ benchmark bench
141
156
142 other-modules: Countdown
157 other-modules: Countdown
143 FileSizes
158 FileSizes
159+ Unlift
src/Effectful.hs+18
@@ -18,6 +18,13 @@ module Effectful
18 , IOE
18 , IOE
19 , runIOE
19 , runIOE
20
20
21+ -- *** Unlift strategies
22+ , UnliftStrategy(..)
23+ , Persistence(..)
24+ , Limit(..)
25+ , unliftStrategy
26+ , withUnliftStrategy
27+
21 -- * Building 'Effect' handlers
28 -- * Building 'Effect' handlers
22
29
23 -- ** Basic handlers
30 -- ** Basic handlers
@@ -27,6 +34,17 @@ module Effectful
27 -- ** Derived handlers
34 -- ** Derived handlers
28 , reinterpret
35 , reinterpret
29 , reinterpretM
36 , reinterpretM
37+
38+ -- ** Local operations
39+ , LocalEnv
40+
41+ -- *** 'Eff'
42+ , localSeqUnlift
43+ , localUnlift
44+
45+ -- *** 'IO'
46+ , localSeqUnliftIO
47+ , localUnliftIO
30 ) where
48 ) where
31
49
32import Effectful.Interpreter
50import Effectful.Interpreter
src/Effectful/Internal/Monad.hs+115−41
@@ -11,11 +11,9 @@ module Effectful.Internal.Monad
11 , runEff
11 , runEff
12
12
13 -- ** Access to the internal representation
13 -- ** Access to the internal representation
14− , RunIn
15 , unEff
14 , unEff
16 , unsafeEff
15 , unsafeEff
17 , unsafeEff_
16 , unsafeEff_
18− , unsafeUnliftEff
19
17
20 -- * Fail
18 -- * Fail
21 , Fail
19 , Fail
@@ -25,7 +23,18 @@ module Effectful.Internal.Monad
25 , IOE
23 , IOE
26 , runIOE
24 , runIOE
27
25
26+ -- ** Unlift strategies
27+ , UnliftStrategy(..)
28+ , Persistence(..)
29+ , Limit(..)
30+ , unliftStrategy
31+ , withUnliftStrategy
32+
28− -- * Low-level helpers
33+ --- *** Low-level helpers
34+ , seqUnliftEff
35+ , concUnliftEff
36+
37+ -- * Primitive effects
29
38
30 -- ** Running
39 -- ** Running
31 , runEffect
40 , runEffect
@@ -50,12 +59,14 @@ import Control.Monad.IO.Class
50import Control.Monad.IO.Unlift
59import Control.Monad.IO.Unlift
51import Control.Monad.Trans.Control
60import Control.Monad.Trans.Control
52import Data.Unique
61import Data.Unique
53−import GHC.Magic (oneShot)
62+import GHC.Exts (oneShot)
54import System.IO.Unsafe (unsafeDupablePerformIO)
63import System.IO.Unsafe (unsafeDupablePerformIO)
55import qualified Control.Monad.Catch as E
64import qualified Control.Monad.Catch as E
56
65
57import Effectful.Internal.Effect
66import Effectful.Internal.Effect
58import Effectful.Internal.Env
67import Effectful.Internal.Env
68+import Effectful.Internal.Unlift
69+import Effectful.Internal.Utils
59
70
60type role Eff nominal representational
71type role Eff nominal representational
61
72
@@ -113,40 +124,32 @@ unsafeEff m = Eff (oneShot m)
113unsafeEff_ :: IO a -> Eff es a
124unsafeEff_ :: IO a -> Eff es a
114unsafeEff_ m = unsafeEff $ \_ -> m
125unsafeEff_ m = unsafeEff $ \_ -> m
115
126
116−-- | A function that runs 'Eff' operations in @m@ (usually a local 'Eff'
117−-- environment or 'IO').
118−type RunIn es m = forall r. Eff es r -> m r
119−
120−-- | Lower 'Eff' operations into 'IO'.
127+-- | Lower 'Eff' operations into 'IO' ('SeqUnlift').
121−--
128+seqUnliftEff :: ((forall r. Eff es r -> IO r) -> IO a) -> Eff es a
122−-- /Note:/ the 'RunIn' argument has to be called from the same thread as
123−-- 'unsafeUnliftEff' ('liftBaseWith' / 'withRunInIO'). Attempting otherwise
124−-- results in a runtime error, because it cannot be done safely in general
125−-- (*). If you need to spawn threads, have a look at "Effectful.Async" or create
126−-- a new 'Eff' environment.
127−--
128−-- (*) Consider the following:
129−--
130−-- @
131−-- runState 'x' $ do
132−-- liftBaseWith $ \run -> forkIO $ do
133−-- threadDelay 1000000
134−-- run $ ... (1)
135−-- ... (2)
136−-- ...
137−-- @
138−--
139−-- If (1) runs after (2) completed, the State Char effect is no longer in scope,
140−-- but (1) assumes otherwise and things break horribly.
141−unsafeUnliftEff :: (RunIn es IO -> IO a) -> Eff es a
142−unsafeUnliftEff f = unsafeEff $ \es -> do
129+seqUnliftEff f = unsafeEff $ \es -> do
143 tid0 <- myThreadId
130 tid0 <- myThreadId
144 f $ \m -> do
131 f $ \m -> do
145 tid <- myThreadId
132 tid <- myThreadId
146− if tid == tid0
133+ if tid `eqThreadId` tid0
147 then unEff m es
134 then unEff m es
148− else error $ "Running Eff operations in a different thread is "
135+ else error
149− ++ "currently not possible with the unlifting function"
136+ $ "If you want to use the unlifting function to run Eff operations "
137+ ++ "in multiple threads, have a look at UnliftStrategy (ConcUnlift)."
138+
139+-- | Lower 'Eff' operations into 'IO' ('ConcUnlift').
140+concUnliftEff
141+ :: Persistence
142+ -> Limit
143+ -> ((forall r. Eff es r -> IO r) -> IO a)
144+ -> Eff es a
145+concUnliftEff Ephemeral (Limited uses) f = unsafeEff $ \es -> do
146+ ephemeralConcUnliftIO uses f es unEff
147+concUnliftEff Ephemeral Unlimited f = unsafeEff $ \es -> do
148+ ephemeralConcUnliftIO maxBound f es unEff
149+concUnliftEff Persistent (Limited threads) f = unsafeEff $ \es -> do
150+ persistentConcUnliftIO False threads f es unEff
151+concUnliftEff Persistent Unlimited f = unsafeEff $ \es -> do
152+ persistentConcUnliftIO True maxBound f es unEff
150
153
151----------------------------------------
154----------------------------------------
152-- Base
155-- Base
@@ -246,14 +249,14 @@ tryFailIO tag m =
246
249
247-- | Run arbitrary 'IO' operations via 'MonadIO', 'MonadBaseControl' 'IO' or
250-- | Run arbitrary 'IO' operations via 'MonadIO', 'MonadBaseControl' 'IO' or
248-- 'MonadUnliftIO'.
251-- 'MonadUnliftIO'.
249−data IOE :: Effect where
252+newtype IOE :: Effect where
250− IOE :: IOE m r
253+ IOE :: UnliftStrategy -> IOE m r
251
254
252-- | Run an 'Eff' operation with side effects.
255-- | Run an 'Eff' operation with side effects.
253--
256--
254-- For the pure version see 'runEff'.
257-- For the pure version see 'runEff'.
255runIOE :: Eff '[IOE] a -> IO a
258runIOE :: Eff '[IOE] a -> IO a
256−runIOE m = unEff (evalEffect (IdE IOE) m) =<< emptyEnv
259+runIOE m = unEff (evalEffect (IdE (IOE SeqUnlift)) m) =<< emptyEnv
257
260
258instance IOE :> es => MonadIO (Eff es) where
261instance IOE :> es => MonadIO (Eff es) where
259 liftIO = unsafeEff_
262 liftIO = unsafeEff_
@@ -263,11 +266,82 @@ instance IOE :> es => MonadBase IO (Eff es) where
263
266
264instance IOE :> es => MonadBaseControl IO (Eff es) where
267instance IOE :> es => MonadBaseControl IO (Eff es) where
265 type StM (Eff es) a = a
268 type StM (Eff es) a = a
266− liftBaseWith = unsafeUnliftEff
269+ liftBaseWith = unliftEff
267 restoreM = pure
270 restoreM = pure
268
271
269instance IOE :> es => MonadUnliftIO (Eff es) where
272instance IOE :> es => MonadUnliftIO (Eff es) where
270− withRunInIO = unsafeUnliftEff
273+ withRunInIO = unliftEff
274+
275+----------------------------------------
276+-- Unlift strategies
277+
278+-- | The strategy to use when unlifting 'Eff' operations via 'liftBaseWith',
279+-- 'withRunInIO' or the 'Effectful.Interpreter.localUnlift' family.
280+data UnliftStrategy
281+ = SeqUnlift
282+ -- ^ The fastest strategy and a default setting for 'IOE'. An attempt to call
283+ -- the unlifting function in threads distinct from its creator will result in
284+ -- a runtime error.
285+ | ConcUnlift Persistence Limit
286+ -- ^ A strategy that makes it possible for the unlifting function to be called
287+ -- in threads distinct from its creator. See 'Persistence' and 'Limit'
288+ -- settings for more information.
289+ deriving (Eq, Ord, Show)
290+
291+-- | Persistence setting for the 'ConcUnlift' strategy.
292+--
293+-- Different functions require different persistence strategies. Examples:
294+--
295+-- - Lifting 'pooledMapConcurrentlyN' from the @unliftio@ library requires the
296+-- 'Ephemeral' strategy as we don't want jobs to share environment changes
297+-- made by previous jobs run in the same worker thread.
298+--
299+-- - Lifting 'Control.Concurrent.forkIOWithUnmask' requires the 'Persistent'
300+-- strategy, otherwise the unmasking function would start with a fresh
301+-- environment each time it's called.
302+data Persistence
303+ = Ephemeral
304+ -- ^ Don't persist the environment between calls to the unlifting function in
305+ -- threads distinct from its creator.
306+ | Persistent
307+ -- ^ Persist the environment between calls to the unlifting function within a
308+ -- particular thread.
309+ deriving (Eq, Ord, Show)
310+
311+-- | Limit setting for the 'ConcUnlift' strategy.
312+data Limit
313+ = Limited Int
314+ -- ^ Behavior dependent on the 'Persistence' setting.
315+ --
316+ -- For 'Ephemeral', it limits the amount of uses of the unlifting function in
317+ -- threads distinct from its creator to @N@. The unlifting function will
318+ -- create @N@ copies of the environment when called @N@ times and @K+1@ copies
319+ -- when called @K < N@ times.
320+ --
321+ -- For 'Persistent', it limits the amount of threads, distinct from the
322+ -- creator of the unlifting function, it can be called in to @N@. The amount
323+ -- of calls to the unlifting function within a particular threads is
324+ -- unlimited. The unlifting function will create @N@ copies of the environment
325+ -- when called in @N@ threads and @K+1@ copies when called in @K < N@ threads.
326+ | Unlimited
327+ -- ^ Unlimited use of the unlifting function.
328+ deriving (Eq, Ord, Show)
329+
330+-- | Get the current 'UnliftStrategy'.
331+unliftStrategy :: IOE :> es => Eff es UnliftStrategy
332+unliftStrategy = readerEffectM $ \(IdE (IOE unlift)) -> pure unlift
333+
334+-- | Locally override the 'UnliftStrategy' with a given value.
335+--
336+-- /Note:/ the strategy is always reset to 'SeqUnlift' for new threads.
337+withUnliftStrategy :: IOE :> es => UnliftStrategy -> Eff es a -> Eff es a
338+withUnliftStrategy unlift = localEffect $ \_ -> IdE (IOE unlift)
339+
340+-- | Helper for 'MonadBaseControl' and 'MonadUnliftIO' instances.
341+unliftEff :: IOE :> es => ((forall r. Eff es r -> IO r) -> IO a) -> Eff es a
342+unliftEff f = unliftStrategy >>= \case
343+ SeqUnlift -> seqUnliftEff f
344+ ConcUnlift p b -> withUnliftStrategy SeqUnlift $ concUnliftEff p b f
271
345
272----------------------------------------
346----------------------------------------
273-- Helpers
347-- Helpers
@@ -317,8 +391,8 @@ readerEffectM f = unsafeEff $ \es -> getEnv es >>= \e -> unEff (f e) es
317{-# INLINE readerEffectM #-}
391{-# INLINE readerEffectM #-}
318
392
319stateEffectM :: e :> es => (i e -> Eff es (a, i e)) -> Eff es a
393stateEffectM :: e :> es => (i e -> Eff es (a, i e)) -> Eff es a
320−stateEffectM f = unsafeEff $ \es -> do
394+stateEffectM f = unsafeEff $ \es -> mask $ \release -> do
321− (a, e) <- (\e -> unEff (f e) es) =<< getEnv es
395+ (a, e) <- (\e -> release $ unEff (f e) es) =<< getEnv es
322 unsafePutEnv e es
396 unsafePutEnv e es
323 pure a
397 pure a
324{-# INLINE stateEffectM #-}
398{-# INLINE stateEffectM #-}
src/Effectful/Internal/Unlift.hsnew file+209
@@ -0,0 +1,209 @@
1+{-# LANGUAGE MagicHash #-}
2+{-# LANGUAGE UnboxedTuples #-}
3+{-# OPTIONS_HADDOCK not-home #-}
4+-- | Implementation of the concurrent unlift.
5+--
6+-- This module is intended for internal use only, and may change without warning
7+-- in subsequent releases.
8+module Effectful.Internal.Unlift
9+ ( ephemeralConcUnliftIO
10+ , persistentConcUnliftIO
11+ ) where
12+
13+import Control.Concurrent
14+import Control.Monad
15+import GHC.Conc.Sync (ThreadId(..))
16+import GHC.Exts (mkWeak#, mkWeakNoFinalizer#)
17+import GHC.IO (IO(..))
18+import GHC.Weak (Weak(..))
19+import System.Mem.Weak (deRefWeak)
20+import qualified Data.IntMap.Strict as IM
21+
22+import Effectful.Internal.Env
23+import Effectful.Internal.Utils
24+
25+-- | Concurrent unlift that doesn't preserve the environment between calls to
26+-- the unlifting function in threads other than its creator.
27+ephemeralConcUnliftIO
28+ :: Int
29+ -> ((forall r. m r -> IO r) -> IO a)
30+ -> Env es
31+ -> (forall r. m r -> Env es -> IO r)
32+ -> IO a
33+ephemeralConcUnliftIO uses f es0 unEff = do
34+ when (uses < 0) $ do
35+ error $ "ephemeralConcUnliftIO: invalid number of uses: " ++ show uses
36+ tid0 <- myThreadId
37+ -- Create a copy of the environment as a template for the other threads to
38+ -- use. This can't be done from inside the callback as the environment might
39+ -- have already changed by then.
40+ esTemplate <- cloneEnv es0
41+ mvUses <- newMVar uses
42+ f $ \m -> do
43+ es <- myThreadId >>= \case
44+ tid | tid0 `eqThreadId` tid -> pure es0
45+ _ -> modifyMVar mvUses $ \case
46+ 0 -> error
47+ $ "Number of permitted calls (" ++ show uses ++ ") to the unlifting "
48+ ++ "function in other threads was exceeded. Please increase the limit "
49+ ++ "or use the unlimited variant."
50+ 1 -> pure (0, esTemplate)
51+ n -> do
52+ let newUses = n - 1
53+ es <- cloneEnv esTemplate
54+ newUses `seq` pure (newUses, es)
55+ unEff m es
56+
57+-- | Concurrent unlift that preserves the environment between calls to the
58+-- unlifting function within a particular thread.
59+persistentConcUnliftIO
60+ :: Bool
61+ -> Int
62+ -> ((forall r. m r -> IO r) -> IO a)
63+ -> Env es
64+ -> (forall r. m r -> Env es -> IO r)
65+ -> IO a
66+persistentConcUnliftIO cleanUp threads f es0 unEff = do
67+ when (threads < 0) $ do
68+ error $ "persistentConcUnliftIO: invalid number of threads: " ++ show threads
69+ tid0 <- myThreadId
70+ -- Create a copy of the environment as a template for the other threads to
71+ -- use. This can't be done from inside the callback as the environment might
72+ -- have already changed by then.
73+ esTemplate <- cloneEnv es0
74+ mvEntries <- newMVar $ ThreadEntries threads IM.empty
75+ f $ \m -> do
76+ es <- myThreadId >>= \case
77+ tid | tid0 `eqThreadId` tid -> pure es0
78+ tid -> modifyMVar mvEntries $ \te -> do
79+ let wkTid = weakThreadId tid
80+ (mes, i) <- case wkTid `IM.lookup` teEntries te of
81+ Just (ThreadEntry i td) -> (, i) <$> lookupEnv tid td
82+ Nothing -> pure (Nothing, newEntryId)
83+ case mes of
84+ Just es -> pure (te, es)
85+ Nothing -> case teCapacity te of
86+ 0 -> error
87+ $ "Number of other threads (" ++ show threads ++ ") permitted to "
88+ ++ "use the unlifting function was exceeded. Please increase the "
89+ ++ "limit or use the unlimited variant."
90+ 1 -> do
91+ wkTidEs <- mkWeakThreadIdEnv tid esTemplate wkTid i mvEntries cleanUp
92+ let newEntries = ThreadEntries
93+ { teCapacity = teCapacity te - 1
94+ , teEntries = addThreadData wkTid i wkTidEs $ teEntries te
95+ }
96+ newEntries `seq` pure (newEntries, esTemplate)
97+ _ -> do
98+ es <- cloneEnv esTemplate
99+ wkTidEs <- mkWeakThreadIdEnv tid es wkTid i mvEntries cleanUp
100+ let newEntries = ThreadEntries
101+ { teCapacity = teCapacity te - 1
102+ , teEntries = addThreadData wkTid i wkTidEs $ teEntries te
103+ }
104+ newEntries `seq` pure (newEntries, es)
105+ unEff m es
106+
107+----------------------------------------
108+-- Data types
109+
110+newtype EntryId = EntryId Int
111+ deriving Eq
112+
113+newEntryId :: EntryId
114+newEntryId = EntryId 0
115+
116+nextEntryId :: EntryId -> EntryId
117+nextEntryId (EntryId i) = EntryId (i + 1)
118+
119+data ThreadEntries es = ThreadEntries
120+ { teCapacity :: Int
121+ , teEntries :: IM.IntMap (ThreadEntry es)
122+ }
123+
124+-- | In GHC < 9 weak thread ids are 32bit long, while ThreadIdS are 64bit long,
125+-- so there is potential for collisions. This is solved by keeping, for a
126+-- particular weak thread id, a list of ThreadIdS with unique EntryIdS.
127+data ThreadEntry es = ThreadEntry EntryId (ThreadData es)
128+
129+data ThreadData es
130+ = ThreadData EntryId (Weak (ThreadId, Env es)) (ThreadData es)
131+ | NoThreadData
132+
133+----------------------------------------
134+-- Weak references to threads
135+
136+mkWeakThreadIdEnv
137+ :: ThreadId
138+ -> Env es
139+ -> Int
140+ -> EntryId
141+ -> MVar (ThreadEntries es)
142+ -> Bool
143+ -> IO (Weak (ThreadId, Env es))
144+mkWeakThreadIdEnv t@(ThreadId t#) es wkTid i v = \case
145+ True -> IO $ \s0 ->
146+ case mkWeak# t# (t, es) finalizer s0 of
147+ (# s1, w #) -> (# s1, Weak w #)
148+ False -> IO $ \s0 ->
149+ case mkWeakNoFinalizer# t# (t, es) s0 of
150+ (# s1, w #) -> (# s1, Weak w #)
151+ where
152+ IO finalizer = deleteThreadData wkTid i v
153+
154+----------------------------------------
155+-- Manipulation of ThreadEntries
156+
157+lookupEnv :: ThreadId -> ThreadData es -> IO (Maybe (Env es))
158+lookupEnv tid0 = \case
159+ NoThreadData -> pure Nothing
160+ ThreadData _ wkTidEs td -> deRefWeak wkTidEs >>= \case
161+ Nothing -> lookupEnv tid0 td
162+ Just (tid, es)
163+ | tid0 `eqThreadId` tid -> pure $ Just es
164+ | otherwise -> lookupEnv tid0 td
165+
166+----------------------------------------
167+
168+addThreadData
169+ :: Int
170+ -> EntryId
171+ -> Weak (ThreadId, Env es)
172+ -> IM.IntMap (ThreadEntry es)
173+ -> IM.IntMap (ThreadEntry es)
174+addThreadData wkTid i w teMap
175+ | i == newEntryId = IM.insert wkTid (newThreadEntry i w) teMap
176+ | otherwise = IM.adjust (consThreadData w) wkTid teMap
177+
178+newThreadEntry :: EntryId -> Weak (ThreadId, Env es) -> ThreadEntry es
179+newThreadEntry i w = ThreadEntry (nextEntryId i) $ ThreadData i w NoThreadData
180+
181+consThreadData :: Weak (ThreadId, Env es) -> ThreadEntry es -> ThreadEntry es
182+consThreadData w (ThreadEntry i td) =
183+ ThreadEntry (nextEntryId i) $ ThreadData i w td
184+
185+----------------------------------------
186+
187+deleteThreadData :: Int -> EntryId -> MVar (ThreadEntries es) -> IO ()
188+deleteThreadData wkTid i v = modifyMVar_ v $ \te -> do
189+ let newEntries = ThreadEntries
190+ { teCapacity = case teCapacity te of
191+ -- If the template copy of the environment hasn't been consumed
192+ -- yet, the capacity can be restored.
193+ 0 -> 0
194+ n -> n + 1
195+ , teEntries = IM.update (cleanThreadEntry i) wkTid $ teEntries te
196+ }
197+ newEntries `seq` pure newEntries
198+
199+cleanThreadEntry :: EntryId -> ThreadEntry es -> Maybe (ThreadEntry es)
200+cleanThreadEntry i0 (ThreadEntry i td0) = case cleanThreadData i0 td0 of
201+ NoThreadData -> Nothing
202+ td -> Just (ThreadEntry i td)
203+
204+cleanThreadData :: EntryId -> ThreadData es -> ThreadData es
205+cleanThreadData i0 = \case
206+ NoThreadData -> NoThreadData
207+ ThreadData i w td
208+ | i0 == i -> td
209+ | otherwise -> ThreadData i w (cleanThreadData i0 td)
src/Effectful/Interpreter.hs+80−17
@@ -3,15 +3,25 @@ module Effectful.Interpreter
3 send
3 send
4
4
5 -- * Handling effects
5 -- * Handling effects
6− , RunIn
7
6
8− -- ** Interpretation
7+ -- ** Basic handlers
9 , interpret
8 , interpret
10 , interpretM
9 , interpretM
11
10
12− -- ** Reinterpretation
11+ -- ** Derived handlers
13 , reinterpret
12 , reinterpret
14 , reinterpretM
13 , reinterpretM
14+
15+ -- ** Local operations
16+ , LocalEnv
17+
18+ -- *** 'Eff'
19+ , localSeqUnlift
20+ , localUnlift
21+
22+ -- *** 'IO'
23+ , localSeqUnliftIO
24+ , localUnliftIO
15 ) where
25 ) where
16
26
17import Control.Exception
27import Control.Exception
@@ -24,10 +34,13 @@ import Effectful.Internal.Monad
24----------------------------------------
34----------------------------------------
25-- Interpreter
35-- Interpreter
26
36
37+-- | Representation of the local 'Eff' environment.
38+newtype LocalEnv es = LocalEnv (Env es)
39+
27data Interpreter (e :: Effect) = forall handlerEs. Interpreter
40data Interpreter (e :: Effect) = forall handlerEs. Interpreter
28 { _env :: Env handlerEs
41 { _env :: Env handlerEs
29 , _interpret :: forall a localEs. HasCallStack
42 , _interpret :: forall a localEs. HasCallStack
30− => RunIn localEs IO
43+ => LocalEnv localEs
31 -> e (Eff localEs) a
44 -> e (Eff localEs) a
32 -> Eff handlerEs a
45 -> Eff handlerEs a
33 }
46 }
@@ -48,12 +61,10 @@ runInterpreter es0 e m = do
48-- Sending commands
61-- Sending commands
49
62
50-- | Send an operation of a given effect to the interpreter.
63-- | Send an operation of a given effect to the interpreter.
51−--
52−-- This is used to connect the definition of an effect to the 'Eff' monad so
53−-- that it can be used and handled.
54send :: (HasCallStack, e :> es) => e (Eff es) a -> Eff es a
64send :: (HasCallStack, e :> es) => e (Eff es) a -> Eff es a
65+send op = unsafeEff $ \es -> do
55−send op = readerEffectM $ \Interpreter{..} -> do
66+ Interpreter{..} <- getEnv es
56− unsafeUnliftEff $ \run -> unEff (_interpret run op) _env
67+ unEff (_interpret (LocalEnv es) op) _env
57
68
58----------------------------------------
69----------------------------------------
59-- Interpretation
70-- Interpretation
@@ -68,16 +79,18 @@ interpret interpreter m = unsafeEff $ \es -> do
68 les <- forkEnv es
79 les <- forkEnv es
69 runInterpreter es (Interpreter les $ \_ -> interpreter) m
80 runInterpreter es (Interpreter les $ \_ -> interpreter) m
70
81
71−-- | Interpret a higher order effect with help of a function running local 'Eff'
82+-- | Interpret a higher order effect.
72−-- operations in the context of an effect handler.
83+--
84+-- /Note:/ 'LocalEnv' is for handling local 'Eff' operations using a function
85+-- from the 'localUnlift' family.
73interpretM
86interpretM
74− :: (forall r localEs. HasCallStack => RunIn localEs (Eff es) -> e (Eff localEs) r -> Eff es r)
87+ :: (forall r localEs. HasCallStack => LocalEnv localEs -> e (Eff localEs) r -> Eff es r)
75 -- ^ The effect handler.
88 -- ^ The effect handler.
76 -> Eff (e : es) a
89 -> Eff (e : es) a
77 -> Eff es a
90 -> Eff es a
78interpretM interpreter m = unsafeEff $ \es -> do
91interpretM interpreter m = unsafeEff $ \es -> do
79 les <- forkEnv es
92 les <- forkEnv es
80− runInterpreter es (Interpreter les $ \k -> interpreter $ unsafeEff_ . k) m
93+ runInterpreter es (Interpreter les interpreter) m
81
94
82----------------------------------------
95----------------------------------------
83-- Reinterpretation
96-- Reinterpretation
@@ -95,16 +108,66 @@ reinterpret runHandlerEs interpreter m = unsafeEff $ \es -> do
95 (`unEff` les0) . runHandlerEs . unsafeEff $ \les -> do
108 (`unEff` les0) . runHandlerEs . unsafeEff $ \les -> do
96 runInterpreter es (Interpreter les $ \_ -> interpreter) m
109 runInterpreter es (Interpreter les $ \_ -> interpreter) m
97
110
98−-- | Interpret a higher order effect using other effects with help of a function
111+-- | Interpret a higher order effect using other effects.
99−-- running local 'Eff' operations in the context of an effect handler.
112+--
113+-- /Note:/ 'LocalEnv' is for handling local 'Eff' operations using a function
114+-- from the 'localUnlift' family.
100reinterpretM
115reinterpretM
101 :: (Eff handlerEs a -> Eff es b)
116 :: (Eff handlerEs a -> Eff es b)
102 -- ^ Introduction of effects encapsulated within the handler.
117 -- ^ Introduction of effects encapsulated within the handler.
103− -> (forall r localEs. HasCallStack => RunIn localEs (Eff handlerEs) -> e (Eff localEs) r -> Eff handlerEs r)
118+ -> (forall r localEs. HasCallStack => LocalEnv localEs -> e (Eff localEs) r -> Eff handlerEs r)
104 -- ^ The effect handler.
119 -- ^ The effect handler.
105 -> Eff (e : es) a
120 -> Eff (e : es) a
106 -> Eff es b
121 -> Eff es b
107reinterpretM runHandlerEs interpreter m = unsafeEff $ \es -> do
122reinterpretM runHandlerEs interpreter m = unsafeEff $ \es -> do
108 les0 <- forkEnv es
123 les0 <- forkEnv es
109 (`unEff` les0) . runHandlerEs . unsafeEff $ \les -> do
124 (`unEff` les0) . runHandlerEs . unsafeEff $ \les -> do
110− runInterpreter es (Interpreter les $ \k -> interpreter $ unsafeEff_ . k) m
125+ runInterpreter es (Interpreter les interpreter) m
126+
127+----------------------------------------
128+-- Unlifting
129+
130+-- | Create a local unlifting function with the 'SeqUnlift' strategy.
131+localSeqUnlift
132+ :: LocalEnv localEs
133+ -- ^ Local environment from the effect handler.
134+ -> ((forall r. Eff localEs r -> Eff es r) -> Eff es a)
135+ -- ^ Continuation with the unlifting function.
136+ -> Eff es a
137+localSeqUnlift (LocalEnv les) f = unsafeEff $ \es -> do
138+ unEff (seqUnliftEff $ \k -> unEff (f $ unsafeEff_ . k) es) les
139+
140+-- | Create a local unlifting function with the 'SeqUnlift' strategy.
141+localSeqUnliftIO
142+ :: IOE :> es
143+ => LocalEnv localEs
144+ -- ^ Local environment from the effect handler.
145+ -> ((forall r. Eff localEs r -> IO r) -> IO a)
146+ -- ^ Continuation with the unlifting function.
147+ -> Eff es a
148+localSeqUnliftIO (LocalEnv les) f = unsafeEff_ $ unEff (seqUnliftEff f) les
149+
150+-- | Create a local unlifting function with the given strategy.
151+localUnlift
152+ :: LocalEnv localEs
153+ -- ^ Local environment from the effect handler.
154+ -> UnliftStrategy
155+ -> ((forall r. Eff localEs r -> Eff es r) -> Eff es a)
156+ -- ^ Continuation with the unlifting function.
157+ -> Eff es a
158+localUnlift (LocalEnv les) unlift f = unsafeEff $ \es -> case unlift of
159+ SeqUnlift -> unEff (seqUnliftEff $ \k -> unEff (f $ unsafeEff_ . k) es) les
160+ ConcUnlift p b -> unEff (concUnliftEff p b $ \k -> unEff (f $ unsafeEff_ . k) es) les
161+
162+-- | Create a local unlifting function with the given strategy.
163+localUnliftIO
164+ :: IOE :> es
165+ => LocalEnv localEs
166+ -- ^ Local environment from the effect handler.
167+ -> UnliftStrategy
168+ -> ((forall r. Eff localEs r -> IO r) -> IO a)
169+ -- ^ Continuation with the unlifting function.
170+ -> Eff es a
171+localUnliftIO (LocalEnv les) unlift f = unsafeEff_ $ case unlift of
172+ SeqUnlift -> unEff (seqUnliftEff f) les
173+ ConcUnlift p b -> unEff (concUnliftEff p b f) les
src/Effectful/Monad.hs+7
@@ -10,6 +10,13 @@ module Effectful.Monad
10 -- * Arbitrary I/O
10 -- * Arbitrary I/O
11 , IOE
11 , IOE
12 , runIOE
12 , runIOE
13+
14+ -- ** Unlift strategies
15+ , UnliftStrategy(..)
16+ , Persistence(..)
17+ , Limit(..)
18+ , unliftStrategy
19+ , withUnliftStrategy
13 ) where
20 ) where
14
21
15import Effectful.Internal.Effect
22import Effectful.Internal.Effect
src/Effectful/State/Dynamic.hs+6−6
@@ -62,14 +62,14 @@ execState s0 = reinterpretM (SP.execState s0) statePure
62
62
63statePure
63statePure
64 :: SP.State s :> es
64 :: SP.State s :> es
65− => RunIn localEs (Eff es)
65+ => LocalEnv localEs
66 -> State s (Eff localEs) a
66 -> State s (Eff localEs) a
67 -> Eff es a
67 -> Eff es a
68−statePure run = \case
68+statePure env = \case
69 Get -> SP.get
69 Get -> SP.get
70 Put s -> SP.put s
70 Put s -> SP.put s
71 State f -> SP.state f
71 State f -> SP.state f
72− StateM f -> SP.stateM (run . f)
72+ StateM f -> localSeqUnlift env $ \run -> SP.stateM (run . f)
73
73
74----------------------------------------
74----------------------------------------
75-- MVar
75-- MVar
@@ -85,14 +85,14 @@ execStateMVar s0 = reinterpretM (SM.execState s0) stateMVar
85
85
86stateMVar
86stateMVar
87 :: SM.State s :> es
87 :: SM.State s :> es
88− => (forall r. Eff localEs r -> Eff es r)
88+ => LocalEnv localEs
89 -> State s (Eff localEs) a
89 -> State s (Eff localEs) a
90 -> Eff es a
90 -> Eff es a
91−stateMVar run = \case
91+stateMVar env = \case
92 Get -> SM.get
92 Get -> SM.get
93 Put s -> SM.put s
93 Put s -> SM.put s
94 State f -> SM.state f
94 State f -> SM.state f
95− StateM f -> SM.stateM (run . f)
95+ StateM f -> localSeqUnlift env $ \run -> SM.stateM (run . f)
96
96
97----------------------------------------
97----------------------------------------
98-- Operations
98-- Operations
src/Effectful/Writer/Dynamic.hs+6−6
@@ -35,12 +35,12 @@ execWriter = reinterpretM WP.execWriter writerPure
35
35
36writerPure
36writerPure
37 :: (WP.Writer w :> es, Monoid w)
37 :: (WP.Writer w :> es, Monoid w)
38− => RunIn localEs (Eff es)
38+ => LocalEnv localEs
39 -> Writer w (Eff localEs) a
39 -> Writer w (Eff localEs) a
40 -> Eff es a
40 -> Eff es a
41−writerPure run = \case
41+writerPure env = \case
42 Tell w -> WP.tell w
42 Tell w -> WP.tell w
43− Listen m -> WP.listen (run m)
43+ Listen m -> localSeqUnlift env $ \run -> WP.listen (run m)
44
44
45----------------------------------------
45----------------------------------------
46-- MVar
46-- MVar
@@ -53,12 +53,12 @@ execWriterMVar = reinterpretM WM.execWriter writerMVar
53
53
54writerMVar
54writerMVar
55 :: (WM.Writer w :> es, Monoid w)
55 :: (WM.Writer w :> es, Monoid w)
56− => RunIn localEs (Eff es)
56+ => LocalEnv localEs
57 -> Writer w (Eff localEs) a
57 -> Writer w (Eff localEs) a
58 -> Eff es a
58 -> Eff es a
59−writerMVar run = \case
59+writerMVar env = \case
60 Tell w -> WM.tell w
60 Tell w -> WM.tell w
61− Listen m -> WM.listen (run m)
61+ Listen m -> localSeqUnlift env $ \run -> WM.listen (run m)
62
62
63----------------------------------------
63----------------------------------------
64-- Operations
64-- Operations
tests/ConcurrencyTests.hsnew file+123
@@ -0,0 +1,123 @@
1+module ConcurrencyTests (concurrencyTests) where
2+
3+import Control.Concurrent (threadDelay)
4+import Control.Monad
5+import Control.Monad.IO.Class
6+import Test.Tasty
7+import Test.Tasty.HUnit
8+import UnliftIO
9+import qualified Data.Set as S
10+
11+import Effectful
12+import Effectful.Error
13+import Effectful.State.Dynamic
14+import qualified Utils as U
15+
16+concurrencyTests :: TestTree
17+concurrencyTests = testGroup "Concurrency"
18+ [ testCase "local state" test_localState
19+ , testCase "shared state" test_sharedState
20+ , testCase "error handling" test_errorHandling
21+ , testCase "unlifting several times" test_unliftMany
22+ , testCase "async with unmask" test_asyncWithUnmask
23+ , testCase "pooled workers" test_pooledWorkers
24+ ]
25+
26+test_localState :: Assertion
27+test_localState = runIOE . evalState x $ do
28+ withUnliftStrategy (ConcUnlift Ephemeral $ Limited 2) $ do
29+ replicateConcurrently_ 2 $ do
30+ r <- goDownward 0
31+ U.assertEqual "expected result" x r
32+ where
33+ x :: Int
34+ x = 100000
35+
36+ goDownward :: State Int :> es => Int -> Eff es Int
37+ goDownward acc = do
38+ end <- state @Int $ \case
39+ 0 -> (True, 0)
40+ n -> (False, n - 1)
41+ if end
42+ then pure acc
43+ else goDownward $ acc + 1
44+
45+test_sharedState :: Assertion
46+test_sharedState = runIOE . evalStateMVar (S.empty @Int) $ do
47+ withUnliftStrategy (ConcUnlift Ephemeral $ Limited 2) $ do
48+ concurrently_ (addWhen even x) (addWhen odd x)
49+ U.assertEqual "expected result" (S.fromList [1..x]) =<< get
50+ where
51+ x :: Int
52+ x = 100
53+
54+ addWhen :: State (S.Set Int) :> es => (Int -> Bool) -> Int -> Eff es ()
55+ addWhen f = \case
56+ 0 -> pure ()
57+ n -> do
58+ when (f n) $ do
59+ modify $ S.insert n
60+ addWhen f $ n - 1
61+
62+test_errorHandling :: Assertion
63+test_errorHandling = runIOE . evalStateMVar (0::Int) $ do
64+ withUnliftStrategy (ConcUnlift Ephemeral $ Limited 2) $ do
65+ r <- runError $ concurrently_
66+ (liftIO (threadDelay 10000) >> throwError err)
67+ (modify (+x))
68+ case r of
69+ Left (_, e) -> U.assertEqual "error caught" err e
70+ Right _ -> U.assertFailure "error not caught"
71+ U.assertEqual "state updated" x =<< get
72+ where
73+ x :: Int
74+ x = 67
75+
76+ err :: String
77+ err = "thrown from async"
78+
79+test_unliftMany :: Assertion
80+test_unliftMany = runIOE . evalState "initial value" $ do
81+ withUnliftStrategy (ConcUnlift Persistent $ Limited 1) $ do
82+ x <- withRunInIO $ \runInIO -> async $ do
83+ v1 <- runInIO $ get @String -- 1
84+ threadDelay 20000
85+ v2 <- runInIO $ get @String -- 3
86+ runInIO $ put "inner change"
87+ v3 <- runInIO $ get @String
88+ return (v1, v2, v3)
89+ liftIO $ threadDelay 10000
90+ put "outer change" -- 2
91+ (v1, v2, v3) <- liftIO $ wait x
92+ v4 <- get -- 4
93+ U.assertEqual "expected result"
94+ ("initial value", "initial value", "inner change", "outer change")
95+ (v1, v2, v3, v4)
96+
97+test_asyncWithUnmask :: Assertion
98+test_asyncWithUnmask = runIOE . evalState "initial" $ do
99+ withUnliftStrategy (ConcUnlift Persistent $ Limited 1) $ do
100+ x <- asyncWithUnmask $ \unmask -> do
101+ liftIO $ threadDelay 10000
102+ r1 <- get @String -- 2
103+ unmask $ put "unmask"
104+ r2 <- get @String -- 3
105+ pure (r1, r2)
106+ put "changed" -- 1
107+ (inner1, inner2) <- liftIO $ wait x
108+ outer <- get -- 4
109+ U.assertEqual "expected result"
110+ ("initial", "unmask", "changed")
111+ (inner1, inner2, outer)
112+
113+test_pooledWorkers :: Assertion
114+test_pooledWorkers = runIOE . evalState (0::Int) $ do
115+ withUnliftStrategy (ConcUnlift Ephemeral $ Limited n) $ do
116+ x <- pooledForConcurrentlyN threads [1..n] $ \k -> do
117+ r <- get @Int
118+ modify @Int (+1)
119+ pure $ k + r
120+ U.assertEqual "expected result" [1..n] x
121+ where
122+ n = 10
123+ threads = 4
tests/Main.hs+2
@@ -3,12 +3,14 @@ module Main (main) where
3import Test.Tasty
3import Test.Tasty
4
4
5import AsyncTests
5import AsyncTests
6+import ConcurrencyTests
6import ErrorTests
7import ErrorTests
7import StateTests
8import StateTests
8
9
9main :: IO ()
10main :: IO ()
10main = defaultMain $ testGroup "effectful"
11main = defaultMain $ testGroup "effectful"
11 [ asyncTests
12 [ asyncTests
13+ , concurrencyTests
12 , errorTests
14 , errorTests
13 , stateTests
15 , stateTests
14 ]
16 ]
utils/Effectful/Internal/Utils.hsnew file+30
@@ -0,0 +1,30 @@
1+{-# LANGUAGE CPP #-}
2+{-# LANGUAGE MagicHash #-}
3+{-# LANGUAGE UnliftedFFITypes #-}
4+module Effectful.Internal.Utils
5+ ( weakThreadId
6+ , eqThreadId
7+ ) where
8+
9+import Foreign.C.Types
10+import GHC.Conc.Sync (ThreadId(..))
11+import GHC.Exts (ThreadId#)
12+
13+-- | Get an id of a thread that doesn't prevent its garbage collection.
14+weakThreadId :: ThreadId -> Int
15+weakThreadId (ThreadId t#) = fromIntegral $ rts_getThreadId t#
16+
17+foreign import ccall unsafe "rts_getThreadId"
18+#if __GLASGOW_HASKELL__ >= 900
19+ rts_getThreadId :: ThreadId# -> CLong
20+#else
21+ rts_getThreadId :: ThreadId# -> CInt
22+#endif
23+
24+-- | 'Eq' instance for 'ThreadId' is broken in GHC < 9, see
25+-- https://gitlab.haskell.org/ghc/ghc/-/issues/16761 for more info.
26+eqThreadId :: ThreadId -> ThreadId -> Bool
27+eqThreadId (ThreadId t1#) (ThreadId t2#) = eq_thread t1# t2# == 1
28+
29+foreign import ccall unsafe "effectful_eq_thread"
30+ eq_thread :: ThreadId# -> ThreadId# -> CLong
utils/utils.cnew file+5
@@ -0,0 +1,5 @@
1+// Correct implementation of ThreadId# equality for GHC < 9.
2+long effectful_eq_thread(void *tso1, void *tso2)
3+{
4+ return tso1 == tso2;
5+}