git.iko.soy/streamly
on 61af9fc help sign in
streamly/src/Streamly/Internal/Data/Stream/Async/Channel/Operations.hs
290 lines 11 kB history raw
1#ifdef __HADDOCK_VERSION__
2#undef INSPECTION
3#endif
5#ifdef INSPECTION
6{-# LANGUAGE TemplateHaskell #-}
7{-# OPTIONS_GHC -fplugin Test.Inspection.Plugin #-}
8#endif
10-- |
11-- Module : Streamly.Internal.Data.Stream.Async.Channel.Operations
12-- Copyright : (c) 2017 Composewell Technologies
13-- License : BSD-3-Clause
14-- Maintainer : streamly@composewell.com
15-- Stability : experimental
16-- Portability : GHC
18module Streamly.Internal.Data.Stream.Async.Channel.Operations
19 (
20 toChannel
21 , toChannelK
22 , fromChannel
23 , fromChannelK
24 )
25where
27#include "inline.hs"
29import Control.Exception (fromException)
30import Control.Monad (when)
31import Control.Monad.Catch (throwM, MonadThrow)
32import Control.Monad.IO.Class (MonadIO(liftIO))
33import Control.Monad.Trans.Control (MonadBaseControl)
34import Data.IORef (newIORef, readIORef, mkWeakIORef, writeIORef)
35import Data.Maybe (isNothing)
36import Streamly.Internal.Control.Concurrent (MonadAsync, askRunInIO)
37import Streamly.Internal.Data.Stream.Type (Stream)
38import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
39import System.Mem (performMajorGC)
41import qualified Streamly.Internal.Data.Stream as Stream
42import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
43import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
45import Streamly.Internal.Data.Stream.Async.Channel.Dispatcher
46import Streamly.Internal.Data.Stream.Async.Channel.Type
47import Streamly.Internal.Data.Stream.Channel.Types
49import Prelude hiding (map, concat, concatMap)
51#if __GLASGOW_HASKELL__ < 810
52#ifdef INSPECTION
53import Control.Exception (Exception)
54import Control.Monad.Catch (MonadThrow)
55import Control.Monad.Trans.Control (MonadBaseControl)
56import Data.Typeable (Typeable)
57import Test.Inspection (inspect, hasNoTypeClassesExcept)
58#endif
59#endif
61------------------------------------------------------------------------------
62-- Generating streams from a channel
63------------------------------------------------------------------------------
65-- $concurrentEval
66--
67-- Usually a channel is used to concurrently evaluate multiple actions in a
68-- stream using many worker threads that push the results to the channel and a
69-- single puller that pulls them from channel generating the evaluated stream.
70--
71-- @
72-- input stream
73-- |
74-- <-----------------|<--------worker
75-- | exceptions |
76-- output stream <---Channel<------worker
77-- |
78-- |<--------worker
79--
80-- @
81--
82-- The puller itself schedules the worker threads based on demand.
83-- Exceptions are propagated from the worker threads to the puller.
85-------------------------------------------------------------------------------
86-- Write a stream to a channel
87-------------------------------------------------------------------------------
89-- XXX Should be a Fold, singleton API could be called joinChannel, or the fold
90-- can be called joinChannel.
92-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
93-- be read back from the SVar using 'fromSVar'.
94{-# INLINE toChannelK #-}
95toChannelK :: (MonadIO m, MonadBaseControl IO m) =>
96 Channel m a -> K.Stream m a -> m ()
97toChannelK sv m = do
98 runIn <- askRunInIO
99 liftIO $ enqueue sv (runIn, m)
101-- INLINE for fromStreamK/toStreamK fusion
103-- | Send a stream to a given channel for concurrent evaluation.
104{-# INLINE toChannel #-}
105toChannel :: (MonadIO m, MonadBaseControl IO m) =>
106 Channel m a -> Stream m a -> m ()
107toChannel chan = toChannelK chan . Stream.toStreamK
109{-
110-- | Send a stream of streams to a concurrent channel for evaluation.
111{-# INLINE joinChannel #-}
112joinChannel :: Channel m a -> Fold m (Stream m a) ()
113joinChannel = undefined
114-}
116-------------------------------------------------------------------------------
117-- Read a stream from a channel
118-------------------------------------------------------------------------------
120-- | Pull a stream from an SVar.
121{-# NOINLINE fromChannelRaw #-}
122fromChannelRaw :: (MonadIO m, MonadThrow m) => Channel m a -> K.Stream m a
123fromChannelRaw sv = K.MkStream $ \st yld sng stp -> do
124 list <- readOutputQ sv
125 -- Reversing the output is important to guarantee that we process the
126 -- outputs in the same order as they were generated by the constituent
127 -- streams.
128 K.foldStream st yld sng stp $ processEvents $ reverse list
130 where
132 allDone stp = do
133 when (svarInspectMode sv) $ do
134 t <- liftIO $ getTime Monotonic
135 liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)
136 liftIO $ printSVar (dumpSVar sv) "SVar Done"
137 stp
139 {-# INLINE processEvents #-}
140 processEvents [] = K.MkStream $ \st yld sng stp -> do
141 done <- postProcess sv
142 if done
143 then allDone stp
144 else K.foldStream st yld sng stp $ fromChannelRaw sv
146 processEvents (ev : es) = K.MkStream $ \st yld sng stp -> do
147 let rest = processEvents es
148 case ev of
149 ChildYield a -> yld a rest
150 ChildStop tid e -> do
151 accountThread sv tid
152 case e of
153 Nothing -> K.foldStream st yld sng stp rest
154 Just ex ->
155 case fromException ex of
156 Just ThreadAbort ->
157 K.foldStream st yld sng stp rest
158 Nothing -> do
159 liftIO (cleanupSVar (workerThreads sv))
160 throwM ex
162#if __GLASGOW_HASKELL__ < 810
163#ifdef INSPECTION
164-- Use of GHC constraint tuple (GHC.Classes.(%,,%)) in fromStreamVar leads to
165-- space leak because the tuple gets allocated in every recursive call and each
166-- allocation holds on to the previous allocation. This test is to make sure
167-- that we do not use the constraint tuple type class.
168--
169inspect $ hasNoTypeClassesExcept 'fromStreamVar
170 [ ''Monad
171 , ''Applicative
172 , ''MonadThrow
173 , ''Exception
174 , ''MonadIO
175 , ''MonadBaseControl
176 , ''Typeable
177 , ''Functor
178 ]
179#endif
180#endif
182-- XXX fromChannel Should not be called multiple times, we can add a
183-- safeguard for that. Or we can replicate the stream so that we can distribute
184-- it to multiple consumers. or should we use an explicit dupChannel for that?
186{-# INLINE fromChannelK #-}
187fromChannelK :: MonadAsync m => Channel m a -> K.Stream m a
188fromChannelK sv =
189 K.mkStream $ \st yld sng stp -> do
190 ref <- liftIO $ newIORef ()
191 _ <- liftIO $ mkWeakIORef ref hook
193 startChannel sv
194 -- We pass a copy of sv to fromStreamVar, so that we know that it has
195 -- no other references, when that copy gets garbage collected "ref"
196 -- will get garbage collected and our hook will be called.
197 K.foldStreamShared st yld sng stp $
198 fromChannelRaw sv{svarRef = Just ref}
199 where
201 hook = do
202 when (svarInspectMode sv) $ do
203 r <- liftIO $ readIORef (svarStopTime (svarStats sv))
204 when (isNothing r) $
205 printSVar (dumpSVar sv) "SVar Garbage Collected"
206 cleanupSVar (workerThreads sv)
207 -- If there are any SVars referenced by this SVar a GC will prompt
208 -- them to be cleaned up quickly.
209 when (svarInspectMode sv) performMajorGC
211-- | Generate a stream of results from concurrent evaluations from a channel.
212-- Evaluation of the channel does not start until this API is called. This API
213-- must not be called more than once on a channel. It kicks off evaluation of
214-- the channel by dispatching concurrent workers and ensures that as long there
215-- is work queued on the channel workers are dispatched proportional to the
216-- demand by the consumer.
217--
218{-# INLINE fromChannel #-}
219fromChannel :: MonadAsync m => Channel m a -> Stream m a
220fromChannel = Stream.fromStreamK . fromChannelK
222data FromSVarState t m a =
223 FromSVarInit
224 | FromSVarRead (Channel m a)
225 | FromSVarLoop (Channel m a) [ChildEvent a]
226 | FromSVarDone (Channel m a)
228-- | Like 'fromSVar' but generates a StreamD style stream instead of CPS.
229--
230{-# INLINE_NORMAL _fromChannelD #-}
231_fromChannelD :: (MonadIO m, MonadThrow m) => Channel m a -> D.Stream m a
232_fromChannelD svar = D.Stream step FromSVarInit
233 where
235 {-# INLINE_LATE step #-}
236 step _ FromSVarInit = do
237 ref <- liftIO $ newIORef ()
238 _ <- liftIO $ mkWeakIORef ref hook
239 -- when this copy of svar gets garbage collected "ref" will get
240 -- garbage collected and our GC hook will be called.
241 let sv = svar{svarRef = Just ref}
242 return $ D.Skip (FromSVarRead sv)
244 where
246 {-# NOINLINE hook #-}
247 hook = do
248 when (svarInspectMode svar) $ do
249 r <- liftIO $ readIORef (svarStopTime (svarStats svar))
250 when (isNothing r) $
251 printSVar (dumpSVar svar) "SVar Garbage Collected"
252 cleanupSVar (workerThreads svar)
253 -- If there are any SVars referenced by this SVar a GC will prompt
254 -- them to be cleaned up quickly.
255 when (svarInspectMode svar) performMajorGC
257 step _ (FromSVarRead sv) = do
258 list <- readOutputQ sv
259 -- Reversing the output is important to guarantee that we process the
260 -- outputs in the same order as they were generated by the constituent
261 -- streams.
262 return $ D.Skip $ FromSVarLoop sv (Prelude.reverse list)
264 step _ (FromSVarLoop sv []) = do
265 done <- postProcess sv
266 return $ D.Skip $ if done
267 then FromSVarDone sv
268 else FromSVarRead sv
270 step _ (FromSVarLoop sv (ev : es)) = do
271 case ev of
272 ChildYield a -> return $ D.Yield a (FromSVarLoop sv es)
273 ChildStop tid e -> do
274 accountThread sv tid
275 case e of
276 Nothing -> return $ D.Skip (FromSVarLoop sv es)
277 Just ex ->
278 case fromException ex of
279 Just ThreadAbort ->
280 return $ D.Skip (FromSVarLoop sv es)
281 Nothing -> do
282 liftIO (cleanupSVar (workerThreads sv))
283 throwM ex
285 step _ (FromSVarDone sv) = do
286 when (svarInspectMode sv) $ do
287 t <- liftIO $ getTime Monotonic
288 liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)
289 liftIO $ printSVar (dumpSVar sv) "SVar Done"
290 return D.Stop