9f74175 · effectful · git.iko.soy

git.iko.soy/effectful

on master help sign in

Add more functions to Effectful.Async

9f74175Andrzej Rybczak5 years ago

bench/Concurrency.hs+4−4

compare with master

unifiedsplit

@@ -23,13 +23,13 @@ concurrencyBenchmark = bgroup "concurrency"

23 [ bench "async (IO)" $ nfAppIO asyncBenchIO (pure 1)

24 , bgroup "shallow"

25 [ bench "async (Eff)" $ nfAppIO (runShallow . A.runAsyncE . asyncBench) (pure 1)

26− , bench "Fork (unlift)" $ nfAppIO (runShallow . runFork1 . forkBench) (pure 1)

26+ , bench "Fork (1)" $ nfAppIO (runShallow . runFork1 . forkBench) (pure 1)

27− , bench "Fork (lift/unlift)" $ nfAppIO (runShallow . runFork2 . forkBench) (pure 1)

27+ , bench "Fork (2)" $ nfAppIO (runShallow . runFork2 . forkBench) (pure 1)

28 ]

29 , bgroup "deep"

30 [ bench "async (Eff)" $ nfAppIO (runDeep . A.runAsyncE . asyncBench) (pure 1)

31− , bench "Fork (unlift)" $ nfAppIO (runDeep . runFork1 . forkBench) (pure 1)

31+ , bench "Fork (1)" $ nfAppIO (runDeep . runFork1 . forkBench) (pure 1)

32− , bench "Fork (lift/unlift)" $ nfAppIO (runDeep . runFork2 . forkBench) (pure 1)

32+ , bench "Fork (2)" $ nfAppIO (runDeep . runFork2 . forkBench) (pure 1)

33 ]

34 ]

bench/Unlift.hs+30−10

@@ -15,6 +15,7 @@ import UnliftIO

15import Effectful

16import Effectful.State.Dynamic

17import Utils

18+import qualified Effectful.Async as A

19unliftBenchmark :: Benchmark

20unliftBenchmark :: Benchmark

20unliftBenchmark = bgroup "unlift"

21unliftBenchmark = bgroup "unlift"

@@ -26,18 +27,18 @@ unliftBenchmark = bgroup "unlift"

26 , bgroup "shallow"

27 , bgroup "shallow"

27 [ bgroup "dummy"

28 [ bgroup "dummy"

28 [ bench "seq" $ nfAppIO (runShallow . withUnliftStrategy SeqUnlift) benchDummy

29 [ bench "seq" $ nfAppIO (runShallow . withUnliftStrategy SeqUnlift) benchDummy

29− , shallowConc "conc" 1 benchDummy

30+ , shallowConc "conc" 1 benchDummyE benchDummy

30 ]

31 ]

31− , shallowConc "async" 1 benchAsync

32+ , shallowConc "async" 1 benchAsyncE benchAsync

32− , shallowConc "concurrently" 2 benchConcurrently

33+ , shallowConc "concurrently" 2 benchConcurrentlyE benchConcurrently

33 ]

34 ]

34 , bgroup "deep"

35 , bgroup "deep"

35 [ bgroup "dummy"

36 [ bgroup "dummy"

36 [ bench "seq" $ nfAppIO (runDeep . withUnliftStrategy SeqUnlift) benchDummy

37 [ bench "seq" $ nfAppIO (runDeep . withUnliftStrategy SeqUnlift) benchDummy

37− , deepConc "conc" 1 benchDummy

38+ , deepConc "conc" 1 benchDummyE benchDummy

38 ]

39 ]

39− , deepConc "async" 1 benchAsync

40+ , deepConc "async" 1 benchAsyncE benchAsync

40− , deepConc "concurrently" 2 benchConcurrently

41+ , deepConc "concurrently" 2 benchConcurrentlyE benchConcurrently

41 ]

42 ]

43 ]

@@ -46,10 +47,12 @@ unliftBenchmark = bgroup "unlift"

46shallowConc

47shallowConc

47 :: String

48 :: String

48 -> Int

49 -> Int

50+ -> Eff '[A.AsyncE, IOE] ()

49 -> Eff '[IOE] ()

51 -> Eff '[IOE] ()

50 -> Benchmark

52 -> Benchmark

51−shallowConc name n f = bgroup name

53+shallowConc name n r f = bgroup name

54+ [ bench "reference" $ nfAppIO (runShallow . A.runAsyncE) r

52− [ bench "ephemeral/limited" $ nfAppIO

55+ , bench "ephemeral/limited" $ nfAppIO

53 (runShallow . withUnliftStrategy (ConcUnlift Ephemeral $ Limited n)) f

56 (runShallow . withUnliftStrategy (ConcUnlift Ephemeral $ Limited n)) f

54 , bench "ephemeral/unlimited" $ nfAppIO

57 , bench "ephemeral/unlimited" $ nfAppIO

55 (runShallow . withUnliftStrategy (ConcUnlift Ephemeral Unlimited)) f

58 (runShallow . withUnliftStrategy (ConcUnlift Ephemeral Unlimited)) f

@@ -62,12 +65,17 @@ shallowConc name n f = bgroup name

62deepConc

65deepConc

63 :: String

66 :: String

64 -> Int

67 -> Int

68+ -> Eff '[ A.AsyncE

69+ , State (), State (), State (), State (), State ()

70+ , State (), State (), State (), State (), State ()

71+ , IOE] ()

65 -> Eff '[ State (), State (), State (), State (), State ()

72 -> Eff '[ State (), State (), State (), State (), State ()

66 , State (), State (), State (), State (), State ()

73 , State (), State (), State (), State (), State ()

67 , IOE] ()

74 , IOE] ()

68 -> Benchmark

75 -> Benchmark

69−deepConc name n f = bgroup name

76+deepConc name n r f = bgroup name

77+ [ bench "reference" $ nfAppIO (runDeep . A.runAsyncE) r

70− [ bench "ephemeral/limited" $ nfAppIO

78+ , bench "ephemeral/limited" $ nfAppIO

71 (runDeep . withUnliftStrategy (ConcUnlift Ephemeral $ Limited n)) f

79 (runDeep . withUnliftStrategy (ConcUnlift Ephemeral $ Limited n)) f

72 , bench "ephemeral/unlimited" $ nfAppIO

80 , bench "ephemeral/unlimited" $ nfAppIO

73 (runDeep . withUnliftStrategy (ConcUnlift Ephemeral Unlimited)) f

81 (runDeep . withUnliftStrategy (ConcUnlift Ephemeral Unlimited)) f

@@ -87,3 +95,15 @@ benchAsync = withAsync (pure ()) wait

88benchConcurrently :: MonadUnliftIO m => m ()

96benchConcurrently :: MonadUnliftIO m => m ()

89benchConcurrently = concurrently_ (pure ()) (pure ())

97benchConcurrently = concurrently_ (pure ()) (pure ())

98+

99+benchDummyE :: Eff es ()

100+benchDummyE = pure ()

101+{-# NOINLINE benchDummyE #-}

102+

103+benchAsyncE :: A.AsyncE :> es => Eff es ()

104+benchAsyncE = A.withAsync (pure ()) A.wait

105+{-# NOINLINE benchAsyncE #-}

106+

107+benchConcurrentlyE :: A.AsyncE :> es => Eff es ()

108+benchConcurrentlyE = A.concurrently_ (pure ()) (pure ())

109+{-# NOINLINE benchConcurrentlyE #-}

effectful.cabal+3

@@ -34,6 +34,7 @@ common language

34 default-extensions: BangPatterns

35 ConstraintKinds

36 DataKinds

37+ DeriveFunctor

37 FlexibleContexts

38 FlexibleContexts

38 FlexibleInstances

39 FlexibleInstances

39 GADTs

40 GADTs

@@ -45,6 +46,7 @@ common language

45 RecordWildCards

46 RecordWildCards

46 RoleAnnotations

47 RoleAnnotations

47 ScopedTypeVariables

48 ScopedTypeVariables

49+ StandaloneDeriving

48 StrictData

50 StrictData

49 TupleSections

51 TupleSections

50 TypeApplications

52 TypeApplications

@@ -77,6 +79,7 @@ library

77 , stm

79 , stm

78 , transformers-base

80 , transformers-base

79 , unliftio-core

81 , unliftio-core

82+ , unliftio

81 hs-source-dirs: src

84 hs-source-dirs: src

src/Effectful/Async.hs+248−77

@@ -4,51 +4,70 @@ module Effectful.Async

4 AsyncE

5 , runAsyncE

7− -- * Asynchronous actions

7+ -- * Asynchronous actions

8 , Async

10− -- * High-level API

10+ -- * High-level API

12− -- ** Spawning with automatic 'cancel'ation

12+ -- ** Spawning with automatic 'cancel'ation

13 , withAsync, withAsyncBound, withAsyncOn, withAsyncWithUnmask

14 , withAsyncOnWithUnmask

16− -- ** Querying 'Async's

16+ -- ** Querying 'Async's

17 , wait, poll, waitCatch, A.asyncThreadId

18 , cancel, uninterruptibleCancel, cancelWith, A.AsyncCancelled(..)

19+ , A.compareAsyncs

20− -- ** #high-level-utilities# High-level utilities

21+ -- ** High-level utilities

21 , race, race_

22 , race, race_

22 , concurrently, concurrently_

23 , concurrently, concurrently_

23 , mapConcurrently, forConcurrently

24 , mapConcurrently, forConcurrently

24 , mapConcurrently_, forConcurrently_

25 , mapConcurrently_, forConcurrently_

25 , replicateConcurrently, replicateConcurrently_

26 , replicateConcurrently, replicateConcurrently_

26− , Concurrently(..)

27− , A.compareAsyncs

29− -- ** Specialised operations

28+ -- *** Concurrently

29+ , Concurrently(..)

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+

31− -- *** STM operations

50+ -- *** STM operations

32 , A.waitSTM, A.pollSTM, A.waitCatchSTM

51 , A.waitSTM, A.pollSTM, A.waitCatchSTM

34− -- *** Waiting for multiple 'Async's

53+ -- *** Waiting for multiple 'Async's

35 , waitAny, waitAnyCatch, waitAnyCancel, waitAnyCatchCancel

54 , waitAny, waitAnyCatch, waitAnyCancel, waitAnyCatchCancel

36 , waitEither, waitEitherCatch, waitEitherCancel, waitEitherCatchCancel

55 , waitEither, waitEitherCatch, waitEitherCancel, waitEitherCatchCancel

37 , waitEither_

56 , waitEither_

38 , waitBoth

57 , waitBoth

40− -- *** Waiting for multiple 'Async's in STM

59+ -- *** Waiting for multiple 'Async's in STM

41 , A.waitAnySTM, A.waitAnyCatchSTM

60 , A.waitAnySTM, A.waitAnyCatchSTM

42 , A.waitEitherSTM, A.waitEitherCatchSTM

61 , A.waitEitherSTM, A.waitEitherCatchSTM

43 , A.waitEitherSTM_

62 , A.waitEitherSTM_

44 , A.waitBothSTM

63 , A.waitBothSTM

46− -- * Low-level API

65+ -- * Low-level API

48− -- ** Spawning (low-level API)

67+ -- ** Spawning (low-level API)

49 , async, asyncBound, asyncOn, asyncWithUnmask, asyncOnWithUnmask

68 , async, asyncBound, asyncOn, asyncWithUnmask, asyncOnWithUnmask

51− -- ** Linking

70+ -- ** Linking

52 , link, linkOnly, link2, link2Only, A.ExceptionInLinkedThread(..)

71 , link, linkOnly, link2, link2Only, A.ExceptionInLinkedThread(..)

53 ) where

72 ) where

@@ -56,9 +75,10 @@ import Control.Applicative

56import Control.Concurrent (threadDelay)

75import Control.Concurrent (threadDelay)

57import Control.Concurrent.Async (Async)

76import Control.Concurrent.Async (Async)

58import Control.Exception (Exception, SomeException)

77import Control.Exception (Exception, SomeException)

59−import Control.Monad (forever, void)

78+import Control.Monad (forever)

60−import Data.Foldable (fold)

61import qualified Control.Concurrent.Async as A

79import qualified Control.Concurrent.Async as A

80+import qualified UnliftIO.Async as U

81+import qualified UnliftIO.Internals.Async as I

63import Effectful.Internal.Effect

83import Effectful.Internal.Effect

64import Effectful.Internal.Env

84import Effectful.Internal.Env

@@ -74,26 +94,29 @@ data AsyncE :: Effect where

74runAsyncE :: IOE :> es => Eff (AsyncE : es) a -> Eff es a

94runAsyncE :: IOE :> es => Eff (AsyncE : es) a -> Eff es a

75runAsyncE = evalEffect (IdE AsyncE)

95runAsyncE = evalEffect (IdE AsyncE)

77−-- | Generalized version of 'A.async'.

97+----------------------------------------

98+-- Operations

99+

100+-- | Lifted 'A.async'.

78async :: AsyncE :> es => Eff es a -> Eff es (Async a)

101async :: AsyncE :> es => Eff es a -> Eff es (Async a)

79async = liftAsync A.async

102async = liftAsync A.async

103

81−-- | Generalized version of 'A.asyncBound'.

104+-- | Lifted 'A.asyncBound'.

82asyncBound :: AsyncE :> es => Eff es a -> Eff es (Async a)

105asyncBound :: AsyncE :> es => Eff es a -> Eff es (Async a)

83asyncBound = liftAsync A.asyncBound

106asyncBound = liftAsync A.asyncBound

107

85−-- | Generalized version of 'A.asyncOn'.

108+-- | Lifted 'A.asyncOn'.

86asyncOn :: AsyncE :> es => Int -> Eff es a -> Eff es (Async a)

109asyncOn :: AsyncE :> es => Int -> Eff es a -> Eff es (Async a)

87asyncOn cpu = liftAsync (A.asyncOn cpu)

110asyncOn cpu = liftAsync (A.asyncOn cpu)

111

89−-- | Generalized version of 'A.asyncWithUnmask'.

112+-- | Lifted 'A.asyncWithUnmask'.

90asyncWithUnmask

113asyncWithUnmask

91 :: AsyncE :> es

114 :: AsyncE :> es

92 => ((forall b. Eff es b -> Eff es b) -> Eff es a)

115 => ((forall b. Eff es b -> Eff es b) -> Eff es a)

93 -> Eff es (Async a)

116 -> Eff es (Async a)

94asyncWithUnmask = liftAsyncWithUnmask A.asyncWithUnmask

117asyncWithUnmask = liftAsyncWithUnmask A.asyncWithUnmask

118

96−-- | Generalized version of 'A.asyncOnWithUnmask'.

119+-- | Lifted 'A.asyncOnWithUnmask'.

97asyncOnWithUnmask

120asyncOnWithUnmask

98 :: AsyncE :> es

121 :: AsyncE :> es

99 => Int

122 => Int

@@ -101,7 +124,7 @@ asyncOnWithUnmask

101 -> Eff es (Async a)

124 -> Eff es (Async a)

102asyncOnWithUnmask cpu = liftAsyncWithUnmask (A.asyncOnWithUnmask cpu)

125asyncOnWithUnmask cpu = liftAsyncWithUnmask (A.asyncOnWithUnmask cpu)

103

126

104−-- | Generalized version of 'A.withAsync'.

127+-- | Lifted 'A.withAsync'.

105withAsync

128withAsync

106 :: AsyncE :> es

129 :: AsyncE :> es

107 => Eff es a

130 => Eff es a

@@ -109,7 +132,7 @@ withAsync

109 -> Eff es b

132 -> Eff es b

110withAsync = liftWithAsync A.withAsync

133withAsync = liftWithAsync A.withAsync

111

134

112−-- | Generalized version of 'A.withAsyncBound'.

135+-- | Lifted 'A.withAsyncBound'.

113withAsyncBound

136withAsyncBound

114 :: AsyncE :> es

137 :: AsyncE :> es

115 => Eff es a

138 => Eff es a

@@ -117,7 +140,7 @@ withAsyncBound

117 -> Eff es b

140 -> Eff es b

118withAsyncBound = liftWithAsync A.withAsyncBound

141withAsyncBound = liftWithAsync A.withAsyncBound

119

142

120−-- | Generalized version of 'A.withAsyncOn'.

143+-- | Lifted 'A.withAsyncOn'.

121withAsyncOn

144withAsyncOn

122 :: AsyncE :> es

145 :: AsyncE :> es

123 => Int

146 => Int

@@ -126,7 +149,7 @@ withAsyncOn

126 -> Eff es b

149 -> Eff es b

127withAsyncOn cpu = liftWithAsync (A.withAsyncOn cpu)

150withAsyncOn cpu = liftWithAsync (A.withAsyncOn cpu)

128

151

129−-- | Generalized version of 'A.withAsyncWithUnmask'.

152+-- | Lifted 'A.withAsyncWithUnmask'.

130withAsyncWithUnmask

153withAsyncWithUnmask

131 :: AsyncE :> es

154 :: AsyncE :> es

132 => ((forall c. Eff es c -> Eff es c) -> Eff es a)

155 => ((forall c. Eff es c -> Eff es c) -> Eff es a)

@@ -134,7 +157,7 @@ withAsyncWithUnmask

134 -> Eff es b

157 -> Eff es b

135withAsyncWithUnmask = liftWithAsyncWithUnmask A.withAsyncWithUnmask

158withAsyncWithUnmask = liftWithAsyncWithUnmask A.withAsyncWithUnmask

136

159

137−-- | Generalized version of 'A.withAsyncOnWithUnmask'.

160+-- | Lifted 'A.withAsyncOnWithUnmask'.

138withAsyncOnWithUnmask

161withAsyncOnWithUnmask

139 :: AsyncE :> es

162 :: AsyncE :> es

140 => Int

163 => Int

@@ -143,59 +166,59 @@ withAsyncOnWithUnmask

143 -> Eff es b

166 -> Eff es b

144withAsyncOnWithUnmask cpu = liftWithAsyncWithUnmask (A.withAsyncOnWithUnmask cpu)

167withAsyncOnWithUnmask cpu = liftWithAsyncWithUnmask (A.withAsyncOnWithUnmask cpu)

145

168

146−-- | Generalized version of 'A.wait'.

169+-- | Lifted 'A.wait'.

147wait :: AsyncE :> es => Async a -> Eff es a

170wait :: AsyncE :> es => Async a -> Eff es a

148wait = unsafeEff_ . A.wait

171wait = unsafeEff_ . A.wait

149

172

150−-- | Generalized version of 'A.poll'.

173+-- | Lifted 'A.poll'.

151poll

174poll

152 :: AsyncE :> es

175 :: AsyncE :> es

153 => Async a

176 => Async a

154 -> Eff es (Maybe (Either SomeException a))

177 -> Eff es (Maybe (Either SomeException a))

155poll = unsafeEff_ . A.poll

178poll = unsafeEff_ . A.poll

156

179

157−-- | Generalized version of 'A.cancel'.

180+-- | Lifted 'A.cancel'.

158cancel :: AsyncE :> es => Async a -> Eff es ()

181cancel :: AsyncE :> es => Async a -> Eff es ()

159cancel = unsafeEff_ . A.cancel

182cancel = unsafeEff_ . A.cancel

160

183

161−-- | Generalized version of 'A.cancelWith'.

184+-- | Lifted 'A.cancelWith'.

162cancelWith :: (Exception e, AsyncE :> es) => Async a -> e -> Eff es ()

185cancelWith :: (Exception e, AsyncE :> es) => Async a -> e -> Eff es ()

163cancelWith a = unsafeEff_ . A.cancelWith a

186cancelWith a = unsafeEff_ . A.cancelWith a

164

187

165−-- | Generalized version of 'A.uninterruptibleCancel'.

188+-- | Lifted 'A.uninterruptibleCancel'.

166uninterruptibleCancel :: AsyncE :> es => Async a -> Eff es ()

189uninterruptibleCancel :: AsyncE :> es => Async a -> Eff es ()

167uninterruptibleCancel = unsafeEff_ . A.uninterruptibleCancel

190uninterruptibleCancel = unsafeEff_ . A.uninterruptibleCancel

168

191

169−-- | Generalized version of 'A.waitCatch'.

192+-- | Lifted 'A.waitCatch'.

170waitCatch

193waitCatch

171 :: AsyncE :> es

194 :: AsyncE :> es

172 => Async a

195 => Async a

173 -> Eff es (Either SomeException a)

196 -> Eff es (Either SomeException a)

174waitCatch = unsafeEff_ . A.waitCatch

197waitCatch = unsafeEff_ . A.waitCatch

175

198

176−-- | Generalized version of 'A.waitAny'.

199+-- | Lifted 'A.waitAny'.

177waitAny :: AsyncE :> es => [Async a] -> Eff es (Async a, a)

200waitAny :: AsyncE :> es => [Async a] -> Eff es (Async a, a)

178waitAny = unsafeEff_ . A.waitAny

201waitAny = unsafeEff_ . A.waitAny

179

202

180−-- | Generalized version of 'A.waitAnyCatch'.

203+-- | Lifted 'A.waitAnyCatch'.

181waitAnyCatch

204waitAnyCatch

182 :: AsyncE :> es

205 :: AsyncE :> es

183 => [Async a]

206 => [Async a]

184 -> Eff es (Async a, Either SomeException a)

207 -> Eff es (Async a, Either SomeException a)

185waitAnyCatch = unsafeEff_ . A.waitAnyCatch

208waitAnyCatch = unsafeEff_ . A.waitAnyCatch

186

209

187−-- | Generalized version of 'A.waitAnyCancel'.

210+-- | Lifted 'A.waitAnyCancel'.

188waitAnyCancel :: AsyncE :> es => [Async a] -> Eff es (Async a, a)

211waitAnyCancel :: AsyncE :> es => [Async a] -> Eff es (Async a, a)

189waitAnyCancel = unsafeEff_ . A.waitAnyCancel

212waitAnyCancel = unsafeEff_ . A.waitAnyCancel

190

213

191−-- | Generalized version of 'A.waitAnyCatchCancel'.

214+-- | Lifted 'A.waitAnyCatchCancel'.

192waitAnyCatchCancel

215waitAnyCatchCancel

193 :: AsyncE :> es

216 :: AsyncE :> es

194 => [Async a]

217 => [Async a]

195 -> Eff es (Async a, Either SomeException a)

218 -> Eff es (Async a, Either SomeException a)

196waitAnyCatchCancel = unsafeEff_ . A.waitAnyCatchCancel

219waitAnyCatchCancel = unsafeEff_ . A.waitAnyCatchCancel

197

220

198−-- | Generalized version of 'A.waitEither'.

221+-- | Lifted 'A.waitEither'.

199waitEither

222waitEither

200 :: AsyncE :> es

223 :: AsyncE :> es

201 => Async a

224 => Async a

@@ -203,7 +226,7 @@ waitEither

203 -> Eff es (Either a b)

226 -> Eff es (Either a b)

204waitEither a b = unsafeEff_ $ A.waitEither a b

227waitEither a b = unsafeEff_ $ A.waitEither a b

205

228

206−-- | Generalized version of 'A.waitEitherCatch'.

229+-- | Lifted 'A.waitEitherCatch'.

207waitEitherCatch

230waitEitherCatch

208 :: AsyncE :> es

231 :: AsyncE :> es

209 => Async a

232 => Async a

@@ -211,7 +234,7 @@ waitEitherCatch

211 -> Eff es (Either (Either SomeException a) (Either SomeException b))

234 -> Eff es (Either (Either SomeException a) (Either SomeException b))

212waitEitherCatch a b = unsafeEff_ $ A.waitEitherCatch a b

235waitEitherCatch a b = unsafeEff_ $ A.waitEitherCatch a b

213

236

214−-- | Generalized version of 'A.waitEitherCancel'.

237+-- | Lifted 'A.waitEitherCancel'.

215waitEitherCancel

238waitEitherCancel

216 :: AsyncE :> es

239 :: AsyncE :> es

217 => Async a

240 => Async a

@@ -219,7 +242,7 @@ waitEitherCancel

219 -> Eff es (Either a b)

242 -> Eff es (Either a b)

220waitEitherCancel a b = unsafeEff_ $ A.waitEitherCancel a b

243waitEitherCancel a b = unsafeEff_ $ A.waitEitherCancel a b

221

244

222−-- | Generalized version of 'A.waitEitherCatchCancel'.

245+-- | Lifted 'A.waitEitherCatchCancel'.

223waitEitherCatchCancel

246waitEitherCatchCancel

224 :: AsyncE :> es

247 :: AsyncE :> es

225 => Async a

248 => Async a

@@ -227,103 +250,251 @@ waitEitherCatchCancel

227 -> Eff es (Either (Either SomeException a) (Either SomeException b))

250 -> Eff es (Either (Either SomeException a) (Either SomeException b))

228waitEitherCatchCancel a b = unsafeEff_ $ A.waitEitherCatch a b

251waitEitherCatchCancel a b = unsafeEff_ $ A.waitEitherCatch a b

229

252

230−-- | Generalized version of 'A.waitEither_'.

253+-- | Lifted 'A.waitEither_'.

231waitEither_ :: AsyncE :> es => Async a -> Async b -> Eff es ()

254waitEither_ :: AsyncE :> es => Async a -> Async b -> Eff es ()

232waitEither_ a b = unsafeEff_ $ A.waitEither_ a b

255waitEither_ a b = unsafeEff_ $ A.waitEither_ a b

233

256

234−-- | Generalized version of 'A.waitBoth'.

257+-- | Lifted 'A.waitBoth'.

235waitBoth :: AsyncE :> es => Async a -> Async b -> Eff es (a, b)

258waitBoth :: AsyncE :> es => Async a -> Async b -> Eff es (a, b)

236waitBoth a b = unsafeEff_ $ A.waitBoth a b

259waitBoth a b = unsafeEff_ $ A.waitBoth a b

237

260

238−-- | Generalized version of 'A.link'.

261+-- | Lifted 'A.link'.

239link :: AsyncE :> es => Async a -> Eff es ()

262link :: AsyncE :> es => Async a -> Eff es ()

240link = unsafeEff_ . A.link

263link = unsafeEff_ . A.link

241

264

242−-- | Generalized version of 'A.linkOnly'.

265+-- | Lifted 'A.linkOnly'.

243linkOnly :: AsyncE :> es => (SomeException -> Bool) -> Async a -> Eff es ()

266linkOnly :: AsyncE :> es => (SomeException -> Bool) -> Async a -> Eff es ()

244linkOnly f = unsafeEff_ . A.linkOnly f

267linkOnly f = unsafeEff_ . A.linkOnly f

245

268

246−-- | Generalized version of 'A.link2'.

269+-- | Lifted 'A.link2'.

247link2 :: AsyncE :> es => Async a -> Async b -> Eff es ()

270link2 :: AsyncE :> es => Async a -> Async b -> Eff es ()

248link2 a b = unsafeEff_ $ A.link2 a b

271link2 a b = unsafeEff_ $ A.link2 a b

249

272

250−-- | Generalized version of 'A.link2Only'.

273+-- | Lifted 'A.link2Only'.

251link2Only :: AsyncE :> es => (SomeException -> Bool) -> Async a -> Async b -> Eff es ()

274link2Only :: AsyncE :> es => (SomeException -> Bool) -> Async a -> Async b -> Eff es ()

252link2Only f a b = unsafeEff_ $ A.link2Only f a b

275link2Only f a b = unsafeEff_ $ A.link2Only f a b

253

276

254−-- | Generalized version of 'A.race'.

277+-- | Lifted 'A.race'.

255race :: AsyncE :> es => Eff es a -> Eff es b -> Eff es (Either a b)

278race :: AsyncE :> es => Eff es a -> Eff es b -> Eff es (Either a b)

256race ma mb = unsafeEff $ \es -> do

279race 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)

260

281

261−-- | Generalized version of 'A.race_'.

282+-- | Lifted 'A.race_'.

262race_ :: AsyncE :> es => Eff es a -> Eff es b -> Eff es ()

283race_ :: AsyncE :> es => Eff es a -> Eff es b -> Eff es ()

263race_ ma mb = unsafeEff $ \es -> do

284race_ 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)

267

286

268−-- | Generalized version of 'A.concurrently'.

287+-- | Lifted 'A.concurrently'.

269concurrently :: AsyncE :> es => Eff es a -> Eff es b -> Eff es (a, b)

288concurrently :: AsyncE :> es => Eff es a -> Eff es b -> Eff es (a, b)

270concurrently ma mb = unsafeEff $ \es -> do

289concurrently 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)

274

291

275−-- | Generalized version of 'A.concurrently_'.

292+-- | Lifted 'A.concurrently_'.

276concurrently_ :: AsyncE :> es => Eff es a -> Eff es b -> Eff es ()

293concurrently_ :: AsyncE :> es => Eff es a -> Eff es b -> Eff es ()

277concurrently_ ma mb = unsafeEff $ \es -> do

294concurrently_ 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.

281

300

282−-- | Generalized version of 'A.mapConcurrently'.

301+-- | Lifted 'A.mapConcurrently'.

283mapConcurrently

302mapConcurrently

284 :: (Traversable f, AsyncE :> es)

303 :: (Traversable f, AsyncE :> es)

285 => (a -> Eff es b)

304 => (a -> Eff es b)

286 -> f a

305 -> f a

287 -> Eff es (f b)

306 -> 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

289

309

290−-- | Generalized version of 'A.mapConcurrently_'.

310+-- | Lifted 'A.mapConcurrently_'.

291mapConcurrently_

311mapConcurrently_

292 :: (Foldable f, AsyncE :> es)

312 :: (Foldable f, AsyncE :> es)

293 => (a -> Eff es b)

313 => (a -> Eff es b)

294 -> f a

314 -> f a

295 -> Eff es ()

315 -> 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

297

318

298−-- | Generalized version of 'A.forConcurrently'.

319+-- | Lifted 'A.forConcurrently'.

299forConcurrently

320forConcurrently

300 :: (Traversable f, AsyncE :> es)

321 :: (Traversable f, AsyncE :> es)

301 => f a

322 => f a

302 -> (a -> Eff es b)

323 -> (a -> Eff es b)

303 -> Eff es (f b)

324 -> 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)

305

327

306−-- | Generalized version of 'A.forConcurrently_'.

328+-- | Lifted 'A.forConcurrently_'.

307forConcurrently_

329forConcurrently_

308 :: (Foldable f, AsyncE :> es)

330 :: (Foldable f, AsyncE :> es)

309 => f a

331 => f a

310 -> (a -> Eff es b)

332 -> (a -> Eff es b)

311 -> Eff es ()

333 -> Eff es ()

312−forConcurrently_ = flip mapConcurrently_

334+forConcurrently_ t f = unsafeEff $ \es -> do

335+ U.forConcurrently_ t (\a -> unEff (f a) =<< cloneEnv es)

313

336

314−-- | Generalized version of 'A.replicateConcurrently'.

337+-- | Lifted 'A.replicateConcurrently'.

315replicateConcurrently :: AsyncE :> es => Int -> Eff es a -> Eff es [a]

338replicateConcurrently :: AsyncE :> es => Int -> Eff es a -> Eff es [a]

316−replicateConcurrently n =

339+replicateConcurrently n f = unsafeEff $ \es -> do

317− runConcurrently . sequenceA . replicate n . Concurrently

340+ U.replicateConcurrently n (unEff f =<< cloneEnv es)

318

341

319−-- | Generalized version of 'A.replicateConcurrently_'.

342+-- | Lifted 'A.replicateConcurrently_'.

320replicateConcurrently_ :: AsyncE :> es => Int -> Eff es a -> Eff es ()

343replicateConcurrently_ :: AsyncE :> es => Int -> Eff es a -> Eff es ()

321−replicateConcurrently_ n =

344+replicateConcurrently_ n f = unsafeEff $ \es -> do

322− runConcurrently . fold . replicate n . Concurrently . void

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

323

493

324----------------------------------------

494----------------------------------------

495+-- Concurrently

325

496

326−-- | Generalized version of 'A.Concurrently'.

497+-- | Lifted 'A.Concurrently'.

327newtype Concurrently es a = Concurrently { runConcurrently :: Eff es a }

498newtype Concurrently es a = Concurrently { runConcurrently :: Eff es a }

328

499

329instance Functor (Concurrently es) where

500instance Functor (Concurrently es) where

tests/AsyncTests.hs+28

@@ -19,6 +19,8 @@ asyncTests = testGroup "Async"

19 [ testCase "local state" test_localState

20 , testCase "shared state" test_sharedState

21 , testCase "error handling" test_errorHandling

22+ , testCase "async with unmask" test_asyncWithUnmask

23+ , testCase "pooled workers" test_pooledWorkers

22 ]

24 ]

24test_localState :: Assertion

26test_localState :: Assertion

@@ -70,3 +72,29 @@ test_errorHandling = runIOE . runAsyncE . evalStateMVar (0::Int) $ do

71 err :: String

73 err :: String

72 err = "thrown from async"

74 err = "thrown from async"

75+

76+test_asyncWithUnmask :: Assertion

77+test_asyncWithUnmask = runIOE . runAsyncE . evalState "initial" $ do

78+ x <- asyncWithUnmask $ \unmask -> do

79+ liftIO $ threadDelay 10000

80+ r1 <- get @String -- 2

81+ unmask $ put "unmask"

82+ r2 <- get @String -- 3

83+ pure (r1, r2)

84+ put "changed" -- 1

85+ (inner1, inner2) <- wait x

86+ outer <- get -- 4

87+ U.assertEqual "expected result"

88+ ("initial", "unmask", "changed")

89+ (inner1, inner2, outer)

90+

91+test_pooledWorkers :: Assertion

92+test_pooledWorkers = runIOE . runAsyncE . evalState (0::Int) $ do

93+ x <- pooledForConcurrentlyN threads [1..n] $ \k -> do

94+ r <- get @Int

95+ modify @Int (+1)

96+ pure $ k + r

97+ U.assertEqual "expected result" [1..n] x

98+ where

99+ n = 10

100+ threads = 4