e7e42cb · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Add parallel and interleaved bind operations

e7e42cbHarendra Kumar9 years ago

src/Asyncly.hs+5

compare with master

unifiedsplit

@@ -133,6 +133,11 @@ module Asyncly

133 , parLeft

134 , (<|)

135

136+ -- * Bind Operators

137+ , (>->)

138+ , (>>|)

139+ , (>|>)

140+

136 -- * Transformation

141 -- * Transformation

137 , take

142 , take

138 , drop

143 , drop

src/Asyncly/AsyncT.hs+35−2

@@ -23,12 +23,18 @@ module Asyncly.AsyncT

23 , toList

24 , take

25 , drop

26+

26 , interleave

27 , interleave

27 , (<=>)

28 , (<=>)

28 , parAhead

29 , parAhead

29 , (<>|)

30 , (<>|)

30 , parLeft

31 , parLeft

31 , (<|)

32 , (<|)

33+

34+ , (>->)

35+ , (>>|)

36+ , (>|>)

37+

32 , foldWith

38 , foldWith

33 , foldMapWith

39 , foldMapWith

34 , forEachWith

40 , forEachWith

@@ -155,17 +161,44 @@ infixr 5 <=>

155-- thread or not based on the level of the bind in the tree. Maybe at some

161-- thread or not based on the level of the bind in the tree. Maybe at some

156-- point we shall be able to do that?

162-- point we shall be able to do that?

157

163

164+-- XXX Using a common bindWith function is 15% slower than duplicating the

165+-- code. In case that performance is important we can duplicate it.

166+--

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

168+-- reset the context to Nothing.

169+{-# INLINE bindWith #-}

170+bindWith

171+ :: (forall c. AsyncT m c -> AsyncT m c -> AsyncT m c)

172+ -> AsyncT m a

173+ -> (a -> AsyncT m b)

174+ -> AsyncT m b

175+bindWith k (AsyncT m) f = AsyncT $ \_ stp yld ->

176+ let run x = (runAsyncT x) Nothing stp yld

177+ yield a _ Nothing = run $ f a

178+ yield a _ (Just r) = run $ f a `k` (bindWith k r f)

179+ in m Nothing stp yield

180+

158instance Monad m => Monad (AsyncT m) where

181instance Monad m => Monad (AsyncT m) where

159 return a = AsyncT $ \ctx _ yld -> yld a ctx Nothing

182 return a = AsyncT $ \ctx _ yld -> yld a ctx Nothing

160

183

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

162− -- reset the context to Nothing.

163 AsyncT m >>= f = AsyncT $ \_ stp yld ->

184 AsyncT m >>= f = AsyncT $ \_ stp yld ->

164 let run x = (runAsyncT x) Nothing stp yld

185 let run x = (runAsyncT x) Nothing stp yld

165 yield a _ Nothing = run $ f a

186 yield a _ Nothing = run $ f a

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

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

167 in m Nothing stp yield

188 in m Nothing stp yield

168

189

190+-- | Serial, interleaved bind

191+(>->) :: AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b

192+(>->) = bindWith (<=>)

193+

194+-- | Parallel, depth first bind

195+(>>|) :: MonadAsync m => AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b

196+(>>|) = bindWith (<|)

197+

198+-- | Parallel, breadth first bind

199+(>|>) :: MonadAsync m => AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b

200+(>|>) = bindWith (<|>)

201+

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

202------------------------------------------------------------------------------

170-- Functor

203-- Functor

171------------------------------------------------------------------------------

204------------------------------------------------------------------------------

test/Main.hs+128−42

@@ -1,6 +1,7 @@

1module Main (main) where

3import Control.Concurrent (threadDelay)

4+import Data.Foldable (forM_)

4import Data.List (sort)

5import Data.List (sort)

5import Prelude hiding (take, drop)

6import Prelude hiding (take, drop)

6import Test.Hspec

7import Test.Hspec

@@ -25,32 +26,74 @@ main = hspec $ do

25 it "MonadPlus - mzero" $

26 it "MonadPlus - mzero" $

26 (toList mzero) `shouldReturn` ([] :: [Int])

27 (toList mzero) `shouldReturn` ([] :: [Int])

29+ ---------------------------------------------------------------------------

30+ -- Binds

31+ ---------------------------------------------------------------------------

32+

28− describe "Bind" bind

33+ describe "Bind then" thenBind

34+ describe "Pure bind serial" $ pureBind (>>=)

35+ describe "Pure bind serial interleaved" $ pureBind (>->)

36+ describe "Pure bind parallel DFS" $ pureBind (>>|)

37+ describe "Pure bind parallel BFS" $ pureBind (>|>)

38+

39+ ---------------------------------------------------------------------------

40+ -- Compositions

41+ ---------------------------------------------------------------------------

30 describe "Serial Composition (<>)" $ compose (<>) id

43 describe "Serial Composition (<>)" $ compose (<>) id

31− describe "Serial interleaved (<=>)" $ interleaved (<=>) id

32 describe "Left biased parallel Composition (<|)" $ compose (<|) sort

44 describe "Left biased parallel Composition (<|)" $ compose (<|) sort

33 describe "Fair parallel Composition (<|>)" $ compose (<|>) sort

45 describe "Fair parallel Composition (<|>)" $ compose (<|>) sort

46+

47+ ---------------------------------------------------------------------------

48+ -- Composition ordering checks

49+ ---------------------------------------------------------------------------

50+

51+ describe "Serial interleaved (<=>)" $ interleaved (<=>) id

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

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

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

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

36 describe "Parallel interleaved (<|>)" $ interleaved (<|>) reverse

54 describe "Parallel interleaved (<|>)" $ interleaved (<|>) reverse

56+ ---------------------------------------------------------------------------

57+ -- Composition recursion loops

58+ ---------------------------------------------------------------------------

59+

38 describe "Serial loops (<>)" $ loops (<>) id reverse

60 describe "Serial loops (<>)" $ loops (<>) id reverse

39 describe "Left biased parallel loops (<|)" $ loops (<|) sort sort

61 describe "Left biased parallel loops (<|)" $ loops (<|) sort sort

40 describe "Fair parallel loops (<|>)" $ loops (<|>) sort sort

62 describe "Fair parallel loops (<|>)" $ loops (<|>) sort sort

64+ ---------------------------------------------------------------------------

65+ -- Bind and Composition combinations

66+ ---------------------------------------------------------------------------

67+

68+ forM_ [(>>=), (>->), (>>|), (>|>)] $ \f ->

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

70+ describe "Bind and compose" $ bindAndComposeSimple f g

71+

72+ let fldr f = foldr f empty

73+ fldl f = foldl f empty

74+ in forM_ [(>>=), (>->), (>>|), (>|>)] $ \f ->

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

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

77+ describe "Bind and compose" $

78+ bindAndComposeHierarchy f (k g)

79+

80+ -- TBD combine all binds and all compose in one example

81+ describe "Miscellaneous combined examples" mixedOps

82+

42 describe "Transformation" $ transformOps (<>)

83 describe "Transformation" $ transformOps (<>)

43− describe "Miscellaneouse combined examples" mixedOps

45−bind :: Spec

85+thenBind :: Spec

46−bind = do

86+thenBind = do

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

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

48 runAsyncly (liftIO (putStrLn "hello") >> liftIO (putStrLn "world"))

88 runAsyncly (liftIO (putStrLn "hello") >> liftIO (putStrLn "world"))

49 `shouldReturn` ()

89 `shouldReturn` ()

50 it "Then and toList" $

90 it "Then and toList" $

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

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

92+

93+pureBind :: (AsyncT IO Int -> (Int -> AsyncT IO Int) -> AsyncT IO Int) -> Spec

94+pureBind f = do

52 it "Bind and toList" $

95 it "Bind and toList" $

53− toList (do x <- return 1; y <- return 2; return (x + y))

96+ toList (return 1 `f` \x -> return 2 `f` \y -> return (x + y))

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

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

56interleaved

99interleaved

@@ -102,13 +145,6 @@ compose f srt = do

102 it "Compose many" $

145 it "Compose many" $

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

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

104 `shouldReturn` [1..100]

147 `shouldReturn` [1..100]

105− it "Compose many (right fold) with bind" $

106− (toList (forEachWith f [1..10 :: Int] $ \x -> return x >>= return . id)

107− >>= return . srt) `shouldReturn` [1..10]

108− it "Compose many (left fold) with bind" $

109− let forL xs k = foldl f empty $ map k xs

110− in (toList (forL [1..10 :: Int] $ \x -> return x >>= return . id)

111− >>= return . srt) `shouldReturn` [1..10]

112 it "Compose hierarchical (multiple levels)" $

148 it "Compose hierarchical (multiple levels)" $

113 ((toList $ (((return 0 `f` return 1) `f` (return 2 `f` return 3))

149 ((toList $ (((return 0 `f` return 1) `f` (return 2 `f` return 3))

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

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

@@ -137,6 +173,85 @@ loops f tsrt hsrt = do

137 liftIO $ putStrLn "LoopTail..."

173 liftIO $ putStrLn "LoopTail..."

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

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

139

175

176+bindAndComposeSimple

177+ :: (AsyncT IO Int -> (Int -> AsyncT IO Int) -> AsyncT IO Int)

178+ -> (AsyncT IO Int -> AsyncT IO Int -> AsyncT IO Int)

179+ -> Spec

180+bindAndComposeSimple f g = do

181+ it "Compose many (right fold) with bind" $

182+ (toList (forEachWith g [1..10 :: Int] $ \x -> return x `f` (return . id))

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

184+

185+ it "Compose many (left fold) with bind" $

186+ let forL xs k = foldl g empty $ map k xs

187+ in (toList (forL [1..10 :: Int] $ \x -> return x `f` (return . id))

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

189+

190+bindAndComposeHierarchy

191+ :: (AsyncT IO Int -> (Int -> AsyncT IO Int) -> AsyncT IO Int)

192+ -> ([AsyncT IO Int] -> AsyncT IO Int)

193+ -> Spec

194+bindAndComposeHierarchy f g = do

195+ it "Bind and compose nested" $

196+ (toList bindComposeNested >>= return . sort)

197+ `shouldReturn` (sort (

198+ [12, 18]

199+ ++ replicate 3 13

200+ ++ replicate 3 17

201+ ++ replicate 6 14

202+ ++ replicate 6 16

203+ ++ replicate 7 15) :: [Int])

204+

205+ where

206+

207+ bindComposeNested :: AsyncT IO Int

208+ bindComposeNested =

209+ let c1 = tripleCompose (return 1) (return 2) (return 3)

210+ c2 = tripleCompose (return 4) (return 5) (return 6)

211+ c3 = tripleCompose (return 7) (return 8) (return 9)

212+ b = tripleBind c1 c2 c3

213+-- it seems to be causing a huge space leak in hspec so disabling this for now

214+-- c = tripleCompose b b b

215+-- m = tripleBind c c c

216+-- in m

217+ in b

218+

219+ tripleCompose a b c = g [a, b, c]

220+ tripleBind mx my mz =

221+ mx `f` \x -> my

222+ `f` \y -> mz

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

224+

225+mixedOps :: Spec

226+mixedOps = do

227+ it "Compose many ops" $

228+ (toList composeMixed >>= return . sort)

229+ `shouldReturn` ([8,9,9,9,9,9,10,10,10,10,10,10,10,10,10,10,11,11

230+ ,11,11,11,11,11,11,11,11,12,12,12,12,12,13

231+ ] :: [Int])

232+ where

233+

234+ composeMixed :: AsyncT IO Int

235+ composeMixed = do

236+ liftIO $ return ()

237+ liftIO $ putStr ""

238+ x <- return 1

239+ y <- return 2

240+ z <- do

241+ x1 <- return 1 <|> return 2

242+ liftIO $ return ()

243+ liftIO $ putStr ""

244+ y1 <- return 1 <| return 2

245+ z1 <- do

246+ x11 <- return 1 <> return 2

247+ y11 <- return 1 <| return 2

248+ z11 <- return 1 <=> return 2

249+ liftIO $ return ()

250+ liftIO $ putStr ""

251+ return (x11 + y11 + z11)

252+ return (x1 + y1 + z1)

253+ return (x + y + z)

254+

140transformOps :: (AsyncT IO Int -> AsyncT IO Int -> AsyncT IO Int) -> Spec

255transformOps :: (AsyncT IO Int -> AsyncT IO Int -> AsyncT IO Int) -> Spec

141transformOps f = do

256transformOps f = do

142 it "take all" $

257 it "take all" $

@@ -158,32 +273,3 @@ transformOps f = do

158 it "drop 5" $

273 it "drop 5" $

159 (toList $ drop 5 $ foldMapWith f return [1..10])

274 (toList $ drop 5 $ foldMapWith f return [1..10])

160 `shouldReturn` [6..10]

275 `shouldReturn` [6..10]

161−

162−mixedOps :: Spec

163−mixedOps = do

164− it "Compose many ops" $

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

166− `shouldReturn` ([8,9,9,9,9,9,10,10,10,10,10,10,10,10,10,10,11,11

167− ,11,11,11,11,11,11,11,11,12,12,12,12,12,13

168− ] :: [Int])

169−

170−composeMixed :: AsyncT IO Int

171−composeMixed = do

172− liftIO $ return ()

173− liftIO $ putStr ""

174− x <- return 1

175− y <- return 2

176− z <- do

177− x1 <- return 1 <|> return 2

178− liftIO $ return ()

179− liftIO $ putStr ""

180− y1 <- return 1 <| return 2

181− z1 <- do

182− x11 <- return 1 <> return 2

183− y11 <- return 1 <| return 2

184− z11 <- return 1 <=> return 2

185− liftIO $ return ()

186− liftIO $ putStr ""

187− return (x11 + y11 + z11)

188− return (x1 + y1 + z1)

189− return (x + y + z)