b2ba739 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Fix parallel binds, pass the context where needed

b2ba739Harendra Kumar9 years ago

benchmark/Main.hs+40−15

compare with master

unifiedsplit

@@ -31,30 +31,55 @@ main = do

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

32 let as = asyncly_basic

33 defaultMain [

34− bgroup "basic"

34+ bgroup "asyncly"

35+ [ bgroup "serial bind"

35− [ bench "asyncly-serial" $ nfIO (as (>>=) (A.<>))

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

36− , bench "asyncly-interleaved" $ nfIO (as (>>=) (A.<=>))

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

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

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

40+ ]

41+

42+ , bgroup "fair bind"

37− , bench "asyncly-serial-fairbind" $ nfIO (as (A.>->) (A.<>))

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

38− , bench "asyncly-interleaved-fairbind" $ nfIO (as (A.>->) (A.<=>))

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

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

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

47+ ]

48+

49+ , bgroup "parallel bind"

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

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

39− , bench "asyncly-parleft" $ nfIO (as (>>=) (A.<|))

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

40− , bench "asyncly-parinterleaved" $ nfIO (as (>>=) (A.<|>))

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

54+ ]

55+

56+ , bgroup "fair parallel bind"

41− , bench "asyncly-parleft-fairbind" $ nfIO (as (A.>->) (A.<|))

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

42− , bench "asyncly-parinterleaved-fairbind" $ nfIO (as (A.>->) (A.<|>))

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

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

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

61+ ]

62+

63+ -- Benchmark smallest possible actions composed together

64+ , bgroup "serial bind nil"

43− , bench "asyncly-serial-nil" $ nfIO (asyncly_nil (A.<>))

65+ [ bench "serial" $ nfIO (asyncly_nil (A.<>))

44− , bench "asyncly-interleaved-nil" $ nfIO (asyncly_nil (A.<=>))

66+ , bench "fair serial" $ nfIO (asyncly_nil (A.<=>))

45− , bench "asyncly-parleft-nil" $ nfIO (asyncly_nil (A.<|))

67+ , bench "left parallel" $ nfIO (asyncly_nil (A.<|))

46− , bench "asyncly-parinterleaved-nil" $ nfIO (asyncly_nil (A.<|>))

68+ , bench "fair parallel" $ nfIO (asyncly_nil (A.<|>))

69+ ]

70+ ]

47#ifdef EXTRA_BENCHMARKS

71#ifdef EXTRA_BENCHMARKS

48#if MIN_VERSION_transient(0,5,1)

72#if MIN_VERSION_transient(0,5,1)

73+ , bgroup "others"

49− , bench "transient" $ nfIO transient_basic

74+ [ bench "transient" $ nfIO transient_basic

50 , bench "transient-nil" $ nfIO transient_nil

75 , bench "transient-nil" $ nfIO transient_nil

51#endif

76#endif

52 , bench "logict" $ nfIO logict_basic

77 , bench "logict" $ nfIO logict_basic

53 , bench "list-t" $ nfIO list_t_basic

78 , bench "list-t" $ nfIO list_t_basic

54 , bench "simple-conduit" $ nfIO simple_conduit_basic

79 , bench "simple-conduit" $ nfIO simple_conduit_basic

55 , bench "machines" $ nfIO machines_basic

80 , bench "machines" $ nfIO machines_basic

56−#endif

57 ]

81 ]

82+#endif

58 ]

83 ]

60{-# INLINABLE map #-}

85{-# INLINABLE map #-}

src/Asyncly/AsyncT.hs+26−14

@@ -205,10 +205,8 @@ bindWith k m f = go m

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

206 in g Nothing stp yield

207

208−-- | '>>=' execute a monadic action for each element generated by the 'AsyncT'

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

209−-- stream, running nested loops sequentially, in a depth first manner i.e. the

210−-- inner loop finishes completely before the next iteration of the outer loop

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

211−-- can run.

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

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

213 return = pure

211 return = pure

214

212

@@ -222,22 +220,36 @@ instance Monad m => Monad (AsyncT m) where

222-- Alternative ways to bind

220-- Alternative ways to bind

223------------------------------------------------------------------------------

221------------------------------------------------------------------------------

224

222

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

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

226−-- loops sequentially, but in a fairly interleaved manner i.e. both inner and

227−-- outer loop yield alternately.

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

228(>->) :: AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b

225(>->) :: AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b

229(>->) = bindWith (<=>)

226(>->) = bindWith (<=>)

230

227

228+{-# INLINE parbind #-}

229+parbind

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

231+ -> AsyncT m a

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

233+ -> AsyncT m b

234+parbind k m f = go m

235+ where

236+ go (AsyncT g) =

237+ AsyncT $ \ctx stp yld ->

238+ let run x = (runAsyncT x) ctx stp yld

239+ yield a Nothing = run $ f a

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

241+ in g Nothing stp yield

242+

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

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

232−-- loops in parallel, but giving preference to inner loops.

244+-- iterations in parallel, but giving preference to iterations started earlier.

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

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

234−(>>|) = bindWith (parallel (CtxType Conjunction LIFO))

246+(>>|) = parbind (parallel (CtxType Conjunction LIFO))

235

247

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

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

237−-- loops in parallel, but in a fairly interleaved manner i.e. both inner and

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

238−-- outer loops yield alternately.

250+-- likely to run.

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

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

240−(>|>) = bindWith (parallel (CtxType Conjunction FIFO))

252+(>|>) = parbind (parallel (CtxType Conjunction FIFO))

241

253

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

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

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

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