git.iko.soy/streamly
on 43dda21 help sign in
streamly/src/Streamly/Internal/Data/Fold/SVar.hs
78 lines 2.3 kB history raw
1-- |
2-- Module : Streamly.Internal.Data.Fold.SVar
3-- Copyright : (c) 2017 Composewell Technologies
4-- License : BSD-3-Clause
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
7-- Portability : GHC
8--
9--
10module Streamly.Internal.Data.Fold.SVar
11 (
12 write
13 , writeLimited
14 )
15where
17#include "inline.hs"
19import Control.Monad (void)
20import Control.Monad.IO.Class (MonadIO(liftIO))
21import Streamly.Internal.Data.Fold.Type (Fold(..))
23import qualified Streamly.Internal.Data.Fold.Type as FL
25import Streamly.Internal.Data.SVar
27-- | A fold to write a stream to an SVar. Unlike 'toSVar' this does not allow
28-- for concurrent evaluation of the stream, as the fold receives the input one
29-- element at a time, it just forwards the elements to the SVar. However, we
30-- can safely execute the fold in an independent thread, the SVar can act as a
31-- buffer decoupling the sender from the receiver. Also, we can have multiple
32-- folds running concurrently pusing the streams to the SVar.
33--
34{-# INLINE write #-}
35write :: MonadIO m => SVar t m a -> Maybe WorkerInfo -> Fold m a ()
36write svar winfo = Fold step initial extract
38 where
40 initial = return $ FL.Partial ()
42 -- XXX we can have a separate fold for unlimited buffer case to avoid a
43 -- branch in the step here.
44 step () x =
45 liftIO $ do
46 decrementBufferLimit svar
47 void $ send svar (ChildYield x)
48 return $ FL.Partial ()
50 extract () = liftIO $ sendStop svar winfo
52-- | Like write, but applies a yield limit.
53--
54{-# INLINE writeLimited #-}
55writeLimited :: MonadIO m
56 => SVar t m a -> Maybe WorkerInfo -> Fold m a ()
57writeLimited svar winfo = Fold step initial extract
59 where
61 initial = return $ FL.Partial True
63 step True x =
64 liftIO $ do
65 yieldLimitOk <- decrementYieldLimit svar
66 if yieldLimitOk
67 then do
68 decrementBufferLimit svar
69 void $ send svar (ChildYield x)
70 return $ FL.Partial True
71 else do
72 cleanupSVarFromWorker svar
73 sendStop svar winfo
74 return $ FL.Done ()
75 step False _ = return $ FL.Done ()
77 extract True = liftIO $ sendStop svar winfo
78 extract False = return ()