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