git.iko.soy/effectful
on master help sign in
Add the Async effect
b5e29e1Andrzej Rybczak
effectful.cabal+3−2
compare with master
unifiedsplit
@@ -52,6 +52,7 @@ library
52 ghc-options: -O2
52 ghc-options: -O2
53
53
54 build-depends: base >= 4.12 && <5
54 build-depends: base >= 4.12 && <5
55+ , async
55 , containers
56 , containers
56 , exceptions
57 , exceptions
57 , ghc-prim
58 , ghc-prim
@@ -64,7 +65,8 @@ library
64
65
65 hs-source-dirs: src
66 hs-source-dirs: src
66
67
67− exposed-modules: Effectful.Class.Reader
68+ exposed-modules: Effectful.Async
69+ Effectful.Class.Reader
68 Effectful.Class.State
70 Effectful.Class.State
69 Effectful.Class.Writer
71 Effectful.Class.Writer
70 Effectful.Error
72 Effectful.Error
@@ -87,7 +89,6 @@ test-suite test
87 ghc-options: -threaded -rtsopts -with-rtsopts=-N4
89 ghc-options: -threaded -rtsopts -with-rtsopts=-N4
88
90
89 build-depends: base <5
91 build-depends: base <5
90− , async
91 , effectful
92 , effectful
92 , exceptions
93 , exceptions
93 , lifted-base
94 , lifted-base
src/Effectful/Async.hsnew file+366
@@ -0,0 +1,366 @@
1+{-# LANGUAGE UndecidableInstances #-}
2+module Effectful.Async
3+ ( -- * Async effect
4+ AsyncE
5+ , runAsyncE
6+
7+ -- * Asynchronous actions
8+ , A.Async
9+ -- ** Spawning
10+ , async, asyncBound, asyncOn
11+ , asyncWithUnmask, asyncOnWithUnmask
12+
13+ -- ** Spawning with automatic 'cancel'ation
14+ , withAsync, withAsyncBound, withAsyncOn
15+ , withAsyncWithUnmask, withAsyncOnWithUnmask
16+
17+ -- ** Quering 'Async's
18+ , wait, poll, waitCatch
19+ , cancel
20+ , uninterruptibleCancel
21+ , cancelWith
22+ , A.asyncThreadId
23+ , A.AsyncCancelled(..)
24+
25+ -- ** STM operations
26+ , A.waitSTM, A.pollSTM, A.waitCatchSTM
27+
28+ -- ** Waiting for multiple 'Async's
29+ , waitAny, waitAnyCatch, waitAnyCancel, waitAnyCatchCancel
30+ , waitEither, waitEitherCatch, waitEitherCancel, waitEitherCatchCancel
31+ , waitEither_
32+ , waitBoth
33+
34+ -- ** Waiting for multiple 'Async's in STM
35+ , A.waitAnySTM
36+ , A.waitAnyCatchSTM
37+ , A.waitEitherSTM
38+ , A.waitEitherCatchSTM
39+ , A.waitEitherSTM_
40+ , A.waitBothSTM
41+
42+ -- ** Linking
43+ , link, link2
44+ , A.ExceptionInLinkedThread(..)
45+
46+ -- * Convenient utilities
47+ , race, race_, concurrently, concurrently_
48+ , mapConcurrently, mapConcurrently_
49+ , forConcurrently, forConcurrently_
50+ , replicateConcurrently, replicateConcurrently_
51+ , Concurrently(..)
52+
53+ , A.compareAsyncs
54+ ) where
55+
56+import Control.Applicative
57+import Control.Concurrent (threadDelay)
58+import Control.Concurrent.Async (Async)
59+import Control.Exception (Exception, SomeException)
60+import Control.Monad (forever, void)
61+import Data.Foldable (fold)
62+import qualified Control.Concurrent.Async as A
63+
64+import Effectful.Internal.Effect
65+import Effectful.Internal.Env
66+import Effectful.Internal.Monad
67+
68+-- | Run 'Eff' computations asynchronously using the @async@ library.
69+--
70+-- /Note:/ Thread-local state changes in 'Eff' computations lifted into an
71+-- 'Async' will not affect the thread that spawned them.
72+data AsyncE :: Effect where
73+ AsyncE :: AsyncE m r
74+
75+runAsyncE :: IOE :> es => Eff (AsyncE : es) a -> Eff es a
76+runAsyncE = evalEffect (IdE AsyncE)
77+
78+-- | Generalized version of 'A.async'.
79+async :: AsyncE :> es => Eff es a -> Eff es (Async a)
80+async = liftAsync A.async
81+
82+-- | Generalized version of 'A.asyncBound'.
83+asyncBound :: AsyncE :> es => Eff es a -> Eff es (Async a)
84+asyncBound = liftAsync A.asyncBound
85+
86+-- | Generalized version of 'A.asyncOn'.
87+asyncOn :: AsyncE :> es => Int -> Eff es a -> Eff es (Async a)
88+asyncOn cpu = liftAsync (A.asyncOn cpu)
89+
90+-- | Generalized version of 'A.asyncWithUnmask'.
91+asyncWithUnmask
92+ :: AsyncE :> es
93+ => ((forall b. Eff es b -> Eff es b) -> Eff es a)
94+ -> Eff es (Async a)
95+asyncWithUnmask = liftAsyncWithUnmask A.asyncWithUnmask
96+
97+-- | Generalized version of 'A.asyncOnWithUnmask'.
98+asyncOnWithUnmask
99+ :: AsyncE :> es
100+ => Int
101+ -> ((forall b. Eff es b -> Eff es b) -> Eff es a)
102+ -> Eff es (Async a)
103+asyncOnWithUnmask cpu = liftAsyncWithUnmask (A.asyncOnWithUnmask cpu)
104+
105+-- | Generalized version of 'A.withAsync'.
106+withAsync
107+ :: AsyncE :> es
108+ => Eff es a
109+ -> (Async a -> Eff es b)
110+ -> Eff es b
111+withAsync = liftWithAsync A.withAsync
112+
113+-- | Generalized version of 'A.withAsyncBound'.
114+withAsyncBound
115+ :: AsyncE :> es
116+ => Eff es a
117+ -> (Async a -> Eff es b)
118+ -> Eff es b
119+withAsyncBound = liftWithAsync A.withAsyncBound
120+
121+-- | Generalized version of 'A.withAsyncOn'.
122+withAsyncOn
123+ :: AsyncE :> es
124+ => Int
125+ -> Eff es a
126+ -> (Async a -> Eff es b)
127+ -> Eff es b
128+withAsyncOn cpu = liftWithAsync (A.withAsyncOn cpu)
129+
130+-- | Generalized version of 'A.withAsyncWithUnmask'.
131+withAsyncWithUnmask
132+ :: AsyncE :> es
133+ => ((forall c. Eff es c -> Eff es c) -> Eff es a)
134+ -> (Async a -> Eff es b)
135+ -> Eff es b
136+withAsyncWithUnmask = liftWithAsyncWithUnmask A.withAsyncWithUnmask
137+
138+-- | Generalized version of 'A.withAsyncOnWithUnmask'.
139+withAsyncOnWithUnmask
140+ :: AsyncE :> es
141+ => Int
142+ -> ((forall c. Eff es c -> Eff es c) -> Eff es a)
143+ -> (Async a -> Eff es b)
144+ -> Eff es b
145+withAsyncOnWithUnmask cpu = liftWithAsyncWithUnmask (A.withAsyncOnWithUnmask cpu)
146+
147+-- | Generalized version of 'A.wait'.
148+wait :: AsyncE :> es => Async a -> Eff es a
149+wait = unsafeEff_ . A.wait
150+
151+-- | Generalized version of 'A.poll'.
152+poll
153+ :: AsyncE :> es
154+ => Async a
155+ -> Eff es (Maybe (Either SomeException a))
156+poll = unsafeEff_ . A.poll
157+
158+-- | Generalized version of 'A.cancel'.
159+cancel :: AsyncE :> es => Async a -> Eff es ()
160+cancel = unsafeEff_ . A.cancel
161+
162+-- | Generalized version of 'A.cancelWith'.
163+cancelWith :: (Exception e, AsyncE :> es) => Async a -> e -> Eff es ()
164+cancelWith a = unsafeEff_ . A.cancelWith a
165+
166+-- | Generalized version of 'A.uninterruptibleCancel'.
167+uninterruptibleCancel :: AsyncE :> es => Async a -> Eff es ()
168+uninterruptibleCancel = unsafeEff_ . A.uninterruptibleCancel
169+
170+-- | Generalized version of 'A.waitCatch'.
171+waitCatch
172+ :: AsyncE :> es
173+ => Async a
174+ -> Eff es (Either SomeException a)
175+waitCatch = unsafeEff_ . A.waitCatch
176+
177+-- | Generalized version of 'A.waitAny'.
178+waitAny :: AsyncE :> es => [Async a] -> Eff es (Async a, a)
179+waitAny = unsafeEff_ . A.waitAny
180+
181+-- | Generalized version of 'A.waitAnyCatch'.
182+waitAnyCatch
183+ :: AsyncE :> es
184+ => [Async a]
185+ -> Eff es (Async a, Either SomeException a)
186+waitAnyCatch = unsafeEff_ . A.waitAnyCatch
187+
188+-- | Generalized version of 'A.waitAnyCancel'.
189+waitAnyCancel :: AsyncE :> es => [Async a] -> Eff es (Async a, a)
190+waitAnyCancel = unsafeEff_ . A.waitAnyCancel
191+
192+-- | Generalized version of 'A.waitAnyCatchCancel'.
193+waitAnyCatchCancel
194+ :: AsyncE :> es
195+ => [Async a]
196+ -> Eff es (Async a, Either SomeException a)
197+waitAnyCatchCancel = unsafeEff_ . A.waitAnyCatchCancel
198+
199+-- | Generalized version of 'A.waitEither'.
200+waitEither
201+ :: AsyncE :> es
202+ => Async a
203+ -> Async b
204+ -> Eff es (Either a b)
205+waitEither a b = unsafeEff_ $ A.waitEither a b
206+
207+-- | Generalized version of 'A.waitEitherCatch'.
208+waitEitherCatch
209+ :: AsyncE :> es
210+ => Async a
211+ -> Async b
212+ -> Eff es (Either (Either SomeException a) (Either SomeException b))
213+waitEitherCatch a b = unsafeEff_ $ A.waitEitherCatch a b
214+
215+-- | Generalized version of 'A.waitEitherCancel'.
216+waitEitherCancel
217+ :: AsyncE :> es
218+ => Async a
219+ -> Async b
220+ -> Eff es (Either a b)
221+waitEitherCancel a b = unsafeEff_ $ A.waitEitherCancel a b
222+
223+-- | Generalized version of 'A.waitEitherCatchCancel'.
224+waitEitherCatchCancel
225+ :: AsyncE :> es
226+ => Async a
227+ -> Async b
228+ -> Eff es (Either (Either SomeException a) (Either SomeException b))
229+waitEitherCatchCancel a b = unsafeEff_ $ A.waitEitherCatch a b
230+
231+-- | Generalized version of 'A.waitEither_'.
232+waitEither_ :: AsyncE :> es => Async a -> Async b -> Eff es ()
233+waitEither_ a b = unsafeEff_ $ A.waitEither_ a b
234+
235+-- | Generalized version of 'A.waitBoth'.
236+waitBoth :: AsyncE :> es => Async a -> Async b -> Eff es (a, b)
237+waitBoth a b = unsafeEff_ $ A.waitBoth a b
238+
239+-- | Generalized version of 'A.link'.
240+link :: AsyncE :> es => Async a -> Eff es ()
241+link = unsafeEff_ . A.link
242+
243+-- | Generalized version of 'A.link2'.
244+link2 :: AsyncE :> es => Async a -> Async b -> Eff es ()
245+link2 a b = unsafeEff_ $ A.link2 a b
246+
247+-- | Generalized version of 'A.race'.
248+race :: AsyncE :> es => Eff es a -> Eff es b -> Eff es (Either a b)
249+race ma mb = withAsync ma $ \a -> withAsync mb $ \b -> waitEither a b
250+
251+-- | Generalized version of 'A.race_'.
252+race_ :: AsyncE :> es => Eff es a -> Eff es b -> Eff es ()
253+race_ ma mb = withAsync ma $ \a -> withAsync mb $ \b -> waitEither_ a b
254+
255+-- | Generalized version of 'A.concurrently'.
256+concurrently :: AsyncE :> es => Eff es a -> Eff es b -> Eff es (a, b)
257+concurrently ma mb = withAsync ma $ \a -> withAsync mb $ \b -> waitBoth a b
258+
259+-- | Generalized version of 'A.concurrently_'.
260+concurrently_ :: AsyncE :> es => Eff es a -> Eff es b -> Eff es ()
261+concurrently_ ma mb = void $ concurrently ma mb
262+
263+-- | Generalized version of 'A.mapConcurrently'.
264+mapConcurrently
265+ :: (Traversable f, AsyncE :> es)
266+ => (a -> Eff es b)
267+ -> f a
268+ -> Eff es (f b)
269+mapConcurrently f = runConcurrently . traverse (Concurrently . f)
270+
271+-- | Generalized version of 'A.mapConcurrently_'.
272+mapConcurrently_
273+ :: (Foldable f, AsyncE :> es)
274+ => (a -> Eff es b)
275+ -> f a
276+ -> Eff es ()
277+mapConcurrently_ f = runConcurrently . foldMap (Concurrently . void . f)
278+
279+-- | Generalized version of 'A.forConcurrently'.
280+forConcurrently
281+ :: (Traversable f, AsyncE :> es)
282+ => f a
283+ -> (a -> Eff es b)
284+ -> Eff es (f b)
285+forConcurrently = flip mapConcurrently
286+
287+-- | Generalized version of 'A.forConcurrently_'.
288+forConcurrently_
289+ :: (Foldable f, AsyncE :> es)
290+ => f a
291+ -> (a -> Eff es b)
292+ -> Eff es ()
293+forConcurrently_ = flip mapConcurrently_
294+
295+-- | Generalized version of 'A.replicateConcurrently'.
296+replicateConcurrently :: AsyncE :> es => Int -> Eff es a -> Eff es [a]
297+replicateConcurrently n =
298+ runConcurrently . sequenceA . replicate n . Concurrently
299+
300+-- | Generalized version of 'A.replicateConcurrently_'.
301+replicateConcurrently_ :: AsyncE :> es => Int -> Eff es a -> Eff es ()
302+replicateConcurrently_ n =
303+ runConcurrently . fold . replicate n . Concurrently . void
304+
305+----------------------------------------
306+
307+-- | Generalized version of 'A.Concurrently'.
308+newtype Concurrently es a = Concurrently { runConcurrently :: Eff es a }
309+
310+instance Functor (Concurrently es) where
311+ fmap f (Concurrently a) = Concurrently (fmap f a)
312+
313+instance AsyncE :> es => Applicative (Concurrently es) where
314+ pure = Concurrently . pure
315+ Concurrently fs <*> Concurrently as =
316+ Concurrently ((\(f, a) -> f a) <$> concurrently fs as)
317+
318+instance AsyncE :> es => Alternative (Concurrently es) where
319+ empty = Concurrently . unsafeEff_ . forever $ threadDelay maxBound
320+ Concurrently as <|> Concurrently bs =
321+ Concurrently (either id id <$> race as bs)
322+
323+instance (AsyncE :> es, Semigroup a) => Semigroup (Concurrently es a) where
324+ (<>) = liftA2 (<>)
325+
326+instance (AsyncE :> es, Monoid a) => Monoid (Concurrently es a) where
327+ mempty = pure mempty
328+
329+----------------------------------------
330+-- Helpers
331+
332+liftAsync
333+ :: (IO a -> IO (Async a))
334+ -> Eff es a
335+ -> Eff es (Async a)
336+liftAsync fork action = unsafeEff $ \es0 -> do
337+ es <- cloneEnv es0
338+ fork $ unEff action es
339+
340+liftAsyncWithUnmask
341+ :: (((forall b. IO b -> IO b) -> IO a) -> IO (Async a))
342+ -> ((forall b. Eff es b -> Eff es b) -> Eff es a)
343+ -> Eff es (Async a)
344+liftAsyncWithUnmask fork action = unsafeEff $ \es0 -> do
345+ es <- cloneEnv es0
346+ fork $ \unmask -> unEff (action $ \m -> unsafeEff $ unmask . unEff m) es
347+
348+liftWithAsync
349+ :: (IO a -> (Async a -> IO b) -> IO b)
350+ -> Eff es a
351+ -> (Async a -> Eff es b)
352+ -> Eff es b
353+liftWithAsync withA action k = unsafeEff $ \es0 -> do
354+ es <- cloneEnv es0
355+ withA (unEff action es)
356+ (\a -> unEff (k a) es0)
357+
358+liftWithAsyncWithUnmask
359+ :: (((forall c. IO c -> IO c) -> IO a) -> (Async a -> IO b) -> IO b)
360+ -> ((forall c. Eff es c -> Eff es c) -> Eff es a)
361+ -> (Async a -> Eff es b)
362+ -> Eff es b
363+liftWithAsyncWithUnmask withA action k = unsafeEff $ \es0 -> do
364+ es <- cloneEnv es0
365+ withA (\unmask -> unEff (action $ \m -> unsafeEff $ unmask . unEff m) es)
366+ (\a -> unEff (k a) es0)
tests/Main.hs+10−8
@@ -1,6 +1,5 @@
1module Main (main) where
1module Main (main) where
2
2
3−import Control.Concurrent.Async
4import Control.Monad.IO.Class
3import Control.Monad.IO.Class
5import Test.Tasty
4import Test.Tasty
6import Test.Tasty.HUnit
5import Test.Tasty.HUnit
@@ -8,6 +7,7 @@ import qualified Control.Monad.Catch as E
8import qualified Control.Exception.Lifted as LE
7import qualified Control.Exception.Lifted as LE
9import qualified UnliftIO.Exception as UE
8import qualified UnliftIO.Exception as UE
10
9
10+import Effectful.Async
11import Effectful.Interpreter
11import Effectful.Interpreter
12import Effectful.State.Dynamic
12import Effectful.State.Dynamic
13import Effectful.Monad
13import Effectful.Monad
@@ -90,8 +90,8 @@ test_exceptions = runIOE $ do
90 modify @Int (+2)
90 modify @Int (+2)
91
91
92test_concurrentState :: Assertion
92test_concurrentState :: Assertion
93−test_concurrentState =
93+test_concurrentState = runIOE . evalState x $ do
94− replicateConcurrently_ 2 . runIOE . runState x $ do
94+ runAsyncE . replicateConcurrently_ 2 $ do
95 r <- goDownward 0
95 r <- goDownward 0
96 assertEqual_ "x = n" x r
96 assertEqual_ "x = n" x r
97 where
97 where
@@ -99,11 +99,13 @@ test_concurrentState =
99 x = 1000000
99 x = 1000000
100
100
101 goDownward :: State Int :> es => Int -> Eff es Int
101 goDownward :: State Int :> es => Int -> Eff es Int
102− goDownward acc = get @Int >>= \case
102+ goDownward acc = do
103− 0 -> pure acc
103+ end <- state @Int $ \case
104− n -> do
104+ 0 -> (True, 0)
105− put @Int $ n - 1
105+ n -> (False, n - 1)
106+ if end
107+ then pure acc
106− goDownward $ acc + 1
108+ else goDownward $ acc + 1
107
109
108test_localEffects :: Assertion
110test_localEffects :: Assertion
109test_localEffects = runIOE $ do
111test_localEffects = runIOE $ do