Concurrency

concurrently and mapConcurrently

Pure functions can be evaluated in any order, so the compiler is free to parallelise them. IO actions are ordered by the program, and the async package is the shortest way to say that two of them may run at the same time:

import Control.Concurrent.Async main :: IO () main = do (a, b) <- concurrently (readFile "a.txt") (readFile "b.txt") putStrLn (show (length a + length b))

concurrently starts both actions, waits for both, and returns the pair of results. If one of them throws, the other is cancelled and the exception is rethrown in the parent, so a failure never leaves a thread running in the background.

-- one action per element, all at once sizes <- mapConcurrently fileSize ["a.txt", "b.txt", "c.txt"] -- keep results in order, discard them forConcurrently_ urls fetch

mapConcurrently keeps the order of the input in the output even though the work finishes out of order. The number of threads is bounded by the runtime, not by the length of the list.

race and timeouts

Two actions where the first answer wins is a race. The loser is cancelled as soon as the winner finishes, which is how a timeout is built:

race (readFile "big.txt") (threadDelay 2000000 >> pure "") -- Left contents, or Right "" after two seconds

The result is an Either, with the left side for the first action and the right side for the second. A timeout is the same shape with a sleep on one side:

timeout :: Int -> IO a -> IO (Maybe a) slow <- timeout 2000000 (readFile "big.txt") case slow of Nothing -> putStrLn "gave up" Just t -> putStrLn (show (length t))

timeout comes from System.Timeout and takes microseconds. It throws an asynchronous exception into the action, so the action must be interruptible, which IO actions generally are.

Software transactional memory

Shared mutable state with locks is where concurrency bugs live. STM replaces locks with transactions: a block of reads and writes either commits as a whole or is retried from the start:

import Control.Concurrent.STM main = do counter <- newTVarIO 0 replicateM_ 1000 (atomically (modifyTVar' counter (+1))) n <- atomically (readTVar counter) print n

TVar is the shared cell, and atomically runs a transaction. The block has the type STM a, which is a monad, so do notation works as usual.

-- transfer: both sides change, or neither does transfer from to amount = atomically (do a <- readTVar from writeTVar from (a - amount) b <- readTVar to writeTVar to (b + amount))

Two threads running transfer at once cannot observe a half-finished state. There is no lock to forget and no ordering to get wrong, which is the whole argument for STM.

retry and blocking

A transaction that cannot proceed calls retry. The runtime parks the thread until one of the variables it read changes, then runs the block again:

takeItem :: TVar [a] -> STM a takeItem queue = do xs <- readTVar queue case xs of [] -> retry (x:rest) -> do writeTVar queue rest pure x

The empty case waits instead of failing, and the wait costs no CPU because the thread is asleep, not spinning. A producer only has to atomically write to the same TVar to wake it.

-- producer atomically (modifyTVar' queue (++ [newItem])) -- consumer, blocks until an item exists item <- atomically (takeItem queue)

That pair is a bounded-free queue in six lines. For production work use TQueue or TBQueue from the same module, which have the same interface and better performance.

Next: Parsing with Megaparsec

Article author: Arthur Isaev