@@ -1,14 +1,5 @@
1−#ifdef __HADDOCK_VERSION__
6−{-# LANGUAGE TemplateHaskell #-}
7−{-# OPTIONS_GHC -fplugin Test.Inspection.Plugin #-}
11−-- Module : Streamly.Internal.Data.Stream.Async.Channel.Core
2+-- Module : Streamly.Internal.Data.Stream.Async.Channel.Append
12-- Copyright : (c) 2017 Composewell Technologies
3-- Copyright : (c) 2017 Composewell Technologies
13-- License : BSD-3-Clause
4-- License : BSD-3-Clause
14-- Maintainer : streamly@composewell.com
5-- Maintainer : streamly@composewell.com
@@ -22,38 +13,23 @@
22-- use INLINABLE and SPECIALIZE on it which makes it specialized but it is not
13-- use INLINABLE and SPECIALIZE on it which makes it specialized but it is not
23-- actually inlined.
14-- actually inlined.
25−module Streamly.Internal.Data.Stream.Async.Channel.Core
16+module Streamly.Internal.Data.Stream.Async.Channel.Append
27 newChannel
18 newChannel
37import Control.Concurrent (myThreadId)
22import Control.Concurrent (myThreadId)
38import Control.Concurrent.MVar (newEmptyMVar)
23import Control.Concurrent.MVar (newEmptyMVar)
39−import Control.Exception (fromException)
40−import Control.Monad (when)
41−import Control.Monad.Catch (throwM, MonadThrow)
42import Control.Monad.IO.Class (MonadIO(liftIO))
24import Control.Monad.IO.Class (MonadIO(liftIO))
43import Control.Monad.Trans.Control (MonadBaseControl, restoreM)
25import Control.Monad.Trans.Control (MonadBaseControl, restoreM)
44−import Data.IORef (IORef, newIORef, readIORef, mkWeakIORef, writeIORef)
26+import Data.IORef (IORef, newIORef, readIORef)
45−import Data.Maybe (isNothing)
46import Streamly.Internal.Control.Concurrent
27import Streamly.Internal.Control.Concurrent
47 (MonadRunInIO, MonadAsync, RunInIO(..), askRunInIO)
28 (MonadRunInIO, MonadAsync, RunInIO(..), askRunInIO)
48import Streamly.Internal.Data.Atomics
29import Streamly.Internal.Data.Atomics
49 (atomicModifyIORefCAS, atomicModifyIORefCAS_)
30 (atomicModifyIORefCAS, atomicModifyIORefCAS_)
50−import Streamly.Internal.Data.Stream.Type (Stream)
51−import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
52−import System.Mem (performMajorGC)
54import qualified Data.Set as Set
32import qualified Data.Set as Set
55−import qualified Streamly.Internal.Data.Stream as Stream
56−import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
57import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
33import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
59import Streamly.Internal.Data.Stream.Async.Channel.Consumer
35import Streamly.Internal.Data.Stream.Async.Channel.Consumer
@@ -61,18 +37,6 @@ import Streamly.Internal.Data.Stream.Async.Channel.Dispatcher
61import Streamly.Internal.Data.Stream.Async.Channel.Type
37import Streamly.Internal.Data.Stream.Async.Channel.Type
62import Streamly.Internal.Data.Stream.Async.Channel.Worker
38import Streamly.Internal.Data.Stream.Async.Channel.Worker
64−import Prelude hiding (map, concat, concatMap)
66−#if __GLASGOW_HASKELL__ < 810
68−import Control.Exception (Exception)
69−import Control.Monad.Catch (MonadThrow)
70−import Control.Monad.Trans.Control (MonadBaseControl)
71−import Data.Typeable (Typeable)
72−import Test.Inspection (inspect, hasNoTypeClassesExcept)
76------------------------------------------------------------------------------
40------------------------------------------------------------------------------
77-- Creating a channel
41-- Creating a channel
78------------------------------------------------------------------------------
42------------------------------------------------------------------------------
@@ -263,7 +227,6 @@ getLifoSVar mrun cfg = do
263 -> (Channel m a -> m Bool)
227 -> (Channel m a -> m Bool)
264 -> (Channel m a -> IO Bool)
228 -> (Channel m a -> IO Bool)
265 -> (IORef [(RunInIO m, K.Stream m a)]
229 -> (IORef [(RunInIO m, K.Stream m a)]
266− -- -> State Stream m a
267 -> Channel m a
230 -> Channel m a
268 -> Maybe WorkerInfo
231 -> Maybe WorkerInfo
@@ -317,11 +280,6 @@ getLifoSVar mrun cfg = do
317 workLoopLIFOLimited
280 workLoopLIFOLimited
318 in return sv
281 in return sv
320−-- type Channel m a = SVar K.Stream m a
321−-- data Config = Config
325-- | Create a new async style concurrent stream evaluation channel. The monad
283-- | Create a new async style concurrent stream evaluation channel. The monad
326-- state used to run the stream actions is taken from the call site of
284-- state used to run the stream actions is taken from the call site of
327-- newChannel.
285-- newChannel.
@@ -332,229 +290,3 @@ newChannel :: (MonadAsync m) =>
332newChannel modifier = do
290newChannel modifier = do
333 mrun <- askRunInIO
291 mrun <- askRunInIO
334 liftIO $ getLifoSVar mrun (modifier defaultConfig)
292 liftIO $ getLifoSVar mrun (modifier defaultConfig)
336−------------------------------------------------------------------------------
337−-- Generating streams from a channel
338−------------------------------------------------------------------------------
342−-- Usually a channel is used to concurrently evaluate multiple actions in a
343−-- stream using many worker threads that push the results to the channel and a
344−-- single puller that pulls them from channel generating the evaluated stream.
349−-- <-----------------|<--------worker
351−-- output stream <---Channel<------worker
357−-- The puller itself schedules the worker threads based on demand.
358−-- Exceptions are propagated from the worker threads to the puller.
360−-------------------------------------------------------------------------------
361−-- Write a stream to a channel
362−-------------------------------------------------------------------------------
364−-- XXX Should be a Fold, singleton API could be called joinChannel, or the fold
365−-- can be called joinChannel.
367−-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
368−-- be read back from the SVar using 'fromSVar'.
369−toChannelK :: (MonadIO m, MonadBaseControl IO m) =>
370− Channel m a -> K.Stream m a -> m ()
373− liftIO $ enqueue sv (runIn, m)
375−-- INLINE for fromStreamK/toStreamK fusion
377−-- | Send a stream to a given channel for concurrent evaluation.
378−{-# INLINE toChannel #-}
379−toChannel :: (MonadIO m, MonadBaseControl IO m) =>
380− Channel m a -> Stream m a -> m ()
381−toChannel chan = toChannelK chan . Stream.toStreamK
384−-- | Send a stream of streams to a concurrent channel for evaluation.
385−{-# INLINE joinChannel #-}
386−joinChannel :: Channel m a -> Fold m (Stream m a) ()
387−joinChannel = undefined
390−-------------------------------------------------------------------------------
391−-- Read a stream from a channel
392−-------------------------------------------------------------------------------
394−-- | Pull a stream from an SVar.
395−{-# NOINLINE fromChannelRaw #-}
396−fromChannelRaw :: (MonadIO m, MonadThrow m) => Channel m a -> K.Stream m a
397−fromChannelRaw sv = K.MkStream $ \st yld sng stp -> do
398− list <- readOutputQ sv
399− -- Reversing the output is important to guarantee that we process the
400− -- outputs in the same order as they were generated by the constituent
402− K.foldStream st yld sng stp $ processEvents $ reverse list
407− when (svarInspectMode sv) $ do
408− t <- liftIO $ getTime Monotonic
409− liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)
410− liftIO $ printSVar sv "SVar Done"
413− {-# INLINE processEvents #-}
414− processEvents [] = K.MkStream $ \st yld sng stp -> do
415− done <- postProcess sv
418− else K.foldStream st yld sng stp $ fromChannelRaw sv
420− processEvents (ev : es) = K.MkStream $ \st yld sng stp -> do
421− let rest = processEvents es
423− ChildYield a -> yld a rest
424− ChildStop tid e -> do
425− accountThread sv tid
427− Nothing -> K.foldStream st yld sng stp rest
429− case fromException ex of
431− K.foldStream st yld sng stp rest
432− Nothing -> liftIO (cleanupSVar sv) >> throwM ex
434−#if __GLASGOW_HASKELL__ < 810
436−-- Use of GHC constraint tuple (GHC.Classes.(%,,%)) in fromStreamVar leads to
437−-- space leak because the tuple gets allocated in every recursive call and each
438−-- allocation holds on to the previous allocation. This test is to make sure
439−-- that we do not use the constraint tuple type class.
441−inspect $ hasNoTypeClassesExcept 'fromStreamVar
447− , ''MonadBaseControl
454−-- XXX fromChannel Should not be called multiple times, we can add a
455−-- safeguard for that. Or we can replicate the stream so that we can distribute
456−-- it to multiple consumers. or should we use an explicit dupChannel for that?
458−{-# INLINE fromChannelK #-}
459−fromChannelK :: MonadAsync m => Channel m a -> K.Stream m a
461− K.mkStream $ \st yld sng stp -> do
462− ref <- liftIO $ newIORef ()
463− _ <- liftIO $ mkWeakIORef ref hook
466− -- We pass a copy of sv to fromStreamVar, so that we know that it has
467− -- no other references, when that copy gets garbage collected "ref"
468− -- will get garbage collected and our hook will be called.
469− K.foldStreamShared st yld sng stp $
470− fromChannelRaw sv{svarRef = Just ref}
474− when (svarInspectMode sv) $ do
475− r <- liftIO $ readIORef (svarStopTime (svarStats sv))
476− when (isNothing r) $
477− printSVar sv "SVar Garbage Collected"
479− -- If there are any SVars referenced by this SVar a GC will prompt
480− -- them to be cleaned up quickly.
481− when (svarInspectMode sv) performMajorGC
483−-- | Generate a stream of results from concurrent evaluations from a channel.
484−-- Evaluation of the channel does not start until this API is called. This API
485−-- must not be called more than once on a channel. It kicks off evaluation of
486−-- the channel by dispatching concurrent workers and ensures that as long there
487−-- is work queued on the channel workers are dispatched proportional to the
488−-- demand by the consumer.
490−{-# INLINE fromChannel #-}
491−fromChannel :: MonadAsync m => Channel m a -> Stream m a
492−fromChannel = Stream.fromStreamK . fromChannelK
494−data FromSVarState t m a =
496− | FromSVarRead (Channel m a)
497− | FromSVarLoop (Channel m a) [ChildEvent a]
498− | FromSVarDone (Channel m a)
500−-- | Like 'fromSVar' but generates a StreamD style stream instead of CPS.
502−{-# INLINE_NORMAL _fromChannelD #-}
503−_fromChannelD :: (MonadIO m, MonadThrow m) => Channel m a -> D.Stream m a
504−_fromChannelD svar = D.Stream step FromSVarInit
507− {-# INLINE_LATE step #-}
508− step _ FromSVarInit = do
509− ref <- liftIO $ newIORef ()
510− _ <- liftIO $ mkWeakIORef ref hook
511− -- when this copy of svar gets garbage collected "ref" will get
512− -- garbage collected and our GC hook will be called.
513− let sv = svar{svarRef = Just ref}
514− return $ D.Skip (FromSVarRead sv)
518− {-# NOINLINE hook #-}
520− when (svarInspectMode svar) $ do
521− r <- liftIO $ readIORef (svarStopTime (svarStats svar))
522− when (isNothing r) $
523− printSVar svar "SVar Garbage Collected"
525− -- If there are any SVars referenced by this SVar a GC will prompt
526− -- them to be cleaned up quickly.
527− when (svarInspectMode svar) performMajorGC
529− step _ (FromSVarRead sv) = do
530− list <- readOutputQ sv
531− -- Reversing the output is important to guarantee that we process the
532− -- outputs in the same order as they were generated by the constituent
534− return $ D.Skip $ FromSVarLoop sv (Prelude.reverse list)
536− step _ (FromSVarLoop sv []) = do
537− done <- postProcess sv
538− return $ D.Skip $ if done
539− then FromSVarDone sv
540− else FromSVarRead sv
542− step _ (FromSVarLoop sv (ev : es)) = do
544− ChildYield a -> return $ D.Yield a (FromSVarLoop sv es)
545− ChildStop tid e -> do
546− accountThread sv tid
548− Nothing -> return $ D.Skip (FromSVarLoop sv es)
550− case fromException ex of
552− return $ D.Skip (FromSVarLoop sv es)
553− Nothing -> liftIO (cleanupSVar sv) >> throwM ex
555− step _ (FromSVarDone sv) = do
556− when (svarInspectMode sv) $ do
557− t <- liftIO $ getTime Monotonic
558− liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)
559− liftIO $ printSVar sv "SVar Done"