git.iko.soy/streamly
on 2ba5f01 help sign in
streamly/src/Streamly/Streams/SVar.hs
104 lines 3.6 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 System.Clock (Clock(Monotonic), getTime)
28import System.Mem (performMajorGC)
30import Streamly.SVar
31import Streamly.Streams.StreamK
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 stp sng yld -> 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 stp sng yld $ 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 stp sng yld -> do
59 done <- postProcess sv
60 if done
61 then allDone stp
62 else foldStream st stp sng yld $ fromStreamVar sv
64 processEvents (ev : es) = mkStream $ \st stp sng yld -> 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 -> foldStream st stp sng yld rest
72 Just ex ->
73 case fromException ex of
74 Just ThreadAbort ->
75 foldStream st stp sng yld rest
76 Nothing -> liftIO (cleanupSVar sv) >> throwM ex
78{-# INLINE fromSVar #-}
79fromSVar :: (MonadAsync m, IsStream t) => SVar Stream m a -> t m a
80fromSVar sv =
81 mkStream $ \st stp sng yld -> do
82 ref <- liftIO $ newIORef ()
83 _ <- liftIO $ mkWeakIORef ref hook
84 -- We pass a copy of sv to fromStreamVar, so that we know that it has
85 -- no other references, when that copy gets garbage collected "ref"
86 -- will get garbage collected and our hook will be called.
87 foldStreamShared st stp sng yld $
88 fromStream $ fromStreamVar sv{svarRef = Just ref}
89 where
91 hook = do
92 when (svarInspectMode sv) $ do
93 r <- liftIO $ readIORef (svarStopTime (svarStats sv))
94 when (isNothing r) $
95 printSVar sv "SVar Garbage Collected"
96 cleanupSVar sv
97 -- If there are any SVars referenced by this SVar a GC will prompt
98 -- them to be cleaned up quickly.
99 when (svarInspectMode sv) performMajorGC
101-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
102-- be read back from the SVar using 'fromSVar'.
103toSVar :: (IsStream t, MonadAsync m) => SVar Stream m a -> t m a -> m ()
104toSVar sv m = toStreamVar sv (toStream m)