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