@@ -4,51 +4,70 @@ module Effectful.Async
44 AsyncE
55 , runAsyncE
66
7− -- * Asynchronous actions
7+ -- * Asynchronous actions
88 , Async
99
10− -- * High-level API
10+ -- * High-level API
1111
12− -- ** Spawning with automatic 'cancel'ation
12+ -- ** Spawning with automatic 'cancel'ation
1313 , withAsync, withAsyncBound, withAsyncOn, withAsyncWithUnmask
1414 , withAsyncOnWithUnmask
1515
16− -- ** Querying 'Async's
16+ -- ** Querying 'Async's
1717 , wait, poll, waitCatch, A.asyncThreadId
1818 , cancel, uninterruptibleCancel, cancelWith, A.AsyncCancelled(..)
19+ , A.compareAsyncs
1920
20− -- ** #high-level-utilities# High-level utilities
21+ -- ** High-level utilities
2122 , race, race_
2223 , concurrently, concurrently_
2324 , mapConcurrently, forConcurrently
2425 , mapConcurrently_, forConcurrently_
2526 , replicateConcurrently, replicateConcurrently_
26− , Concurrently(..)
27− , A.compareAsyncs
2827
29− -- ** Specialised operations
28+ -- *** Concurrently
29+ , Concurrently(..)
3030
31− -- *** STM operations
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
3251 , A.waitSTM, A.pollSTM, A.waitCatchSTM
3352
34− -- *** Waiting for multiple 'Async's
53+ -- *** Waiting for multiple 'Async's
3554 , waitAny, waitAnyCatch, waitAnyCancel, waitAnyCatchCancel
3655 , waitEither, waitEitherCatch, waitEitherCancel, waitEitherCatchCancel
3756 , waitEither_
3857 , waitBoth
3958
40− -- *** Waiting for multiple 'Async's in STM
59+ -- *** Waiting for multiple 'Async's in STM
4160 , A.waitAnySTM, A.waitAnyCatchSTM
4261 , A.waitEitherSTM, A.waitEitherCatchSTM
4362 , A.waitEitherSTM_
4463 , A.waitBothSTM
4564
46− -- * Low-level API
65+ -- * Low-level API
4766
48− -- ** Spawning (low-level API)
67+ -- ** Spawning (low-level API)
4968 , async, asyncBound, asyncOn, asyncWithUnmask, asyncOnWithUnmask
5069
51− -- ** Linking
70+ -- ** Linking
5271 , link, linkOnly, link2, link2Only, A.ExceptionInLinkedThread(..)
5372 ) where
5473
@@ -56,9 +75,10 @@ import Control.Applicative
5675import Control.Concurrent (threadDelay)
5776import Control.Concurrent.Async (Async)
5877import Control.Exception (Exception, SomeException)
59−import Control.Monad (forever, void)
60−import Data.Foldable (fold)
78+import Control.Monad (forever)
6179import qualified Control.Concurrent.Async as A
80+import qualified UnliftIO.Async as U
81+import qualified UnliftIO.Internals.Async as I
6282
6383import Effectful.Internal.Effect
6484import Effectful.Internal.Env
@@ -74,26 +94,29 @@ data AsyncE :: Effect where
7494runAsyncE :: IOE :> es => Eff (AsyncE : es) a -> Eff es a
7595runAsyncE = evalEffect (IdE AsyncE)
7696
77−-- | Generalized version of 'A.async'.
97+----------------------------------------
98+-- Operations
99+
100+-- | Lifted 'A.async'.
78101async :: AsyncE :> es => Eff es a -> Eff es (Async a)
79102async = liftAsync A.async
80103
81−-- | Generalized version of 'A.asyncBound'.
104+-- | Lifted 'A.asyncBound'.
82105asyncBound :: AsyncE :> es => Eff es a -> Eff es (Async a)
83106asyncBound = liftAsync A.asyncBound
84107
85−-- | Generalized version of 'A.asyncOn'.
108+-- | Lifted 'A.asyncOn'.
86109asyncOn :: AsyncE :> es => Int -> Eff es a -> Eff es (Async a)
87110asyncOn cpu = liftAsync (A.asyncOn cpu)
88111
89−-- | Generalized version of 'A.asyncWithUnmask'.
112+-- | Lifted 'A.asyncWithUnmask'.
90113asyncWithUnmask
91114 :: AsyncE :> es
92115 => ((forall b. Eff es b -> Eff es b) -> Eff es a)
93116 -> Eff es (Async a)
94117asyncWithUnmask = liftAsyncWithUnmask A.asyncWithUnmask
95118
96−-- | Generalized version of 'A.asyncOnWithUnmask'.
119+-- | Lifted 'A.asyncOnWithUnmask'.
97120asyncOnWithUnmask
98121 :: AsyncE :> es
99122 => Int
@@ -101,7 +124,7 @@ asyncOnWithUnmask
101124 -> Eff es (Async a)
102125asyncOnWithUnmask cpu = liftAsyncWithUnmask (A.asyncOnWithUnmask cpu)
103126
104−-- | Generalized version of 'A.withAsync'.
127+-- | Lifted 'A.withAsync'.
105128withAsync
106129 :: AsyncE :> es
107130 => Eff es a
@@ -109,7 +132,7 @@ withAsync
109132 -> Eff es b
110133withAsync = liftWithAsync A.withAsync
111134
112−-- | Generalized version of 'A.withAsyncBound'.
135+-- | Lifted 'A.withAsyncBound'.
113136withAsyncBound
114137 :: AsyncE :> es
115138 => Eff es a
@@ -117,7 +140,7 @@ withAsyncBound
117140 -> Eff es b
118141withAsyncBound = liftWithAsync A.withAsyncBound
119142
120−-- | Generalized version of 'A.withAsyncOn'.
143+-- | Lifted 'A.withAsyncOn'.
121144withAsyncOn
122145 :: AsyncE :> es
123146 => Int
@@ -126,7 +149,7 @@ withAsyncOn
126149 -> Eff es b
127150withAsyncOn cpu = liftWithAsync (A.withAsyncOn cpu)
128151
129−-- | Generalized version of 'A.withAsyncWithUnmask'.
152+-- | Lifted 'A.withAsyncWithUnmask'.
130153withAsyncWithUnmask
131154 :: AsyncE :> es
132155 => ((forall c. Eff es c -> Eff es c) -> Eff es a)
@@ -134,7 +157,7 @@ withAsyncWithUnmask
134157 -> Eff es b
135158withAsyncWithUnmask = liftWithAsyncWithUnmask A.withAsyncWithUnmask
136159
137−-- | Generalized version of 'A.withAsyncOnWithUnmask'.
160+-- | Lifted 'A.withAsyncOnWithUnmask'.
138161withAsyncOnWithUnmask
139162 :: AsyncE :> es
140163 => Int
@@ -143,59 +166,59 @@ withAsyncOnWithUnmask
143166 -> Eff es b
144167withAsyncOnWithUnmask cpu = liftWithAsyncWithUnmask (A.withAsyncOnWithUnmask cpu)
145168
146−-- | Generalized version of 'A.wait'.
169+-- | Lifted 'A.wait'.
147170wait :: AsyncE :> es => Async a -> Eff es a
148171wait = unsafeEff_ . A.wait
149172
150−-- | Generalized version of 'A.poll'.
173+-- | Lifted 'A.poll'.
151174poll
152175 :: AsyncE :> es
153176 => Async a
154177 -> Eff es (Maybe (Either SomeException a))
155178poll = unsafeEff_ . A.poll
156179
157−-- | Generalized version of 'A.cancel'.
180+-- | Lifted 'A.cancel'.
158181cancel :: AsyncE :> es => Async a -> Eff es ()
159182cancel = unsafeEff_ . A.cancel
160183
161−-- | Generalized version of 'A.cancelWith'.
184+-- | Lifted 'A.cancelWith'.
162185cancelWith :: (Exception e, AsyncE :> es) => Async a -> e -> Eff es ()
163186cancelWith a = unsafeEff_ . A.cancelWith a
164187
165−-- | Generalized version of 'A.uninterruptibleCancel'.
188+-- | Lifted 'A.uninterruptibleCancel'.
166189uninterruptibleCancel :: AsyncE :> es => Async a -> Eff es ()
167190uninterruptibleCancel = unsafeEff_ . A.uninterruptibleCancel
168191
169−-- | Generalized version of 'A.waitCatch'.
192+-- | Lifted 'A.waitCatch'.
170193waitCatch
171194 :: AsyncE :> es
172195 => Async a
173196 -> Eff es (Either SomeException a)
174197waitCatch = unsafeEff_ . A.waitCatch
175198
176−-- | Generalized version of 'A.waitAny'.
199+-- | Lifted 'A.waitAny'.
177200waitAny :: AsyncE :> es => [Async a] -> Eff es (Async a, a)
178201waitAny = unsafeEff_ . A.waitAny
179202
180−-- | Generalized version of 'A.waitAnyCatch'.
203+-- | Lifted 'A.waitAnyCatch'.
181204waitAnyCatch
182205 :: AsyncE :> es
183206 => [Async a]
184207 -> Eff es (Async a, Either SomeException a)
185208waitAnyCatch = unsafeEff_ . A.waitAnyCatch
186209
187−-- | Generalized version of 'A.waitAnyCancel'.
210+-- | Lifted 'A.waitAnyCancel'.
188211waitAnyCancel :: AsyncE :> es => [Async a] -> Eff es (Async a, a)
189212waitAnyCancel = unsafeEff_ . A.waitAnyCancel
190213
191−-- | Generalized version of 'A.waitAnyCatchCancel'.
214+-- | Lifted 'A.waitAnyCatchCancel'.
192215waitAnyCatchCancel
193216 :: AsyncE :> es
194217 => [Async a]
195218 -> Eff es (Async a, Either SomeException a)
196219waitAnyCatchCancel = unsafeEff_ . A.waitAnyCatchCancel
197220
198−-- | Generalized version of 'A.waitEither'.
221+-- | Lifted 'A.waitEither'.
199222waitEither
200223 :: AsyncE :> es
201224 => Async a
@@ -203,7 +226,7 @@ waitEither
203226 -> Eff es (Either a b)
204227waitEither a b = unsafeEff_ $ A.waitEither a b
205228
206−-- | Generalized version of 'A.waitEitherCatch'.
229+-- | Lifted 'A.waitEitherCatch'.
207230waitEitherCatch
208231 :: AsyncE :> es
209232 => Async a
@@ -211,7 +234,7 @@ waitEitherCatch
211234 -> Eff es (Either (Either SomeException a) (Either SomeException b))
212235waitEitherCatch a b = unsafeEff_ $ A.waitEitherCatch a b
213236
214−-- | Generalized version of 'A.waitEitherCancel'.
237+-- | Lifted 'A.waitEitherCancel'.
215238waitEitherCancel
216239 :: AsyncE :> es
217240 => Async a
@@ -219,7 +242,7 @@ waitEitherCancel
219242 -> Eff es (Either a b)
220243waitEitherCancel a b = unsafeEff_ $ A.waitEitherCancel a b
221244
222−-- | Generalized version of 'A.waitEitherCatchCancel'.
245+-- | Lifted 'A.waitEitherCatchCancel'.
223246waitEitherCatchCancel
224247 :: AsyncE :> es
225248 => Async a
@@ -227,103 +250,251 @@ waitEitherCatchCancel
227250 -> Eff es (Either (Either SomeException a) (Either SomeException b))
228251waitEitherCatchCancel a b = unsafeEff_ $ A.waitEitherCatch a b
229252
230−-- | Generalized version of 'A.waitEither_'.
253+-- | Lifted 'A.waitEither_'.
231254waitEither_ :: AsyncE :> es => Async a -> Async b -> Eff es ()
232255waitEither_ a b = unsafeEff_ $ A.waitEither_ a b
233256
234−-- | Generalized version of 'A.waitBoth'.
257+-- | Lifted 'A.waitBoth'.
235258waitBoth :: AsyncE :> es => Async a -> Async b -> Eff es (a, b)
236259waitBoth a b = unsafeEff_ $ A.waitBoth a b
237260
238−-- | Generalized version of 'A.link'.
261+-- | Lifted 'A.link'.
239262link :: AsyncE :> es => Async a -> Eff es ()
240263link = unsafeEff_ . A.link
241264
242−-- | Generalized version of 'A.linkOnly'.
265+-- | Lifted 'A.linkOnly'.
243266linkOnly :: AsyncE :> es => (SomeException -> Bool) -> Async a -> Eff es ()
244267linkOnly f = unsafeEff_ . A.linkOnly f
245268
246−-- | Generalized version of 'A.link2'.
269+-- | Lifted 'A.link2'.
247270link2 :: AsyncE :> es => Async a -> Async b -> Eff es ()
248271link2 a b = unsafeEff_ $ A.link2 a b
249272
250−-- | Generalized version of 'A.link2Only'.
273+-- | Lifted 'A.link2Only'.
251274link2Only :: AsyncE :> es => (SomeException -> Bool) -> Async a -> Async b -> Eff es ()
252275link2Only f a b = unsafeEff_ $ A.link2Only f a b
253276
254−-- | Generalized version of 'A.race'.
277+-- | Lifted 'A.race'.
255278race :: AsyncE :> es => Eff es a -> Eff es b -> Eff es (Either a b)
256279race ma mb = unsafeEff $ \es -> do
257− esA <- cloneEnv es
258− esB <- cloneEnv es
259− A.race (unEff ma esA) (unEff mb esB)
280+ A.race (unEff ma =<< cloneEnv es) (unEff mb =<< cloneEnv es)
260281
261−-- | Generalized version of 'A.race_'.
282+-- | Lifted 'A.race_'.
262283race_ :: AsyncE :> es => Eff es a -> Eff es b -> Eff es ()
263284race_ ma mb = unsafeEff $ \es -> do
264− esA <- cloneEnv es
265− esB <- cloneEnv es
266− A.race_ (unEff ma esA) (unEff mb esB)
285+ A.race_ (unEff ma =<< cloneEnv es) (unEff mb =<< cloneEnv es)
267286
268−-- | Generalized version of 'A.concurrently'.
287+-- | Lifted 'A.concurrently'.
269288concurrently :: AsyncE :> es => Eff es a -> Eff es b -> Eff es (a, b)
270289concurrently ma mb = unsafeEff $ \es -> do
271− esA <- cloneEnv es
272− esB <- cloneEnv es
273− A.concurrently (unEff ma esA) (unEff mb esB)
290+ A.concurrently (unEff ma =<< cloneEnv es) (unEff mb =<< cloneEnv es)
274291
275−-- | Generalized version of 'A.concurrently_'.
292+-- | Lifted 'A.concurrently_'.
276293concurrently_ :: AsyncE :> es => Eff es a -> Eff es b -> Eff es ()
277294concurrently_ ma mb = unsafeEff $ \es -> do
278− esA <- cloneEnv es
279− esB <- cloneEnv es
280− A.concurrently_ (unEff ma esA) (unEff mb esB)
295+ A.concurrently_ (unEff ma =<< cloneEnv es) (unEff mb =<< cloneEnv es)
296+
297+-- Below functions use variants from the @unliftio@ library as they minimize the
298+-- amount of spawned threads and are thus much more efficient than the ones from
299+-- the @async@ library.
281300
282−-- | Generalized version of 'A.mapConcurrently'.
301+-- | Lifted 'A.mapConcurrently'.
283302mapConcurrently
284303 :: (Traversable f, AsyncE :> es)
285304 => (a -> Eff es b)
286305 -> f a
287306 -> Eff es (f b)
288−mapConcurrently f = runConcurrently . traverse (Concurrently . f)
307+mapConcurrently f t = unsafeEff $ \es -> do
308+ U.mapConcurrently (\a -> unEff (f a) =<< cloneEnv es) t
289309
290−-- | Generalized version of 'A.mapConcurrently_'.
310+-- | Lifted 'A.mapConcurrently_'.
291311mapConcurrently_
292312 :: (Foldable f, AsyncE :> es)
293313 => (a -> Eff es b)
294314 -> f a
295315 -> Eff es ()
296−mapConcurrently_ f = runConcurrently . foldMap (Concurrently . void . f)
316+mapConcurrently_ f t = unsafeEff $ \es -> do
317+ U.mapConcurrently_ (\a -> unEff (f a) =<< cloneEnv es) t
297318
298−-- | Generalized version of 'A.forConcurrently'.
319+-- | Lifted 'A.forConcurrently'.
299320forConcurrently
300321 :: (Traversable f, AsyncE :> es)
301322 => f a
302323 -> (a -> Eff es b)
303324 -> Eff es (f b)
304−forConcurrently = flip mapConcurrently
325+forConcurrently t f = unsafeEff $ \es -> do
326+ U.forConcurrently t (\a -> unEff (f a) =<< cloneEnv es)
305327
306−-- | Generalized version of 'A.forConcurrently_'.
328+-- | Lifted 'A.forConcurrently_'.
307329forConcurrently_
308330 :: (Foldable f, AsyncE :> es)
309331 => f a
310332 -> (a -> Eff es b)
311333 -> Eff es ()
312−forConcurrently_ = flip mapConcurrently_
334+forConcurrently_ t f = unsafeEff $ \es -> do
335+ U.forConcurrently_ t (\a -> unEff (f a) =<< cloneEnv es)
313336
314−-- | Generalized version of 'A.replicateConcurrently'.
337+-- | Lifted 'A.replicateConcurrently'.
315338replicateConcurrently :: AsyncE :> es => Int -> Eff es a -> Eff es [a]
316−replicateConcurrently n =
317− runConcurrently . sequenceA . replicate n . Concurrently
339+replicateConcurrently n f = unsafeEff $ \es -> do
340+ U.replicateConcurrently n (unEff f =<< cloneEnv es)
318341
319−-- | Generalized version of 'A.replicateConcurrently_'.
342+-- | Lifted 'A.replicateConcurrently_'.
320343replicateConcurrently_ :: AsyncE :> es => Int -> Eff es a -> Eff es ()
321−replicateConcurrently_ n =
322− runConcurrently . fold . replicate n . Concurrently . void
344+replicateConcurrently_ n f = unsafeEff $ \es -> do
345+ U.replicateConcurrently_ n (unEff f =<< cloneEnv es)
346+
347+----------------------------------------
348+-- Pooled concurrency (unliftio)
349+
350+-- | Lifted 'U.pooledMapConcurrentlyN'.
351+pooledMapConcurrentlyN
352+ :: (AsyncE :> es, Traversable t)
353+ => Int
354+ -> (a -> Eff es b)
355+ -> t a
356+ -> Eff es (t b)
357+pooledMapConcurrentlyN threads f t = unsafeEff $ \es -> do
358+ U.pooledMapConcurrentlyN threads (\a -> unEff (f a) =<< cloneEnv es) t
359+
360+-- | Lifted 'U.pooledMapConcurrently'.
361+pooledMapConcurrently
362+ :: (AsyncE :> es, Traversable t)
363+ => (a -> Eff es b)
364+ -> t a
365+ -> Eff es (t b)
366+pooledMapConcurrently f t = unsafeEff $ \es -> do
367+ U.pooledMapConcurrently (\a -> unEff (f a) =<< cloneEnv es) t
368+
369+-- | Lifted 'U.pooledMapConcurrentlyN'.
370+pooledMapConcurrentlyN_
371+ :: (AsyncE :> es, Foldable f)
372+ => Int
373+ -> (a -> Eff es b)
374+ -> f a
375+ -> Eff es ()
376+pooledMapConcurrentlyN_ threads f t = unsafeEff $ \es -> do
377+ U.pooledMapConcurrentlyN_ threads (\a -> unEff (f a) =<< cloneEnv es) t
378+
379+-- | Lifted 'U.pooledMapConcurrently_'.
380+pooledMapConcurrently_
381+ :: (AsyncE :> es, Foldable f)
382+ => (a -> Eff es b)
383+ -> f a
384+ -> Eff es ()
385+pooledMapConcurrently_ f t = unsafeEff $ \es -> do
386+ U.pooledMapConcurrently_ (\a -> unEff (f a) =<< cloneEnv es) t
387+
388+-- | Lifted 'U.pooledForConcurrentlyN'.
389+pooledForConcurrentlyN
390+ :: (AsyncE :> es, Traversable t)
391+ => Int
392+ -> t a
393+ -> (a -> Eff es b)
394+ -> Eff es (t b)
395+pooledForConcurrentlyN threads t f = unsafeEff $ \es -> do
396+ U.pooledForConcurrentlyN threads t (\a -> unEff (f a) =<< cloneEnv es)
397+
398+-- | Lifted 'U.pooledForConcurrently'.
399+pooledForConcurrently
400+ :: (AsyncE :> es, Traversable t)
401+ => t a
402+ -> (a -> Eff es b)
403+ -> Eff es (t b)
404+pooledForConcurrently t f = unsafeEff $ \es -> do
405+ U.pooledForConcurrently t (\a -> unEff (f a) =<< cloneEnv es)
406+
407+-- | Lifted 'U.pooledForConcurrentlyN'.
408+pooledForConcurrentlyN_
409+ :: (AsyncE :> es, Foldable f)
410+ => Int
411+ -> f a
412+ -> (a -> Eff es b)
413+ -> Eff es ()
414+pooledForConcurrentlyN_ threads t f = unsafeEff $ \es -> do
415+ U.pooledForConcurrentlyN_ threads t (\a -> unEff (f a) =<< cloneEnv es)
416+
417+-- | Lifted 'U.pooledForConcurrently_'.
418+pooledForConcurrently_
419+ :: (AsyncE :> es, Foldable f)
420+ => f a
421+ -> (a -> Eff es b)
422+ -> Eff es ()
423+pooledForConcurrently_ t f = unsafeEff $ \es -> do
424+ U.pooledForConcurrently_ t (\a -> unEff (f a) =<< cloneEnv es)
425+
426+-- | Lifted 'U.pooledReplicateConcurrentlyN'.
427+pooledReplicateConcurrentlyN :: AsyncE :> es => Int -> Int -> Eff es a -> Eff es [a]
428+pooledReplicateConcurrentlyN threads n f = unsafeEff $ \es -> do
429+ U.pooledReplicateConcurrentlyN threads n (unEff f =<< cloneEnv es)
430+
431+-- | Lifted 'U.pooledReplicateConcurrently'.
432+pooledReplicateConcurrently :: AsyncE :> es => Int -> Eff es a -> Eff es [a]
433+pooledReplicateConcurrently n f = unsafeEff $ \es -> do
434+ U.pooledReplicateConcurrently n (unEff f =<< cloneEnv es)
435+
436+-- | Lifted 'U.pooledReplicateConcurrentlyN_'.
437+pooledReplicateConcurrentlyN_ :: AsyncE :> es => Int -> Int -> Eff es a -> Eff es ()
438+pooledReplicateConcurrentlyN_ threads n f = unsafeEff $ \es -> do
439+ U.pooledReplicateConcurrentlyN_ threads n (unEff f =<< cloneEnv es)
440+
441+-- | Lifted 'U.pooledReplicateConcurrently_'.
442+pooledReplicateConcurrently_ :: AsyncE :> es => Int -> Eff es a -> Eff es ()
443+pooledReplicateConcurrently_ n f = unsafeEff $ \es -> do
444+ U.pooledReplicateConcurrently_ n (unEff f =<< cloneEnv es)
445+
446+----------------------------------------
447+-- Conc
448+
449+-- | Lifted 'U.Conc'.
450+data Conc es a where
451+ Action :: Eff es a -> Conc es a
452+ Apply :: Conc es (v -> a) -> Conc es v -> Conc es a
453+ LiftA2 :: (x -> y -> a) -> Conc es x -> Conc es y -> Conc es a
454+ Pure :: a -> Conc es a
455+ Alt :: Conc es a -> Conc es a -> Conc es a
456+ Empty :: Conc es a
457+
458+deriving instance Functor (Conc es)
459+
460+instance AsyncE :> es => Applicative (Conc es) where
461+ pure = Pure
462+ (<*>) = Apply
463+ (<*) = LiftA2 (\x _ -> x)
464+ (*>) = LiftA2 (\_ y -> y)
465+ liftA2 = LiftA2
466+
467+instance AsyncE :> es => Alternative (Conc es) where
468+ empty = Empty
469+ (<|>) = Alt
470+
471+instance (AsyncE :> es, Semigroup a) => Semigroup (Conc es a) where
472+ (<>) = liftA2 (<>)
473+
474+instance (AsyncE :> es, Monoid a) => Monoid (Conc es a) where
475+ mempty = pure mempty
476+
477+-- | Lifted 'U.conc'.
478+conc :: AsyncE :> es => Eff es a -> Conc es a
479+conc = Action
480+
481+-- | Lifted 'U.runConc'.
482+runConc :: AsyncE :> es => Conc es a -> Eff es a
483+runConc m = unsafeEff $ \es -> U.runConc (unliftConc es m)
484+ where
485+ unliftConc :: Env es -> Conc es a -> U.Conc IO a
486+ unliftConc es = \case
487+ Action f -> I.Action $ unEff f =<< cloneEnv es
488+ Apply a b -> I.Apply (unliftConc es a) (unliftConc es b)
489+ LiftA2 f a b -> I.LiftA2 f (unliftConc es a) (unliftConc es b)
490+ Pure a -> I.Pure a
491+ Alt a b -> I.Alt (unliftConc es a) (unliftConc es b)
492+ Empty -> I.Empty
323493
324494----------------------------------------
495+-- Concurrently
325496
326−-- | Generalized version of 'A.Concurrently'.
497+-- | Lifted 'A.Concurrently'.
327498newtype Concurrently es a = Concurrently { runConcurrently :: Eff es a }
328499
329500instance Functor (Concurrently es) where