29f64fe · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Separate types for different styles of product composition

29f64feHarendra Kumar9 years ago

benchmark/Main.hs+57−28

compare with master

unifiedsplit

@@ -1,4 +1,5 @@

1{-# LANGUAGE CPP #-}

2+{-# LANGUAGE RankNTypes #-}

3module Main where

4module Main where

@@ -30,35 +31,38 @@ main :: IO ()

30main = do

31main = do

31 -- XXX due to a GHC bug passing bind as an argument causes perf

32 -- XXX due to a GHC bug passing bind as an argument causes perf

32 -- degradation, so we should keep that in account when comparing.

33 -- degradation, so we should keep that in account when comparing.

33− let as = asyncly_basic

34+ let as = asyncly_serial

35+ ai = asyncly_interleaved

36+ aa = asyncly_async

37+ ap = asyncly_parallel

34 defaultMain [

38 defaultMain [

35 bgroup "asyncly"

39 bgroup "asyncly"

36 [ bgroup "serial bind"

40 [ bgroup "serial bind"

37− [ bench "serial" $ nfIO (as (>>=) (A.<>))

41+ [ bench "serial" $ nfIO (as (A.<>))

38− , bench "fair serial" $ nfIO (as (>>=) (A.<=>))

42+ , bench "fair serial" $ nfIO (as (A.<=>))

39− , bench "left parallel" $ nfIO (as (>>=) (A.<|))

43+ , bench "left parallel" $ nfIO (as (A.<|))

40− , bench "fair parallel" $ nfIO (as (>>=) (A.<|>))

44+ , bench "fair parallel" $ nfIO (as (A.<|>))

41 ]

45 ]

43− , bgroup "fair bind"

47+ , bgroup "interleaved bind"

44− [ bench "serial" $ nfIO (as (A.>->) (A.<>))

48+ [ bench "serial" $ nfIO (ai (A.<>))

45− , bench "fair serial" $ nfIO (as (A.>->) (A.<=>))

49+ , bench "fair serial" $ nfIO (ai (A.<=>))

46− , bench "left parallel" $ nfIO (as (A.>->) (A.<|))

50+ , bench "left parallel" $ nfIO (ai (A.<|))

47− , bench "fair parallel" $ nfIO (as (A.>->) (A.<|>))

51+ , bench "fair parallel" $ nfIO (ai (A.<|>))

48 ]

52 ]

50− , bgroup "parallel bind"

54+ , bgroup "async bind"

51− [ bench "serial" $ nfIO (as (A.>>|) (A.<>))

55+ [ bench "serial" $ nfIO (aa (A.<>))

52− , bench "fair serial" $ nfIO (as (A.>>|) (A.<=>))

56+ , bench "fair serial" $ nfIO (aa (A.<=>))

53− , bench "left parallel" $ nfIO (as (A.>>|) (A.<|))

57+ , bench "left parallel" $ nfIO (aa (A.<|))

54− , bench "fair parallel" $ nfIO (as (A.>>|) (A.<|>))

58+ , bench "fair parallel" $ nfIO (aa (A.<|>))

55 ]

59 ]

57− , bgroup "fair parallel bind"

61+ , bgroup "parallel bind"

58− [ bench "serial" $ nfIO (as (A.>|>) (A.<>))

62+ [ bench "serial" $ nfIO (ap (A.<>))

59− , bench "fair serial" $ nfIO (as (A.>|>) (A.<=>))

63+ , bench "fair serial" $ nfIO (ap (A.<=>))

60− , bench "left parallel" $ nfIO (as (A.>|>) (A.<|))

64+ , bench "left parallel" $ nfIO (ap (A.<|))

61− , bench "fair parallel" $ nfIO (as (A.>|>) (A.<|>))

65+ , bench "fair parallel" $ nfIO (ap (A.<|>))

62 ]

66 ]

64 -- Benchmark smallest possible actions composed together

68 -- Benchmark smallest possible actions composed together

@@ -91,31 +95,56 @@ map f x = return $ f x

91filter :: (Monad m, Alternative m) => (a -> Bool) -> a -> m a

95filter :: (Monad m, Alternative m) => (a -> Bool) -> a -> m a

92filter cond x = guard (not $ cond x) >> return x

96filter cond x = guard (not $ cond x) >> return x

94−amap :: (Int -> Int) -> Int -> A.AsyncT IO Int

98+amap :: Monad (s IO) => (Int -> Int) -> Int -> s IO Int

95amap = Main.map

99amap = Main.map

100

97−afilter :: (Int -> Bool) -> Int -> A.AsyncT IO Int

101+afilter :: (Alternative (s IO), Monad (s IO)) => (Int -> Bool) -> Int -> s IO Int

98afilter = Main.filter

102afilter = Main.filter

103

100{-# INLINE asyncly_basic #-}

104{-# INLINE asyncly_basic #-}

101asyncly_basic

105asyncly_basic

102− :: (A.AsyncT IO Int -> (Int -> A.AsyncT IO Int) -> A.AsyncT IO Int)

106+ :: (Alternative (s IO), Monad (s IO), Monoid (s IO Int), A.Streaming s) => (forall a. s IO a -> IO [a])

103− -> (A.AsyncT IO Int -> A.AsyncT IO Int -> A.AsyncT IO Int)

107+ -> (s IO Int -> s IO Int -> s IO Int)

104 -> IO Int

108 -> IO Int

105−asyncly_basic f g = do

109+asyncly_basic tl g = do

106− xs <- A.toList $ do

110+ xs <- tl $ do

107 A.drop 100 (A.forEachWith g [1..100000 :: Int] $ \x ->

111 A.drop 100 (A.forEachWith g [1..100000 :: Int] $ \x ->

108 afilter even x `f` amap (+1))

112 afilter even x `f` amap (+1))

109 `f` amap (+1)

113 `f` amap (+1)

110 `f` afilter (\y -> y `mod` 2 == 0)

114 `f` afilter (\y -> y `mod` 2 == 0)

111 assert (Prelude.length xs == 49900) $

115 assert (Prelude.length xs == 49900) $

112 return (Prelude.length xs)

116 return (Prelude.length xs)

117+ where f = (>>=)

118+

119+{-# INLINE asyncly_serial #-}

120+asyncly_serial

121+ :: (A.StreamT IO Int -> A.StreamT IO Int -> A.StreamT IO Int)

122+ -> IO Int

123+asyncly_serial = asyncly_basic A.toListSerial

124+

125+{-# INLINE asyncly_interleaved #-}

126+asyncly_interleaved

127+ :: (A.InterleavedT IO Int -> A.InterleavedT IO Int -> A.InterleavedT IO Int)

128+ -> IO Int

129+asyncly_interleaved = asyncly_basic A.toListInterleaved

130+

131+{-# INLINE asyncly_async #-}

132+asyncly_async

133+ :: (A.AsyncT IO Int -> A.AsyncT IO Int -> A.AsyncT IO Int)

134+ -> IO Int

135+asyncly_async = asyncly_basic A.toListAsync

136+

137+{-# INLINE asyncly_parallel #-}

138+asyncly_parallel

139+ :: (A.ParallelT IO Int -> A.ParallelT IO Int -> A.ParallelT IO Int)

140+ -> IO Int

141+asyncly_parallel = asyncly_basic A.toListParallel

113

142

114{-# INLINE asyncly_nil #-}

143{-# INLINE asyncly_nil #-}

115−asyncly_nil :: (A.AsyncT IO Int -> A.AsyncT IO Int -> A.AsyncT IO Int)

144+asyncly_nil :: (A.StreamT IO Int -> A.StreamT IO Int -> A.StreamT IO Int)

116 -> IO Int

145 -> IO Int

117asyncly_nil f = do

146asyncly_nil f = do

118− xs <- A.toList $ do

147+ xs <- A.toListSerial $ do

119 (A.forEachWith f [1..100000:: Int] $

148 (A.forEachWith f [1..100000:: Int] $

120 \x -> return x >>= return . id)

149 \x -> return x >>= return . id)

121 assert (Prelude.length xs == 100000) $

150 assert (Prelude.length xs == 100000) $

src/Asyncly.hs+28−15

@@ -52,24 +52,36 @@

52-- A simple inline example here illustrating applicative, monad and alternative

53-- compositions.

54module Asyncly

55− ( AsyncT

56− , MonadAsync

55+ ( MonadAsync

56+ , Streaming

57+ ( fromStream

58+ , fromInterleaved

59+ , fromAsync

60+ , fromParallel

61+ , fromZipStream

62+ , fromZipAsync

63+ , (<=>)

64+ , (<|)

65+ )

58− -- * Running

67+ -- * Stream Styles

59− , runAsyncly

68+ , StreamT

60− , toList

69+ , InterleavedT

61− , async

70+ , AsyncT

71+ , ParallelT

72+ , ZipStream

73+ , ZipAsync

63− -- * Monadic Composition (Conjunction)

75+ -- * Running Streams

64− -- $bind

76+ , serially

65− , (>->)

77+ , interleaved

66− , (>>|)

78+ , asyncly

67− , (>|>)

79+ , parallely

69− -- * Monoidal Composition (Disjunction)

81+ , toListSerial

70− -- $monoidal

82+ , toListInterleaved

71− , (<=>)

83+ , toListAsync

72− , (<|)

84+ , toListParallel

74 -- * General Fold Utilities

86 -- * General Fold Utilities

75 , foldWith

87 , foldWith

@@ -87,6 +99,7 @@ module Asyncly

87where

99where

100

89import Asyncly.AsyncT

101import Asyncly.AsyncT

102+import Asyncly.Prelude

90import Data.Semigroup (Semigroup(..))

103import Data.Semigroup (Semigroup(..))

91import Control.Applicative (Alternative(..))

104import Control.Applicative (Alternative(..))

92import Control.Monad (MonadPlus(..))

105import Control.Monad (MonadPlus(..))

src/Asyncly/AsyncT.hs+84−166

@@ -16,27 +16,25 @@

16--

17--

18module Asyncly.AsyncT

19− ( AsyncT (..)

19+ ( Stream (..)

20 , MonadAsync

21− , runAsyncly

21+

22+ , fromCallback

23+

24+ , runStreaming

22 , toList

25 , toList

23 , async

26 , async

25 , interleave

28 , interleave

26− , (<=>)

29+ , parallel

27− , parAhead

28− , (<>|)

29 , parLeft

30 , parLeft

30− , (<|)

31+ , CtxType (..)

31−

32+ , SchedPolicy (..)

32− , (>->)

33+ , Dimension (..)

33− , (>>|)

34− , (>|>)

36 , foldWith

35 , foldWith

37 , foldMapWith

36 , foldMapWith

38 , forEachWith

37 , forEachWith

39− , fromCallback

40 )

38 )

41where

39where

@@ -101,7 +99,7 @@ data CtxType = CtxType Dimension SchedPolicy deriving Eq

101data Context m a =

99data Context m a =

102 Context { outputQueue :: IORef [ChildEvent a]

100 Context { outputQueue :: IORef [ChildEvent a]

103 , doorBell :: MVar Bool -- wakeup mechanism for outQ

101 , doorBell :: MVar Bool -- wakeup mechanism for outQ

104− , enqueue :: AsyncT m a -> IO ()

102+ , enqueue :: Stream m a -> IO ()

105 , runqueue :: m ()

103 , runqueue :: m ()

106 , runningThreads :: IORef (Set ThreadId)

104 , runningThreads :: IORef (Set ThreadId)

107 , queueEmpty :: m Bool

105 , queueEmpty :: m Bool

@@ -112,12 +110,12 @@ data Context m a =

112-- in monad 'm'. 'AsyncT' streams can be composed sequentially or in parallel

110-- in monad 'm'. 'AsyncT' streams can be composed sequentially or in parallel

113-- in various ways using monadic bind or variants of it and using monoidal

111-- in various ways using monadic bind or variants of it and using monoidal

114-- compositions like 'Monoid', 'Alternative' or variants of these.

112-- compositions like 'Monoid', 'Alternative' or variants of these.

115−newtype AsyncT m a =

113+newtype Stream m a =

116− AsyncT {

114+ Stream {

117− runAsyncT :: forall r.

115+ runStream :: forall r.

118 Maybe (Context m a) -- local state

116 Maybe (Context m a) -- local state

119 -> m r -- stop

117 -> m r -- stop

120− -> (a -> Maybe (AsyncT m a) -> m r) -- yield

118+ -> (a -> Maybe (Stream m a) -> m r) -- yield

121 -> m r

119 -> m r

122 }

120 }

123

121

@@ -130,40 +128,31 @@ type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)

130

128

131-- | '<>' concatenates two 'AsyncT' streams sequentially i.e. the first stream

129-- | '<>' concatenates two 'AsyncT' streams sequentially i.e. the first stream

132-- is exhausted completely before yielding any element from the second stream.

130-- is exhausted completely before yielding any element from the second stream.

133−instance Semigroup (AsyncT m a) where

131+instance Semigroup (Stream m a) where

134 m1 <> m2 = go m1

132 m1 <> m2 = go m1

135 where

133 where

136− go (AsyncT m) = AsyncT $ \_ stp yld ->

134+ go (Stream m) = Stream $ \_ stp yld ->

137− let stop = (runAsyncT m2) Nothing stp yld

135+ let stop = (runStream m2) Nothing stp yld

138 yield a Nothing = yld a (Just m2)

136 yield a Nothing = yld a (Just m2)

139 yield a (Just r) = yld a (Just (go r))

137 yield a (Just r) = yld a (Just (go r))

140 in m Nothing stop yield

138 in m Nothing stop yield

141

139

142−instance Monoid (AsyncT m a) where

140+instance Monoid (Stream m a) where

143− mempty = AsyncT $ \_ stp _ -> stp

141+ mempty = Stream $ \_ stp _ -> stp

144 mappend = (<>)

142 mappend = (<>)

145

143

146−-- | Same as '<=>'.

147−interleave :: AsyncT m a -> AsyncT m a -> AsyncT m a

144+interleave :: Stream m a -> Stream m a -> Stream m a

148−interleave m1 m2 = AsyncT $ \_ stp yld -> do

145+interleave m1 m2 = Stream $ \_ stp yld -> do

149− let stop = (runAsyncT m2) Nothing stp yld

146+ let stop = (runStream m2) Nothing stp yld

150 yield a Nothing = yld a (Just m2)

147 yield a Nothing = yld a (Just m2)

151 yield a (Just r) = yld a (Just (interleave m2 r))

148 yield a (Just r) = yld a (Just (interleave m2 r))

152− (runAsyncT m1) Nothing stop yield

149+ (runStream m1) Nothing stop yield

153−

154−infixr 5 <=>

155−

156−-- | Sequential interleaved composition, similar to '<>' except that it

157−-- fairly interleaves the two 'AsyncT' streams, yielding an element from each

158−-- stream alternately.

159−(<=>) :: AsyncT m a -> AsyncT m a -> AsyncT m a

160−(<=>) = interleave

161

150

162------------------------------------------------------------------------------

151------------------------------------------------------------------------------

163-- Functor

152-- Functor

164------------------------------------------------------------------------------

153------------------------------------------------------------------------------

165

154

166−instance Monad m => Functor (AsyncT m) where

155+instance Monad m => Functor (Stream m) where

167 fmap = liftM

156 fmap = liftM

168

157

169------------------------------------------------------------------------------

158------------------------------------------------------------------------------

@@ -172,88 +161,25 @@ instance Monad m => Functor (AsyncT m) where

172

161

173-- The newtype ZipAsync provides a parallel zip applicative instance

162-- The newtype ZipAsync provides a parallel zip applicative instance

174

163

175−instance Monad m => Applicative (AsyncT m) where

164+instance Monad m => Applicative (Stream m) where

176− pure a = AsyncT $ \_ _ yld -> yld a Nothing

165+ pure a = Stream $ \_ _ yld -> yld a Nothing

177 (<*>) = ap

166 (<*>) = ap

178

167

179------------------------------------------------------------------------------

168------------------------------------------------------------------------------

180-- Monad

169-- Monad

181------------------------------------------------------------------------------

170------------------------------------------------------------------------------

182

171

183−-- | A thread context is valid only until the next bind. Upon a bind we

184−-- reset the context to Nothing.

185−{-# INLINE bindWith #-}

186−bindWith

187− :: (forall c. AsyncT m c -> AsyncT m c -> AsyncT m c)

188− -> AsyncT m a

189− -> (a -> AsyncT m b)

190− -> AsyncT m b

191−bindWith k m f = go m

192− where

193− go (AsyncT g) =

194− AsyncT $ \_ stp yld ->

195− let run x = (runAsyncT x) Nothing stp yld

196− yield a Nothing = run $ f a

197− yield a (Just r) = run $ f a `k` (go r)

198− in g Nothing stp yield

199−

200-- | Execute a monadic action sequentially for each element in the 'AsyncT'

172-- | Execute a monadic action sequentially for each element in the 'AsyncT'

201-- stream, i.e. an iteration finishes completely before the next one starts.

173-- stream, i.e. an iteration finishes completely before the next one starts.

202−instance Monad m => Monad (AsyncT m) where

174+instance Monad m => Monad (Stream m) where

203 return = pure

175 return = pure

204

176

205− AsyncT m >>= f = AsyncT $ \_ stp yld ->

177+ Stream m >>= f = Stream $ \_ stp yld ->

206− let run x = (runAsyncT x) Nothing stp yld

178+ let run x = (runStream x) Nothing stp yld

207 yield a Nothing = run $ f a

179 yield a Nothing = run $ f a

208 yield a (Just r) = run $ f a <> (r >>= f)

180 yield a (Just r) = run $ f a <> (r >>= f)

209 in m Nothing stp yield

181 in m Nothing stp yield

210

182

211−------------------------------------------------------------------------------

212−-- Alternative ways to bind

213−------------------------------------------------------------------------------

214−

215−-- XXX Do we need newtype wrappers for these?

216−

217−-- | Execute a monadic action for each element in the stream, in a fairly

218−-- interleaved manner i.e. iterations yield alternately.

219−(>->) :: AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b

220−(>->) = bindWith (<=>)

221−

222−{-# INLINE parbind #-}

223−parbind

224− :: (forall c. AsyncT m c -> AsyncT m c -> AsyncT m c)

225− -> AsyncT m a

226− -> (a -> AsyncT m b)

227− -> AsyncT m b

228−parbind k m f = go m

229− where

230− go (AsyncT g) =

231− AsyncT $ \ctx stp yld ->

232− let run x = (runAsyncT x) ctx stp yld

233− yield a Nothing = run $ f a

234− yield a (Just r) = run $ f a `k` (go r)

235− in g Nothing stp yield

236−

237−-- | Execute a monadic action for each element in the stream, running

238−-- iterations in parallel, but giving preference to iterations started earlier.

239−(>>|) :: MonadAsync m => AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b

240−(>>|) = parbind (parallel (CtxType Conjunction LIFO))

241−

242−-- | Execute a monadic action for each element in the stream, running

243−-- iterations in a fairly parallel manner, i.e. all iterations are equally

244−-- likely to run.

245−(>|>) :: MonadAsync m => AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b

246−(>|>) = parbind (parallel (CtxType Conjunction FIFO))

247−

248−-- TBD Ideally we would want parallelism to be completely hidden from the user,

249−-- for example we can automatically decide where to start an action in parallel

250−-- so that it is most efficient. We can decide whether a particular bind should

251−-- fork a parallel thread or not based on the level of the bind in the tree.

252−-- This is equivalent to dynamically deciding (maybe based user policies

253−-- applied using combinators) which type of bind or monoidal composition we

254−-- should use. We can have another bind operator (maybe the standard >>= should

255−-- be used for that) for this type of behavior.

256−

257------------------------------------------------------------------------------

183------------------------------------------------------------------------------

258-- Alternative

184-- Alternative

259------------------------------------------------------------------------------

185------------------------------------------------------------------------------

@@ -287,10 +213,10 @@ sendStop ctx = liftIO myThreadId >>= \tid -> send ctx (ChildStop tid Nothing)

287

213

288-- Note: Left associated operations can grow this queue to a large size

214-- Note: Left associated operations can grow this queue to a large size

289{-# INLINE enqueueLIFO #-}

215{-# INLINE enqueueLIFO #-}

290−enqueueLIFO :: IORef [AsyncT m a] -> AsyncT m a -> IO ()

216+enqueueLIFO :: IORef [Stream m a] -> Stream m a -> IO ()

291enqueueLIFO q m = atomicModifyIORefCAS_ q $ \ ms -> m : ms

217enqueueLIFO q m = atomicModifyIORefCAS_ q $ \ ms -> m : ms

292

218

293−runqueueLIFO :: MonadIO m => Context m a -> IORef [AsyncT m a] -> m ()

219+runqueueLIFO :: MonadIO m => Context m a -> IORef [Stream m a] -> m ()

294runqueueLIFO ctx q = run

220runqueueLIFO ctx q = run

295

221

296 where

222 where

@@ -299,11 +225,11 @@ runqueueLIFO ctx q = run

299 work <- dequeue

225 work <- dequeue

300 case work of

226 case work of

301 Nothing -> sendStop ctx

227 Nothing -> sendStop ctx

302− Just m -> (runAsyncT m) (Just ctx) run yield

228+ Just m -> (runStream m) (Just ctx) run yield

303

229

304 sendit a = send ctx (ChildYield a)

230 sendit a = send ctx (ChildYield a)

305 yield a Nothing = sendit a >> run

231 yield a Nothing = sendit a >> run

306− yield a (Just r) = sendit a >> (runAsyncT r) (Just ctx) run yield

232+ yield a (Just r) = sendit a >> (runStream r) (Just ctx) run yield

307

233

308 dequeue = liftIO $ atomicModifyIORefCAS q $ \ ms ->

234 dequeue = liftIO $ atomicModifyIORefCAS q $ \ ms ->

309 case ms of

235 case ms of

@@ -311,10 +237,10 @@ runqueueLIFO ctx q = run

311 x : xs -> (xs, Just x)

237 x : xs -> (xs, Just x)

312

238

313{-# INLINE enqueueFIFO #-}

239{-# INLINE enqueueFIFO #-}

314−enqueueFIFO :: LinkedQueue (AsyncT m a) -> AsyncT m a -> IO ()

240+enqueueFIFO :: LinkedQueue (Stream m a) -> Stream m a -> IO ()

315enqueueFIFO = pushL

241enqueueFIFO = pushL

316

242

317−runqueueFIFO :: MonadIO m => Context m a -> LinkedQueue (AsyncT m a) -> m ()

243+runqueueFIFO :: MonadIO m => Context m a -> LinkedQueue (Stream m a) -> m ()

318runqueueFIFO ctx q = run

244runqueueFIFO ctx q = run

319

245

320 where

246 where

@@ -323,7 +249,7 @@ runqueueFIFO ctx q = run

323 work <- dequeue

249 work <- dequeue

324 case work of

250 case work of

325 Nothing -> sendStop ctx

251 Nothing -> sendStop ctx

326− Just m -> (runAsyncT m) (Just ctx) run yield

252+ Just m -> (runStream m) (Just ctx) run yield

327

253

328 dequeue = liftIO $ tryPopR q

254 dequeue = liftIO $ tryPopR q

329 sendit a = send ctx (ChildYield a)

255 sendit a = send ctx (ChildYield a)

@@ -387,8 +313,8 @@ instance Exception ContextUsedAfterEOF

387

313

388-- | Pull an AsyncT stream from a context

314-- | Pull an AsyncT stream from a context

389{-# NOINLINE pullFromCtx #-}

315{-# NOINLINE pullFromCtx #-}

390−pullFromCtx :: MonadAsync m => Context m a -> AsyncT m a

316+pullFromCtx :: MonadAsync m => Context m a -> Stream m a

391−pullFromCtx ctx = AsyncT $ \_ stp yld -> do

317+pullFromCtx ctx = Stream $ \_ stp yld -> do

392 -- When using an async handle to the context, one may keep using a stale

318 -- When using an async handle to the context, one may keep using a stale

393 -- context even after it has been fully drained. To detect it gracefully we

319 -- context even after it has been fully drained. To detect it gracefully we

394 -- raise an explicit exception.

320 -- raise an explicit exception.

@@ -403,7 +329,7 @@ pullFromCtx ctx = AsyncT $ \_ stp yld -> do

403 -- To avoid lock overhead we read all events at once instead of reading one

329 -- To avoid lock overhead we read all events at once instead of reading one

404 -- at a time. We just reverse the list to process the events in the order

330 -- at a time. We just reverse the list to process the events in the order

405 -- they arrived. Maybe we can use a queue instead?

331 -- they arrived. Maybe we can use a queue instead?

406− (runAsyncT $ processEvents (reverse list)) Nothing stp yld

332+ (runStream $ processEvents (reverse list)) Nothing stp yld

407

333

408 where

334 where

409

335

@@ -414,14 +340,14 @@ pullFromCtx ctx = AsyncT $ \_ stp yld -> do

414 throwM e

340 throwM e

415

341

416 {-# INLINE processEvents #-}

342 {-# INLINE processEvents #-}

417− processEvents [] = AsyncT $ \_ stp yld -> do

343+ processEvents [] = Stream $ \_ stp yld -> do

418 done <- allThreadsDone ctx

344 done <- allThreadsDone ctx

419 if not done

345 if not done

420− then (runAsyncT (pullFromCtx ctx)) Nothing stp yld

346+ then (runStream (pullFromCtx ctx)) Nothing stp yld

421 else stp

347 else stp

422

348

423− processEvents (ev : es) = AsyncT $ \_ stp yld -> do

349+ processEvents (ev : es) = Stream $ \_ stp yld -> do

424− let continue = (runAsyncT (processEvents es)) Nothing stp yld

350+ let continue = (runStream (processEvents es)) Nothing stp yld

425 yield a = yld a (Just (processEvents es))

351 yield a = yld a (Just (processEvents es))

426

352

427 case ev of

353 case ev of

@@ -469,7 +395,7 @@ getLifoCtx ctype = do

469-- | Split the original computation in a pull-push pair. The original

395-- | Split the original computation in a pull-push pair. The original

470-- computation pulls from a Channel while m1 and m2 push to the channel.

396-- computation pulls from a Channel while m1 and m2 push to the channel.

471pushPairToCtx :: MonadAsync m

397pushPairToCtx :: MonadAsync m

472− => CtxType -> AsyncT m a -> AsyncT m a -> m (Context m a)

398+ => CtxType -> Stream m a -> Stream m a -> m (Context m a)

473pushPairToCtx ctype m1 m2 = do

399pushPairToCtx ctype m1 m2 = do

474 -- Note: We must have all the work on the queue before sending the

400 -- Note: We must have all the work on the queue before sending the

475 -- pushworker, otherwise the pushworker may exit before we even get a

401 -- pushworker, otherwise the pushworker may exit before we even get a

@@ -489,10 +415,10 @@ pushPairToCtx ctype m1 m2 = do

489

415

490{-# NOINLINE pushPullFork #-}

416{-# NOINLINE pushPullFork #-}

491pushPullFork :: MonadAsync m

417pushPullFork :: MonadAsync m

492− => CtxType -> AsyncT m a -> AsyncT m a -> AsyncT m a

418+ => CtxType -> Stream m a -> Stream m a -> Stream m a

493−pushPullFork ct m1 m2 = AsyncT $ \_ stp yld -> do

419+pushPullFork ct m1 m2 = Stream $ \_ stp yld -> do

494 ctx <- pushPairToCtx ct m1 m2

420 ctx <- pushPairToCtx ct m1 m2

495− (runAsyncT (pullFromCtx ctx)) Nothing stp yld

421+ (runStream (pullFromCtx ctx)) Nothing stp yld

496

422

497-- Concurrency rate control. Our objective is to create more threads on

423-- Concurrency rate control. Our objective is to create more threads on

498-- demand if the consumer is running faster than us. As soon as we

424-- demand if the consumer is running faster than us. As soon as we

@@ -528,20 +454,20 @@ pushPullFork ct m1 m2 = AsyncT $ \_ stp yld -> do

528-- Note: This is designed to scale for right associated compositions,

454-- Note: This is designed to scale for right associated compositions,

529-- therefore always use a right fold for folding bigger structures.

455-- therefore always use a right fold for folding bigger structures.

530{-# INLINE parallel #-}

456{-# INLINE parallel #-}

531−parallel :: MonadAsync m => CtxType -> AsyncT m a -> AsyncT m a -> AsyncT m a

457+parallel :: MonadAsync m => CtxType -> Stream m a -> Stream m a -> Stream m a

532−parallel ct m1 m2 = AsyncT $ \ctx stp yld -> do

458+parallel ct m1 m2 = Stream $ \ctx stp yld -> do

533 case ctx of

459 case ctx of

534− Nothing -> (runAsyncT (pushPullFork ct m1 m2)) Nothing stp yld

460+ Nothing -> (runStream (pushPullFork ct m1 m2)) Nothing stp yld

535 Just c | ctxType c /= ct ->

461 Just c | ctxType c /= ct ->

536− (runAsyncT (pushPullFork ct m1 m2)) Nothing stp yld

462+ (runStream (pushPullFork ct m1 m2)) Nothing stp yld

537− Just c -> liftIO ((enqueue c) m2) >> (runAsyncT m1) ctx stp yld

463+ Just c -> liftIO ((enqueue c) m2) >> (runStream m1) ctx stp yld

538

464

539-- | `empty` represents an action that takes non-zero time to complete. Since

465-- | `empty` represents an action that takes non-zero time to complete. Since

540-- all actions take non-zero time, an `Alternative` composition ('<|>') is a

466-- all actions take non-zero time, an `Alternative` composition ('<|>') is a

541-- monoidal composition executing all actions in parallel, it is similar to

467-- monoidal composition executing all actions in parallel, it is similar to

542-- '<>' except that it runs all the actions in parallel and interleaves their

468-- '<>' except that it runs all the actions in parallel and interleaves their

543-- results fairly.

469-- results fairly.

544−instance MonadAsync m => Alternative (AsyncT m) where

470+instance MonadAsync m => Alternative (Stream m) where

545 empty = mempty

471 empty = mempty

546

472

547 {-# INLINE (<|>) #-}

473 {-# INLINE (<|>) #-}

@@ -551,37 +477,30 @@ instance MonadAsync m => Alternative (AsyncT m) where

551-- Other monoidal compositions for parallel actions

477-- Other monoidal compositions for parallel actions

552------------------------------------------------------------------------------

478------------------------------------------------------------------------------

553

479

480+{-

554-- | Same as '<>|'.

481-- | Same as '<>|'.

555−parAhead :: AsyncT m a -> AsyncT m a -> AsyncT m a

482+parAhead :: Stream m a -> Stream m a -> Stream m a

556parAhead = undefined

483parAhead = undefined

557

484

558-- | Sequential composition similar to '<>' except that it can execute the

485-- | Sequential composition similar to '<>' except that it can execute the

559-- action on the right in parallel ahead of time. Returns the results in

486-- action on the right in parallel ahead of time. Returns the results in

560-- sequential order like '<>' from left to right.

487-- sequential order like '<>' from left to right.

561−(<>|) :: AsyncT m a -> AsyncT m a -> AsyncT m a

488+(<>|) :: Stream m a -> Stream m a -> Stream m a

562(<>|) = parAhead

489(<>|) = parAhead

490+-}

563

491

564-- | Same as '<|'.

492-- | Same as '<|'.

565{-# INLINE parLeft #-}

493{-# INLINE parLeft #-}

566−parLeft :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a

494+parLeft :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

567parLeft = parallel (CtxType Disjunction LIFO)

495parLeft = parallel (CtxType Disjunction LIFO)

568

496

569−-- | Parallel composition similar to `<|>` except that it is left-biased

570−-- instead of being fair. Action on the left is likely to be given a chance to

571−-- execute before the action on the right. If the left action keeps yielding

572−-- results without blocking it continues running until it finishes and only

573−-- then the right action runs.

574−{-# INLINE (<|) #-}

575−(<|) :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a

576−(<|) = parLeft

577−

578−instance MonadAsync m => MonadPlus (AsyncT m) where

497+instance MonadAsync m => MonadPlus (Stream m) where

579 mzero = empty

498 mzero = empty

580 mplus = (<|>)

499 mplus = (<|>)

581

500

582-- | The computation specified in the argument is pushed to a new thread and

501-- | The computation specified in the argument is pushed to a new thread and

583-- the context is computation pulls output from the thread.

502-- the context is computation pulls output from the thread.

584−pushOneToCtx :: MonadAsync m => CtxType -> AsyncT m a -> m (Context m a)

503+pushOneToCtx :: MonadAsync m => CtxType -> Stream m a -> m (Context m a)

585pushOneToCtx ctype m = do

504pushOneToCtx ctype m = do

586 ctx <- liftIO $

505 ctx <- liftIO $

587 case ctype of

506 case ctype of

@@ -610,7 +529,7 @@ pushOneToCtx ctype m = do

610-- fully we will may have a thread blocked forever and if executed more than

529-- fully we will may have a thread blocked forever and if executed more than

611-- once a ContextUsedAfterEOF exception will be raised.

530-- once a ContextUsedAfterEOF exception will be raised.

612

531

613−async :: MonadAsync m => AsyncT m a -> m (AsyncT m a)

532+async :: MonadAsync m => Stream m a -> m (Stream m a)

614async m = do

533async m = do

615 ctx <- pushOneToCtx (CtxType Disjunction LIFO) m

534 ctx <- pushOneToCtx (CtxType Disjunction LIFO) m

616 return $ pullFromCtx ctx

535 return $ pullFromCtx ctx

@@ -619,7 +538,7 @@ async m = do

619-- Num

538-- Num

620------------------------------------------------------------------------------

539------------------------------------------------------------------------------

621

540

622−instance (Monad m, Num a) => Num (AsyncT m a) where

541+instance (Monad m, Num a) => Num (Stream m a) where

623 fromInteger n = pure (fromInteger n)

542 fromInteger n = pure (fromInteger n)

624

543

625 negate = fmap negate

544 negate = fmap negate

@@ -630,14 +549,14 @@ instance (Monad m, Num a) => Num (AsyncT m a) where

630 (*) = liftA2 (*)

549 (*) = liftA2 (*)

631 (-) = liftA2 (-)

550 (-) = liftA2 (-)

632

551

633−instance (Monad m, Fractional a) => Fractional (AsyncT m a) where

552+instance (Monad m, Fractional a) => Fractional (Stream m a) where

634 fromRational n = pure (fromRational n)

553 fromRational n = pure (fromRational n)

635

554

636 recip = fmap recip

555 recip = fmap recip

637

556

638 (/) = liftA2 (/)

557 (/) = liftA2 (/)

639

558

640−instance (Monad m, Floating a) => Floating (AsyncT m a) where

559+instance (Monad m, Floating a) => Floating (Stream m a) where

641 pi = pure pi

560 pi = pure pi

642

561

643 exp = fmap exp

562 exp = fmap exp

@@ -663,39 +582,39 @@ instance (Monad m, Floating a) => Floating (AsyncT m a) where

663-- AsyncT transformer

582-- AsyncT transformer

664-------------------------------------------------------------------------------

583-------------------------------------------------------------------------------

665

584

666−instance MonadTrans AsyncT where

585+instance MonadTrans Stream where

667− lift mx = AsyncT $ \_ _ yld -> mx >>= (\a -> (yld a Nothing))

586+ lift mx = Stream $ \_ _ yld -> mx >>= (\a -> (yld a Nothing))

668

587

669−instance (MonadBase b m, Monad m) => MonadBase b (AsyncT m) where

588+instance (MonadBase b m, Monad m) => MonadBase b (Stream m) where

670 liftBase = liftBaseDefault

589 liftBase = liftBaseDefault

671

590

672------------------------------------------------------------------------------

591------------------------------------------------------------------------------

673-- Standard transformer instances

592-- Standard transformer instances

674------------------------------------------------------------------------------

593------------------------------------------------------------------------------

675

594

676−instance MonadIO m => MonadIO (AsyncT m) where

595+instance MonadIO m => MonadIO (Stream m) where

677 liftIO = lift . liftIO

596 liftIO = lift . liftIO

678

597

679−instance MonadThrow m => MonadThrow (AsyncT m) where

598+instance MonadThrow m => MonadThrow (Stream m) where

680 throwM = lift . throwM

599 throwM = lift . throwM

681

600

682-- XXX handle and test cross thread state transfer

601-- XXX handle and test cross thread state transfer

683−instance MonadError e m => MonadError e (AsyncT m) where

602+instance MonadError e m => MonadError e (Stream m) where

684 throwError = lift . throwError

603 throwError = lift . throwError

685− catchError m h = AsyncT $ \ctx stp yld ->

604+ catchError m h = Stream $ \ctx stp yld ->

686− let handle r = r `catchError` \e -> (runAsyncT (h e)) ctx stp yld

605+ let handle r = r `catchError` \e -> (runStream (h e)) ctx stp yld

687 yield a Nothing = yld a Nothing

606 yield a Nothing = yld a Nothing

688 yield a (Just r) = yld a (Just (catchError r h))

607 yield a (Just r) = yld a (Just (catchError r h))

689− in handle $ (runAsyncT m) ctx stp yield

608+ in handle $ (runStream m) ctx stp yield

690

609

691−instance MonadReader r m => MonadReader r (AsyncT m) where

610+instance MonadReader r m => MonadReader r (Stream m) where

692 ask = lift ask

611 ask = lift ask

693− local f m = AsyncT $ \ctx stp yld ->

612+ local f m = Stream $ \ctx stp yld ->

694 let yield a Nothing = local f $ yld a Nothing

613 let yield a Nothing = local f $ yld a Nothing

695 yield a (Just r) = local f $ yld a (Just (local f r))

614 yield a (Just r) = local f $ yld a (Just (local f r))

696− in (runAsyncT m) ctx (local f stp) yield

615+ in (runStream m) ctx (local f stp) yield

697

616

698−instance MonadState s m => MonadState s (AsyncT m) where

617+instance MonadState s m => MonadState s (Stream m) where

699 get = lift get

618 get = lift get

700 put x = lift (put x)

619 put x = lift (put x)

701 state k = lift (state k)

620 state k = lift (state k)

@@ -706,8 +625,8 @@ instance MonadState s m => MonadState s (AsyncT m) where

706

625

707-- | Run an 'AsyncT' computation, wait for it to finish and discard the

626-- | Run an 'AsyncT' computation, wait for it to finish and discard the

708-- results.

627-- results.

709−runAsyncly :: MonadAsync m => AsyncT m a -> m ()

628+runStreaming :: Monad m => Stream m a -> m ()

710−runAsyncly m = (runAsyncT m) Nothing stop yield

629+runStreaming m = (runStream m) Nothing stop yield

711

630

712 where

631 where

713

632

@@ -715,13 +634,12 @@ runAsyncly m = (runAsyncT m) Nothing stop yield

715

634

716 {-# INLINE yield #-}

635 {-# INLINE yield #-}

717 yield _ Nothing = stop

636 yield _ Nothing = stop

718− yield _ (Just x) = runAsyncly x

637+ yield _ (Just x) = runStreaming x

719−

720

638

721-- | Collect the results of an 'AsyncT' stream into a list.

639-- | Collect the results of an 'AsyncT' stream into a list.

722{-# INLINABLE toList #-}

640{-# INLINABLE toList #-}

723−toList :: MonadAsync m => AsyncT m a -> m [a]

641+toList :: Monad m => Stream m a -> m [a]

724−toList m = (runAsyncT m) Nothing stop yield

642+toList m = (runStream m) Nothing stop yield

725

643

726 where

644 where

727

645

@@ -759,5 +677,5 @@ forEachWith f xs g = foldr (f . g) mempty xs

759------------------------------------------------------------------------------

677------------------------------------------------------------------------------

760

678

761-- | Convert a callback into an 'AsyncT' stream.

679-- | Convert a callback into an 'AsyncT' stream.

762−fromCallback :: (forall r. (a -> m r) -> m r) -> AsyncT m a

680+fromCallback :: (forall r. (a -> m r) -> m r) -> Stream m a

763−fromCallback k = AsyncT $ \_ _ yld -> k (\a -> yld a Nothing)

681+fromCallback k = Stream $ \_ _ yld -> k (\a -> yld a Nothing)

src/Asyncly/Prelude.hs+396−64

@@ -1,5 +1,9 @@

1{-# LANGUAGE FlexibleContexts #-}

2+{-# LANGUAGE FlexibleInstances #-}

2{-# LANGUAGE GeneralizedNewtypeDeriving#-}

3{-# LANGUAGE GeneralizedNewtypeDeriving#-}

4+{-# LANGUAGE MultiParamTypeClasses #-}

5+{-# LANGUAGE RankNTypes #-}

6+{-# LANGUAGE StandaloneDeriving #-}

3{-# LANGUAGE UndecidableInstances #-} -- XXX

7{-# LANGUAGE UndecidableInstances #-} -- XXX

5-- |

9-- |

@@ -13,32 +17,310 @@

13--

17--

14--

18--

15module Asyncly.Prelude

19module Asyncly.Prelude

16− (

20+ ( Streaming (..)

17− -- * Construction

21+ , StreamT

18− unfoldr

22+ , InterleavedT

23+ , AsyncT

24+ , ParallelT

25+ , ZipStream

26+ , ZipAsync

27+ , serially

28+ , toListSerial

29+ , interleaved

30+ , toListInterleaved

31+ , asyncly

32+ , toListAsync

33+ , parallely

34+ , toListParallel

35+ )

36+where

37+

38+import Control.Applicative (Alternative (..), liftA2)

39+import Control.Arrow (second)

40+import Data.Semigroup (Semigroup(..), cycle1)

41+import Control.Monad (MonadPlus(..))

42+import Control.Monad.Base (MonadBase (..))

43+import Control.Monad.Catch (MonadThrow)

44+import Control.Monad.Error.Class (MonadError(..))

45+import Control.Monad.IO.Class (MonadIO(..))

46+import Control.Monad.Reader.Class (MonadReader(..))

47+import Control.Monad.State.Class (MonadState(..))

48+import Control.Monad.Trans.Class (MonadTrans)

49+import Prelude hiding (drop, take, zipWith)

50+import Asyncly.AsyncT

51+

52+------------------------------------------------------------------------------

53+-- Streams with different styles of product composition

54+------------------------------------------------------------------------------

55+

56+newtype StreamT m a = StreamT {getStreamT :: Stream m a}

57+ deriving (Functor, Applicative, Monad, Semigroup, Monoid, Num, Fractional,

58+ Floating, MonadTrans, MonadIO, MonadThrow)

59+

60+deriving instance MonadAsync m => Alternative (StreamT m)

61+deriving instance MonadAsync m => MonadPlus (StreamT m)

62+deriving instance (MonadBase b m, Monad m) => MonadBase b (StreamT m)

63+deriving instance MonadError e m => MonadError e (StreamT m)

64+deriving instance MonadReader r m => MonadReader r (StreamT m)

65+deriving instance MonadState s m => MonadState s (StreamT m)

66+

67+-- XXX the applicative, functor and num instances may not be correct

68+newtype InterleavedT m a = InterleavedT {getInterleavedT :: Stream m a}

69+ deriving (Functor, Applicative, Semigroup, Monoid, Num, Fractional,

70+ Floating, MonadTrans, MonadIO, MonadThrow)

71+

72+deriving instance MonadAsync m => Alternative (InterleavedT m)

73+deriving instance MonadAsync m => MonadPlus (InterleavedT m)

74+deriving instance (MonadBase b m, Monad m) => MonadBase b (InterleavedT m)

75+deriving instance MonadError e m => MonadError e (InterleavedT m)

76+deriving instance MonadReader r m => MonadReader r (InterleavedT m)

77+deriving instance MonadState s m => MonadState s (InterleavedT m)

78+

79+newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}

80+ deriving (Functor, Applicative, Semigroup, Monoid, Num, Fractional,

81+ Floating, MonadTrans)

82+

83+deriving instance MonadAsync m => MonadIO (AsyncT m)

84+deriving instance MonadAsync m => MonadThrow (AsyncT m)

85+deriving instance (MonadBase b m, MonadAsync m) => MonadBase b (AsyncT m)

86+deriving instance MonadAsync m => Alternative (AsyncT m)

87+deriving instance MonadAsync m => MonadPlus (AsyncT m)

88+deriving instance (MonadError e m, MonadAsync m) => MonadError e (AsyncT m)

89+deriving instance (MonadReader r m, MonadAsync m) => MonadReader r (AsyncT m)

90+deriving instance (MonadState s m, MonadAsync m) => MonadState s (AsyncT m)

91+

92+newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}

93+ deriving (Functor, Applicative, Semigroup, Monoid, Num, Fractional,

94+ Floating, MonadTrans)

95+

96+deriving instance MonadAsync m => MonadIO (ParallelT m)

97+deriving instance MonadAsync m => MonadThrow (ParallelT m)

98+deriving instance (MonadBase b m, MonadAsync m) => MonadBase b (ParallelT m)

99+deriving instance MonadAsync m => Alternative (ParallelT m)

100+deriving instance MonadAsync m => MonadPlus (ParallelT m)

101+deriving instance (MonadError e m, MonadAsync m) => MonadError e (ParallelT m)

102+deriving instance (MonadReader r m, MonadAsync m) => MonadReader r (ParallelT m)

103+deriving instance (MonadState s m, MonadAsync m) => MonadState s (ParallelT m)

104+

105+infixr 5 <=>

106+

107+class Streaming t where

108+ -- sum style operations

109+

110+ -- | Sequential interleaved composition, similar to '<>' except that it

111+ -- fairly interleaves the two 'AsyncT' streams, yielding an element from

112+ -- each stream alternately. Should be used only on finite streams.

113+ (<=>) :: t m a -> t m a -> t m a

114+

115+ -- | Parallel composition similar to '<|>' except that it is left-biased

116+ -- instead of being fair. Action on the left is likely to be given a

117+ -- chance to execute before the action on the right. If the left action

118+ -- keeps yielding results without blocking it continues running until it

119+ -- finishes and only then the right action runs. Unlike '<|>' it can be

120+ -- used on infinite streams.

121+ (<|) :: MonadAsync m => t m a -> t m a -> t m a

122+

123+ -- Conversions

124+ fromStream :: StreamT m a -> t m a

125+ fromInterleaved :: InterleavedT m a -> t m a

126+ fromAsync :: AsyncT m a -> t m a

127+ fromParallel :: ParallelT m a -> t m a

128+ fromZipStream :: ZipStream m a -> t m a

129+ fromZipAsync :: ZipAsync m a -> t m a

130

20 -- * Elimination

131 -- * Elimination

21− , uncons

132+

133+ -- | Decompose a stream into its head and tail. If the stream is empty,

134+ -- returns 'Nothing'. If the stream is non-empty, returns 'Just (a, ma)',

135+ -- where 'a' is the head of the stream and 'ma' its tail.

136+ uncons :: MonadAsync m => t m a -> m (Maybe (a, t m a))

137+

138+ -- * Construction

139+ unfoldr :: MonadAsync m => (b -> m (Maybe (a, b))) -> b -> t m a

140

23 -- * Special folds

141 -- * Special folds

24− , take

142+ take :: MonadAsync m => Int -> t m a -> t m a

25− , drop

143+ drop :: MonadAsync m => Int -> t m a -> t m a

144

27 -- * Zipping

145 -- * Zipping

28− , ZipSerial (..)

146+ zipWith :: Monad m => (a -> b -> c) -> t m a -> t m b -> t m c

29− , zipWith

147+ zipWithM :: Monad m =>

30− , zipWithM

148+ (a -> b -> t m c) -> t m a -> t m b -> t m c

149+

150+ zipAsyncWithM :: MonadAsync m

151+ => (a -> b -> t m c) -> t m a -> t m b -> t m c

152+ zipAsyncWith :: MonadAsync m

153+ => (a -> b -> c) -> t m a -> t m b -> t m c

154+

155+instance Streaming StreamT where

156+ {-# INLINE (<=>) #-}

157+ m1 <=> m2 = StreamT $ interleave (getStreamT m1) (getStreamT m2)

158+

159+ {-# INLINE (<|) #-}

160+ m1 <| m2 = StreamT $ parLeft (getStreamT m1) (getStreamT m2)

161+

162+ fromStream = id

163+ fromInterleaved = StreamT . getInterleavedT

164+ fromAsync = StreamT . getAsyncT

165+ fromParallel = StreamT . getParallelT

166+ fromZipStream = StreamT . getZipStream

167+ fromZipAsync = StreamT . getZipAsync

168+

169+ uncons (StreamT s) = fmap (fmap (second StreamT)) (next s)

170+ unfoldr step seed = StreamT $ unfold step seed

171+ take n (StreamT s) = StreamT $ taking n s

172+ drop n (StreamT s) = StreamT $ dropping n s

173+ zipWith f (StreamT s1) (StreamT s2) = StreamT $ zippingWith f s1 s2

174+ zipWithM f (StreamT s1) (StreamT s2) = StreamT $ zippingWithM g s1 s2

175+ where g a b = getStreamT (f a b)

176+ zipAsyncWith f (StreamT s1) (StreamT s2) = StreamT $ zippingAsyncWith f s1 s2

177+ zipAsyncWithM f (StreamT s1) (StreamT s2) = StreamT $ zippingAsyncWithM g s1 s2

178+ where g a b = getStreamT (f a b)

179+

180+instance Streaming InterleavedT where

181+ {-# INLINE (<=>) #-}

182+ m1 <=> m2 = InterleavedT $ interleave (getInterleavedT m1) (getInterleavedT m2)

183+

184+ {-# INLINE (<|) #-}

185+ m1 <| m2 = InterleavedT $ parLeft (getInterleavedT m1) (getInterleavedT m2)

186+

187+ fromStream = InterleavedT . getStreamT

188+ fromInterleaved = id

189+ fromAsync = InterleavedT . getAsyncT

190+ fromParallel = InterleavedT . getParallelT

191+ fromZipStream = InterleavedT . getZipStream

192+ fromZipAsync = InterleavedT . getZipAsync

193+

194+ uncons (InterleavedT s) = fmap (fmap (second InterleavedT)) (next s)

195+ unfoldr step seed = InterleavedT $ unfold step seed

196+ take n (InterleavedT s) = InterleavedT $ taking n s

197+ drop n (InterleavedT s) = InterleavedT $ dropping n s

198+ zipWith f (InterleavedT s1) (InterleavedT s2) = InterleavedT $ zippingWith f s1 s2

199+ zipWithM f (InterleavedT s1) (InterleavedT s2) = InterleavedT $ zippingWithM g s1 s2

200+ where g a b = getInterleavedT (f a b)

201+ zipAsyncWith f (InterleavedT s1) (InterleavedT s2) = InterleavedT $ zippingAsyncWith f s1 s2

202+ zipAsyncWithM f (InterleavedT s1) (InterleavedT s2) = InterleavedT $ zippingAsyncWithM g s1 s2

203+ where g a b = getInterleavedT (f a b)

204+

205+instance Streaming AsyncT where

206+ {-# INLINE (<=>) #-}

207+ m1 <=> m2 = AsyncT $ interleave (getAsyncT m1) (getAsyncT m2)

208+

209+ {-# INLINE (<|) #-}

210+ m1 <| m2 = AsyncT $ parLeft (getAsyncT m1) (getAsyncT m2)

211+

212+ fromStream = AsyncT . getStreamT

213+ fromInterleaved = AsyncT . getInterleavedT

214+ fromAsync = id

215+ fromParallel = AsyncT . getParallelT

216+ fromZipStream = AsyncT . getZipStream

217+ fromZipAsync = AsyncT . getZipAsync

218+

219+ uncons (AsyncT s) = fmap (fmap (second AsyncT)) (next s)

220+ unfoldr step seed = AsyncT $ unfold step seed

221+ take n (AsyncT s) = AsyncT $ taking n s

222+ drop n (AsyncT s) = AsyncT $ dropping n s

223+ zipWith f (AsyncT s1) (AsyncT s2) = AsyncT $ zippingWith f s1 s2

224+ zipWithM f (AsyncT s1) (AsyncT s2) = AsyncT $ zippingWithM g s1 s2

225+ where g a b = getAsyncT (f a b)

226+ zipAsyncWith f (AsyncT s1) (AsyncT s2) = AsyncT $ zippingAsyncWith f s1 s2

227+ zipAsyncWithM f (AsyncT s1) (AsyncT s2) = AsyncT $ zippingAsyncWithM g s1 s2

228+ where g a b = getAsyncT (f a b)

229+

230+instance Streaming ParallelT where

231+ {-# INLINE (<=>) #-}

232+ m1 <=> m2 = ParallelT $ interleave (getParallelT m1) (getParallelT m2)

233+

234+ {-# INLINE (<|) #-}

235+ m1 <| m2 = ParallelT $ parLeft (getParallelT m1) (getParallelT m2)

236+

237+ fromStream = ParallelT . getStreamT

238+ fromInterleaved = ParallelT . getInterleavedT

239+ fromAsync = ParallelT . getAsyncT

240+ fromParallel = id

241+ fromZipStream = ParallelT . getZipStream

242+ fromZipAsync = ParallelT . getZipAsync

243+

244+ uncons (ParallelT s) = fmap (fmap (second ParallelT)) (next s)

245+ unfoldr step seed = ParallelT $ unfold step seed

246+ take n (ParallelT s) = ParallelT $ taking n s

247+ drop n (ParallelT s) = ParallelT $ dropping n s

248+ zipWith f (ParallelT s1) (ParallelT s2) = ParallelT $ zippingWith f s1 s2

249+ zipWithM f (ParallelT s1) (ParallelT s2) = ParallelT $ zippingWithM g s1 s2

250+ where g a b = getParallelT (f a b)

251+ zipAsyncWith f (ParallelT s1) (ParallelT s2) = ParallelT $ zippingAsyncWith f s1 s2

252+ zipAsyncWithM f (ParallelT s1) (ParallelT s2) = ParallelT $ zippingAsyncWithM g s1 s2

253+ where g a b = getParallelT (f a b)

254

32− , ZipAsync (..)

255+------------------------------------------------------------------------------

33− , zipAsyncWith

256+-- Alternative ways to bind

34− , zipAsyncWithM

257+------------------------------------------------------------------------------

35− )

36−where

258

38−import Control.Applicative (Alternative (..), liftA2)

259+instance Monad m => Monad (InterleavedT m) where

39−import Data.Semigroup (Semigroup(..), cycle1)

260+ -- | Execute a monadic action for each element in the stream, in a fairly

40−import Prelude hiding (drop, take, zipWith)

261+ -- interleaved manner i.e. iterations yield alternately.

41−import Asyncly.AsyncT

262+ (InterleavedT (Stream m)) >>= f = InterleavedT $ Stream $ \_ stp yld ->

263+ let run x = (runStream x) Nothing stp yld

264+ yield a Nothing = run $ getInterleavedT (f a)

265+ yield a (Just r) = run $ getInterleavedT (f a)

266+ `interleave`

267+ getInterleavedT ((InterleavedT r) >>= f)

268+ in m Nothing stp yield

269+

270+{-# INLINE parbind #-}

271+parbind

272+ :: (forall c. Stream m c -> Stream m c -> Stream m c)

273+ -> Stream m a

274+ -> (a -> Stream m b)

275+ -> Stream m b

276+parbind k m f = go m

277+ where

278+ go (Stream g) =

279+ Stream $ \ctx stp yld ->

280+ let run x = (runStream x) ctx stp yld

281+ yield a Nothing = run $ f a

282+ yield a (Just r) = run $ f a `k` (go r)

283+ in g Nothing stp yield

284+

285+instance MonadAsync m => Monad (AsyncT m) where

286+ -- | Execute a monadic action for each element in the stream, running

287+ -- iterations in parallel, but giving preference to iterations started

288+ -- earlier.

289+ (AsyncT m) >>= f = AsyncT $ parbind par m g

290+ where g x = getAsyncT (f x)

291+ par = parallel (CtxType Conjunction LIFO)

292+

293+instance MonadAsync m => Monad (ParallelT m) where

294+ -- | Execute a monadic action for each element in the stream, running

295+ -- iterations in a fairly parallel manner, i.e. all iterations are equally

296+ -- likely to run.

297+ (ParallelT m) >>= f = ParallelT $ parbind par m g

298+ where g x = getParallelT (f x)

299+ par = parallel (CtxType Conjunction FIFO)

300+

301+serially :: Monad m => StreamT m a -> m ()

302+serially = runStreaming . getStreamT

303+

304+toListSerial :: Monad m => StreamT m a -> m [a]

305+toListSerial = toList . getStreamT

306+

307+interleaved :: Monad m => InterleavedT m a -> m ()

308+interleaved = runStreaming . getInterleavedT

309+

310+toListInterleaved :: Monad m => InterleavedT m a -> m [a]

311+toListInterleaved = toList . getInterleavedT

312+

313+asyncly :: Monad m => AsyncT m a -> m ()

314+asyncly = runStreaming . getAsyncT

315+

316+toListAsync :: Monad m => AsyncT m a -> m [a]

317+toListAsync = toList . getAsyncT

318+

319+parallely :: Monad m => ParallelT m a -> m ()

320+parallely = runStreaming . getParallelT

321+

322+toListParallel :: Monad m => ParallelT m a -> m [a]

323+toListParallel = toList . getParallelT

324

43------------------------------------------------------------------------------

325------------------------------------------------------------------------------

44-- Construct and deconstruct

326-- Construct and deconstruct

@@ -47,8 +329,8 @@ import Asyncly.AsyncT

47-- | Decompose a stream into its head and tail. If the stream is empty, returns

329-- | Decompose a stream into its head and tail. If the stream is empty, returns

48-- 'Nothing'. If the stream is non-empty, returns 'Just (a, ma)', where 'a' is

330-- 'Nothing'. If the stream is non-empty, returns 'Just (a, ma)', where 'a' is

49-- the head of the stream and 'ma' its tail.

331-- the head of the stream and 'ma' its tail.

50−uncons :: MonadAsync m => AsyncT m a -> m (Maybe (a, AsyncT m a))

332+next :: MonadAsync m => Stream m a -> m (Maybe (a, Stream m a))

51−uncons m = (runAsyncT m) Nothing stop yield

333+next m = (runStream m) Nothing stop yield

334

53 where

335 where

336

@@ -59,10 +341,10 @@ uncons m = (runAsyncT m) Nothing stop yield

59 yield a (Just x) = return (Just (a, x))

341 yield a (Just x) = return (Just (a, x))

342

61-- | Build a Stream by unfolding steps starting from a seed.

343-- | Build a Stream by unfolding steps starting from a seed.

62−unfoldr :: MonadAsync m => (b -> m (Maybe (a, b))) -> b -> AsyncT m a

344+unfold :: MonadAsync m => (b -> m (Maybe (a, b))) -> b -> Stream m a

63−unfoldr step = go

345+unfold step = go

64 where

346 where

65− go s = AsyncT $ \_ stp yld -> do

347+ go s = Stream $ \_ stp yld -> do

66 mayb <- step s

348 mayb <- step s

67 case mayb of

349 case mayb of

68 Nothing -> stp

350 Nothing -> stp

@@ -73,64 +355,89 @@ unfoldr step = go

73------------------------------------------------------------------------------

355------------------------------------------------------------------------------

356

75-- | Take first 'n' elements from the stream and discard the rest.

357-- | Take first 'n' elements from the stream and discard the rest.

76−take :: MonadAsync m => Int -> AsyncT m a -> AsyncT m a

358+taking :: MonadAsync m => Int -> Stream m a -> Stream m a

77−take n m = AsyncT $ \ctx stp yld -> do

359+taking n m = Stream $ \ctx stp yld -> do

78 let yield a Nothing = yld a Nothing

360 let yield a Nothing = yld a Nothing

79− yield a (Just x) = yld a (Just (take (n - 1) x))

361+ yield a (Just x) = yld a (Just (taking (n - 1) x))

80 if (n <= 0)

362 if (n <= 0)

81 then stp

363 then stp

82− else (runAsyncT m) ctx stp yield

364+ else (runStream m) ctx stp yield

365

84-- | Discard first 'n' elements from the stream and take the rest.

366-- | Discard first 'n' elements from the stream and take the rest.

85−drop :: MonadAsync m => Int -> AsyncT m a -> AsyncT m a

367+dropping :: MonadAsync m => Int -> Stream m a -> Stream m a

86−drop n m = AsyncT $ \ctx stp yld -> do

368+dropping n m = Stream $ \ctx stp yld -> do

87 let yield _ Nothing = stp

369 let yield _ Nothing = stp

88− yield _ (Just x) = (runAsyncT $ drop (n - 1) x) ctx stp yld

370+ yield _ (Just x) = (runStream $ dropping (n - 1) x) ctx stp yld

89 if (n <= 0)

371 if (n <= 0)

90− then (runAsyncT m) ctx stp yld

372+ then (runStream m) ctx stp yld

91− else (runAsyncT m) ctx stp yield

373+ else (runStream m) ctx stp yield

374

93------------------------------------------------------------------------------

375------------------------------------------------------------------------------

94-- Serially Zipping Streams

376-- Serially Zipping Streams

95------------------------------------------------------------------------------

377------------------------------------------------------------------------------

378

97-- | Zip two AsyncT streams serially using a monadic zipping function.

379-- | Zip two AsyncT streams serially using a monadic zipping function.

98−zipWithM :: Monad m =>

380+zippingWithM :: Monad m =>

99− (a -> b -> AsyncT m c) -> AsyncT m a -> AsyncT m b -> AsyncT m c

381+ (a -> b -> Stream m c) -> Stream m a -> Stream m b -> Stream m c

100−zipWithM f m1 m2 = AsyncT $ \_ stp yld -> do

382+zippingWithM f m1 m2 = Stream $ \_ stp yld -> do

101 let merge a ra =

383 let merge a ra =

102− let yield2 b Nothing = (runAsyncT (f a b)) Nothing stp yld

384+ let yield2 b Nothing = (runStream (f a b)) Nothing stp yld

103 yield2 b (Just rb) =

385 yield2 b (Just rb) =

104− (runAsyncT ((f a b) <> (zipWithM f ra rb))) Nothing stp yld

386+ (runStream ((f a b) <> (zippingWithM f ra rb))) Nothing stp yld

105− in (runAsyncT m2) Nothing stp yield2

387+ in (runStream m2) Nothing stp yield2

106 let yield1 a Nothing = merge a mempty

388 let yield1 a Nothing = merge a mempty

107 yield1 a (Just ra) = merge a ra

389 yield1 a (Just ra) = merge a ra

108− (runAsyncT m1) Nothing stp yield1

390+ (runStream m1) Nothing stp yield1

109

391

110-- | Zip two AsyncT streams serially using a pure zipping function.

392-- | Zip two AsyncT streams serially using a pure zipping function.

111−zipWith :: Monad m => (a -> b -> c) -> AsyncT m a -> AsyncT m b -> AsyncT m c

393+zippingWith :: Monad m => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c

112−zipWith f m1 m2 = AsyncT $ \_ stp yld -> do

394+zippingWith f m1 m2 = Stream $ \_ stp yld -> do

113 let merge a ra =

395 let merge a ra =

114 let yield2 b Nothing = yld (f a b) Nothing

396 let yield2 b Nothing = yld (f a b) Nothing

115− yield2 b (Just rb) = yld (f a b) (Just (zipWith f ra rb))

397+ yield2 b (Just rb) = yld (f a b) (Just (zippingWith f ra rb))

116− in (runAsyncT m2) Nothing stp yield2

398+ in (runStream m2) Nothing stp yield2

117 let yield1 a Nothing = merge a mempty

399 let yield1 a Nothing = merge a mempty

118 yield1 a (Just ra) = merge a ra

400 yield1 a (Just ra) = merge a ra

119− (runAsyncT m1) Nothing stp yield1

401+ (runStream m1) Nothing stp yield1

120

402

121-- | Wrapper around AsyncT type with a serial zipping Applicative instance.

403-- | Wrapper around AsyncT type with a serial zipping Applicative instance.

122-- Note that the binary function interleave (\<=>) is a special case of

404-- Note that the binary function interleave (\<=>) is a special case of

123−-- ZipSerial Applicative.

405+-- ZipStream Applicative.

124--

406--

125−-- > f <$> ZipSerial xs1 <*> ... <*> ZipSerial xsN

407+-- > f <$> ZipStream xs1 <*> ... <*> ZipStream xsN

126−newtype ZipSerial m a = ZipSerial {getZipSerial :: AsyncT m a}

408+newtype ZipStream m a = ZipStream {getZipStream :: Stream m a}

127 deriving (Functor)

409 deriving (Functor)

128

410

129−instance Monad m => Applicative (ZipSerial m) where

411+instance Monad m => Applicative (ZipStream m) where

130− pure a = ZipSerial $ cycle1 (pure a)

412+ pure a = ZipStream $ cycle1 (pure a)

131− (ZipSerial xs) <*> (ZipSerial ys) = ZipSerial (zipWith id xs ys)

413+ (ZipStream xs) <*> (ZipStream ys) = ZipStream (zippingWith id xs ys)

132−

414+

415+instance Streaming ZipStream where

416+ {-# INLINE (<=>) #-}

417+ m1 <=> m2 = ZipStream $ interleave (getZipStream m1) (getZipStream m2)

418+

419+ {-# INLINE (<|) #-}

420+ m1 <| m2 = ZipStream $ parLeft (getZipStream m1) (getZipStream m2)

421+

422+ fromStream = ZipStream . getStreamT

423+ fromInterleaved = ZipStream . getInterleavedT

424+ fromAsync = ZipStream . getAsyncT

425+ fromParallel = ZipStream . getParallelT

426+ fromZipStream = ZipStream . getZipStream

427+ fromZipAsync = ZipStream . getZipAsync

428+

429+ uncons (ZipStream s) = fmap (fmap (second ZipStream)) (next s)

430+ unfoldr step seed = ZipStream $ unfold step seed

431+ take n (ZipStream s) = ZipStream $ taking n s

432+ drop n (ZipStream s) = ZipStream $ dropping n s

433+ zipWith f (ZipStream s1) (ZipStream s2) = ZipStream $ zippingWith f s1 s2

434+ zipWithM f (ZipStream s1) (ZipStream s2) = ZipStream $ zippingWithM g s1 s2

435+ where g a b = getZipStream (f a b)

436+ zipAsyncWith f (ZipStream s1) (ZipStream s2) = ZipStream $ zippingAsyncWith f s1 s2

437+ zipAsyncWithM f (ZipStream s1) (ZipStream s2) = ZipStream $ zippingAsyncWithM g s1 s2

438+ where g a b = getZipStream (f a b)

439+

133−instance (Monad m, Num a) => Num (ZipSerial m a) where

440+instance (Monad m, Num a) => Num (ZipStream m a) where

134 fromInteger n = pure (fromInteger n)

441 fromInteger n = pure (fromInteger n)

135

442

136 negate = fmap negate

443 negate = fmap negate

@@ -141,14 +448,14 @@ instance (Monad m, Num a) => Num (ZipSerial m a) where

141 (*) = liftA2 (*)

448 (*) = liftA2 (*)

142 (-) = liftA2 (-)

449 (-) = liftA2 (-)

143

450

144−instance (Monad m, Fractional a) => Fractional (ZipSerial m a) where

451+instance (Monad m, Fractional a) => Fractional (ZipStream m a) where

145 fromRational n = pure (fromRational n)

452 fromRational n = pure (fromRational n)

146

453

147 recip = fmap recip

454 recip = fmap recip

148

455

149 (/) = liftA2 (/)

456 (/) = liftA2 (/)

150

457

151−instance (Monad m, Floating a) => Floating (ZipSerial m a) where

458+instance (Monad m, Floating a) => Floating (ZipStream m a) where

152 pi = pure pi

459 pi = pure pi

153

460

154 exp = fmap exp

461 exp = fmap exp

@@ -176,33 +483,58 @@ instance (Monad m, Floating a) => Floating (ZipSerial m a) where

176

483

177-- | Zip two AsyncT streams asyncly (i.e. both the streams are generated

484-- | Zip two AsyncT streams asyncly (i.e. both the streams are generated

178-- concurrently) using a monadic zipping function.

485-- concurrently) using a monadic zipping function.

179−zipAsyncWithM :: MonadAsync m

486+zippingAsyncWithM :: MonadAsync m

180− => (a -> b -> AsyncT m c) -> AsyncT m a -> AsyncT m b -> AsyncT m c

487+ => (a -> b -> Stream m c) -> Stream m a -> Stream m b -> Stream m c

181−zipAsyncWithM f m1 m2 = AsyncT $ \_ stp yld -> do

488+zippingAsyncWithM f m1 m2 = Stream $ \_ stp yld -> do

182 ma <- async m1

489 ma <- async m1

183 mb <- async m2

490 mb <- async m2

184− (runAsyncT (zipWithM f ma mb)) Nothing stp yld

491+ (runStream (zippingWithM f ma mb)) Nothing stp yld

185

492

186-- | Zip two AsyncT streams asyncly (i.e. both the streams are generated

493-- | Zip two AsyncT streams asyncly (i.e. both the streams are generated

187-- concurrently) using a pure zipping function.

494-- concurrently) using a pure zipping function.

188−zipAsyncWith :: MonadAsync m

495+zippingAsyncWith :: MonadAsync m

189− => (a -> b -> c) -> AsyncT m a -> AsyncT m b -> AsyncT m c

496+ => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c

190−zipAsyncWith f m1 m2 = AsyncT $ \_ stp yld -> do

497+zippingAsyncWith f m1 m2 = Stream $ \_ stp yld -> do

191 ma <- async m1

498 ma <- async m1

192 mb <- async m2

499 mb <- async m2

193− (runAsyncT (zipWith f ma mb)) Nothing stp yld

500+ (runStream (zippingWith f ma mb)) Nothing stp yld

194

501

195-- | Wrapper around AsyncT type with a parallel zipping Applicative instance.

502-- | Wrapper around AsyncT type with a parallel zipping Applicative instance.

196-- Note that the binary operator (\<|>) from the Alternative instance of AsyncT

503-- Note that the binary operator (\<|>) from the Alternative instance of AsyncT

197-- is a special case of ZipAsync Applicative.

504-- is a special case of ZipAsync Applicative.

198--

505--

199-- > f <$> ZipAsync xs1 <*> ... <*> ZipAsync xsN

506-- > f <$> ZipAsync xs1 <*> ... <*> ZipAsync xsN

200−newtype ZipAsync m a = ZipAsync {getZipAsync :: AsyncT m a}

507+newtype ZipAsync m a = ZipAsync {getZipAsync :: Stream m a}

201 deriving (Functor)

508 deriving (Functor)

202

509

203instance MonadAsync m => Applicative (ZipAsync m) where

510instance MonadAsync m => Applicative (ZipAsync m) where

204 pure a = ZipAsync $ cycle1 (pure a)

511 pure a = ZipAsync $ cycle1 (pure a)

205− (ZipAsync xs) <*> (ZipAsync ys) = ZipAsync (zipAsyncWith id xs ys)

512+ (ZipAsync xs) <*> (ZipAsync ys) = ZipAsync (zippingAsyncWith id xs ys)

513+

514+instance Streaming ZipAsync where

515+ {-# INLINE (<=>) #-}

516+ m1 <=> m2 = ZipAsync $ interleave (getZipAsync m1) (getZipAsync m2)

517+

518+ {-# INLINE (<|) #-}

519+ m1 <| m2 = ZipAsync $ parLeft (getZipAsync m1) (getZipAsync m2)

520+

521+ fromStream = ZipAsync . getStreamT

522+ fromInterleaved = ZipAsync . getInterleavedT

523+ fromAsync = ZipAsync . getAsyncT

524+ fromParallel = ZipAsync . getParallelT

525+ fromZipStream = ZipAsync . getZipStream

526+ fromZipAsync = ZipAsync . getZipAsync

527+

528+ uncons (ZipAsync s) = fmap (fmap (second ZipAsync)) (next s)

529+ unfoldr step seed = ZipAsync $ unfold step seed

530+ take n (ZipAsync s) = ZipAsync $ taking n s

531+ drop n (ZipAsync s) = ZipAsync $ dropping n s

532+ zipWith f (ZipAsync s1) (ZipAsync s2) = ZipAsync $ zippingWith f s1 s2

533+ zipWithM f (ZipAsync s1) (ZipAsync s2) = ZipAsync $ zippingWithM g s1 s2

534+ where g a b = getZipAsync (f a b)

535+ zipAsyncWith f (ZipAsync s1) (ZipAsync s2) = ZipAsync $ zippingAsyncWith f s1 s2

536+ zipAsyncWithM f (ZipAsync s1) (ZipAsync s2) = ZipAsync $ zippingAsyncWithM g s1 s2

537+ where g a b = getZipAsync (f a b)

206

538

207instance (MonadAsync m, Num a) => Num (ZipAsync m a) where

539instance (MonadAsync m, Num a) => Num (ZipAsync m a) where

208 fromInteger n = pure (fromInteger n)

540 fromInteger n = pure (fromInteger n)

test/Main.hs+130−104

@@ -14,20 +14,20 @@ import qualified Asyncly.Prelude as A

14main :: IO ()

15main = hspec $ do

16 describe "Runners" $ do

17− it "simple runAsyncly" $

17+ it "simple serially" $

18− runAsyncly (return (0 :: Int)) `shouldReturn` ()

18+ serially (return (0 :: Int)) `shouldReturn` ()

19− it "simple runAsyncly with IO" $

19+ it "simple serially with IO" $

20− runAsyncly (liftIO $ putStrLn "hello") `shouldReturn` ()

20+ serially (liftIO $ putStrLn "hello") `shouldReturn` ()

21 it "Captures a return value using toList" $

22− toList (return 0) `shouldReturn` ([0] :: [Int])

22+ toListSerial (return 0) `shouldReturn` ([0] :: [Int])

24 describe "Empty" $ do

25 it "Monoid - mempty" $

26− (toList mempty) `shouldReturn` ([] :: [Int])

26+ (toListSerial mempty) `shouldReturn` ([] :: [Int])

27 it "Alternative - empty" $

28− (toList empty) `shouldReturn` ([] :: [Int])

28+ (toListSerial empty) `shouldReturn` ([] :: [Int])

29 it "MonadPlus - mzero" $

30− (toList mzero) `shouldReturn` ([] :: [Int])

30+ (toListSerial mzero) `shouldReturn` ([] :: [Int])

32 ---------------------------------------------------------------------------

33 -- Functor

@@ -35,12 +35,12 @@ main = hspec $ do

36 describe "Functor (fmap)" $ do

37 it "Simple fmap" $

38− (toList $ fmap (+1) (return 1)) `shouldReturn` ([2] :: [Int])

38+ (toListSerial $ fmap (+1) (return 1)) `shouldReturn` ([2] :: [Int])

39 it "fmap on composed (<>)" $

40− (toList $ fmap (+1) (return 1 <> return 2))

40+ (toListSerial $ fmap (+1) (return 1 <> return 2))

41 `shouldReturn` ([2,3] :: [Int])

42 it "fmap on composed (<|>)" $

43− (toList $ fmap (+1) (return 1 <|> return 2))

43+ (toListSerial $ fmap (+1) (return 1 <|> return 2))

44 `shouldReturn` ([2,3] :: [Int])

46 ---------------------------------------------------------------------------

@@ -49,23 +49,23 @@ main = hspec $ do

50 describe "Applicative" $ do

51 it "Simple apply" $

52− (toList $ (,) <$> (return 1) <*> (return 2))

52+ (toListSerial $ (,) <$> (return 1) <*> (return 2))

53 `shouldReturn` ([(1,2)] :: [(Int, Int)])

55 it "Apply - serial composed first argument" $

56− (toList $ (,) <$> (return 1 <> return 2) <*> (return 3))

56+ (toListSerial $ (,) <$> (return 1 <> return 2) <*> (return 3))

57 `shouldReturn` ([(1,3),(2,3)] :: [(Int, Int)])

59 it "Apply - serial composed second argument" $

60− (toList $ (,) <$> (return 1) <*> (return 2 <> return 3))

60+ (toListSerial $ (,) <$> (return 1) <*> (return 2 <> return 3))

61 `shouldReturn` ([(1,2),(1,3)] :: [(Int, Int)])

63 it "Apply - parallel composed first argument" $

64− (toList $ (,) <$> (return 1 <|> return 2) <*> (return 3))

64+ (toListSerial $ (,) <$> (return 1 <|> return 2) <*> (return 3))

65 `shouldReturn` ([(1,3),(2,3)] :: [(Int, Int)])

67 it "Apply - parallel composed second argument" $

68− (toList $ (,) <$> (return 1) <*> (return 2 <|> return 3))

68+ (toListSerial $ (,) <$> (return 1) <*> (return 2 <|> return 3))

69 `shouldReturn` ([(1,2),(1,3)] :: [(Int, Int)])

71 ---------------------------------------------------------------------------

@@ -73,15 +73,15 @@ main = hspec $ do

73 ---------------------------------------------------------------------------

75 describe "Bind then" thenBind

76− describe "Pure bind serial" $ pureBind (>>=)

76+ describe "Pure bind serial" $ pureBind toListSerial

77− describe "Pure bind serial interleaved" $ pureBind (>->)

77+ describe "Pure bind serial interleaved" $ pureBind toListInterleaved

78− describe "Pure bind parallel DFS" $ pureBind (>>|)

78+ describe "Pure bind parallel DFS" $ pureBind toListAsync

79− describe "Pure bind parallel BFS" $ pureBind (>|>)

79+ describe "Pure bind parallel BFS" $ pureBind toListParallel

81− describe "Bind (>>=) with empty" $ bindEmpty (>>=)

81+ describe "Bind (>>=) with empty" $ bindEmpty toListSerial

82− describe "Bind (>->) with empty" $ bindEmpty (>->)

82+ describe "Bind (>->) with empty" $ bindEmpty toListInterleaved

83− describe "Bind (>|>) with empty" $ bindEmpty (>|>)

83+ describe "Bind (>|>) with empty" $ bindEmpty toListAsync

84− describe "Bind (>>|) with empty" $ bindEmpty (>>|)

84+ describe "Bind (>>|) with empty" $ bindEmpty toListParallel

86 ---------------------------------------------------------------------------

87 -- Monoidal Compositions

@@ -98,8 +98,8 @@ main = hspec $ do

98 -- Monoidal Composition ordering checks

99 ---------------------------------------------------------------------------

100

101− describe "Serial interleaved ordering check (<=>)" $ interleaved (<=>)

101+ describe "Serial interleaved ordering check (<=>)" $ interleaveCheck (<=>)

102− describe "Parallel interleaved ordering check (<|>)" $ interleaved (<|>)

102+ describe "Parallel interleaved ordering check (<|>)" $ interleaveCheck (<|>)

103 describe "Left biased parallel time order check" $ parallelCheck (<|)

104 describe "Fair parallel time order check" $ parallelCheck (<|>)

105

@@ -140,13 +140,13 @@ main = hspec $ do

140 describe "Nested parallel and serial compositions" $ do

141 it "Nest <|>, <>, <|> (1)" $

142 let t = timed

143− in toList (

143+ in toListSerial (

144 ((t 8 <|> t 4) <> (t 2 <|> t 0))

145 <|> ((t 8 <|> t 4) <> (t 2 <|> t 0)))

146 `shouldReturn` ([4,4,8,8,0,0,2,2])

147 it "Nest <|>, <>, <|> (2)" $

148 let t = timed

149− in toList (

149+ in toListSerial (

150 ((t 4 <|> t 8) <> (t 1 <|> t 2))

151 <|> ((t 4 <|> t 8) <> (t 1 <|> t 2)))

152 `shouldReturn` ([4,4,8,8,1,1,2,2])

@@ -155,20 +155,20 @@ main = hspec $ do

155 {-

156 it "Nest <|>, <=>, <|> (1)" $

157 let t = timed

158− in toList (

158+ in toListSerial (

159 ((t 8 <|> t 4) <=> (t 2 <|> t 0))

160 <|> ((t 9 <|> t 4) <=> (t 2 <|> t 0)))

161 `shouldReturn` ([4,4,0,0,8,2,9,2])

162 it "Nest <|>, <=>, <|> (2)" $

163 let t = timed

164− in toList (

164+ in toListSerial (

165 ((t 4 <|> t 8) <=> (t 1 <|> t 2))

166 <|> ((t 4 <|> t 9) <=> (t 1 <|> t 2)))

167 `shouldReturn` ([4,4,1,1,8,2,9,2])

168 -}

169 it "Nest <|>, <|>, <|>" $

170 let t = timed

171− in toList (

171+ in toListSerial (

172 ((t 4 <|> t 8) <|> (t 0 <|> t 2))

173 <|> ((t 4 <|> t 8) <|> (t 0 <|> t 2)))

174 `shouldReturn` ([0,0,2,2,4,4,8,8])

@@ -185,17 +185,37 @@ main = hspec $ do

185 -- Bind and monoidal composition combinations

186 ---------------------------------------------------------------------------

187

188− forM_ [(>>=), (>->), (>>|), (>|>)] $ \f ->

188+ forM_ [(<>), (<=>), (<|), (<|>)] $ \g ->

189+ describe "Bind and compose" $ bindAndComposeSimple toListSerial g

190+

191+ forM_ [(<>), (<=>), (<|), (<|>)] $ \g ->

192+ describe "Bind and compose" $ bindAndComposeSimple toListInterleaved g

193+

194+ forM_ [(<>), (<=>), (<|), (<|>)] $ \g ->

195+ describe "Bind and compose" $ bindAndComposeSimple toListAsync g

196+

189− forM_ [(<>), (<=>), (<|), (<|>)] $ \g ->

197+ forM_ [(<>), (<=>), (<|), (<|>)] $ \g ->

190− describe "Bind and compose" $ bindAndComposeSimple f g

198+ describe "Bind and compose" $ bindAndComposeSimple toListParallel g

191

199

192 let fldr f = foldr f empty

200 let fldr f = foldr f empty

193 fldl f = foldl f empty

201 fldl f = foldl f empty

202+ in do

194− in forM_ [(>>=), (>->), (>>|), (>|>)] $ \f ->

203+ forM_ [(<>), (<=>), (<|), (<|>)] $ \g ->

195− forM_ [(<>), (<=>), (<|), (<|>)] $ \g ->

204+ forM_ [fldr, fldl] $ \k ->

205+ describe "Bind and compose" $

206+ bindAndComposeHierarchy toListSerial (k g)

207+ forM_ [(<>), (<=>), (<|), (<|>)] $ \g ->

208+ forM_ [fldr, fldl] $ \k ->

209+ describe "Bind and compose" $

210+ bindAndComposeHierarchy toListInterleaved (k g)

211+ forM_ [(<>), (<=>), (<|), (<|>)] $ \g ->

196− forM_ [fldr, fldl] $ \k ->

212+ forM_ [fldr, fldl] $ \k ->

213+ describe "Bind and compose" $

214+ bindAndComposeHierarchy toListAsync (k g)

215+ forM_ [(<>), (<=>), (<|), (<|>)] $ \g ->

216+ forM_ [fldr, fldl] $ \k ->

197− describe "Bind and compose" $

217+ describe "Bind and compose" $

198− bindAndComposeHierarchy f (k g)

218+ bindAndComposeHierarchy toListParallel (k g)

199

219

200 ---------------------------------------------------------------------------

220 ---------------------------------------------------------------------------

201 -- TBD Bind and Bind combinations

221 -- TBD Bind and Bind combinations

@@ -206,149 +226,153 @@ main = hspec $ do

206

226

207 describe "Transformation" $ transformOps (<>)

227 describe "Transformation" $ transformOps (<>)

208 describe "Serial zipping" $

228 describe "Serial zipping" $

209− zipOps A.zipWith A.zipWithM A.ZipSerial A.getZipSerial

229+ zipOps A.zipWith A.zipWithM fromStream fromZipStream

210 describe "Async zipping" $

230 describe "Async zipping" $

211− zipOps A.zipAsyncWith A.zipAsyncWithM A.ZipAsync A.getZipAsync

231+ zipOps A.zipAsyncWith A.zipAsyncWithM fromStream fromZipAsync

212

232

213zipOps :: Applicative (f IO)

233zipOps :: Applicative (f IO)

214 => (forall a b c. (a -> b -> c)

234 => (forall a b c. (a -> b -> c)

215− -> AsyncT IO a -> AsyncT IO b -> AsyncT IO c)

235+ -> StreamT IO a -> StreamT IO b -> StreamT IO c)

216− -> (forall a b c. (a -> b -> AsyncT IO c)

236+ -> (forall a b c. (a -> b -> StreamT IO c)

217− -> AsyncT IO a -> AsyncT IO b -> AsyncT IO c)

237+ -> StreamT IO a -> StreamT IO b -> StreamT IO c)

218− -> (forall a. AsyncT IO a -> f IO a)

238+ -> (forall a. StreamT IO a -> f IO a)

219− -> (forall a. f IO a -> AsyncT IO a)

239+ -> (forall a. f IO a -> StreamT IO a)

220 -> Spec

240 -> Spec

221zipOps z zM app getz = do

241zipOps z zM app getz = do

222 it "zipWith" $

242 it "zipWith" $

223 let s1 = foldMapWith (<|>) return [1..10]

243 let s1 = foldMapWith (<|>) return [1..10]

224 s2 = foldMapWith (<>) return [1..]

244 s2 = foldMapWith (<>) return [1..]

225− in toList (z (+) s1 s2)

245+ in toListSerial (z (+) s1 s2)

226 `shouldReturn` ([2,4..20] :: [Int])

246 `shouldReturn` ([2,4..20] :: [Int])

227

247

228 it "zipWithM" $

248 it "zipWithM" $

229 let s1 = foldMapWith (<|>) return [1..10]

249 let s1 = foldMapWith (<|>) return [1..10]

230 s2 = foldMapWith (<>) return [1..]

250 s2 = foldMapWith (<>) return [1..]

231− in toList (zM (\a b -> return (a + b)) s1 s2)

251+ in toListSerial (zM (\a b -> return (a + b)) s1 s2)

232 `shouldReturn` ([2,4..20] :: [Int])

252 `shouldReturn` ([2,4..20] :: [Int])

233

253

234 it "Applicative zip" $

254 it "Applicative zip" $

235 let s1 = foldMapWith (<|>) return [1..10]

255 let s1 = foldMapWith (<|>) return [1..10]

236 s2 = foldMapWith (<>) return [1..]

256 s2 = foldMapWith (<>) return [1..]

237− in toList (getz ((+) <$> app s1 <*> app s2))

257+ in toListSerial (getz ((+) <$> app s1 <*> app s2))

238 `shouldReturn` ([2,4..20] :: [Int])

258 `shouldReturn` ([2,4..20] :: [Int])

239

259

240−timed :: Int -> AsyncT IO Int

260+timed :: Int -> StreamT IO Int

241timed x = liftIO (threadDelay (x * 100000)) >> return x

261timed x = liftIO (threadDelay (x * 100000)) >> return x

242

262

243thenBind :: Spec

263thenBind :: Spec

244thenBind = do

264thenBind = do

245 it "Simple runAsyncly and 'then' with IO" $

265 it "Simple runAsyncly and 'then' with IO" $

246− runAsyncly (liftIO (putStrLn "hello") >> liftIO (putStrLn "world"))

266+ serially (liftIO (putStrLn "hello") >> liftIO (putStrLn "world"))

247 `shouldReturn` ()

267 `shouldReturn` ()

248 it "Then and toList" $

268 it "Then and toList" $

249− toList (return (1 :: Int) >> return 2) `shouldReturn` ([2] :: [Int])

269+ toListSerial (return (1 :: Int) >> return 2) `shouldReturn` ([2] :: [Int])

250

270

251−pureBind :: (AsyncT IO Int -> (Int -> AsyncT IO Int) -> AsyncT IO Int) -> Spec

271+type ToListType s = (forall a. s IO a -> IO [a])

272+pureBind :: Monad (s IO) => ToListType s -> Spec

252−pureBind f = do

273+pureBind l = do

253 it "Bind and toList" $

274 it "Bind and toList" $

254− toList (return 1 `f` \x -> return 2 `f` \y -> return (x + y))

275+ l (return 1 `f` \x -> return 2 `f` \y -> return (x + y))

255 `shouldReturn` ([3] :: [Int])

276 `shouldReturn` ([3] :: [Int])

277+ where f = (>>=)

256

278

257−bindEmpty :: (AsyncT IO Int -> (Int -> AsyncT IO Int) -> AsyncT IO Int) -> Spec

279+bindEmpty :: (Monad (s IO), Alternative (s IO)) => ToListType s -> Spec

258−bindEmpty f = it "Binds with empty" $

280+bindEmpty l = it "Binds with empty" $

259− (toList (return 1 `f` \_ -> empty `f` \_ -> return 2))

281+ (l (return (1 :: Int) `f` \_ -> empty `f` \_ -> return 2))

260 `shouldReturn` ([] :: [Int])

282 `shouldReturn` ([] :: [Int])

283+ where f = (>>=)

261

284

262−interleaved

285+interleaveCheck

263− :: (AsyncT IO Int -> AsyncT IO Int -> AsyncT IO Int)

286+ :: (StreamT IO Int -> StreamT IO Int -> StreamT IO Int)

264 -> Spec

287 -> Spec

265−interleaved f =

288+interleaveCheck f =

266 it "Interleave four" $

289 it "Interleave four" $

267− toList ((return 0 <> return 1) `f` (return 100 <> return 101))

290+ toListSerial ((return 0 <> return 1) `f` (return 100 <> return 101))

268 `shouldReturn` ([0, 100, 1, 101])

291 `shouldReturn` ([0, 100, 1, 101])

269

292

270−parallelCheck :: (AsyncT IO Int -> AsyncT IO Int -> AsyncT IO Int) -> Spec

293+parallelCheck :: (StreamT IO Int -> StreamT IO Int -> StreamT IO Int) -> Spec

271parallelCheck f = do

294parallelCheck f = do

272 it "Parallel ordering left associated" $

295 it "Parallel ordering left associated" $

273− toList (((event 4 `f` event 3) `f` event 2) `f` event 1)

296+ toListSerial (((event 4 `f` event 3) `f` event 2) `f` event 1)

274 `shouldReturn` ([1..4])

297 `shouldReturn` ([1..4])

275

298

276 it "Parallel ordering right associated" $

299 it "Parallel ordering right associated" $

277− toList (event 4 `f` (event 3 `f` (event 2 `f` event 1)))

300+ toListSerial (event 4 `f` (event 3 `f` (event 2 `f` event 1)))

278 `shouldReturn` ([1..4])

301 `shouldReturn` ([1..4])

279

302

280 where event n = (liftIO $ threadDelay (n * 100000)) >> (return n)

303 where event n = (liftIO $ threadDelay (n * 100000)) >> (return n)

281

304

282compose

305compose

283− :: (AsyncT IO Int -> AsyncT IO Int -> AsyncT IO Int)

306+ :: (StreamT IO Int -> StreamT IO Int -> StreamT IO Int)

284 -> ([Int] -> [Int])

307 -> ([Int] -> [Int])

285 -> Spec

308 -> Spec

286compose f srt = do

309compose f srt = do

287 it "Compose mempty, mempty" $

310 it "Compose mempty, mempty" $

288− (toList (mempty `f` mempty)) `shouldReturn` []

311+ (tl (mempty `f` mempty)) `shouldReturn` []

289 it "Compose empty, empty" $

312 it "Compose empty, empty" $

290− (toList (empty `f` empty)) `shouldReturn` []

313+ (tl (empty `f` empty)) `shouldReturn` []

291 it "Compose empty at the beginning" $

314 it "Compose empty at the beginning" $

292− (toList $ (empty `f` return 1)) `shouldReturn` [1]

315+ (tl $ (empty `f` return 1)) `shouldReturn` [1]

293 it "Compose empty at the end" $

316 it "Compose empty at the end" $

294− (toList $ (return 1 `f` empty)) `shouldReturn` [1]

317+ (tl $ (return 1 `f` empty)) `shouldReturn` [1]

295 it "Compose two" $

318 it "Compose two" $

296− (toList (return 0 `f` return 1) >>= return . srt)

319+ (tl (return 0 `f` return 1) >>= return . srt)

297 `shouldReturn` [0, 1]

320 `shouldReturn` [0, 1]

298 it "Compose three - empty in the middle" $

321 it "Compose three - empty in the middle" $

299− ((toList $ (return 0 `f` empty `f` return 1)) >>= return . srt)

322+ ((tl $ (return 0 `f` empty `f` return 1)) >>= return . srt)

300 `shouldReturn` [0, 1]

323 `shouldReturn` [0, 1]

301 it "Compose left associated" $

324 it "Compose left associated" $

302− ((toList $ (((return 0 `f` return 1) `f` return 2) `f` return 3))

325+ ((tl $ (((return 0 `f` return 1) `f` return 2) `f` return 3))

303 >>= return . srt) `shouldReturn` [0, 1, 2, 3]

326 >>= return . srt) `shouldReturn` [0, 1, 2, 3]

304 it "Compose right associated" $

327 it "Compose right associated" $

305− ((toList $ (return 0 `f` (return 1 `f` (return 2 `f` return 3))))

328+ ((tl $ (return 0 `f` (return 1 `f` (return 2 `f` return 3))))

306 >>= return . srt) `shouldReturn` [0, 1, 2, 3]

329 >>= return . srt) `shouldReturn` [0, 1, 2, 3]

307 it "Compose many" $

330 it "Compose many" $

308− ((toList $ forEachWith f [1..100] return) >>= return . srt)

331+ ((tl $ forEachWith f [1..100] return) >>= return . srt)

309 `shouldReturn` [1..100]

332 `shouldReturn` [1..100]

310 it "Compose hierarchical (multiple levels)" $

333 it "Compose hierarchical (multiple levels)" $

311− ((toList $ (((return 0 `f` return 1) `f` (return 2 `f` return 3))

334+ ((tl $ (((return 0 `f` return 1) `f` (return 2 `f` return 3))

312 `f` ((return 4 `f` return 5) `f` (return 6 `f` return 7)))

335 `f` ((return 4 `f` return 5) `f` (return 6 `f` return 7)))

313 ) >>= return . srt) `shouldReturn` [0..7]

336 ) >>= return . srt) `shouldReturn` [0..7]

337+ where tl = toListSerial

314

338

315composeAndComposeSimple

339composeAndComposeSimple

316− :: (AsyncT IO Int -> AsyncT IO Int -> AsyncT IO Int)

340+ :: (StreamT IO Int -> StreamT IO Int -> StreamT IO Int)

317− -> (AsyncT IO Int -> AsyncT IO Int -> AsyncT IO Int)

341+ -> (StreamT IO Int -> StreamT IO Int -> StreamT IO Int)

318 -> [[Int]]

342 -> [[Int]]

319 -> Spec

343 -> Spec

320composeAndComposeSimple f g answer = do

344composeAndComposeSimple f g answer = do

321 it "Compose right associated outer expr, right folded inner" $

345 it "Compose right associated outer expr, right folded inner" $

322 let fold = foldMapWith g return

346 let fold = foldMapWith g return

323− in (toList (fold [1,2,3] `f` (fold [4,5,6] `f` fold [7,8,9])))

347+ in (toListSerial (fold [1,2,3] `f` (fold [4,5,6] `f` fold [7,8,9])))

324 `shouldReturn` (answer !! 0)

348 `shouldReturn` (answer !! 0)

325

349

326 it "Compose left associated outer expr, right folded inner" $

350 it "Compose left associated outer expr, right folded inner" $

327 let fold = foldMapWith g return

351 let fold = foldMapWith g return

328− in (toList ((fold [1,2,3] `f` fold [4,5,6]) `f` fold [7,8,9]))

352+ in (toListSerial ((fold [1,2,3] `f` fold [4,5,6]) `f` fold [7,8,9]))

329 `shouldReturn` (answer !! 1)

353 `shouldReturn` (answer !! 1)

330

354

331 it "Compose right associated outer expr, left folded inner" $

355 it "Compose right associated outer expr, left folded inner" $

332 let fold xs = foldl g empty $ map return xs

356 let fold xs = foldl g empty $ map return xs

333− in (toList (fold [1,2,3] `f` (fold [4,5,6] `f` fold [7,8,9])))

357+ in (toListSerial (fold [1,2,3] `f` (fold [4,5,6] `f` fold [7,8,9])))

334 `shouldReturn` (answer !! 2)

358 `shouldReturn` (answer !! 2)

335

359

336 it "Compose left associated outer expr, left folded inner" $

360 it "Compose left associated outer expr, left folded inner" $

337 let fold xs = foldl g empty $ map return xs

361 let fold xs = foldl g empty $ map return xs

338− in (toList ((fold [1,2,3] `f` fold [4,5,6]) `f` fold [7,8,9]))

362+ in (toListSerial ((fold [1,2,3] `f` fold [4,5,6]) `f` fold [7,8,9]))

339 `shouldReturn` (answer !! 3)

363 `shouldReturn` (answer !! 3)

340

364

341

365

342loops

366loops

343− :: (AsyncT IO Int -> AsyncT IO Int -> AsyncT IO Int)

367+ :: (StreamT IO Int -> StreamT IO Int -> StreamT IO Int)

344 -> ([Int] -> [Int])

368 -> ([Int] -> [Int])

345 -> ([Int] -> [Int])

369 -> ([Int] -> [Int])

346 -> Spec

370 -> Spec

347loops f tsrt hsrt = do

371loops f tsrt hsrt = do

348− it "Tail recursive loop" $ (toList (loopTail 0) >>= return . tsrt)

372+ it "Tail recursive loop" $ (toListSerial (loopTail 0) >>= return . tsrt)

349 `shouldReturn` [0..3]

373 `shouldReturn` [0..3]

350

374

351− it "Head recursive loop" $ (toList (loopHead 0) >>= return . hsrt)

375+ it "Head recursive loop" $ (toListSerial (loopHead 0) >>= return . hsrt)

352 `shouldReturn` [0..3]

376 `shouldReturn` [0..3]

353

377

354 where

378 where

@@ -363,26 +387,27 @@ loops f tsrt hsrt = do

363 return x `f` (if x < 3 then loopTail (x + 1) else empty)

387 return x `f` (if x < 3 then loopTail (x + 1) else empty)

364

388

365bindAndComposeSimple

389bindAndComposeSimple

366− :: (AsyncT IO Int -> (Int -> AsyncT IO Int) -> AsyncT IO Int)

390+ :: (Alternative (s IO), Monad (s IO), Monoid (s IO Int)) => (forall a. s IO a -> IO [a])

367− -> (AsyncT IO Int -> AsyncT IO Int -> AsyncT IO Int)

391+ -> (s IO Int -> s IO Int -> s IO Int)

368 -> Spec

392 -> Spec

369−bindAndComposeSimple f g = do

393+bindAndComposeSimple tl g = do

370 it "Compose many (right fold) with bind" $

394 it "Compose many (right fold) with bind" $

371− (toList (forEachWith g [1..10 :: Int] $ \x -> return x `f` (return . id))

395+ (tl (forEachWith g [1..10 :: Int] $ \x -> return x `f` (return . id))

372 >>= return . sort) `shouldReturn` [1..10]

396 >>= return . sort) `shouldReturn` [1..10]

373

397

374 it "Compose many (left fold) with bind" $

398 it "Compose many (left fold) with bind" $

375 let forL xs k = foldl g empty $ map k xs

399 let forL xs k = foldl g empty $ map k xs

376− in (toList (forL [1..10 :: Int] $ \x -> return x `f` (return . id))

400+ in (tl (forL [1..10 :: Int] $ \x -> return x `f` (return . id))

377 >>= return . sort) `shouldReturn` [1..10]

401 >>= return . sort) `shouldReturn` [1..10]

402+ where f = (>>=)

378

403

379bindAndComposeHierarchy

404bindAndComposeHierarchy

380− :: (AsyncT IO Int -> (Int -> AsyncT IO Int) -> AsyncT IO Int)

405+ :: Monad (s IO) => (forall a. s IO a -> IO [a])

381− -> ([AsyncT IO Int] -> AsyncT IO Int)

406+ -> ([s IO Int] -> s IO Int)

382 -> Spec

407 -> Spec

383−bindAndComposeHierarchy f g = do

408+bindAndComposeHierarchy tl g = do

384 it "Bind and compose nested" $

409 it "Bind and compose nested" $

385− (toList bindComposeNested >>= return . sort)

410+ (tl bindComposeNested >>= return . sort)

386 `shouldReturn` (sort (

411 `shouldReturn` (sort (

387 [12, 18]

412 [12, 18]

388 ++ replicate 3 13

413 ++ replicate 3 13

@@ -393,7 +418,7 @@ bindAndComposeHierarchy f g = do

393

418

394 where

419 where

395

420

396− bindComposeNested :: AsyncT IO Int

421+ -- bindComposeNested :: AsyncT IO Int

397 bindComposeNested =

422 bindComposeNested =

398 let c1 = tripleCompose (return 1) (return 2) (return 3)

423 let c1 = tripleCompose (return 1) (return 2) (return 3)

399 c2 = tripleCompose (return 4) (return 5) (return 6)

424 c2 = tripleCompose (return 4) (return 5) (return 6)

@@ -410,17 +435,18 @@ bindAndComposeHierarchy f g = do

410 mx `f` \x -> my

435 mx `f` \x -> my

411 `f` \y -> mz

436 `f` \y -> mz

412 `f` \z -> return (x + y + z)

437 `f` \z -> return (x + y + z)

438+ f = (>>=)

413

439

414mixedOps :: Spec

440mixedOps :: Spec

415mixedOps = do

441mixedOps = do

416 it "Compose many ops" $

442 it "Compose many ops" $

417− (toList composeMixed >>= return . sort)

443+ (toListSerial composeMixed >>= return . sort)

418 `shouldReturn` ([8,9,9,9,9,9,10,10,10,10,10,10,10,10,10,10,11,11

444 `shouldReturn` ([8,9,9,9,9,9,10,10,10,10,10,10,10,10,10,10,11,11

419 ,11,11,11,11,11,11,11,11,12,12,12,12,12,13

445 ,11,11,11,11,11,11,11,11,12,12,12,12,12,13

420 ] :: [Int])

446 ] :: [Int])

421 where

447 where

422

448

423− composeMixed :: AsyncT IO Int

449+ composeMixed :: StreamT IO Int

424 composeMixed = do

450 composeMixed = do

425 liftIO $ return ()

451 liftIO $ return ()

426 liftIO $ putStr ""

452 liftIO $ putStr ""

@@ -441,24 +467,24 @@ mixedOps = do

441 return (x1 + y1 + z1)

467 return (x1 + y1 + z1)

442 return (x + y + z)

468 return (x + y + z)

443

469

444−transformOps :: (AsyncT IO Int -> AsyncT IO Int -> AsyncT IO Int) -> Spec

470+transformOps :: (StreamT IO Int -> StreamT IO Int -> StreamT IO Int) -> Spec

445transformOps f = do

471transformOps f = do

446 it "take all" $

472 it "take all" $

447− (toList $ A.take 10 $ foldMapWith f return [1..10])

473+ (toListSerial $ A.take 10 $ foldMapWith f return [1..10])

448 `shouldReturn` [1..10]

474 `shouldReturn` [1..10]

449 it "take none" $

475 it "take none" $

450− (toList $ A.take 0 $ foldMapWith f return [1..10])

476+ (toListSerial $ A.take 0 $ foldMapWith f return [1..10])

451 `shouldReturn` []

477 `shouldReturn` []

452 it "take 5" $

478 it "take 5" $

453− (toList $ A.take 5 $ foldMapWith f return [1..10])

479+ (toListSerial $ A.take 5 $ foldMapWith f return [1..10])

454 `shouldReturn` [1..5]

480 `shouldReturn` [1..5]

455

481

456 it "drop all" $

482 it "drop all" $

457− (toList $ A.drop 10 $ foldMapWith f return [1..10])

483+ (toListSerial $ A.drop 10 $ foldMapWith f return [1..10])

458 `shouldReturn` []

484 `shouldReturn` []

459 it "drop none" $

485 it "drop none" $

460− (toList $ A.drop 0 $ foldMapWith f return [1..10])

486+ (toListSerial $ A.drop 0 $ foldMapWith f return [1..10])

461 `shouldReturn` [1..10]

487 `shouldReturn` [1..10]

462 it "drop 5" $

488 it "drop 5" $

463− (toList $ A.drop 5 $ foldMapWith f return [1..10])

489+ (toListSerial $ A.drop 5 $ foldMapWith f return [1..10])

464 `shouldReturn` [6..10]

490 `shouldReturn` [6..10]