content: {-# LANGUAGE Unsafe #-}
  {-# LANGUAGE RankNTypes #-} 
  {-# LANGUAGE CPP #-}
  {-# LANGUAGE NamedFieldPuns, BangPatterns #-}
  {-# LANGUAGE ExistentialQuantification #-}
  {-# LANGUAGE MultiParamTypeClasses #-}
  {-# LANGUAGE ScopedTypeVariables #-}
  {-# LANGUAGE DoAndIfThenElse #-}
  {-# LANGUAGE TypeFamilies #-}
  {-# LANGUAGE TypeOperators #-}
  {-# LANGUAGE UndecidableInstances #-}
  {-# LANGUAGE DeriveDataTypeable #-} 
  {-# LANGUAGE GeneralizedNewtypeDeriving #-}
  {-# LANGUAGE FlexibleInstances #-} -- For DeepFreeze
  
  {-# OPTIONS_GHC -fno-warn-name-shadowing -fno-warn-unused-do-bind #-}
  
  -- | This is an internal module that provides the core parallel scheduler.
  --   It is /not/ for end-users.
  
  module Control.LVish.Internal.SchedIdempotent
    (
      -- * Basic types and accessors
      LVar(..), HandlerPool(),
      Par(..), ClosedPar(..),
      
      -- * Safe, deterministic operations
      yield, newPool, fork, forkHP,
      runPar, runParIO,
      runParDetailed, runParLogged,
      withNewPool, withNewPool_,
      forkWithExceptions,
      
      -- * Quasi-deterministic operations
      quiesce, quiesceAll,
  
      -- * Re-exported debug facilities
      logStrLn, dbgLvl, getLogger,
         
      -- * Unsafe operations; should be used only by experts to build new abstractions
      newLV, getLV, putLV, putLV_, freezeLV, freezeLVAfter,
      addHandler, liftIO, toss,
  
      -- * Internal, private bits.
      mkPar, Status(..), sched, Listener(..), lvarDbgName
    ) where
  
  import qualified Control.Monad.IO.Class as IOClass
  import           Control.Monad hiding (sequence, join)
  import           Control.Concurrent hiding (yield)
  -- import qualified Control.Concurrent as Conc
  import qualified Control.Exception as E
  import qualified Control.Concurrent.Async as A
  import           Control.DeepSeq
  import           Control.Applicative
  -- import           Control.LVish.Internal.Logging as L
  import           Data.Concurrent.Internal.MonadToss
  -- import           Debug.Trace(trace)
  import           Data.IORef
  import           Data.Atomics
  -- import           Data.Typeable
  import qualified Data.Atomics.Counter as C
  import qualified Data.Concurrent.Bag as B
  import           GHC.Conc hiding (yield, withMVar)
  -- import qualified GHC.Conc 
  import           System.IO
  import           System.IO.Unsafe (unsafePerformIO)
  -- import           System.Environment(getEnvironment)
  import           System.Mem.StableName (makeStableName, hashStableName)
  import           Prelude  hiding (mapM, sequence, head, tail)
  import qualified Prelude
  --import           System.Random (random)
  import           System.Random.PCG.Fast.Pure (uniformBool)
  import           System.Log.TSLogger as L
  #ifdef DEBUG_LVAR
  import           Text.Printf (hPrintf)
  #endif
  -- import Control.Compose ((:.), unO)
  -- import           Data.Traversable  hiding (forM)
  
  import Control.LVish.Internal.Types
  import qualified Control.LVish.Internal.SchedUtils as Sched
  
  ------------------------------------------------------------------------------
  -- LVar and Par monad representation
  ------------------------------------------------------------------------------
  
  -- | LVars are parameterized by two types:
  -- 
  --     * The first, @a@, characterizes the \"state\" of the LVar (i.e., the lattice
  --     element), and should be a concurrently mutable data type.  That means, in
  --     particular, that only a /transient snapshot/ of the state can be
  --     obtained in general.  But the information in such a snapshot is always a
  --     lower bound on the current value of the LVar.
  -- 
  --     * The second, @d@, characterizes the \"delta\" associated with a @putLV@
  --     operation (i.e., the actual change, if any, to the LVar's state).
  --     In many cases such deltas allow far more efficient communication between
  --     @putLV@s and blocked @getLV@s or handlers.  It is crucial, however, that
  --     the behavior of a @get@ or handler does not depend on the /particular/
  --     choice of @putLV@ operations (and hence deltas) that moved the LVar over
  --     the threshold.  For simple data structures, the delta may just be the
  --     entire LVar state, but for, e.g., collection data structures, delta will
  --     generally represent a single insertion.
  data LVar a d = LVar {
    state  :: !a,                                          -- the current, "global" state of the LVar
    status :: {-# UNPACK #-} !(IORef (Status d)),          -- is the LVar active or frozen?  
    name   :: {-# UNPACK #-} !LVarID                       -- a unique identifier for this LVar
  #ifdef NONIDEM
    , handlerStatus :: {-# UNPACK #-} !(IORef HandlerStatus) -- are handlers being installed?
  #endif
  }
  
  #ifdef NONIDEM
  data HandlerStatus
   = Dormant                    -- no handlers currently being installed
   | Installing Int [ClosedPar] -- some number of handlers being installed, with
                                -- a list of blocked puts waiting on completion
  #endif
  
  type LVarID = IORef ()
  newLVID :: IO (IORef ())
  newLVID = newIORef ()
  
  -- | a global ID that is *not* the name of any LVar.  Makes it possible to
  -- represent Maybe (LVarID) with the type LVarID -- i.e., without any allocation.
  noName :: LVarID
  noName = unsafePerformIO $ newLVID
  {-# NOINLINE noName #-}
  
  -- | The frozen bit of an LVar is tied together with the bag of waiting listeners,
  -- which allows the entire bag to become garbage immediately after freezing.
  -- (Note, however, that outstanding @put@s that occurred just before freezing
  -- may still reference the bag, which is necessary to ensure that all listeners
  -- are informed of the @put@ prior to freezing.)
  data Status d 
    = Freezing                     -- ^ further changes to the state are forbidden
    | Frozen                       -- ^ further changes to the state are forbidden
    | Active (B.Bag (Listener d))  -- ^ bag of blocked threshold reads and handlers
  
  -- | A listener for an LVar is informed of each change to the LVar's state
  -- (represented as a delta) and the event of the LVar freezing.  The listener is
  -- given access to a bag token, allowing it to remove itself from the bag of
  -- listeners, after unblocking a threshold read, for example.  It is also given
  -- access to the scheduler queue for the CPU that generated the event, which it
  -- can use to add threads.
  data Listener d = Listener {
    onUpdate :: d -> B.Token (Listener d) -> SchedState -> IO (),
    onFreeze ::      B.Token (Listener d) -> SchedState -> IO ()
  }
  
  -- | A @HandlerPool@ contains a way to count outstanding parallel computations that
  -- are affiliated with the pool.  It detects the condition where all such threads
  -- have completed.
  data HandlerPool = HandlerPool {
    numHandlers      :: C.AtomicCounter,   -- How many handler callbacks are currently
                                     -- running?
    blockedOnQuiesce :: B.Bag ClosedPar
  }
  
  -- | A monadic type constructor for parallel computations producing an answer @a@.
  -- This is the internal, unsafe type.
  newtype Par a = Par {
    -- the computation is represented in CPS
    close :: (a -> ClosedPar) -> ClosedPar  
  }
  
  -- A "closed" Par computation is one that has been plugged into a continuation.
  -- It is represented in a "Church encoded" style, i.e., directly in terms of its
  -- interpretation into the IO monad.  Since the continuation has already been
  -- plugged into the computation, there is no answer type here.
  newtype ClosedPar = ClosedPar {
    exec :: SchedState -> IO ()
  }
  
  type SchedState = Sched.State ClosedPar LVarID
  
  instance Functor Par where
    fmap f m = Par $ \k -> close m (k . f)
  
  instance Monad Par where
    return a = Par $ \k -> k a
    m >>= c  = Par $ \k -> close m $ \a -> close (c a) k
  
  instance Applicative Par where
    (<*>) = ap
    pure  = return
  
  
  ------------------------------------------------------------------------------
  -- A few auxiliary functions
  ------------------------------------------------------------------------------  
  
  mkPar :: ((a -> ClosedPar) -> SchedState -> IO ()) -> Par a
  mkPar f = Par $ \k -> ClosedPar $ \q -> f k q
  
  whenJust :: Maybe a -> (a -> IO ()) -> IO ()
  whenJust Nothing  _ = return ()
  whenJust (Just a) f = f a
  
  isFrozen :: LVar a d -> IO Bool
  isFrozen (LVar {status}) = do
    curStatus <- readIORef status
    case curStatus of
      Active _ -> return False
      _        -> return True
  
  -- | Logging within the (internal) Par monad.
  logStrLn  :: Int -> String -> Par ()
  #ifdef DEBUG_LVAR
  -- logStrLn = liftIO . logStrLn_
  logStrLn lvl str = do
    l <- getLogger
    case l of
      Nothing  -> liftIO$ lockedPrint ("WARNING, compiled in debug mode but no logger: "++show(lvl,str))
      Just lgr -> 
       when (lvl >= L.minLvl lgr && lvl <= L.maxLvl lgr) $ do
         num <- getWorkerNum
         if lvl < 0
          then liftIO$ logHelper (Just lgr) num (L.OffTheRecord (-lvl) str)
          else liftIO$ logHelper (Just lgr) num (L.StrMsg lvl str)    
  #else
  logStrLn _ _  = return ()
  #endif
  
  #ifdef DEBUG_LVAR
  -- | This is inefficient, but we should only use it in exceptional
  -- circumstances.
  lockedPrint :: String -> IO () 
  lockedPrint m =
    withMVar printLock (\() -> hPutStrLn stderr m)
  
  {-# NOINLINE printLock #-}
  printLock :: MVar ()
  printLock = unsafePerformIO (newMVar ())
   
  logHelper :: Maybe Logger -> Int -> LogMsg -> IO ()
  logHelper lgr num msg = do
    let msg' = L.mapMsg (("wrkr"++show num++" ")++) msg
    case lgr of 
      Just lgr -> L.logOn lgr msg' -- logOn will do the lvl check
      Nothing  -> lockedPrint ("WARNING/nologger:"++show msg')
  #endif
  
  logWith      :: Sched.State a s -> Int -> String -> IO ()
  logOffRecord :: Sched.State a s -> Int -> String -> IO ()
  #ifdef DEBUG_LVAR
  -- Only when the debug level is 1 or higher is the logger even initialized:
  logWith      q lvl str = logHelper (Sched.logger q) (Sched.no q) (L.StrMsg lvl str)
  logOffRecord q lvl str = logHelper (Sched.logger q) (Sched.no q) (L.OffTheRecord lvl str)
  #else
  logWith _ _ _ = return ()
  logOffRecord  _ _ _  = return ()
  #endif
  
  ------------------------------------------------------------------------------
  -- LVar operations
  ------------------------------------------------------------------------------
  
  -- | Debugging only -- create some kind of printable identifier for
  -- the LVar (uses StableName).
  lvarDbgName :: LVar a d -> String
  lvarDbgName (LVar {state}) = (show$ unsafeName state)
      
  -- | Create an LVar.
  newLV :: IO a -> Par (LVar a d)
  newLV init = mkPar $ \k q -> do  
    logOffRecord q 7$ " [dbg-lvish] newLV: allocating... "
    state     <- init
    listeners <- B.new
    status    <- newIORef $ Active listeners
    name      <- newLVID
  #ifdef NONIDEM
    handlerStatus <- newIORef Dormant
    exec (k $ LVar {state, status, name, handlerStatus}) q
  #else
    exec (k $ LVar {state, status, name}) q
  #endif
  
  -- | Do a threshold read on an LVar
  getLV :: (LVar a d)                  -- ^ the LVar 
        -> (a -> Bool -> IO (Maybe b)) -- ^ already past threshold?
                                       -- The @Bool@ indicates whether the LVar is FROZEN.
        -> (d ->         IO (Maybe b)) -- ^ does @d@ pass the threshold?
        -> Par b
  getLV lv@(LVar {state, status}) globalThresh deltaThresh = mkPar $ \k q -> do
    -- tradeoff: we fastpath the case where the LVar is already beyond the
    -- threshhold by polling *before* enrolling the callback.  The price is
    -- that, if we are not currently above the threshhold, we will have to poll
    -- /again/ after enrolling the callback.  This race may also result in the
    -- continuation being executed twice, which is permitted by idempotence.
    let uniqsuf = ", lv "++lvarDbgName lv++" on worker "++(show$ Sched.no q)
    
    logWith q 7$ " [dbg-lvish] getLV: first readIORef "++uniqsuf
    curStatus <- readIORef status
    case curStatus of
      Active listeners -> do
        logWith q 7$ " [dbg-lvish] getLV (active): check globalThresh"++uniqsuf
        tripped <- globalThresh state False
        case tripped of
          Just b -> exec (k b) q -- already past the threshold; invoke the
                                 -- continuation immediately        
  
          Nothing -> do          -- /transiently/ not past the threshhold; block        
  
            execFlag <- newDedupCheck
            let onUpdate d = unblockWhen $ deltaThresh d
                onFreeze   = unblockWhen $ globalThresh state True
                {-# INLINE unblockWhen #-}
                unblockWhen thresh tok q = do
                  let uniqsuf = ", lv "++(lvarDbgName lv)++" on worker "++(show$ Sched.no q)
                  logWith q 7$ " [dbg-lvish] getLV (active): callback: check thresh"++uniqsuf
                  tripped <- thresh
                  whenJust tripped $ \b -> do        
                    B.remove tok
                    winnerCheck execFlag q (Sched.pushWork q (k b)) (return ())
  
            logWith q 8$ " [dbg-lvish] getLV "++show(unsafeName execFlag)++
                         ": blocking on LVar, registering listeners..."
            -- add listener, i.e., move the continuation to the waiting bag
            tok <- B.put listeners $ Listener onUpdate onFreeze
  
            -- but there's a race: the threshold might be passed (or the LVar
            -- frozen) between our check and the enrollment as a listener, so we
            -- must poll again
            logWith q 8$ " [dbg-lvish] getLV (active): second frozen check"++uniqsuf
            frozen <- isFrozen lv
            logWith q 7$ " [dbg-lvish] getLV (active): second globalThresh check"++uniqsuf
            tripped' <- globalThresh state frozen
            case tripped' of
              Just b -> do
                logWith q 7$ " [dbg-lvish] getLV (active): second globalThresh tripped, remove tok"++uniqsuf
                B.remove tok  -- remove the listener we just added, and
  
                winnerCheck execFlag q (exec (k b) q) (sched q)
                        -- execute the continuation. this work might be
                        -- redundant, but in idempotence-mode that's OK
              Nothing -> sched q
  
      --------------------------------------------------------------------------------
      -- Freezing or Frozen:
      _ -> do
        logWith q 7$ " [dbg-lvish] getLV (frozen): about to check globalThresh"++uniqsuf
        tripped <- globalThresh state True
        case tripped of
          Just b -> do -- logWith q 9$ " [dbg-lvish] getLV (frozen): thresh met, invoking continuation "++uniqsuf
                       exec (k b) q -- already past the threshold; invoke the
                                    -- continuation immediately                    
          Nothing -> sched q     -- We'll NEVER be above the threshold.
                                 -- Shouldn't this be an ERROR? (blocked-indefinitely)
                                 -- Depends on our semantics for runPar quiescence / errors states.
  
  
  {-# INLINE newDedupCheck #-}
  {-# INLINE winnerCheck #-}
  winnerCheck :: DedupCell -> Sched.State a s  -> IO () -> IO () -> IO ()
  newDedupCheck :: IO DedupCell
  
  #if GET_ONCE
  
  #  if 0
  type DedupCell = IORef Bool
  newDedupCheck = newIORef False -- True means someone has already won.
  winnerCheck execFlag q tru fal = do                
    ticket <- readForCAS execFlag
    if (peekTicket ticket) 
      then do logWith q 8 $ " [dbg-lvish] dedup winnerCheck failed.."
              fal
      else do
        (winner, _) <- casIORef execFlag ticket True
        logWith q 8 $ " [dbg-lvish] dedup "++show(unsafeName execFlag)
                   ++" on worker "++ (show$ Sched.no q) ++": winner check? " ++show winner
                   ++ ", ticks " ++ show (ticket, peekTicket ticket)
        if winner then tru else fal
  #  else
  
  type DedupCell = C.AtomicCounter
  newDedupCheck = C.newCounter 0 
  winnerCheck execFlag q tru fal = do
    cnt <- C.incrCounter 1 execFlag
    logWith q 8 $ " [dbg-lvish] dedup "++show(unsafeName execFlag)
               ++" on worker "++ (show$ Sched.no q) ++": winner check? " ++show (cnt==1)
               ++ ", counter val " ++ show cnt
    if cnt==1 then tru else fal
  
  #  endif
  #else
  type DedupCell = ()
  newDedupCheck = return ()
  winnerCheck _ _ tr _ = tr
  #endif
  
  -- | A version of the winner check that either lets the current thread
  -- through, or kills it and returns to the scheduler.
  -- winnerCheckPar :: DedupCell -> Par ()
  -- winnerCheckPar execFlag = mkPar $ \k q ->
  --   undefined
  
  
  
  -- | Update an LVar.
  putLV_ :: LVar a d                 -- ^ the LVar
         -> (a -> Par (Maybe d, b))  -- ^ how to do the put, and whether the LVar's
                                     -- value changed
         -> Par b
  #ifdef NONIDEM
  putLV_ lv@(LVar {state, status, name, handlerStatus}) doPut = mkPar body  where
    body k q = 
      let uniqsuf = ", lv "++(lvarDbgName lv)++" on worker "++(show$ Sched.no q)
          putAfterFrzExn = E.throw$ PutAfterFreezeExn "Attempt to change a frozen LVar"
  
          setPutFlag   = Sched.setStatus q name
          clearPutFlag = Sched.setStatus q noName
  
          cont (delta, ret) = ClosedPar $ \q -> do
            logWith q 8 $ " [dbg-lvish] putLV: status read"++uniqsuf
            curStatus <- readIORef status
            clearPutFlag          -- retract our modification intent
            whenJust delta $ \d -> do
              case curStatus of
                Freezing -> putAfterFrzExn
                Frozen   -> putAfterFrzExn
                Active listeners -> do
                  -- FIXME: need finer granularity here:
                  logWith q 9 $ " [dbg-lvish] putLV: calling each listener's onUpdate"++uniqsuf
                  B.foreach listeners $ \(Listener onUpdate _) tok -> onUpdate d tok q
            exec (k ret) q
  
          putNonidemp = do
            logWith q 8 $ " [dbg-lvish] putLV/nonidem: setPutFlag"++uniqsuf
            setPutFlag    -- publish our intent to modify the LVar
            logWith q 8 $ " [dbg-lvish] putLV/nonidem: initial handlerStatus read"++uniqsuf
            ticket    <- readForCAS handlerStatus
            case peekTicket ticket of
              Dormant -> do
                logWith q 8 $ " [dbg-lvish] putLV/nonidem: about to exec the real mutation"++uniqsuf
                exec (close (doPut state) cont) q  -- possibly modify the LVar
              Installing n ps -> do
                logWith q 8 $ " [dbg-lvish] putLV/nonidem: casIORef handlerStatus"++uniqsuf
                (success, _) <- casIORef handlerStatus ticket $! Installing n $! (ClosedPar $ body k):ps
                logWith q 8 $ " [dbg-lvish] putLV/nonidem: clearPutFlag"++uniqsuf
                clearPutFlag
                if success then sched q else putNonidemp
      in putNonidemp
  #else
  putLV_ lv@(LVar {state, status, name}) doPut = mkPar $ \k q -> do
    let uniqsuf = ", lv "++(lvarDbgName lv)++" on worker "++(show$ Sched.no q)
        putAfterFrzExn = E.throw$ PutAfterFreezeExn "Attempt to change a frozen LVar"
    logWith q 8 $ " [dbg-lvish] putLV: initial lvar status read"++uniqsuf
    fstStatus <- readIORef status
    case fstStatus of
      Freezing -> putAfterFrzExn
      Frozen   -> putAfterFrzExn
      Active listeners -> do
        logWith q 8 $ " [dbg-lvish] putLV: setStatus,"++uniqsuf
        Sched.setStatus q name         -- publish our intent to modify the LVar
        let cont (delta, ret) = ClosedPar $ \q -> do
              logWith q 8 $ " [dbg-lvish] putLV: read final status before unsetting"++uniqsuf
              sndStatus <- readIORef status  -- read the frozen bit *while q's status is marked*
              logWith q 8 $ " [dbg-lvish] putLV: UN-setStatus"++uniqsuf
              Sched.setStatus q noName       -- retract our modification intent
              -- AFTER the retraction, freezeLV is allowed to set the state to Frozen.
              whenJust delta $ \d -> do
                case sndStatus of
                  Frozen -> putAfterFrzExn
                  _ -> do
                    logWith q 9 $ " [dbg-lvish] putLV: calling each listener's onUpdate"++uniqsuf
                    B.foreach listeners $ \(Listener onUpdate _) tok -> onUpdate d tok q
              exec (k ret) q
        logWith q 5 $ " [dbg-lvish] putLV: about to mutate lvar"++uniqsuf
        exec (close (doPut state) cont) q
  #endif
  
  -- | Update an LVar without generating a result.  
  putLV :: LVar a d             -- ^ the LVar
        -> (a -> IO (Maybe d))  -- ^ how to do the put, and whether the LVar's
                                 -- value changed
        -> Par ()
  putLV lv doPut = putLV_ lv doPut'
    where doPut' a = do r <- liftIO (doPut a); return (r, ())
  
  -- | Freeze an LVar (introducing quasi-determinism).
  --   It is the data structure implementor's responsibility to expose this as quasi-deterministc.
  freezeLV :: LVar a d -> Par ()
  freezeLV lv@(LVar {name, status}) = mkPar $ \k q -> do
    let uniqsuf = ", lv "++(lvarDbgName lv)++" on worker "++(show$ Sched.no q)
    logWith q 5 $ " [dbg-lvish] freezeLV: atomic modify status to Freezing"++uniqsuf
    oldStatus <- atomicModifyIORef' status $ \s -> (Freezing, s)    
    case oldStatus of
      Frozen   -> return ()
      Freezing -> return ()
      Active listeners -> do
        logWith q 7 $ " [dbg-lvish] freezeLV: begin busy-wait for putter status"++uniqsuf
        Sched.await q (name /=)  -- wait until all currently-running puts have
                                 -- snapshotted the active status
        logWith q 7 $ " [dbg-lvish] freezeLV: calling each listener's onFreeze"++uniqsuf
        B.foreach listeners $ \Listener {onFreeze} tok -> onFreeze tok q
        logWith q 7 $ " [dbg-lvish] freezeLV: finalizing status as Frozen"++uniqsuf
        writeIORef status Frozen
    exec (k ()) q
    
  ------------------------------------------------------------------------------
  -- Handler pool operations
  ------------------------------------------------------------------------------  
  
  -- | Create a handler pool.
  newPool :: Par HandlerPool
  newPool = mkPar $ \k q -> do
    cnt <- C.newCounter 0
    bag <- B.new
    let hp = HandlerPool cnt bag
    hpMsg q " [dbg-lvish] Created new pool" hp
    exec (k hp) q
    
  -- | Convenience function.  Execute a @Par@ computation in the context of a fresh handler pool.
  withNewPool :: (HandlerPool -> Par a) -> Par (a, HandlerPool)
  withNewPool f = do
    hp <- newPool
    a  <- f hp
    return (a, hp)
    
  -- | Convenience function.  Execute a @Par@ computation in the context of a fresh
  -- handler pool, while ignoring the result of the computation.
  withNewPool_ :: (HandlerPool -> Par ()) -> Par HandlerPool
  withNewPool_ f = do
    hp <- newPool
    f hp
    return hp
  
  data DecStatus = HasDec | HasNotDec
  
  -- | Close a @Par@ task so that it is properly registered with a handler pool.
  closeInPool :: Maybe HandlerPool -> Par () -> IO ClosedPar
  closeInPool Nothing c = return $ close c $ const (ClosedPar sched)
  closeInPool (Just hp) c = do
    decRef <- newIORef HasNotDec      -- in case the thread is duplicated, ensure
                                      -- that the counter is decremented only once
                                      -- on termination
    let cnt = numHandlers hp
        
        tryDecRef = do                -- attempt to claim the role of decrementer
          ticket <- readForCAS decRef
          case peekTicket ticket of
            HasDec    -> return False
            HasNotDec -> do
              (firstToDec, _) <- casIORef decRef ticket HasDec
              return firstToDec
              
        onFinishHandler _ = ClosedPar $ \q -> do
          shouldDec <- tryDecRef      -- are we the first copy of the thread to
                                      -- terminate?
          when shouldDec $ do
            C.incrCounter (-1) cnt                 -- record handler completion in pool
            quiescent <- C.readCounter cnt   -- check for (transient) quiescence
            when (quiescent==0) $ do       -- wake any threads waiting on quiescence
              hpMsg q " [dbg-lvish] -> Quiescent now.. waking conts" hp 
              let invoke t tok = do
                    B.remove tok
                    Sched.pushWork q t                
              B.foreach (blockedOnQuiesce hp) invoke
          sched q
    C.incrCounter 1 $ numHandlers hp            -- record handler invocation in pool
    return $ close c onFinishHandler  -- close the task with a special "done"
                                      -- continuation that clears it from the
                                      -- handler pool
  
  -- | Add a handler to an existing pool.
  --
  --   The initial snapshot action must be carefully designed.  It needs to:
  --   (1) Read the data structure contents and fork tasks (NONBLOCKING)
  --   (2) Release the lock on the data structure (i.e. call the unlock action passed to it)
  --   (3) Do whatever else it feels like. (possibly blocking)
  {-# INLINE addHandler #-}
  addHandler :: Maybe HandlerPool           -- ^ pool to enroll in, if any
             -> LVar a d                    -- ^ LVar to listen to
             -> (a -> IO () -> Par ())     -- ^ initial snapshot callback on handler registration
             -> (d -> IO (Maybe (Par ())))  -- ^ subsequent callbacks: updates
             -> Par ()
  #ifdef NONIDEM
  addHandler hp LVar {state, status, name, handlerStatus} globalCB updateThresh =
    let acqLock q = do
          ticket <- readForCAS handlerStatus
          let (newStatus, wait) = case peekTicket ticket of
                Dormant         -> (Installing 1 [],     True)
                Installing n ps -> (Installing (n+1) ps, False)
          (success, _) <- casIORef handlerStatus ticket newStatus
          logWith q 4 $ " [dbg-lvish] addHandler/attempt to acqLock, CAS: "++show success
          if success
            then when wait $ Sched.await q (name /=) -- first handler installation; wait
                                                     -- for existing puts to complete
            else acqLock q  -- retry lock acquisition
  
        relLock q = do
          ticket <- readForCAS handlerStatus
          let (newStatus, ps) = case peekTicket ticket of
                Dormant         -> error "BUG: acq/rel mismatch on handler lock"
                Installing 1 ps -> (Dormant, ps)
                Installing n ps -> (Installing (n-1) ps, [])
          (success, _) <- casIORef handlerStatus ticket newStatus
          logWith q 4 $ " [dbg-lvish] addHandler/attempt to relLock, CAS: "++show success
          if success
            then forM_ ps $ Sched.pushWork q 
            else relLock q -- retry lock release
            
        spawnWhen thresh q = do
          tripped <- thresh
          whenJust tripped $ \cb -> do
            logWith q 5 " [dbg-lvish] addHandler: Delta threshold triggered, pushing work.."
            closed <- closeInPool hp cb
            Sched.pushWork q closed
            
        onUpdate d _ q = spawnWhen (updateThresh d) q
        onFreeze   _ _ = return ()
    in mkPar $ \k q -> do
      acqLock q
      curStatus <- readIORef status 
      case curStatus of
        Active listeners ->             -- enroll the handler as a listener
          do B.put listeners $ Listener onUpdate onFreeze; return ()
        Frozen   -> return ()           -- frozen, so no need to enroll
        Freezing -> return ()           -- frozen, so no need to enroll
        
      logWith q 4 " [dbg-lvish] addHandler: calling globalCB.."
      -- At registration time, traverse (globally) over the previously inserted items
      -- to launch any required callbacks.
      let k2 :: () -> ClosedPar
          k2 () = case k () of 
                    ClosedPar go -> ClosedPar $ \ q2 -> do 
                      logWith q 5 " [dbg-lvish] addHandler: inside k2"
                      go q2 -- Continue after the addHandler.
      -- Ported over bugfix here from master branch.
      -- There's a quirk here where we need to stick in the lock release
      -- to happen afetr the globalCB is done (in the continuation).
      exec (close (globalCB state (relLock q)) k2) q
      -- Warning! What happens if the globalCB blocks and then wakes on
      -- a different thread? => Should be ok.
  #else
  addHandler hp LVar {state, status} globalCB updateThresh = 
    let spawnWhen thresh q = do
          tripped <- thresh
          whenJust tripped $ \cb -> do
            logWith q 5 " [dbg-lvish] addHandler: Delta threshold triggered, pushing work.."
            closed <- closeInPool hp cb
            Sched.pushWork q closed
        onUpdate d _ q = spawnWhen (updateThresh d) q
        onFreeze   _ _ = return ()
  
    in mkPar $ \k q -> do
      curStatus <- readIORef status 
      case curStatus of
        Active listeners ->             -- enroll the handler as a listener
          do B.put listeners $ Listener onUpdate onFreeze; return ()
        Frozen   -> return ()           -- frozen, so no need to enroll
        Freezing -> return ()           -- frozen, so no need to enroll
  
      logWith q 4 " [dbg-lvish] addHandler: calling globalCB.."
      -- At registration time, traverse (globally) over the previously inserted items
      -- to launch any required callbacks.
      exec (close (globalCB state (return ())) k) q
  #endif
  
  -- | Block until a handler pool is quiescent.
  quiesce :: HandlerPool -> Par ()
  quiesce hp@(HandlerPool cnt bag) = mkPar $ \k q -> do
    hpMsg q " [dbg-lvish] Begin quiescing pool, identity= " hp
  
    flg <- newDedupCheck
    -- Wrap the continuation with a dedup Check:
    -- TODO: Issue #99 addresses whether this could be combined with the B.remove step:
    let k' = ClosedPar$ \q -> winnerCheck flg q
                                 (exec (k ()) q) (sched q)
    -- tradeoff: we assume that the pool is not yet quiescent, and thus enroll as
    -- a blocked thread prior to checking for quiescence
    tok <- B.put bag k'
    hpMsg q " [dbg-lvish] quiesce: poll count" hp
    quiescent <- C.readCounter cnt
    if (quiescent==0) then do
      hpMsg q " [dbg-lvish] already quiesced, remove token from bag" hp
      B.remove tok
      exec k' q 
    else do 
      logOffRecord q 4 " [dbg-lvish] -> Not quiescent yet, back to sched"
      sched q
  
  -- | A global barrier.
  quiesceAll :: Par ()
  quiesceAll = mkPar $ \k q -> do
    logWith q 1 " [dbg-lvish] quiesceAll: initiating global barrier."
    sched q
    logWith q 1 " [dbg-lvish] quiesceAll: Past global barrier."
    exec (k ()) q
  
  -- | Freeze an LVar after a given handler quiesces.
  -- 
  -- This is quasi-deterministic.
  --
  -- The initial snapshot action must fulfill the same contract as in addHandler.
  freezeLVAfter :: LVar a d                    -- ^ the LVar of interest
                -> (a -> IO () -> Par ())      -- ^ initial snapshot callback on handler registration
                -> (d -> IO (Maybe (Par ())))  -- ^ subsequent callbacks: updates
                -> Par ()
  freezeLVAfter lv globalCB updateCB = do
    hp <- newPool
    addHandler (Just hp) lv globalCB updateCB
    logStrLn 4 " [dbg-lvish] freezeLVAfter: addHandler complete, calling quiesce"
    quiesce hp  
    freezeLV lv
  
  ------------------------------------------------------------------------------
  -- Par monad operations
  ------------------------------------------------------------------------------
  
  -- | Fork a child thread, optionally in the context of a handler pool.
  forkHP :: Maybe HandlerPool -> Par () -> Par ()
  forkHP mh child = mkPar $ \k q -> do
    closed <- closeInPool mh child
    Sched.pushWork q (k ()) -- "Work-first" policy.
  --  hpMsg q " [dbg-lvish] incremented and pushed work in forkInPool, now running cont" hp   
    exec closed q  
    
  -- | Fork a child thread.
  fork :: Par () -> Par ()
  fork f = forkHP Nothing f
  
  -- | Perform an @IO@ action.
  liftIO :: IO a -> Par a
  liftIO io = mkPar $ \k q -> do
    r <- io
    exec (k r) q
  
  instance IOClass.MonadIO Par where
    liftIO = liftIO
  
  -- | IF compiled with debugging support, this will return the Logger used by the
  -- current Par session, otherwise it will return Nothing.
  getLogger :: Par (Maybe L.Logger)
  getLogger = mkPar $ \k q -> 
    let mlgr = Sched.logger q in
    exec (k mlgr) q
  
  #ifdef DEBUG_LVAR
  -- | Return the worker that we happen to be running on.  (NONDETERMINISTIC.)
  getWorkerNum :: Par Int
  getWorkerNum = mkPar $ \k q -> exec (k (Sched.no q)) q
  #endif
  
  -- | Generate a random boolean in a core-local way.  Fully nondeterministic!
  instance MonadToss Par where  
    toss = mkPar $ \k q -> do  
  --    g <- readIORef $ Sched.prng q
      b <- uniformBool $ Sched.prng q
  --    let (b, g' ) = random g
  --    writeIORef (Sched.prng q) g'
      exec (k b) q
  
  -- | Cooperatively schedule other threads.
  yield :: Par ()  
  yield = mkPar $ \k q -> do
    Sched.yieldWork q (k ())
    sched q
    
  {-# INLINE sched #-}
  -- | Contract: This scheduler function only returns when ALL worker threads have
  -- completed their work and idled.
  sched :: SchedState -> IO ()
  sched q = do
    n <- Sched.next q
    case n of
      Just t  -> exec t q
      Nothing -> return ()
  
  -- Forcing evaluation of a LVar is fruitless.
  instance NFData (LVar a d) where
    rnf _ = ()
  
  
  -- | A variant with full control over the relevant knobs.
  --   
  --   Returns a list of flushed debug messages at the end (if in-memory logging was
  --   enabled, otherwise the list is empty).
  --
  --   This version of runPar catches ALL exceptions that occur within the runPar, and
  --   returns them via an Either.  The reason for this is that even if an error
  --   occurs, it is still useful to observe the log messages that lead to the failure.
  --
  --  WARNING: if LVish was installed without `-fdebug`, then the only
  --  log messages generated will be those explictly created by the user
  --  with `logDbgLn`.  All messages from the runtime system and core
  --  LVar data structures will be absent.
  -- 
  runParDetailed :: DbgCfg  -- ^ Debugging config
                 -> Int           -- ^ How many worker threads to use. 
                 -> Par a         -- ^ The computation to run.
                 -> IO ([String], Either E.SomeException a)
  runParDetailed cfg numWrkrs comp = do
  -- ifndef DEBUG_LVAR
  --   when (dbgScheduling /= True) $ -- || dbgRange /= Nothing  
  --     error "runParDetailed: asked to control scheduling, but compiled without debugging support."
  -- endif
    (lgr,queues) <- Sched.new cfg numWrkrs noName 
    case lgr of
      Nothing -> return ()
      Just l  -> L.logOn l (L.OffTheRecord 1 " [dbg-lvi6sh] New logger & scheds created... entering main loop.")
     
    -- We create a thread on each CPU with forkOn.  The CPU on which
    -- the current thread is running will host the main thread; the
    -- other CPUs will host worker threads.
    main_cpu <- Sched.currentCPU
    answerMV <- newEmptyMVar
  
    let grabLogs = do  
          logOffRecord (Prelude.head queues) 1 " [dbg-lvish] parent thread escaped unscathed.  Optionally closing logger."
          case lgr of 
            Nothing -> return []
            Just lgr -> do L.closeIt lgr
                           L.flushLogs lgr -- If in-memory logging is off, this will be empty.
        mlog s = case lgr of 
                   Nothing -> return ()
                   Just l  -> L.logOn l (L.OffTheRecord 4 s)
  
    -- Use Control.Concurrent.Async to deal with exceptions:
    ----------------------------------------------------------------------------------
    let runWorker :: (Int,Sched.State ClosedPar LVarID) -> IO ()
        runWorker (cpu, q) = do 
          if (cpu /= main_cpu)
             then do logOffRecord q 3 $  " [dbg-lvish] Auxillary worker #"++show cpu++" starting (main cpu "++show main_cpu++")."
                     sched q
                     logOffRecord q 3 $  " [dbg-lvish] Auxillary worker #"++show cpu++" exitting."
             else let k x = ClosedPar $ \q -> do                       
                        logOffRecord q 3 " [dbg-lvish] Final continuation of main worker: reenter sched to cleanup."
                        sched q      -- ensure any remaining, enabled threads run to 
                                     -- completion prior to returning the result
                        -- FIXME: this continuation gets duplicated.
                        logOffRecord q 3 " [dbg-lvish] Main worker: past global barrier, putting answer."
                        b <- tryPutMVar answerMV x
  #ifdef GET_ONCE
                        unless b $ error "Final continuation of Par computation was duplicated, in spite of GET_ONCE!"
  #endif
                        return ()
                  in do logOffRecord q 3 " [dbg-lvish] Main worker thread starting."
                        exec (close comp k) q
  
    -- Here we want a traditional, fork-join parallel loop with proper exception handling:
    let loop [] asyncs = do tid <- myThreadId
                            mlog $ " [dbg-lvish] (main tid "++show tid++") Wait on at least one async to complete.."
                            (_,x) <- A.waitAnyCatch asyncs
                            -- We could do a binary tree of waitBoth here, but this should work for now:
                            case x of
                              Left e -> return $! Left e
                              Right () -> waitloop asyncs -- If one finishes, all are trying to.
        loop ((cpu,q):tl) asyncs = 
          A.withAsyncOn cpu (runWorker (cpu,q))
                        (\a -> loop tl (a:asyncs))
        waitloop [] = do 
                         mlog " [dbg-lvish] All asyncs complete, read final answer MVar."
                         fmap Right (dbgTakeMVar [] "retrieve final runPar answer, after workers complete" answerMV)
  --                       fmap Right (takeMVar answerMV)
        waitloop (hd:tl) = do mlog (" [dbg-lvish] Waiting for one async.. "++show(1+length tl)++" remaining")
                              me <- A.waitCatch hd
                              case me of 
                                Left e    -> return $! Left e
                                Right ()  -> waitloop tl
    ----------------------------------------
    -- (1) There was a BUG in 'loop' at some point:
    --    "thread blocked indefinitely in an STM transaction"
    ans <- loop (zip (main_cpu : filter (/= main_cpu) [0..]) queues) []
    ----------------------------------------
    -- (2) This has the same problem as 'loop':
    --  ls <- mapM (\ pr@(cpu,_) -> Async.asyncOn cpu (runWorker pr)) (zip [0..] queues)
    --  mapM_ wait ls
    ----------------------------------------
    -- (3) Using this FOR NOW, but it does NOT pin to the right processors:
    --  A.mapConcurrently runWorker (zip [0..] queues)
    ----------------------------------------
    logs <- grabLogs
    return $! (logs,ans)
  
  
  
  defaultRun :: Par b -> IO b
  defaultRun = fmap (fromRight . snd) .
               runParDetailed cfg numCapabilities
    where
     cfg = DbgCfg { dbgRange = Just (0,dbgLvl)
                  , dbgDests = [L.OutputTo stderr, L.OutputEvents]
                  , dbgScheduling  = False }
  
  -- | Run a deterministic parallel computation as pure.
  runPar :: Par a -> a
  runPar = unsafePerformIO . defaultRun
  
  -- | A version that avoids an internal `unsafePerformIO` for calling
  -- contexts that are already in the `IO` monad.
  runParIO :: Par a -> IO a
  runParIO = defaultRun
  
  -- | Debugging aid.  Return debugging logs, in realtime order, in addition to the
  -- final result.  This is like `runParDetailed` but uses the default settings.
  --
  --  WARNING: if LVish was installed without `-fdebug`, then the only
  --  log messages generated will be those explictly created by the user
  --  with `logDbgLn`.  All messages from the runtime system and core
  --  LVar data structures will be absent.
  -- 
  runParLogged :: Par a -> IO ([String],a)
  runParLogged comp = do
  -- ifndef DEBUG_LVAR
  --   error "runParLogged: this function is disabled when LVish is compiled without debugging support."
  -- endif
    (logs,ans) <- runParDetailed 
                     DbgCfg { dbgRange = (Just (0,dbgLvl))
                            , dbgDests = [L.OutputEvents, L.OutputInMemory]
                            , dbgScheduling = False }  
                     numCapabilities comp
    return $! (logs,fromRight ans)
  
  -- | Convert from a Maybe back to an exception.
  fromRight :: Either E.SomeException a -> a
  fromRight (Right x) = x
  fromRight (Left e) = E.throw e
  
  -- {-# INLINE atomicModifyIORef_ #-}
  -- atomicModifyIORef_ :: IORef a -> (a -> a) -> IO ()
  -- atomicModifyIORef_ ref fn = atomicModifyIORef' ref (\ x -> (fn x,()))
  
  {-# NOINLINE unsafeName #-}
  unsafeName :: a -> Int
  unsafeName x = unsafePerformIO $ do 
     sn <- makeStableName x
     return (hashStableName sn)
  
  {-# INLINE hpMsg #-}
  hpMsg :: Sched.State a s -> String -> HandlerPool -> IO ()
  #ifdef DEBUG_LVAR
  hpMsg q msg hp = do
      s <- hpId_ hp
      logWith q 3 $ msg++", pool identity= " ++s
  #else
  hpMsg _ _ _ = do
       return ()
  #endif
  
  -- {-# NOINLINE hpId #-}   
  -- hpId :: HandlerPool -> String
  -- hpId hp = unsafePerformIO (hpId_ hp)
  
  #ifdef DEBUG_LVAR
  -- | Debugging tool for printing which HandlerPool
  hpId_ :: HandlerPool -> IO String
  hpId_ (HandlerPool cnt bag) = do
    sn1 <- makeStableName cnt
    sn2 <- makeStableName bag
    c   <- C.readCounter cnt
    return $ show (hashStableName sn1) ++"/"++ show (hashStableName sn2) ++
             " transient cnt "++show c
  #endif
  
  #ifdef DEBUG_LVAR
  -- | For debugging purposes.  This can help us figure out (by an ugly
  --   process of elimination) which MVar reads are leading to a "Thread
  --   blocked indefinitely" exception.
  --
  -- FIXME: This function is suppressing "thread blocked indefinitely in an MVar
  -- operation" exceptions and continue trying to read from the LVar.
  busyTakeMVar :: [ThreadId] -> String -> MVar a -> IO a
  busyTakeMVar tids msg mv = 
    do b <- L.newBackoff maxWait
       try b
   where
   maxWait = 10000 -- nanoseconds
   timeOut = (3 * 1000 * 1000) -- three seconds, only for debugging.
   try bkoff | totalWait bkoff >= timeOut = do
  --     error "busyTakeMVar (debugging): time-out expired for waiting on MVar"
     -- when dbg $ do
       tid <- myThreadId
       -- After we've failed enough times, start complaining:
       hPrintf stderr "%s not unblocked yet, for: %s\n" (show tid) msg
       stats <- Prelude.mapM threadStatus tids 
       hPrintf stderr $ "Worker statuses: " ++ show (zip tids stats) ++"\n"
       try =<< L.backoff bkoff 
   try bkoff = do
     x <- tryTakeMVar mv
     case x of
       Just y  -> return y
       Nothing -> try =<< L.backoff bkoff 
  #endif
  
  dbgTakeMVar :: forall a. [ThreadId] -> String -> MVar a -> IO a
  #ifdef DEBUG_LVAR
  dbgTakeMVar = busyTakeMVar
  #else
  dbgTakeMVar _ _ = takeMVar
  #endif
  
  