composewell / streamly / 3171

#include "inline.hs"

-- |

-- Module      : Streamly.Internal.Data.Stream.Combinators

-- Copyright   : (c) 2017 Harendra Kumar

--

-- License     : BSD3

-- Maintainer  : streamly@composewell.com

-- Stability   : experimental

-- Portability : GHC

--

--

module Streamly.Internal.Data.Stream.Combinators

    ( maxThreads

    , maxBuffer

    , maxYields

    , rate

    , avgRate

    , minRate

    , maxRate

    , constRate

    , inspectMode

    , printState

    )

where

import Control.Monad.IO.Class (MonadIO(liftIO))

import Data.Int (Int64)

import Streamly.Internal.Data.SVar

import Streamly.Internal.Data.Stream.StreamK

import Streamly.Internal.Data.Stream.Serial (SerialT)

-------------------------------------------------------------------------------

-- Concurrency control

-------------------------------------------------------------------------------

--

-- XXX need to write these in direct style otherwise they will break fusion.

--

-- | Specify the maximum number of threads that can be spawned concurrently for

-- any concurrent combinator in a stream.

-- A value of 0 resets the thread limit to default, a negative value means

-- there is no limit. The default value is 1500. 'maxThreads' does not affect

-- 'ParallelT' streams as they can use unbounded number of threads.

--

-- When the actions in a stream are IO bound, having blocking IO calls, this

-- option can be used to control the maximum number of in-flight IO requests.

-- When the actions are CPU bound this option can be used to

-- control the amount of CPU used by the stream.

--

-- @since 0.8.0

--

-- /Since: 0.4.0 ("Streamly")/

{-# INLINE_NORMAL maxThreads #-}

maxThreads :: IsStream t => Int -> t m a -> t m a

{-

{-# RULES "maxThreadsSerial serial" maxThreads = maxThreadsSerial #-}

maxThreadsSerial :: Int -> SerialT m a -> SerialT m a

maxThreadsSerial _ = id

-}

-- | Specify the maximum size of the buffer for storing the results from

-- concurrent computations. If the buffer becomes full we stop spawning more

-- concurrent tasks until there is space in the buffer.

-- A value of 0 resets the buffer size to default, a negative value means

-- there is no limit. The default value is 1500.

--

-- CAUTION! using an unbounded 'maxBuffer' value (i.e. a negative value)

-- coupled with an unbounded 'maxThreads' value is a recipe for disaster in

-- presence of infinite streams, or very large streams.  Especially, it must

-- not be used when 'pure' is used in 'ZipAsyncM' streams as 'pure' in

-- applicative zip streams generates an infinite stream causing unbounded

-- concurrent generation with no limit on the buffer or threads.

--

-- @since 0.8.0

--

-- /Since: 0.4.0 ("Streamly")/

{-# INLINE_NORMAL maxBuffer #-}

maxBuffer :: IsStream t => Int -> t m a -> t m a

{-

{-# RULES "maxBuffer serial" maxBuffer = maxBufferSerial #-}

maxBufferSerial :: Int -> SerialT m a -> SerialT m a

maxBufferSerial _ = id

-}

-- | Specify the pull rate of a stream.

-- A 'Nothing' value resets the rate to default which is unlimited.  When the

-- rate is specified, concurrent production may be ramped up or down

-- automatically to achieve the specified yield rate. The specific behavior for

-- different styles of 'Rate' specifications is documented under 'Rate'.  The

-- effective maximum production rate achieved by a stream is governed by:

--

-- * The 'maxThreads' limit

-- * The 'maxBuffer' limit

-- * The maximum rate that the stream producer can achieve

-- * The maximum rate that the stream consumer can achieve

--

-- @since 0.8.0

--

-- /Since: 0.5.0 ("Streamly")/

{-# INLINE_NORMAL rate #-}

rate :: IsStream t => Maybe Rate -> t m a -> t m a

-- XXX implement for serial streams as well, as a simple delay

{-

{-# RULES "rate serial" rate = yieldRateSerial #-}

yieldRateSerial :: Double -> SerialT m a -> SerialT m a

yieldRateSerial _ = id

-}

-- | Same as @rate (Just $ Rate (r/2) r (2*r) maxBound)@

--

-- Specifies the average production rate of a stream in number of yields

-- per second (i.e.  @Hertz@).  Concurrent production is ramped up or down

-- automatically to achieve the specified average yield rate. The rate can

-- go down to half of the specified rate on the lower side and double of

-- the specified rate on the higher side.

--

-- @since 0.8.0

--

-- /Since: 0.5.0 ("Streamly")/

avgRate :: IsStream t => Double -> t m a -> t m a

-- | Same as @rate (Just $ Rate r r (2*r) maxBound)@

--

-- Specifies the minimum rate at which the stream should yield values. As

-- far as possible the yield rate would never be allowed to go below the

-- specified rate, even though it may possibly go above it at times, the

-- upper limit is double of the specified rate.

--

-- @since 0.8.0

--

-- /Since: 0.5.0 ("Streamly")/

minRate :: IsStream t => Double -> t m a -> t m a

-- | Same as @rate (Just $ Rate (r/2) r r maxBound)@

--

-- Specifies the maximum rate at which the stream should yield values. As

-- far as possible the yield rate would never be allowed to go above the

-- specified rate, even though it may possibly go below it at times, the

-- lower limit is half of the specified rate. This can be useful in

-- applications where certain resource usage must not be allowed to go

-- beyond certain limits.

--

-- @since 0.8.0

--

-- /Since: 0.5.0 ("Streamly")/

maxRate :: IsStream t => Double -> t m a -> t m a

-- | Same as @rate (Just $ Rate r r r 0)@

--

-- Specifies a constant yield rate. If for some reason the actual rate

-- goes above or below the specified rate we do not try to recover it by

-- increasing or decreasing the rate in future.  This can be useful in

-- applications like graphics frame refresh where we need to maintain a

-- constant refresh rate.

--

-- @since 0.8.0

--

-- /Since: 0.5.0 ("Streamly")/

constRate :: IsStream t => Double -> t m a -> t m a

-- | Specify the average latency, in nanoseconds, of a single threaded action

-- in a concurrent composition. Streamly can measure the latencies, but that is

-- possible only after at least one task has completed. This combinator can be

-- used to provide a latency hint so that rate control using 'rate' can take

-- that into account right from the beginning. When not specified then a

-- default behavior is chosen which could be too slow or too fast, and would be

-- restricted by any other control parameters configured.

-- A value of 0 indicates default behavior, a negative value means there is no

-- limit i.e. zero latency.

-- This would normally be useful only in high latency and high throughput

-- cases.

--

{-# INLINE_NORMAL _serialLatency #-}

_serialLatency :: IsStream t => Int -> t m a -> t m a

{-

{-# RULES "serialLatency serial" _serialLatency = serialLatencySerial #-}

serialLatencySerial :: Int -> SerialT m a -> SerialT m a

serialLatencySerial _ = id

-}

-- Stop concurrent dispatches after this limit. This is useful in API's like

-- "take" where we want to dispatch only upto the number of elements "take"

-- needs.  This value applies only to the immediate next level and is not

-- inherited by everything in enclosed scope.

{-# INLINE_NORMAL maxYields #-}

maxYields :: IsStream t => Maybe Int64 -> t m a -> t m a

{-# RULES "maxYields serial" maxYields = maxYieldsSerial #-}

maxYieldsSerial :: Maybe Int64 -> SerialT m a -> SerialT m a

printState :: MonadIO m => State Stream m a -> m ()

-- | Print debug information about an SVar when the stream ends

--

-- /Internal/

--

inspectMode :: IsStream t => t m a -> t m a