git.iko.soy/streamly
on 04c2ae0 help sign in
streamly/src/Streamly/Streams/SVar.hs
119 lines 4.2 kB history raw
1{-# LANGUAGE CPP #-}
2{-# LANGUAGE FlexibleContexts #-}
4-- |
5-- Module : Streamly.Streams.SVar
6-- Copyright : (c) 2017 Harendra Kumar
7--
8-- License : BSD3
9-- Maintainer : streamly@composewell.com
10-- Stability : experimental
11-- Portability : GHC
12--
13--
14module Streamly.Streams.SVar
15 ( fromSVar
16 , fromStreamVar
17 , toSVar
18 )
19where
21import Control.Exception (fromException)
22import Control.Monad (when)
23import Control.Monad.Catch (throwM)
24import Control.Monad.IO.Class (MonadIO(liftIO))
25import Data.IORef (newIORef, readIORef, mkWeakIORef, writeIORef)
26import Data.Maybe (isNothing)
27#if __GLASGOW_HASKELL__ < 808
28import Data.Semigroup ((<>))
29#endif
30import System.IO (hPutStrLn, stderr)
31import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
32import System.Mem (performMajorGC)
34import Streamly.Internal.Data.SVar
35import Streamly.Streams.StreamK hiding (reverse)
37printSVar :: SVar t m a -> String -> IO ()
38printSVar sv how = do
39 svInfo <- dumpSVar sv
40 hPutStrLn stderr $ "\n" <> how <> "\n" <> svInfo
42-- | Pull a stream from an SVar.
43{-# NOINLINE fromStreamVar #-}
44fromStreamVar :: MonadAsync m => SVar Stream m a -> Stream m a
45fromStreamVar sv = mkStream $ \st yld sng stp -> do
46 list <- readOutputQ sv
47 -- Reversing the output is important to guarantee that we process the
48 -- outputs in the same order as they were generated by the constituent
49 -- streams.
50 foldStream st yld sng stp $ processEvents $ reverse list
52 where
54 allDone stp = do
55 when (svarInspectMode sv) $ do
56 t <- liftIO $ getTime Monotonic
57 liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)
58 liftIO $ printSVar sv "SVar Done"
59 stp
61 {-# INLINE processEvents #-}
62 processEvents [] = mkStream $ \st yld sng stp -> do
63 done <- postProcess sv
64 if done
65 then allDone stp
66 else foldStream st yld sng stp $ fromStreamVar sv
68 processEvents (ev : es) = mkStream $ \st yld sng stp -> do
69 let rest = processEvents es
70 case ev of
71 ChildYield a -> yld a rest
72 ChildStop tid e -> do
73 accountThread sv tid
74 case e of
75 Nothing -> do
76 stop <- shouldStop tid
77 if stop
78 then liftIO (cleanupSVar sv) >> allDone stp
79 else foldStream st yld sng stp rest
80 Just ex ->
81 case fromException ex of
82 Just ThreadAbort ->
83 foldStream st yld sng stp rest
84 Nothing -> liftIO (cleanupSVar sv) >> throwM ex
85 shouldStop tid =
86 case svarStopStyle sv of
87 StopNone -> return False
88 StopAny -> return True
89 StopBy -> do
90 sid <- liftIO $ readIORef (svarStopBy sv)
91 return $ if tid == sid then True else False
93{-# INLINE fromSVar #-}
94fromSVar :: (MonadAsync m, IsStream t) => SVar Stream m a -> t m a
95fromSVar sv =
96 mkStream $ \st yld sng stp -> do
97 ref <- liftIO $ newIORef ()
98 _ <- liftIO $ mkWeakIORef ref hook
99 -- We pass a copy of sv to fromStreamVar, so that we know that it has
100 -- no other references, when that copy gets garbage collected "ref"
101 -- will get garbage collected and our hook will be called.
102 foldStreamShared st yld sng stp $
103 fromStream $ fromStreamVar sv{svarRef = Just ref}
104 where
106 hook = do
107 when (svarInspectMode sv) $ do
108 r <- liftIO $ readIORef (svarStopTime (svarStats sv))
109 when (isNothing r) $
110 printSVar sv "SVar Garbage Collected"
111 cleanupSVar sv
112 -- If there are any SVars referenced by this SVar a GC will prompt
113 -- them to be cleaned up quickly.
114 when (svarInspectMode sv) performMajorGC
116-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
117-- be read back from the SVar using 'fromSVar'.
118toSVar :: (IsStream t, MonadAsync m) => SVar Stream m a -> t m a -> m ()
119toSVar sv m = toStreamVar sv (toStream m)