3014

pending completion

Build # 3014

Build Type

Pull #623

travis-ci-com

Committed by

web-flow

Commit Message

<a href="https://github.com/composewell/streamly/commit/<a class=hub.com/composewell/streamly/commit/39a6719d9f2c15a16ccf917377f12e54db883c20">39a6719d9<a href="https://github.com/composewell/streamly/commit/39a6719d9f2c15a16ccf917377f12e54db883c20">">Merge </a><a class="double-link" href="https://github.com/composewell/streamly/commit/<a class="double-link" href="https://github.com/composewell/streamly/commit/41626c52c2e5d34a0e9aee0a845147f965b1600f">41626c52c</a>">41626c52c</a><a href="https://github.com/composewell/streamly/commit/39a6719d9f2c15a16ccf917377f12e54db883c20"> into 9851d18f9">9851d18f9</a>

Pull Request Pull Request #623: Clean up hlint warnings

Run Details

127 of 127 new or added lines in 9 files covered. (100.0%)

3360 of 9371 relevant lines covered (35.86%)

0.36 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

27.37

/src/Streamly/Internal/Data/Stream/SVar.hs

#ifdef __HADDOCK_VERSION__
#undef INSPECTION
#endif

#ifdef INSPECTION
{-# LANGUAGE TemplateHaskell #-}
{-# OPTIONS_GHC -fplugin Test.Inspection.Plugin #-}
#endif

-- |
-- Module      : Streamly.Internal.Data.Stream.SVar
-- Copyright   : (c) 2017 Harendra Kumar
--
-- License     : BSD3
-- Maintainer  : streamly@composewell.com
-- Stability   : experimental
-- Portability : GHC
--
--
module Streamly.Internal.Data.Stream.SVar
    ( fromSVar
    , fromStreamVar
    , fromProducer
    , fromConsumer
    , toSVar
    , pushToFold
    )
where

import Control.Exception (fromException)
import Control.Monad (when, void)
import Control.Monad.Catch (throwM)
import Control.Monad.IO.Class (MonadIO(liftIO))
import Data.IORef (newIORef, readIORef, mkWeakIORef, writeIORef)
import Data.Maybe (isNothing)
import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
import System.Mem (performMajorGC)

import Streamly.Internal.Data.SVar
import Streamly.Internal.Data.Stream.StreamK hiding (reverse)

#if __GLASGOW_HASKELL__ < 810
#ifdef INSPECTION
import Control.Exception (Exception)
import Control.Monad.Catch (MonadThrow)
import Control.Monad.Trans.Control (MonadBaseControl)
import Data.Typeable (Typeable)
import Test.Inspection (inspect, hasNoTypeClassesExcept)
#endif
#endif

-- | Pull a stream from an SVar.
{-# NOINLINE fromStreamVar #-}
fromStreamVar :: MonadAsync m => SVar Stream m a -> Stream m a
fromStreamVar sv = MkStream $ \st yld sng stp -> do
    list <- readOutputQ sv
    -- Reversing the output is important to guarantee that we process the
    -- outputs in the same order as they were generated by the constituent
    -- streams.
    foldStream st yld sng stp $ processEvents $ reverse list

    where

    allDone stp = do
        when (svarInspectMode sv) $ do
            t <- liftIO $ getTime Monotonic
            liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)
            liftIO $ printSVar sv "SVar Done"
        stp

    {-# INLINE processEvents #-}
    processEvents [] = MkStream $ \st yld sng stp -> do
        done <- postProcess sv
        if done
        then allDone stp
        else foldStream st yld sng stp $ fromStreamVar sv

    processEvents (ev : es) = MkStream $ \st yld sng stp -> do
        let rest = processEvents es
        case ev of
            ChildYield a -> yld a rest
            ChildStop tid e -> do
                accountThread sv tid
                case e of
                    Nothing -> do
                        stop <- shouldStop tid
                        if stop
                        then liftIO (cleanupSVar sv) >> allDone stp
                        else foldStream st yld sng stp rest
                    Just ex ->
                        case fromException ex of
                            Just ThreadAbort ->
                                foldStream st yld sng stp rest
                            Nothing -> liftIO (cleanupSVar sv) >> throwM ex
    shouldStop tid =
        case svarStopStyle sv of
            StopNone -> return False
            StopAny -> return True
            StopBy -> do
                sid <- liftIO $ readIORef (svarStopBy sv)
                return (tid == sid)

#if __GLASGOW_HASKELL__ < 810
#ifdef INSPECTION
-- Use of GHC constraint tuple (GHC.Classes.(%,,%)) in fromStreamVar leads to
-- space leak because the tuple gets allocated in every recursive call and each
-- allocation holds on to the previous allocation. This test is to make sure
-- that we do not use the constraint tuple type class.
--
inspect $ hasNoTypeClassesExcept 'fromStreamVar
    [ ''Monad
    , ''Applicative
    , ''MonadThrow
    , ''Exception
    , ''MonadIO
    , ''MonadBaseControl
    , ''Typeable
    , ''Functor
    ]
#endif
#endif

{-# INLINE fromSVar #-}
fromSVar :: (MonadAsync m, IsStream t) => SVar Stream m a -> t m a
fromSVar sv =
    mkStream $ \st yld sng stp -> do
        ref <- liftIO $ newIORef ()
        _ <- liftIO $ mkWeakIORef ref hook
        -- We pass a copy of sv to fromStreamVar, so that we know that it has
        -- no other references, when that copy gets garbage collected "ref"
        -- will get garbage collected and our hook will be called.
        foldStreamShared st yld sng stp $
            fromStream $ fromStreamVar sv{svarRef = Just ref}
    where

    hook = do
        when (svarInspectMode sv) $ do
            r <- liftIO $ readIORef (svarStopTime (svarStats sv))
            when (isNothing r) $
                printSVar sv "SVar Garbage Collected"
        cleanupSVar sv
        -- If there are any SVars referenced by this SVar a GC will prompt
        -- them to be cleaned up quickly.
        when (svarInspectMode sv) performMajorGC

-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
-- be read back from the SVar using 'fromSVar'.
toSVar :: (IsStream t, MonadAsync m) => SVar Stream m a -> t m a -> m ()
toSVar sv m = toStreamVar sv (toStream m)

-------------------------------------------------------------------------------
-- Process events received by a fold consumer from a stream producer
-------------------------------------------------------------------------------

-- | Pull a stream from an SVar.
{-# NOINLINE fromProducer #-}
fromProducer :: forall m a . MonadAsync m => SVar Stream m a -> Stream m a
fromProducer sv = mkStream $ \st yld sng stp -> do
    list <- readOutputQ sv
    -- Reversing the output is important to guarantee that we process the
    -- outputs in the same order as they were generated by the constituent
    -- streams.
    foldStream st yld sng stp $ processEvents $ reverse list

    where

    allDone :: m r -> m r
    allDone stp = do
        when (svarInspectMode sv) $ do
            t <- liftIO $ getTime Monotonic
            liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)
            liftIO $ printSVar sv "SVar Done"
        sendStopToProducer sv
        stp

    {-# INLINE processEvents #-}
    processEvents :: [ChildEvent a] -> Stream m a
    processEvents [] = mkStream $ \st yld sng stp -> do
        foldStream st yld sng stp $ fromProducer sv

    processEvents (ev : es) = mkStream $ \_ yld _ stp -> do
        let rest = processEvents es
        case ev of
            ChildYield a -> yld a rest
            ChildStop tid e -> do
                accountThread sv tid
                case e of
                    Nothing -> allDone stp
                    Just _ -> error "Bug: fromProducer: received exception"

-------------------------------------------------------------------------------
-- Process events received by the producer thread from the consumer side
-------------------------------------------------------------------------------

-- XXX currently only one event is sent by a fold consumer to the stream
-- producer. But we can potentially have multiple events e.g. the fold step can
-- generate exception more than once and the producer can ignore those
-- exceptions or handle them and still keep driving the fold.
--
{-# NOINLINE fromConsumer #-}
fromConsumer :: MonadAsync m => SVar Stream m a -> m Bool
fromConsumer sv = do
    (list, _) <- liftIO $ readOutputQBasic (outputQueueFromConsumer sv)
    -- Reversing the output is important to guarantee that we process the
    -- outputs in the same order as they were generated by the constituent
    -- streams.
    processEvents $ reverse list

    where

    {-# INLINE processEvents #-}
    processEvents [] = return False
    processEvents (ev : _) = do
        case ev of
            ChildStop _ e -> do
                case e of
                    Nothing -> return True
                    Just ex -> throwM ex
            ChildYield _ -> error "Bug: fromConsumer: invalid ChildYield event"

-- push values to a fold worker via an SVar. Returns whether the fold is done.
{-# INLINE pushToFold #-}
pushToFold :: MonadAsync m => SVar Stream m a -> a -> m Bool
pushToFold sv a = do
    -- Check for exceptions before decrement so that we do not
    -- block forever if the child already exited with an exception.
    --
    -- We avoid a race between the consumer fold sending an event and we
    -- blocking on decrementBufferLimit by waking up the producer thread in
    -- sendToProducer before any event is sent by the fold to the producer
    -- stream.
    let qref = outputQueueFromConsumer sv
    done <- do
        (_, n) <- liftIO $ readIORef qref
        if n > 0
        then fromConsumer sv
        else return False
    if done
    then return True
    else liftIO $ do
        decrementBufferLimit sv
        void $ send sv (ChildYield a)
        return False

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
10	-- \|
11	-- Module : Streamly.Internal.Data.Stream.SVar
12	-- Copyright : (c) 2017 Harendra Kumar
13	--
14	-- License : BSD3
15	-- Maintainer : streamly@composewell.com
16	-- Stability : experimental
17	-- Portability : GHC
18	--
19	--
20	module Streamly.Internal.Data.Stream.SVar
21	( fromSVar
22	, fromStreamVar
23	, fromProducer
24	, fromConsumer
25	, toSVar
26	, pushToFold
27	)
28	where
29
30	import Control.Exception (fromException)
31	import Control.Monad (when, void)
32	import Control.Monad.Catch (throwM)
33	import Control.Monad.IO.Class (MonadIO(liftIO))
34	import Data.IORef (newIORef, readIORef, mkWeakIORef, writeIORef)
35	import Data.Maybe (isNothing)
36	import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
37	import System.Mem (performMajorGC)
38
39	import Streamly.Internal.Data.SVar
40	import Streamly.Internal.Data.Stream.StreamK hiding (reverse)
41
42	#if __GLASGOW_HASKELL__ < 810
43	#ifdef INSPECTION
44	import Control.Exception (Exception)
45	import Control.Monad.Catch (MonadThrow)
46	import Control.Monad.Trans.Control (MonadBaseControl)
47	import Data.Typeable (Typeable)
48	import Test.Inspection (inspect, hasNoTypeClassesExcept)
49	#endif
50	#endif
51
52	-- \| Pull a stream from an SVar.
53	{-# NOINLINE fromStreamVar #-}
54	fromStreamVar :: MonadAsync m => SVar Stream m a -> Stream m a
55	fromStreamVar sv = MkStream $ \st yld sng stp -> do	1✔
56	list <- readOutputQ sv	1✔
57	-- Reversing the output is important to guarantee that we process the
58	-- outputs in the same order as they were generated by the constituent
59	-- streams.
60	foldStream st yld sng stp $ processEvents $ reverse list	×
61
62	where
63
64	allDone stp = do	1✔
65	when (svarInspectMode sv) $ do	×
66	t <- liftIO $ getTime Monotonic	×
67	liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)	×
68	liftIO $ printSVar sv "SVar Done"	×
69	stp	1✔
70
71	{-# INLINE processEvents #-}
72	processEvents [] = MkStream $ \st yld sng stp -> do	1✔
73	done <- postProcess sv	1✔
74	if done	1✔
75	then allDone stp	1✔
76	else foldStream st yld sng stp $ fromStreamVar sv	×
77
78	processEvents (ev : es) = MkStream $ \st yld sng stp -> do	1✔
79	let rest = processEvents es	1✔
80	case ev of	1✔
81	ChildYield a -> yld a rest	1✔
82	ChildStop tid e -> do	1✔
83	accountThread sv tid	1✔
84	case e of	1✔
85	Nothing -> do	1✔
86	stop <- shouldStop tid	×
87	if stop	×
88	then liftIO (cleanupSVar sv) >> allDone stp	×
89	else foldStream st yld sng stp rest	×
90	Just ex ->
91	case fromException ex of	1✔
92	Just ThreadAbort ->
93	foldStream st yld sng stp rest	×
94	Nothing -> liftIO (cleanupSVar sv) >> throwM ex	1✔
95	shouldStop tid =	1✔
96	case svarStopStyle sv of	1✔
97	StopNone -> return False	1✔
98	StopAny -> return True	×
99	StopBy -> do	×
100	sid <- liftIO $ readIORef (svarStopBy sv)	×
101	return (tid == sid)	×
102
103	#if __GLASGOW_HASKELL__ < 810
104	#ifdef INSPECTION
105	-- Use of GHC constraint tuple (GHC.Classes.(%,,%)) in fromStreamVar leads to
106	-- space leak because the tuple gets allocated in every recursive call and each
107	-- allocation holds on to the previous allocation. This test is to make sure
108	-- that we do not use the constraint tuple type class.
109	--
110	inspect $ hasNoTypeClassesExcept 'fromStreamVar
111	[ ''Monad
112	, ''Applicative
113	, ''MonadThrow
114	, ''Exception
115	, ''MonadIO
116	, ''MonadBaseControl
117	, ''Typeable
118	, ''Functor
119	]
120	#endif
121	#endif
122
123	{-# INLINE fromSVar #-}
124	fromSVar :: (MonadAsync m, IsStream t) => SVar Stream m a -> t m a
125	fromSVar sv =	1✔
126	mkStream $ \st yld sng stp -> do	1✔
127	ref <- liftIO $ newIORef ()	×
128	_ <- liftIO $ mkWeakIORef ref hook	1✔
129	-- We pass a copy of sv to fromStreamVar, so that we know that it has
130	-- no other references, when that copy gets garbage collected "ref"
131	-- will get garbage collected and our hook will be called.
132	foldStreamShared st yld sng stp $	×
133	fromStream $ fromStreamVar sv{svarRef = Just ref}	×
134	where
135
136	hook = do	1✔
137	when (svarInspectMode sv) $ do	×
138	r <- liftIO $ readIORef (svarStopTime (svarStats sv))	×
139	when (isNothing r) $	×
140	printSVar sv "SVar Garbage Collected"	×
141	cleanupSVar sv	1✔
142	-- If there are any SVars referenced by this SVar a GC will prompt
143	-- them to be cleaned up quickly.
144	when (svarInspectMode sv) performMajorGC	×
145
146	-- \| Write a stream to an 'SVar' in a non-blocking manner. The stream can then
147	-- be read back from the SVar using 'fromSVar'.
148	toSVar :: (IsStream t, MonadAsync m) => SVar Stream m a -> t m a -> m ()
149	toSVar sv m = toStreamVar sv (toStream m)	×
150
151	-------------------------------------------------------------------------------
152	-- Process events received by a fold consumer from a stream producer
153	-------------------------------------------------------------------------------
154
155	-- \| Pull a stream from an SVar.
156	{-# NOINLINE fromProducer #-}
157	fromProducer :: forall m a . MonadAsync m => SVar Stream m a -> Stream m a
158	fromProducer sv = mkStream $ \st yld sng stp -> do	×
159	list <- readOutputQ sv	×
160	-- Reversing the output is important to guarantee that we process the
161	-- outputs in the same order as they were generated by the constituent
162	-- streams.
163	foldStream st yld sng stp $ processEvents $ reverse list	×
164
165	where
166
167	allDone :: m r -> m r
168	allDone stp = do	×
169	when (svarInspectMode sv) $ do	×
170	t <- liftIO $ getTime Monotonic	×
171	liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)	×
172	liftIO $ printSVar sv "SVar Done"	×
173	sendStopToProducer sv	×
174	stp	×
175
176	{-# INLINE processEvents #-}
177	processEvents :: [ChildEvent a] -> Stream m a
178	processEvents [] = mkStream $ \st yld sng stp -> do	×
179	foldStream st yld sng stp $ fromProducer sv	×
180
181	processEvents (ev : es) = mkStream $ \_ yld _ stp -> do	×
182	let rest = processEvents es	×
183	case ev of	×
184	ChildYield a -> yld a rest	×
185	ChildStop tid e -> do	×
186	accountThread sv tid	×
187	case e of	×
188	Nothing -> allDone stp	×
189	Just _ -> error "Bug: fromProducer: received exception"	×
190
191	-------------------------------------------------------------------------------
192	-- Process events received by the producer thread from the consumer side
193	-------------------------------------------------------------------------------
194
195	-- XXX currently only one event is sent by a fold consumer to the stream
196	-- producer. But we can potentially have multiple events e.g. the fold step can
197	-- generate exception more than once and the producer can ignore those
198	-- exceptions or handle them and still keep driving the fold.
199	--
200	{-# NOINLINE fromConsumer #-}
201	fromConsumer :: MonadAsync m => SVar Stream m a -> m Bool
202	fromConsumer sv = do	×
203	(list, _) <- liftIO $ readOutputQBasic (outputQueueFromConsumer sv)	×
204	-- Reversing the output is important to guarantee that we process the
205	-- outputs in the same order as they were generated by the constituent
206	-- streams.
207	processEvents $ reverse list	×
208
209	where
210
211	{-# INLINE processEvents #-}
212	processEvents [] = return False	×
213	processEvents (ev : _) = do	×
214	case ev of	×
215	ChildStop _ e -> do	×
216	case e of	×
217	Nothing -> return True	×
218	Just ex -> throwM ex	×
219	ChildYield _ -> error "Bug: fromConsumer: invalid ChildYield event"	×
220
221	-- push values to a fold worker via an SVar. Returns whether the fold is done.
222	{-# INLINE pushToFold #-}
223	pushToFold :: MonadAsync m => SVar Stream m a -> a -> m Bool
224	pushToFold sv a = do	×
225	-- Check for exceptions before decrement so that we do not
226	-- block forever if the child already exited with an exception.
227	--
228	-- We avoid a race between the consumer fold sending an event and we
229	-- blocking on decrementBufferLimit by waking up the producer thread in
230	-- sendToProducer before any event is sent by the fold to the producer
231	-- stream.
232	let qref = outputQueueFromConsumer sv	×
233	done <- do	×
234	(_, n) <- liftIO $ readIORef qref	×
235	if n > 0	×
236	then fromConsumer sv	×
237	else return False	×
238	if done	×
239	then return True	×
240	else liftIO $ do	×
241	decrementBufferLimit sv	×
242	void $ send sv (ChildYield a)	×
243	return False	×

composewell / streamly / 3014

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous