@@ -1,19 +1,33 @@
1+{-# LANGUAGE DeriveGeneric #-}
12{-# LANGUAGE MagicHash #-}
23{-# LANGUAGE UnboxedTuples #-}
34{-# OPTIONS_HADDOCK not-home #-}
4−-- | Implementation of concurrent unlifts.
5+-- | Implementation of sequential and concurrent unlifts.
56--
67-- This module is intended for internal use only, and may change without warning
78-- in subsequent releases.
89module Effectful.Internal.Unlift
9− ( ephemeralConcUnliftIO
10+ ( -- * Unlifting strategies
11+ UnliftStrategy(..)
12+ , Persistence(..)
13+ , Limit(..)
14+
15+ -- * Unlifting functions
16+ , seqUnliftIO
17+ , concUnliftIO
18+ , ephemeralConcUnliftIO
1019 , persistentConcUnliftIO
20+
21+ -- * Unlifting errors
22+ , UnliftError(..)
1123 ) where
1224
1325import Control.Concurrent
26+import Control.Exception
1427import Control.Monad
1528import GHC.Conc.Sync (ThreadId(..))
1629import GHC.Exts (mkWeak#, mkWeakNoFinalizer#)
30+import GHC.Generics (Generic)
1731import GHC.IO (IO(..))
1832import GHC.Stack (HasCallStack)
1933import GHC.Weak (Weak(..))
@@ -23,18 +37,119 @@ import qualified Data.IntMap.Strict as IM
2337import Effectful.Internal.Env
2438import Effectful.Internal.Utils
2539
40+----------------------------------------
41+-- Unlift strategies
42+
43+-- | The strategy to use when unlifting 'Eff' operations via 'withRunInIO' or
44+-- the 'Effectful.Interpreter.localUnlift' family.
45+data UnliftStrategy
46+ = SeqUnlift
47+ -- ^ The fastest strategy and a default setting for 'IOE'. An attempt to call
48+ -- the unlifting function in threads distinct from its creator will result in
49+ -- a runtime error.
50+ | ConcUnlift !Persistence !Limit
51+ -- ^ A strategy that makes it possible for the unlifting function to be called
52+ -- in threads distinct from its creator. See 'Persistence' and 'Limit'
53+ -- settings for more information.
54+ deriving (Eq, Generic, Ord, Show)
55+
56+-- | Persistence setting for the 'ConcUnlift' strategy.
57+--
58+-- Different functions require different persistence strategies. Examples:
59+--
60+-- - Lifting 'pooledMapConcurrentlyN' from the @unliftio@ library requires the
61+-- 'Ephemeral' strategy as we don't want jobs to share environment changes
62+-- made by previous jobs run in the same worker thread.
63+--
64+-- - Lifting 'Control.Concurrent.forkIOWithUnmask' requires the 'Persistent'
65+-- strategy, otherwise the unmasking function would start with a fresh
66+-- environment each time it's called.
67+data Persistence
68+ = Ephemeral
69+ -- ^ Don't persist the environment between calls to the unlifting function in
70+ -- threads distinct from its creator.
71+ | Persistent
72+ -- ^ Persist the environment between calls to the unlifting function within a
73+ -- particular thread.
74+ deriving (Eq, Generic, Ord, Show)
75+
76+-- | Limit setting for the 'ConcUnlift' strategy.
77+data Limit
78+ = Limited !Int
79+ -- ^ Behavior dependent on the 'Persistence' setting.
80+ --
81+ -- For 'Ephemeral', it limits the amount of uses of the unlifting function in
82+ -- threads distinct from its creator to @N@. The unlifting function will
83+ -- create @N@ copies of the environment when called @N@ times and @K+1@ copies
84+ -- when called @K < N@ times.
85+ --
86+ -- For 'Persistent', it limits the amount of threads, distinct from the
87+ -- creator of the unlifting function, it can be called in to @N@. The amount
88+ -- of calls to the unlifting function within a particular threads is
89+ -- unlimited. The unlifting function will create @N@ copies of the environment
90+ -- when called in @N@ threads and @K+1@ copies when called in @K < N@ threads.
91+ | Unlimited
92+ -- ^ Unlimited use of the unlifting function.
93+ deriving (Eq, Generic, Ord, Show)
94+
95+----------------------------------------
96+-- Unlift functions
97+
98+-- | Sequential unlift.
99+--
100+-- Exceptions thrown by this function:
101+--
102+-- - 'InvalidUseOfSeqUnlift' if the unlift function is used in another thread.
103+seqUnliftIO
104+ :: HasCallStack
105+ => ((forall r. m r -> IO r) -> IO a)
106+ -> Env es
107+ -> (forall r. m r -> Env es -> IO r)
108+ -> IO a
109+seqUnliftIO k es unEff = do
110+ tid0 <- myThreadId
111+ k $ \m -> do
112+ tid <- myThreadId
113+ if tid `eqThreadId` tid0
114+ then unEff m es
115+ else throwIO InvalidUseOfSeqUnlift
116+
117+-- | Concurrent unlift for various strategies and limits.
118+concUnliftIO
119+ :: HasCallStack
120+ => Persistence
121+ -> Limit
122+ -> ((forall r. m r -> IO r) -> IO a)
123+ -> Env es
124+ -> (forall r. m r -> Env es -> IO r)
125+ -> IO a
126+concUnliftIO Ephemeral (Limited uses) k =
127+ ephemeralConcUnliftIO uses k
128+concUnliftIO Ephemeral Unlimited k =
129+ ephemeralConcUnliftIO maxBound k
130+concUnliftIO Persistent (Limited threads) k =
131+ persistentConcUnliftIO False threads k
132+concUnliftIO Persistent Unlimited k =
133+ persistentConcUnliftIO True maxBound k
134+
26135-- | Concurrent unlift that doesn't preserve the environment between calls to
27136-- the unlifting function in threads other than its creator.
137+--
138+-- Exceptions thrown by this function:
139+--
140+-- - 'InvalidNumberOfUses' if the number of uses is less or equal zero.
141+-- - 'ExceededNumberOfUses' if the unlift function is called too often.
28142ephemeralConcUnliftIO
29143 :: HasCallStack
30144 => Int
145+ -- ^ Number of permitted uses of the unlift function.
31146 -> ((forall r. m r -> IO r) -> IO a)
32147 -> Env es
33148 -> (forall r. m r -> Env es -> IO r)
34149 -> IO a
35150ephemeralConcUnliftIO uses k es0 unEff = do
36151 unless (uses > 0) $ do
37− error $ "Invalid number of uses: " ++ show uses
152+ throwIO $ InvalidNumberOfUses uses
38153 tid0 <- myThreadId
39154 -- Create a copy of the environment as a template for the other threads to
40155 -- use. This can't be done from inside the callback as the environment might
@@ -45,10 +160,7 @@ ephemeralConcUnliftIO uses k es0 unEff = do
45160 es <- myThreadId >>= \case
46161 tid | tid0 `eqThreadId` tid -> pure es0
47162 _ -> modifyMVar mvUses $ \case
48− 0 -> error
49− $ "Number of permitted calls (" ++ show uses ++ ") to the unlifting "
50− ++ "function in other threads was exceeded. Please increase the limit "
51− ++ "or use the unlimited variant."
163+ 0 -> throwIO $ ExceededNumberOfUses uses
52164 1 -> pure (0, esTemplate)
53165 n -> do
54166 let newUses = n - 1
@@ -58,17 +170,24 @@ ephemeralConcUnliftIO uses k es0 unEff = do
58170
59171-- | Concurrent unlift that preserves the environment between calls to the
60172-- unlifting function within a particular thread.
173+--
174+-- Exceptions thrown by this function:
175+--
176+-- - 'InvalidNumberOfThreads' if the number of threads is less or equal zero.
177+-- - 'ExceededNumberOfThreads' if the unlift function is called by too many
178+-- threads.
61179persistentConcUnliftIO
62180 :: HasCallStack
63181 => Bool
64182 -> Int
183+ -- ^ Number of threads that are allowed to use the unlift function.
65184 -> ((forall r. m r -> IO r) -> IO a)
66185 -> Env es
67186 -> (forall r. m r -> Env es -> IO r)
68187 -> IO a
69188persistentConcUnliftIO cleanUp threads k es0 unEff = do
70189 unless (threads > 0) $ do
71− error $ "Invalid number of threads: " ++ show threads
190+ throwIO $ InvalidNumberOfThreads threads
72191 tid0 <- myThreadId
73192 -- Create a copy of the environment as a template for the other threads to
74193 -- use. This can't be done from inside the callback as the environment might
@@ -86,10 +205,7 @@ persistentConcUnliftIO cleanUp threads k es0 unEff = do
86205 case mes of
87206 Just es -> pure (te, es)
88207 Nothing -> case teCapacity te of
89− 0 -> error
90− $ "Number of other threads (" ++ show threads ++ ") permitted to "
91− ++ "use the unlifting function was exceeded. Please increase the "
92− ++ "limit or use the unlimited variant."
208+ 0 -> throwIO $ ExceededNumberOfThreads threads
93209 1 -> do
94210 wkTidEs <- mkWeakThreadIdEnv tid esTemplate wkTid i mvEntries cleanUp
95211 let newEntries = ThreadEntries
@@ -107,6 +223,38 @@ persistentConcUnliftIO cleanUp threads k es0 unEff = do
107223 newEntries `seq` pure (newEntries, es)
108224 unEff m es
109225
226+----------------------------------------
227+-- Errors
228+
229+-- | An type used for errors occuring in unlifting functions.
230+--
231+-- Use the 'displayException' method of the 'Exception' type class to get
232+-- descriptive error messages.
233+data UnliftError
234+ = InvalidNumberOfThreads !Int
235+ | InvalidNumberOfUses !Int
236+ | InvalidUseOfSeqUnlift
237+ | ExceededNumberOfThreads !Int
238+ | ExceededNumberOfUses !Int
239+ deriving (Eq, Generic, Show)
240+
241+instance Exception UnliftError where
242+ displayException (InvalidNumberOfThreads threads) =
243+ "Invalid number of threads: " ++ show threads
244+ displayException (InvalidNumberOfUses uses) =
245+ "Invalid number of uses: " ++ show uses
246+ displayException (ExceededNumberOfThreads threads) =
247+ "Number of other threads (" ++ show threads ++ ") permitted to use the " ++
248+ "unlifting function was exceeded. Please increase the limit or use the " ++
249+ "unlimited variant."
250+ displayException (ExceededNumberOfUses uses) =
251+ "Number of permitted calls (" ++ show uses ++ ") to the unlifting " ++
252+ "function in other threads was exceeded. Please increase the limit or " ++
253+ "use the unlimited variant."
254+ displayException InvalidUseOfSeqUnlift =
255+ "If you want to use the unlifting function to run Eff operations in " ++
256+ "multiple threads, have a look at UnliftStrategy (ConcUnlift)."
257+
110258----------------------------------------
111259-- Data types
112260