@@ -0,0 +1,544 @@
1+{-# LANGUAGE UndecidableInstances #-}
2+-- | Lifted version of "Control.Concurrent.Async".
3+module Effectful.Concurrent.Async
4+ ( Concurrent
5+ , runConcurrent
6+
7+ -- * Asynchronous actions
8+ , Async
9+
10+ -- * High-level API
11+
12+ -- ** Spawning with automatic 'cancel'ation
13+ , withAsync, withAsyncBound, withAsyncOn, withAsyncWithUnmask
14+ , withAsyncOnWithUnmask
15+
16+ -- ** Querying 'Async's
17+ , wait, poll, waitCatch, A.asyncThreadId
18+ , cancel, uninterruptibleCancel, cancelWith, A.AsyncCancelled(..)
19+ , A.compareAsyncs
20+
21+ -- ** High-level utilities
22+ , race, race_
23+ , concurrently, concurrently_
24+ , mapConcurrently, forConcurrently
25+ , mapConcurrently_, forConcurrently_
26+ , replicateConcurrently, replicateConcurrently_
27+
28+ -- *** Concurrently
29+ , Concurrently(..)
30+
31+ -- *** Conc
32+ , Conc, conc, runConc, U.ConcException(..)
33+
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_
47+
48+ -- ** Specialised operations
49+
50+ -- *** STM operations
51+ , A.waitSTM, A.pollSTM, A.waitCatchSTM
52+
53+ -- *** Waiting for multiple 'Async's
54+ , waitAny, waitAnyCatch, waitAnyCancel, waitAnyCatchCancel
55+ , waitEither, waitEitherCatch, waitEitherCancel, waitEitherCatchCancel
56+ , waitEither_
57+ , waitBoth
58+
59+ -- *** Waiting for multiple 'Async's in STM
60+ , A.waitAnySTM, A.waitAnyCatchSTM
61+ , A.waitEitherSTM, A.waitEitherCatchSTM
62+ , A.waitEitherSTM_
63+ , A.waitBothSTM
64+
65+ -- * Low-level API
66+
67+ -- ** Spawning (low-level API)
68+ , async, asyncBound, asyncOn, asyncWithUnmask, asyncOnWithUnmask
69+
70+ -- ** Linking
71+ , link, linkOnly, link2, link2Only, A.ExceptionInLinkedThread(..)
72+ ) where
73+
74+import Control.Applicative
75+import Control.Concurrent (threadDelay)
76+import Control.Concurrent.Async (Async)
77+import Control.Exception (Exception, SomeException)
78+import Control.Monad (forever)
79+import Data.Kind (Type)
80+import qualified Control.Concurrent.Async as A
81+import qualified UnliftIO.Async as U
82+import qualified UnliftIO.Internals.Async as I
83+
84+import Effectful.Concurrent.Effect
85+import Effectful.Dispatch.Static
86+import Effectful.Monad
87+
88+-- | Lifted 'A.async'.
89+async :: Concurrent :> es => Eff es a -> Eff es (Async a)
90+async = liftAsync A.async
91+
92+-- | Lifted 'A.asyncBound'.
93+asyncBound :: Concurrent :> es => Eff es a -> Eff es (Async a)
94+asyncBound = liftAsync A.asyncBound
95+
96+-- | Lifted 'A.asyncOn'.
97+asyncOn :: Concurrent :> es => Int -> Eff es a -> Eff es (Async a)
98+asyncOn cpu = liftAsync (A.asyncOn cpu)
99+
100+-- | Lifted 'A.asyncWithUnmask'.
101+asyncWithUnmask
102+ :: Concurrent :> es
103+ => ((forall b. Eff es b -> Eff es b) -> Eff es a)
104+ -> Eff es (Async a)
105+asyncWithUnmask = liftAsyncWithUnmask A.asyncWithUnmask
106+
107+-- | Lifted 'A.asyncOnWithUnmask'.
108+asyncOnWithUnmask
109+ :: Concurrent :> es
110+ => Int
111+ -> ((forall b. Eff es b -> Eff es b) -> Eff es a)
112+ -> Eff es (Async a)
113+asyncOnWithUnmask cpu = liftAsyncWithUnmask (A.asyncOnWithUnmask cpu)
114+
115+-- | Lifted 'A.withAsync'.
116+withAsync
117+ :: Concurrent :> es
118+ => Eff es a
119+ -> (Async a -> Eff es b)
120+ -> Eff es b
121+withAsync = liftWithAsync A.withAsync
122+
123+-- | Lifted 'A.withAsyncBound'.
124+withAsyncBound
125+ :: Concurrent :> es
126+ => Eff es a
127+ -> (Async a -> Eff es b)
128+ -> Eff es b
129+withAsyncBound = liftWithAsync A.withAsyncBound
130+
131+-- | Lifted 'A.withAsyncOn'.
132+withAsyncOn
133+ :: Concurrent :> es
134+ => Int
135+ -> Eff es a
136+ -> (Async a -> Eff es b)
137+ -> Eff es b
138+withAsyncOn cpu = liftWithAsync (A.withAsyncOn cpu)
139+
140+-- | Lifted 'A.withAsyncWithUnmask'.
141+withAsyncWithUnmask
142+ :: Concurrent :> es
143+ => ((forall c. Eff es c -> Eff es c) -> Eff es a)
144+ -> (Async a -> Eff es b)
145+ -> Eff es b
146+withAsyncWithUnmask = liftWithAsyncWithUnmask A.withAsyncWithUnmask
147+
148+-- | Lifted 'A.withAsyncOnWithUnmask'.
149+withAsyncOnWithUnmask
150+ :: Concurrent :> es
151+ => Int
152+ -> ((forall c. Eff es c -> Eff es c) -> Eff es a)
153+ -> (Async a -> Eff es b)
154+ -> Eff es b
155+withAsyncOnWithUnmask cpu = liftWithAsyncWithUnmask (A.withAsyncOnWithUnmask cpu)
156+
157+-- | Lifted 'A.wait'.
158+wait :: Concurrent :> es => Async a -> Eff es a
159+wait = unsafeEff_ . A.wait
160+
161+-- | Lifted 'A.poll'.
162+poll
163+ :: Concurrent :> es
164+ => Async a
165+ -> Eff es (Maybe (Either SomeException a))
166+poll = unsafeEff_ . A.poll
167+
168+-- | Lifted 'A.cancel'.
169+cancel :: Concurrent :> es => Async a -> Eff es ()
170+cancel = unsafeEff_ . A.cancel
171+
172+-- | Lifted 'A.cancelWith'.
173+cancelWith :: (Exception e, Concurrent :> es) => Async a -> e -> Eff es ()
174+cancelWith a = unsafeEff_ . A.cancelWith a
175+
176+-- | Lifted 'A.uninterruptibleCancel'.
177+uninterruptibleCancel :: Concurrent :> es => Async a -> Eff es ()
178+uninterruptibleCancel = unsafeEff_ . A.uninterruptibleCancel
179+
180+-- | Lifted 'A.waitCatch'.
181+waitCatch
182+ :: Concurrent :> es
183+ => Async a
184+ -> Eff es (Either SomeException a)
185+waitCatch = unsafeEff_ . A.waitCatch
186+
187+-- | Lifted 'A.waitAny'.
188+waitAny :: Concurrent :> es => [Async a] -> Eff es (Async a, a)
189+waitAny = unsafeEff_ . A.waitAny
190+
191+-- | Lifted 'A.waitAnyCatch'.
192+waitAnyCatch
193+ :: Concurrent :> es
194+ => [Async a]
195+ -> Eff es (Async a, Either SomeException a)
196+waitAnyCatch = unsafeEff_ . A.waitAnyCatch
197+
198+-- | Lifted 'A.waitAnyCancel'.
199+waitAnyCancel :: Concurrent :> es => [Async a] -> Eff es (Async a, a)
200+waitAnyCancel = unsafeEff_ . A.waitAnyCancel
201+
202+-- | Lifted 'A.waitAnyCatchCancel'.
203+waitAnyCatchCancel
204+ :: Concurrent :> es
205+ => [Async a]
206+ -> Eff es (Async a, Either SomeException a)
207+waitAnyCatchCancel = unsafeEff_ . A.waitAnyCatchCancel
208+
209+-- | Lifted 'A.waitEither'.
210+waitEither
211+ :: Concurrent :> es
212+ => Async a
213+ -> Async b
214+ -> Eff es (Either a b)
215+waitEither a b = unsafeEff_ $ A.waitEither a b
216+
217+-- | Lifted 'A.waitEitherCatch'.
218+waitEitherCatch
219+ :: Concurrent :> es
220+ => Async a
221+ -> Async b
222+ -> Eff es (Either (Either SomeException a) (Either SomeException b))
223+waitEitherCatch a b = unsafeEff_ $ A.waitEitherCatch a b
224+
225+-- | Lifted 'A.waitEitherCancel'.
226+waitEitherCancel
227+ :: Concurrent :> es
228+ => Async a
229+ -> Async b
230+ -> Eff es (Either a b)
231+waitEitherCancel a b = unsafeEff_ $ A.waitEitherCancel a b
232+
233+-- | Lifted 'A.waitEitherCatchCancel'.
234+waitEitherCatchCancel
235+ :: Concurrent :> es
236+ => Async a
237+ -> Async b
238+ -> Eff es (Either (Either SomeException a) (Either SomeException b))
239+waitEitherCatchCancel a b = unsafeEff_ $ A.waitEitherCatch a b
240+
241+-- | Lifted 'A.waitEither_'.
242+waitEither_ :: Concurrent :> es => Async a -> Async b -> Eff es ()
243+waitEither_ a b = unsafeEff_ $ A.waitEither_ a b
244+
245+-- | Lifted 'A.waitBoth'.
246+waitBoth :: Concurrent :> es => Async a -> Async b -> Eff es (a, b)
247+waitBoth a b = unsafeEff_ $ A.waitBoth a b
248+
249+-- | Lifted 'A.link'.
250+link :: Concurrent :> es => Async a -> Eff es ()
251+link = unsafeEff_ . A.link
252+
253+-- | Lifted 'A.linkOnly'.
254+linkOnly :: Concurrent :> es => (SomeException -> Bool) -> Async a -> Eff es ()
255+linkOnly f = unsafeEff_ . A.linkOnly f
256+
257+-- | Lifted 'A.link2'.
258+link2 :: Concurrent :> es => Async a -> Async b -> Eff es ()
259+link2 a b = unsafeEff_ $ A.link2 a b
260+
261+-- | Lifted 'A.link2Only'.
262+link2Only :: Concurrent :> es => (SomeException -> Bool) -> Async a -> Async b -> Eff es ()
263+link2Only f a b = unsafeEff_ $ A.link2Only f a b
264+
265+-- | Lifted 'A.race'.
266+race :: Concurrent :> es => Eff es a -> Eff es b -> Eff es (Either a b)
267+race ma mb = unsafeEff $ \es -> do
268+ A.race (unEff ma =<< cloneEnv es) (unEff mb =<< cloneEnv es)
269+
270+-- | Lifted 'A.race_'.
271+race_ :: Concurrent :> es => Eff es a -> Eff es b -> Eff es ()
272+race_ ma mb = unsafeEff $ \es -> do
273+ A.race_ (unEff ma =<< cloneEnv es) (unEff mb =<< cloneEnv es)
274+
275+-- | Lifted 'A.concurrently'.
276+concurrently :: Concurrent :> es => Eff es a -> Eff es b -> Eff es (a, b)
277+concurrently ma mb = unsafeEff $ \es -> do
278+ A.concurrently (unEff ma =<< cloneEnv es) (unEff mb =<< cloneEnv es)
279+
280+-- | Lifted 'A.concurrently_'.
281+concurrently_ :: Concurrent :> es => Eff es a -> Eff es b -> Eff es ()
282+concurrently_ ma mb = unsafeEff $ \es -> do
283+ A.concurrently_ (unEff ma =<< cloneEnv es) (unEff mb =<< cloneEnv es)
284+
285+-- Below functions use variants from the @unliftio@ library as they minimize the
286+-- amount of spawned threads and are thus much more efficient than the ones from
287+-- the @async@ library.
288+
289+-- | Lifted 'A.mapConcurrently'.
290+mapConcurrently
291+ :: (Traversable f, Concurrent :> es)
292+ => (a -> Eff es b)
293+ -> f a
294+ -> Eff es (f b)
295+mapConcurrently f t = unsafeEff $ \es -> do
296+ U.mapConcurrently (\a -> unEff (f a) =<< cloneEnv es) t
297+
298+-- | Lifted 'A.mapConcurrently_'.
299+mapConcurrently_
300+ :: (Foldable f, Concurrent :> es)
301+ => (a -> Eff es b)
302+ -> f a
303+ -> Eff es ()
304+mapConcurrently_ f t = unsafeEff $ \es -> do
305+ U.mapConcurrently_ (\a -> unEff (f a) =<< cloneEnv es) t
306+
307+-- | Lifted 'A.forConcurrently'.
308+forConcurrently
309+ :: (Traversable f, Concurrent :> es)
310+ => f a
311+ -> (a -> Eff es b)
312+ -> Eff es (f b)
313+forConcurrently t f = unsafeEff $ \es -> do
314+ U.forConcurrently t (\a -> unEff (f a) =<< cloneEnv es)
315+
316+-- | Lifted 'A.forConcurrently_'.
317+forConcurrently_
318+ :: (Foldable f, Concurrent :> es)
319+ => f a
320+ -> (a -> Eff es b)
321+ -> Eff es ()
322+forConcurrently_ t f = unsafeEff $ \es -> do
323+ U.forConcurrently_ t (\a -> unEff (f a) =<< cloneEnv es)
324+
325+-- | Lifted 'A.replicateConcurrently'.
326+replicateConcurrently :: Concurrent :> es => Int -> Eff es a -> Eff es [a]
327+replicateConcurrently n f = unsafeEff $ \es -> do
328+ U.replicateConcurrently n (unEff f =<< cloneEnv es)
329+
330+-- | Lifted 'A.replicateConcurrently_'.
331+replicateConcurrently_ :: Concurrent :> es => Int -> Eff es a -> Eff es ()
332+replicateConcurrently_ n f = unsafeEff $ \es -> do
333+ U.replicateConcurrently_ n (unEff f =<< cloneEnv es)
334+
335+----------------------------------------
336+-- Pooled concurrency (unliftio)
337+
338+-- | Lifted 'U.pooledMapConcurrentlyN'.
339+pooledMapConcurrentlyN
340+ :: (Concurrent :> es, Traversable t)
341+ => Int
342+ -> (a -> Eff es b)
343+ -> t a
344+ -> Eff es (t b)
345+pooledMapConcurrentlyN threads f t = unsafeEff $ \es -> do
346+ U.pooledMapConcurrentlyN threads (\a -> unEff (f a) =<< cloneEnv es) t
347+
348+-- | Lifted 'U.pooledMapConcurrently'.
349+pooledMapConcurrently
350+ :: (Concurrent :> es, Traversable t)
351+ => (a -> Eff es b)
352+ -> t a
353+ -> Eff es (t b)
354+pooledMapConcurrently f t = unsafeEff $ \es -> do
355+ U.pooledMapConcurrently (\a -> unEff (f a) =<< cloneEnv es) t
356+
357+-- | Lifted 'U.pooledMapConcurrentlyN'.
358+pooledMapConcurrentlyN_
359+ :: (Concurrent :> es, Foldable f)
360+ => Int
361+ -> (a -> Eff es b)
362+ -> f a
363+ -> Eff es ()
364+pooledMapConcurrentlyN_ threads f t = unsafeEff $ \es -> do
365+ U.pooledMapConcurrentlyN_ threads (\a -> unEff (f a) =<< cloneEnv es) t
366+
367+-- | Lifted 'U.pooledMapConcurrently_'.
368+pooledMapConcurrently_
369+ :: (Concurrent :> es, Foldable f)
370+ => (a -> Eff es b)
371+ -> f a
372+ -> Eff es ()
373+pooledMapConcurrently_ f t = unsafeEff $ \es -> do
374+ U.pooledMapConcurrently_ (\a -> unEff (f a) =<< cloneEnv es) t
375+
376+-- | Lifted 'U.pooledForConcurrentlyN'.
377+pooledForConcurrentlyN
378+ :: (Concurrent :> es, Traversable t)
379+ => Int
380+ -> t a
381+ -> (a -> Eff es b)
382+ -> Eff es (t b)
383+pooledForConcurrentlyN threads t f = unsafeEff $ \es -> do
384+ U.pooledForConcurrentlyN threads t (\a -> unEff (f a) =<< cloneEnv es)
385+
386+-- | Lifted 'U.pooledForConcurrently'.
387+pooledForConcurrently
388+ :: (Concurrent :> es, Traversable t)
389+ => t a
390+ -> (a -> Eff es b)
391+ -> Eff es (t b)
392+pooledForConcurrently t f = unsafeEff $ \es -> do
393+ U.pooledForConcurrently t (\a -> unEff (f a) =<< cloneEnv es)
394+
395+-- | Lifted 'U.pooledForConcurrentlyN'.
396+pooledForConcurrentlyN_
397+ :: (Concurrent :> es, Foldable f)
398+ => Int
399+ -> f a
400+ -> (a -> Eff es b)
401+ -> Eff es ()
402+pooledForConcurrentlyN_ threads t f = unsafeEff $ \es -> do
403+ U.pooledForConcurrentlyN_ threads t (\a -> unEff (f a) =<< cloneEnv es)
404+
405+-- | Lifted 'U.pooledForConcurrently_'.
406+pooledForConcurrently_
407+ :: (Concurrent :> es, Foldable f)
408+ => f a
409+ -> (a -> Eff es b)
410+ -> Eff es ()
411+pooledForConcurrently_ t f = unsafeEff $ \es -> do
412+ U.pooledForConcurrently_ t (\a -> unEff (f a) =<< cloneEnv es)
413+
414+-- | Lifted 'U.pooledReplicateConcurrentlyN'.
415+pooledReplicateConcurrentlyN :: Concurrent :> es => Int -> Int -> Eff es a -> Eff es [a]
416+pooledReplicateConcurrentlyN threads n f = unsafeEff $ \es -> do
417+ U.pooledReplicateConcurrentlyN threads n (unEff f =<< cloneEnv es)
418+
419+-- | Lifted 'U.pooledReplicateConcurrently'.
420+pooledReplicateConcurrently :: Concurrent :> es => Int -> Eff es a -> Eff es [a]
421+pooledReplicateConcurrently n f = unsafeEff $ \es -> do
422+ U.pooledReplicateConcurrently n (unEff f =<< cloneEnv es)
423+
424+-- | Lifted 'U.pooledReplicateConcurrentlyN_'.
425+pooledReplicateConcurrentlyN_ :: Concurrent :> es => Int -> Int -> Eff es a -> Eff es ()
426+pooledReplicateConcurrentlyN_ threads n f = unsafeEff $ \es -> do
427+ U.pooledReplicateConcurrentlyN_ threads n (unEff f =<< cloneEnv es)
428+
429+-- | Lifted 'U.pooledReplicateConcurrently_'.
430+pooledReplicateConcurrently_ :: Concurrent :> es => Int -> Eff es a -> Eff es ()
431+pooledReplicateConcurrently_ n f = unsafeEff $ \es -> do
432+ U.pooledReplicateConcurrently_ n (unEff f =<< cloneEnv es)
433+
434+----------------------------------------
435+-- Conc
436+
437+-- | Lifted 'U.Conc'.
438+data Conc :: [Effect] -> Type -> Type where
439+ Action :: Eff es a -> Conc es a
440+ Apply :: Conc es (v -> a) -> Conc es v -> Conc es a
441+ LiftA2 :: (x -> y -> a) -> Conc es x -> Conc es y -> Conc es a
442+ Pure :: a -> Conc es a
443+ Alt :: Conc es a -> Conc es a -> Conc es a
444+ Empty :: Conc es a
445+
446+deriving instance Functor (Conc es)
447+
448+instance Applicative (Conc es) where
449+ pure = Pure
450+ (<*>) = Apply
451+ (<*) = LiftA2 (\x _ -> x)
452+ (*>) = LiftA2 (\_ y -> y)
453+ liftA2 = LiftA2
454+
455+instance Alternative (Conc es) where
456+ empty = Empty
457+ (<|>) = Alt
458+
459+instance Semigroup a => Semigroup (Conc es a) where
460+ (<>) = liftA2 (<>)
461+
462+instance Monoid a => Monoid (Conc es a) where
463+ mempty = pure mempty
464+
465+-- | Lifted 'U.conc'.
466+conc :: Eff es a -> Conc es a
467+conc = Action
468+
469+-- | Lifted 'U.runConc'.
470+runConc :: Concurrent :> es => Conc es a -> Eff es a
471+runConc m = unsafeEff $ \es -> U.runConc (unliftConc es m)
472+ where
473+ unliftConc :: Env es -> Conc es a -> U.Conc IO a
474+ unliftConc es = \case
475+ Action f -> I.Action $ unEff f =<< cloneEnv es
476+ Apply a b -> I.Apply (unliftConc es a) (unliftConc es b)
477+ LiftA2 f a b -> I.LiftA2 f (unliftConc es a) (unliftConc es b)
478+ Pure a -> I.Pure a
479+ Alt a b -> I.Alt (unliftConc es a) (unliftConc es b)
480+ Empty -> I.Empty
481+
482+----------------------------------------
483+-- Concurrently
484+
485+-- | Lifted 'A.Concurrently'.
486+newtype Concurrently es a = Concurrently { runConcurrently :: Eff es a }
487+
488+instance Functor (Concurrently es) where
489+ fmap f (Concurrently a) = Concurrently (fmap f a)
490+
491+instance Concurrent :> es => Applicative (Concurrently es) where
492+ pure = Concurrently . pure
493+ Concurrently fs <*> Concurrently as =
494+ Concurrently ((\(f, a) -> f a) <$> concurrently fs as)
495+
496+instance Concurrent :> es => Alternative (Concurrently es) where
497+ empty = Concurrently . unsafeEff_ . forever $ threadDelay maxBound
498+ Concurrently as <|> Concurrently bs =
499+ Concurrently (either id id <$> race as bs)
500+
501+instance (Concurrent :> es, Semigroup a) => Semigroup (Concurrently es a) where
502+ (<>) = liftA2 (<>)
503+
504+instance (Concurrent :> es, Monoid a) => Monoid (Concurrently es a) where
505+ mempty = pure mempty
506+
507+----------------------------------------
508+-- Helpers
509+
510+liftAsync
511+ :: (IO a -> IO (Async a))
512+ -> Eff es a
513+ -> Eff es (Async a)
514+liftAsync fork action = unsafeEff $ \es -> do
515+ esA <- cloneEnv es
516+ fork $ unEff action esA
517+
518+liftAsyncWithUnmask
519+ :: (((forall b. IO b -> IO b) -> IO a) -> IO (Async a))
520+ -> ((forall b. Eff es b -> Eff es b) -> Eff es a)
521+ -> Eff es (Async a)
522+liftAsyncWithUnmask fork action = unsafeEff $ \es -> do
523+ esA <- cloneEnv es
524+ fork $ \unmask -> unEff (action $ unsafeLiftMapIO unmask) esA
525+
526+liftWithAsync
527+ :: (IO a -> (Async a -> IO b) -> IO b)
528+ -> Eff es a
529+ -> (Async a -> Eff es b)
530+ -> Eff es b
531+liftWithAsync withA action k = unsafeEff $ \es -> do
532+ esA <- cloneEnv es
533+ withA (unEff action esA)
534+ (\a -> unEff (k a) es)
535+
536+liftWithAsyncWithUnmask
537+ :: (((forall c. IO c -> IO c) -> IO a) -> (Async a -> IO b) -> IO b)
538+ -> ((forall c. Eff es c -> Eff es c) -> Eff es a)
539+ -> (Async a -> Eff es b)
540+ -> Eff es b
541+liftWithAsyncWithUnmask withA action k = unsafeEff $ \es -> do
542+ esA <- cloneEnv es
543+ withA (\unmask -> unEff (action $ unsafeLiftMapIO unmask) esA)
544+ (\a -> unEff (k a) es)