git.iko.soy/streamly
on e856425 help sign in
streamly/src/Streamly/Internal/Data/Stream/Concurrent/Channel/Operations.hs
295 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.Concurrent.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.Concurrent.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.Concurrent.Channel.Dispatcher
46import Streamly.Internal.Data.Stream.Concurrent.Channel.Type
47import Streamly.Internal.Data.Stream.Channel.Types hiding (inspect)
49import Prelude hiding (map, concat, concatMap)
51#ifdef INSPECTION
52import Control.Exception (Exception)
53import Data.Typeable (Typeable)
54import Test.Inspection (inspect, hasNoTypeClassesExcept)
55#endif
57------------------------------------------------------------------------------
58-- Generating streams from a channel
59------------------------------------------------------------------------------
61-- $concurrentEval
62--
63-- Usually a channel is used to concurrently evaluate multiple actions in a
64-- stream using many worker threads that push the results to the channel and a
65-- single puller that pulls them from channel generating the evaluated stream.
66--
67-- @
68-- input stream
69-- |
70-- <-----------------|<--------worker
71-- | exceptions |
72-- output stream <---Channel<------worker
73-- |
74-- |<--------worker
75--
76-- @
77--
78-- The puller itself schedules the worker threads based on demand.
79-- Exceptions are propagated from the worker threads to the puller.
81-------------------------------------------------------------------------------
82-- Write a stream to a channel
83-------------------------------------------------------------------------------
85-- XXX Should be a Fold, singleton API could be called joinChannel, or the fold
86-- can be called joinChannel.
88-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
89-- be read back from the SVar using 'fromSVar'.
90{-# INLINE toChannelK #-}
91toChannelK :: (MonadIO m, MonadBaseControl IO m) =>
92 Channel m a -> K.Stream m a -> m ()
93toChannelK sv m = do
94 runIn <- askRunInIO
95 liftIO $ enqueue sv False (runIn, m)
97-- INLINE for fromStreamK/toStreamK fusion
99-- | Send a stream to a given channel for concurrent evaluation.
100{-# INLINE toChannel #-}
101toChannel :: (MonadIO m, MonadBaseControl IO m) =>
102 Channel m a -> Stream m a -> m ()
103toChannel chan = toChannelK chan . Stream.toStreamK
105{-
106-- | Send a stream of streams to a concurrent channel for evaluation.
107{-# INLINE joinChannel #-}
108joinChannel :: Channel m a -> Fold m (Stream m a) ()
109joinChannel = undefined
110-}
112-------------------------------------------------------------------------------
113-- Read a stream from a channel
114-------------------------------------------------------------------------------
116-- | Pull a stream from an SVar.
117{-# NOINLINE fromChannelRaw #-}
118fromChannelRaw :: (MonadIO m, MonadThrow m) => Channel m a -> K.Stream m a
119fromChannelRaw sv = K.MkStream $ \st yld sng stp -> do
120 list <- readOutputQ sv
121 -- Reversing the output is important to guarantee that we process the
122 -- outputs in the same order as they were generated by the constituent
123 -- streams.
124 K.foldStream st yld sng stp $ processEvents $ reverse list
126 where
128 cleanup = do
129 when (svarInspectMode sv) $ liftIO $ do
130 t <- getTime Monotonic
131 writeIORef (svarStopTime (svarStats sv)) (Just t)
132 printSVar (dumpSVar sv) "SVar Done"
134 {-# INLINE processEvents #-}
135 processEvents [] = K.MkStream $ \st yld sng stp -> do
136 done <- postProcess sv
137 if done
138 then cleanup >> stp
139 else K.foldStream st yld sng stp $ fromChannelRaw sv
141 processEvents (ev : es) = K.MkStream $ \st yld sng stp -> do
142 let rest = processEvents es
143 case ev of
144 ChildYield a -> yld a rest
145 ChildStopChannel -> do
146 liftIO (cleanupSVar (workerThreads sv))
147 cleanup >> stp
148 ChildStop tid e -> do
149 accountThread sv tid
150 case e of
151 Nothing -> K.foldStream st yld sng stp rest
152 Just ex ->
153 case fromException ex of
154 Just ThreadAbort ->
155 -- We terminate the loop after sending
156 -- ThreadAbort to workers so we should never
157 -- get it unless it is thrown from inside a
158 -- worker thread or by someone else to our
159 -- thread.
160 error "processEvents: got ThreadAbort"
161 -- K.foldStream st yld sng stp rest
162 Nothing -> do
163 liftIO (cleanupSVar (workerThreads sv))
164 cleanup >> throwM ex
166#ifdef INSPECTION
167-- Use of GHC constraint tuple (GHC.Classes.(%,,%)) in fromStreamVar leads to
168-- space leak because the tuple gets allocated in every recursive call and each
169-- allocation holds on to the previous allocation. This test is to make sure
170-- that we do not use the constraint tuple type class.
171--
172inspect $ hasNoTypeClassesExcept 'fromChannelRaw
173 [ ''Monad
174 , ''Applicative
175 , ''MonadThrow
176 , ''Exception
177 , ''MonadIO
178 , ''MonadBaseControl
179 , ''Typeable
180 , ''Functor
181 ]
182#endif
184-- XXX fromChannel Should not be called multiple times, we can add a
185-- safeguard for that. Or we can replicate the stream so that we can distribute
186-- it to multiple consumers. or should we use an explicit dupChannel for that?
188{-# INLINE fromChannelK #-}
189fromChannelK :: MonadAsync m => Channel m a -> K.Stream m a
190fromChannelK sv =
191 K.mkStream $ \st yld sng stp -> do
192 ref <- liftIO $ newIORef ()
193 _ <- liftIO $ mkWeakIORef ref hook
195 startChannel sv
196 -- We pass a copy of sv to fromStreamVar, so that we know that it has
197 -- no other references, when that copy gets garbage collected "ref"
198 -- will get garbage collected and our hook will be called.
199 K.foldStreamShared st yld sng stp $
200 fromChannelRaw sv{svarRef = Just ref}
201 where
203 hook = do
204 when (svarInspectMode sv) $ do
205 r <- liftIO $ readIORef (svarStopTime (svarStats sv))
206 when (isNothing r) $
207 printSVar (dumpSVar sv) "SVar Garbage Collected"
208 cleanupSVar (workerThreads sv)
209 -- If there are any SVars referenced by this SVar a GC will prompt
210 -- them to be cleaned up quickly.
211 when (svarInspectMode sv) performMajorGC
213-- | Generate a stream of results from concurrent evaluations from a channel.
214-- Evaluation of the channel does not start until this API is called. This API
215-- must not be called more than once on a channel. It kicks off evaluation of
216-- the channel by dispatching concurrent workers and ensures that as long there
217-- is work queued on the channel workers are dispatched proportional to the
218-- demand by the consumer.
219--
220{-# INLINE fromChannel #-}
221fromChannel :: MonadAsync m => Channel m a -> Stream m a
222fromChannel = Stream.fromStreamK . fromChannelK
224data FromSVarState t m a =
225 FromSVarInit
226 | FromSVarRead (Channel m a)
227 | FromSVarLoop (Channel m a) [ChildEvent a]
228 | FromSVarDone (Channel m a)
230-- | Like 'fromSVar' but generates a StreamD style stream instead of CPS.
231--
232{-# INLINE_NORMAL _fromChannelD #-}
233_fromChannelD :: (MonadIO m, MonadThrow m) => Channel m a -> D.Stream m a
234_fromChannelD svar = D.Stream step FromSVarInit
235 where
237 {-# INLINE_LATE step #-}
238 step _ FromSVarInit = do
239 ref <- liftIO $ newIORef ()
240 _ <- liftIO $ mkWeakIORef ref hook
241 -- when this copy of svar gets garbage collected "ref" will get
242 -- garbage collected and our GC hook will be called.
243 let sv = svar{svarRef = Just ref}
244 return $ D.Skip (FromSVarRead sv)
246 where
248 {-# NOINLINE hook #-}
249 hook = do
250 when (svarInspectMode svar) $ do
251 r <- liftIO $ readIORef (svarStopTime (svarStats svar))
252 when (isNothing r) $
253 printSVar (dumpSVar svar) "SVar Garbage Collected"
254 cleanupSVar (workerThreads svar)
255 -- If there are any SVars referenced by this SVar a GC will prompt
256 -- them to be cleaned up quickly.
257 when (svarInspectMode svar) performMajorGC
259 step _ (FromSVarRead sv) = do
260 list <- readOutputQ sv
261 -- Reversing the output is important to guarantee that we process the
262 -- outputs in the same order as they were generated by the constituent
263 -- streams.
264 return $ D.Skip $ FromSVarLoop sv (Prelude.reverse list)
266 step _ (FromSVarLoop sv []) = do
267 done <- postProcess sv
268 return $ D.Skip $ if done
269 then FromSVarDone sv
270 else FromSVarRead sv
272 step _ (FromSVarLoop sv (ev : es)) = do
273 case ev of
274 ChildYield a -> return $ D.Yield a (FromSVarLoop sv es)
275 ChildStopChannel -> do
276 liftIO (cleanupSVar (workerThreads sv))
277 return $ D.Skip (FromSVarDone sv)
278 ChildStop tid e -> do
279 accountThread sv tid
280 case e of
281 Nothing -> return $ D.Skip (FromSVarLoop sv es)
282 Just ex ->
283 case fromException ex of
284 Just ThreadAbort ->
285 return $ D.Skip (FromSVarLoop sv es)
286 Nothing -> do
287 liftIO (cleanupSVar (workerThreads sv))
288 throwM ex
290 step _ (FromSVarDone sv) = do
291 when (svarInspectMode sv) $ do
292 t <- liftIO $ getTime Monotonic
293 liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)
294 liftIO $ printSVar (dumpSVar sv) "SVar Done"
295 return D.Stop