@@ -11,11 +11,9 @@ module Effectful.Internal.Monad
1111 , runEff
1212
1313 -- ** Access to the internal representation
14− , RunIn
1514 , unEff
1615 , unsafeEff
1716 , unsafeEff_
18− , unsafeUnliftEff
1917
2018 -- * Fail
2119 , Fail
@@ -25,7 +23,18 @@ module Effectful.Internal.Monad
2523 , IOE
2624 , runIOE
2725
28− -- * Low-level helpers
26+ -- ** Unlift strategies
27+ , UnliftStrategy(..)
28+ , Persistence(..)
29+ , Limit(..)
30+ , unliftStrategy
31+ , withUnliftStrategy
32+
33+ --- *** Low-level helpers
34+ , seqUnliftEff
35+ , concUnliftEff
36+
37+ -- * Primitive effects
2938
3039 -- ** Running
3140 , runEffect
@@ -50,12 +59,14 @@ import Control.Monad.IO.Class
5059import Control.Monad.IO.Unlift
5160import Control.Monad.Trans.Control
5261import Data.Unique
53−import GHC.Magic (oneShot)
62+import GHC.Exts (oneShot)
5463import System.IO.Unsafe (unsafeDupablePerformIO)
5564import qualified Control.Monad.Catch as E
5665
5766import Effectful.Internal.Effect
5867import Effectful.Internal.Env
68+import Effectful.Internal.Unlift
69+import Effectful.Internal.Utils
5970
6071type role Eff nominal representational
6172
@@ -113,40 +124,32 @@ unsafeEff m = Eff (oneShot m)
113124unsafeEff_ :: IO a -> Eff es a
114125unsafeEff_ m = unsafeEff $ \_ -> m
115126
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'.
121−--
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
127+-- | Lower 'Eff' operations into 'IO' ('SeqUnlift').
128+seqUnliftEff :: ((forall r. Eff es r -> IO r) -> IO a) -> Eff es a
129+seqUnliftEff f = unsafeEff $ \es -> do
143130 tid0 <- myThreadId
144131 f $ \m -> do
145132 tid <- myThreadId
146− if tid == tid0
133+ if tid `eqThreadId` tid0
147134 then unEff m es
148− else error $ "Running Eff operations in a different thread is "
149− ++ "currently not possible with the unlifting function"
135+ else error
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
150153
151154----------------------------------------
152155-- Base
@@ -246,14 +249,14 @@ tryFailIO tag m =
246249
247250-- | Run arbitrary 'IO' operations via 'MonadIO', 'MonadBaseControl' 'IO' or
248251-- 'MonadUnliftIO'.
249−data IOE :: Effect where
250− IOE :: IOE m r
252+newtype IOE :: Effect where
253+ IOE :: UnliftStrategy -> IOE m r
251254
252255-- | Run an 'Eff' operation with side effects.
253256--
254257-- For the pure version see 'runEff'.
255258runIOE :: Eff '[IOE] a -> IO a
256−runIOE m = unEff (evalEffect (IdE IOE) m) =<< emptyEnv
259+runIOE m = unEff (evalEffect (IdE (IOE SeqUnlift)) m) =<< emptyEnv
257260
258261instance IOE :> es => MonadIO (Eff es) where
259262 liftIO = unsafeEff_
@@ -263,11 +266,82 @@ instance IOE :> es => MonadBase IO (Eff es) where
263266
264267instance IOE :> es => MonadBaseControl IO (Eff es) where
265268 type StM (Eff es) a = a
266− liftBaseWith = unsafeUnliftEff
269+ liftBaseWith = unliftEff
267270 restoreM = pure
268271
269272instance 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
271345
272346----------------------------------------
273347-- Helpers
@@ -317,8 +391,8 @@ readerEffectM f = unsafeEff $ \es -> getEnv es >>= \e -> unEff (f e) es
317391{-# INLINE readerEffectM #-}
318392
319393stateEffectM :: e :> es => (i e -> Eff es (a, i e)) -> Eff es a
320−stateEffectM f = unsafeEff $ \es -> do
321− (a, e) <- (\e -> unEff (f e) es) =<< getEnv es
394+stateEffectM f = unsafeEff $ \es -> mask $ \release -> do
395+ (a, e) <- (\e -> release $ unEff (f e) es) =<< getEnv es
322396 unsafePutEnv e es
323397 pure a
324398{-# INLINE stateEffectM #-}