@@ -0,0 +1,544 @@
1+{-# LANGUAGE UndecidableInstances #-}
2+-- | Lifted version of "Control.Concurrent.Async".
3+module Effectful.Concurrent.Async
7+ -- * Asynchronous actions
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(..)
21+ -- ** High-level utilities
23+ , concurrently, concurrently_
24+ , mapConcurrently, forConcurrently
25+ , mapConcurrently_, forConcurrently_
26+ , replicateConcurrently, replicateConcurrently_
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
59+ -- *** Waiting for multiple 'Async's in STM
60+ , A.waitAnySTM, A.waitAnyCatchSTM
61+ , A.waitEitherSTM, A.waitEitherCatchSTM
67+ -- ** Spawning (low-level API)
68+ , async, asyncBound, asyncOn, asyncWithUnmask, asyncOnWithUnmask
71+ , link, linkOnly, link2, link2Only, A.ExceptionInLinkedThread(..)
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
84+import Effectful.Concurrent.Effect
85+import Effectful.Dispatch.Static
86+import Effectful.Monad
88+-- | Lifted 'A.async'.
89+async :: Concurrent :> es => Eff es a -> Eff es (Async a)
90+async = liftAsync A.async
92+-- | Lifted 'A.asyncBound'.
93+asyncBound :: Concurrent :> es => Eff es a -> Eff es (Async a)
94+asyncBound = liftAsync A.asyncBound
96+-- | Lifted 'A.asyncOn'.
97+asyncOn :: Concurrent :> es => Int -> Eff es a -> Eff es (Async a)
98+asyncOn cpu = liftAsync (A.asyncOn cpu)
100+-- | Lifted 'A.asyncWithUnmask'.
103+ => ((forall b. Eff es b -> Eff es b) -> Eff es a)
105+asyncWithUnmask = liftAsyncWithUnmask A.asyncWithUnmask
107+-- | Lifted 'A.asyncOnWithUnmask'.
111+ -> ((forall b. Eff es b -> Eff es b) -> Eff es a)
113+asyncOnWithUnmask cpu = liftAsyncWithUnmask (A.asyncOnWithUnmask cpu)
115+-- | Lifted 'A.withAsync'.
119+ -> (Async a -> Eff es b)
121+withAsync = liftWithAsync A.withAsync
123+-- | Lifted 'A.withAsyncBound'.
127+ -> (Async a -> Eff es b)
129+withAsyncBound = liftWithAsync A.withAsyncBound
131+-- | Lifted 'A.withAsyncOn'.
136+ -> (Async a -> Eff es b)
138+withAsyncOn cpu = liftWithAsync (A.withAsyncOn cpu)
140+-- | Lifted 'A.withAsyncWithUnmask'.
143+ => ((forall c. Eff es c -> Eff es c) -> Eff es a)
144+ -> (Async a -> Eff es b)
146+withAsyncWithUnmask = liftWithAsyncWithUnmask A.withAsyncWithUnmask
148+-- | Lifted 'A.withAsyncOnWithUnmask'.
149+withAsyncOnWithUnmask
152+ -> ((forall c. Eff es c -> Eff es c) -> Eff es a)
153+ -> (Async a -> Eff es b)
155+withAsyncOnWithUnmask cpu = liftWithAsyncWithUnmask (A.withAsyncOnWithUnmask cpu)
157+-- | Lifted 'A.wait'.
158+wait :: Concurrent :> es => Async a -> Eff es a
159+wait = unsafeEff_ . A.wait
161+-- | Lifted 'A.poll'.
165+ -> Eff es (Maybe (Either SomeException a))
166+poll = unsafeEff_ . A.poll
168+-- | Lifted 'A.cancel'.
169+cancel :: Concurrent :> es => Async a -> Eff es ()
170+cancel = unsafeEff_ . A.cancel
172+-- | Lifted 'A.cancelWith'.
173+cancelWith :: (Exception e, Concurrent :> es) => Async a -> e -> Eff es ()
174+cancelWith a = unsafeEff_ . A.cancelWith a
176+-- | Lifted 'A.uninterruptibleCancel'.
177+uninterruptibleCancel :: Concurrent :> es => Async a -> Eff es ()
178+uninterruptibleCancel = unsafeEff_ . A.uninterruptibleCancel
180+-- | Lifted 'A.waitCatch'.
184+ -> Eff es (Either SomeException a)
185+waitCatch = unsafeEff_ . A.waitCatch
187+-- | Lifted 'A.waitAny'.
188+waitAny :: Concurrent :> es => [Async a] -> Eff es (Async a, a)
189+waitAny = unsafeEff_ . A.waitAny
191+-- | Lifted 'A.waitAnyCatch'.
195+ -> Eff es (Async a, Either SomeException a)
196+waitAnyCatch = unsafeEff_ . A.waitAnyCatch
198+-- | Lifted 'A.waitAnyCancel'.
199+waitAnyCancel :: Concurrent :> es => [Async a] -> Eff es (Async a, a)
200+waitAnyCancel = unsafeEff_ . A.waitAnyCancel
202+-- | Lifted 'A.waitAnyCatchCancel'.
206+ -> Eff es (Async a, Either SomeException a)
207+waitAnyCatchCancel = unsafeEff_ . A.waitAnyCatchCancel
209+-- | Lifted 'A.waitEither'.
214+ -> Eff es (Either a b)
215+waitEither a b = unsafeEff_ $ A.waitEither a b
217+-- | Lifted 'A.waitEitherCatch'.
222+ -> Eff es (Either (Either SomeException a) (Either SomeException b))
223+waitEitherCatch a b = unsafeEff_ $ A.waitEitherCatch a b
225+-- | Lifted 'A.waitEitherCancel'.
230+ -> Eff es (Either a b)
231+waitEitherCancel a b = unsafeEff_ $ A.waitEitherCancel a b
233+-- | Lifted 'A.waitEitherCatchCancel'.
234+waitEitherCatchCancel
238+ -> Eff es (Either (Either SomeException a) (Either SomeException b))
239+waitEitherCatchCancel a b = unsafeEff_ $ A.waitEitherCatch a b
241+-- | Lifted 'A.waitEither_'.
242+waitEither_ :: Concurrent :> es => Async a -> Async b -> Eff es ()
243+waitEither_ a b = unsafeEff_ $ A.waitEither_ a b
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
249+-- | Lifted 'A.link'.
250+link :: Concurrent :> es => Async a -> Eff es ()
251+link = unsafeEff_ . A.link
253+-- | Lifted 'A.linkOnly'.
254+linkOnly :: Concurrent :> es => (SomeException -> Bool) -> Async a -> Eff es ()
255+linkOnly f = unsafeEff_ . A.linkOnly f
257+-- | Lifted 'A.link2'.
258+link2 :: Concurrent :> es => Async a -> Async b -> Eff es ()
259+link2 a b = unsafeEff_ $ A.link2 a b
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
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)
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)
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)
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)
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.
289+-- | Lifted 'A.mapConcurrently'.
291+ :: (Traversable f, Concurrent :> es)
295+mapConcurrently f t = unsafeEff $ \es -> do
296+ U.mapConcurrently (\a -> unEff (f a) =<< cloneEnv es) t
298+-- | Lifted 'A.mapConcurrently_'.
300+ :: (Foldable f, Concurrent :> es)
304+mapConcurrently_ f t = unsafeEff $ \es -> do
305+ U.mapConcurrently_ (\a -> unEff (f a) =<< cloneEnv es) t
307+-- | Lifted 'A.forConcurrently'.
309+ :: (Traversable f, Concurrent :> es)
313+forConcurrently t f = unsafeEff $ \es -> do
314+ U.forConcurrently t (\a -> unEff (f a) =<< cloneEnv es)
316+-- | Lifted 'A.forConcurrently_'.
318+ :: (Foldable f, Concurrent :> es)
322+forConcurrently_ t f = unsafeEff $ \es -> do
323+ U.forConcurrently_ t (\a -> unEff (f a) =<< cloneEnv es)
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)
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)
335+----------------------------------------
336+-- Pooled concurrency (unliftio)
338+-- | Lifted 'U.pooledMapConcurrentlyN'.
339+pooledMapConcurrentlyN
340+ :: (Concurrent :> es, Traversable t)
345+pooledMapConcurrentlyN threads f t = unsafeEff $ \es -> do
346+ U.pooledMapConcurrentlyN threads (\a -> unEff (f a) =<< cloneEnv es) t
348+-- | Lifted 'U.pooledMapConcurrently'.
349+pooledMapConcurrently
350+ :: (Concurrent :> es, Traversable t)
354+pooledMapConcurrently f t = unsafeEff $ \es -> do
355+ U.pooledMapConcurrently (\a -> unEff (f a) =<< cloneEnv es) t
357+-- | Lifted 'U.pooledMapConcurrentlyN'.
358+pooledMapConcurrentlyN_
359+ :: (Concurrent :> es, Foldable f)
364+pooledMapConcurrentlyN_ threads f t = unsafeEff $ \es -> do
365+ U.pooledMapConcurrentlyN_ threads (\a -> unEff (f a) =<< cloneEnv es) t
367+-- | Lifted 'U.pooledMapConcurrently_'.
368+pooledMapConcurrently_
369+ :: (Concurrent :> es, Foldable f)
373+pooledMapConcurrently_ f t = unsafeEff $ \es -> do
374+ U.pooledMapConcurrently_ (\a -> unEff (f a) =<< cloneEnv es) t
376+-- | Lifted 'U.pooledForConcurrentlyN'.
377+pooledForConcurrentlyN
378+ :: (Concurrent :> es, Traversable t)
383+pooledForConcurrentlyN threads t f = unsafeEff $ \es -> do
384+ U.pooledForConcurrentlyN threads t (\a -> unEff (f a) =<< cloneEnv es)
386+-- | Lifted 'U.pooledForConcurrently'.
387+pooledForConcurrently
388+ :: (Concurrent :> es, Traversable t)
392+pooledForConcurrently t f = unsafeEff $ \es -> do
393+ U.pooledForConcurrently t (\a -> unEff (f a) =<< cloneEnv es)
395+-- | Lifted 'U.pooledForConcurrentlyN'.
396+pooledForConcurrentlyN_
397+ :: (Concurrent :> es, Foldable f)
402+pooledForConcurrentlyN_ threads t f = unsafeEff $ \es -> do
403+ U.pooledForConcurrentlyN_ threads t (\a -> unEff (f a) =<< cloneEnv es)
405+-- | Lifted 'U.pooledForConcurrently_'.
406+pooledForConcurrently_
407+ :: (Concurrent :> es, Foldable f)
411+pooledForConcurrently_ t f = unsafeEff $ \es -> do
412+ U.pooledForConcurrently_ t (\a -> unEff (f a) =<< cloneEnv es)
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)
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)
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)
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)
434+----------------------------------------
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
446+deriving instance Functor (Conc es)
448+instance Applicative (Conc es) where
451+ (<*) = LiftA2 (\x _ -> x)
452+ (*>) = LiftA2 (\_ y -> y)
455+instance Alternative (Conc es) where
459+instance Semigroup a => Semigroup (Conc es a) where
462+instance Monoid a => Monoid (Conc es a) where
463+ mempty = pure mempty
465+-- | Lifted 'U.conc'.
466+conc :: Eff es a -> Conc es a
469+-- | Lifted 'U.runConc'.
470+runConc :: Concurrent :> es => Conc es a -> Eff es a
471+runConc m = unsafeEff $ \es -> U.runConc (unliftConc es m)
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)
479+ Alt a b -> I.Alt (unliftConc es a) (unliftConc es b)
482+----------------------------------------
485+-- | Lifted 'A.Concurrently'.
486+newtype Concurrently es a = Concurrently { runConcurrently :: Eff es a }
488+instance Functor (Concurrently es) where
489+ fmap f (Concurrently a) = Concurrently (fmap f a)
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)
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)
501+instance (Concurrent :> es, Semigroup a) => Semigroup (Concurrently es a) where
504+instance (Concurrent :> es, Monoid a) => Monoid (Concurrently es a) where
505+ mempty = pure mempty
507+----------------------------------------
511+ :: (IO a -> IO (Async a))
514+liftAsync fork action = unsafeEff $ \es -> do
516+ fork $ unEff action esA
519+ :: (((forall b. IO b -> IO b) -> IO a) -> IO (Async a))
520+ -> ((forall b. Eff es b -> Eff es b) -> Eff es a)
522+liftAsyncWithUnmask fork action = unsafeEff $ \es -> do
524+ fork $ \unmask -> unEff (action $ unsafeLiftMapIO unmask) esA
527+ :: (IO a -> (Async a -> IO b) -> IO b)
529+ -> (Async a -> Eff es b)
531+liftWithAsync withA action k = unsafeEff $ \es -> do
533+ withA (unEff action esA)
534+ (\a -> unEff (k a) es)
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)
541+liftWithAsyncWithUnmask withA action k = unsafeEff $ \es -> do
543+ withA (\unmask -> unEff (action $ unsafeLiftMapIO unmask) esA)
544+ (\a -> unEff (k a) es)