b2ba739 · streamly · git.iko.soy

git.iko.soy/streamly

Fix parallel binds, pass the context where needed

commitb2ba739bcc83195fe928a8a4c28f36472f35b71b authorHarendra Kumar <harendra.kumar@gmail.com> dateFri 22 Sep 2017, 10:33 (9 years ago) parentd1cb4d8

2 files changed+66−29

compare with master

unifiedsplit

benchmark/Main.hs+40−15 src/Asyncly/AsyncT.hs+26−14

benchmark/Main.hs+40−15

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

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

3232 let as = asyncly_basic

3333 defaultMain [

34− bgroup "basic"

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

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

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

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

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

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

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

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

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

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

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

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

34+ bgroup "asyncly"

35+ [ bgroup "serial bind"

36+ [ bench "serial" $ 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"

43+ [ bench "serial" $ 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.<=>))

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

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

54+ ]

55+

56+ , bgroup "fair parallel bind"

57+ [ bench "serial" $ 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"

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

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

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

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

69+ ]

70+ ]

4771#ifdef EXTRA_BENCHMARKS

4872#if MIN_VERSION_transient(0,5,1)

49− , bench "transient" $ nfIO transient_basic

73+ , bgroup "others"

74+ [ bench "transient" $ nfIO transient_basic

5075 , bench "transient-nil" $ nfIO transient_nil

5176#endif

5277 , bench "logict" $ nfIO logict_basic

5378 , bench "list-t" $ nfIO list_t_basic

5479 , bench "simple-conduit" $ nfIO simple_conduit_basic

5580 , bench "machines" $ nfIO machines_basic

56−#endif

5781 ]

82+#endif

5883 ]

5984

6085{-# INLINABLE map #-}

src/Asyncly/AsyncT.hs+26−14

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

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

206206 in g Nothing stp yield

207207

208−-- | '>>=' execute a monadic action for each element generated by 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

211−-- can run.

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

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

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

213211 return = pure

214212

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

222220-- Alternative ways to bind

223221------------------------------------------------------------------------------

224222

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

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

227−-- outer loop yield alternately.

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

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

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

229226(>->) = bindWith (<=>)

230227

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

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

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+

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

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

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

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

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

235247

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

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

238−-- outer loops yield alternately.

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

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

250+-- likely to run.

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

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

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

241253

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

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