@@ -1,8 +1,8 @@
11{-# LANGUAGE UndecidableInstances #-}
2−-- | Lifted version of "Control.Concurrent.Async".
3−module Effectful.Concurrent.Async
4− ( Concurrent
5− , runConcurrent
2+module Effectful.Async
3+ ( -- * Async effect
4+ AsyncE
5+ , runAsyncE
66
77 -- * Asynchronous actions
88 , Async
@@ -76,37 +76,49 @@ import Control.Concurrent (threadDelay)
7676import Control.Concurrent.Async (Async)
7777import Control.Exception (Exception, SomeException)
7878import Control.Monad (forever)
79−import Data.Kind (Type)
8079import qualified Control.Concurrent.Async as A
8180import qualified UnliftIO.Async as U
8281import qualified UnliftIO.Internals.Async as I
8382
84−import Effectful.Concurrent.Effect
85−import Effectful.Dispatch.Static
86−import Effectful.Monad
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
8799
88100-- | Lifted 'A.async'.
89−async :: Concurrent :> es => Eff es a -> Eff es (Async a)
101+async :: AsyncE :> es => Eff es a -> Eff es (Async a)
90102async = liftAsync A.async
91103
92104-- | Lifted 'A.asyncBound'.
93−asyncBound :: Concurrent :> es => Eff es a -> Eff es (Async a)
105+asyncBound :: AsyncE :> es => Eff es a -> Eff es (Async a)
94106asyncBound = liftAsync A.asyncBound
95107
96108-- | Lifted 'A.asyncOn'.
97−asyncOn :: Concurrent :> es => Int -> Eff es a -> Eff es (Async a)
109+asyncOn :: AsyncE :> es => Int -> Eff es a -> Eff es (Async a)
98110asyncOn cpu = liftAsync (A.asyncOn cpu)
99111
100112-- | Lifted 'A.asyncWithUnmask'.
101113asyncWithUnmask
102− :: Concurrent :> es
114+ :: AsyncE :> es
103115 => ((forall b. Eff es b -> Eff es b) -> Eff es a)
104116 -> Eff es (Async a)
105117asyncWithUnmask = liftAsyncWithUnmask A.asyncWithUnmask
106118
107119-- | Lifted 'A.asyncOnWithUnmask'.
108120asyncOnWithUnmask
109− :: Concurrent :> es
121+ :: AsyncE :> es
110122 => Int
111123 -> ((forall b. Eff es b -> Eff es b) -> Eff es a)
112124 -> Eff es (Async a)
@@ -114,7 +126,7 @@ asyncOnWithUnmask cpu = liftAsyncWithUnmask (A.asyncOnWithUnmask cpu)
114126
115127-- | Lifted 'A.withAsync'.
116128withAsync
117− :: Concurrent :> es
129+ :: AsyncE :> es
118130 => Eff es a
119131 -> (Async a -> Eff es b)
120132 -> Eff es b
@@ -122,7 +134,7 @@ withAsync = liftWithAsync A.withAsync
122134
123135-- | Lifted 'A.withAsyncBound'.
124136withAsyncBound
125− :: Concurrent :> es
137+ :: AsyncE :> es
126138 => Eff es a
127139 -> (Async a -> Eff es b)
128140 -> Eff es b
@@ -130,7 +142,7 @@ withAsyncBound = liftWithAsync A.withAsyncBound
130142
131143-- | Lifted 'A.withAsyncOn'.
132144withAsyncOn
133− :: Concurrent :> es
145+ :: AsyncE :> es
134146 => Int
135147 -> Eff es a
136148 -> (Async a -> Eff es b)
@@ -139,7 +151,7 @@ withAsyncOn cpu = liftWithAsync (A.withAsyncOn cpu)
139151
140152-- | Lifted 'A.withAsyncWithUnmask'.
141153withAsyncWithUnmask
142− :: Concurrent :> es
154+ :: AsyncE :> es
143155 => ((forall c. Eff es c -> Eff es c) -> Eff es a)
144156 -> (Async a -> Eff es b)
145157 -> Eff es b
@@ -147,7 +159,7 @@ withAsyncWithUnmask = liftWithAsyncWithUnmask A.withAsyncWithUnmask
147159
148160-- | Lifted 'A.withAsyncOnWithUnmask'.
149161withAsyncOnWithUnmask
150− :: Concurrent :> es
162+ :: AsyncE :> es
151163 => Int
152164 -> ((forall c. Eff es c -> Eff es c) -> Eff es a)
153165 -> (Async a -> Eff es b)
@@ -155,60 +167,60 @@ withAsyncOnWithUnmask
155167withAsyncOnWithUnmask cpu = liftWithAsyncWithUnmask (A.withAsyncOnWithUnmask cpu)
156168
157169-- | Lifted 'A.wait'.
158−wait :: Concurrent :> es => Async a -> Eff es a
170+wait :: AsyncE :> es => Async a -> Eff es a
159171wait = unsafeEff_ . A.wait
160172
161173-- | Lifted 'A.poll'.
162174poll
163− :: Concurrent :> es
175+ :: AsyncE :> es
164176 => Async a
165177 -> Eff es (Maybe (Either SomeException a))
166178poll = unsafeEff_ . A.poll
167179
168180-- | Lifted 'A.cancel'.
169−cancel :: Concurrent :> es => Async a -> Eff es ()
181+cancel :: AsyncE :> es => Async a -> Eff es ()
170182cancel = unsafeEff_ . A.cancel
171183
172184-- | Lifted 'A.cancelWith'.
173−cancelWith :: (Exception e, Concurrent :> es) => Async a -> e -> Eff es ()
185+cancelWith :: (Exception e, AsyncE :> es) => Async a -> e -> Eff es ()
174186cancelWith a = unsafeEff_ . A.cancelWith a
175187
176188-- | Lifted 'A.uninterruptibleCancel'.
177−uninterruptibleCancel :: Concurrent :> es => Async a -> Eff es ()
189+uninterruptibleCancel :: AsyncE :> es => Async a -> Eff es ()
178190uninterruptibleCancel = unsafeEff_ . A.uninterruptibleCancel
179191
180192-- | Lifted 'A.waitCatch'.
181193waitCatch
182− :: Concurrent :> es
194+ :: AsyncE :> es
183195 => Async a
184196 -> Eff es (Either SomeException a)
185197waitCatch = unsafeEff_ . A.waitCatch
186198
187199-- | Lifted 'A.waitAny'.
188−waitAny :: Concurrent :> es => [Async a] -> Eff es (Async a, a)
200+waitAny :: AsyncE :> es => [Async a] -> Eff es (Async a, a)
189201waitAny = unsafeEff_ . A.waitAny
190202
191203-- | Lifted 'A.waitAnyCatch'.
192204waitAnyCatch
193− :: Concurrent :> es
205+ :: AsyncE :> es
194206 => [Async a]
195207 -> Eff es (Async a, Either SomeException a)
196208waitAnyCatch = unsafeEff_ . A.waitAnyCatch
197209
198210-- | Lifted 'A.waitAnyCancel'.
199−waitAnyCancel :: Concurrent :> es => [Async a] -> Eff es (Async a, a)
211+waitAnyCancel :: AsyncE :> es => [Async a] -> Eff es (Async a, a)
200212waitAnyCancel = unsafeEff_ . A.waitAnyCancel
201213
202214-- | Lifted 'A.waitAnyCatchCancel'.
203215waitAnyCatchCancel
204− :: Concurrent :> es
216+ :: AsyncE :> es
205217 => [Async a]
206218 -> Eff es (Async a, Either SomeException a)
207219waitAnyCatchCancel = unsafeEff_ . A.waitAnyCatchCancel
208220
209221-- | Lifted 'A.waitEither'.
210222waitEither
211− :: Concurrent :> es
223+ :: AsyncE :> es
212224 => Async a
213225 -> Async b
214226 -> Eff es (Either a b)
@@ -216,7 +228,7 @@ waitEither a b = unsafeEff_ $ A.waitEither a b
216228
217229-- | Lifted 'A.waitEitherCatch'.
218230waitEitherCatch
219− :: Concurrent :> es
231+ :: AsyncE :> es
220232 => Async a
221233 -> Async b
222234 -> Eff es (Either (Either SomeException a) (Either SomeException b))
@@ -224,7 +236,7 @@ waitEitherCatch a b = unsafeEff_ $ A.waitEitherCatch a b
224236
225237-- | Lifted 'A.waitEitherCancel'.
226238waitEitherCancel
227− :: Concurrent :> es
239+ :: AsyncE :> es
228240 => Async a
229241 -> Async b
230242 -> Eff es (Either a b)
@@ -232,53 +244,53 @@ waitEitherCancel a b = unsafeEff_ $ A.waitEitherCancel a b
232244
233245-- | Lifted 'A.waitEitherCatchCancel'.
234246waitEitherCatchCancel
235− :: Concurrent :> es
247+ :: AsyncE :> es
236248 => Async a
237249 -> Async b
238250 -> Eff es (Either (Either SomeException a) (Either SomeException b))
239251waitEitherCatchCancel a b = unsafeEff_ $ A.waitEitherCatch a b
240252
241253-- | Lifted 'A.waitEither_'.
242−waitEither_ :: Concurrent :> es => Async a -> Async b -> Eff es ()
254+waitEither_ :: AsyncE :> es => Async a -> Async b -> Eff es ()
243255waitEither_ a b = unsafeEff_ $ A.waitEither_ a b
244256
245257-- | Lifted 'A.waitBoth'.
246−waitBoth :: Concurrent :> es => Async a -> Async b -> Eff es (a, b)
258+waitBoth :: AsyncE :> es => Async a -> Async b -> Eff es (a, b)
247259waitBoth a b = unsafeEff_ $ A.waitBoth a b
248260
249261-- | Lifted 'A.link'.
250−link :: Concurrent :> es => Async a -> Eff es ()
262+link :: AsyncE :> es => Async a -> Eff es ()
251263link = unsafeEff_ . A.link
252264
253265-- | Lifted 'A.linkOnly'.
254−linkOnly :: Concurrent :> es => (SomeException -> Bool) -> Async a -> Eff es ()
266+linkOnly :: AsyncE :> es => (SomeException -> Bool) -> Async a -> Eff es ()
255267linkOnly f = unsafeEff_ . A.linkOnly f
256268
257269-- | Lifted 'A.link2'.
258−link2 :: Concurrent :> es => Async a -> Async b -> Eff es ()
270+link2 :: AsyncE :> es => Async a -> Async b -> Eff es ()
259271link2 a b = unsafeEff_ $ A.link2 a b
260272
261273-- | Lifted 'A.link2Only'.
262−link2Only :: Concurrent :> es => (SomeException -> Bool) -> Async a -> Async b -> Eff es ()
274+link2Only :: AsyncE :> es => (SomeException -> Bool) -> Async a -> Async b -> Eff es ()
263275link2Only f a b = unsafeEff_ $ A.link2Only f a b
264276
265277-- | Lifted 'A.race'.
266−race :: Concurrent :> es => Eff es a -> Eff es b -> Eff es (Either a b)
278+race :: AsyncE :> es => Eff es a -> Eff es b -> Eff es (Either a b)
267279race ma mb = unsafeEff $ \es -> do
268280 A.race (unEff ma =<< cloneEnv es) (unEff mb =<< cloneEnv es)
269281
270282-- | Lifted 'A.race_'.
271−race_ :: Concurrent :> es => Eff es a -> Eff es b -> Eff es ()
283+race_ :: AsyncE :> es => Eff es a -> Eff es b -> Eff es ()
272284race_ ma mb = unsafeEff $ \es -> do
273285 A.race_ (unEff ma =<< cloneEnv es) (unEff mb =<< cloneEnv es)
274286
275287-- | Lifted 'A.concurrently'.
276−concurrently :: Concurrent :> es => Eff es a -> Eff es b -> Eff es (a, b)
288+concurrently :: AsyncE :> es => Eff es a -> Eff es b -> Eff es (a, b)
277289concurrently ma mb = unsafeEff $ \es -> do
278290 A.concurrently (unEff ma =<< cloneEnv es) (unEff mb =<< cloneEnv es)
279291
280292-- | Lifted 'A.concurrently_'.
281−concurrently_ :: Concurrent :> es => Eff es a -> Eff es b -> Eff es ()
293+concurrently_ :: AsyncE :> es => Eff es a -> Eff es b -> Eff es ()
282294concurrently_ ma mb = unsafeEff $ \es -> do
283295 A.concurrently_ (unEff ma =<< cloneEnv es) (unEff mb =<< cloneEnv es)
284296
@@ -288,7 +300,7 @@ concurrently_ ma mb = unsafeEff $ \es -> do
288300
289301-- | Lifted 'A.mapConcurrently'.
290302mapConcurrently
291− :: (Traversable f, Concurrent :> es)
303+ :: (Traversable f, AsyncE :> es)
292304 => (a -> Eff es b)
293305 -> f a
294306 -> Eff es (f b)
@@ -297,7 +309,7 @@ mapConcurrently f t = unsafeEff $ \es -> do
297309
298310-- | Lifted 'A.mapConcurrently_'.
299311mapConcurrently_
300− :: (Foldable f, Concurrent :> es)
312+ :: (Foldable f, AsyncE :> es)
301313 => (a -> Eff es b)
302314 -> f a
303315 -> Eff es ()
@@ -306,7 +318,7 @@ mapConcurrently_ f t = unsafeEff $ \es -> do
306318
307319-- | Lifted 'A.forConcurrently'.
308320forConcurrently
309− :: (Traversable f, Concurrent :> es)
321+ :: (Traversable f, AsyncE :> es)
310322 => f a
311323 -> (a -> Eff es b)
312324 -> Eff es (f b)
@@ -315,7 +327,7 @@ forConcurrently t f = unsafeEff $ \es -> do
315327
316328-- | Lifted 'A.forConcurrently_'.
317329forConcurrently_
318− :: (Foldable f, Concurrent :> es)
330+ :: (Foldable f, AsyncE :> es)
319331 => f a
320332 -> (a -> Eff es b)
321333 -> Eff es ()
@@ -323,12 +335,12 @@ forConcurrently_ t f = unsafeEff $ \es -> do
323335 U.forConcurrently_ t (\a -> unEff (f a) =<< cloneEnv es)
324336
325337-- | Lifted 'A.replicateConcurrently'.
326−replicateConcurrently :: Concurrent :> es => Int -> Eff es a -> Eff es [a]
338+replicateConcurrently :: AsyncE :> es => Int -> Eff es a -> Eff es [a]
327339replicateConcurrently n f = unsafeEff $ \es -> do
328340 U.replicateConcurrently n (unEff f =<< cloneEnv es)
329341
330342-- | Lifted 'A.replicateConcurrently_'.
331−replicateConcurrently_ :: Concurrent :> es => Int -> Eff es a -> Eff es ()
343+replicateConcurrently_ :: AsyncE :> es => Int -> Eff es a -> Eff es ()
332344replicateConcurrently_ n f = unsafeEff $ \es -> do
333345 U.replicateConcurrently_ n (unEff f =<< cloneEnv es)
334346
@@ -337,7 +349,7 @@ replicateConcurrently_ n f = unsafeEff $ \es -> do
337349
338350-- | Lifted 'U.pooledMapConcurrentlyN'.
339351pooledMapConcurrentlyN
340− :: (Concurrent :> es, Traversable t)
352+ :: (AsyncE :> es, Traversable t)
341353 => Int
342354 -> (a -> Eff es b)
343355 -> t a
@@ -347,7 +359,7 @@ pooledMapConcurrentlyN threads f t = unsafeEff $ \es -> do
347359
348360-- | Lifted 'U.pooledMapConcurrently'.
349361pooledMapConcurrently
350− :: (Concurrent :> es, Traversable t)
362+ :: (AsyncE :> es, Traversable t)
351363 => (a -> Eff es b)
352364 -> t a
353365 -> Eff es (t b)
@@ -356,7 +368,7 @@ pooledMapConcurrently f t = unsafeEff $ \es -> do
356368
357369-- | Lifted 'U.pooledMapConcurrentlyN'.
358370pooledMapConcurrentlyN_
359− :: (Concurrent :> es, Foldable f)
371+ :: (AsyncE :> es, Foldable f)
360372 => Int
361373 -> (a -> Eff es b)
362374 -> f a
@@ -366,7 +378,7 @@ pooledMapConcurrentlyN_ threads f t = unsafeEff $ \es -> do
366378
367379-- | Lifted 'U.pooledMapConcurrently_'.
368380pooledMapConcurrently_
369− :: (Concurrent :> es, Foldable f)
381+ :: (AsyncE :> es, Foldable f)
370382 => (a -> Eff es b)
371383 -> f a
372384 -> Eff es ()
@@ -375,7 +387,7 @@ pooledMapConcurrently_ f t = unsafeEff $ \es -> do
375387
376388-- | Lifted 'U.pooledForConcurrentlyN'.
377389pooledForConcurrentlyN
378− :: (Concurrent :> es, Traversable t)
390+ :: (AsyncE :> es, Traversable t)
379391 => Int
380392 -> t a
381393 -> (a -> Eff es b)
@@ -385,7 +397,7 @@ pooledForConcurrentlyN threads t f = unsafeEff $ \es -> do
385397
386398-- | Lifted 'U.pooledForConcurrently'.
387399pooledForConcurrently
388− :: (Concurrent :> es, Traversable t)
400+ :: (AsyncE :> es, Traversable t)
389401 => t a
390402 -> (a -> Eff es b)
391403 -> Eff es (t b)
@@ -394,7 +406,7 @@ pooledForConcurrently t f = unsafeEff $ \es -> do
394406
395407-- | Lifted 'U.pooledForConcurrentlyN'.
396408pooledForConcurrentlyN_
397− :: (Concurrent :> es, Foldable f)
409+ :: (AsyncE :> es, Foldable f)
398410 => Int
399411 -> f a
400412 -> (a -> Eff es b)
@@ -404,7 +416,7 @@ pooledForConcurrentlyN_ threads t f = unsafeEff $ \es -> do
404416
405417-- | Lifted 'U.pooledForConcurrently_'.
406418pooledForConcurrently_
407− :: (Concurrent :> es, Foldable f)
419+ :: (AsyncE :> es, Foldable f)
408420 => f a
409421 -> (a -> Eff es b)
410422 -> Eff es ()
@@ -412,22 +424,22 @@ pooledForConcurrently_ t f = unsafeEff $ \es -> do
412424 U.pooledForConcurrently_ t (\a -> unEff (f a) =<< cloneEnv es)
413425
414426-- | Lifted 'U.pooledReplicateConcurrentlyN'.
415−pooledReplicateConcurrentlyN :: Concurrent :> es => Int -> Int -> Eff es a -> Eff es [a]
427+pooledReplicateConcurrentlyN :: AsyncE :> es => Int -> Int -> Eff es a -> Eff es [a]
416428pooledReplicateConcurrentlyN threads n f = unsafeEff $ \es -> do
417429 U.pooledReplicateConcurrentlyN threads n (unEff f =<< cloneEnv es)
418430
419431-- | Lifted 'U.pooledReplicateConcurrently'.
420−pooledReplicateConcurrently :: Concurrent :> es => Int -> Eff es a -> Eff es [a]
432+pooledReplicateConcurrently :: AsyncE :> es => Int -> Eff es a -> Eff es [a]
421433pooledReplicateConcurrently n f = unsafeEff $ \es -> do
422434 U.pooledReplicateConcurrently n (unEff f =<< cloneEnv es)
423435
424436-- | Lifted 'U.pooledReplicateConcurrentlyN_'.
425−pooledReplicateConcurrentlyN_ :: Concurrent :> es => Int -> Int -> Eff es a -> Eff es ()
437+pooledReplicateConcurrentlyN_ :: AsyncE :> es => Int -> Int -> Eff es a -> Eff es ()
426438pooledReplicateConcurrentlyN_ threads n f = unsafeEff $ \es -> do
427439 U.pooledReplicateConcurrentlyN_ threads n (unEff f =<< cloneEnv es)
428440
429441-- | Lifted 'U.pooledReplicateConcurrently_'.
430−pooledReplicateConcurrently_ :: Concurrent :> es => Int -> Eff es a -> Eff es ()
442+pooledReplicateConcurrently_ :: AsyncE :> es => Int -> Eff es a -> Eff es ()
431443pooledReplicateConcurrently_ n f = unsafeEff $ \es -> do
432444 U.pooledReplicateConcurrently_ n (unEff f =<< cloneEnv es)
433445
@@ -435,7 +447,7 @@ pooledReplicateConcurrently_ n f = unsafeEff $ \es -> do
435447-- Conc
436448
437449-- | Lifted 'U.Conc'.
438−data Conc :: [Effect] -> Type -> Type where
450+data Conc es a where
439451 Action :: Eff es a -> Conc es a
440452 Apply :: Conc es (v -> a) -> Conc es v -> Conc es a
441453 LiftA2 :: (x -> y -> a) -> Conc es x -> Conc es y -> Conc es a
@@ -467,7 +479,7 @@ conc :: Eff es a -> Conc es a
467479conc = Action
468480
469481-- | Lifted 'U.runConc'.
470−runConc :: Concurrent :> es => Conc es a -> Eff es a
482+runConc :: AsyncE :> es => Conc es a -> Eff es a
471483runConc m = unsafeEff $ \es -> U.runConc (unliftConc es m)
472484 where
473485 unliftConc :: Env es -> Conc es a -> U.Conc IO a
@@ -488,20 +500,20 @@ newtype Concurrently es a = Concurrently { runConcurrently :: Eff es a }
488500instance Functor (Concurrently es) where
489501 fmap f (Concurrently a) = Concurrently (fmap f a)
490502
491−instance Concurrent :> es => Applicative (Concurrently es) where
503+instance AsyncE :> es => Applicative (Concurrently es) where
492504 pure = Concurrently . pure
493505 Concurrently fs <*> Concurrently as =
494506 Concurrently ((\(f, a) -> f a) <$> concurrently fs as)
495507
496−instance Concurrent :> es => Alternative (Concurrently es) where
508+instance AsyncE :> es => Alternative (Concurrently es) where
497509 empty = Concurrently . unsafeEff_ . forever $ threadDelay maxBound
498510 Concurrently as <|> Concurrently bs =
499511 Concurrently (either id id <$> race as bs)
500512
501−instance (Concurrent :> es, Semigroup a) => Semigroup (Concurrently es a) where
513+instance (AsyncE :> es, Semigroup a) => Semigroup (Concurrently es a) where
502514 (<>) = liftA2 (<>)
503515
504−instance (Concurrent :> es, Monoid a) => Monoid (Concurrently es a) where
516+instance (AsyncE :> es, Monoid a) => Monoid (Concurrently es a) where
505517 mempty = pure mempty
506518
507519----------------------------------------
@@ -519,9 +531,10 @@ liftAsyncWithUnmask
519531 :: (((forall b. IO b -> IO b) -> IO a) -> IO (Async a))
520532 -> ((forall b. Eff es b -> Eff es b) -> Eff es a)
521533 -> Eff es (Async a)
522−liftAsyncWithUnmask fork action = unsafeEff $ \es -> do
523− esA <- cloneEnv es
524− fork $ \unmask -> unEff (action $ unsafeLiftMapIO unmask) esA
534+liftAsyncWithUnmask fork action = do
535+ unsafeWithLiftMapIO $ \liftMap -> unsafeEff $ \es -> do
536+ esA <- cloneEnv es
537+ fork $ \unmask -> unEff (action $ liftMap unmask) esA
525538
526539liftWithAsync
527540 :: (IO a -> (Async a -> IO b) -> IO b)
@@ -538,7 +551,8 @@ liftWithAsyncWithUnmask
538551 -> ((forall c. Eff es c -> Eff es c) -> Eff es a)
539552 -> (Async a -> Eff es b)
540553 -> 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)
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)