[Haskell-cafe] Barrier implementation

[Peter Eriksen]

Hi again,

Now I've actually tested the barrier implementation by counting the
number of times each worker thread reaches the barrier.  It's not a
proof, but I take it as strong indication, that it's not as bad, as I
first thought.  If all workers have run the same number of times 
(that is a maximum of one apart), then at least that's one good 
feature of the barrier.  I think it works though and also keeps that
invariant (max one iteration apart) all the time.

Thank you for your kind help.

Regards,

Peter

****************
Here are the counts for runs with the barrier in different places:

=============
No barrier: 

> worker :: Int -> TVar Int -> TVar Int -> IO ()
> worker id tv ic = do
>       sleepingTime <- randomRIO (0, 50000)
>       threadDelay sleepingTime
>       putStr $ show id
>	atomically $ (inc ic)
>       worker id tv ic

(0,274)
(1,272)
(2,274)
(3,278)
(4,269)
(5,287)
(6,287)
(7,275)
(8,281)
(9,274)

================
The barrier after putStr:

> worker :: Int -> TVar Int -> TVar Int -> IO ()
> worker id tv ic = do
>       sleepingTime <- randomRIO (0, 50000)
>       threadDelay sleepingTime
>       putStr $ show id
>	atomically $ barrier tv id 
>	atomically $ (inc ic)
>       worker id tv ic

(0,199)
(1,199)
(2,199)
(3,199)
(4,198)
(5,198)
(6,198)
(7,198)
(8,198)
(9,198)

===============
The thread between threadDelay and putStr:

> worker :: Int -> TVar Int -> TVar Int -> IO ()
> worker id tv ic = do
>       sleepingTime <- randomRIO (0, 50000)
>       threadDelay sleepingTime
>	atomically $ barrier tv id 
>       putStr $ show id
>	atomically $ (inc ic)
>       worker id tv ic

(0,202)
(1,201)
(2,201)
(3,201)
(4,201)
(5,201)
(6,201)
(7,201)
(8,201)
(9,201)

Note: This is the one looking most like 0123456789012345... as I
initially wanted, but of course there is a chance of a race where
all worker threads wait before putStr after they are in sequence
from the barrier.  Then it would be random which one executed putStr
first.

===============
The barrier is placed in the beginning before threadDelay:

> worker :: Int -> TVar Int -> TVar Int -> IO ()
> worker id tv ic = do
>       sleepingTime <- randomRIO (0, 50000)
>	atomically $ barrier tv id 
>       threadDelay sleepingTime
>       putStr $ show id
>	atomically $ (inc ic)
>       worker id tv ic

(0,200)
(1,200)
(2,200)
(3,200)
(4,200)
(5,199)
(6,200)
(7,200)
(8,199)
(9,199)

=======================
=======================
Here's the full program:

> module Main where
>
> import Control.Concurrent
> import Control.Concurrent.STM
> import System.Random
>
> worker :: Int -> TVar Int -> TVar Int -> IO ()
> worker id tv ic = do
>       sleepingTime <- randomRIO (0, 50000)
>       threadDelay sleepingTime
>       putStr $ show id
>	atomically $ barrier tv id 
>	atomically $ (inc ic)
>       worker id tv ic

Each worker sleeps for some time, then outputs its id and waits at
the barrier for all the other workers to finish their sleep+output.

> barrier :: TVar Int -> Int -> STM ()
> barrier tv id = do
>       passed <- readTVar tv
>       if (passed `mod` 10 == id)
>               then writeTVar tv (passed+1)
>               else retry

The barrier is simply a global variable, tv, which holds the number of
times any worker passed the barrier.  Now, a worker may only pass the
barrier iff the worker with an id one less just passed, or else it
should block.

> main :: IO ()
> main = do
>       tv <- atomically $ newTVar 0
>	idCounts <- mapM (atomically . newTVar) [0,0,0,0,0,0,0,0,0,0]
>       for [0..9] $ \i -> forkIO $ worker i tv (idCounts!!i)
>       threadDelay (10*10^6)
>	mapM_ (\(i,ic) -> (atomically $ readTVar ic) >>= \n -> print
>	(i,n)) (zip [0..9]  idCounts)
>
> for = flip mapM_
> 
> inc tvar = readTVar tvar >>= \n -> writeTVar tvar (n+1)