9a5a5e8 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Simplify SVar types - remove the Conjunction/Disjunction tags

9a5a5e8Harendra Kumar8 years ago

src/Streamly/Core.hs+17−29

compare with master

unifiedsplit

@@ -54,8 +54,6 @@ module Streamly.Core

55 -- * Concurrent Stream Vars (SVars)

56 , SVar

57− , SVarSched (..)

58− , SVarTag (..)

59 , SVarStyle (..)

57 , SVarStyle (..)

60 , newEmptySVar

58 , newEmptySVar

61 , newStreamVar1

59 , newStreamVar1

@@ -114,23 +112,15 @@ data AheadHeapEntry a =

114-- State threaded around the monad for thread management

112-- State threaded around the monad for thread management

115------------------------------------------------------------------------------

113------------------------------------------------------------------------------

116

114

117−-- XXX this is redundant as we always start a new SVar on bind

118−-- | Conjunction is used for monadic/product style composition. Disjunction is

119−-- used for fold/sum style composition. We need to distinguish the two types of

120−-- SVars so that the scheduling of the two is independent.

121−data SVarTag = Conjunction | Disjunction deriving Eq

122−

123-- XXX use a separate data structure for each type of SVar

115-- XXX use a separate data structure for each type of SVar

124−data SVarSched =

125− LIFO -- depth first concurrent

126− | FIFO -- breadth first concurrent

127− | Par -- all parallel

128− | SVarAhead

129− deriving Eq

130−

131-- | Identify the type of the SVar. Two computations using the same style can

116-- | Identify the type of the SVar. Two computations using the same style can

132-- be scheduled on the same SVar.

117-- be scheduled on the same SVar.

133−data SVarStyle = SVarStyle SVarTag SVarSched deriving Eq

118+data SVarStyle =

119+ AsyncVar -- depth first concurrent

120+ | WAsyncVar -- breadth first concurrent

121+ | ParallelVar -- all parallel

122+ | AheadVar -- Concurrent look ahead

123+ deriving Eq

134

124

135-- | An SVar or a Stream Var is a conduit to the output from multiple streams

125-- | An SVar or a Stream Var is a conduit to the output from multiple streams

136-- running concurrently and asynchronously. An SVar can be thought of as an

126-- running concurrently and asynchronously. An SVar can be thought of as an

@@ -756,8 +746,7 @@ sendWorkerWait sv = do

756{-# NOINLINE fromStreamVar #-}

746{-# NOINLINE fromStreamVar #-}

757fromStreamVar :: MonadAsync m => SVar m a -> Stream m a

747fromStreamVar :: MonadAsync m => SVar m a -> Stream m a

758fromStreamVar sv = Stream $ \_ stp sng yld -> do

748fromStreamVar sv = Stream $ \_ stp sng yld -> do

759− let SVarStyle _ sched = svarStyle sv

749+ if svarStyle sv == ParallelVar

760− if sched == Par

761 then liftIO $ takeMVar (doorBell sv)

750 then liftIO $ takeMVar (doorBell sv)

762 else do

751 else do

763 res <- liftIO $ tryTakeMVar (doorBell sv)

752 res <- liftIO $ tryTakeMVar (doorBell sv)

@@ -906,10 +895,10 @@ newEmptySVar :: MonadAsync m => SVarStyle -> m (SVar m a)

906newEmptySVar style = do

895newEmptySVar style = do

907 liftIO $

896 liftIO $

908 case style of

897 case style of

909− SVarStyle _ FIFO -> getFifoSVar style

898+ WAsyncVar -> getFifoSVar style

910− SVarStyle _ LIFO -> getLifoSVar style

899+ AsyncVar -> getLifoSVar style

911− SVarStyle _ Par -> getParSVar style

900+ ParallelVar -> getParSVar style

912− SVarStyle _ SVarAhead -> getAheadSVar style

901+ AheadVar -> getAheadSVar style

913

902

914-- | Create a new SVar and enqueue one stream computation on it.

903-- | Create a new SVar and enqueue one stream computation on it.

915newStreamVar1 :: MonadAsync m => SVarStyle -> Stream m a -> m (SVar m a)

904newStreamVar1 :: MonadAsync m => SVarStyle -> Stream m a -> m (SVar m a)

@@ -925,7 +914,7 @@ newStreamVar1 style m = do

925-- | Create a new SVar and enqueue one stream computation on it.

914-- | Create a new SVar and enqueue one stream computation on it.

926newStreamVarAhead :: MonadAsync m => Stream m a -> m (SVar m a)

915newStreamVarAhead :: MonadAsync m => Stream m a -> m (SVar m a)

927newStreamVarAhead m = do

916newStreamVarAhead m = do

928− sv <- newEmptySVar (SVarStyle Disjunction SVarAhead)

917+ sv <- newEmptySVar AheadVar

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

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

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

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

931 -- chance to push.

920 -- chance to push.

@@ -1039,9 +1028,8 @@ joinStreamVarPar style m1 m2 = Stream $ \svr stp sng yld ->

1039{-# INLINE ahead #-}

1028{-# INLINE ahead #-}

1040ahead :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

1029ahead :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

1041ahead m1 m2 = Stream $ \svr stp sng yld -> do

1030ahead m1 m2 = Stream $ \svr stp sng yld -> do

1042− let style = SVarStyle Disjunction SVarAhead

1043 case svr of

1031 case svr of

1044− Just sv | svarStyle sv == style -> do

1032+ Just sv | svarStyle sv == AheadVar -> do

1045 liftIO $ enqueueAhead (workQueue sv) m2

1033 liftIO $ enqueueAhead (workQueue sv) m2

1046 -- Always run the left side on a new SVar to avoid complexity in

1034 -- Always run the left side on a new SVar to avoid complexity in

1047 -- sequencing results. This means the left side cannot further

1035 -- sequencing results. This means the left side cannot further

@@ -1061,15 +1049,15 @@ ahead m1 m2 = Stream $ \svr stp sng yld -> do

1061

1049

1062{-# INLINE async #-}

1050{-# INLINE async #-}

1063async :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

1051async :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

1064−async = joinStreamVar2 (SVarStyle Disjunction LIFO)

1052+async = joinStreamVar2 AsyncVar

1065

1053

1066{-# INLINE wAsync #-}

1054{-# INLINE wAsync #-}

1067wAsync :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

1055wAsync :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

1068−wAsync = joinStreamVar2 (SVarStyle Disjunction FIFO)

1056+wAsync = joinStreamVar2 WAsyncVar

1069

1057

1070{-# INLINE parallel #-}

1058{-# INLINE parallel #-}

1071parallel :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

1059parallel :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

1072−parallel = joinStreamVarPar (SVarStyle Disjunction Par)

1060+parallel = joinStreamVarPar ParallelVar

1073

1061

1074-------------------------------------------------------------------------------

1062-------------------------------------------------------------------------------

1075-- Functor instace is the same for all types

1063-- Functor instace is the same for all types

@@ -1107,7 +1095,7 @@ zipWith f m1 m2 = go m1 m2

1107 (runStream mx) Nothing stp single1 yield1

1095 (runStream mx) Nothing stp single1 yield1

1108

1096

1109mkAsync :: MonadAsync m => Stream m a -> m (Stream m a)

1097mkAsync :: MonadAsync m => Stream m a -> m (Stream m a)

1110−mkAsync m = newStreamVar1 (SVarStyle Disjunction LIFO) m

1098+mkAsync m = newStreamVar1 AsyncVar m

1111 >>= return . fromStreamVar

1099 >>= return . fromStreamVar

1112

1100

1113zipAsyncWith :: MonadAsync m

1101zipAsyncWith :: MonadAsync m

src/Streamly/Streams.hs+3−6

@@ -25,8 +25,6 @@ module Streamly.Streams

25 , S.MonadAsync

27 -- * SVars

28− , SVarSched (..)

29− , SVarTag (..)

30 , SVarStyle (..)

28 , SVarStyle (..)

31 , SVar

29 , SVar

32 , S.newEmptySVar

30 , S.newEmptySVar

@@ -121,9 +119,8 @@ import Control.Monad.Reader.Class (MonadReader(..))

121import Control.Monad.State.Class (MonadState(..))

119import Control.Monad.State.Class (MonadState(..))

122import Control.Monad.Trans.Class (MonadTrans (lift))

120import Control.Monad.Trans.Class (MonadTrans (lift))

123import Data.Semigroup (Semigroup(..))

121import Data.Semigroup (Semigroup(..))

124−import Streamly.Core ( MonadAsync

122+import Streamly.Core ( MonadAsync , SVar,

125− , SVar, SVarStyle(..)

123+ SVarStyle(..))

126− , SVarTag(..), SVarSched(..))

127import qualified Streamly.Core as S

124import qualified Streamly.Core as S

128

125

129------------------------------------------------------------------------------

126------------------------------------------------------------------------------

@@ -300,7 +297,7 @@ toSVar sv m = S.toStreamVar sv (toStream m)

300-- @since 0.2.0

297-- @since 0.2.0

301mkAsync :: (IsStream t, MonadAsync m) => t m a -> m (t m a)

298mkAsync :: (IsStream t, MonadAsync m) => t m a -> m (t m a)

302mkAsync m = do

299mkAsync m = do

303− sv <- S.newStreamVar1 (SVarStyle Disjunction LIFO) (toStream m)

300+ sv <- S.newStreamVar1 AsyncVar (toStream m)

304 return $ fromSVar sv

301 return $ fromSVar sv

305

302

306------------------------------------------------------------------------------

303------------------------------------------------------------------------------