content: {-# LANGUAGE ScopedTypeVariables, BangPatterns, ConstraintKinds, TypeFamilies #-}
  {-# LANGUAGE DataKinds #-}
  
  -- | EXPERIMENTAL version which eventually should be made generic across Par monads
  -- (i.e. a BulkRetryT transformer), and should thus be extended to transparently
  -- catch any attempts by a thread to block, not just the special non-blocking calls
  -- provided by *this* library.
  
  module Control.LVish.BulkRetry
  
         where
  
  import qualified Data.Bits.Atomic as B
  import Foreign.Storable (sizeOf, Storable)
  import Control.Monad (unless) 
  import Control.LVish
  import Control.LVish.Internal (unsafeDet)
  import Control.Par.Class (LVarSched(returnToSched))
  -- import Data.LVar.NatArray
  import Data.LVar.NatArray.Unsafe (NatArray, unsafePeek)
  
  import Data.Par.Splittable (pforEach)
  import Data.Par.Range (range)
  import Data.Par.Set () -- Instances only.
  
  import qualified Data.Foldable as F
  import qualified Data.Set as S
  -- import           Data.LVar.PureSet as IS
  import           Data.LVar.SLSet as IS
  import           Data.LVar.Generic (freeze)
  
  -- import Data.Par.Range
  
  --------------------------------------------------------------------------------
  
  -- | The point where users send abort messages.
  data RetryHub s = RetryHub (ISet s Int) -- ^ This stores the iterations that fail.
                             Int -- ^ This is the current iteration
  
  -- -- | Non-blocking get on a `NatArray`.
  -- getNB :: forall s d elt . (Storable elt, B.AtomicBits elt, Num elt) =>
  --          RetryHub s -> NatArray s elt -> Int -> Par e s elt
  -- -- LVarSched (Par e s)         
  -- getNB (RetryHub fails) arr ind = do
  --   x <- unsafePeek arr ind
  --   -- if empty, don't block, do this:
  --   case x of
  --     Nothing  -> do logDbgLn 4 $ " [dbg-lvish] getNB: iteration failed, enqueue for retry: "++show ind
  --                    insert ind fails
  --                    returnToSched
  --     Just res -> return res
  
  
  -- | Non-blocking get on a `NatArray`.  In this prototype we require that the user
  -- manually CPS the computation, so that the delimited continuation between this get
  -- and the end of the loop iteration is passed explicitly as an argument.
  --
  -- The current reason for this compromise is that the HandlerPool mechanism is not
  -- robust to us dropping the current continuation with `returnToSched`.  We would
  -- need a version of HandlerPool's that interoperates with a user-level callCC, that is
  -- we would need something like bracket/dynamic-wind for our continuation monad.
  getNB_cps :: forall s e elt . (Storable elt, B.AtomicBits elt, Num elt) =>
           RetryHub s
           -> NatArray s elt      -- ^ Array to dereference
           -> Int                 -- ^ Which index to get
           -> (elt -> Par e s ()) -- ^ Delimited continuation.
           -> Par e s ()
  -- LVarSched (Par e s)         
  getNB_cps (RetryHub fails thisiter) arr ind cont = do
    x <- unsafePeek arr ind
    -- if empty, don't block, do this:
    case x of
      Nothing  -> do logDbgLn 4 $ " [dbg-lvish] getNB: iteration "++ show thisiter
                                  ++" failed, due to get on index "++show ind
                     insert thisiter fails
                     return ()
      Just res -> do logDbgLn 4 $ " [dbg-lvish] getNB: result available, calling continuation (iter "++show thisiter++")"
                     cont res
  {-# INLINE getNB_cps #-}
  
  desired_tasks :: Int
  desired_tasks = 16 -- FIXME: num procs * overpartition
  
  -- | A parallel for-loop which aborts and retries failed iterations in bulk, rather
  -- than allowing them to "block" and suffering the overhead of capturing and storing
  -- their continuations.
  -- 
  -- `forSpeculative` continues retrying until ALL iterations have completed.  It is
  -- thus a *synchronous* parallel for loop.
  forSpeculative :: Idempotent e => (Int, Int)  -- ^ Inclusive/Exclusive range to run.
                    -> (RetryHub s -> Int -> Par e s ()) -- ^ Body of the loop
                    -> Par e s ()
  -- forSpeculative :: (Int, Int) -> (RetryHub s -> Int -> Par QuasiDet s ()) -> Par QuasiDet s ()
  -- TODO: Requires idempotency!!
  forSpeculative (st,end) bodyfn = do
    logDbgLn 2 $ " [dbg-lvish] Begin forSpeculative, bounds "++show (st,end)
    let sz = end - st
        -- Even in a trivial loop, 2000 iters per task should be enough:
        prefix = min sz (2000 * desired_tasks)
        -- TODO: automatic strategies for tuning the input prefix size would be helpful.
        -- One approach that might make sense would be to auto-tune based on the
        -- time/iteration observed.  That is, gradually increase to try to approximate a
        -- minimum reasonable task size and no bigger.  
  
        body' = bodyfn
        -- body' retry ix = bodyfn retry ix
    
    let flush leftover fails = 
          -- unless (S.null leftover) $ do
            -- TODO: need parallel fold, this is sequential...
            F.foldlM (\ () ix -> do
                         logDbgLn 3 $ " [dbg-lvish] forSpeculative: flushing iter "++show ix
                         body' (RetryHub fails ix) ix)
                     () leftover
    let flushLoop leftover =  do
          fails <- newEmptySet
          -- FIXME: Add parallelism
          flush leftover fails -- Sequential...        
          snap <- unsafeDet $ freeze fails
          logDbgLn 3 $ " [dbg-lvish] forSpeculative: did one sequential flush, remaining: "++show snap
          unless (S.null snap) $
            -- error$ "forSpeculative: failures not flushed with a sequential run!:\n "++show snap
            flushLoop snap
        
    -- Outer loop of "rounds", in which we try a prefix of the iteration space.  
    let loop !round leftover offset 0 = do
          logDbgLn 3 $ " [dbg-lvish] forSpeculative: got to the end, only failures left."
          flushLoop leftover
          
        loop !round leftover offset remain = do
          logDbgLn 3 $ " [dbg-lvish] forSpeculative starting round "++
                       show round++": offset "++show offset++", remaining "++show remain
          -- Set of iterations that failed in THIS upcoming round:
          fails <- newEmptySet
          let chunkend = offset + (min prefix remain)        
  
          hp <- newPool
          -- Here we keep the failed iterations "to the left" of the new batch, i.e. we
          -- fork them first.
          
          -- FINISHME: need Split instance.
          logDbgLn 4 $ " [dbg-lvish] forSpeculative RElaunching failures: "++show leftover
          -- This version is poor because it forks on a per-iteration basis upon retry:
          -- F.foldrM (\ ix () -> forkHP (Just hp) (body' (RetryHub fails ix) ix)) () leftover
          -- F.foldrM (\ ix () -> body' (RetryHub fails ix) ix) () leftover
          -- pforEach leftover $ bodyfn (RetryHub fails)
          asyncForEachHP (Just hp) leftover $ \ ix -> bodyfn (RetryHub fails ix) ix
          
          -- TODO: if we keep failing it's better to expand the prefix.  That way we
          -- end up with a logarithmic number of retries for each iterate in the worst
          -- case, rather than linear (making the whole loop unnecessarily quadratic).
  
          logDbgLn 4 $ " [dbg-lvish] forSpeculative launching new batch: "++show (offset,chunkend)
          asyncForEachHP (Just hp) (range offset chunkend) $ \ ix -> 
            body' (RetryHub fails ix) ix
          logDbgLn 4 $ " [dbg-lvish] forSpeculative: return from par for-loop; now quiesce."
          quiesce hp
          logDbgLn 4 $ " [dbg-lvish] forSpeculative: quiesce finished, next freeze failed set."
          snap <- unsafeDet $ freeze fails
          logDbgLn 4 $ " [dbg-lvish] forSpeculative finish round; failed iterates: "++show snap
          loop (round+1) snap chunkend (remain - (chunkend - offset))
    loop 0 S.empty 0 sz       
    -- After the last quiesce, we're done.
  