29f64fe · streamly · git.iko.soy

git.iko.soy/streamly

Separate types for different styles of product composition

commit29f64fe8d1a3553f2d385d397ff2d8e03b6d53a0 authorHarendra Kumar <harendra.kumar@gmail.com> dateWed 25 Oct 2017, 08:01 (9 years ago) parentff35f64

5 files changed+695−377

compare with master

unifiedsplit

benchmark/Main.hs+57−28 src/Asyncly.hs+28−15 src/Asyncly/AsyncT.hs+84−166 src/Asyncly/Prelude.hs+396−64 test/Main.hs+130−104

benchmark/Main.hs+57−28

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

11{-# LANGUAGE CPP #-}

2+{-# LANGUAGE RankNTypes #-}

34module Main where

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

3031main = do

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

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

3438 defaultMain [

3539 bgroup "asyncly"

3640 [ bgroup "serial bind"

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

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

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

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

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

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

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

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

4145 ]

4246

43− , bgroup "fair bind"

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

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

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

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

47+ , bgroup "interleaved bind"

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

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

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

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

4852 ]

4953

50− , bgroup "parallel bind"

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

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

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

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

54+ , bgroup "async bind"

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

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

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

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

5559 ]

5660

57− , bgroup "fair parallel bind"

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

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

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

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

61+ , bgroup "parallel bind"

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

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

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

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

6266 ]

6367

6468 -- Benchmark smallest possible actions composed together

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

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

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

9397

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

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

9599amap = Main.map

96100

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

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

98102afilter = Main.filter

99103

100104{-# INLINE asyncly_basic #-}

101105asyncly_basic

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

103− -> (A.AsyncT IO 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])

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

104108 -> IO Int

105−asyncly_basic f g = do

106− xs <- A.toList $ do

109+asyncly_basic tl g = do

110+ xs <- tl $ do

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

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

109113 `f` amap (+1)

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

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

112116 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

113142

114143{-# 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)

116145 -> IO Int

117146asyncly_nil f = do

118− xs <- A.toList $ do

147+ xs <- A.toListSerial $ do

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

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

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

src/Asyncly.hs+28−15

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

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

5353-- compositions.

5454module Asyncly

55− ( AsyncT

56− , MonadAsync

55+ ( MonadAsync

56+ , Streaming

57+ ( fromStream

58+ , fromInterleaved

59+ , fromAsync

60+ , fromParallel

61+ , fromZipStream

62+ , fromZipAsync

63+ , (<=>)

64+ , (<|)

65+ )

5766

58− -- * Running

59− , runAsyncly

60− , toList

61− , async

67+ -- * Stream Styles

68+ , StreamT

69+ , InterleavedT

70+ , AsyncT

71+ , ParallelT

72+ , ZipStream

73+ , ZipAsync

6274

63− -- * Monadic Composition (Conjunction)

64− -- $bind

65− , (>->)

66− , (>>|)

67− , (>|>)

75+ -- * Running Streams

76+ , serially

77+ , interleaved

78+ , asyncly

79+ , parallely

6880

69− -- * Monoidal Composition (Disjunction)

70− -- $monoidal

71− , (<=>)

72− , (<|)

81+ , toListSerial

82+ , toListInterleaved

83+ , toListAsync

84+ , toListParallel

7385

7486 -- * General Fold Utilities

7587 , foldWith

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

8799where

88100

89101import Asyncly.AsyncT

102+import Asyncly.Prelude

90103import Data.Semigroup (Semigroup(..))

91104import Control.Applicative (Alternative(..))

92105import Control.Monad (MonadPlus(..))

src/Asyncly/AsyncT.hs+84−166

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

1616--

1717--

1818module Asyncly.AsyncT

19− ( AsyncT (..)

19+ ( Stream (..)

2020 , MonadAsync

21− , runAsyncly

21+

22+ , fromCallback

23+

24+ , runStreaming

2225 , toList

2326 , async

2427

2528 , interleave

26− , (<=>)

27− , parAhead

28− , (<>|)

29+ , parallel

2930 , parLeft

30− , (<|)

31−

32− , (>->)

33− , (>>|)

34− , (>|>)

31+ , CtxType (..)

32+ , SchedPolicy (..)

33+ , Dimension (..)

3534

3635 , foldWith

3736 , foldMapWith

3837 , forEachWith

39− , fromCallback

4038 )

4139where

4240

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

10199data Context m a =

102100 Context { outputQueue :: IORef [ChildEvent a]

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

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

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

105103 , runqueue :: m ()

106104 , runningThreads :: IORef (Set ThreadId)

107105 , queueEmpty :: m Bool

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

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

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

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

115−newtype AsyncT m a =

116− AsyncT {

117− runAsyncT :: forall r.

113+newtype Stream m a =

114+ Stream {

115+ runStream :: forall r.

118116 Maybe (Context m a) -- local state

119117 -> m r -- stop

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

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

121119 -> m r

122120 }

123121

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

130128

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

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

134132 m1 <> m2 = go m1

135133 where

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

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

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

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

138136 yield a Nothing = yld a (Just m2)

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

140138 in m Nothing stop yield

141139

142−instance Monoid (AsyncT m a) where

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

140+instance Monoid (Stream m a) where

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

144142 mappend = (<>)

145143

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

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

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

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

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

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

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

150147 yield a Nothing = yld a (Just m2)

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

152− (runAsyncT 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

149+ (runStream m1) Nothing stop yield

161150

162151------------------------------------------------------------------------------

163152-- Functor

164153------------------------------------------------------------------------------

165154

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

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

167156 fmap = liftM

168157

169158------------------------------------------------------------------------------

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

172161

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

174163

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

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

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

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

177166 (<*>) = ap

178167

179168------------------------------------------------------------------------------

180169-- Monad

181170------------------------------------------------------------------------------

182171

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−

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

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

203175 return = pure

204176

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

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

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

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

207179 yield a Nothing = run $ f a

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

209181 in m Nothing stp yield

210182

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−

257183------------------------------------------------------------------------------

258184-- Alternative

259185------------------------------------------------------------------------------

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

287213

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

289215{-# INLINE enqueueLIFO #-}

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

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

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

292218

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

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

294220runqueueLIFO ctx q = run

295221

296222 where

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

299225 work <- dequeue

300226 case work of

301227 Nothing -> sendStop ctx

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

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

303229

304230 sendit a = send ctx (ChildYield a)

305231 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

307233

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

309235 case ms of

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

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

312238

313239{-# INLINE enqueueFIFO #-}

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

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

315241enqueueFIFO = pushL

316242

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

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

318244runqueueFIFO ctx q = run

319245

320246 where

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

323249 work <- dequeue

324250 case work of

325251 Nothing -> sendStop ctx

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

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

327253

328254 dequeue = liftIO $ tryPopR q

329255 sendit a = send ctx (ChildYield a)

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

387313

388314-- | Pull an AsyncT stream from a context

389315{-# NOINLINE pullFromCtx #-}

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

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

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

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

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

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

394320 -- raise an explicit exception.

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

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

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

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

407333

408334 where

409335

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

414340 throwM e

415341

416342 {-# INLINE processEvents #-}

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

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

418344 done <- allThreadsDone ctx

419345 if not done

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

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

421347 else stp

422348

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

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

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

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

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

426352

427353 case ev of

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

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

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

471397pushPairToCtx :: 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)

473399pushPairToCtx ctype m1 m2 = do

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

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

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

489415

490416{-# NOINLINE pushPullFork #-}

491417pushPullFork :: MonadAsync m

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

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

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

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

494420 ctx <- pushPairToCtx ct m1 m2

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

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

496422

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

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

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

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

530456{-# INLINE parallel #-}

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

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

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

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

533459 case ctx of

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

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

535461 Just c | ctxType c /= ct ->

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

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

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

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

538464

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

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

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

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

543469-- results fairly.

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

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

545471 empty = mempty

546472

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

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

551477-- Other monoidal compositions for parallel actions

552478------------------------------------------------------------------------------

553479

480+{-

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

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

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

556483parAhead = undefined

557484

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

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

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

562489(<>|) = parAhead

490+-}

563491

564492-- | Same as '<|'.

565493{-# 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

567495parLeft = parallel (CtxType Disjunction LIFO)

568496

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

579498 mzero = empty

580499 mplus = (<|>)

581500

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

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

585504pushOneToCtx ctype m = do

586505 ctx <- liftIO $

587506 case ctype of

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

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

611530-- once a ContextUsedAfterEOF exception will be raised.

612531

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

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

614533async m = do

615534 ctx <- pushOneToCtx (CtxType Disjunction LIFO) m

616535 return $ pullFromCtx ctx

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

619538-- Num

620539------------------------------------------------------------------------------

621540

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

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

623542 fromInteger n = pure (fromInteger n)

624543

625544 negate = fmap negate

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

630549 (*) = liftA2 (*)

631550 (-) = liftA2 (-)

632551

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

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

634553 fromRational n = pure (fromRational n)

635554

636555 recip = fmap recip

637556

638557 (/) = liftA2 (/)

639558

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

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

641560 pi = pure pi

642561

643562 exp = fmap exp

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

663582-- AsyncT transformer

664583-------------------------------------------------------------------------------

665584

666−instance MonadTrans AsyncT where

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

585+instance MonadTrans Stream where

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

668587

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

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

670589 liftBase = liftBaseDefault

671590

672591------------------------------------------------------------------------------

673592-- Standard transformer instances

674593------------------------------------------------------------------------------

675594

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

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

677596 liftIO = lift . liftIO

678597

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

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

680599 throwM = lift . throwM

681600

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

684603 throwError = lift . throwError

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

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

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

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

687606 yield a Nothing = yld a Nothing

688607 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

690609

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

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

692611 ask = lift ask

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

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

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

695614 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

697616

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

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

699618 get = lift get

700619 put x = lift (put x)

701620 state k = lift (state k)

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

706625

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

708627-- results.

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

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

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

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

711630

712631 where

713632

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

715634

716635 {-# INLINE yield #-}

717636 yield _ Nothing = stop

718− yield _ (Just x) = runAsyncly x

719−

637+ yield _ (Just x) = runStreaming x

720638

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

722640{-# INLINABLE toList #-}

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

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

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

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

725643

726644 where

727645

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

759677------------------------------------------------------------------------------

760678

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

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

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

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

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

src/Asyncly/Prelude.hs+396−64

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

11{-# LANGUAGE FlexibleContexts #-}

2+{-# LANGUAGE FlexibleInstances #-}

23{-# LANGUAGE GeneralizedNewtypeDeriving#-}

4+{-# LANGUAGE MultiParamTypeClasses #-}

5+{-# LANGUAGE RankNTypes #-}

6+{-# LANGUAGE StandaloneDeriving #-}

37{-# LANGUAGE UndecidableInstances #-} -- XXX

59-- |

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

1317--

1418--

1519module Asyncly.Prelude

16− (

17− -- * Construction

18− unfoldr

20+ ( Streaming (..)

21+ , StreamT

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

19130

20131 -- * 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

22140

23141 -- * Special folds

24− , take

25− , drop

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

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

26144

27145 -- * Zipping

28− , ZipSerial (..)

29− , zipWith

30− , zipWithM

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

147+ zipWithM :: Monad m =>

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)

31254

32− , ZipAsync (..)

33− , zipAsyncWith

34− , zipAsyncWithM

35− )

36−where

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

256+-- Alternative ways to bind

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

37258

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

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

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

41−import Asyncly.AsyncT

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

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

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

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

42324

43325------------------------------------------------------------------------------

44326-- Construct and deconstruct

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

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

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

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

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

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

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

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

52334

53335 where

54336

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

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

60342

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

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

63−unfoldr step = go

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

345+unfold step = go

64346 where

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

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

66348 mayb <- step s

67349 case mayb of

68350 Nothing -> stp

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

73355------------------------------------------------------------------------------

74356

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

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

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

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

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

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

80362 if (n <= 0)

81363 then stp

82− else (runAsyncT m) ctx stp yield

364+ else (runStream m) ctx stp yield

83365

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

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

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

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

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

87369 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

89371 if (n <= 0)

90− then (runAsyncT m) ctx stp yld

91− else (runAsyncT m) ctx stp yield

372+ then (runStream m) ctx stp yld

373+ else (runStream m) ctx stp yield

92374

93375------------------------------------------------------------------------------

94376-- Serially Zipping Streams

95377------------------------------------------------------------------------------

96378

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

98−zipWithM :: Monad m =>

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

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

380+zippingWithM :: Monad m =>

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

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

101383 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

103385 yield2 b (Just rb) =

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

105− in (runAsyncT m2) Nothing stp yield2

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

387+ in (runStream m2) Nothing stp yield2

106388 let yield1 a Nothing = merge a mempty

107389 yield1 a (Just ra) = merge a ra

108− (runAsyncT m1) Nothing stp yield1

390+ (runStream m1) Nothing stp yield1

109391

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

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

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

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

113395 let merge a ra =

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

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

116− in (runAsyncT m2) Nothing stp yield2

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

398+ in (runStream m2) Nothing stp yield2

117399 let yield1 a Nothing = merge a mempty

118400 yield1 a (Just ra) = merge a ra

119− (runAsyncT m1) Nothing stp yield1

401+ (runStream m1) Nothing stp yield1

120402

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

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

123−-- ZipSerial Applicative.

405+-- ZipStream Applicative.

124406--

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

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

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

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

127409 deriving (Functor)

128410

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

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

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

132−

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

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

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

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

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+

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

134441 fromInteger n = pure (fromInteger n)

135442

136443 negate = fmap negate

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

141448 (*) = liftA2 (*)

142449 (-) = liftA2 (-)

143450

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

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

145452 fromRational n = pure (fromRational n)

146453

147454 recip = fmap recip

148455

149456 (/) = liftA2 (/)

150457

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

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

152459 pi = pure pi

153460

154461 exp = fmap exp

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

176483

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

178485-- concurrently) using a monadic zipping function.

179−zipAsyncWithM :: MonadAsync m

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

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

486+zippingAsyncWithM :: MonadAsync m

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

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

182489 ma <- async m1

183490 mb <- async m2

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

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

185492

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

187494-- concurrently) using a pure zipping function.

188−zipAsyncWith :: MonadAsync m

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

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

495+zippingAsyncWith :: MonadAsync m

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

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

191498 ma <- async m1

192499 mb <- async m2

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

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

194501

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

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

197504-- is a special case of ZipAsync Applicative.

198505--

199506-- > 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}

201508 deriving (Functor)

202509

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

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

206538

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

208540 fromInteger n = pure (fromInteger n)

test/Main.hs+130−104

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

1414main :: IO ()

1515main = hspec $ do

1616 describe "Runners" $ do

17− it "simple runAsyncly" $

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

19− it "simple runAsyncly with IO" $

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

17+ it "simple serially" $

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

19+ it "simple serially with IO" $

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

2121 it "Captures a return value using toList" $

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

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

2323

2424 describe "Empty" $ do

2525 it "Monoid - mempty" $

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

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

2727 it "Alternative - empty" $

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

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

2929 it "MonadPlus - mzero" $

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

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

3131

3232 ---------------------------------------------------------------------------

3333 -- Functor

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

3535

3636 describe "Functor (fmap)" $ do

3737 it "Simple fmap" $

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

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

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

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

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

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

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

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

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

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

4545

4646 ---------------------------------------------------------------------------

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

4949

5050 describe "Applicative" $ do

5151 it "Simple apply" $

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

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

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

5454

5555 it "Apply - serial composed first argument" $

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

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

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

5858

5959 it "Apply - serial composed second argument" $

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

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

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

6262

6363 it "Apply - parallel composed first argument" $

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

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

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

6666

6767 it "Apply - parallel composed second argument" $

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

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

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

7070

7171 ---------------------------------------------------------------------------

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

7373 ---------------------------------------------------------------------------

7474

7575 describe "Bind then" thenBind

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

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

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

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

76+ describe "Pure bind serial" $ pureBind toListSerial

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

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

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

8080

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

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

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

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

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

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

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

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

8585

8686 ---------------------------------------------------------------------------

8787 -- Monoidal Compositions

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

9898 -- Monoidal Composition ordering checks

9999 ---------------------------------------------------------------------------

100100

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

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

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

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

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

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

105105

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

140140 describe "Nested parallel and serial compositions" $ do

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

142142 let t = timed

143− in toList (

143+ in toListSerial (

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

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

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

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

148148 let t = timed

149− in toList (

149+ in toListSerial (

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

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

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

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

155155 {-

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

157157 let t = timed

158− in toList (

158+ in toListSerial (

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

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

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

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

163163 let t = timed

164− in toList (

164+ in toListSerial (

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

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

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

168168 -}

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

170170 let t = timed

171− in toList (

171+ in toListSerial (

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

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

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

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

185185 -- Bind and monoidal composition combinations

186186 ---------------------------------------------------------------------------

187187

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

189− forM_ [(<>), (<=>), (<|), (<|>)] $ \g ->

190− describe "Bind and compose" $ bindAndComposeSimple f g

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+

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

198+ describe "Bind and compose" $ bindAndComposeSimple toListParallel g

191199

192200 let fldr f = foldr f empty

193201 fldl f = foldl f empty

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

195− forM_ [(<>), (<=>), (<|), (<|>)] $ \g ->

196− forM_ [fldr, fldl] $ \k ->

197− describe "Bind and compose" $

198− bindAndComposeHierarchy f (k g)

202+ in do

203+ 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 ->

212+ forM_ [fldr, fldl] $ \k ->

213+ describe "Bind and compose" $

214+ bindAndComposeHierarchy toListAsync (k g)

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

216+ forM_ [fldr, fldl] $ \k ->

217+ describe "Bind and compose" $

218+ bindAndComposeHierarchy toListParallel (k g)

199219

200220 ---------------------------------------------------------------------------

201221 -- TBD Bind and Bind combinations

@@ -206,149 +226,153 @@ main = hspec $ do

206226

207227 describe "Transformation" $ transformOps (<>)

208228 describe "Serial zipping" $

209− zipOps A.zipWith A.zipWithM A.ZipSerial A.getZipSerial

229+ zipOps A.zipWith A.zipWithM fromStream fromZipStream

210230 describe "Async zipping" $

211− zipOps A.zipAsyncWith A.zipAsyncWithM A.ZipAsync A.getZipAsync

231+ zipOps A.zipAsyncWith A.zipAsyncWithM fromStream fromZipAsync

212232

213233zipOps :: Applicative (f IO)

214234 => (forall a b c. (a -> b -> c)

215− -> AsyncT IO a -> AsyncT IO b -> AsyncT IO c)

216− -> (forall a b c. (a -> b -> AsyncT IO c)

217− -> AsyncT IO a -> AsyncT IO b -> AsyncT IO c)

218− -> (forall a. AsyncT IO a -> f IO a)

219− -> (forall a. f IO a -> AsyncT IO a)

235+ -> StreamT IO a -> StreamT IO b -> StreamT IO c)

236+ -> (forall a b c. (a -> b -> StreamT IO c)

237+ -> StreamT IO a -> StreamT IO b -> StreamT IO c)

238+ -> (forall a. StreamT IO a -> f IO a)

239+ -> (forall a. f IO a -> StreamT IO a)

220240 -> Spec

221241zipOps z zM app getz = do

222242 it "zipWith" $

223243 let s1 = foldMapWith (<|>) return [1..10]

224244 s2 = foldMapWith (<>) return [1..]

225− in toList (z (+) s1 s2)

245+ in toListSerial (z (+) s1 s2)

226246 `shouldReturn` ([2,4..20] :: [Int])

227247

228248 it "zipWithM" $

229249 let s1 = foldMapWith (<|>) return [1..10]

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

232252 `shouldReturn` ([2,4..20] :: [Int])

233253

234254 it "Applicative zip" $

235255 let s1 = foldMapWith (<|>) return [1..10]

236256 s2 = foldMapWith (<>) return [1..]

237− in toList (getz ((+) <$> app s1 <*> app s2))

257+ in toListSerial (getz ((+) <$> app s1 <*> app s2))

238258 `shouldReturn` ([2,4..20] :: [Int])

239259

240−timed :: Int -> AsyncT IO Int

260+timed :: Int -> StreamT IO Int

241261timed x = liftIO (threadDelay (x * 100000)) >> return x

242262

243263thenBind :: Spec

244264thenBind = do

245265 it "Simple runAsyncly and 'then' with IO" $

246− runAsyncly (liftIO (putStrLn "hello") >> liftIO (putStrLn "world"))

266+ serially (liftIO (putStrLn "hello") >> liftIO (putStrLn "world"))

247267 `shouldReturn` ()

248268 it "Then and toList" $

249− toList (return (1 :: Int) >> return 2) `shouldReturn` ([2] :: [Int])

269+ toListSerial (return (1 :: Int) >> return 2) `shouldReturn` ([2] :: [Int])

250270

251−pureBind :: (AsyncT IO Int -> (Int -> AsyncT IO Int) -> AsyncT IO Int) -> Spec

252−pureBind f = do

271+type ToListType s = (forall a. s IO a -> IO [a])

272+pureBind :: Monad (s IO) => ToListType s -> Spec

273+pureBind l = do

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

255276 `shouldReturn` ([3] :: [Int])

277+ where f = (>>=)

256278

257−bindEmpty :: (AsyncT IO Int -> (Int -> AsyncT IO Int) -> AsyncT IO Int) -> Spec

258−bindEmpty f = it "Binds with empty" $

259− (toList (return 1 `f` \_ -> empty `f` \_ -> return 2))

279+bindEmpty :: (Monad (s IO), Alternative (s IO)) => ToListType s -> Spec

280+bindEmpty l = it "Binds with empty" $

281+ (l (return (1 :: Int) `f` \_ -> empty `f` \_ -> return 2))

260282 `shouldReturn` ([] :: [Int])

283+ where f = (>>=)

261284

262−interleaved

263− :: (AsyncT IO Int -> AsyncT IO Int -> AsyncT IO Int)

285+interleaveCheck

286+ :: (StreamT IO Int -> StreamT IO Int -> StreamT IO Int)

264287 -> Spec

265−interleaved f =

288+interleaveCheck f =

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

268291 `shouldReturn` ([0, 100, 1, 101])

269292

270−parallelCheck :: (AsyncT IO Int -> AsyncT IO Int -> AsyncT IO Int) -> Spec

293+parallelCheck :: (StreamT IO Int -> StreamT IO Int -> StreamT IO Int) -> Spec

271294parallelCheck f = do

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

274297 `shouldReturn` ([1..4])

275298

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

278301 `shouldReturn` ([1..4])

279302

280303 where event n = (liftIO $ threadDelay (n * 100000)) >> (return n)

281304

282305compose

283− :: (AsyncT IO Int -> AsyncT IO Int -> AsyncT IO Int)

306+ :: (StreamT IO Int -> StreamT IO Int -> StreamT IO Int)

284307 -> ([Int] -> [Int])

285308 -> Spec

286309compose f srt = do

287310 it "Compose mempty, mempty" $

288− (toList (mempty `f` mempty)) `shouldReturn` []

311+ (tl (mempty `f` mempty)) `shouldReturn` []

289312 it "Compose empty, empty" $

290− (toList (empty `f` empty)) `shouldReturn` []

313+ (tl (empty `f` empty)) `shouldReturn` []

291314 it "Compose empty at the beginning" $

292− (toList $ (empty `f` return 1)) `shouldReturn` [1]

315+ (tl $ (empty `f` return 1)) `shouldReturn` [1]

293316 it "Compose empty at the end" $

294− (toList $ (return 1 `f` empty)) `shouldReturn` [1]

317+ (tl $ (return 1 `f` empty)) `shouldReturn` [1]

295318 it "Compose two" $

296− (toList (return 0 `f` return 1) >>= return . srt)

319+ (tl (return 0 `f` return 1) >>= return . srt)

297320 `shouldReturn` [0, 1]

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

300323 `shouldReturn` [0, 1]

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

303326 >>= return . srt) `shouldReturn` [0, 1, 2, 3]

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

306329 >>= return . srt) `shouldReturn` [0, 1, 2, 3]

307330 it "Compose many" $

308− ((toList $ forEachWith f [1..100] return) >>= return . srt)

331+ ((tl $ forEachWith f [1..100] return) >>= return . srt)

309332 `shouldReturn` [1..100]

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

312335 `f` ((return 4 `f` return 5) `f` (return 6 `f` return 7)))

313336 ) >>= return . srt) `shouldReturn` [0..7]

337+ where tl = toListSerial

314338

315339composeAndComposeSimple

316− :: (AsyncT IO Int -> AsyncT IO Int -> AsyncT IO Int)

317− -> (AsyncT IO Int -> AsyncT IO Int -> AsyncT IO Int)

340+ :: (StreamT IO Int -> StreamT IO Int -> StreamT IO Int)

341+ -> (StreamT IO Int -> StreamT IO Int -> StreamT IO Int)

318342 -> [[Int]]

319343 -> Spec

320344composeAndComposeSimple f g answer = do

321345 it "Compose right associated outer expr, right folded inner" $

322346 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])))

324348 `shouldReturn` (answer !! 0)

325349

326350 it "Compose left associated outer expr, right folded inner" $

327351 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]))

329353 `shouldReturn` (answer !! 1)

330354

331355 it "Compose right associated outer expr, left folded inner" $

332356 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])))

334358 `shouldReturn` (answer !! 2)

335359

336360 it "Compose left associated outer expr, left folded inner" $

337361 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]))

339363 `shouldReturn` (answer !! 3)

340364

341365

342366loops

343− :: (AsyncT IO Int -> AsyncT IO Int -> AsyncT IO Int)

367+ :: (StreamT IO Int -> StreamT IO Int -> StreamT IO Int)

344368 -> ([Int] -> [Int])

345369 -> ([Int] -> [Int])

346370 -> Spec

347371loops f tsrt hsrt = do

348− it "Tail recursive loop" $ (toList (loopTail 0) >>= return . tsrt)

372+ it "Tail recursive loop" $ (toListSerial (loopTail 0) >>= return . tsrt)

349373 `shouldReturn` [0..3]

350374

351− it "Head recursive loop" $ (toList (loopHead 0) >>= return . hsrt)

375+ it "Head recursive loop" $ (toListSerial (loopHead 0) >>= return . hsrt)

352376 `shouldReturn` [0..3]

353377

354378 where

@@ -363,26 +387,27 @@ loops f tsrt hsrt = do

363387 return x `f` (if x < 3 then loopTail (x + 1) else empty)

364388

365389bindAndComposeSimple

366− :: (AsyncT IO Int -> (Int -> AsyncT IO Int) -> AsyncT IO Int)

367− -> (AsyncT IO 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])

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

368392 -> Spec

369−bindAndComposeSimple f g = do

393+bindAndComposeSimple tl g = do

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

372396 >>= return . sort) `shouldReturn` [1..10]

373397

374398 it "Compose many (left fold) with bind" $

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

377401 >>= return . sort) `shouldReturn` [1..10]

402+ where f = (>>=)

378403

379404bindAndComposeHierarchy

380− :: (AsyncT IO Int -> (Int -> AsyncT IO Int) -> AsyncT IO Int)

381− -> ([AsyncT IO Int] -> AsyncT IO Int)

405+ :: Monad (s IO) => (forall a. s IO a -> IO [a])

406+ -> ([s IO Int] -> s IO Int)

382407 -> Spec

383−bindAndComposeHierarchy f g = do

408+bindAndComposeHierarchy tl g = do

384409 it "Bind and compose nested" $

385− (toList bindComposeNested >>= return . sort)

410+ (tl bindComposeNested >>= return . sort)

386411 `shouldReturn` (sort (

387412 [12, 18]

388413 ++ replicate 3 13

@@ -393,7 +418,7 @@ bindAndComposeHierarchy f g = do

393418

394419 where

395420

396− bindComposeNested :: AsyncT IO Int

421+ -- bindComposeNested :: AsyncT IO Int

397422 bindComposeNested =

398423 let c1 = tripleCompose (return 1) (return 2) (return 3)

399424 c2 = tripleCompose (return 4) (return 5) (return 6)

@@ -410,17 +435,18 @@ bindAndComposeHierarchy f g = do

410435 mx `f` \x -> my

411436 `f` \y -> mz

412437 `f` \z -> return (x + y + z)

438+ f = (>>=)

413439

414440mixedOps :: Spec

415441mixedOps = do

416442 it "Compose many ops" $

417− (toList composeMixed >>= return . sort)

443+ (toListSerial composeMixed >>= return . sort)

418444 `shouldReturn` ([8,9,9,9,9,9,10,10,10,10,10,10,10,10,10,10,11,11

419445 ,11,11,11,11,11,11,11,11,12,12,12,12,12,13

420446 ] :: [Int])

421447 where

422448

423− composeMixed :: AsyncT IO Int

449+ composeMixed :: StreamT IO Int

424450 composeMixed = do

425451 liftIO $ return ()

426452 liftIO $ putStr ""

@@ -441,24 +467,24 @@ mixedOps = do

441467 return (x1 + y1 + z1)

442468 return (x + y + z)

443469

444−transformOps :: (AsyncT IO Int -> AsyncT IO Int -> AsyncT IO Int) -> Spec

470+transformOps :: (StreamT IO Int -> StreamT IO Int -> StreamT IO Int) -> Spec

445471transformOps f = do

446472 it "take all" $

447− (toList $ A.take 10 $ foldMapWith f return [1..10])

473+ (toListSerial $ A.take 10 $ foldMapWith f return [1..10])

448474 `shouldReturn` [1..10]

449475 it "take none" $

450− (toList $ A.take 0 $ foldMapWith f return [1..10])

476+ (toListSerial $ A.take 0 $ foldMapWith f return [1..10])

451477 `shouldReturn` []

452478 it "take 5" $

453− (toList $ A.take 5 $ foldMapWith f return [1..10])

479+ (toListSerial $ A.take 5 $ foldMapWith f return [1..10])

454480 `shouldReturn` [1..5]

455481

456482 it "drop all" $

457− (toList $ A.drop 10 $ foldMapWith f return [1..10])

483+ (toListSerial $ A.drop 10 $ foldMapWith f return [1..10])

458484 `shouldReturn` []

459485 it "drop none" $

460− (toList $ A.drop 0 $ foldMapWith f return [1..10])

486+ (toListSerial $ A.drop 0 $ foldMapWith f return [1..10])

461487 `shouldReturn` [1..10]

462488 it "drop 5" $

463− (toList $ A.drop 5 $ foldMapWith f return [1..10])

489+ (toListSerial $ A.drop 5 $ foldMapWith f return [1..10])

464490 `shouldReturn` [6..10]