[Haskell-cafe] synchronous channels in STM

[Arnar Birgisson]

On Thu, Oct 9, 2008 at 18:10, Arnar Birgisson <arnarbi at gmail.com> wrote:
>> But I did write a concurrent prime sieve with it:
>
> I did the same, with the one-place-buffers (the MVars implemented over
> STM). Be warned that there is no stop condition, this just keeps
> printing primes forever.

Please forgive me for reposting, but the last one exited quite prematurely :)


module Main where

import Control.Concurrent (forkIO)
import Control.Concurrent.STM
import System (getArgs)

-- MVars from the STM paper
type MVar a = TVar (Maybe a)

newEmptyMVar :: STM (MVar a)
newEmptyMVar = newTVar Nothing

takeMVar :: MVar a -> STM a
takeMVar mv
 = do v <- readTVar mv
      case v of
        Nothing  -> retry
        Just val -> do writeTVar mv Nothing
                       return val

putMVar :: MVar a -> a -> STM ()
putMVar mv val
 = do v <- readTVar mv
      case v of
        Nothing  -> writeTVar mv (Just val)
        Just _   -> retry

-- Sieve
forever a = do a; forever a

pfilter :: Int -> MVar Int -> MVar Int -> IO ()
pfilter p in_ out =
    forever $ do atomically $ do v <- takeMVar in_
                                 if v `mod` p /= 0
                                    then putMVar out v
                                    else return ()

sieve :: MVar Int -> MVar Int -> IO ()
sieve in_ out =
    do p <- atomically $ takeMVar in_
       atomically $ putMVar out p
       ch <- atomically $ newEmptyMVar
       forkIO $ pfilter p in_ ch
       sieve ch out

feeder :: MVar Int -> IO ()
feeder out = feed' 2
           where
             feed' i = do atomically $ putMVar out i
                          feed' (i+1)

printer :: MVar () -> MVar Int -> Int -> IO ()
printer stop in_ max =
    do v <- atomically $ takeMVar in_
       putStrLn $ show v
       if v > max then atomically $ putMVar stop ()
                  else printer stop in_ max

main :: IO ()
main = do max:_ <- getArgs
          in_ <- atomically newEmptyMVar
          out <- atomically newEmptyMVar
          stop <- atomically newEmptyMVar
          forkIO $ feeder in_
          forkIO $ printer stop out (read max)
          forkIO $ sieve in_ out
          atomically $ takeMVar stop
          return ()