ℹ️ Select 'Choose Exercise', or randomize 'Next Random Exercise' in selected language.

Choose Exercise:
Timer 00:00
WPM --
Score --
Acc --
Correct chars --

Transactional Queue with STM

Haskell

Goal -- WPM

Ready
Exercise Algorithm Area
1module STMQueue where
2
3import Control.Concurrent.STM
4import Control.Monad (replicateM, forever)
5import Control.Concurrent (forkIO, threadDelay)
6import System.Random (randomRIO)
7
8-- | A thread-safe queue implemented using STM.
9-- It uses a TVar holding a list of elements.
10newtype TQueue a = TQueue (TVar [a])
11
12-- | Creates a new empty transactional queue.
13newTQueue :: STM (TQueue a)
14newTQueue = do
15tvar <- newTVar []
16return $ TQueue tvar
17
18-- | Enqueues an element into the transactional queue.
19-- This operation is atomic.
20enqueue :: TQueue a -> a -> STM ()
21enqueue (TQueue tvar) x = do
22list <- readTVar tvar
23writeTVar tvar (list ++ [x]) -- Append to the end
24
25-- | Dequeues an element from the transactional queue.
26-- If the queue is empty, it retries the transaction.
27-- This operation is atomic.
28dequeue :: TQueue a -> STM a
29dequeue (TQueue tvar) = do
30list <- readTVar tvar
31case list of
32[] -> retry -- Queue is empty, retry transaction
33(x:xs) -> do
34writeTVar tvar xs
35return x
36
37-- | Checks if the transactional queue is empty.
38isEmptyTQueue :: TQueue a -> STM Bool
39isEmptyTQueue (TQueue tvar) = do
40list <- readTVar tvar
41return (null list)
42
43-- | Producer thread function.
44producer :: TQueue Int -> IO ()
45producer q = forever $ do
46val <- randomRIO (1, 100 :: Int)
47atomically $ enqueue q val
48putStrLn $ "Produced: " ++ show val
49delay <- randomRIO (100000, 500000) -- 0.1 to 0.5 seconds
50threadDelay delay
51
52-- | Consumer thread function.
53consumer :: Int -> TQueue Int -> IO ()
54consumer cid q = forever $ do
55val <- atomically $ dequeue q
56putStrLn $ "Consumer " ++ show cid ++ " consumed: " ++ show val
57delay <- randomRIO (200000, 700000) -- 0.2 to 0.7 seconds
58threadDelay delay
59
60-- | Main function to demonstrate the transactional queue.
61main :: IO ()
62main = do
63putStrLn "Starting STM Queue Demo..."
64q <- atomically newTQueue
65
66-- Start producer thread
67_ <- forkIO $ producer q
68
69-- Start multiple consumer threads
70replicateM 3 $ forkIO $ consumer 1 q -- Example with 3 consumers
71
72-- Keep the main thread alive for a while
73threadDelay (5 * 1000000) -- Run for 5 seconds
74putStrLn "STM Queue Demo finished."
Algorithm description viewbox

Transactional Queue with STM

Algorithm description:

This Haskell code demonstrates a thread-safe queue using Software Transactional Memory (STM). The `TQueue` is implemented using a `TVar` holding a list. `enqueue` and `dequeue` operations are performed within `STM` transactions, ensuring atomicity and consistency. `dequeue` uses `retry` to block if the queue is empty, automatically retrying when the queue becomes non-empty. This pattern is crucial for building concurrent applications where shared mutable state needs to be managed safely, such as in message queues or producer-consumer scenarios.

Algorithm explanation:

The `TQueue` is a wrapper around a `TVar [a]`, where `TVar` is a transactional variable. `newTQueue` creates a new `TVar` initialized with an empty list within an `STM` transaction. `enqueue` reads the current list from the `TVar`, appends the new element, and writes the updated list back. This is all done atomically within `STM`. `dequeue` reads the list; if it's empty, `retry` is called. `retry` aborts the current transaction and blocks the thread until one of the `TVar`s read in the transaction is modified, at which point the transaction is retried. If the list is not empty, it takes the head element, writes the tail back to the `TVar`, and returns the head. `isEmptyTQueue` simply checks if the list inside the `TVar` is empty. The `producer` and `consumer` functions demonstrate concurrent access to the queue. The `producer` adds random numbers, and `consumer`s remove them. The `atomically` function executes an `STM` action as a single, atomic transaction. Time complexity for `enqueue` and `dequeue` is O(1) on average for list operations (though append is O(N) for standard lists, a `Seq` would be better for true O(1)). Space complexity is O(N) where N is the number of elements in the queue.

Pseudocode:

define TQueue as a TVar holding a list of elements

function newTQueue():
  create a new TVar initialized with an empty list
  return TQueue containing the TVar

function enqueue(queue, element):
  read the list from the queue's TVar
  append the element to the list
  write the new list back to the TVar

function dequeue(queue):
  read the list from the queue's TVar
  if the list is empty:
    retry the transaction
  else:
    take the first element (head) and the rest of the list (tail)
    write the tail back to the TVar
    return the head element

function isEmptyTQueue(queue):
  read the list from the queue's TVar
  return true if the list is empty, false otherwise

producer thread:
  loop forever:
    generate a random value
    atomically enqueue the value into the queue
    print "Produced: value"
    wait for a random delay

consumer thread:
  loop forever:
    atomically dequeue an element from the queue
    print "Consumer X consumed: value"
    wait for a random delay

main:
  create a new TQueue
  fork a producer thread
  fork multiple consumer threads
  wait for a duration