@@ -54,8 +54,6 @@ module Streamly.Core
5454
5555 -- * Concurrent Stream Vars (SVars)
5656 , SVar
57− , SVarSched (..)
58− , SVarTag (..)
5957 , SVarStyle (..)
6058 , newEmptySVar
6159 , newStreamVar1
@@ -114,23 +112,15 @@ data AheadHeapEntry a =
114112-- State threaded around the monad for thread management
115113------------------------------------------------------------------------------
116114
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−
123115-- 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−
131116-- | Identify the type of the SVar. Two computations using the same style can
132117-- 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
134124
135125-- | An SVar or a Stream Var is a conduit to the output from multiple streams
136126-- running concurrently and asynchronously. An SVar can be thought of as an
@@ -756,8 +746,7 @@ sendWorkerWait sv = do
756746{-# NOINLINE fromStreamVar #-}
757747fromStreamVar :: MonadAsync m => SVar m a -> Stream m a
758748fromStreamVar sv = Stream $ \_ stp sng yld -> do
759− let SVarStyle _ sched = svarStyle sv
760− if sched == Par
749+ if svarStyle sv == ParallelVar
761750 then liftIO $ takeMVar (doorBell sv)
762751 else do
763752 res <- liftIO $ tryTakeMVar (doorBell sv)
@@ -906,10 +895,10 @@ newEmptySVar :: MonadAsync m => SVarStyle -> m (SVar m a)
906895newEmptySVar style = do
907896 liftIO $
908897 case style of
909− SVarStyle _ FIFO -> getFifoSVar style
910− SVarStyle _ LIFO -> getLifoSVar style
911− SVarStyle _ Par -> getParSVar style
912− SVarStyle _ SVarAhead -> getAheadSVar style
898+ WAsyncVar -> getFifoSVar style
899+ AsyncVar -> getLifoSVar style
900+ ParallelVar -> getParSVar style
901+ AheadVar -> getAheadSVar style
913902
914903-- | Create a new SVar and enqueue one stream computation on it.
915904newStreamVar1 :: MonadAsync m => SVarStyle -> Stream m a -> m (SVar m a)
@@ -925,7 +914,7 @@ newStreamVar1 style m = do
925914-- | Create a new SVar and enqueue one stream computation on it.
926915newStreamVarAhead :: MonadAsync m => Stream m a -> m (SVar m a)
927916newStreamVarAhead m = do
928− sv <- newEmptySVar (SVarStyle Disjunction SVarAhead)
917+ sv <- newEmptySVar AheadVar
929918 -- Note: We must have all the work on the queue before sending the
930919 -- pushworker, otherwise the pushworker may exit before we even get a
931920 -- chance to push.
@@ -1039,9 +1028,8 @@ joinStreamVarPar style m1 m2 = Stream $ \svr stp sng yld ->
10391028{-# INLINE ahead #-}
10401029ahead :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
10411030ahead m1 m2 = Stream $ \svr stp sng yld -> do
1042− let style = SVarStyle Disjunction SVarAhead
10431031 case svr of
1044− Just sv | svarStyle sv == style -> do
1032+ Just sv | svarStyle sv == AheadVar -> do
10451033 liftIO $ enqueueAhead (workQueue sv) m2
10461034 -- Always run the left side on a new SVar to avoid complexity in
10471035 -- sequencing results. This means the left side cannot further
@@ -1061,15 +1049,15 @@ ahead m1 m2 = Stream $ \svr stp sng yld -> do
10611049
10621050{-# INLINE async #-}
10631051async :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1064−async = joinStreamVar2 (SVarStyle Disjunction LIFO)
1052+async = joinStreamVar2 AsyncVar
10651053
10661054{-# INLINE wAsync #-}
10671055wAsync :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1068−wAsync = joinStreamVar2 (SVarStyle Disjunction FIFO)
1056+wAsync = joinStreamVar2 WAsyncVar
10691057
10701058{-# INLINE parallel #-}
10711059parallel :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1072−parallel = joinStreamVarPar (SVarStyle Disjunction Par)
1060+parallel = joinStreamVarPar ParallelVar
10731061
10741062-------------------------------------------------------------------------------
10751063-- Functor instace is the same for all types
@@ -1107,7 +1095,7 @@ zipWith f m1 m2 = go m1 m2
11071095 (runStream mx) Nothing stp single1 yield1
11081096
11091097mkAsync :: MonadAsync m => Stream m a -> m (Stream m a)
1110−mkAsync m = newStreamVar1 (SVarStyle Disjunction LIFO) m
1098+mkAsync m = newStreamVar1 AsyncVar m
11111099 >>= return . fromStreamVar
11121100
11131101zipAsyncWith :: MonadAsync m