git.iko.soy/effectful
on 01bdb2a help sign in
effectful/src/Effectful/Async.hs
366 lines 11 kB history raw
1{-# LANGUAGE UndecidableInstances #-}
2module Effectful.Async
3 ( -- * Async effect
4 AsyncE
5 , runAsyncE
7 -- * Asynchronous actions
8 , A.Async
9 -- ** Spawning
10 , async, asyncBound, asyncOn
11 , asyncWithUnmask, asyncOnWithUnmask
13 -- ** Spawning with automatic 'cancel'ation
14 , withAsync, withAsyncBound, withAsyncOn
15 , withAsyncWithUnmask, withAsyncOnWithUnmask
17 -- ** Quering 'Async's
18 , wait, poll, waitCatch
19 , cancel
20 , uninterruptibleCancel
21 , cancelWith
22 , A.asyncThreadId
23 , A.AsyncCancelled(..)
25 -- ** STM operations
26 , A.waitSTM, A.pollSTM, A.waitCatchSTM
28 -- ** Waiting for multiple 'Async's
29 , waitAny, waitAnyCatch, waitAnyCancel, waitAnyCatchCancel
30 , waitEither, waitEitherCatch, waitEitherCancel, waitEitherCatchCancel
31 , waitEither_
32 , waitBoth
34 -- ** Waiting for multiple 'Async's in STM
35 , A.waitAnySTM
36 , A.waitAnyCatchSTM
37 , A.waitEitherSTM
38 , A.waitEitherCatchSTM
39 , A.waitEitherSTM_
40 , A.waitBothSTM
42 -- ** Linking
43 , link, link2
44 , A.ExceptionInLinkedThread(..)
46 -- * Convenient utilities
47 , race, race_, concurrently, concurrently_
48 , mapConcurrently, mapConcurrently_
49 , forConcurrently, forConcurrently_
50 , replicateConcurrently, replicateConcurrently_
51 , Concurrently(..)
53 , A.compareAsyncs
54 ) where
56import Control.Applicative
57import Control.Concurrent (threadDelay)
58import Control.Concurrent.Async (Async)
59import Control.Exception (Exception, SomeException)
60import Control.Monad (forever, void)
61import Data.Foldable (fold)
62import qualified Control.Concurrent.Async as A
64import Effectful.Internal.Effect
65import Effectful.Internal.Env
66import Effectful.Internal.Monad
68-- | Run 'Eff' computations asynchronously using the @async@ library.
69--
70-- /Note:/ thread local state changes in 'Eff' computations lifted into an
71-- 'Async' will not affect the parent thread.
72data AsyncE :: Effect where
73 AsyncE :: AsyncE m r
75runAsyncE :: IOE :> es => Eff (AsyncE : es) a -> Eff es a
76runAsyncE = evalEffect (IdE AsyncE)
78-- | Generalized version of 'A.async'.
79async :: AsyncE :> es => Eff es a -> Eff es (Async a)
80async = liftAsync A.async
82-- | Generalized version of 'A.asyncBound'.
83asyncBound :: AsyncE :> es => Eff es a -> Eff es (Async a)
84asyncBound = liftAsync A.asyncBound
86-- | Generalized version of 'A.asyncOn'.
87asyncOn :: AsyncE :> es => Int -> Eff es a -> Eff es (Async a)
88asyncOn cpu = liftAsync (A.asyncOn cpu)
90-- | Generalized version of 'A.asyncWithUnmask'.
91asyncWithUnmask
92 :: AsyncE :> es
93 => ((forall b. Eff es b -> Eff es b) -> Eff es a)
94 -> Eff es (Async a)
95asyncWithUnmask = liftAsyncWithUnmask A.asyncWithUnmask
97-- | Generalized version of 'A.asyncOnWithUnmask'.
98asyncOnWithUnmask
99 :: AsyncE :> es
100 => Int
101 -> ((forall b. Eff es b -> Eff es b) -> Eff es a)
102 -> Eff es (Async a)
103asyncOnWithUnmask cpu = liftAsyncWithUnmask (A.asyncOnWithUnmask cpu)
105-- | Generalized version of 'A.withAsync'.
106withAsync
107 :: AsyncE :> es
108 => Eff es a
109 -> (Async a -> Eff es b)
110 -> Eff es b
111withAsync = liftWithAsync A.withAsync
113-- | Generalized version of 'A.withAsyncBound'.
114withAsyncBound
115 :: AsyncE :> es
116 => Eff es a
117 -> (Async a -> Eff es b)
118 -> Eff es b
119withAsyncBound = liftWithAsync A.withAsyncBound
121-- | Generalized version of 'A.withAsyncOn'.
122withAsyncOn
123 :: AsyncE :> es
124 => Int
125 -> Eff es a
126 -> (Async a -> Eff es b)
127 -> Eff es b
128withAsyncOn cpu = liftWithAsync (A.withAsyncOn cpu)
130-- | Generalized version of 'A.withAsyncWithUnmask'.
131withAsyncWithUnmask
132 :: AsyncE :> es
133 => ((forall c. Eff es c -> Eff es c) -> Eff es a)
134 -> (Async a -> Eff es b)
135 -> Eff es b
136withAsyncWithUnmask = liftWithAsyncWithUnmask A.withAsyncWithUnmask
138-- | Generalized version of 'A.withAsyncOnWithUnmask'.
139withAsyncOnWithUnmask
140 :: AsyncE :> es
141 => Int
142 -> ((forall c. Eff es c -> Eff es c) -> Eff es a)
143 -> (Async a -> Eff es b)
144 -> Eff es b
145withAsyncOnWithUnmask cpu = liftWithAsyncWithUnmask (A.withAsyncOnWithUnmask cpu)
147-- | Generalized version of 'A.wait'.
148wait :: AsyncE :> es => Async a -> Eff es a
149wait = unsafeEff_ . A.wait
151-- | Generalized version of 'A.poll'.
152poll
153 :: AsyncE :> es
154 => Async a
155 -> Eff es (Maybe (Either SomeException a))
156poll = unsafeEff_ . A.poll
158-- | Generalized version of 'A.cancel'.
159cancel :: AsyncE :> es => Async a -> Eff es ()
160cancel = unsafeEff_ . A.cancel
162-- | Generalized version of 'A.cancelWith'.
163cancelWith :: (Exception e, AsyncE :> es) => Async a -> e -> Eff es ()
164cancelWith a = unsafeEff_ . A.cancelWith a
166-- | Generalized version of 'A.uninterruptibleCancel'.
167uninterruptibleCancel :: AsyncE :> es => Async a -> Eff es ()
168uninterruptibleCancel = unsafeEff_ . A.uninterruptibleCancel
170-- | Generalized version of 'A.waitCatch'.
171waitCatch
172 :: AsyncE :> es
173 => Async a
174 -> Eff es (Either SomeException a)
175waitCatch = unsafeEff_ . A.waitCatch
177-- | Generalized version of 'A.waitAny'.
178waitAny :: AsyncE :> es => [Async a] -> Eff es (Async a, a)
179waitAny = unsafeEff_ . A.waitAny
181-- | Generalized version of 'A.waitAnyCatch'.
182waitAnyCatch
183 :: AsyncE :> es
184 => [Async a]
185 -> Eff es (Async a, Either SomeException a)
186waitAnyCatch = unsafeEff_ . A.waitAnyCatch
188-- | Generalized version of 'A.waitAnyCancel'.
189waitAnyCancel :: AsyncE :> es => [Async a] -> Eff es (Async a, a)
190waitAnyCancel = unsafeEff_ . A.waitAnyCancel
192-- | Generalized version of 'A.waitAnyCatchCancel'.
193waitAnyCatchCancel
194 :: AsyncE :> es
195 => [Async a]
196 -> Eff es (Async a, Either SomeException a)
197waitAnyCatchCancel = unsafeEff_ . A.waitAnyCatchCancel
199-- | Generalized version of 'A.waitEither'.
200waitEither
201 :: AsyncE :> es
202 => Async a
203 -> Async b
204 -> Eff es (Either a b)
205waitEither a b = unsafeEff_ $ A.waitEither a b
207-- | Generalized version of 'A.waitEitherCatch'.
208waitEitherCatch
209 :: AsyncE :> es
210 => Async a
211 -> Async b
212 -> Eff es (Either (Either SomeException a) (Either SomeException b))
213waitEitherCatch a b = unsafeEff_ $ A.waitEitherCatch a b
215-- | Generalized version of 'A.waitEitherCancel'.
216waitEitherCancel
217 :: AsyncE :> es
218 => Async a
219 -> Async b
220 -> Eff es (Either a b)
221waitEitherCancel a b = unsafeEff_ $ A.waitEitherCancel a b
223-- | Generalized version of 'A.waitEitherCatchCancel'.
224waitEitherCatchCancel
225 :: AsyncE :> es
226 => Async a
227 -> Async b
228 -> Eff es (Either (Either SomeException a) (Either SomeException b))
229waitEitherCatchCancel a b = unsafeEff_ $ A.waitEitherCatch a b
231-- | Generalized version of 'A.waitEither_'.
232waitEither_ :: AsyncE :> es => Async a -> Async b -> Eff es ()
233waitEither_ a b = unsafeEff_ $ A.waitEither_ a b
235-- | Generalized version of 'A.waitBoth'.
236waitBoth :: AsyncE :> es => Async a -> Async b -> Eff es (a, b)
237waitBoth a b = unsafeEff_ $ A.waitBoth a b
239-- | Generalized version of 'A.link'.
240link :: AsyncE :> es => Async a -> Eff es ()
241link = unsafeEff_ . A.link
243-- | Generalized version of 'A.link2'.
244link2 :: AsyncE :> es => Async a -> Async b -> Eff es ()
245link2 a b = unsafeEff_ $ A.link2 a b
247-- | Generalized version of 'A.race'.
248race :: AsyncE :> es => Eff es a -> Eff es b -> Eff es (Either a b)
249race ma mb = withAsync ma $ \a -> withAsync mb $ \b -> waitEither a b
251-- | Generalized version of 'A.race_'.
252race_ :: AsyncE :> es => Eff es a -> Eff es b -> Eff es ()
253race_ ma mb = withAsync ma $ \a -> withAsync mb $ \b -> waitEither_ a b
255-- | Generalized version of 'A.concurrently'.
256concurrently :: AsyncE :> es => Eff es a -> Eff es b -> Eff es (a, b)
257concurrently ma mb = withAsync ma $ \a -> withAsync mb $ \b -> waitBoth a b
259-- | Generalized version of 'A.concurrently_'.
260concurrently_ :: AsyncE :> es => Eff es a -> Eff es b -> Eff es ()
261concurrently_ ma mb = void $ concurrently ma mb
263-- | Generalized version of 'A.mapConcurrently'.
264mapConcurrently
265 :: (Traversable f, AsyncE :> es)
266 => (a -> Eff es b)
267 -> f a
268 -> Eff es (f b)
269mapConcurrently f = runConcurrently . traverse (Concurrently . f)
271-- | Generalized version of 'A.mapConcurrently_'.
272mapConcurrently_
273 :: (Foldable f, AsyncE :> es)
274 => (a -> Eff es b)
275 -> f a
276 -> Eff es ()
277mapConcurrently_ f = runConcurrently . foldMap (Concurrently . void . f)
279-- | Generalized version of 'A.forConcurrently'.
280forConcurrently
281 :: (Traversable f, AsyncE :> es)
282 => f a
283 -> (a -> Eff es b)
284 -> Eff es (f b)
285forConcurrently = flip mapConcurrently
287-- | Generalized version of 'A.forConcurrently_'.
288forConcurrently_
289 :: (Foldable f, AsyncE :> es)
290 => f a
291 -> (a -> Eff es b)
292 -> Eff es ()
293forConcurrently_ = flip mapConcurrently_
295-- | Generalized version of 'A.replicateConcurrently'.
296replicateConcurrently :: AsyncE :> es => Int -> Eff es a -> Eff es [a]
297replicateConcurrently n =
298 runConcurrently . sequenceA . replicate n . Concurrently
300-- | Generalized version of 'A.replicateConcurrently_'.
301replicateConcurrently_ :: AsyncE :> es => Int -> Eff es a -> Eff es ()
302replicateConcurrently_ n =
303 runConcurrently . fold . replicate n . Concurrently . void
305----------------------------------------
307-- | Generalized version of 'A.Concurrently'.
308newtype Concurrently es a = Concurrently { runConcurrently :: Eff es a }
310instance Functor (Concurrently es) where
311 fmap f (Concurrently a) = Concurrently (fmap f a)
313instance AsyncE :> es => Applicative (Concurrently es) where
314 pure = Concurrently . pure
315 Concurrently fs <*> Concurrently as =
316 Concurrently ((\(f, a) -> f a) <$> concurrently fs as)
318instance AsyncE :> es => Alternative (Concurrently es) where
319 empty = Concurrently . unsafeEff_ . forever $ threadDelay maxBound
320 Concurrently as <|> Concurrently bs =
321 Concurrently (either id id <$> race as bs)
323instance (AsyncE :> es, Semigroup a) => Semigroup (Concurrently es a) where
324 (<>) = liftA2 (<>)
326instance (AsyncE :> es, Monoid a) => Monoid (Concurrently es a) where
327 mempty = pure mempty
329----------------------------------------
330-- Helpers
332liftAsync
333 :: (IO a -> IO (Async a))
334 -> Eff es a
335 -> Eff es (Async a)
336liftAsync fork action = unsafeEff $ \es0 -> do
337 es <- cloneEnv es0
338 fork $ unEff action es
340liftAsyncWithUnmask
341 :: (((forall b. IO b -> IO b) -> IO a) -> IO (Async a))
342 -> ((forall b. Eff es b -> Eff es b) -> Eff es a)
343 -> Eff es (Async a)
344liftAsyncWithUnmask fork action = unsafeEff $ \es0 -> do
345 es <- cloneEnv es0
346 fork $ \unmask -> unEff (action $ \m -> unsafeEff $ unmask . unEff m) es
348liftWithAsync
349 :: (IO a -> (Async a -> IO b) -> IO b)
350 -> Eff es a
351 -> (Async a -> Eff es b)
352 -> Eff es b
353liftWithAsync withA action k = unsafeEff $ \es0 -> do
354 es <- cloneEnv es0
355 withA (unEff action es)
356 (\a -> unEff (k a) es0)
358liftWithAsyncWithUnmask
359 :: (((forall c. IO c -> IO c) -> IO a) -> (Async a -> IO b) -> IO b)
360 -> ((forall c. Eff es c -> Eff es c) -> Eff es a)
361 -> (Async a -> Eff es b)
362 -> Eff es b
363liftWithAsyncWithUnmask withA action k = unsafeEff $ \es0 -> do
364 es <- cloneEnv es0
365 withA (\unmask -> unEff (action $ \m -> unsafeEff $ unmask . unEff m) es)
366 (\a -> unEff (k a) es0)