content: -----------------------------------------------------------------------------
  -- |
  -- Module      :  Control.Concurrent.SHFSTM.Internal
  -- Copyright   :  (c) D. Sabel, Goethe-University, Frankfurt a.M., Germany
  -- License     :  BSD-style 
  -- 
  -- Maintainer  :  sabel <at> ki.cs.uni-frankfurt.de
  -- Stability   :  experimental
  -- Portability :  non-portable (needs GHC and extensions)
  --
  --  
  -- This module implements transaction execution
  -----------------------------------------------------------------------------
  
  module Control.Concurrent.STM.Internal (
    TLOG,
    emptyTLOG,
    RetryException(..),
    globalRetry,
    commit,
    newTVarWithLog,
    readTVarWithLog,
    writeTVarWithLog,
    orElseWithLog,
    orRetryWithLog
    ) where
  
  import Prelude hiding(catch)
  import Control.Exception 
  import Control.Concurrent
  import Control.Concurrent.STM.Internal.TVar  
  import Control.Concurrent.STM.Internal.TransactionLog
  import qualified Control.Concurrent.STM.Internal.Map as Map
  import qualified Control.Concurrent.STM.Internal.Set as Set
  import Data.List
  import Data.IORef
  import Data.Maybe
  import Data.Typeable     
  -- | The 'RetryException' is thrown from the committing transaction
  -- to conflicting transactions
  
  
  
  
    
  data RetryException = RetryException 
       deriving (Typeable,Show)
  
  instance Exception RetryException
  
  -- | 'newTVarWithLog' creates a new TVar
  newTVarWithLog :: TLOG      -- ^ the transaction log
                 -> a         -- ^ the content of the TVar
                 -> IO (TVar a)
  newTVarWithLog (TLOG tlog) content =
    uninterruptibleMask_ $
     do
      mid <- myThreadId
      tvar_Id <- nextCounter
      lg <- readIORef tlog                   -- access the Transaction-Log
      let ((la,ln,lw):xs) = tripelStack lg   -- access the La,Ln,Lw lists
      -- Create the TVar  ...
      content_global <- newMVar content       -- set global content
      pointer_local_content <- newIORef [content]
      let mp = Map.insert mid pointer_local_content Map.empty
      content_local   <- seq mp (newMVar mp)
      notify_list     <- newMVar Set.empty
      unset_lock      <- newEmptyMVar
      -- loc <- newIORef (Just mid)
      content_waiting_queue <- newMVar []                  -- empty broadcast list
      content_tvarx <-         newMVar (TV {globalContent = content_global,
                                    localContent  = content_local,
                                    notifyList    = notify_list,
                                    lock          = unset_lock,   
                                    waitingQueue  = content_waiting_queue
                                    -- local = loc
                                    })
                                           
      let tvany = TVarAny (tvar_Id,content_tvarx)
      let tva   = TVarA content_tvarx
      let tvar = TVar (tva,tvany)
      -- ------------- 
      writeIORef tlog (lg{tripelStack=(Set.insert tvany la,Set.insert tvany ln,lw):xs}) -- adjust the Transaction Log
      return tvar
  
      
  
  -- | 'readTVarWithLog' performs the readTVar-operation
  readTVarWithLog :: TLOG   -- ^ the transaction log
                  -> TVar a -- ^ the  'TVar'
                  -> IO a   -- ^ the content of the 'TVar' (local)
  readTVarWithLog (TLOG tlog) ptvar@(TVar (_,tvany)) =
   do
    
    res <- tryReadTVarWithLog (TLOG tlog) ptvar
    case res of
      Right r -> return r
      Left blockvar -> do
                       mid <- myThreadId
                       takeMVar blockvar
                       readTVarWithLog (TLOG tlog) ptvar
      
  tryReadTVarWithLog (TLOG tlog) ptvar@(TVar (TVarA tva,tvany@(TVarAny tx))) = uninterruptibleMask_ $
    do
      _tva <-  takeMVar tva  -- access the TVar
      lg <- readIORef tlog  -- access the Log-File
      let ((la,ln,lw):xs) = tripelStack lg
      mid <- myThreadId        -- the ThreadId
      if tvany `Set.member` la then do
        -- x in L_a, local copy exists
        uninterruptibleMask_ (
         do       
          localmap <- readMVar  (localContent _tva)
          lk <- readIORef $ fromJust $ Map.lookup mid localmap
          let (x:xs) = lk
          putMVar tva _tva
          return (Right x)        
          )
        else                  
        -- TVar not in read TVars
         do
          uninterruptibleMask_ ( 
           do 
            b <- isEmptyMVar (lock _tva)
            if b -- not locked
             then 
                  do
                   nl <- takeMVar (notifyList _tva) 
                   putMVar (notifyList _tva) (Set.insert mid nl) -- add to notifyList
                   globalC <- readMVar  (globalContent _tva) -- read global content
                   content_local <- newIORef [globalC]
                   mp <- takeMVar (localContent _tva)
                   writeIORef tlog (lg{readTVars = Set.insert tvany (readTVars lg),tripelStack = (Set.insert tvany la,ln,lw):xs})
                   putMVar (localContent _tva) (Map.insert mid content_local mp)        -- copy to local tvar stack
                               -- adjust the transaction log
                   putMVar tva _tva
                   return (Right globalC)
        
             else -- locked
                  do 
                   blockvar <- newEmptyMVar
                   wq <- takeMVar (waitingQueue _tva)
                   putMVar (waitingQueue _tva) (blockvar:wq)
                   putMVar tva _tva
                   return (Left blockvar)
             )
  
  -- | 'writeTVarWithLog' performs the writeTVar operation and 
  --   adjusts the transaction log accordingly
  writeTVarWithLog :: TLOG   -- ^ the transaction log
                    -> TVar a -- ^ the  'TVar'
                    -> a      -- ^ the new content
                    -> IO ()  
  
  writeTVarWithLog (TLOG tlog) ptvar@(TVar (_,tvany)) con =
   do
    res <- tryWriteTVarWithLog (TLOG tlog) ptvar con
    case res of
      Right r -> return r
      Left blockvar -> do
                        takeMVar blockvar
                        writeTVarWithLog (TLOG tlog) ptvar con
     
  tryWriteTVarWithLog (TLOG tlog) ptvar@(TVar (TVarA tva,tvany@(TVarAny (id,m)))) con =
   uninterruptibleMask_ $ do
      _tva <-  takeMVar tva  -- access the TVar
      lg <- readIORef tlog  -- access the Log-File
      let ((la,ln,lw):xs) = tripelStack lg
      mid <- myThreadId        -- the ThreadId
      if tvany `Set.member` la then do
        -- x in L_a, local copy exists
        uninterruptibleMask_ (
         do       
          localmap <- readMVar  (localContent _tva)
          let ioref_with_old_content = fromJust $ Map.lookup mid localmap
          lk <- readIORef ioref_with_old_content 
          let (x:ys) = lk
          writeIORef  ioref_with_old_content (con:ys)
          writeIORef tlog (lg{tripelStack = (la,ln,Set.insert tvany lw):xs})
          putMVar tva _tva
          return $ Right ()
          )
       else                  
        -- TVar not in read TVars
         do
          uninterruptibleMask_ ( 
           do 
            b <- isEmptyMVar (lock _tva)
            if b -- not locked
             then 
                  do
                   globalC <- readMVar  (globalContent _tva) -- read global content
                   content_local <- newIORef [con]
                   mp <- takeMVar (localContent _tva)
                   putMVar (localContent _tva) (Map.insert mid content_local mp)        -- copy to local tvar stack
                               -- adjust the transaction log
                   writeIORef tlog (lg{readTVars = Set.insert tvany (readTVars lg),tripelStack = (Set.insert tvany la,ln,Set.insert tvany lw):xs})
                   putMVar tva _tva
                   return (Right ())
        
             else -- locked
                  do 
                   blockvar <- newEmptyMVar
                   wq <- takeMVar (waitingQueue _tva)
                   putMVar (waitingQueue _tva) (wq ++ [blockvar])
                   putMVar tva _tva
                   return (Left blockvar)
  
             )
  
  
             
  writeStartWithLog' (TLOG tlog) =
   uninterruptibleMask_ $
     do
      mid <- myThreadId    
      lg <- readIORef tlog  -- access the Log-File
      mid <- myThreadId
      let ((la,ln,lw):_)  = tripelStack lg
      let t               = readTVars lg
      let xs = t `Set.union` (la Set.\\ ln)  -- x1,...,xn
      res <- uninterruptibleMask_ (grabLocks mid (Data.List.sort $ Set.elems xs)  [])
      case res of
         Right _ -> do 
                     writeIORef tlog (lg{lockingSet = xs}) -- K := xs
                     return (Right ())
         Left lock -> return (Left lock)
      
  -- | 'writeStartWithLog' starts the commit phase, by 
  -- locking the read and written TVars
  
  writeStartWithLog :: TLOG -- ^ the transaction log
                       -> IO () 
  writeStartWithLog (TLOG tlog) =
    do
      res <- writeStartWithLog' (TLOG tlog)
      case res of 
       Left lock -> 
        do
         mid <- myThreadId
         takeMVar lock
         yield
         threadDelay 1000
         writeStartWithLog (TLOG tlog) 
       Right () -> return ()
       
  grabLocks mid [] _  = return (Right ())     
  grabLocks mid (ptvar@(TVarAny (i,tvany)):xs) held = 
    uninterruptibleMask_ $ do
            _tvany <- takeMVar tvany
            b <- tryPutMVar (lock _tvany) mid
            if b -- not locked
             then  do
               putMVar tvany _tvany
               grabLocks mid xs (ptvar:held)
              
              else  do -- already locked
               waiton <- newEmptyMVar
               l <- takeMVar (waitingQueue _tvany)
               putMVar (waitingQueue _tvany) (l ++ [waiton])
               putMVar tvany _tvany
               mapM_ (\(TVarAny (i,tvany)) -> do
                                                            _tv <- takeMVar tvany
                                                            takeMVar (lock _tv)
                                                            putMVar tvany _tv) (reverse held)
                                                            
               return (Left waiton)
                            
       
      
             
             
             
             
  iterateClearWithLog (TLOG tlog) [] = return ()
  iterateClearWithLog (TLOG tlog) ((TVarAny (id,tvany)):xs) =
    do
      mid <- myThreadId
      uninterruptibleMask_ $
        do 
         lg <- readIORef tlog
         _tvany <- takeMVar tvany
         ns <- takeMVar (notifyList _tvany)
         -- remove thread id from notify list:
         putMVar  (notifyList _tvany) (Set.delete mid ns)
         putMVar tvany _tvany
      iterateClearWithLog (TLOG tlog) xs    
  -- iterate clearWrite as long as possible (i.e. until the T-List is empty)
  -- Note: the iteration is not atomic (as in the paper)
  
  -- | 'writeClearWithLog' removes the notify entries of the committing transaction    
  writeClearWithLog (TLOG tlog) =
    do 
      lg <- readIORef tlog  -- access the Log-File
      let xs =  Set.elems (readTVars lg)
      iterateClearWithLog (TLOG tlog) xs
      writeIORef tlog (lg{readTVars=Set.empty})
  
       
             
  getIds []                       ls = return (Set.elems ls)          
  getIds ((TVarAny (_,tvany)):xs) ls =
   do
    l <- uninterruptibleMask_
     (do 
       _tvany <- takeMVar tvany 
       l <- takeMVar (notifyList _tvany)
       putMVar (notifyList _tvany) Set.empty
       putMVar tvany _tvany
       return l
      )
    getIds xs (Set.union l ls)
  -- | 'sendRetryWithLog' sends exceptions to the conflicting threads
    
  sendRetryWithLog :: TLOG -- ^ the transaction log
                   -> IO ()
                   
  sendRetryWithLog (TLOG tlog) =
    uninterruptibleMask_ $ 
     do
      lg <- readIORef tlog  -- access the Log-File
      mid <- myThreadId
      let ((la,ln,lw):xs) = tripelStack lg
      openLW <- getIds (Set.elems lw) Set.empty
      notify openLW
  
  notify [] = return ()
  notify (tid:xs) =
   do        
     mid <- myThreadId
     -- ack <- newEmptyMVar 
     throwTo tid RetryException -- send retry, 
     notify xs
  
             
  -- | 'writeTVWithLog' performs the write-Operations of the committing thread.
  writeTVWithLog :: TLOG -- ^ the transaction log
                 -> IO ()
                 
  writeTVWithLog (TLOG tlog) =
    do
      lg <- readIORef tlog  -- access the Log-File
      mid <- myThreadId
      let ((la,ln,lw):xs) = tripelStack lg
      let tobewritten = lw Set.\\ ln
      writeTVars (Set.elems tobewritten)
      writeIORef tlog lg{tripelStack=(la,ln,Set.empty):xs}
  
  writeTVars [] = return ()
  writeTVars ((TVarAny (tid,tvany)):xs) =
    do
     mid <- myThreadId
     _tvany <- takeMVar tvany
     localMap <- takeMVar (localContent _tvany)
     case Map.lookup mid localMap of
      Just conp -> 
        do
          con <- readIORef conp
          let (ltv:stackedcontent) = con 
          putMVar (localContent _tvany) (Map.delete mid localMap)
          -- copy to global storage
          takeMVar (globalContent _tvany)
          putMVar  (globalContent _tvany) ltv
          putMVar tvany _tvany
          writeTVars xs
  
     
  
  -- ------------------------------------------------
  -- unlockTVWithLog  implements (unlockTV)
  -- ------------------------------------------------
  
  -- | 'unlockTVWithLog' remove the locks of the TVars during commit
  unlockTVWithLog :: TLOG -- ^ the transaction log
                 -> IO ()
  
  unlockTVWithLog (TLOG tlog) =
    uninterruptibleMask_ $ do
      lg <- readIORef tlog  -- access the Log-File
      mid <- myThreadId
      let ((la,ln,lw):xs) = tripelStack lg
      let k = lockingSet lg
      unlockTVars (Set.elems k)
      writeIORef tlog lg{lockingSet = Set.empty}
      
  unlockTVars [] = return ()
  unlockTVars ((TVarAny (i,tvany)):xs) =    
     uninterruptibleMask_ $
      do
         mid <- myThreadId
         _tvany <- takeMVar tvany
         wq <- takeMVar (waitingQueue _tvany)    
         takeMVar (lock _tvany)
         putMVar (waitingQueue _tvany) []
         mapM_ (`putMVar` ()) wq
         putMVar tvany _tvany
         unlockTVars xs
  
  -- | 'writeTVnWithLog' writes the newly created TVars during commit    
  writeTVnWithLog :: TLOG -- ^ the transaction log
                 -> IO ()
  
  writeTVnWithLog (TLOG tlog) =
    do
      lg <- readIORef tlog  -- access the Log-File
      mid <- myThreadId
      let ((la,ln,lw):xs) = tripelStack lg
      let t = readTVars lg
      let k = lockingSet lg
      let toBeWritten = Set.elems ln
      writeNew toBeWritten
      writeIORef tlog lg{tripelStack=(la,Set.empty,lw):xs}
      
  writeNew [] = return ()
  writeNew ((TVarAny (i,tvany)):xs) =
   do
    mid <- myThreadId
    _tvany <- takeMVar tvany
    lmap <- takeMVar (localContent _tvany)
    if Map.null (Map.delete mid lmap) then
     case Map.lookup mid lmap of
      Just conp ->
        do 
         (con:_) <- readIORef conp
         takeMVar (globalContent _tvany)
         putMVar (globalContent _tvany) con
         putMVar (localContent _tvany) Map.empty
         putMVar tvany _tvany
         writeNew xs
      else 
         error "panic"
         
        
  -- ------------------------------------------------
  -- writeEndWithLog  implements (writeEnd)
  -- ------------------------------------------------
  
  -- | 'writeEndWithLog' clears the local TVars, during commit
  
  writeEndWithLog :: TLOG -- ^ the transaction log
                 -> IO ()
  
  writeEndWithLog (TLOG tlog) =
    do
      lg <- readIORef tlog  -- access the Log-File
      mid <- myThreadId
      let ((la,ln,lw):xs) = tripelStack lg
      let t = readTVars lg
      let k = lockingSet lg
      clearEntries (Set.elems la)
  
  clearEntries [] = return ()
  clearEntries ((TVarAny (_,tvany)):xs) =
   do
    mid <- myThreadId
    _tvany <- takeMVar tvany
    localmap <- takeMVar (localContent _tvany)
    putMVar (localContent _tvany) (Map.delete mid localmap)
    putMVar tvany _tvany
    
  -- ------------------------------------------------
  -- commit performs all the operations for committing
  -- ------------------------------------------------
  
  
  -- | 'commit' performs the operations for committing
  --
  --   - 'writeStartWithLog' to lock the read and to-be-written TVars
  --
  --   - 'writeClearWithLog' (iteratively) to remove the notify entries of the committing transaction
  --
  --   - 'sendRetryWithLog' (iteratively) to abort conflicting transactions
  --
  --   - 'writeTVWithLog' (iteratively) to write the local contents in to the global memory
  --
  --   - 'writeTVnWithLog' (iteratively) to create the newly created TVars in the global memory
  --
  --   - 'writeEndWithLog'  to clear the local TVar stack
  --
  --   - 'unlockTVWithLog' (iteratively) to unlock the global TVars
   
  commit :: TLOG -- ^ the transaction logs
         -> IO () 
  commit (TLOG tlog) =
   do
     mid <- myThreadId
  --    yield 
     writeStartWithLog (TLOG tlog)       -- writeStart 
  --    yield 
     writeClearWithLog (TLOG tlog)     -- clearWrite phase
  --    yield 
     sendRetryWithLog  (TLOG tlog) -- sendRetry phase
  --    yield 
     writeTVWithLog (TLOG tlog)   -- writeTV phase
  --    yield   
     writeTVnWithLog (TLOG tlog) -- writeTVn phase
  --    yield    
     writeEndWithLog (TLOG tlog) -- writeTVn phase
     unlockTVWithLog (TLOG tlog) -- unlockTV phase
  
             
     
  
  
  
  
      
  -- ------------------------------------------------
  -- retryCGlobWithLog  implements (retryCGlob)
  -- ------------------------------------------------
  -- | 'retryCGlobWithLog' performs the removal of notify entries during retry
  retryCGlobWithLog :: TLOG -- ^ the transaction log
                    -> IO ()
  retryCGlobWithLog (TLOG tlog) =
    do
      lg <- readIORef tlog  -- access the Log-File
      mid <- myThreadId
      let t = Set.elems (readTVars lg)
      uninterruptibleMask_ $ 
        do
         removeNotifyEntries mid t
         writeIORef tlog lg{readTVars = Set.empty}
         
  removeNotifyEntries mid [] = return ()
  removeNotifyEntries mid ((TVarAny (i,tvany)):xs) =
    uninterruptibleMask_ $
     do
      _tvany <- takeMVar tvany
      nlist <- takeMVar (notifyList _tvany)
      putMVar (notifyList _tvany)  (Set.delete mid nlist)
      putMVar tvany _tvany
      removeNotifyEntries mid xs
  
    
  
      
  -- ------------------------------------------------
  -- retryEndWithLog  implements (retryEnd) (slightly modified)
  -- ------------------------------------------------
  -- | 'retryEndWithLog' resets the transaction log and the local tvar content during retry
  
  retryEndWithLog :: TLOG -- ^ the transaction log
                  -> IO ()
  retryEndWithLog (TLOG tlog) =
    do
      lg <- readIORef tlog  -- access the Log-File
      mid <- myThreadId
      let ((la,ln,lw):xs) = tripelStack lg
      let tvset = la `Set.union` ln `Set.union` lw
      let toBeResetted = Set.elems tvset
      uninterruptibleMask_ $
        do 
         resetEntries mid toBeResetted    
         writeIORef tlog (lg{tripelStack = (Set.empty,Set.empty, Set.empty):xs})
      
  resetEntries mid [] = return ()
  resetEntries mid ((TVarAny (_,tvany)):xs) =    
    uninterruptibleMask_ $
         do
           _tvany <- takeMVar tvany
           localmap <- takeMVar (localContent _tvany)
           putMVar (localContent _tvany) (Map.delete mid localmap)
           putMVar tvany _tvany
           resetEntries mid xs
  
           
  -- ----------------------------
  -- globalRetry should be called when the transaction retries
  -------------------------------
  -- | 'globalRetry' should be called to retry a transaction, it iteratively 
  -- perform 'retryCGlobWithLog' and then 'retryEndWithLog'
  
  globalRetry :: TLOG -- ^ the transaction log
              -> IO () 
  globalRetry (TLOG tlog) =
   catch (
    do
     mid <- myThreadId
     uninterruptibleMask_ (retryCGlobWithLog (TLOG tlog))
     uninterruptibleMask_ (retryEndWithLog  (TLOG tlog))
      ) (\ RetryException -> globalRetry (TLOG tlog))
     
  -- ------------------------------------------------
  -- orElseWithLog (performs (orElse), i.e duplication of stacks etc.
  -- ------------------------------------------------
  
  -- | 'orElseWithLog' should be called for the evaluation of 'orElse' to duplicate the local TVar stacks and
  --    the entries in the transaction log
  orElseWithLog :: TLOG -- ^ the transaction log 
                ->  IO ()
  orElseWithLog (TLOG tlog) =
    do
      lg <- readIORef tlog  -- access the Log-File
      mid <- myThreadId
      let ((la,ln,lw):xs) = tripelStack lg
  
      -- double all local TVars
      uninterruptibleMask_ (
        do 
         doubleLocalTVars mid (Set.elems la)
         writeIORef tlog (lg{tripelStack=(la,ln,lw):((la,ln,lw):xs)})
       )
       
  doubleLocalTVars mid [] = return ()
  doubleLocalTVars mid ((TVarAny (_,tvany)):xs) =
   uninterruptibleMask_ $
     do
      _tvany <- takeMVar tvany
      localmap <- takeMVar (localContent _tvany)
      case Map.lookup mid localmap of
        Just conp -> 
          do 
           (x:ys) <- readIORef conp
           writeIORef conp (x:x:ys)
           putMVar (localContent _tvany) localmap
           putMVar tvany _tvany
           doubleLocalTVars mid xs
  
  -- ------------------------------------------------
  -- orRetryWithLog (performs (orRetry), i.e removal
  -- ------------------------------------------------
  
  -- | 'orRetryWithLog' should be called when the left expression of an 'orElse' evaluates to 'retry'
  --   it pops all stacks (local TVars and transaction log)
  orRetryWithLog :: TLOG -- ^ the transaction log
                 -> IO ()
  orRetryWithLog (TLOG tlog) =
    do
      lg <- readIORef tlog  -- access the Log-File
      mid <- myThreadId
      let ((la,ln,lw):xs) = tripelStack lg
      uninterruptibleMask_ $ 
        do
          undoubleLocalTVars mid (Set.elems la)    
          writeIORef tlog (lg{tripelStack=xs})
  
      
  undoubleLocalTVars mid [] = return []
  undoubleLocalTVars mid ((TVarAny (_,tvany)):xs) =
   uninterruptibleMask_ $
      do
        _tvany <- takeMVar tvany
        localmap <- takeMVar (localContent _tvany)
        case Map.lookup mid localmap of
          Just conp -> do
           (l:ltv) <- readIORef conp
           writeIORef conp ltv
           putMVar tvany _tvany
           putMVar (localContent _tvany) localmap
           undoubleLocalTVars mid xs
           
  -- *************************************************************************************           
  -- only for debugging: generate a global TVar quickly           
  newGlobalTVar content =
   do
      tvar_Id <- nextCounter
      -- Create the TVar  ...
      content_global <- newMVar content       -- set global content
      content_local   <- newMVar Map.empty     
      notify_list     <- newMVar Set.empty
      unset_lock      <- newEmptyMVar
      -- loc <- newIORef Nothing
      content_waiting_queue <- newMVar []                  -- empty broadcast list
      content_tvarx <-         newMVar (TV {globalContent = content_global,
                                    localContent  = content_local,
                                    notifyList    = notify_list,
                                    lock          = unset_lock,   
                                    waitingQueue  = content_waiting_queue
                                    -- local = loc
                                    })
      let tvany = TVarAny (tvar_Id,content_tvarx)
      let tva   = TVarA content_tvarx
      let tvar = TVar (tva,tvany)
      return tvar
  
      
  
  
              
  