使用 MVar 的并发堆栈实现

Question

我正在尝试实现一个堆栈以在并发应用程序中使用。我想要以下语义：push 永远不应该阻塞，pop 应该阻塞空堆栈上的调用线程，但仍然允许 pushes。我实现如下（底部无关位）：

data Stream a = Stream a (MVar (Stream a))
data Stack a = Stack (MVar (Int, MVar (Stream a)))

popStack :: Stack a -> IO a 
popStack (Stack stack) = do 
  (sz, mvar) <- takeMVar stack
  mbStream <- tryTakeMVar mvar 
  case mbStream of 
    Nothing -> putMVar stack (sz, mvar) >> popStack (Stack stack)
    Just (Stream x xs) -> putMVar stack (sz-1, xs) >> return x

如果流MVar 为空，我必须释放堆栈上的锁并重试。然而，这看起来像是一个杂物：如果一个线程在一个空堆栈上调用pop，它可能会在被挂起之前循环几次，即使在该线程正在执行时MVar 不会变满。有没有更好的方法利用MVars 来编写具有所需语义的pop？

---

import Control.Concurrent.MVar 
import Control.Monad 
import Control.Concurrent
import Text.Printf

newStack :: IO (Stack a) 
newStack = do 
  stream <- newEmptyMVar 
  Stack <$> newMVar (0, stream)

pushStack :: Stack a -> a -> IO ()
pushStack (Stack stack) val = do 
  (sz, stream) <- takeMVar stack
  stream' <- newMVar (Stream val stream)
  putMVar stack (sz+1, stream')

test = do 
  s <- newStack
  _ <- forkIO $ mapM_ (\a -> printf "pushing %c... " a >> pushStack s a >> threadDelay 100000) ['a' .. 'z']
  _ <- forkIO $ do 
         replicateM 13 (popStack s) >>= printf "\npopped 13 elems: %s\n"
         replicateM 13 (popStack s) >>= printf "\npopped 13 elems: %s\n"
  threadDelay (5*10^6)
  putStrLn "Done"

Answer 1

这不是很令人兴奋，但最简单的解决方案是使用 STM（如果您使用的是 cabal，则需要依赖项列表中的 stm 包）。

import Control.Concurrent.STM

newtype Stack a = Stack (TVar [a])

new :: STM (Stack a)
new = fmap Stack $ newTVar []

put :: a -> Stack a -> STM ()
put a (Stack v) = modifyTVar' v (a:)

get :: Stack a -> STM a
get (Stack v) = do
    stack <- readTVar v
    case stack of
         [] -> retry
         (a:as) -> do writeTVar v as
                      return a

您可以使用retry 获得您想要的阻塞行为，它的实现方式是在TVar 更改为[] 以外的其他内容之前不会唤醒线程。这也很好，因为您现在可以在更大的组合原子事务中使用您的堆栈，并且您不必担心确保异常不会破坏您的结构。

如果您尝试通过大量线程争用读取和/或写入来实现高性能并发，您可能会发现这样做不够聪明。在这种情况下，您可能会很高兴设计一个基于来自 atomic-primops 的基于 fetch-and-add 的计数器的结构，或者查看 hackage 上还有什么可用的。

Answer 2

快速评论：

1. “推送不应该阻塞”不是您要实际实现的目标。尽管您可能对“块”有一个不同于 GHC 含义的个人定义。例如，您的 pushStack 确实会阻塞。
2. 空堆栈上的popStack 进入一个非常繁忙的循环，反复获取和放置堆栈MVar。你不想这样做，甚至说“pop 应该阻止”。
3. 您使用 takeMVar 和 putMVar 而不是 withMVar 或 modifyMVar。这意味着您没有考虑异常，并且堆栈在通用库中不会很好。

所以您已经了解了 MVar，并且您正在使用它们来了解更多信息。

这里的 StackData 是一个有数据的堆栈（Full）或没有数据的堆栈（Empty）。当 Empty 时，有一个初始为空的 MVar 供饥饿的 poppers 等待。

type Lock = MVar ()
type Some a = (a, [a]) -- non empty version of list
data StackData a = Full !(Some a)
                 | Empty !Lock
data Stack a = Stack { stack :: MVar (StackData a) }

pop s = do
    x <- modifyMVar (stack s) $ \ sd ->
           case sd of
               Empty lock -> do
                   return (Empty lock, Left lock)
               Full (a, []) -> do
                   lock <- newEmptyMVar
                   return (Empty lock, Right a)
               Full (a, (b:bs)) -> return (Full (b, bs), Right a)
    case x of
        Left lock -> do
            withMVar lock return  -- wait on next pusher
            pop s
        Right a -> return a


 push s a = modifyMVar_ (stack s) $ \ sd ->
           case sd of
               Empty lock -> do
                   tryPutMVar lock () -- should succeed, releases waiting poppers
                   evaluate Full (a,[]) -- do not accumulate lazy thunks
               Full (b, bs) -> do
                   xs <- evaluate (b:bs) -- do not accumulate lazy thunks
                   evaluate (Full (a, xs)) -- do not accumulate lazy thunks

注意：我没有尝试编译这个。

编辑：更安全的 push 版本只需要在成功将堆栈从 Empty 修改为 Full 时将 () 放入锁中。这种确定性可以通过“掩码”操作来实现。 'restore' 在 'modifyMVar' 中使用，但不是必需的：

push s a = mask $ \restore -> do
    mLock <- modifyMVar (stack s) $ \ sd -> restore $
           case sd of
               Empty lock -> do
                   n <- evaluate Full (a,[]) -- do not accumulate lazy thunks
                   return (n, Just lock)
               Full (b, bs) -> do
                   xs <- evaluate (b:bs) -- do not accumulate lazy thunks
                   n <- evaluate (Full (a, xs))
                   return (n, Nothing)
    whenJust mLock $ \ lock -> tryPutMVar lock ()