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