主页 / coding / 问题 / 79591505
Accepted
Enlico
Asked: 2025-04-25 05:49:56 +0800 CST

在 STM 的背景下，事务日志在概念上是什么样的，以及当事务在几次重试后成功时它会如何演变？

  • 772

例如，考虑这个函数，它可以在 WM 中使用，允许在给定的显示器上将窗口从一个桌面移动到另一个桌面，

moveWindowSTM :: Display -> Window -> Desktop -> Desktop -> STM ()
moveWindowSTM disp win a b = do
  wa <- readTVar ma
  wb <- readTVar mb
  writeTVar ma (Set.delete win wa)
  writeTVar mb (Set.insert win wb)
 where
  ma = disp ! a
  mb = disp ! b

显然还有它的IO包装，

moveWindow :: Display -> Window -> Desktop -> Desktop -> IO ()
moveWindow disp win a b = atomically $ moveWindowSTM disp win a b

然后假设

  • 我们的交易在第三次尝试时成功了，
  • ma因为在第一次尝试时，它会因另一个并发事务提交而失效，该事务在我们的事务之后立即更改了内部的值wa <- readTVar ma，
  • 而在第二次尝试时，它被另一个并发事务无效，该事务开始提交并更改了我们的事务中的一个或ma两个的内容。mbwriteTVar ma (Set.delete win wa)

在这种情况下，交易日志将如何演变，以及验证在什么时候会失败？

Simon Marlow 所著《Haskell 中的并行和并发编程》的相关摘录如下（但 Simon Peyton Jones 所著的论文《美丽的并发性》中也提供了类似的信息）：

STM 事务的工作原理是累积事务期间迄今为止发生的操作的日志。日志有三种readTVar用途：writeTVar

  • 通过将writeTVar操作存储在日志中而不是立即将其应用到主内存中，丢弃事务的影响很容易；我们只需丢弃日志即可。因此，中止事务的代价很小且固定。

  • 每个人都readTVar必须遍历日志来检查 是否是TVar由之前的 写入的writeTVar。因此，readTVar是日志长度上的 O(n) 操作。

  • 因为日志包含所有操作的记录readTVar，所以可以使用它来发现事务期间的完整TVars读取集，我们需要知道这些才能实现retry。

当事务结束时，STM 实现会将日志与内存内容进行比较。如果当前内存内容与 读取的值匹配readTVar，则将事务的结果提交到内存；如果不匹配，则丢弃日志并从头开始重新运行事务。此过程以原子方式进行，方法是在TVars事务执行期间锁定所有涉及的 。GHC 中的 STM 实现不使用全局锁；TVars提交期间仅锁定涉及事务的 ，因此对不相交的 集合进行操作的事务TVars可以不受干扰地继续进行。

haskell

2 个回答

  1. STM 事务运行完整，将每个事件记录到日志中，最后检查一致性以决定是否提交事务。

    因此，对于您的示例交易

    do
  wa <- readTVar ma
  wb <- readTVar mb
  writeTVar ma (Set.delete win wa)
  writeTVar mb (Set.insert win wb)

    每次尝试都运行readTVar一次writeTVar。这些函数会写入日志，记录每个事件是读取还是写入、受影响的 TVar 以及读取或写入的值。

    READ ma a   -- value a is read from ma (note that values are just pointers as far as STM is concerned)
READ mb b   -- value b is read from mb
WRITE ma c  -- value c is written to ma
WRITE mb d  -- value d is written to mb

    在该事务期间，其他线程可能会提交写入相同 TVars 的事务。无论如何，当前事务仍将继续进行。

    一旦atomically完成运行事务，它会锁定其日志中出现的所有 TVars，以确保接下来的操作以原子方式发生：

    • atomically检查日志中读取的值是否与 TVars 中的当前值匹配，这意味着它们没有被覆盖（或者其他事务将相同的值写入 TVar，这很好）。

    • 如果日志与 TVar 匹配，则写入事件实际上会应用于其 TVar。这就是“提交”事务的含义。锁被释放并atomically返回。

    • 否则，日志和 TVar 的内容不匹配；日志将被丢弃，锁将被释放，并atomically从头开始重新启动。

    在您的示例中，如果有三次尝试，则意味着创建了三个类似上述的完整日志。前两次日志的唯一区别在于，当atomically回顾这些日志时，它无法假装它们是原子发生的，因此事务会重新启动。

    • 2
  2. Best Answer

    我认为这样的事情应该可以作为一个相当不错的心理模型：

    data Generational a = Generational
    { generation :: Int
    , value :: a
    }
instance Eq (Generational a) where (==) = (==) `on` generation

-- choose a generation number that's never been chosen before
-- e.g. by starting at minBound and incrementing one at each call
freshGeneration :: IO Int

newtype TVar a = TVar (IORef (Generational a))

unsafeReadTVar :: MonadIO m => TVar a -> m (Generational a)
unsafeReadTVar (TVar ref) = liftIO $ readIORef ref

unsafeWriteTVar :: Int -> TVar a -> Identity a -> IO ()
unsafeWriteTVar g (TVar ref) (Identity a) = writeIORef ref (Generational g a)

lock, unlock :: TVar a -> IO ()

-- like Map, but with an extra type argument
data Map1 (key :: k -> *) (value :: k -> *)

data STMLog = STMLog
    { reads :: Map1 TVar Generational
    , writes :: Map1 TVar Identity
    }

emptyLog :: STMLog
emptyLog = STMLog empty1 empty1

-- throw away the generation
readsId :: STMLog -> Map1 TVar Identity
readsId = fmap1 (Identity . value) . reads

newtype STM a = STM (StateT STMLog IO a)
    deriving (Functor, Applicative, Monad)

readTVar :: TVar a -> STM a
readTVar v = STM do
   log <- get
   case lookup1 v (writes log) <|> lookup1 v (readsId log) of
        Just (Identity a) -> pure a
        Nothing -> do
            ga <- unsafeReadTVar v
            put log { reads = insert1 v ga (reads log) }
            pure (value ga)

writeTVar :: TVar a -> a -> STM ()
writeTVar v a = STM $ modify \log ->
    log { writes = insert1 v (Identity a) (writes log) }

atomically :: STM a -> IO a
atomically (STM act) = do
    (a, log) <- runStateT act emptyLog
    let allVars = keys1 $ union1 (readsId log) (writes log)
    traverse1_ lock allVars
    reads' <- traverseWithKey1 (\v _ -> unsafeReadTVar v) (reads log)
    -- N.B. (==) only compares generations here
    -- (the real implementation probably uses some form
    -- of pointer equality instead of storing generations)
    if reads log == reads'
        then do
            g <- freshGeneration
            traverseWithKey1_ (unsafeWriteTVar g) (writes log)
            traverse1_ unlock allVars
            pure a
        else do
            traverse1_ unlock allVars
            atomically (STM act) -- try again

    实际情况可能更加复杂/优化/健壮。我这里根本不考虑处理异常，我假设一次性锁定一组变量很容易，我没有展示具体的实现retry，我也没有考虑是否需要任何内存隔离等等。

    • 2

相关问题