Parallel stack implementation using MVar

Question

I am trying to implement a stack for use in a parallel application. I need the following semantics: pushit should never block, but popshould block the calling thread on an empty stack, but still allow pushes. I implemented it as follows (non-essential bits below):

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

If the thread is MVarempty, I have to release the lock on the stack and try again. However, this is similar to kludge: if a thread calls popon an empty stack, it can loop several times before pausing, even if MVarit is not full during the execution of this thread. Is there a better way to use MVarfor recording popwith the right semantics?

---

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

:

• "push " - , . "", GHC. , pushStack .
• popStack , Stack MVar. , , " ".
• takeMVar putMVar withMVar modifyMVar. , , .

, MVars, , .

StackData - (), (). , MVar , .

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

. .

EDIT: push () , . "". "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 ()

Answer 2

, 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 , []. , , , .

concurrency , / , , . , fetch-and-add atomic-primops, , hackage.