git.iko.soy/effectful
on 043bffe help sign in
effectful/src/Effectful/Async.hs
558 lines 16 kB history raw
1{-# LANGUAGE UndecidableInstances #-}
2module Effectful.Async
3 ( -- * Async effect
4 AsyncE
5 , runAsyncE
7 -- * Asynchronous actions
8 , Async
10 -- * High-level API
12 -- ** Spawning with automatic 'cancel'ation
13 , withAsync, withAsyncBound, withAsyncOn, withAsyncWithUnmask
14 , withAsyncOnWithUnmask
16 -- ** Querying 'Async's
17 , wait, poll, waitCatch, A.asyncThreadId
18 , cancel, uninterruptibleCancel, cancelWith, A.AsyncCancelled(..)
19 , A.compareAsyncs
21 -- ** High-level utilities
22 , race, race_
23 , concurrently, concurrently_
24 , mapConcurrently, forConcurrently
25 , mapConcurrently_, forConcurrently_
26 , replicateConcurrently, replicateConcurrently_
28 -- *** Concurrently
29 , Concurrently(..)
31 -- *** Conc
32 , Conc, conc, runConc, U.ConcException(..)
34 -- ** Pooled concurrency
35 , pooledMapConcurrentlyN
36 , pooledMapConcurrently
37 , pooledMapConcurrentlyN_
38 , pooledMapConcurrently_
39 , pooledForConcurrentlyN
40 , pooledForConcurrently
41 , pooledForConcurrentlyN_
42 , pooledForConcurrently_
43 , pooledReplicateConcurrentlyN
44 , pooledReplicateConcurrently
45 , pooledReplicateConcurrentlyN_
46 , pooledReplicateConcurrently_
48 -- ** Specialised operations
50 -- *** STM operations
51 , A.waitSTM, A.pollSTM, A.waitCatchSTM
53 -- *** Waiting for multiple 'Async's
54 , waitAny, waitAnyCatch, waitAnyCancel, waitAnyCatchCancel
55 , waitEither, waitEitherCatch, waitEitherCancel, waitEitherCatchCancel
56 , waitEither_
57 , waitBoth
59 -- *** Waiting for multiple 'Async's in STM
60 , A.waitAnySTM, A.waitAnyCatchSTM
61 , A.waitEitherSTM, A.waitEitherCatchSTM
62 , A.waitEitherSTM_
63 , A.waitBothSTM
65 -- * Low-level API
67 -- ** Spawning (low-level API)
68 , async, asyncBound, asyncOn, asyncWithUnmask, asyncOnWithUnmask
70 -- ** Linking
71 , link, linkOnly, link2, link2Only, A.ExceptionInLinkedThread(..)
72 ) where
74import Control.Applicative
75import Control.Concurrent (threadDelay)
76import Control.Concurrent.Async (Async)
77import Control.Exception (Exception, SomeException)
78import Control.Monad (forever)
79import qualified Control.Concurrent.Async as A
80import qualified UnliftIO.Async as U
81import qualified UnliftIO.Internals.Async as I
83import Effectful.Internal.Effect
84import Effectful.Internal.Env
85import Effectful.Internal.Monad
87-- | Run 'Eff' operations asynchronously via the @async@ library.
88--
89-- /Note:/ thread local state changes in 'Eff' operations run asynchronously
90-- will not affect the parent thread.
91data AsyncE :: Effect where
92 AsyncE :: AsyncE m r
94runAsyncE :: IOE :> es => Eff (AsyncE : es) a -> Eff es a
95runAsyncE = evalEffect (IdE AsyncE)
97----------------------------------------
98-- Operations
100-- | Lifted 'A.async'.
101async :: AsyncE :> es => Eff es a -> Eff es (Async a)
102async = liftAsync A.async
104-- | Lifted 'A.asyncBound'.
105asyncBound :: AsyncE :> es => Eff es a -> Eff es (Async a)
106asyncBound = liftAsync A.asyncBound
108-- | Lifted 'A.asyncOn'.
109asyncOn :: AsyncE :> es => Int -> Eff es a -> Eff es (Async a)
110asyncOn cpu = liftAsync (A.asyncOn cpu)
112-- | Lifted 'A.asyncWithUnmask'.
113asyncWithUnmask
114 :: AsyncE :> es
115 => ((forall b. Eff es b -> Eff es b) -> Eff es a)
116 -> Eff es (Async a)
117asyncWithUnmask = liftAsyncWithUnmask A.asyncWithUnmask
119-- | Lifted 'A.asyncOnWithUnmask'.
120asyncOnWithUnmask
121 :: AsyncE :> es
122 => Int
123 -> ((forall b. Eff es b -> Eff es b) -> Eff es a)
124 -> Eff es (Async a)
125asyncOnWithUnmask cpu = liftAsyncWithUnmask (A.asyncOnWithUnmask cpu)
127-- | Lifted 'A.withAsync'.
128withAsync
129 :: AsyncE :> es
130 => Eff es a
131 -> (Async a -> Eff es b)
132 -> Eff es b
133withAsync = liftWithAsync A.withAsync
135-- | Lifted 'A.withAsyncBound'.
136withAsyncBound
137 :: AsyncE :> es
138 => Eff es a
139 -> (Async a -> Eff es b)
140 -> Eff es b
141withAsyncBound = liftWithAsync A.withAsyncBound
143-- | Lifted 'A.withAsyncOn'.
144withAsyncOn
145 :: AsyncE :> es
146 => Int
147 -> Eff es a
148 -> (Async a -> Eff es b)
149 -> Eff es b
150withAsyncOn cpu = liftWithAsync (A.withAsyncOn cpu)
152-- | Lifted 'A.withAsyncWithUnmask'.
153withAsyncWithUnmask
154 :: AsyncE :> es
155 => ((forall c. Eff es c -> Eff es c) -> Eff es a)
156 -> (Async a -> Eff es b)
157 -> Eff es b
158withAsyncWithUnmask = liftWithAsyncWithUnmask A.withAsyncWithUnmask
160-- | Lifted 'A.withAsyncOnWithUnmask'.
161withAsyncOnWithUnmask
162 :: AsyncE :> es
163 => Int
164 -> ((forall c. Eff es c -> Eff es c) -> Eff es a)
165 -> (Async a -> Eff es b)
166 -> Eff es b
167withAsyncOnWithUnmask cpu = liftWithAsyncWithUnmask (A.withAsyncOnWithUnmask cpu)
169-- | Lifted 'A.wait'.
170wait :: AsyncE :> es => Async a -> Eff es a
171wait = unsafeEff_ . A.wait
173-- | Lifted 'A.poll'.
174poll
175 :: AsyncE :> es
176 => Async a
177 -> Eff es (Maybe (Either SomeException a))
178poll = unsafeEff_ . A.poll
180-- | Lifted 'A.cancel'.
181cancel :: AsyncE :> es => Async a -> Eff es ()
182cancel = unsafeEff_ . A.cancel
184-- | Lifted 'A.cancelWith'.
185cancelWith :: (Exception e, AsyncE :> es) => Async a -> e -> Eff es ()
186cancelWith a = unsafeEff_ . A.cancelWith a
188-- | Lifted 'A.uninterruptibleCancel'.
189uninterruptibleCancel :: AsyncE :> es => Async a -> Eff es ()
190uninterruptibleCancel = unsafeEff_ . A.uninterruptibleCancel
192-- | Lifted 'A.waitCatch'.
193waitCatch
194 :: AsyncE :> es
195 => Async a
196 -> Eff es (Either SomeException a)
197waitCatch = unsafeEff_ . A.waitCatch
199-- | Lifted 'A.waitAny'.
200waitAny :: AsyncE :> es => [Async a] -> Eff es (Async a, a)
201waitAny = unsafeEff_ . A.waitAny
203-- | Lifted 'A.waitAnyCatch'.
204waitAnyCatch
205 :: AsyncE :> es
206 => [Async a]
207 -> Eff es (Async a, Either SomeException a)
208waitAnyCatch = unsafeEff_ . A.waitAnyCatch
210-- | Lifted 'A.waitAnyCancel'.
211waitAnyCancel :: AsyncE :> es => [Async a] -> Eff es (Async a, a)
212waitAnyCancel = unsafeEff_ . A.waitAnyCancel
214-- | Lifted 'A.waitAnyCatchCancel'.
215waitAnyCatchCancel
216 :: AsyncE :> es
217 => [Async a]
218 -> Eff es (Async a, Either SomeException a)
219waitAnyCatchCancel = unsafeEff_ . A.waitAnyCatchCancel
221-- | Lifted 'A.waitEither'.
222waitEither
223 :: AsyncE :> es
224 => Async a
225 -> Async b
226 -> Eff es (Either a b)
227waitEither a b = unsafeEff_ $ A.waitEither a b
229-- | Lifted 'A.waitEitherCatch'.
230waitEitherCatch
231 :: AsyncE :> es
232 => Async a
233 -> Async b
234 -> Eff es (Either (Either SomeException a) (Either SomeException b))
235waitEitherCatch a b = unsafeEff_ $ A.waitEitherCatch a b
237-- | Lifted 'A.waitEitherCancel'.
238waitEitherCancel
239 :: AsyncE :> es
240 => Async a
241 -> Async b
242 -> Eff es (Either a b)
243waitEitherCancel a b = unsafeEff_ $ A.waitEitherCancel a b
245-- | Lifted 'A.waitEitherCatchCancel'.
246waitEitherCatchCancel
247 :: AsyncE :> es
248 => Async a
249 -> Async b
250 -> Eff es (Either (Either SomeException a) (Either SomeException b))
251waitEitherCatchCancel a b = unsafeEff_ $ A.waitEitherCatch a b
253-- | Lifted 'A.waitEither_'.
254waitEither_ :: AsyncE :> es => Async a -> Async b -> Eff es ()
255waitEither_ a b = unsafeEff_ $ A.waitEither_ a b
257-- | Lifted 'A.waitBoth'.
258waitBoth :: AsyncE :> es => Async a -> Async b -> Eff es (a, b)
259waitBoth a b = unsafeEff_ $ A.waitBoth a b
261-- | Lifted 'A.link'.
262link :: AsyncE :> es => Async a -> Eff es ()
263link = unsafeEff_ . A.link
265-- | Lifted 'A.linkOnly'.
266linkOnly :: AsyncE :> es => (SomeException -> Bool) -> Async a -> Eff es ()
267linkOnly f = unsafeEff_ . A.linkOnly f
269-- | Lifted 'A.link2'.
270link2 :: AsyncE :> es => Async a -> Async b -> Eff es ()
271link2 a b = unsafeEff_ $ A.link2 a b
273-- | Lifted 'A.link2Only'.
274link2Only :: AsyncE :> es => (SomeException -> Bool) -> Async a -> Async b -> Eff es ()
275link2Only f a b = unsafeEff_ $ A.link2Only f a b
277-- | Lifted 'A.race'.
278race :: AsyncE :> es => Eff es a -> Eff es b -> Eff es (Either a b)
279race ma mb = unsafeEff $ \es -> do
280 A.race (unEff ma =<< cloneEnv es) (unEff mb =<< cloneEnv es)
282-- | Lifted 'A.race_'.
283race_ :: AsyncE :> es => Eff es a -> Eff es b -> Eff es ()
284race_ ma mb = unsafeEff $ \es -> do
285 A.race_ (unEff ma =<< cloneEnv es) (unEff mb =<< cloneEnv es)
287-- | Lifted 'A.concurrently'.
288concurrently :: AsyncE :> es => Eff es a -> Eff es b -> Eff es (a, b)
289concurrently ma mb = unsafeEff $ \es -> do
290 A.concurrently (unEff ma =<< cloneEnv es) (unEff mb =<< cloneEnv es)
292-- | Lifted 'A.concurrently_'.
293concurrently_ :: AsyncE :> es => Eff es a -> Eff es b -> Eff es ()
294concurrently_ ma mb = unsafeEff $ \es -> do
295 A.concurrently_ (unEff ma =<< cloneEnv es) (unEff mb =<< cloneEnv es)
297-- Below functions use variants from the @unliftio@ library as they minimize the
298-- amount of spawned threads and are thus much more efficient than the ones from
299-- the @async@ library.
301-- | Lifted 'A.mapConcurrently'.
302mapConcurrently
303 :: (Traversable f, AsyncE :> es)
304 => (a -> Eff es b)
305 -> f a
306 -> Eff es (f b)
307mapConcurrently f t = unsafeEff $ \es -> do
308 U.mapConcurrently (\a -> unEff (f a) =<< cloneEnv es) t
310-- | Lifted 'A.mapConcurrently_'.
311mapConcurrently_
312 :: (Foldable f, AsyncE :> es)
313 => (a -> Eff es b)
314 -> f a
315 -> Eff es ()
316mapConcurrently_ f t = unsafeEff $ \es -> do
317 U.mapConcurrently_ (\a -> unEff (f a) =<< cloneEnv es) t
319-- | Lifted 'A.forConcurrently'.
320forConcurrently
321 :: (Traversable f, AsyncE :> es)
322 => f a
323 -> (a -> Eff es b)
324 -> Eff es (f b)
325forConcurrently t f = unsafeEff $ \es -> do
326 U.forConcurrently t (\a -> unEff (f a) =<< cloneEnv es)
328-- | Lifted 'A.forConcurrently_'.
329forConcurrently_
330 :: (Foldable f, AsyncE :> es)
331 => f a
332 -> (a -> Eff es b)
333 -> Eff es ()
334forConcurrently_ t f = unsafeEff $ \es -> do
335 U.forConcurrently_ t (\a -> unEff (f a) =<< cloneEnv es)
337-- | Lifted 'A.replicateConcurrently'.
338replicateConcurrently :: AsyncE :> es => Int -> Eff es a -> Eff es [a]
339replicateConcurrently n f = unsafeEff $ \es -> do
340 U.replicateConcurrently n (unEff f =<< cloneEnv es)
342-- | Lifted 'A.replicateConcurrently_'.
343replicateConcurrently_ :: AsyncE :> es => Int -> Eff es a -> Eff es ()
344replicateConcurrently_ n f = unsafeEff $ \es -> do
345 U.replicateConcurrently_ n (unEff f =<< cloneEnv es)
347----------------------------------------
348-- Pooled concurrency (unliftio)
350-- | Lifted 'U.pooledMapConcurrentlyN'.
351pooledMapConcurrentlyN
352 :: (AsyncE :> es, Traversable t)
353 => Int
354 -> (a -> Eff es b)
355 -> t a
356 -> Eff es (t b)
357pooledMapConcurrentlyN threads f t = unsafeEff $ \es -> do
358 U.pooledMapConcurrentlyN threads (\a -> unEff (f a) =<< cloneEnv es) t
360-- | Lifted 'U.pooledMapConcurrently'.
361pooledMapConcurrently
362 :: (AsyncE :> es, Traversable t)
363 => (a -> Eff es b)
364 -> t a
365 -> Eff es (t b)
366pooledMapConcurrently f t = unsafeEff $ \es -> do
367 U.pooledMapConcurrently (\a -> unEff (f a) =<< cloneEnv es) t
369-- | Lifted 'U.pooledMapConcurrentlyN'.
370pooledMapConcurrentlyN_
371 :: (AsyncE :> es, Foldable f)
372 => Int
373 -> (a -> Eff es b)
374 -> f a
375 -> Eff es ()
376pooledMapConcurrentlyN_ threads f t = unsafeEff $ \es -> do
377 U.pooledMapConcurrentlyN_ threads (\a -> unEff (f a) =<< cloneEnv es) t
379-- | Lifted 'U.pooledMapConcurrently_'.
380pooledMapConcurrently_
381 :: (AsyncE :> es, Foldable f)
382 => (a -> Eff es b)
383 -> f a
384 -> Eff es ()
385pooledMapConcurrently_ f t = unsafeEff $ \es -> do
386 U.pooledMapConcurrently_ (\a -> unEff (f a) =<< cloneEnv es) t
388-- | Lifted 'U.pooledForConcurrentlyN'.
389pooledForConcurrentlyN
390 :: (AsyncE :> es, Traversable t)
391 => Int
392 -> t a
393 -> (a -> Eff es b)
394 -> Eff es (t b)
395pooledForConcurrentlyN threads t f = unsafeEff $ \es -> do
396 U.pooledForConcurrentlyN threads t (\a -> unEff (f a) =<< cloneEnv es)
398-- | Lifted 'U.pooledForConcurrently'.
399pooledForConcurrently
400 :: (AsyncE :> es, Traversable t)
401 => t a
402 -> (a -> Eff es b)
403 -> Eff es (t b)
404pooledForConcurrently t f = unsafeEff $ \es -> do
405 U.pooledForConcurrently t (\a -> unEff (f a) =<< cloneEnv es)
407-- | Lifted 'U.pooledForConcurrentlyN'.
408pooledForConcurrentlyN_
409 :: (AsyncE :> es, Foldable f)
410 => Int
411 -> f a
412 -> (a -> Eff es b)
413 -> Eff es ()
414pooledForConcurrentlyN_ threads t f = unsafeEff $ \es -> do
415 U.pooledForConcurrentlyN_ threads t (\a -> unEff (f a) =<< cloneEnv es)
417-- | Lifted 'U.pooledForConcurrently_'.
418pooledForConcurrently_
419 :: (AsyncE :> es, Foldable f)
420 => f a
421 -> (a -> Eff es b)
422 -> Eff es ()
423pooledForConcurrently_ t f = unsafeEff $ \es -> do
424 U.pooledForConcurrently_ t (\a -> unEff (f a) =<< cloneEnv es)
426-- | Lifted 'U.pooledReplicateConcurrentlyN'.
427pooledReplicateConcurrentlyN :: AsyncE :> es => Int -> Int -> Eff es a -> Eff es [a]
428pooledReplicateConcurrentlyN threads n f = unsafeEff $ \es -> do
429 U.pooledReplicateConcurrentlyN threads n (unEff f =<< cloneEnv es)
431-- | Lifted 'U.pooledReplicateConcurrently'.
432pooledReplicateConcurrently :: AsyncE :> es => Int -> Eff es a -> Eff es [a]
433pooledReplicateConcurrently n f = unsafeEff $ \es -> do
434 U.pooledReplicateConcurrently n (unEff f =<< cloneEnv es)
436-- | Lifted 'U.pooledReplicateConcurrentlyN_'.
437pooledReplicateConcurrentlyN_ :: AsyncE :> es => Int -> Int -> Eff es a -> Eff es ()
438pooledReplicateConcurrentlyN_ threads n f = unsafeEff $ \es -> do
439 U.pooledReplicateConcurrentlyN_ threads n (unEff f =<< cloneEnv es)
441-- | Lifted 'U.pooledReplicateConcurrently_'.
442pooledReplicateConcurrently_ :: AsyncE :> es => Int -> Eff es a -> Eff es ()
443pooledReplicateConcurrently_ n f = unsafeEff $ \es -> do
444 U.pooledReplicateConcurrently_ n (unEff f =<< cloneEnv es)
446----------------------------------------
447-- Conc
449-- | Lifted 'U.Conc'.
450data Conc es a where
451 Action :: Eff es a -> Conc es a
452 Apply :: Conc es (v -> a) -> Conc es v -> Conc es a
453 LiftA2 :: (x -> y -> a) -> Conc es x -> Conc es y -> Conc es a
454 Pure :: a -> Conc es a
455 Alt :: Conc es a -> Conc es a -> Conc es a
456 Empty :: Conc es a
458deriving instance Functor (Conc es)
460instance Applicative (Conc es) where
461 pure = Pure
462 (<*>) = Apply
463 (<*) = LiftA2 (\x _ -> x)
464 (*>) = LiftA2 (\_ y -> y)
465 liftA2 = LiftA2
467instance Alternative (Conc es) where
468 empty = Empty
469 (<|>) = Alt
471instance Semigroup a => Semigroup (Conc es a) where
472 (<>) = liftA2 (<>)
474instance Monoid a => Monoid (Conc es a) where
475 mempty = pure mempty
477-- | Lifted 'U.conc'.
478conc :: Eff es a -> Conc es a
479conc = Action
481-- | Lifted 'U.runConc'.
482runConc :: AsyncE :> es => Conc es a -> Eff es a
483runConc m = unsafeEff $ \es -> U.runConc (unliftConc es m)
484 where
485 unliftConc :: Env es -> Conc es a -> U.Conc IO a
486 unliftConc es = \case
487 Action f -> I.Action $ unEff f =<< cloneEnv es
488 Apply a b -> I.Apply (unliftConc es a) (unliftConc es b)
489 LiftA2 f a b -> I.LiftA2 f (unliftConc es a) (unliftConc es b)
490 Pure a -> I.Pure a
491 Alt a b -> I.Alt (unliftConc es a) (unliftConc es b)
492 Empty -> I.Empty
494----------------------------------------
495-- Concurrently
497-- | Lifted 'A.Concurrently'.
498newtype Concurrently es a = Concurrently { runConcurrently :: Eff es a }
500instance Functor (Concurrently es) where
501 fmap f (Concurrently a) = Concurrently (fmap f a)
503instance AsyncE :> es => Applicative (Concurrently es) where
504 pure = Concurrently . pure
505 Concurrently fs <*> Concurrently as =
506 Concurrently ((\(f, a) -> f a) <$> concurrently fs as)
508instance AsyncE :> es => Alternative (Concurrently es) where
509 empty = Concurrently . unsafeEff_ . forever $ threadDelay maxBound
510 Concurrently as <|> Concurrently bs =
511 Concurrently (either id id <$> race as bs)
513instance (AsyncE :> es, Semigroup a) => Semigroup (Concurrently es a) where
514 (<>) = liftA2 (<>)
516instance (AsyncE :> es, Monoid a) => Monoid (Concurrently es a) where
517 mempty = pure mempty
519----------------------------------------
520-- Helpers
522liftAsync
523 :: (IO a -> IO (Async a))
524 -> Eff es a
525 -> Eff es (Async a)
526liftAsync fork action = unsafeEff $ \es -> do
527 esA <- cloneEnv es
528 fork $ unEff action esA
530liftAsyncWithUnmask
531 :: (((forall b. IO b -> IO b) -> IO a) -> IO (Async a))
532 -> ((forall b. Eff es b -> Eff es b) -> Eff es a)
533 -> Eff es (Async a)
534liftAsyncWithUnmask fork action = do
535 unsafeWithLiftSeqOp $ \liftSeqOp -> unsafeEff $ \es -> do
536 esA <- cloneEnv es
537 fork $ \unmask -> unEff (action $ liftSeqOp unmask) esA
539liftWithAsync
540 :: (IO a -> (Async a -> IO b) -> IO b)
541 -> Eff es a
542 -> (Async a -> Eff es b)
543 -> Eff es b
544liftWithAsync withA action k = unsafeEff $ \es -> do
545 esA <- cloneEnv es
546 withA (unEff action esA)
547 (\a -> unEff (k a) es)
549liftWithAsyncWithUnmask
550 :: (((forall c. IO c -> IO c) -> IO a) -> (Async a -> IO b) -> IO b)
551 -> ((forall c. Eff es c -> Eff es c) -> Eff es a)
552 -> (Async a -> Eff es b)
553 -> Eff es b
554liftWithAsyncWithUnmask withA action k = do
555 unsafeWithLiftSeqOp $ \liftSeqOp -> unsafeEff $ \es -> do
556 esA <- cloneEnv es
557 withA (\unmask -> unEff (action $ liftSeqOp unmask) esA)
558 (\a -> unEff (k a) es)