git.iko.soy/effectful
on b42e0dc help sign in
effectful/src/Effectful/Async.hs
373 lines 11 kB history raw
1{-# LANGUAGE UndecidableInstances #-}
2module Effectful.Async
3 ( -- * Async effect
4 AsyncE
5 , runAsyncE
7 -- * Asynchronous actions
8 , Async
10 -- * High-level API
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(..)
20 -- ** #high-level-utilities# High-level utilities
21 , race, race_
22 , concurrently, concurrently_
23 , mapConcurrently, forConcurrently
24 , mapConcurrently_, forConcurrently_
25 , replicateConcurrently, replicateConcurrently_
26 , Concurrently(..)
27 , A.compareAsyncs
29 -- ** Specialised operations
31 -- *** STM operations
32 , A.waitSTM, A.pollSTM, A.waitCatchSTM
34 -- *** Waiting for multiple 'Async's
35 , waitAny, waitAnyCatch, waitAnyCancel, waitAnyCatchCancel
36 , waitEither, waitEitherCatch, waitEitherCancel, waitEitherCatchCancel
37 , waitEither_
38 , waitBoth
40 -- *** Waiting for multiple 'Async's in STM
41 , A.waitAnySTM, A.waitAnyCatchSTM
42 , A.waitEitherSTM, A.waitEitherCatchSTM
43 , A.waitEitherSTM_
44 , A.waitBothSTM
46 -- * Low-level API
48 -- ** Spawning (low-level API)
49 , async, asyncBound, asyncOn, asyncWithUnmask, asyncOnWithUnmask
51 -- ** Linking
52 , link, linkOnly, link2, link2Only, A.ExceptionInLinkedThread(..)
53 ) where
55import Control.Applicative
56import Control.Concurrent (threadDelay)
57import Control.Concurrent.Async (Async)
58import Control.Exception (Exception, SomeException)
59import Control.Monad (forever, void)
60import Data.Foldable (fold)
61import qualified Control.Concurrent.Async as A
63import Effectful.Internal.Effect
64import Effectful.Internal.Env
65import Effectful.Internal.Monad
67-- | Run 'Eff' computations asynchronously using the @async@ library.
68--
69-- /Note:/ thread local state changes in 'Eff' computations run asynchronously
70-- as 'Async' will not affect the parent thread.
71data AsyncE :: Effect where
72 AsyncE :: AsyncE m r
74runAsyncE :: IOE :> es => Eff (AsyncE : es) a -> Eff es a
75runAsyncE = evalEffect (IdE AsyncE)
77-- | Generalized version of 'A.async'.
78async :: AsyncE :> es => Eff es a -> Eff es (Async a)
79async = liftAsync A.async
81-- | Generalized version of 'A.asyncBound'.
82asyncBound :: AsyncE :> es => Eff es a -> Eff es (Async a)
83asyncBound = liftAsync A.asyncBound
85-- | Generalized version of 'A.asyncOn'.
86asyncOn :: AsyncE :> es => Int -> Eff es a -> Eff es (Async a)
87asyncOn cpu = liftAsync (A.asyncOn cpu)
89-- | Generalized version of 'A.asyncWithUnmask'.
90asyncWithUnmask
91 :: AsyncE :> es
92 => ((forall b. Eff es b -> Eff es b) -> Eff es a)
93 -> Eff es (Async a)
94asyncWithUnmask = liftAsyncWithUnmask A.asyncWithUnmask
96-- | Generalized version of 'A.asyncOnWithUnmask'.
97asyncOnWithUnmask
98 :: AsyncE :> es
99 => Int
100 -> ((forall b. Eff es b -> Eff es b) -> Eff es a)
101 -> Eff es (Async a)
102asyncOnWithUnmask cpu = liftAsyncWithUnmask (A.asyncOnWithUnmask cpu)
104-- | Generalized version of 'A.withAsync'.
105withAsync
106 :: AsyncE :> es
107 => Eff es a
108 -> (Async a -> Eff es b)
109 -> Eff es b
110withAsync = liftWithAsync A.withAsync
112-- | Generalized version of 'A.withAsyncBound'.
113withAsyncBound
114 :: AsyncE :> es
115 => Eff es a
116 -> (Async a -> Eff es b)
117 -> Eff es b
118withAsyncBound = liftWithAsync A.withAsyncBound
120-- | Generalized version of 'A.withAsyncOn'.
121withAsyncOn
122 :: AsyncE :> es
123 => Int
124 -> Eff es a
125 -> (Async a -> Eff es b)
126 -> Eff es b
127withAsyncOn cpu = liftWithAsync (A.withAsyncOn cpu)
129-- | Generalized version of 'A.withAsyncWithUnmask'.
130withAsyncWithUnmask
131 :: AsyncE :> es
132 => ((forall c. Eff es c -> Eff es c) -> Eff es a)
133 -> (Async a -> Eff es b)
134 -> Eff es b
135withAsyncWithUnmask = liftWithAsyncWithUnmask A.withAsyncWithUnmask
137-- | Generalized version of 'A.withAsyncOnWithUnmask'.
138withAsyncOnWithUnmask
139 :: AsyncE :> es
140 => Int
141 -> ((forall c. Eff es c -> Eff es c) -> Eff es a)
142 -> (Async a -> Eff es b)
143 -> Eff es b
144withAsyncOnWithUnmask cpu = liftWithAsyncWithUnmask (A.withAsyncOnWithUnmask cpu)
146-- | Generalized version of 'A.wait'.
147wait :: AsyncE :> es => Async a -> Eff es a
148wait = unsafeEff_ . A.wait
150-- | Generalized version of 'A.poll'.
151poll
152 :: AsyncE :> es
153 => Async a
154 -> Eff es (Maybe (Either SomeException a))
155poll = unsafeEff_ . A.poll
157-- | Generalized version of 'A.cancel'.
158cancel :: AsyncE :> es => Async a -> Eff es ()
159cancel = unsafeEff_ . A.cancel
161-- | Generalized version of 'A.cancelWith'.
162cancelWith :: (Exception e, AsyncE :> es) => Async a -> e -> Eff es ()
163cancelWith a = unsafeEff_ . A.cancelWith a
165-- | Generalized version of 'A.uninterruptibleCancel'.
166uninterruptibleCancel :: AsyncE :> es => Async a -> Eff es ()
167uninterruptibleCancel = unsafeEff_ . A.uninterruptibleCancel
169-- | Generalized version of 'A.waitCatch'.
170waitCatch
171 :: AsyncE :> es
172 => Async a
173 -> Eff es (Either SomeException a)
174waitCatch = unsafeEff_ . A.waitCatch
176-- | Generalized version of 'A.waitAny'.
177waitAny :: AsyncE :> es => [Async a] -> Eff es (Async a, a)
178waitAny = unsafeEff_ . A.waitAny
180-- | Generalized version of 'A.waitAnyCatch'.
181waitAnyCatch
182 :: AsyncE :> es
183 => [Async a]
184 -> Eff es (Async a, Either SomeException a)
185waitAnyCatch = unsafeEff_ . A.waitAnyCatch
187-- | Generalized version of 'A.waitAnyCancel'.
188waitAnyCancel :: AsyncE :> es => [Async a] -> Eff es (Async a, a)
189waitAnyCancel = unsafeEff_ . A.waitAnyCancel
191-- | Generalized version of 'A.waitAnyCatchCancel'.
192waitAnyCatchCancel
193 :: AsyncE :> es
194 => [Async a]
195 -> Eff es (Async a, Either SomeException a)
196waitAnyCatchCancel = unsafeEff_ . A.waitAnyCatchCancel
198-- | Generalized version of 'A.waitEither'.
199waitEither
200 :: AsyncE :> es
201 => Async a
202 -> Async b
203 -> Eff es (Either a b)
204waitEither a b = unsafeEff_ $ A.waitEither a b
206-- | Generalized version of 'A.waitEitherCatch'.
207waitEitherCatch
208 :: AsyncE :> es
209 => Async a
210 -> Async b
211 -> Eff es (Either (Either SomeException a) (Either SomeException b))
212waitEitherCatch a b = unsafeEff_ $ A.waitEitherCatch a b
214-- | Generalized version of 'A.waitEitherCancel'.
215waitEitherCancel
216 :: AsyncE :> es
217 => Async a
218 -> Async b
219 -> Eff es (Either a b)
220waitEitherCancel a b = unsafeEff_ $ A.waitEitherCancel a b
222-- | Generalized version of 'A.waitEitherCatchCancel'.
223waitEitherCatchCancel
224 :: AsyncE :> es
225 => Async a
226 -> Async b
227 -> Eff es (Either (Either SomeException a) (Either SomeException b))
228waitEitherCatchCancel a b = unsafeEff_ $ A.waitEitherCatch a b
230-- | Generalized version of 'A.waitEither_'.
231waitEither_ :: AsyncE :> es => Async a -> Async b -> Eff es ()
232waitEither_ a b = unsafeEff_ $ A.waitEither_ a b
234-- | Generalized version of 'A.waitBoth'.
235waitBoth :: AsyncE :> es => Async a -> Async b -> Eff es (a, b)
236waitBoth a b = unsafeEff_ $ A.waitBoth a b
238-- | Generalized version of 'A.link'.
239link :: AsyncE :> es => Async a -> Eff es ()
240link = unsafeEff_ . A.link
242-- | Generalized version of 'A.linkOnly'.
243linkOnly :: AsyncE :> es => (SomeException -> Bool) -> Async a -> Eff es ()
244linkOnly f = unsafeEff_ . A.linkOnly f
246-- | Generalized version of 'A.link2'.
247link2 :: AsyncE :> es => Async a -> Async b -> Eff es ()
248link2 a b = unsafeEff_ $ A.link2 a b
250-- | Generalized version of 'A.link2Only'.
251link2Only :: AsyncE :> es => (SomeException -> Bool) -> Async a -> Async b -> Eff es ()
252link2Only f a b = unsafeEff_ $ A.link2Only f a b
254-- | Generalized version of 'A.race'.
255race :: AsyncE :> es => Eff es a -> Eff es b -> Eff es (Either a b)
256race ma mb = withAsync ma $ \a -> withAsync mb $ \b -> waitEither a b
258-- | Generalized version of 'A.race_'.
259race_ :: AsyncE :> es => Eff es a -> Eff es b -> Eff es ()
260race_ ma mb = withAsync ma $ \a -> withAsync mb $ \b -> waitEither_ a b
262-- | Generalized version of 'A.concurrently'.
263concurrently :: AsyncE :> es => Eff es a -> Eff es b -> Eff es (a, b)
264concurrently ma mb = withAsync ma $ \a -> withAsync mb $ \b -> waitBoth a b
266-- | Generalized version of 'A.concurrently_'.
267concurrently_ :: AsyncE :> es => Eff es a -> Eff es b -> Eff es ()
268concurrently_ ma mb = void $ concurrently ma mb
270-- | Generalized version of 'A.mapConcurrently'.
271mapConcurrently
272 :: (Traversable f, AsyncE :> es)
273 => (a -> Eff es b)
274 -> f a
275 -> Eff es (f b)
276mapConcurrently f = runConcurrently . traverse (Concurrently . f)
278-- | Generalized version of 'A.mapConcurrently_'.
279mapConcurrently_
280 :: (Foldable f, AsyncE :> es)
281 => (a -> Eff es b)
282 -> f a
283 -> Eff es ()
284mapConcurrently_ f = runConcurrently . foldMap (Concurrently . void . f)
286-- | Generalized version of 'A.forConcurrently'.
287forConcurrently
288 :: (Traversable f, AsyncE :> es)
289 => f a
290 -> (a -> Eff es b)
291 -> Eff es (f b)
292forConcurrently = flip mapConcurrently
294-- | Generalized version of 'A.forConcurrently_'.
295forConcurrently_
296 :: (Foldable f, AsyncE :> es)
297 => f a
298 -> (a -> Eff es b)
299 -> Eff es ()
300forConcurrently_ = flip mapConcurrently_
302-- | Generalized version of 'A.replicateConcurrently'.
303replicateConcurrently :: AsyncE :> es => Int -> Eff es a -> Eff es [a]
304replicateConcurrently n =
305 runConcurrently . sequenceA . replicate n . Concurrently
307-- | Generalized version of 'A.replicateConcurrently_'.
308replicateConcurrently_ :: AsyncE :> es => Int -> Eff es a -> Eff es ()
309replicateConcurrently_ n =
310 runConcurrently . fold . replicate n . Concurrently . void
312----------------------------------------
314-- | Generalized version of 'A.Concurrently'.
315newtype Concurrently es a = Concurrently { runConcurrently :: Eff es a }
317instance Functor (Concurrently es) where
318 fmap f (Concurrently a) = Concurrently (fmap f a)
320instance AsyncE :> es => Applicative (Concurrently es) where
321 pure = Concurrently . pure
322 Concurrently fs <*> Concurrently as =
323 Concurrently ((\(f, a) -> f a) <$> concurrently fs as)
325instance AsyncE :> es => Alternative (Concurrently es) where
326 empty = Concurrently . unsafeEff_ . forever $ threadDelay maxBound
327 Concurrently as <|> Concurrently bs =
328 Concurrently (either id id <$> race as bs)
330instance (AsyncE :> es, Semigroup a) => Semigroup (Concurrently es a) where
331 (<>) = liftA2 (<>)
333instance (AsyncE :> es, Monoid a) => Monoid (Concurrently es a) where
334 mempty = pure mempty
336----------------------------------------
337-- Helpers
339liftAsync
340 :: (IO a -> IO (Async a))
341 -> Eff es a
342 -> Eff es (Async a)
343liftAsync fork action = unsafeEff $ \es0 -> do
344 es <- cloneEnv es0
345 fork $ unEff action es
347liftAsyncWithUnmask
348 :: (((forall b. IO b -> IO b) -> IO a) -> IO (Async a))
349 -> ((forall b. Eff es b -> Eff es b) -> Eff es a)
350 -> Eff es (Async a)
351liftAsyncWithUnmask fork action = unsafeEff $ \es0 -> do
352 es <- cloneEnv es0
353 fork $ \unmask -> unEff (action $ \m -> unsafeEff $ unmask . unEff m) es
355liftWithAsync
356 :: (IO a -> (Async a -> IO b) -> IO b)
357 -> Eff es a
358 -> (Async a -> Eff es b)
359 -> Eff es b
360liftWithAsync withA action k = unsafeEff $ \es0 -> do
361 es <- cloneEnv es0
362 withA (unEff action es)
363 (\a -> unEff (k a) es0)
365liftWithAsyncWithUnmask
366 :: (((forall c. IO c -> IO c) -> IO a) -> (Async a -> IO b) -> IO b)
367 -> ((forall c. Eff es c -> Eff es c) -> Eff es a)
368 -> (Async a -> Eff es b)
369 -> Eff es b
370liftWithAsyncWithUnmask withA action k = unsafeEff $ \es0 -> do
371 es <- cloneEnv es0
372 withA (\unmask -> unEff (action $ \m -> unsafeEff $ unmask . unEff m) es)
373 (\a -> unEff (k a) es0)