git.iko.soy/streamly
on 6b192fb help sign in
streamly/src/Streamly/Streams/SVar.hs
143 lines 5.3 kB history raw
1{-# LANGUAGE CPP #-}
2{-# LANGUAGE ConstraintKinds #-}
3{-# LANGUAGE FlexibleContexts #-}
4{-# LANGUAGE FlexibleInstances #-}
5{-# LANGUAGE LambdaCase #-}
6{-# LANGUAGE MagicHash #-}
7{-# LANGUAGE MultiParamTypeClasses #-}
8{-# LANGUAGE RankNTypes #-}
9{-# LANGUAGE UnboxedTuples #-}
10{-# LANGUAGE UndecidableInstances #-} -- XXX
12#include "inline.h"
14-- |
15-- Module : Streamly.Streams.SVar
16-- Copyright : (c) 2017 Harendra Kumar
17--
18-- License : BSD3
19-- Maintainer : harendra.kumar@gmail.com
20-- Stability : experimental
21-- Portability : GHC
22--
23--
24module Streamly.Streams.SVar
25 (
26 fromSVar
27 , toSVar
28 , maxThreads
29 , maxBuffer
30 , maxYields
31 )
32where
34import Control.Monad.Catch (throwM)
36import Streamly.SVar
37import Streamly.Streams.StreamK
38import Streamly.Streams.Serial (SerialT)
40-- MVar diagnostics has some overhead - around 5% on asyncly null benchmark, we
41-- can keep it on in production to debug problems quickly if and when they
42-- happen, but it may result in unexpected output when threads are left hanging
43-- until they are GCed because the consumer went away.
45-- | Pull a stream from an SVar.
46{-# NOINLINE fromStreamVar #-}
47fromStreamVar :: MonadAsync m => SVar Stream m a -> Stream m a
48fromStreamVar sv = Stream $ \st stp sng yld -> do
49 list <- readOutputQ sv
50 -- Reversing the output is important to guarantee that we process the
51 -- outputs in the same order as they were generated by the constituent
52 -- streams.
53 unStream (processEvents $ reverse list) (rstState st) stp sng yld
55 where
57 allDone stp = do
58#ifdef DIAGNOSTICS
59#ifdef DIAGNOSTICS_VERBOSE
60 svInfo <- liftIO $ dumpSVar sv
61 liftIO $ hPutStrLn stderr $ "fromStreamVar done\n" ++ svInfo
62#endif
63#endif
64 stp
66 {-# INLINE processEvents #-}
67 processEvents [] = Stream $ \st stp sng yld -> do
68 done <- postProcess sv
69 if done
70 then allDone stp
71 else unStream (fromStreamVar sv) (rstState st) stp sng yld
73 processEvents (ev : es) = Stream $ \st stp sng yld -> do
74 let rest = processEvents es
75 case ev of
76 ChildYield a -> yld a rest
77 ChildStop tid e -> do
78 accountThread sv tid
79 case e of
80 Nothing -> unStream rest (rstState st) stp sng yld
81 Just ex -> throwM ex
83{-# INLINE fromSVar #-}
84fromSVar :: (MonadAsync m, IsStream t) => SVar Stream m a -> t m a
85fromSVar sv = fromStream $ fromStreamVar sv
87-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
88-- be read back from the SVar using 'fromSVar'.
89toSVar :: (IsStream t, MonadAsync m) => SVar Stream m a -> t m a -> m ()
90toSVar sv m = toStreamVar sv (toStream m)
92-------------------------------------------------------------------------------
93-- Concurrency control
94-------------------------------------------------------------------------------
95--
96-- XXX need to write these in direct style otherwise they will break fusion.
97--
98-- | Specify the maximum number of threads that can be spawned concurrently
99-- when using concurrent streams. This is not the grand total number of threads
100-- but the maximum number of threads at each point of concurrency.
101-- A value of 0 resets the thread limit to default, a negative value means
102-- there is no limit. The default value is 1500.
103--
104-- @since 0.4.0
105{-# INLINE_NORMAL maxThreads #-}
106maxThreads :: IsStream t => Int -> t m a -> t m a
107maxThreads n m = fromStream $ Stream $ \st stp sng yld -> do
108 let n' = if n == 0 then defaultMaxThreads else n
109 unStream (toStream m) (st {threadsHigh = n'}) stp sng yld
111{-# RULES "maxThreadsSerial serial" maxThreads = maxThreadsSerial #-}
112maxThreadsSerial :: Int -> SerialT m a -> SerialT m a
113maxThreadsSerial _ = id
115-- | Specify the maximum size of the buffer for storing the results from
116-- concurrent computations. If the buffer becomes full we stop spawning more
117-- concurrent tasks until there is space in the buffer.
118-- A value of 0 resets the buffer size to default, a negative value means
119-- there is no limit. The default value is 1500.
120--
121-- @since 0.4.0
122{-# INLINE_NORMAL maxBuffer #-}
123maxBuffer :: IsStream t => Int -> t m a -> t m a
124maxBuffer n m = fromStream $ Stream $ \st stp sng yld -> do
125 let n' = if n == 0 then defaultMaxBuffer else n
126 unStream (toStream m) (st {bufferHigh = n'}) stp sng yld
128{-# RULES "maxBuffer serial" maxBuffer = maxBufferSerial #-}
129maxBufferSerial :: Int -> SerialT m a -> SerialT m a
130maxBufferSerial _ = id
132-- Stop concurrent dispatches after this limit. This is useful in API's like
133-- "take" where we want to dispatch only upto the number of elements "take"
134-- needs. This value applies only to the immediate next level and is not
135-- inherited by everything in enclosed scope.
136{-# INLINE_NORMAL maxYields #-}
137maxYields :: IsStream t => Maybe Int -> t m a -> t m a
138maxYields n m = fromStream $ Stream $ \st stp sng yld -> do
139 unStream (toStream m) (st {yieldLimit = n}) stp sng yld
141{-# RULES "maxYields serial" maxYields = maxYieldsSerial #-}
142maxYieldsSerial :: Maybe Int -> SerialT m a -> SerialT m a
143maxYieldsSerial _ = id