content: -- |
  -- Module      : Streamly.Internal.Data.Channel.Dispatcher
  -- Copyright   : (c) 2017 Composewell Technologies
  -- License     : BSD-3-Clause
  -- Maintainer  : streamly@composewell.com
  -- Stability   : experimental
  -- Portability : GHC
  --
  --
  module Streamly.Internal.Data.Channel.Dispatcher
      (
      -- ** Latency collection
        minThreadDelay
      , collectLatency
  
      -- ** Thread accounting
      , addThread
      , delThread
      , modifyThread
      , allThreadsDone
      , recordMaxWorkers
  
      -- ** Diagnostics
      , dumpSVarStats
      )
  where
  
  import Data.Set (Set)
  import Control.Concurrent (MVar, ThreadId)
  import Control.Concurrent.MVar (tryPutMVar)
  import Control.Exception (assert)
  import Control.Monad (when, void)
  import Control.Monad.IO.Class (MonadIO(liftIO))
  import Data.IORef (IORef, modifyIORef, readIORef, writeIORef)
  import Streamly.Internal.Data.Atomics
      (atomicModifyIORefCAS, writeBarrier, atomicModifyIORefCAS_)
  import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
  import Streamly.Internal.Data.Time.Units
         ( AbsTime, NanoSecond64(..), diffAbsTime64, showNanoSecond64
         , showRelTime64)
  
  import qualified Data.Set as S
  
  import Streamly.Internal.Data.Channel.Types
  
  -------------------------------------------------------------------------------
  -- Worker latency data processing
  -------------------------------------------------------------------------------
  
  -- | This is a magic number and it is overloaded, and used at several places to
  -- achieve batching:
  --
  -- 1. If we have to sleep to slowdown this is the minimum period that we
  --    accumulate before we sleep. Also, workers do not stop until this much
  --    sleep time is accumulated.
  -- 3. Collected latencies are computed and transferred to measured latency
  --    after a minimum of this period.
  minThreadDelay :: NanoSecond64
  minThreadDelay = 1000000
  
  -- | 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 ()
  updateWorkerPollingInterval yinfo latency = do
      let periodRef = workerPollingInterval yinfo
          -- This depends on the rate, if the rate is low, latencies are
          -- small, by the time we poll it might be too late and we may have
          -- yielded too many results.
          -- cnt = max 1 $ minThreadDelay `div` latency
          cnt = max 1 (latency `div` svarLatencyTarget yinfo)
          period = min cnt (fromIntegral magicMaxBuffer)
  
      writeIORef periodRef (fromIntegral period)
  
  {-# INLINE recordMinMaxLatency #-}
  recordMinMaxLatency :: SVarStats -> NanoSecond64 -> IO ()
  recordMinMaxLatency ss new = do
      minLat <- readIORef (minWorkerLatency ss)
      when (new < minLat || minLat == 0) $
          writeIORef (minWorkerLatency ss) new
  
      maxLat <- readIORef (maxWorkerLatency ss)
      when (new > maxLat) $ writeIORef (maxWorkerLatency ss) new
  
  recordAvgLatency :: SVarStats -> (Count, NanoSecond64) -> IO ()
  recordAvgLatency ss (count, time) = do
      modifyIORef (avgWorkerLatency ss) $
          \(cnt, t) -> (cnt + count, t + time)
  
  -- | Add the 'workerPendingLatency' to 'workerCollectedLatency' and reset it to
  -- zeroes. Return the added counts.
  {-# INLINE collectWorkerPendingLatency #-}
  collectWorkerPendingLatency
      :: IORef (Count, Count, NanoSecond64) -- ^ 'workerPendingLatency'
      -> IORef (Count, Count, NanoSecond64) -- ^ 'workerCollectedLatency'
      -> IO (Count, Maybe (Count, NanoSecond64))
      -- ^ (total yield count, Maybe (total latency count, total latency time)).
      -- Latency count and time are reported only when both are non-zero to avoid
      -- arithemetic exceptions in calculations.
  collectWorkerPendingLatency cur col = do
      (fcount, count, time) <- atomicModifyIORefCAS cur $ \v -> ((0,0,0), v)
  
      (fcnt, cnt, t) <- readIORef col
      let totalCount = fcnt + fcount
          latCount   = cnt + count
          latTime    = t + time
      writeIORef col (totalCount, latCount, latTime)
  
      assert (latCount == 0 || latTime /= 0) (return ())
      let latPair =
              if latCount > 0 && latTime > 0
              then Just (latCount, latTime)
              else Nothing
      return (totalCount, latPair)
  
  {-# INLINE shouldUseCollectedBatch #-}
  shouldUseCollectedBatch
      :: Count
      -> NanoSecond64
      -> NanoSecond64
      -> NanoSecond64
      -> Bool
  shouldUseCollectedBatch collectedYields collectedTime newLat prevLat =
      let r = fromIntegral newLat / fromIntegral prevLat :: Double
      in     (collectedYields > fromIntegral magicMaxBuffer)
          || (collectedTime > minThreadDelay)
          || (prevLat > 0 && (r > 2 || r < 0.5))
          || (prevLat == 0)
  
  -- CAUTION! keep it in sync with getWorkerLatency
  
  -- | Always moves 'workerPendingLatency' to 'workerCollectedLatency':
  --
  --  * 'workerCollectedLatency' always incremented by 'workerPendingLatency'
  --  * 'workerPendingLatency' always reset to 0
  --
  -- Moves 'workerCollectedLatency' to 'svarAllTimeLatency' periodically, when
  -- the collected batch size hits a limit, or time limit is over, or latency
  -- changes beyond a limit. Updates done when the batch is collected:
  --
  --  * 'svarAllTimeLatency' yield count updated
  --  * 'workerMeasuredLatency' set to (new+prev)/2
  --  * 'workerPollingInterval' set using max of new/prev worker latency
  --  * 'workerCollectedLatency' reset to 0
  --
  -- See also 'getWorkerLatency'.
  --
  collectLatency ::
         Bool -- ^ stat inspection mode
      -> SVarStats -- ^ Channel stats
      -> YieldRateInfo -- ^ Channel rate control info
      -> Bool -- ^ Force batch collection
      -> IO (Count, AbsTime, NanoSecond64)
      -- ^ (channel yield count since beginning, beginning timestamp, 'workerMeasuredLatency')
  collectLatency inspecting ss yinfo drain = do
      let cur      = workerPendingLatency yinfo
          col      = workerCollectedLatency yinfo
          longTerm = svarAllTimeLatency yinfo
          measured = workerMeasuredLatency yinfo
  
      (newCount, newLatPair) <- collectWorkerPendingLatency cur col
      (lcount, ltime) <- readIORef longTerm
      prevLat <- readIORef measured
  
      let newLcount = lcount + newCount
          retWith lat = return (newLcount, ltime, lat)
  
      case newLatPair of
          Nothing -> retWith prevLat
          Just (count, time) -> do
              let newLat = time `div` fromIntegral count
              when inspecting $ recordMinMaxLatency ss newLat
              -- 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.
              if shouldUseCollectedBatch newCount time newLat prevLat || drain
              then do
                  -- XXX make this NOINLINE?
                  updateWorkerPollingInterval yinfo (max newLat prevLat)
                  when inspecting $ recordAvgLatency ss (count, time)
                  writeIORef col (0, 0, 0)
                  writeIORef measured ((prevLat + newLat) `div` 2)
                  modifyIORef longTerm $ \(_, t) -> (newLcount, t)
                  retWith newLat
              else retWith prevLat
  
  -------------------------------------------------------------------------------
  -- Dumping the SVar for debug/diag
  -------------------------------------------------------------------------------
  
  dumpSVarStats :: Bool -> Maybe YieldRateInfo -> SVarStats -> IO String
  dumpSVarStats inspecting rateInfo ss = do
      case rateInfo of
          Nothing -> return ()
          Just yinfo -> do
              _ <- liftIO $ collectLatency inspecting ss yinfo True
              return ()
  
      dispatches <- readIORef $ totalDispatches ss
      maxWrk <- readIORef $ maxWorkers ss
      maxOq <- readIORef $ maxOutQSize ss
      -- maxHp <- readIORef $ maxHeapSize ss
      minLat <- readIORef $ minWorkerLatency ss
      maxLat <- readIORef $ maxWorkerLatency ss
      (avgCnt, avgTime) <- readIORef $ avgWorkerLatency ss
      stopTime <- readIORef (svarStopTime ss)
      let stopReason =
              case stopTime of
                  Nothing -> "stream abandoned"
                  Just _ -> "stream finished"
      (svarCnt, svarGainLossCnt, svarLat, interval) <- case rateInfo of
          Nothing -> return (0, 0, 0, 0)
          Just yinfo -> do
              (cnt, startTime) <- readIORef $ svarAllTimeLatency yinfo
              interval <-
                  case stopTime of
                      Nothing -> do
                          now <- getTime Monotonic
                          return (diffAbsTime64 now startTime)
                      Just t -> do
                          return (diffAbsTime64 t startTime)
              if cnt > 0
              then do
                  gl <- readIORef (svarGainedLostYields yinfo)
                  return (cnt, gl, interval `div` fromIntegral cnt, interval)
              else return (0, 0, 0, interval)
  
      return $ concat
          [ "stop reason = " <> stopReason
          ,  if interval > 0
             then "\nlife time = " <> showRelTime64 interval
             else ""
          , "\ntotal dispatches = " <> show dispatches
          , "\nmax workers = " <> show maxWrk
          , "\nmax outQSize = " <> show maxOq
          , if minLat > 0
            then "\nmin worker latency = " <> showNanoSecond64 minLat
            else ""
          , if maxLat > 0
            then "\nmax worker latency = " <> showNanoSecond64 maxLat
            else ""
          , if avgCnt > 0
            then let lat = avgTime `div` fromIntegral avgCnt
                  in "\navg worker latency = " <> showNanoSecond64 lat
            else ""
          , if svarLat > 0
            then "\nchannel latency = " <> showRelTime64 svarLat
            else ""
          , if svarCnt > 0
            then "\nchannel yield count = " <> show svarCnt
            else ""
          , if svarGainLossCnt > 0
            then "\nchannel gain/loss yield count = " <> show svarGainLossCnt
            else ""
          ]
  
  -------------------------------------------------------------------------------
  -- Thread accounting
  -------------------------------------------------------------------------------
  
  -- Thread tracking is needed for two reasons:
  --
  -- 1) Killing threads on exceptions. Threads may not be left to go away by
  -- themselves because they may run for significant times before going away or
  -- worse they may be stuck in IO and never go away.
  --
  -- 2) To know when all threads are done and the stream has ended.
  
  {-# NOINLINE addThread #-}
  addThread :: MonadIO m => IORef (Set ThreadId) -> ThreadId -> m ()
  addThread workerSet tid =
      liftIO $ atomicModifyIORefCAS_ workerSet (S.insert tid)
  
  -- This is cheaper than modifyThread because we do not have to send a
  -- outputDoorBell This can make a difference when more workers are being
  -- dispatched.
  {-# INLINE delThread #-}
  delThread :: MonadIO m => IORef (Set ThreadId) -> ThreadId -> m ()
  delThread workerSet tid =
      liftIO $ atomicModifyIORefCAS_ workerSet (S.delete tid)
  
  -- If present then delete else add. This takes care of out of order add and
  -- delete i.e. a delete arriving before we even added a thread.
  -- This occurs when the forked thread is done even before the 'addThread' right
  -- after the fork gets a chance to run.
  {-# INLINE modifyThread #-}
  modifyThread :: MonadIO m => IORef (Set ThreadId) -> MVar () -> ThreadId -> m ()
  modifyThread workerSet bell tid = do
      changed <- liftIO $ atomicModifyIORefCAS workerSet $ \old ->
          if S.member tid old
          then let new = S.delete tid old in (new, new)
          else let new = S.insert tid old in (new, old)
      when (null changed) $
           liftIO $ do
              writeBarrier
              void $ tryPutMVar bell ()
  
  -- | This is safe even if we are adding more threads concurrently because if
  -- a child thread is adding another thread then anyway 'workerThreads' will
  -- not be empty.
  {-# INLINE allThreadsDone #-}
  allThreadsDone :: MonadIO m => IORef (Set ThreadId) -> m Bool
  allThreadsDone ref = liftIO $ S.null <$> readIORef ref
  
  -------------------------------------------------------------------------------
  -- Dispatching workers
  -------------------------------------------------------------------------------
  
  {-# NOINLINE recordMaxWorkers #-}
  recordMaxWorkers :: MonadIO m => IORef Int -> SVarStats -> m ()
  recordMaxWorkers countRef ss = liftIO $ do
      active <- readIORef countRef
      maxWrk <- readIORef (maxWorkers ss)
      when (active > maxWrk) $ writeIORef (maxWorkers ss) active
      modifyIORef (totalDispatches ss) (+1)
  