-- |
  Module       : Streamly . Internal . Data . SVar . Pull
  Copyright    : ( c ) 2017 Composewell Technologies
-- License     : BSD-3-Clause
-- Maintainer  : 
-- Stability   : experimental
  Portability : GHC
--
--
module Streamly.Internal.Data.SVar.Pull
    (
    -- * Read Output
      readOutputQBasic
    , readOutputQRaw
    , readOutputQPaced
    , readOutputQBounded

  *   After Reading
    , postProcessPaced
    , postProcessBounded

    -- * Release Resources
    , cleanupSVar
    , cleanupSVarFromWorker
    )
where

#include "inline.hs"

import Control.Concurrent (myThreadId, throwTo)
import Control.Monad (when, void)
import Control.Monad.IO.Class (MonadIO(liftIO))
import Data.IORef (readIORef, writeIORef)
import Data.IORef (IORef)
import Streamly.Internal.Control.Concurrent (MonadAsync)
import Streamly.Internal.Data.Atomics (atomicModifyIORefCAS)

import qualified Data.Set as S

import Streamly.Internal.Data.SVar.Type
import Streamly.Internal.Data.SVar.Dispatch

-------------------------------------------------------------------------------
-- Reading from the workers' output queue/buffer
-------------------------------------------------------------------------------

# INLINE readOutputQBasic #
readOutputQBasic :: IORef ([ChildEvent a], Int) -> IO ([ChildEvent a], Int)
readOutputQBasic q = atomicModifyIORefCAS q $ \x -> (([],0), x)

# INLINE readOutputQRaw #
readOutputQRaw :: SVar t m a -> IO ([ChildEvent a], Int)
readOutputQRaw sv = do
    (list, len) <- readOutputQBasic (outputQueue sv)
    when (svarInspectMode sv) $ do
        let ref = maxOutQSize $ svarStats sv
        oqLen <- readIORef ref
        when (len > oqLen) $ writeIORef ref len
    return (list, len)

readOutputQBounded :: MonadAsync m => SVar t m a -> m [ChildEvent a]
readOutputQBounded sv = do
    (list, len) <- liftIO $ readOutputQRaw sv
    -- When there is no output seen we dispatch more workers to help
    -- out if there is work pending in the work queue.
    if len <= 0
    then blockingRead
    else do
        -- send a worker proactively, if needed, even before we start
        -- processing the output.  This may degrade single processor
        -- perf but improves multi-processor, because of more
        -- parallelism
        sendOneWorker
        return list

    where

    sendOneWorker = do
        cnt <- liftIO $ readIORef $ workerCount sv
        when (cnt <= 0) $ do
            done <- liftIO $ isWorkDone sv
            when (not done) (pushWorker 0 sv)

# INLINE blockingRead #
    blockingRead = do
        sendWorkerWait sendWorkerDelay (dispatchWorker 0) sv
        liftIO (fst `fmap` readOutputQRaw sv)

readOutputQPaced :: MonadAsync m => SVar t m a -> m [ChildEvent a]
readOutputQPaced sv = do
    (list, len) <- liftIO $ readOutputQRaw sv
    if len <= 0
    then blockingRead
    else do
        -- XXX send a worker proactively, if needed, even before we start
        -- processing the output.
        void $ dispatchWorkerPaced sv
        return list

    where

# INLINE blockingRead #
    blockingRead = do
        sendWorkerWait sendWorkerDelayPaced dispatchWorkerPaced sv
        liftIO (fst `fmap` readOutputQRaw sv)

postProcessPaced :: MonadAsync m => SVar t m a -> m Bool
postProcessPaced sv = do
    workersDone <- allThreadsDone sv
    -- XXX If during consumption we figure out we are getting delayed then we
    -- should trigger dispatch there as well.  We should try to check on the
    -- workers after consuming every n item from the buffer?
    if workersDone
    then do
        r <- liftIO $ isWorkDone sv
        when (not r) $ do
            void $ dispatchWorkerPaced sv
            -- Note that we need to guarantee a worker since the work is not
            -- finished, therefore we cannot just rely on dispatchWorkerPaced
            -- which may or may not send a worker.
            noWorker <- allThreadsDone sv
            when noWorker $ pushWorker 0 sv
        return r
    else return False

postProcessBounded :: MonadAsync m => SVar t m a -> m Bool
postProcessBounded sv = do
    workersDone <- allThreadsDone sv
    -- There may still be work pending even if there are no workers pending
    -- because all the workers may return if the outputQueue becomes full. In
    -- that case send off a worker to kickstart the work again.
    --
    -- Note that isWorkDone can only be safely checked if all workers are done.
    -- When some workers are in progress they may have decremented the yield
    -- Limit and later ending up incrementing it again. If we look at the yield
    -- limit in that window we may falsely say that it is 0 and therefore we
    -- are done.
    if workersDone
    then do
        r <- liftIO $ isWorkDone sv
        -- Note that we need to guarantee a worker, therefore we cannot just
  use dispatchWorker which may or may not send a worker .
        when (not r) (pushWorker 0 sv)
        -- XXX do we need to dispatch many here?
  void $ dispatchWorker sv
        return r
    else return False

-------------------------------------------------------------------------------
-- Cleanup
-------------------------------------------------------------------------------

cleanupSVar :: SVar t m a -> IO ()
cleanupSVar sv = do
    workers <- readIORef (workerThreads sv)
    Prelude.mapM_ (`throwTo` ThreadAbort)
          workers

cleanupSVarFromWorker :: SVar t m a -> IO ()
cleanupSVarFromWorker sv = do
    workers <- readIORef (workerThreads sv)
    self <- myThreadId
    Prelude.mapM_ (`throwTo` ThreadAbort)
          (Prelude.filter (/= self) $ S.toList workers)

 |
 License     : BSD-3-Clause
 Maintainer  : 
 Stability   : experimental


 * Read Output
 * Release Resources
-----------------------------------------------------------------------------
 Reading from the workers' output queue/buffer
-----------------------------------------------------------------------------
 When there is no output seen we dispatch more workers to help
 out if there is work pending in the work queue.
 send a worker proactively, if needed, even before we start
 processing the output.  This may degrade single processor
 perf but improves multi-processor, because of more
 parallelism
 XXX send a worker proactively, if needed, even before we start
 processing the output.
 XXX If during consumption we figure out we are getting delayed then we
 should trigger dispatch there as well.  We should try to check on the
 workers after consuming every n item from the buffer?
 Note that we need to guarantee a worker since the work is not
 finished, therefore we cannot just rely on dispatchWorkerPaced
 which may or may not send a worker.
 There may still be work pending even if there are no workers pending
 because all the workers may return if the outputQueue becomes full. In
 that case send off a worker to kickstart the work again.

 Note that isWorkDone can only be safely checked if all workers are done.
 When some workers are in progress they may have decremented the yield
 Limit and later ending up incrementing it again. If we look at the yield
 limit in that window we may falsely say that it is 0 and therefore we
 are done.
 Note that we need to guarantee a worker, therefore we cannot just
 XXX do we need to dispatch many here?
-----------------------------------------------------------------------------
 Cleanup
-----------------------------------------------------------------------------
  Module       : Streamly . Internal . Data . SVar . Pull
  Copyright    : ( c ) 2017 Composewell Technologies
  Portability : GHC
module Streamly.Internal.Data.SVar.Pull
    (
      readOutputQBasic
    , readOutputQRaw
    , readOutputQPaced
    , readOutputQBounded

  *   After Reading
    , postProcessPaced
    , postProcessBounded

    , cleanupSVar
    , cleanupSVarFromWorker
    )
where

#include "inline.hs"

import Control.Concurrent (myThreadId, throwTo)
import Control.Monad (when, void)
import Control.Monad.IO.Class (MonadIO(liftIO))
import Data.IORef (readIORef, writeIORef)
import Data.IORef (IORef)
import Streamly.Internal.Control.Concurrent (MonadAsync)
import Streamly.Internal.Data.Atomics (atomicModifyIORefCAS)

import qualified Data.Set as S

import Streamly.Internal.Data.SVar.Type
import Streamly.Internal.Data.SVar.Dispatch


# INLINE readOutputQBasic #
readOutputQBasic :: IORef ([ChildEvent a], Int) -> IO ([ChildEvent a], Int)
readOutputQBasic q = atomicModifyIORefCAS q $ \x -> (([],0), x)

# INLINE readOutputQRaw #
readOutputQRaw :: SVar t m a -> IO ([ChildEvent a], Int)
readOutputQRaw sv = do
    (list, len) <- readOutputQBasic (outputQueue sv)
    when (svarInspectMode sv) $ do
        let ref = maxOutQSize $ svarStats sv
        oqLen <- readIORef ref
        when (len > oqLen) $ writeIORef ref len
    return (list, len)

readOutputQBounded :: MonadAsync m => SVar t m a -> m [ChildEvent a]
readOutputQBounded sv = do
    (list, len) <- liftIO $ readOutputQRaw sv
    if len <= 0
    then blockingRead
    else do
        sendOneWorker
        return list

    where

    sendOneWorker = do
        cnt <- liftIO $ readIORef $ workerCount sv
        when (cnt <= 0) $ do
            done <- liftIO $ isWorkDone sv
            when (not done) (pushWorker 0 sv)

# INLINE blockingRead #
    blockingRead = do
        sendWorkerWait sendWorkerDelay (dispatchWorker 0) sv
        liftIO (fst `fmap` readOutputQRaw sv)

readOutputQPaced :: MonadAsync m => SVar t m a -> m [ChildEvent a]
readOutputQPaced sv = do
    (list, len) <- liftIO $ readOutputQRaw sv
    if len <= 0
    then blockingRead
    else do
        void $ dispatchWorkerPaced sv
        return list

    where

# INLINE blockingRead #
    blockingRead = do
        sendWorkerWait sendWorkerDelayPaced dispatchWorkerPaced sv
        liftIO (fst `fmap` readOutputQRaw sv)

postProcessPaced :: MonadAsync m => SVar t m a -> m Bool
postProcessPaced sv = do
    workersDone <- allThreadsDone sv
    if workersDone
    then do
        r <- liftIO $ isWorkDone sv
        when (not r) $ do
            void $ dispatchWorkerPaced sv
            noWorker <- allThreadsDone sv
            when noWorker $ pushWorker 0 sv
        return r
    else return False

postProcessBounded :: MonadAsync m => SVar t m a -> m Bool
postProcessBounded sv = do
    workersDone <- allThreadsDone sv
    if workersDone
    then do
        r <- liftIO $ isWorkDone sv
  use dispatchWorker which may or may not send a worker .
        when (not r) (pushWorker 0 sv)
  void $ dispatchWorker sv
        return r
    else return False


cleanupSVar :: SVar t m a -> IO ()
cleanupSVar sv = do
    workers <- readIORef (workerThreads sv)
    Prelude.mapM_ (`throwTo` ThreadAbort)
          workers

cleanupSVarFromWorker :: SVar t m a -> IO ()
cleanupSVarFromWorker sv = do
    workers <- readIORef (workerThreads sv)
    self <- myThreadId
    Prelude.mapM_ (`throwTo` ThreadAbort)
          (Prelude.filter (/= self) $ S.toList workers)
