git.iko.soy/streamly
on master help sign in
Change the order of yld and stp continuations
d35674eHarendra Kumar
src/Streamly/Internal/Data/Stream/Ahead.hs+7−7
compare with master
unifiedsplit
@@ -580,14 +580,14 @@ workLoopAhead q heap st sv winfo = do
580-- The only difference between forkSVarAsync and this is that we run the left
580-- The only difference between forkSVarAsync and this is that we run the left
581-- computation without a shared SVar.
581-- computation without a shared SVar.
582forkSVarAhead :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
582forkSVarAhead :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
583−forkSVarAhead m1 m2 = mkStream $ \st stp sng yld -> do
583+forkSVarAhead m1 m2 = mkStream $ \st yld sng stp -> do
584 sv <- newAheadVar st (concurrently (toStream m1) (toStream m2))
584 sv <- newAheadVar st (concurrently (toStream m1) (toStream m2))
585 workLoopAhead
585 workLoopAhead
586− foldStream st stp sng yld (fromSVar sv)
586+ foldStream st yld sng stp (fromSVar sv)
587 where
587 where
588− concurrently ma mb = mkStream $ \st stp sng yld -> do
588+ concurrently ma mb = mkStream $ \st yld sng stp -> do
589 liftIO $ enqueue (fromJust $ streamVar st) mb
589 liftIO $ enqueue (fromJust $ streamVar st) mb
590− foldStream st stp sng yld ma
590+ foldStream st yld sng stp ma
591
591
592-- | Polymorphic version of the 'Semigroup' operation '<>' of 'AheadT'.
592-- | Polymorphic version of the 'Semigroup' operation '<>' of 'AheadT'.
593-- Merges two streams sequentially but with concurrent lookahead.
593-- Merges two streams sequentially but with concurrent lookahead.
@@ -595,15 +595,15 @@ forkSVarAhead m1 m2 = mkStream $ \st stp sng yld -> do
595-- @since 0.3.0
595-- @since 0.3.0
596{-# INLINE ahead #-}
596{-# INLINE ahead #-}
597ahead :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
597ahead :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
598−ahead m1 m2 = mkStream $ \st stp sng yld ->
598+ahead m1 m2 = mkStream $ \st yld sng stp ->
599 case streamVar st of
599 case streamVar st of
600 Just sv | svarStyle sv == AheadVar -> do
600 Just sv | svarStyle sv == AheadVar -> do
601 liftIO $ enqueue sv (toStream m2)
601 liftIO $ enqueue sv (toStream m2)
602 -- Always run the left side on a new SVar to avoid complexity in
602 -- Always run the left side on a new SVar to avoid complexity in
603 -- sequencing results. This means the left side cannot further
603 -- sequencing results. This means the left side cannot further
604 -- split into more ahead computations on the same SVar.
604 -- split into more ahead computations on the same SVar.
605− foldStream st stp sng yld m1
605+ foldStream st yld sng stp m1
606− _ -> foldStreamShared st stp sng yld (forkSVarAhead m1 m2)
606+ _ -> foldStreamShared st yld sng stp (forkSVarAhead m1 m2)
607
607
608-- | XXX we can implement it more efficienty by directly implementing instead
608-- | XXX we can implement it more efficienty by directly implementing instead
609-- of combining streams using ahead.
609-- of combining streams using ahead.
src/Streamly/Internal/Data/Stream/Async.hs+7−7
@@ -584,26 +584,26 @@ newWAsyncVar st m = do
584
584
585forkSVarAsync :: (IsStream t, MonadAsync m)
585forkSVarAsync :: (IsStream t, MonadAsync m)
586 => SVarStyle -> t m a -> t m a -> t m a
586 => SVarStyle -> t m a -> t m a -> t m a
587−forkSVarAsync style m1 m2 = mkStream $ \st stp sng yld -> do
587+forkSVarAsync style m1 m2 = mkStream $ \st yld sng stp -> do
588 sv <- case style of
588 sv <- case style of
589 AsyncVar -> newAsyncVar st (concurrently (toStream m1) (toStream m2))
589 AsyncVar -> newAsyncVar st (concurrently (toStream m1) (toStream m2))
590 WAsyncVar -> newWAsyncVar st (concurrently (toStream m1) (toStream m2))
590 WAsyncVar -> newWAsyncVar st (concurrently (toStream m1) (toStream m2))
591 _ -> error "illegal svar type"
591 _ -> error "illegal svar type"
592− foldStream st stp sng yld $ fromSVar sv
592+ foldStream st yld sng stp $ fromSVar sv
593 where
593 where
594− concurrently ma mb = mkStream $ \st stp sng yld -> do
594+ concurrently ma mb = mkStream $ \st yld sng stp -> do
595 liftIO $ enqueue (fromJust $ streamVar st) mb
595 liftIO $ enqueue (fromJust $ streamVar st) mb
596− foldStreamShared st stp sng yld ma
596+ foldStreamShared st yld sng stp ma
597
597
598{-# INLINE joinStreamVarAsync #-}
598{-# INLINE joinStreamVarAsync #-}
599joinStreamVarAsync :: (IsStream t, MonadAsync m)
599joinStreamVarAsync :: (IsStream t, MonadAsync m)
600 => SVarStyle -> t m a -> t m a -> t m a
600 => SVarStyle -> t m a -> t m a -> t m a
601−joinStreamVarAsync style m1 m2 = mkStream $ \st stp sng yld ->
601+joinStreamVarAsync style m1 m2 = mkStream $ \st yld sng stp ->
602 case streamVar st of
602 case streamVar st of
603 Just sv | svarStyle sv == style -> do
603 Just sv | svarStyle sv == style -> do
604 liftIO $ enqueue sv (toStream m2)
604 liftIO $ enqueue sv (toStream m2)
605− foldStreamShared st stp sng yld m1
605+ foldStreamShared st yld sng stp m1
606− _ -> foldStreamShared st stp sng yld (forkSVarAsync style m1 m2)
606+ _ -> foldStreamShared st yld sng stp (forkSVarAsync style m1 m2)
607
607
608------------------------------------------------------------------------------
608------------------------------------------------------------------------------
609-- Semigroup and Monoid style compositions for parallel actions
609-- Semigroup and Monoid style compositions for parallel actions