git.iko.soy/streamly
on 2ca80e7 help sign in
streamly/src/Streamly/Internal/Data/Stream/Async/Channel/Operations.hs
285 lines 10 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.Consumer
46import Streamly.Internal.Data.Stream.Async.Channel.Dispatcher
47import Streamly.Internal.Data.Stream.Async.Channel.Type
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'.
94toChannelK :: (MonadIO m, MonadBaseControl IO m) =>
95 Channel m a -> K.Stream m a -> m ()
96toChannelK sv m = do
97 runIn <- askRunInIO
98 liftIO $ enqueue sv (runIn, m)
100-- INLINE for fromStreamK/toStreamK fusion
102-- | Send a stream to a given channel for concurrent evaluation.
103{-# INLINE toChannel #-}
104toChannel :: (MonadIO m, MonadBaseControl IO m) =>
105 Channel m a -> Stream m a -> m ()
106toChannel chan = toChannelK chan . Stream.toStreamK
108{-
109-- | Send a stream of streams to a concurrent channel for evaluation.
110{-# INLINE joinChannel #-}
111joinChannel :: Channel m a -> Fold m (Stream m a) ()
112joinChannel = undefined
113-}
115-------------------------------------------------------------------------------
116-- Read a stream from a channel
117-------------------------------------------------------------------------------
119-- | Pull a stream from an SVar.
120{-# NOINLINE fromChannelRaw #-}
121fromChannelRaw :: (MonadIO m, MonadThrow m) => Channel m a -> K.Stream m a
122fromChannelRaw sv = K.MkStream $ \st yld sng stp -> do
123 list <- readOutputQ sv
124 -- Reversing the output is important to guarantee that we process the
125 -- outputs in the same order as they were generated by the constituent
126 -- streams.
127 K.foldStream st yld sng stp $ processEvents $ reverse list
129 where
131 allDone stp = do
132 when (svarInspectMode sv) $ do
133 t <- liftIO $ getTime Monotonic
134 liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)
135 liftIO $ printSVar sv "SVar Done"
136 stp
138 {-# INLINE processEvents #-}
139 processEvents [] = K.MkStream $ \st yld sng stp -> do
140 done <- postProcess sv
141 if done
142 then allDone stp
143 else K.foldStream st yld sng stp $ fromChannelRaw sv
145 processEvents (ev : es) = K.MkStream $ \st yld sng stp -> do
146 let rest = processEvents es
147 case ev of
148 ChildYield a -> yld a rest
149 ChildStop tid e -> do
150 accountThread sv tid
151 case e of
152 Nothing -> K.foldStream st yld sng stp rest
153 Just ex ->
154 case fromException ex of
155 Just ThreadAbort ->
156 K.foldStream st yld sng stp rest
157 Nothing -> liftIO (cleanupSVar sv) >> throwM ex
159#if __GLASGOW_HASKELL__ < 810
160#ifdef INSPECTION
161-- Use of GHC constraint tuple (GHC.Classes.(%,,%)) in fromStreamVar leads to
162-- space leak because the tuple gets allocated in every recursive call and each
163-- allocation holds on to the previous allocation. This test is to make sure
164-- that we do not use the constraint tuple type class.
165--
166inspect $ hasNoTypeClassesExcept 'fromStreamVar
167 [ ''Monad
168 , ''Applicative
169 , ''MonadThrow
170 , ''Exception
171 , ''MonadIO
172 , ''MonadBaseControl
173 , ''Typeable
174 , ''Functor
175 ]
176#endif
177#endif
179-- XXX fromChannel Should not be called multiple times, we can add a
180-- safeguard for that. Or we can replicate the stream so that we can distribute
181-- it to multiple consumers. or should we use an explicit dupChannel for that?
183{-# INLINE fromChannelK #-}
184fromChannelK :: MonadAsync m => Channel m a -> K.Stream m a
185fromChannelK sv =
186 K.mkStream $ \st yld sng stp -> do
187 ref <- liftIO $ newIORef ()
188 _ <- liftIO $ mkWeakIORef ref hook
190 startChannel sv
191 -- We pass a copy of sv to fromStreamVar, so that we know that it has
192 -- no other references, when that copy gets garbage collected "ref"
193 -- will get garbage collected and our hook will be called.
194 K.foldStreamShared st yld sng stp $
195 fromChannelRaw sv{svarRef = Just ref}
196 where
198 hook = do
199 when (svarInspectMode sv) $ do
200 r <- liftIO $ readIORef (svarStopTime (svarStats sv))
201 when (isNothing r) $
202 printSVar sv "SVar Garbage Collected"
203 cleanupSVar sv
204 -- If there are any SVars referenced by this SVar a GC will prompt
205 -- them to be cleaned up quickly.
206 when (svarInspectMode sv) performMajorGC
208-- | Generate a stream of results from concurrent evaluations from a channel.
209-- Evaluation of the channel does not start until this API is called. This API
210-- must not be called more than once on a channel. It kicks off evaluation of
211-- the channel by dispatching concurrent workers and ensures that as long there
212-- is work queued on the channel workers are dispatched proportional to the
213-- demand by the consumer.
214--
215{-# INLINE fromChannel #-}
216fromChannel :: MonadAsync m => Channel m a -> Stream m a
217fromChannel = Stream.fromStreamK . fromChannelK
219data FromSVarState t m a =
220 FromSVarInit
221 | FromSVarRead (Channel m a)
222 | FromSVarLoop (Channel m a) [ChildEvent a]
223 | FromSVarDone (Channel m a)
225-- | Like 'fromSVar' but generates a StreamD style stream instead of CPS.
226--
227{-# INLINE_NORMAL _fromChannelD #-}
228_fromChannelD :: (MonadIO m, MonadThrow m) => Channel m a -> D.Stream m a
229_fromChannelD svar = D.Stream step FromSVarInit
230 where
232 {-# INLINE_LATE step #-}
233 step _ FromSVarInit = do
234 ref <- liftIO $ newIORef ()
235 _ <- liftIO $ mkWeakIORef ref hook
236 -- when this copy of svar gets garbage collected "ref" will get
237 -- garbage collected and our GC hook will be called.
238 let sv = svar{svarRef = Just ref}
239 return $ D.Skip (FromSVarRead sv)
241 where
243 {-# NOINLINE hook #-}
244 hook = do
245 when (svarInspectMode svar) $ do
246 r <- liftIO $ readIORef (svarStopTime (svarStats svar))
247 when (isNothing r) $
248 printSVar svar "SVar Garbage Collected"
249 cleanupSVar svar
250 -- If there are any SVars referenced by this SVar a GC will prompt
251 -- them to be cleaned up quickly.
252 when (svarInspectMode svar) performMajorGC
254 step _ (FromSVarRead sv) = do
255 list <- readOutputQ sv
256 -- Reversing the output is important to guarantee that we process the
257 -- outputs in the same order as they were generated by the constituent
258 -- streams.
259 return $ D.Skip $ FromSVarLoop sv (Prelude.reverse list)
261 step _ (FromSVarLoop sv []) = do
262 done <- postProcess sv
263 return $ D.Skip $ if done
264 then FromSVarDone sv
265 else FromSVarRead sv
267 step _ (FromSVarLoop sv (ev : es)) = do
268 case ev of
269 ChildYield a -> return $ D.Yield a (FromSVarLoop sv es)
270 ChildStop tid e -> do
271 accountThread sv tid
272 case e of
273 Nothing -> return $ D.Skip (FromSVarLoop sv es)
274 Just ex ->
275 case fromException ex of
276 Just ThreadAbort ->
277 return $ D.Skip (FromSVarLoop sv es)
278 Nothing -> liftIO (cleanupSVar sv) >> throwM ex
280 step _ (FromSVarDone sv) = do
281 when (svarInspectMode sv) $ do
282 t <- liftIO $ getTime Monotonic
283 liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)
284 liftIO $ printSVar sv "SVar Done"
285 return D.Stop