@@ -1,14 +1,5 @@
1−#ifdef __HADDOCK_VERSION__
2−#undef INSPECTION
3−#endif
4−
5−#ifdef INSPECTION
6−{-# LANGUAGE TemplateHaskell #-}
7−{-# OPTIONS_GHC -fplugin Test.Inspection.Plugin #-}
8−#endif
9−
101-- |
11−-- Module : Streamly.Internal.Data.Stream.Async.Channel.Core
2+-- Module : Streamly.Internal.Data.Stream.Async.Channel.Append
123-- Copyright : (c) 2017 Composewell Technologies
134-- License : BSD-3-Clause
145-- Maintainer : streamly@composewell.com
@@ -22,38 +13,23 @@
2213-- use INLINABLE and SPECIALIZE on it which makes it specialized but it is not
2314-- actually inlined.
2415
25−module Streamly.Internal.Data.Stream.Async.Channel.Core
16+module Streamly.Internal.Data.Stream.Async.Channel.Append
2617 (
2718 newChannel
28− , toChannel
29− , toChannelK
30− , fromChannel
31− , fromChannelK
3219 )
3320where
3421
35−#include "inline.hs"
36−
3722import Control.Concurrent (myThreadId)
3823import Control.Concurrent.MVar (newEmptyMVar)
39−import Control.Exception (fromException)
40−import Control.Monad (when)
41−import Control.Monad.Catch (throwM, MonadThrow)
4224import Control.Monad.IO.Class (MonadIO(liftIO))
4325import Control.Monad.Trans.Control (MonadBaseControl, restoreM)
44−import Data.IORef (IORef, newIORef, readIORef, mkWeakIORef, writeIORef)
45−import Data.Maybe (isNothing)
26+import Data.IORef (IORef, newIORef, readIORef)
4627import Streamly.Internal.Control.Concurrent
4728 (MonadRunInIO, MonadAsync, RunInIO(..), askRunInIO)
4829import Streamly.Internal.Data.Atomics
4930 (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)
5331
5432import 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
5733import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
5834
5935import Streamly.Internal.Data.Stream.Async.Channel.Consumer
@@ -61,18 +37,6 @@ import Streamly.Internal.Data.Stream.Async.Channel.Dispatcher
6137import Streamly.Internal.Data.Stream.Async.Channel.Type
6238import Streamly.Internal.Data.Stream.Async.Channel.Worker
6339
64−import Prelude hiding (map, concat, concatMap)
65−
66−#if __GLASGOW_HASKELL__ < 810
67−#ifdef INSPECTION
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)
73−#endif
74−#endif
75−
7640------------------------------------------------------------------------------
7741-- Creating a channel
7842------------------------------------------------------------------------------
@@ -263,7 +227,6 @@ getLifoSVar mrun cfg = do
263227 -> (Channel m a -> m Bool)
264228 -> (Channel m a -> IO Bool)
265229 -> (IORef [(RunInIO m, K.Stream m a)]
266− -- -> State Stream m a
267230 -> Channel m a
268231 -> Maybe WorkerInfo
269232 -> m())
@@ -317,11 +280,6 @@ getLifoSVar mrun cfg = do
317280 workLoopLIFOLimited
318281 in return sv
319282
320−-- type Channel m a = SVar K.Stream m a
321−-- data Config = Config
322−
323−-- XXX openChannel?
324−
325283-- | Create a new async style concurrent stream evaluation channel. The monad
326284-- state used to run the stream actions is taken from the call site of
327285-- newChannel.
@@ -332,229 +290,3 @@ newChannel :: (MonadAsync m) =>
332290newChannel modifier = do
333291 mrun <- askRunInIO
334292 liftIO $ getLifoSVar mrun (modifier defaultConfig)
335−
336−------------------------------------------------------------------------------
337−-- Generating streams from a channel
338−------------------------------------------------------------------------------
339−
340−-- $concurrentEval
341−--
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.
345−--
346−-- @
347−-- input stream
348−-- |
349−-- <-----------------|<--------worker
350−-- | exceptions |
351−-- output stream <---Channel<------worker
352−-- |
353−-- |<--------worker
354−--
355−-- @
356−--
357−-- The puller itself schedules the worker threads based on demand.
358−-- Exceptions are propagated from the worker threads to the puller.
359−
360−-------------------------------------------------------------------------------
361−-- Write a stream to a channel
362−-------------------------------------------------------------------------------
363−
364−-- XXX Should be a Fold, singleton API could be called joinChannel, or the fold
365−-- can be called joinChannel.
366−
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 ()
371−toChannelK sv m = do
372− runIn <- askRunInIO
373− liftIO $ enqueue sv (runIn, m)
374−
375−-- INLINE for fromStreamK/toStreamK fusion
376−
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
382−
383−{-
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
388−-}
389−
390−-------------------------------------------------------------------------------
391−-- Read a stream from a channel
392−-------------------------------------------------------------------------------
393−
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
401− -- streams.
402− K.foldStream st yld sng stp $ processEvents $ reverse list
403−
404− where
405−
406− allDone stp = do
407− when (svarInspectMode sv) $ do
408− t <- liftIO $ getTime Monotonic
409− liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)
410− liftIO $ printSVar sv "SVar Done"
411− stp
412−
413− {-# INLINE processEvents #-}
414− processEvents [] = K.MkStream $ \st yld sng stp -> do
415− done <- postProcess sv
416− if done
417− then allDone stp
418− else K.foldStream st yld sng stp $ fromChannelRaw sv
419−
420− processEvents (ev : es) = K.MkStream $ \st yld sng stp -> do
421− let rest = processEvents es
422− case ev of
423− ChildYield a -> yld a rest
424− ChildStop tid e -> do
425− accountThread sv tid
426− case e of
427− Nothing -> K.foldStream st yld sng stp rest
428− Just ex ->
429− case fromException ex of
430− Just ThreadAbort ->
431− K.foldStream st yld sng stp rest
432− Nothing -> liftIO (cleanupSVar sv) >> throwM ex
433−
434−#if __GLASGOW_HASKELL__ < 810
435−#ifdef INSPECTION
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.
440−--
441−inspect $ hasNoTypeClassesExcept 'fromStreamVar
442− [ ''Monad
443− , ''Applicative
444− , ''MonadThrow
445− , ''Exception
446− , ''MonadIO
447− , ''MonadBaseControl
448− , ''Typeable
449− , ''Functor
450− ]
451−#endif
452−#endif
453−
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?
457−
458−{-# INLINE fromChannelK #-}
459−fromChannelK :: MonadAsync m => Channel m a -> K.Stream m a
460−fromChannelK sv =
461− K.mkStream $ \st yld sng stp -> do
462− ref <- liftIO $ newIORef ()
463− _ <- liftIO $ mkWeakIORef ref hook
464−
465− startChannel sv
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}
471− where
472−
473− hook = do
474− when (svarInspectMode sv) $ do
475− r <- liftIO $ readIORef (svarStopTime (svarStats sv))
476− when (isNothing r) $
477− printSVar sv "SVar Garbage Collected"
478− cleanupSVar sv
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
482−
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.
489−--
490−{-# INLINE fromChannel #-}
491−fromChannel :: MonadAsync m => Channel m a -> Stream m a
492−fromChannel = Stream.fromStreamK . fromChannelK
493−
494−data FromSVarState t m a =
495− FromSVarInit
496− | FromSVarRead (Channel m a)
497− | FromSVarLoop (Channel m a) [ChildEvent a]
498− | FromSVarDone (Channel m a)
499−
500−-- | Like 'fromSVar' but generates a StreamD style stream instead of CPS.
501−--
502−{-# INLINE_NORMAL _fromChannelD #-}
503−_fromChannelD :: (MonadIO m, MonadThrow m) => Channel m a -> D.Stream m a
504−_fromChannelD svar = D.Stream step FromSVarInit
505− where
506−
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)
515−
516− where
517−
518− {-# NOINLINE hook #-}
519− hook = do
520− when (svarInspectMode svar) $ do
521− r <- liftIO $ readIORef (svarStopTime (svarStats svar))
522− when (isNothing r) $
523− printSVar svar "SVar Garbage Collected"
524− cleanupSVar svar
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
528−
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
533− -- streams.
534− return $ D.Skip $ FromSVarLoop sv (Prelude.reverse list)
535−
536− step _ (FromSVarLoop sv []) = do
537− done <- postProcess sv
538− return $ D.Skip $ if done
539− then FromSVarDone sv
540− else FromSVarRead sv
541−
542− step _ (FromSVarLoop sv (ev : es)) = do
543− case ev of
544− ChildYield a -> return $ D.Yield a (FromSVarLoop sv es)
545− ChildStop tid e -> do
546− accountThread sv tid
547− case e of
548− Nothing -> return $ D.Skip (FromSVarLoop sv es)
549− Just ex ->
550− case fromException ex of
551− Just ThreadAbort ->
552− return $ D.Skip (FromSVarLoop sv es)
553− Nothing -> liftIO (cleanupSVar sv) >> throwM ex
554−
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"
560− return D.Stop