@@ -404,7 +404,7 @@ doFork action exHandler =
404404
405405-- | This function is used by the producer threads to queue output for the
406406-- consumer thread to consume.
407−{-# INLINE send #-}
407+{-# NOINLINE send #-}
408408send :: SVar m a -> ChildEvent a -> IO ()
409409send sv msg = do
410410 len <- atomicModifyIORefCAS (outputQueue sv) $ \(es, n) ->
@@ -432,7 +432,7 @@ send sv msg = do
432432 (_, n) <- readIORef (outputQueue sv)
433433 when (n <= 1500) $ void $ tryPutMVar (siren sv) ()
434434
435−{-# INLINE sendStop #-}
435+{-# NOINLINE sendStop #-}
436436sendStop :: SVar m a -> IO ()
437437sendStop sv = myThreadId >>= \tid -> send sv (ChildStop tid Nothing)
438438
@@ -520,7 +520,7 @@ runqueueFIFO sv q = run
520520-- Parallel
521521-------------------------------------------------------------------------------
522522
523−{-# INLINE runOne #-}
523+{-# NOINLINE runOne #-}
524524runOne :: MonadIO m => SVar m a -> Stream m a -> m ()
525525runOne sv m = (runStream m) (Just sv) stop single yield
526526
@@ -822,7 +822,7 @@ pushWorkerPar :: MonadAsync m => SVar m a -> Stream m a -> m ()
822822pushWorkerPar sv m = do
823823 -- We do not use activeWorkers in case of ParallelVar but still there is no
824824 -- harm in maintaining it correctly.
825− liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n + 1
825+ -- liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n + 1
826826 doFork (runOne sv m) (handleChildException sv) >>= modifyThread sv
827827
828828-- XXX When the queue is LIFO we can put a limit on the number of dispatches.
@@ -1169,7 +1169,16 @@ toStreamVar sv m = do
11691169-- * When the stream is switching from disjunctive composition to conjunctive
11701170-- composition and vice-versa we create a new SVar to isolate the scheduling
11711171-- of the two.
1172−--
1172+
1173+forkSVarAsync :: MonadAsync m => SVarStyle -> Stream m a -> Stream m a -> Stream m a
1174+forkSVarAsync style m1 m2 = Stream $ \_ stp sng yld -> do
1175+ sv <- newStreamVar1 style (concurrently m1 m2)
1176+ (runStream (fromStreamVar sv)) Nothing stp sng yld
1177+ where
1178+ concurrently ma mb = Stream $ \svr stp sng yld -> do
1179+ liftIO $ enqueue (fromJust svr) mb
1180+ (runStream ma) svr stp sng yld
1181+
11731182{-# INLINE joinStreamVarAsync #-}
11741183joinStreamVarAsync :: MonadAsync m
11751184 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
@@ -1177,13 +1186,15 @@ joinStreamVarAsync style m1 m2 = Stream $ \svr stp sng yld ->
11771186 case svr of
11781187 Just sv | svarStyle sv == style ->
11791188 liftIO ((enqueue sv) m2) >> (runStream m1) svr stp sng yld
1180− _ -> do
1181− sv <- newStreamVar1 style (concurrently m1 m2)
1182− (runStream (fromStreamVar sv)) Nothing stp sng yld
1183− where
1184− concurrently ma mb = Stream $ \svr stp sng yld -> do
1185− liftIO $ enqueue (fromJust svr) mb
1186− (runStream ma) svr stp sng yld
1189+ _ -> runStream (forkSVarAsync style m1 m2) Nothing stp sng yld
1190+
1191+{-# NOINLINE forkSVarPar #-}
1192+forkSVarPar :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1193+forkSVarPar m r = Stream $ \_ stp sng yld -> do
1194+ sv <- newEmptySVar ParallelVar
1195+ pushWorkerPar sv m
1196+ pushWorkerPar sv r
1197+ (runStream (fromStreamVar sv)) Nothing stp sng yld
11871198
11881199{-# INLINE joinStreamVarPar #-}
11891200joinStreamVarPar :: MonadAsync m
@@ -1192,11 +1203,16 @@ joinStreamVarPar style m1 m2 = Stream $ \svr stp sng yld ->
11921203 case svr of
11931204 Just sv | svarStyle sv == style -> do
11941205 pushWorkerPar sv m1 >> (runStream m2) svr stp sng yld
1195− _ -> do
1196− sv <- newEmptySVar style
1197− pushWorkerPar sv m1
1198− pushWorkerPar sv m2
1199− (runStream (fromStreamVar sv)) Nothing stp sng yld
1206+ _ -> runStream (forkSVarPar m1 m2) Nothing stp sng yld
1207+
1208+forkSVarAhead :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1209+forkSVarAhead m1 m2 = Stream $ \_ stp sng yld -> do
1210+ sv <- newStreamVarAhead (concurrently m1 m2)
1211+ (runStream (fromStreamVar sv)) Nothing stp sng yld
1212+ where
1213+ concurrently ma mb = Stream $ \svr stp sng yld -> do
1214+ liftIO $ enqueueAhead (fromJust svr) (workQueue (fromJust svr)) mb
1215+ (runStream ma) Nothing stp sng yld
12001216
12011217{-# INLINE ahead #-}
12021218ahead :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
@@ -1208,13 +1224,7 @@ ahead m1 m2 = Stream $ \svr stp sng yld -> do
12081224 -- sequencing results. This means the left side cannot further
12091225 -- split into more ahead computations on the same SVar.
12101226 (runStream m1) Nothing stp sng yld
1211− _ -> do
1212− sv <- newStreamVarAhead (concurrently m1 m2)
1213− (runStream (fromStreamVar sv)) Nothing stp sng yld
1214− where
1215− concurrently ma mb = Stream $ \svr stp sng yld -> do
1216− liftIO $ enqueueAhead (fromJust svr) (workQueue (fromJust svr)) mb
1217− (runStream ma) Nothing stp sng yld
1227+ _ -> runStream (forkSVarAhead m1 m2) Nothing stp sng yld
12181228
12191229-- | XXX we can implement it more efficienty by directly implementing instead
12201230-- of combining streams using ahead.