@@ -1,544 +0,0 @@
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)