@@ -1,8 +1,8 @@
11{-# LANGUAGE UndecidableInstances #-}
2−module Effectful.Async
3− ( -- * Async effect
4− AsyncE
5− , runAsyncE
2+-- | Lifted version of "Control.Concurrent.Async".
3+module Effectful.Concurrent.Async
4+ ( Concurrent
5+ , runConcurrent
66
77 -- * Asynchronous actions
88 , Async
@@ -76,49 +76,37 @@ import Control.Concurrent (threadDelay)
7676import Control.Concurrent.Async (Async)
7777import Control.Exception (Exception, SomeException)
7878import Control.Monad (forever)
79+import Data.Kind (Type)
7980import qualified Control.Concurrent.Async as A
8081import qualified UnliftIO.Async as U
8182import qualified UnliftIO.Internals.Async as I
8283
83−import Effectful.Internal.Effect
84−import Effectful.Internal.Env
85−import Effectful.Internal.Monad
86−
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.
91−data AsyncE :: Effect where
92− AsyncE :: AsyncE m r
93−
94−runAsyncE :: IOE :> es => Eff (AsyncE : es) a -> Eff es a
95−runAsyncE = evalEffect (IdE AsyncE)
96−
97−----------------------------------------
98−-- Operations
84+import Effectful.Concurrent.Effect
85+import Effectful.Dispatch.Static
86+import Effectful.Monad
9987
10088-- | Lifted 'A.async'.
101−async :: AsyncE :> es => Eff es a -> Eff es (Async a)
89+async :: Concurrent :> es => Eff es a -> Eff es (Async a)
10290async = liftAsync A.async
10391
10492-- | Lifted 'A.asyncBound'.
105−asyncBound :: AsyncE :> es => Eff es a -> Eff es (Async a)
93+asyncBound :: Concurrent :> es => Eff es a -> Eff es (Async a)
10694asyncBound = liftAsync A.asyncBound
10795
10896-- | Lifted 'A.asyncOn'.
109−asyncOn :: AsyncE :> es => Int -> Eff es a -> Eff es (Async a)
97+asyncOn :: Concurrent :> es => Int -> Eff es a -> Eff es (Async a)
11098asyncOn cpu = liftAsync (A.asyncOn cpu)
11199
112100-- | Lifted 'A.asyncWithUnmask'.
113101asyncWithUnmask
114− :: AsyncE :> es
102+ :: Concurrent :> es
115103 => ((forall b. Eff es b -> Eff es b) -> Eff es a)
116104 -> Eff es (Async a)
117105asyncWithUnmask = liftAsyncWithUnmask A.asyncWithUnmask
118106
119107-- | Lifted 'A.asyncOnWithUnmask'.
120108asyncOnWithUnmask
121− :: AsyncE :> es
109+ :: Concurrent :> es
122110 => Int
123111 -> ((forall b. Eff es b -> Eff es b) -> Eff es a)
124112 -> Eff es (Async a)
@@ -126,7 +114,7 @@ asyncOnWithUnmask cpu = liftAsyncWithUnmask (A.asyncOnWithUnmask cpu)
126114
127115-- | Lifted 'A.withAsync'.
128116withAsync
129− :: AsyncE :> es
117+ :: Concurrent :> es
130118 => Eff es a
131119 -> (Async a -> Eff es b)
132120 -> Eff es b
@@ -134,7 +122,7 @@ withAsync = liftWithAsync A.withAsync
134122
135123-- | Lifted 'A.withAsyncBound'.
136124withAsyncBound
137− :: AsyncE :> es
125+ :: Concurrent :> es
138126 => Eff es a
139127 -> (Async a -> Eff es b)
140128 -> Eff es b
@@ -142,7 +130,7 @@ withAsyncBound = liftWithAsync A.withAsyncBound
142130
143131-- | Lifted 'A.withAsyncOn'.
144132withAsyncOn
145− :: AsyncE :> es
133+ :: Concurrent :> es
146134 => Int
147135 -> Eff es a
148136 -> (Async a -> Eff es b)
@@ -151,7 +139,7 @@ withAsyncOn cpu = liftWithAsync (A.withAsyncOn cpu)
151139
152140-- | Lifted 'A.withAsyncWithUnmask'.
153141withAsyncWithUnmask
154− :: AsyncE :> es
142+ :: Concurrent :> es
155143 => ((forall c. Eff es c -> Eff es c) -> Eff es a)
156144 -> (Async a -> Eff es b)
157145 -> Eff es b
@@ -159,7 +147,7 @@ withAsyncWithUnmask = liftWithAsyncWithUnmask A.withAsyncWithUnmask
159147
160148-- | Lifted 'A.withAsyncOnWithUnmask'.
161149withAsyncOnWithUnmask
162− :: AsyncE :> es
150+ :: Concurrent :> es
163151 => Int
164152 -> ((forall c. Eff es c -> Eff es c) -> Eff es a)
165153 -> (Async a -> Eff es b)
@@ -167,60 +155,60 @@ withAsyncOnWithUnmask
167155withAsyncOnWithUnmask cpu = liftWithAsyncWithUnmask (A.withAsyncOnWithUnmask cpu)
168156
169157-- | Lifted 'A.wait'.
170−wait :: AsyncE :> es => Async a -> Eff es a
158+wait :: Concurrent :> es => Async a -> Eff es a
171159wait = unsafeEff_ . A.wait
172160
173161-- | Lifted 'A.poll'.
174162poll
175− :: AsyncE :> es
163+ :: Concurrent :> es
176164 => Async a
177165 -> Eff es (Maybe (Either SomeException a))
178166poll = unsafeEff_ . A.poll
179167
180168-- | Lifted 'A.cancel'.
181−cancel :: AsyncE :> es => Async a -> Eff es ()
169+cancel :: Concurrent :> es => Async a -> Eff es ()
182170cancel = unsafeEff_ . A.cancel
183171
184172-- | Lifted 'A.cancelWith'.
185−cancelWith :: (Exception e, AsyncE :> es) => Async a -> e -> Eff es ()
173+cancelWith :: (Exception e, Concurrent :> es) => Async a -> e -> Eff es ()
186174cancelWith a = unsafeEff_ . A.cancelWith a
187175
188176-- | Lifted 'A.uninterruptibleCancel'.
189−uninterruptibleCancel :: AsyncE :> es => Async a -> Eff es ()
177+uninterruptibleCancel :: Concurrent :> es => Async a -> Eff es ()
190178uninterruptibleCancel = unsafeEff_ . A.uninterruptibleCancel
191179
192180-- | Lifted 'A.waitCatch'.
193181waitCatch
194− :: AsyncE :> es
182+ :: Concurrent :> es
195183 => Async a
196184 -> Eff es (Either SomeException a)
197185waitCatch = unsafeEff_ . A.waitCatch
198186
199187-- | Lifted 'A.waitAny'.
200−waitAny :: AsyncE :> es => [Async a] -> Eff es (Async a, a)
188+waitAny :: Concurrent :> es => [Async a] -> Eff es (Async a, a)
201189waitAny = unsafeEff_ . A.waitAny
202190
203191-- | Lifted 'A.waitAnyCatch'.
204192waitAnyCatch
205− :: AsyncE :> es
193+ :: Concurrent :> es
206194 => [Async a]
207195 -> Eff es (Async a, Either SomeException a)
208196waitAnyCatch = unsafeEff_ . A.waitAnyCatch
209197
210198-- | Lifted 'A.waitAnyCancel'.
211−waitAnyCancel :: AsyncE :> es => [Async a] -> Eff es (Async a, a)
199+waitAnyCancel :: Concurrent :> es => [Async a] -> Eff es (Async a, a)
212200waitAnyCancel = unsafeEff_ . A.waitAnyCancel
213201
214202-- | Lifted 'A.waitAnyCatchCancel'.
215203waitAnyCatchCancel
216− :: AsyncE :> es
204+ :: Concurrent :> es
217205 => [Async a]
218206 -> Eff es (Async a, Either SomeException a)
219207waitAnyCatchCancel = unsafeEff_ . A.waitAnyCatchCancel
220208
221209-- | Lifted 'A.waitEither'.
222210waitEither
223− :: AsyncE :> es
211+ :: Concurrent :> es
224212 => Async a
225213 -> Async b
226214 -> Eff es (Either a b)
@@ -228,7 +216,7 @@ waitEither a b = unsafeEff_ $ A.waitEither a b
228216
229217-- | Lifted 'A.waitEitherCatch'.
230218waitEitherCatch
231− :: AsyncE :> es
219+ :: Concurrent :> es
232220 => Async a
233221 -> Async b
234222 -> Eff es (Either (Either SomeException a) (Either SomeException b))
@@ -236,7 +224,7 @@ waitEitherCatch a b = unsafeEff_ $ A.waitEitherCatch a b
236224
237225-- | Lifted 'A.waitEitherCancel'.
238226waitEitherCancel
239− :: AsyncE :> es
227+ :: Concurrent :> es
240228 => Async a
241229 -> Async b
242230 -> Eff es (Either a b)
@@ -244,53 +232,53 @@ waitEitherCancel a b = unsafeEff_ $ A.waitEitherCancel a b
244232
245233-- | Lifted 'A.waitEitherCatchCancel'.
246234waitEitherCatchCancel
247− :: AsyncE :> es
235+ :: Concurrent :> es
248236 => Async a
249237 -> Async b
250238 -> Eff es (Either (Either SomeException a) (Either SomeException b))
251239waitEitherCatchCancel a b = unsafeEff_ $ A.waitEitherCatch a b
252240
253241-- | Lifted 'A.waitEither_'.
254−waitEither_ :: AsyncE :> es => Async a -> Async b -> Eff es ()
242+waitEither_ :: Concurrent :> es => Async a -> Async b -> Eff es ()
255243waitEither_ a b = unsafeEff_ $ A.waitEither_ a b
256244
257245-- | Lifted 'A.waitBoth'.
258−waitBoth :: AsyncE :> es => Async a -> Async b -> Eff es (a, b)
246+waitBoth :: Concurrent :> es => Async a -> Async b -> Eff es (a, b)
259247waitBoth a b = unsafeEff_ $ A.waitBoth a b
260248
261249-- | Lifted 'A.link'.
262−link :: AsyncE :> es => Async a -> Eff es ()
250+link :: Concurrent :> es => Async a -> Eff es ()
263251link = unsafeEff_ . A.link
264252
265253-- | Lifted 'A.linkOnly'.
266−linkOnly :: AsyncE :> es => (SomeException -> Bool) -> Async a -> Eff es ()
254+linkOnly :: Concurrent :> es => (SomeException -> Bool) -> Async a -> Eff es ()
267255linkOnly f = unsafeEff_ . A.linkOnly f
268256
269257-- | Lifted 'A.link2'.
270−link2 :: AsyncE :> es => Async a -> Async b -> Eff es ()
258+link2 :: Concurrent :> es => Async a -> Async b -> Eff es ()
271259link2 a b = unsafeEff_ $ A.link2 a b
272260
273261-- | Lifted 'A.link2Only'.
274−link2Only :: AsyncE :> es => (SomeException -> Bool) -> Async a -> Async b -> Eff es ()
262+link2Only :: Concurrent :> es => (SomeException -> Bool) -> Async a -> Async b -> Eff es ()
275263link2Only f a b = unsafeEff_ $ A.link2Only f a b
276264
277265-- | Lifted 'A.race'.
278−race :: AsyncE :> es => Eff es a -> Eff es b -> Eff es (Either a b)
266+race :: Concurrent :> es => Eff es a -> Eff es b -> Eff es (Either a b)
279267race ma mb = unsafeEff $ \es -> do
280268 A.race (unEff ma =<< cloneEnv es) (unEff mb =<< cloneEnv es)
281269
282270-- | Lifted 'A.race_'.
283−race_ :: AsyncE :> es => Eff es a -> Eff es b -> Eff es ()
271+race_ :: Concurrent :> es => Eff es a -> Eff es b -> Eff es ()
284272race_ ma mb = unsafeEff $ \es -> do
285273 A.race_ (unEff ma =<< cloneEnv es) (unEff mb =<< cloneEnv es)
286274
287275-- | Lifted 'A.concurrently'.
288−concurrently :: AsyncE :> es => Eff es a -> Eff es b -> Eff es (a, b)
276+concurrently :: Concurrent :> es => Eff es a -> Eff es b -> Eff es (a, b)
289277concurrently ma mb = unsafeEff $ \es -> do
290278 A.concurrently (unEff ma =<< cloneEnv es) (unEff mb =<< cloneEnv es)
291279
292280-- | Lifted 'A.concurrently_'.
293−concurrently_ :: AsyncE :> es => Eff es a -> Eff es b -> Eff es ()
281+concurrently_ :: Concurrent :> es => Eff es a -> Eff es b -> Eff es ()
294282concurrently_ ma mb = unsafeEff $ \es -> do
295283 A.concurrently_ (unEff ma =<< cloneEnv es) (unEff mb =<< cloneEnv es)
296284
@@ -300,7 +288,7 @@ concurrently_ ma mb = unsafeEff $ \es -> do
300288
301289-- | Lifted 'A.mapConcurrently'.
302290mapConcurrently
303− :: (Traversable f, AsyncE :> es)
291+ :: (Traversable f, Concurrent :> es)
304292 => (a -> Eff es b)
305293 -> f a
306294 -> Eff es (f b)
@@ -309,7 +297,7 @@ mapConcurrently f t = unsafeEff $ \es -> do
309297
310298-- | Lifted 'A.mapConcurrently_'.
311299mapConcurrently_
312− :: (Foldable f, AsyncE :> es)
300+ :: (Foldable f, Concurrent :> es)
313301 => (a -> Eff es b)
314302 -> f a
315303 -> Eff es ()
@@ -318,7 +306,7 @@ mapConcurrently_ f t = unsafeEff $ \es -> do
318306
319307-- | Lifted 'A.forConcurrently'.
320308forConcurrently
321− :: (Traversable f, AsyncE :> es)
309+ :: (Traversable f, Concurrent :> es)
322310 => f a
323311 -> (a -> Eff es b)
324312 -> Eff es (f b)
@@ -327,7 +315,7 @@ forConcurrently t f = unsafeEff $ \es -> do
327315
328316-- | Lifted 'A.forConcurrently_'.
329317forConcurrently_
330− :: (Foldable f, AsyncE :> es)
318+ :: (Foldable f, Concurrent :> es)
331319 => f a
332320 -> (a -> Eff es b)
333321 -> Eff es ()
@@ -335,12 +323,12 @@ forConcurrently_ t f = unsafeEff $ \es -> do
335323 U.forConcurrently_ t (\a -> unEff (f a) =<< cloneEnv es)
336324
337325-- | Lifted 'A.replicateConcurrently'.
338−replicateConcurrently :: AsyncE :> es => Int -> Eff es a -> Eff es [a]
326+replicateConcurrently :: Concurrent :> es => Int -> Eff es a -> Eff es [a]
339327replicateConcurrently n f = unsafeEff $ \es -> do
340328 U.replicateConcurrently n (unEff f =<< cloneEnv es)
341329
342330-- | Lifted 'A.replicateConcurrently_'.
343−replicateConcurrently_ :: AsyncE :> es => Int -> Eff es a -> Eff es ()
331+replicateConcurrently_ :: Concurrent :> es => Int -> Eff es a -> Eff es ()
344332replicateConcurrently_ n f = unsafeEff $ \es -> do
345333 U.replicateConcurrently_ n (unEff f =<< cloneEnv es)
346334
@@ -349,7 +337,7 @@ replicateConcurrently_ n f = unsafeEff $ \es -> do
349337
350338-- | Lifted 'U.pooledMapConcurrentlyN'.
351339pooledMapConcurrentlyN
352− :: (AsyncE :> es, Traversable t)
340+ :: (Concurrent :> es, Traversable t)
353341 => Int
354342 -> (a -> Eff es b)
355343 -> t a
@@ -359,7 +347,7 @@ pooledMapConcurrentlyN threads f t = unsafeEff $ \es -> do
359347
360348-- | Lifted 'U.pooledMapConcurrently'.
361349pooledMapConcurrently
362− :: (AsyncE :> es, Traversable t)
350+ :: (Concurrent :> es, Traversable t)
363351 => (a -> Eff es b)
364352 -> t a
365353 -> Eff es (t b)
@@ -368,7 +356,7 @@ pooledMapConcurrently f t = unsafeEff $ \es -> do
368356
369357-- | Lifted 'U.pooledMapConcurrentlyN'.
370358pooledMapConcurrentlyN_
371− :: (AsyncE :> es, Foldable f)
359+ :: (Concurrent :> es, Foldable f)
372360 => Int
373361 -> (a -> Eff es b)
374362 -> f a
@@ -378,7 +366,7 @@ pooledMapConcurrentlyN_ threads f t = unsafeEff $ \es -> do
378366
379367-- | Lifted 'U.pooledMapConcurrently_'.
380368pooledMapConcurrently_
381− :: (AsyncE :> es, Foldable f)
369+ :: (Concurrent :> es, Foldable f)
382370 => (a -> Eff es b)
383371 -> f a
384372 -> Eff es ()
@@ -387,7 +375,7 @@ pooledMapConcurrently_ f t = unsafeEff $ \es -> do
387375
388376-- | Lifted 'U.pooledForConcurrentlyN'.
389377pooledForConcurrentlyN
390− :: (AsyncE :> es, Traversable t)
378+ :: (Concurrent :> es, Traversable t)
391379 => Int
392380 -> t a
393381 -> (a -> Eff es b)
@@ -397,7 +385,7 @@ pooledForConcurrentlyN threads t f = unsafeEff $ \es -> do
397385
398386-- | Lifted 'U.pooledForConcurrently'.
399387pooledForConcurrently
400− :: (AsyncE :> es, Traversable t)
388+ :: (Concurrent :> es, Traversable t)
401389 => t a
402390 -> (a -> Eff es b)
403391 -> Eff es (t b)
@@ -406,7 +394,7 @@ pooledForConcurrently t f = unsafeEff $ \es -> do
406394
407395-- | Lifted 'U.pooledForConcurrentlyN'.
408396pooledForConcurrentlyN_
409− :: (AsyncE :> es, Foldable f)
397+ :: (Concurrent :> es, Foldable f)
410398 => Int
411399 -> f a
412400 -> (a -> Eff es b)
@@ -416,7 +404,7 @@ pooledForConcurrentlyN_ threads t f = unsafeEff $ \es -> do
416404
417405-- | Lifted 'U.pooledForConcurrently_'.
418406pooledForConcurrently_
419− :: (AsyncE :> es, Foldable f)
407+ :: (Concurrent :> es, Foldable f)
420408 => f a
421409 -> (a -> Eff es b)
422410 -> Eff es ()
@@ -424,22 +412,22 @@ pooledForConcurrently_ t f = unsafeEff $ \es -> do
424412 U.pooledForConcurrently_ t (\a -> unEff (f a) =<< cloneEnv es)
425413
426414-- | Lifted 'U.pooledReplicateConcurrentlyN'.
427−pooledReplicateConcurrentlyN :: AsyncE :> es => Int -> Int -> Eff es a -> Eff es [a]
415+pooledReplicateConcurrentlyN :: Concurrent :> es => Int -> Int -> Eff es a -> Eff es [a]
428416pooledReplicateConcurrentlyN threads n f = unsafeEff $ \es -> do
429417 U.pooledReplicateConcurrentlyN threads n (unEff f =<< cloneEnv es)
430418
431419-- | Lifted 'U.pooledReplicateConcurrently'.
432−pooledReplicateConcurrently :: AsyncE :> es => Int -> Eff es a -> Eff es [a]
420+pooledReplicateConcurrently :: Concurrent :> es => Int -> Eff es a -> Eff es [a]
433421pooledReplicateConcurrently n f = unsafeEff $ \es -> do
434422 U.pooledReplicateConcurrently n (unEff f =<< cloneEnv es)
435423
436424-- | Lifted 'U.pooledReplicateConcurrentlyN_'.
437−pooledReplicateConcurrentlyN_ :: AsyncE :> es => Int -> Int -> Eff es a -> Eff es ()
425+pooledReplicateConcurrentlyN_ :: Concurrent :> es => Int -> Int -> Eff es a -> Eff es ()
438426pooledReplicateConcurrentlyN_ threads n f = unsafeEff $ \es -> do
439427 U.pooledReplicateConcurrentlyN_ threads n (unEff f =<< cloneEnv es)
440428
441429-- | Lifted 'U.pooledReplicateConcurrently_'.
442−pooledReplicateConcurrently_ :: AsyncE :> es => Int -> Eff es a -> Eff es ()
430+pooledReplicateConcurrently_ :: Concurrent :> es => Int -> Eff es a -> Eff es ()
443431pooledReplicateConcurrently_ n f = unsafeEff $ \es -> do
444432 U.pooledReplicateConcurrently_ n (unEff f =<< cloneEnv es)
445433
@@ -447,7 +435,7 @@ pooledReplicateConcurrently_ n f = unsafeEff $ \es -> do
447435-- Conc
448436
449437-- | Lifted 'U.Conc'.
450−data Conc es a where
438+data Conc :: [Effect] -> Type -> Type where
451439 Action :: Eff es a -> Conc es a
452440 Apply :: Conc es (v -> a) -> Conc es v -> Conc es a
453441 LiftA2 :: (x -> y -> a) -> Conc es x -> Conc es y -> Conc es a
@@ -479,7 +467,7 @@ conc :: Eff es a -> Conc es a
479467conc = Action
480468
481469-- | Lifted 'U.runConc'.
482−runConc :: AsyncE :> es => Conc es a -> Eff es a
470+runConc :: Concurrent :> es => Conc es a -> Eff es a
483471runConc m = unsafeEff $ \es -> U.runConc (unliftConc es m)
484472 where
485473 unliftConc :: Env es -> Conc es a -> U.Conc IO a
@@ -500,20 +488,20 @@ newtype Concurrently es a = Concurrently { runConcurrently :: Eff es a }
500488instance Functor (Concurrently es) where
501489 fmap f (Concurrently a) = Concurrently (fmap f a)
502490
503−instance AsyncE :> es => Applicative (Concurrently es) where
491+instance Concurrent :> es => Applicative (Concurrently es) where
504492 pure = Concurrently . pure
505493 Concurrently fs <*> Concurrently as =
506494 Concurrently ((\(f, a) -> f a) <$> concurrently fs as)
507495
508−instance AsyncE :> es => Alternative (Concurrently es) where
496+instance Concurrent :> es => Alternative (Concurrently es) where
509497 empty = Concurrently . unsafeEff_ . forever $ threadDelay maxBound
510498 Concurrently as <|> Concurrently bs =
511499 Concurrently (either id id <$> race as bs)
512500
513−instance (AsyncE :> es, Semigroup a) => Semigroup (Concurrently es a) where
501+instance (Concurrent :> es, Semigroup a) => Semigroup (Concurrently es a) where
514502 (<>) = liftA2 (<>)
515503
516−instance (AsyncE :> es, Monoid a) => Monoid (Concurrently es a) where
504+instance (Concurrent :> es, Monoid a) => Monoid (Concurrently es a) where
517505 mempty = pure mempty
518506
519507----------------------------------------
@@ -531,10 +519,9 @@ liftAsyncWithUnmask
531519 :: (((forall b. IO b -> IO b) -> IO a) -> IO (Async a))
532520 -> ((forall b. Eff es b -> Eff es b) -> Eff es a)
533521 -> Eff es (Async a)
534−liftAsyncWithUnmask fork action = do
535− unsafeWithLiftMapIO $ \liftMap -> unsafeEff $ \es -> do
536− esA <- cloneEnv es
537− fork $ \unmask -> unEff (action $ liftMap unmask) esA
522+liftAsyncWithUnmask fork action = unsafeEff $ \es -> do
523+ esA <- cloneEnv es
524+ fork $ \unmask -> unEff (action $ unsafeLiftMapIO unmask) esA
538525
539526liftWithAsync
540527 :: (IO a -> (Async a -> IO b) -> IO b)
@@ -551,8 +538,7 @@ liftWithAsyncWithUnmask
551538 -> ((forall c. Eff es c -> Eff es c) -> Eff es a)
552539 -> (Async a -> Eff es b)
553540 -> Eff es b
554−liftWithAsyncWithUnmask withA action k = do
555− unsafeWithLiftMapIO $ \liftMap -> unsafeEff $ \es -> do
556− esA <- cloneEnv es
557− withA (\unmask -> unEff (action $ liftMap unmask) esA)
558− (\a -> unEff (k a) es)
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)