@@ -304,8 +304,8 @@ where
304304
305305import Control.Concurrent (killThread, myThreadId, takeMVar, threadDelay)
306306import Control.Exception
307− (assert, Exception, SomeException, AsyncException, fromException, mask_)
308−import Control.Monad (void, when, forever)
307+ (assert, Exception, SomeException, AsyncException, mask_)
308+import Control.Monad (void, when)
309309import Control.Monad.Catch (MonadCatch, MonadThrow, throwM)
310310import Control.Monad.IO.Class (MonadIO(..))
311311import Control.Monad.Trans.Reader (ReaderT)
@@ -314,26 +314,21 @@ import Control.Monad.Trans.Class (MonadTrans(lift))
314314import Control.Monad.Trans.Control (MonadBaseControl, liftBaseOp_, control)
315315import Data.Bits (shiftR, shiftL, (.|.), (.&.))
316316import Data.Functor.Identity (Identity(..))
317−import Data.Int (Int64)
318317import Data.IORef (newIORef, readIORef, mkWeakIORef, writeIORef, IORef)
319−import Data.Maybe (fromJust, isJust, isNothing)
318+import Data.Maybe (fromJust, isJust)
320319import Data.Word (Word32)
321320import Foreign.Storable (Storable(..))
322321import GHC.Exts (SpecConstrAnnotation(..))
323322import GHC.Types (SPEC(..))
324−import System.Mem (performMajorGC)
325323import Fusion.Plugin.Types (Fuse(..))
326−import Streamly.Internal.Data.IORef.Prim (Prim)
327324import Streamly.Internal.Data.Time.Units
328− (TimeUnit64, toRelTime64, diffAbsTime64, RelTime64)
325+ (TimeUnit64, toRelTime64, diffAbsTime64)
329326import Streamly.Internal.Data.Atomics (atomicModifyIORefCAS_)
330327import Streamly.Internal.Data.Array.Storable.Foreign.Types (Array(..))
331328import Streamly.Internal.Data.Fold.Types (Fold(..))
332329import Streamly.Internal.Data.Parser (ParseError(..))
333330import Streamly.Internal.Data.Pipe.Types (Pipe(..), PipeState(..))
334331import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
335−import Streamly.Internal.Data.Time.Units
336− (MicroSecond64(..), fromAbsTime, toAbsTime, AbsTime)
337332import Streamly.Internal.Data.Unfold.Types (Unfold(..))
338333import Streamly.Internal.Data.Stream.SVar (fromConsumer, pushToFold)
339334
@@ -357,6 +352,7 @@ import Prelude hiding
357352 , (!!), scanl, scanl1, concatMap, replicate, enumFromTo, concat
358353 , reverse, iterate, splitAt)
359354import Streamly.Internal.Data.Stream.StreamD.Type
355+import Streamly.Internal.Data.Stream.StreamD.Generate
360356import Streamly.Internal.Data.SVar
361357
362358------------------------------------------------------------------------------
@@ -421,89 +417,6 @@ uncons (UnStream step state) = go state
421417 Skip s -> go s
422418 Stop -> return Nothing
423419
424−------------------------------------------------------------------------------
425−-- Generation by unfold
426−------------------------------------------------------------------------------
427−
428−{-# INLINE_NORMAL unfoldrM #-}
429−unfoldrM :: Monad m => (s -> m (Maybe (a, s))) -> s -> Stream m a
430−unfoldrM next state = Stream step state
431− where
432− {-# INLINE_LATE step #-}
433− step _ st = do
434− r <- next st
435− return $ case r of
436− Just (x, s) -> Yield x s
437− Nothing -> Stop
438−
439−{-# INLINE_LATE unfoldr #-}
440−unfoldr :: Monad m => (s -> Maybe (a, s)) -> s -> Stream m a
441−unfoldr f = unfoldrM (return . f)
442−
443−data UnfoldState s = UnfoldNothing | UnfoldJust s
444−
445−-- | Convert an 'Unfold' into a 'Stream' by supplying it a seed.
446−--
447−{-# INLINE_NORMAL unfold #-}
448−unfold :: Monad m => Unfold m a b -> a -> Stream m b
449−unfold (Unfold ustep inject) seed = Stream step UnfoldNothing
450− where
451− {-# INLINE_LATE step #-}
452− step _ UnfoldNothing = inject seed >>= return . Skip . UnfoldJust
453− step _ (UnfoldJust st) = do
454− r <- ustep st
455− return $ case r of
456− Yield x s -> Yield x (UnfoldJust s)
457− Skip s -> Skip (UnfoldJust s)
458− Stop -> Stop
459−
460−------------------------------------------------------------------------------
461−-- Specialized Generation
462−------------------------------------------------------------------------------
463−
464−{-# INLINE_NORMAL repeatM #-}
465−repeatM :: Monad m => m a -> Stream m a
466−repeatM x = Stream (\_ _ -> x >>= \r -> return $ Yield r ()) ()
467−
468−{-# INLINE_NORMAL repeat #-}
469−repeat :: Monad m => a -> Stream m a
470−repeat x = Stream (\_ _ -> return $ Yield x ()) ()
471−
472−{-# INLINE_NORMAL iterateM #-}
473−iterateM :: Monad m => (a -> m a) -> m a -> Stream m a
474−iterateM step = Stream (\_ st -> st >>= \x -> return $ Yield x (step x))
475−
476−{-# INLINE_NORMAL iterate #-}
477−iterate :: Monad m => (a -> a) -> a -> Stream m a
478−iterate step st = iterateM (return . step) (return st)
479−
480−{-# INLINE_NORMAL replicateM #-}
481−replicateM :: forall m a. Monad m => Int -> m a -> Stream m a
482−replicateM n p = Stream step n
483− where
484− {-# INLINE_LATE step #-}
485− step _ (i :: Int)
486− | i <= 0 = return Stop
487− | otherwise = do
488− x <- p
489− return $ Yield x (i - 1)
490−
491−{-# INLINE_NORMAL replicate #-}
492−replicate :: Monad m => Int -> a -> Stream m a
493−replicate n x = replicateM n (return x)
494−
495−-- This would not work properly for floats, therefore we put an Integral
496−-- constraint.
497−-- | Can be used to enumerate unbounded integrals. This does not check for
498−-- overflow or underflow for bounded integrals.
499−{-# INLINE_NORMAL enumerateFromStepIntegral #-}
500−enumerateFromStepIntegral :: (Integral a, Monad m) => a -> a -> Stream m a
501−enumerateFromStepIntegral from stride =
502− from `seq` stride `seq` Stream step from
503− where
504− {-# INLINE_LATE step #-}
505− step _ !x = return $ Yield x $! (x + stride)
506−
507420-- We are assuming that "to" is constrained by the type to be within
508421-- max/min bounds.
509422{-# INLINE enumerateFromToIntegral #-}
@@ -515,100 +428,6 @@ enumerateFromToIntegral from to =
515428enumerateFromIntegral :: (Monad m, Integral a, Bounded a) => a -> Stream m a
516429enumerateFromIntegral from = enumerateFromToIntegral from maxBound
517430
518−data EnumState a = EnumInit | EnumYield a a a | EnumStop
519−
520−{-# INLINE_NORMAL enumerateFromThenToIntegralUp #-}
521−enumerateFromThenToIntegralUp
522− :: (Monad m, Integral a)
523− => a -> a -> a -> Stream m a
524−enumerateFromThenToIntegralUp from next to = Stream step EnumInit
525− where
526− {-# INLINE_LATE step #-}
527− step _ EnumInit =
528− return $
529− if to < next
530− then if to < from
531− then Stop
532− else Yield from EnumStop
533− else -- from <= next <= to
534− let stride = next - from
535− in Skip $ EnumYield from stride (to - stride)
536−
537− step _ (EnumYield x stride toMinus) =
538− return $
539− if x > toMinus
540− then Yield x EnumStop
541− else Yield x $ EnumYield (x + stride) stride toMinus
542−
543− step _ EnumStop = return Stop
544−
545−{-# INLINE_NORMAL enumerateFromThenToIntegralDn #-}
546−enumerateFromThenToIntegralDn
547− :: (Monad m, Integral a)
548− => a -> a -> a -> Stream m a
549−enumerateFromThenToIntegralDn from next to = Stream step EnumInit
550− where
551− {-# INLINE_LATE step #-}
552− step _ EnumInit =
553− return $ if to > next
554− then if to > from
555− then Stop
556− else Yield from EnumStop
557− else -- from >= next >= to
558− let stride = next - from
559− in Skip $ EnumYield from stride (to - stride)
560−
561− step _ (EnumYield x stride toMinus) =
562− return $
563− if x < toMinus
564− then Yield x EnumStop
565− else Yield x $ EnumYield (x + stride) stride toMinus
566−
567− step _ EnumStop = return Stop
568−
569−{-# INLINE_NORMAL enumerateFromThenToIntegral #-}
570−enumerateFromThenToIntegral
571− :: (Monad m, Integral a)
572− => a -> a -> a -> Stream m a
573−enumerateFromThenToIntegral from next to
574− | next >= from = enumerateFromThenToIntegralUp from next to
575− | otherwise = enumerateFromThenToIntegralDn from next to
576−
577−{-# INLINE_NORMAL enumerateFromThenIntegral #-}
578−enumerateFromThenIntegral
579− :: (Monad m, Integral a, Bounded a)
580− => a -> a -> Stream m a
581−enumerateFromThenIntegral from next =
582− if next > from
583− then enumerateFromThenToIntegralUp from next maxBound
584− else enumerateFromThenToIntegralDn from next minBound
585−
586−-- For floating point numbers if the increment is less than the precision then
587−-- it just gets lost. Therefore we cannot always increment it correctly by just
588−-- repeated addition.
589−-- 9007199254740992 + 1 + 1 :: Double => 9.007199254740992e15
590−-- 9007199254740992 + 2 :: Double => 9.007199254740994e15
591−
592−-- Instead we accumulate the increment counter and compute the increment
593−-- every time before adding it to the starting number.
594−--
595−-- This works for Integrals as well as floating point numbers, but
596−-- enumerateFromStepIntegral is faster for integrals.
597−{-# INLINE_NORMAL enumerateFromStepNum #-}
598−enumerateFromStepNum :: (Monad m, Num a) => a -> a -> Stream m a
599−enumerateFromStepNum from stride = Stream step 0
600− where
601− {-# INLINE_LATE step #-}
602− step _ !i = return $ (Yield $! (from + i * stride)) $! (i + 1)
603−
604−{-# INLINE_NORMAL numFrom #-}
605−numFrom :: (Monad m, Num a) => a -> Stream m a
606−numFrom from = enumerateFromStepNum from 1
607−
608−{-# INLINE_NORMAL numFromThen #-}
609−numFromThen :: (Monad m, Num a) => a -> a -> Stream m a
610−numFromThen from next = enumerateFromStepNum from (next - from)
611−
612431-- We cannot write a general function for Num. The only way to write code
613432-- portable between the two is to use a 'Real' constraint and convert between
614433-- Fractional and Integral using fromRational which is horribly slow.
@@ -630,181 +449,6 @@ enumerateFromThenToFractional from next to =
630449 predicate | next >= from = (<= to + mid)
631450 | otherwise = (>= to + mid)
632451
633−-------------------------------------------------------------------------------
634−-- Generation by Conversion
635−-------------------------------------------------------------------------------
636−
637−{-# INLINE_NORMAL fromIndicesM #-}
638−fromIndicesM :: Monad m => (Int -> m a) -> Stream m a
639−fromIndicesM gen = Stream step 0
640− where
641− {-# INLINE_LATE step #-}
642− step _ i = do
643− x <- gen i
644− return $ Yield x (i + 1)
645−
646−{-# INLINE fromIndices #-}
647−fromIndices :: Monad m => (Int -> a) -> Stream m a
648−fromIndices gen = fromIndicesM (return . gen)
649−
650−{-# INLINE_NORMAL generateM #-}
651−generateM :: Monad m => Int -> (Int -> m a) -> Stream m a
652−generateM n gen = n `seq` Stream step 0
653− where
654− {-# INLINE_LATE step #-}
655− step _ i | i < n = do
656− x <- gen i
657− return $ Yield x (i + 1)
658− | otherwise = return Stop
659−
660−{-# INLINE generate #-}
661−generate :: Monad m => Int -> (Int -> a) -> Stream m a
662−generate n gen = generateM n (return . gen)
663−
664−-- XXX we need the MonadAsync constraint because of a rewrite rule.
665−-- | Convert a list of monadic actions to a 'Stream'
666−{-# INLINE_LATE fromListM #-}
667−fromListM :: MonadAsync m => [m a] -> Stream m a
668−fromListM = Stream step
669− where
670− {-# INLINE_LATE step #-}
671− step _ (m:ms) = m >>= \x -> return $ Yield x ms
672− step _ [] = return Stop
673−
674−{-# INLINE toStreamD #-}
675−toStreamD :: (K.IsStream t, Monad m) => t m a -> Stream m a
676−toStreamD = fromStreamK . K.toStream
677−
678−{-# INLINE_NORMAL fromPrimIORef #-}
679−fromPrimIORef :: (MonadIO m, Prim a) => Prim.IORef a -> Stream m a
680−fromPrimIORef var = Stream step ()
681− where
682− {-# INLINE_LATE step #-}
683− step _ () = liftIO (Prim.readIORef var) >>= \x -> return $ Yield x ()
684−
685−-------------------------------------------------------------------------------
686−-- Generation from SVar
687−-------------------------------------------------------------------------------
688−
689−data FromSVarState t m a =
690− FromSVarInit
691− | FromSVarRead (SVar t m a)
692− | FromSVarLoop (SVar t m a) [ChildEvent a]
693− | FromSVarDone (SVar t m a)
694−
695−{-# INLINE_NORMAL fromSVar #-}
696−fromSVar :: (MonadAsync m) => SVar t m a -> Stream m a
697−fromSVar svar = Stream step FromSVarInit
698− where
699−
700− {-# INLINE_LATE step #-}
701− step _ FromSVarInit = do
702− ref <- liftIO $ newIORef ()
703− _ <- liftIO $ mkWeakIORef ref hook
704− -- when this copy of svar gets garbage collected "ref" will get
705− -- garbage collected and our GC hook will be called.
706− let sv = svar{svarRef = Just ref}
707− return $ Skip (FromSVarRead sv)
708−
709− where
710−
711− {-# NOINLINE hook #-}
712− hook = do
713− when (svarInspectMode svar) $ do
714− r <- liftIO $ readIORef (svarStopTime (svarStats svar))
715− when (isNothing r) $
716− printSVar svar "SVar Garbage Collected"
717− cleanupSVar svar
718− -- If there are any SVars referenced by this SVar a GC will prompt
719− -- them to be cleaned up quickly.
720− when (svarInspectMode svar) performMajorGC
721−
722− step _ (FromSVarRead sv) = do
723− list <- readOutputQ sv
724− -- Reversing the output is important to guarantee that we process the
725− -- outputs in the same order as they were generated by the constituent
726− -- streams.
727− return $ Skip $ FromSVarLoop sv (Prelude.reverse list)
728−
729− step _ (FromSVarLoop sv []) = do
730− done <- postProcess sv
731− return $ Skip $ if done
732− then (FromSVarDone sv)
733− else (FromSVarRead sv)
734−
735− step _ (FromSVarLoop sv (ev : es)) = do
736− case ev of
737− ChildYield a -> return $ Yield a (FromSVarLoop sv es)
738− ChildStop tid e -> do
739− accountThread sv tid
740− case e of
741− Nothing -> do
742− stop <- shouldStop tid
743− if stop
744− then do
745− liftIO (cleanupSVar sv)
746− return $ Skip (FromSVarDone sv)
747− else return $ Skip (FromSVarLoop sv es)
748− Just ex ->
749− case fromException ex of
750− Just ThreadAbort ->
751− return $ Skip (FromSVarLoop sv es)
752− Nothing -> liftIO (cleanupSVar sv) >> throwM ex
753− where
754−
755− shouldStop tid =
756− case svarStopStyle sv of
757− StopNone -> return False
758− StopAny -> return True
759− StopBy -> do
760− sid <- liftIO $ readIORef (svarStopBy sv)
761− return $ if tid == sid then True else False
762−
763− step _ (FromSVarDone sv) = do
764− when (svarInspectMode sv) $ do
765− t <- liftIO $ getTime Monotonic
766− liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)
767− liftIO $ printSVar sv "SVar Done"
768− return Stop
769−
770−-------------------------------------------------------------------------------
771−-- Process events received by a fold consumer from a stream producer
772−-------------------------------------------------------------------------------
773−
774−{-# INLINE_NORMAL fromProducer #-}
775−fromProducer :: (MonadAsync m) => SVar t m a -> Stream m a
776−fromProducer svar = Stream step (FromSVarRead svar)
777− where
778−
779− {-# INLINE_LATE step #-}
780− step _ (FromSVarRead sv) = do
781− list <- readOutputQ sv
782− -- Reversing the output is important to guarantee that we process the
783− -- outputs in the same order as they were generated by the constituent
784− -- streams.
785− return $ Skip $ FromSVarLoop sv (Prelude.reverse list)
786−
787− step _ (FromSVarLoop sv []) = return $ Skip $ FromSVarRead sv
788− step _ (FromSVarLoop sv (ev : es)) = do
789− case ev of
790− ChildYield a -> return $ Yield a (FromSVarLoop sv es)
791− ChildStop tid e -> do
792− accountThread sv tid
793− case e of
794− Nothing -> do
795− sendStopToProducer sv
796− return $ Skip (FromSVarDone sv)
797− Just _ -> error "Bug: fromProducer: received exception"
798−
799− step _ (FromSVarDone sv) = do
800− when (svarInspectMode sv) $ do
801− t <- liftIO $ getTime Monotonic
802− liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)
803− liftIO $ printSVar sv "SVar Done"
804− return Stop
805−
806− step _ FromSVarInit = undefined
807−
808452-------------------------------------------------------------------------------
809453-- Hoisting the inner monad
810454-------------------------------------------------------------------------------
@@ -4896,56 +4540,3 @@ dropByTime duration (Stream step1 state1) = Stream step (DropByTimeInit state1)
48964540 Yield x s -> Yield x (DropByTimeYield s)
48974541 Skip s -> Skip (DropByTimeYield s)
48984542 Stop -> Stop
4899−
4900−{-# INLINE updateTimeVar #-}
4901−updateTimeVar :: Prim.IORef Int64 -> IO ()
4902−updateTimeVar timeVar = do
4903− MicroSecond64 t <- fromAbsTime <$> getTime Monotonic
4904− Prim.modifyIORef' timeVar (const t)
4905−
4906−{-# INLINE updateWithDelay #-}
4907−updateWithDelay :: RealFrac a => a -> Prim.IORef Int64 -> IO ()
4908−updateWithDelay precision timeVar = do
4909− threadDelay (delayTime precision)
4910− updateTimeVar timeVar
4911−
4912− where
4913−
4914− -- Keep the minimum at least a millisecond to avoid high CPU usage
4915− {-# INLINE delayTime #-}
4916− delayTime g
4917− | g' >= fromIntegral (maxBound :: Int) = maxBound
4918− | g' < 1000 = 1000
4919− | otherwise = round g'
4920−
4921− where
4922−
4923− g' = g * 10 ^ (6 :: Int)
4924−
4925−-- XXX we should move this to stream generation section of this file. Also, the
4926−-- take/drop combinators above should be moved to filtering section.
4927−
4928−{-# INLINE_NORMAL times #-}
4929−times :: MonadAsync m => Double -> Stream m (AbsTime, RelTime64)
4930−times g = Stream step Nothing
4931−
4932− where
4933−
4934− {-# INLINE_LATE step #-}
4935− step _ Nothing = do
4936− -- XXX note that this is safe only on a 64-bit machine. On a 32-bit
4937− -- machine a 64-bit 'Var' cannot be read consistently without a lock
4938− -- while another thread is writing to it.
4939− timeVar <- liftIO $ Prim.newIORef (0 :: Int64)
4940− liftIO $ updateTimeVar timeVar
4941− tid <- forkManaged $ liftIO $ forever (updateWithDelay g timeVar)
4942− a <- liftIO $ Prim.readIORef timeVar
4943− return $ Skip $ Just (timeVar, tid, a)
4944−
4945− step _ s@(Just (timeVar, _, t0)) = do
4946− a <- liftIO $ Prim.readIORef timeVar
4947− -- XXX we can perhaps use an AbsTime64 using a 64 bit Int for
4948− -- efficiency. or maybe we can use a representation using Double for
4949− -- floating precision time
4950− return $ Yield (toAbsTime (MicroSecond64 t0),
4951− (toRelTime64 (MicroSecond64 (a - t0)))) s