composewell / streamly / 2967

{-# LANGUAGE UnboxedTuples #-}

-- |

-- Module      : Streamly.Internal.Data.SVar

-- Copyright   : (c) 2017 Harendra Kumar

--

-- License     : BSD3

-- Maintainer  : streamly@composewell.com

-- Stability   : experimental

-- Portability : GHC

module Streamly.Internal.Data.SVar

    (

      MonadAsync

    , SVarStyle (..)

    , SVarStopStyle (..)

    , SVar (..)

    -- State threaded around the stream

    , Limit (..)

    , State (streamVar)

    , defState

    , adaptState

    , getMaxThreads

    , setMaxThreads

    , getMaxBuffer

    , setMaxBuffer

    , getStreamRate

    , setStreamRate

    , setStreamLatency

    , getYieldLimit

    , setYieldLimit

    , getInspectMode

    , setInspectMode

    , recordMaxWorkers

    , cleanupSVar

    , cleanupSVarFromWorker

    -- SVar related

    , newAheadVar

    , newParallelVar

    , captureMonadState

    , RunInIO (..)

    , WorkerInfo (..)

    , YieldRateInfo (..)

    , ThreadAbort (..)

    , ChildEvent (..)

    , AheadHeapEntry (..)

    , send

    , sendToProducer

    , sendYield

    , sendStop

    , sendStopToProducer

    , enqueueLIFO

    , enqueueFIFO

    , enqueueAhead

    , reEnqueueAhead

    , pushWorkerPar

    , handleChildException

    , handleFoldException

    , queueEmptyAhead

    , dequeueAhead

    , HeapDequeueResult(..)

    , dequeueFromHeap

    , dequeueFromHeapSeq

    , requeueOnHeapTop

    , updateHeapSeq

    , withIORef

    , heapIsSane

    , Rate (..)

    , getYieldRateInfo

    , newSVarStats

    , collectLatency

    , workerUpdateLatency

    , isBeyondMaxRate

    , workerRateControl

    , updateYieldCount

    , decrementYieldLimit

    , incrementYieldLimit

    , decrementBufferLimit

    , incrementBufferLimit

    , postProcessBounded

    , postProcessPaced

    , readOutputQBounded

    , readOutputQPaced

    , readOutputQBasic

    , dispatchWorkerPaced

    , sendFirstWorker

    , delThread

    , modifyThread

    , doFork

    , fork

    , forkManaged

    , toStreamVar

    , SVarStats (..)

    , dumpSVar

    , printSVar

    , withDiagMVar

    )

where

import Control.Concurrent

       (ThreadId, myThreadId, threadDelay, throwTo, forkIO, killThread)

import Control.Concurrent.MVar

       (MVar, newEmptyMVar, tryPutMVar, takeMVar, tryTakeMVar, newMVar,

        tryReadMVar)

import Control.Exception

       (SomeException(..), catch, mask, assert, Exception, catches,

        throwIO, Handler(..), BlockedIndefinitelyOnMVar(..),

        BlockedIndefinitelyOnSTM(..))

import Control.Monad (when)

import Control.Monad.Catch (MonadThrow)

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

import Control.Monad.Trans.Control

       (MonadBaseControl, control, StM, liftBaseDiscard)

import Streamly.Internal.Data.Atomics

       (atomicModifyIORefCAS, atomicModifyIORefCAS_, writeBarrier,

        storeLoadBarrier)

import Data.Concurrent.Queue.MichaelScott (LinkedQueue, pushL)

import Data.Functor (void)

import Data.Heap (Heap, Entry(..))

import Data.Int (Int64)

#if __GLASGOW_HASKELL__ >= 800

import Data.Kind (Type)

#endif

import Data.IORef

       (IORef, modifyIORef, newIORef, readIORef, writeIORef, atomicModifyIORef)

import Data.Maybe (fromJust, fromMaybe)

#if __GLASGOW_HASKELL__ < 808

import Data.Semigroup ((<>))

#endif

import Data.Set (Set)

import GHC.Conc (ThreadId(..))

import GHC.Exts

import GHC.IO (IO(..))

import System.IO (hPutStrLn, stderr)

import System.Mem.Weak (addFinalizer)

import Streamly.Internal.Data.Time.Clock (Clock(..), getTime)

import Streamly.Internal.Data.Time.Units

       (AbsTime, NanoSecond64(..), MicroSecond64(..), diffAbsTime64,

        fromRelTime64, toRelTime64, showNanoSecond64, showRelTime64)

import qualified Data.Heap as H

import qualified Data.Set                    as S

newtype Count = Count Int64

             )

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

-- Parent child thread communication type

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

instance Exception ThreadAbort

-- | Events that a child thread may send to a parent thread.

data ChildEvent a =

      ChildYield a

    | ChildStop ThreadId (Maybe SomeException)

#if __GLASGOW_HASKELL__ < 800

#define Type *

#endif

-- | Sorting out-of-turn outputs in a heap for Ahead style streams

data AheadHeapEntry (t :: (Type -> Type) -> Type -> Type) m a =

      AheadEntryNull

    | AheadEntryPure a

    | AheadEntryStream (t m a)

#undef Type

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

-- State threaded around the monad for thread management

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

-- | Identify the type of the SVar. Two computations using the same style can

-- be scheduled on the same SVar.

data SVarStyle =

      AsyncVar             -- depth first concurrent

    | WAsyncVar            -- breadth first concurrent

    | ParallelVar          -- all parallel

    | AheadVar             -- Concurrent look ahead

-- | An SVar or a Stream Var is a conduit to the output from multiple streams

-- running concurrently and asynchronously. An SVar can be thought of as an

-- asynchronous IO handle. We can write any number of streams to an SVar in a

-- non-blocking manner and then read them back at any time at any pace.  The

-- SVar would run the streams asynchronously and accumulate results. An SVar

-- may not really execute the stream completely and accumulate all the results.

-- However, it ensures that the reader can read the results at whatever paces

-- it wants to read. The SVar monitors and adapts to the consumer's pace.

--

-- An SVar is a mini scheduler, it has an associated workLoop that holds the

-- stream tasks to be picked and run by a pool of worker threads. It has an

-- associated output queue where the output stream elements are placed by the

-- worker threads. A outputDoorBell is used by the worker threads to intimate the

-- consumer thread about availability of new results in the output queue. More

-- workers are added to the SVar by 'fromStreamVar' on demand if the output

-- produced is not keeping pace with the consumer. On bounded SVars, workers

-- block on the output queue to provide throttling of the producer  when the

-- consumer is not pulling fast enough.  The number of workers may even get

-- reduced depending on the consuming pace.

--

-- New work is enqueued either at the time of creation of the SVar or as a

-- result of executing the parallel combinators i.e. '<|' and '<|>' when the

-- already enqueued computations get evaluated. See 'joinStreamVarAsync'.

-- We measure the individual worker latencies to estimate the number of workers

-- needed or the amount of time we have to sleep between dispatches to achieve

-- a particular rate when controlled pace mode it used.

data WorkerInfo = WorkerInfo

    -- total number of yields by the worker till now

    -- yieldCount at start, timestamp

    }

-- | Specifies the stream yield rate in yields per second (@Hertz@).

-- We keep accumulating yield credits at 'rateGoal'. At any point of time we

-- allow only as many yields as we have accumulated as per 'rateGoal' since the

-- start of time. If the consumer or the producer is slower or faster, the

-- actual rate may fall behind or exceed 'rateGoal'.  We try to recover the gap

-- between the two by increasing or decreasing the pull rate from the producer.

-- However, if the gap becomes more than 'rateBuffer' we try to recover only as

-- much as 'rateBuffer'.

--

-- 'rateLow' puts a bound on how low the instantaneous rate can go when

-- recovering the rate gap.  In other words, it determines the maximum yield

-- latency.  Similarly, 'rateHigh' puts a bound on how high the instantaneous

-- rate can go when recovering the rate gap.  In other words, it determines the

-- minimum yield latency. We reduce the latency by increasing concurrency,

-- therefore we can say that it puts an upper bound on concurrency.

--

-- If the 'rateGoal' is 0 or negative the stream never yields a value.

-- If the 'rateBuffer' is 0 or negative we do not attempt to recover.

--

-- @since 0.5.0

data Rate = Rate

    }

data LatencyRange = LatencyRange

-- Rate control.

data YieldRateInfo = YieldRateInfo

    -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely

    -- read by the worker threads

    -- Actual latency/througput as seen from the consumer side, we count the

    -- yields and the time it took to generates those yields. This is used to

    -- increase or decrease the number of workers needed to achieve the desired

    -- rate. The idle time of workers is adjusted in this, so that we only

    -- account for the rate when the consumer actually demands data.

    -- XXX interval latency is enough, we can move this under diagnostics build

    -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely

    -- read by the worker threads

    -- XXX Worker latency specified by the user to be used before the first

    -- actual measurement arrives. Not yet implemented

    -- After how many yields the worker should update the latency information.

    -- If the latency is high, this count is kept lower and vice-versa.  XXX If

    -- the latency suddenly becomes too high this count may remain too high for

    -- long time, in such cases the consumer can change it.

    -- 0 means no latency computation

    -- XXX this is derivable from workerMeasuredLatency, can be removed.

    -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely

    -- read by the worker threads

    -- This is in progress latency stats maintained by the workers which we

    -- empty into workerCollectedLatency stats at certain intervals - whenever

    -- we process the stream elements yielded in this period. The first count

    -- is all yields, the second count is only those yields for which the

    -- latency was measured to be non-zero (note that if the timer resolution

    -- is low the measured latency may be zero e.g. on JS platform).

    -- [LOCKING] Locked access. Modified by the consumer thread as well as

    -- worker threads. Workers modify it periodically based on

    -- workerPollingInterval and not on every yield to reduce the locking

    -- overhead.

    -- (allYieldCount, yieldCount, timeTaken)

    -- This is the second level stat which is an accmulation from

    -- workerPendingLatency stats. We keep accumulating latencies in this

    -- bucket until we have stats for a sufficient period and then we reset it

    -- to start collecting for the next period and retain the computed average

    -- latency for the last period in workerMeasuredLatency.

    -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely

    -- read by the worker threads

    -- (allYieldCount, yieldCount, timeTaken)

    -- Latency as measured by workers, aggregated for the last period.

    -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely

    -- read by the worker threads

    }

data SVarStats = SVarStats {

}

-- This is essentially a 'Maybe Word' type

instance Eq Limit where

instance Ord Limit where

-- When to stop the composed stream.

data SVarStopStyle =

      StopNone -- stops only when all streams are finished

    | StopAny  -- stop when any stream finishes

    | StopBy   -- stop when a specific stream finishes

-- | Buffering policy for persistent push workers (in ParallelT).  In a pull

-- style SVar (in AsyncT, AheadT etc.), the consumer side dispatches workers on

-- demand, workers terminate if the buffer is full or if the consumer is not

-- cosuming fast enough.  In a push style SVar, a worker is dispatched only

-- once, workers are persistent and keep pushing work to the consumer via a

-- bounded buffer. If the buffer becomes full the worker either blocks, or it

-- can drop an item from the buffer to make space.

--

-- Pull style SVars are useful in lazy stream evaluation whereas push style

-- SVars are useful in strict left Folds.

--

-- XXX Maybe we can separate the implementation in two different types instead

-- of using a common SVar type.

--

data PushBufferPolicy =

      PushBufferDropNew  -- drop the latest element and continue

    | PushBufferDropOld  -- drop the oldest element and continue

    | PushBufferBlock    -- block the thread until space

                         -- becomes available

-- IMPORTANT NOTE: we cannot update the SVar after generating it as we have

-- references to the original SVar stored in several functions which will keep

-- pointing to the original data and the new updates won't reflect there.

-- Any updateable parts must be kept in mutable references (IORef).

--

data SVar t m a = SVar

    {

    -- Read only state

    -- Shared output queue (events, length)

    -- XXX For better efficiency we can try a preallocated array type (perhaps

    -- something like a vector) that allows an O(1) append. That way we will

    -- avoid constructing and reversing the list. Possibly we can also avoid

    -- the GC copying overhead. When the size increases we should be able to

    -- allocate the array in chunks.

    --

    -- [LOCKING] Frequent locked access. This is updated by workers on each

    -- yield and once in a while read by the consumer thread. This could have

    -- big locking overhead if the number of workers is high.

    --

    -- XXX We can use a per-CPU data structure to reduce the locking overhead.

    -- However, a per-cpu structure cannot guarantee the exact sequence in

    -- which the elements were added, though that may not be important.

    -- [LOCKING] Infrequent MVar. Used when the outputQ transitions from empty

    -- to non-empty, or a work item is queued by a worker to the work queue and

    -- needDoorBell is set by the consumer.

    -- channel to send events from the consumer to the worker. Used to send

    -- exceptions from a fold driver to the fold computation running as a

    -- consumer thread in the concurrent fold cases. Currently only one event

    -- is sent by the fold so we do not really need a queue for it.

    -- Combined/aggregate parameters

    -- This is truncated to maxBufferLimit if set to more than that. Otherwise

    -- potentially each worker may yield one value to the buffer in the worst

    -- case exceeding the requested buffer size.

    -- These two are valid and used only when maxBufferLimit is Limited.

    -- [LOCKING] The consumer puts this MVar after emptying the buffer, workers

    -- block on it when the buffer becomes full. No overhead unless the buffer

    -- becomes full.

    -- [LOCKING] Read only access by consumer when dispatching a worker.

    -- Decremented by workers when picking work and undo decrement if the

    -- worker does not yield a value.

    -- Used only by bounded SVar types

    -- Shared, thread tracking

    -- [LOCKING] Updated unlocked only by consumer thread in case of

    -- Async/Ahead style SVars. Updated locked by worker threads in case of

    -- Parallel style.

    -- [LOCKING] Updated locked by consumer thread when dispatching a worker

    -- and by the worker threads when the thread stops. This is read unsafely

    -- at several places where we want to rely on an approximate value.

    -- to track garbage collection of SVar

    -- Only for diagnostics

                              , Maybe Int)

    -- Shared work queue (stream, seqNo)

    }

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

-- State for concurrency control

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

-- XXX we can put the resettable fields in a oneShotConfig field and others in

-- a persistentConfig field. That way reset would be fast and scalable

-- irrespective of the number of fields.

--

-- XXX make all these Limited types and use phantom types to distinguish them

data State t m a = State

    { -- one shot configuration, automatically reset for each API call

    -- persistent configuration, state that remains valid until changed by

    -- an explicit setting via a combinator.

    -- XXX these two can be collapsed into a single type

    }

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

-- State defaults and reset

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

-- A magical value for the buffer size arrived at by running the smallest

-- possible task and measuring the optimal value of the buffer for that.  This

-- is obviously dependent on hardware, this figure is based on a 2.2GHz intel

-- core-i7 processor.

magicMaxBuffer :: Word

defaultMaxThreads, defaultMaxBuffer :: Limit

-- The fields prefixed by an _ are not to be accessed or updated directly but

-- via smart accessor APIs.

defState :: State t m a

    }

-- XXX if perf gets affected we can have all the Nothing params in a single

-- structure so that we reset is fast. We can also use rewrite rules such that

-- reset occurs only in concurrent streams to reduce the impact on serial

-- streams.

-- We can optimize this so that we clear it only if it is a Just value, it

-- results in slightly better perf for zip/zipM but the performance of scan

-- worsens a lot, it does not fuse.

--

-- XXX This has a side effect of clearing the SVar and yieldLimit, therefore it

-- should not be used to convert from the same type to the same type, unless

-- you want to clear the SVar. For clearing the SVar you should be using the

-- appropriate unStream functions instead.

--

-- | Adapt the stream state from one type to another.

adaptState :: State t m a -> State t n b

    }

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

-- Smart get/set routines for State

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

-- Use get/set routines instead of directly accessing the State fields

setYieldLimit :: Maybe Int64 -> State t m a -> State t m a

                Just n  ->

       }

getYieldLimit :: State t m a -> Maybe Count

setMaxThreads :: Int -> State t m a -> State t m a

       }

getMaxThreads :: State t m a -> Limit

setMaxBuffer :: Int -> State t m a -> State t m a

       }

getMaxBuffer :: State t m a -> Limit

setStreamRate :: Maybe Rate -> State t m a -> State t m a

getStreamRate :: State t m a -> Maybe Rate

setStreamLatency :: Int -> State t m a -> State t m a

       }

getStreamLatency :: State t m a -> Maybe NanoSecond64

setInspectMode :: State t m a -> State t m a

getInspectMode :: State t m a -> Bool

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

-- Cleanup

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

cleanupSVar :: SVar t m a -> IO ()

cleanupSVarFromWorker :: SVar t m a -> IO ()

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

-- Worker latency data collection

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

-- Every once in a while workers update the latencies and check the yield rate.

-- They return if we are above the expected yield rate. If we check too often

-- it may impact performance, if we check less often we may have a stale

-- picture. We update every minThreadDelay but we translate that into a yield

-- count based on latency so that the checking overhead is little.

--

-- XXX use a generation count to indicate that the value is updated. If the

-- value is updated an existing worker must check it again on the next yield.

-- Otherwise it is possible that we may keep updating it and because of the mod

-- worker keeps skipping it.

updateWorkerPollingInterval :: YieldRateInfo -> NanoSecond64 -> IO ()

{-# INLINE recordMinMaxLatency #-}

recordMinMaxLatency :: SVar t m a -> NanoSecond64 -> IO ()

recordAvgLatency :: SVar t m a -> (Count, NanoSecond64) -> IO ()

-- Pour the pending latency stats into a collection bucket

{-# INLINE collectWorkerPendingLatency #-}

collectWorkerPendingLatency

    :: IORef (Count, Count, NanoSecond64)

    -> IORef (Count, Count, NanoSecond64)

    -> IO (Count, Maybe (Count, NanoSecond64))

{-# INLINE shouldUseCollectedBatch #-}

shouldUseCollectedBatch

    :: Count

    -> NanoSecond64

    -> NanoSecond64

    -> NanoSecond64

    -> Bool

-- Returns a triple, (1) yield count since last collection, (2) the base time

-- when we started counting, (3) average latency in the last measurement

-- period. The former two are used for accurate measurement of the going rate

-- whereas the average is used for future estimates e.g. how many workers

-- should be maintained to maintain the rate.

-- CAUTION! keep it in sync with getWorkerLatency

collectLatency :: SVar t m a

               -> YieldRateInfo

               -> Bool

               -> IO (Count, AbsTime, NanoSecond64)

            -- When we have collected a significant sized batch we compute the

            -- new latency using that batch and return the new latency,

            -- otherwise we return the previous latency derived from the

            -- previous batch.

                -- XXX make this NOINLINE?

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

-- Dumping the SVar for debug/diag

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

dumpSVarStats :: SVar t m a -> SVarStats -> SVarStyle -> IO String

        ]

{-# NOINLINE dumpSVar #-}

dumpSVar :: SVar t m a -> IO String

    aheadDump <-

                ]

    waiting <-

        -- XXX print the types of events in the outputQueue, first 5

        ]

        -- XXX print the status of first 5 threads

        ]

printSVar :: SVar t m a -> String -> IO ()

-- MVar diagnostics has some overhead - around 5% on asyncly null benchmark, we

-- can keep it on in production to debug problems quickly if and when they

-- happen, but it may result in unexpected output when threads are left hanging

-- until they are GCed because the consumer went away.

{-# NOINLINE mvarExcHandler #-}

mvarExcHandler :: SVar t m a -> String -> BlockedIndefinitelyOnMVar -> IO ()

{-# NOINLINE stmExcHandler #-}

stmExcHandler :: SVar t m a -> String -> BlockedIndefinitelyOnSTM -> IO ()

withDiagMVar :: SVar t m a -> String -> IO () -> IO ()

    then

                         ]

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

-- Spawning threads

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

-- | A monad that can perform concurrent or parallel IO operations. Streams

-- that can be composed concurrently require the underlying monad to be

-- 'MonadAsync'.

--

-- @since 0.1.0

type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)

-- When we run computations concurrently, we completely isolate the state of

-- the concurrent computations from the parent computation.  The invariant is

-- that we should never be running two concurrent computations in the same

-- thread without using the runInIO function.  Also, we should never be running

-- a concurrent computation in the parent thread, otherwise it may affect the

-- state of the parent which is against the defined semantics of concurrent

-- execution.

captureMonadState :: MonadBaseControl IO m => m (RunInIO m)

-- Stolen from the async package. The perf improvement is modest, 2% on a

-- thread heavy benchmark (parallel composition using noop computations).

-- A version of forkIO that does not include the outer exception

-- handler: saves a bit of time when we will be installing our own

-- exception handler.

{-# INLINE rawForkIO #-}

rawForkIO :: IO () -> IO ThreadId

{-# INLINE doFork #-}

doFork :: MonadBaseControl IO m

    => m ()

    -> RunInIO m

    -> (SomeException -> IO ())

    -> m ThreadId

{-# INLINABLE fork #-}

fork :: MonadBaseControl IO m => m () -> m ThreadId

-- | Fork a thread that is automatically killed as soon as the reference to the

-- returned threadId is garbage collected.

--

{-# INLINABLE forkManaged #-}

forkManaged :: (MonadIO m, MonadBaseControl IO m) => m () -> m ThreadId

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

-- Collecting results from child workers in a streamed fashion

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

-- XXX Can we make access to remainingWork and yieldRateInfo fields in sv

-- faster, along with the fields in sv required by send?

-- XXX make it noinline

--

-- XXX we may want to employ an increment and decrement in batches when the

-- througput is high or when the cost of synchronization is high. For example

-- if the application is distributed then inc/dec of a shared variable may be

-- very costly.

--

-- A worker decrements the yield limit before it executes an action. However,

-- the action may not result in an element being yielded, in that case we have

-- to increment the yield limit.

--

-- Note that we need it to be an Int type so that we have the ability to undo a

-- decrement that takes it below zero.

{-# INLINE decrementYieldLimit #-}

decrementYieldLimit :: SVar t m a -> IO Bool

{-# INLINE incrementYieldLimit #-}

incrementYieldLimit :: SVar t m a -> IO ()

-- XXX exception safety of all atomic/MVar operations

-- TBD Each worker can have their own queue and the consumer can empty one

-- queue at a time, that way contention can be reduced.

-- XXX Only yields should be counted in the buffer limit and not the Stop

-- events.

{-# INLINE decrementBufferLimit #-}