haskell - 使用 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 不会变满，而该线程正在执行。有没有更好的方法利用 MVar 来编写具有所需语义的 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更改为 [] 以外的内容。这也很好，因为您现在可以在更大的组合原子事务中使用堆栈，并且不必担心确保异常不会破坏您的结构。

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