da31ef1 · streamly · git.iko.soy

404

405-- | This function is used by the producer threads to queue output for the

406-- consumer thread to consume.

407−{-# INLINE send #-}

407+{-# NOINLINE send #-}

408send :: SVar m a -> ChildEvent a -> IO ()

409send sv msg = do

410 len <- atomicModifyIORefCAS (outputQueue sv) $ \(es, n) ->

432 (_, n) <- readIORef (outputQueue sv)

433 when (n <= 1500) $ void $ tryPutMVar (siren sv) ()

434

435−{-# INLINE sendStop #-}

435+{-# NOINLINE sendStop #-}

436sendStop :: SVar m a -> IO ()

437sendStop sv = myThreadId >>= \tid -> send sv (ChildStop tid Nothing)

438

520-- Parallel

521-------------------------------------------------------------------------------

522

523−{-# INLINE runOne #-}

523+{-# NOINLINE runOne #-}

524runOne :: MonadIO m => SVar m a -> Stream m a -> m ()

525runOne sv m = (runStream m) (Just sv) stop single yield

526

822pushWorkerPar sv m = do

823 -- We do not use activeWorkers in case of ParallelVar but still there is no

824 -- harm in maintaining it correctly.

825− liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n + 1

825+ -- liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n + 1

826 doFork (runOne sv m) (handleChildException sv) >>= modifyThread sv

827

828-- XXX When the queue is LIFO we can put a limit on the number of dispatches.

1169-- * When the stream is switching from disjunctive composition to conjunctive

1170-- composition and vice-versa we create a new SVar to isolate the scheduling

1171-- 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+

1173{-# INLINE joinStreamVarAsync #-}

1182{-# INLINE joinStreamVarAsync #-}

1174joinStreamVarAsync :: MonadAsync m

1183joinStreamVarAsync :: MonadAsync m

1175 => SVarStyle -> Stream m a -> Stream m a -> Stream m a

1184 => SVarStyle -> Stream m a -> Stream m a -> Stream m a

1177 case svr of

1186 case svr of

1178 Just sv | svarStyle sv == style ->

1187 Just sv | svarStyle sv == style ->

1179 liftIO ((enqueue sv) m2) >> (runStream m1) svr stp sng yld

1188 liftIO ((enqueue sv) m2) >> (runStream m1) svr stp sng yld

1180− _ -> do

1189+ _ -> runStream (forkSVarAsync style m1 m2) Nothing stp sng yld

1181− sv <- newStreamVar1 style (concurrently m1 m2)

1190+

1182− (runStream (fromStreamVar sv)) Nothing stp sng yld

1191+{-# NOINLINE forkSVarPar #-}

1183− where

1192+forkSVarPar :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

1184− concurrently ma mb = Stream $ \svr stp sng yld -> do

1193+forkSVarPar m r = Stream $ \_ stp sng yld -> do

1185− liftIO $ enqueue (fromJust svr) mb

1194+ sv <- newEmptySVar ParallelVar

1195+ pushWorkerPar sv m

1196+ pushWorkerPar sv r

1186− (runStream ma) svr stp sng yld

1197+ (runStream (fromStreamVar sv)) Nothing stp sng yld

1187

1198

1188{-# INLINE joinStreamVarPar #-}

1199{-# INLINE joinStreamVarPar #-}

1189joinStreamVarPar :: MonadAsync m

1200joinStreamVarPar :: MonadAsync m

1192 case svr of

1203 case svr of

1193 Just sv | svarStyle sv == style -> do

1204 Just sv | svarStyle sv == style -> do

1194 pushWorkerPar sv m1 >> (runStream m2) svr stp sng yld

1205 pushWorkerPar sv m1 >> (runStream m2) svr stp sng yld

1195− _ -> do

1206+ _ -> runStream (forkSVarPar m1 m2) Nothing stp sng yld

1196− sv <- newEmptySVar style

1207+

1197− pushWorkerPar sv m1

1208+forkSVarAhead :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

1198− pushWorkerPar sv m2

1209+forkSVarAhead m1 m2 = Stream $ \_ stp sng yld -> do

1210+ sv <- newStreamVarAhead (concurrently m1 m2)

1199− (runStream (fromStreamVar sv)) Nothing stp sng yld

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

1200

1216

1201{-# INLINE ahead #-}

1217{-# INLINE ahead #-}

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

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

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

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

1209 -- split into more ahead computations on the same SVar.

1225 -- split into more ahead computations on the same SVar.

1210 (runStream m1) Nothing stp sng yld

1226 (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

1218

1228

1219-- | XXX we can implement it more efficienty by directly implementing instead

1229-- | XXX we can implement it more efficienty by directly implementing instead

1220-- of combining streams using ahead.

1230-- of combining streams using ahead.