Concurrency | Haskell - Wyatt's Notes

-- Concurrency: interleaved execution
-- Task A: [----]
-- Task B:   [----]

-- Parallelism: simultaneous execution
-- Core 1: [----]
-- Core 2: [----]

import Control.Concurrent (forkIO, threadDelay)

main :: IO ()
main = do
  putStrLn "Starting..."

  _ <- forkIO $ do
    threadDelay 1000000  -- 1 second (in microseconds)
    putStrLn "Thread 1 done"

  _ <- forkIO $ do
    threadDelay 500000   -- 0.5 seconds
    putStrLn "Thread 2 done"

  threadDelay 2000000  -- wait for both threads
  putStrLn "All done"

-- Compile with threading support
-- ghc -threaded -with-rtsopts=-N4 MyProgram.hs
-- -N4 means use 4 OS threads (one per CPU core)

-- Creating many threads
spawnMany :: Int -> IO ()
spawnMany n = do
  mapM_ (\i -> forkIO (worker i)) [1..n]
  threadDelay 1000000
  where
    worker i = do
      threadDelay 100000
      putStrLn ("Worker " ++ show i)

import Control.Concurrent.MVar

-- newEmptyMVar :: IO (MVar a)
-- newMVar :: a -> IO (MVar a)
-- takeMVar :: MVar a -> IO a      -- blocks if empty
-- putMVar :: MVar a -> a -> IO () -- blocks if full
-- tryTakeMVar :: MVar a -> IO (Maybe a)
-- tryPutMVar :: MVar a -> a -> IO Bool

-- Simple counter with MVar
counterExample :: IO ()
counterExample = do
  counter <- newMVar 0

  _ <- forkIO $ do
    forM_ [1..100] $ \_ -> do
      val <- takeMVar counter
      putMVar counter (val + 1)

  _ <- forkIO $ do
    forM_ [1..100] $ \_ -> do
      val <- takeMVar counter
      putMVar counter (val + 1)

  threadDelay 1000000
  result <- readMVar counter
  -- This may NOT be 200 due to interleaving
  -- (takeMVar + putMVar is atomic, but the gap
  -- between them is vulnerable)
  putStrLn ("Counter: " ++ show result)

-- Producer-consumer with MVar
channel :: IO ()
channel = do
  mvar <- newEmptyMVar

  -- Producer
  _ <- forkIO $ do
    mapM_ (\x -> putMVar mvar x) [1..10]
    putMVar mvar (-1)  -- sentinel value

  -- Consumer
  go mvar
  where
    go m = do
      val <- takeMVar m
      if val == (-1)
        then putStrLn "Done"
        else do
          putStrLn ("Received: " ++ show val)
          go m

-- Use MVar () as a mutex (binary semaphore)
type Lock = MVar ()

withLock :: Lock -> IO a -> IO a
withLock lock action = do
  takeMVar lock
  result <- action
  putMVar lock ()
  return result

-- Usage
lockedCounter :: IO ()
lockedCounter = do
  lock <- newMVar ()
  counter <- newMVar 0

  let increment = withLock lock $ do
        val <- takeMVar counter
        putMVar counter (val + 1)

  mapM_ (\_ -> forkIO increment) [1..1000]
  threadDelay 1000000
  result <- readMVar counter
  putStrLn ("Counter: " ++ show result)

import Control.Concurrent.Chan

-- newChan :: IO (Chan a)
-- writeChan :: Chan a -> a -> IO ()
-- readChan :: Chan a -> IO a       -- blocks if empty
-- dupChan :: Chan a -> IO (Chan a)  -- duplicate for multiple readers

-- Multiple producers, single consumer
workQueue :: IO ()
workQueue = do
  chan <- newChan

  -- Spawn 3 producers
  mapM_ (\i -> forkIO (producer chan i)) [1..3]

  -- Single consumer
  consumer chan

  where
    producer ch i = do
      mapM_ (\j -> writeChan ch (i, j)) [1..5]

    consumer ch = do
      forM_ [1..15] $ \_ -> do
        (i, j) <- readChan ch
        putStrLn ("Got: producer " ++ show i ++ " item " ++ show j)

-- Broadcast: one producer, multiple consumers
broadcast :: IO ()
broadcast = do
  chan <- newChan

  -- Create duplicate channels for each consumer
  ch1 <- dupChan chan
  ch2 <- dupChan chan

  -- Producer writes to the original
  _ <- forkIO $ mapM_ (writeChan chan) [1..5]

  -- Consumer 1 reads from duplicate 1
  _ <- forkIO $ do
    forM_ [1..5] $ \_ -> do
      val <- readChan ch1
      putStrLn ("Consumer 1: " ++ show val)

  -- Consumer 2 reads from duplicate 2
  _ <- forkIO $ do
    forM_ [1..5] $ \_ -> do
      val <- readChan ch2
      putStrLn ("Consumer 2: " ++ show val)

  threadDelay 1000000

import Control.Concurrent.STM

-- newTVarIO :: a -> IO (TVar a)
-- newTVar :: a -> STM (TVar a)
-- readTVar :: TVar a -> STM a
-- writeTVar :: TVar a -> a -> STM ()
-- modifyTVar :: TVar a -> (a -> a) -> STM ()

-- atomically :: STM a -> IO a
-- Executes an STM transaction atomically

-- Transfer money between two accounts
type Account = TVar Int

transfer :: Account -> Account -> Int -> STM ()
transfer from to amount = do
  fromBal <- readTVar from
  toBal   <- readTVar to
  if fromBal >= amount
    then do
      writeTVar from (fromBal - amount)
      writeTVar to (toBal + amount)
    else retry  -- retry the whole transaction

-- Lock-based approach (CAN deadlock):
-- Thread 1: lock(A); lock(B)
-- Thread 2: lock(B); lock(A)
-- If both threads hold one lock and wait for the other => deadlock

-- STM approach (NEVER deadlocks):
-- Thread 1: atomically $ do readTVar a; readTVar b; ...
-- Thread 2: atomically $ do readTVar b; readTVar a; ...
-- Both transactions see consistent snapshots
-- No ordering issues possible

twoAccountTransfer :: Account -> Account -> Account -> Int -> STM ()
twoAccountTransfer a b c amount = do
  balA <- readTVar a
  balB <- readTVar b
  balC <- readTVar c
  if balA >= amount && balB >= amount
    then do
      writeTVar a (balA - amount)
      writeTVar b (balB - amount)
      writeTVar c (balC + amount + amount)
    else retry

import Control.Concurrent.STM

-- retry: block until at least one read TVar changes
withdraw :: Account -> Int -> STM ()
withdraw acc amount = do
  bal <- readTVar acc
  if bal >= amount
    then writeTVar acc (bal - amount)
    else retry  -- wait until balance changes (deposit?)

-- orElse: try first action, if it retries, try the second
withdrawEither :: Account -> Account -> Int -> STM ()
withdrawEither acc1 acc2 amount = do
  withdraw acc1 amount
  `orElse`
  withdraw acc2 amount
  -- Tries to withdraw from acc1; if that retries (insufficient funds),
  -- tries acc2 instead; if that also retries, waits for either account
  -- to change

-- A thread-safe queue using STM
data TQueue a = TQueue (TVar [a]) (TVar [a])

newTQueue :: STM (TQueue a)
newTQueue = do
  readEnd  <- newTVar []
  writeEnd <- newTVar []
  return (TQueue readEnd writeEnd)

writeTQueue :: TQueue a -> a -> STM ()
writeTQueue (TQueue _ writeEnd) x = do
  xs <- readTVar writeEnd
  writeTVar writeEnd (x : xs)

readTQueue :: TQueue a -> STM a
readTQueue (TQueue readEnd writeEnd) = do
  xs <- readTVar readEnd
  case xs of
    [] -> do
      -- Move elements from write end to read end
      ys <- readTVar writeEnd
      case ys of
        [] -> retry
        _  -> do
          writeTVar writeEnd []
          writeTVar readEnd (reverse ys)
          readTQueue (TQueue readEnd writeEnd)
    (x:rest) -> do
      writeTVar readEnd rest
      return x

import Control.Concurrent.Async

-- async :: IO a -> IO (Async a)
-- Launch a computation asynchronously, returning a handle

-- wait :: Async a -> IO a
-- Wait for the async to complete and return its result

-- waitCatch :: Async a -> IO (Either SomeException a)
-- Wait and catch any exceptions

-- race :: IO a -> IO b -> IO (Either a b)
-- Run two IO actions concurrently, return the first to finish

-- concurrently :: IO a -> IO b -> IO (a, b)
-- Run two IO actions concurrently, return both results

-- Fetch multiple URLs concurrently
fetchUrls :: [String] -> IO [String]
fetchUrls urls = do
  results <- mapConcurrently fetchUrl urls
  return results

fetchUrl :: String -> IO String
fetchUrl url = do
  putStrLn ("Fetching: " ++ url)
  threadDelay 500000  -- simulate network delay
  return ("Response from " ++ url)

-- mapConcurrently: parallel map with bounded concurrency
-- mapConcurrently_ :: (a -> IO b) -> [a] -> IO ()
-- Runs the IO action for each element, up to N at a time (based on RTS -N)

-- race: return the result of whichever action finishes first
import Control.Concurrent.Async (race)

main :: IO ()
main = do
  result <- race
    (do threadDelay 1000000; return "slow")
    (do threadDelay 500000; return "fast")
  case result of
    Left val  -> putStrLn ("Left finished first: " ++ val)
    Right val -> putStrLn ("Right finished first: " ++ val)
  -- Output: Left finished first: fast

-- concurrently runs two IO actions in parallel and waits for both
import Control.Concurrent.Async (concurrently)

main :: IO ()
main = do
  (a, b) <- concurrently
    (do threadDelay 500000; putStrLn "Task A done"; return 1)
    (do threadDelay 300000; putStrLn "Task B done"; return 2)
  putStrLn ("Results: " ++ show a ++ ", " ++ show b)
  -- Task B done, Task A done, Results: 1, 2

import Control.Concurrent.Async (mapConcurrently)

-- Process items in parallel
processItems :: [Item] -> IO [Result]
processItems = mapConcurrently processItem

-- If you only need side effects (no return values)
-- Use mapConcurrently_
printItems :: [String] -> IO ()
printItems = mapConcurrently_ (\s -> do
  threadDelay 100000
  putStrLn s
  )

import Control.Exception (bracket, finally, onException)
import Control.Concurrent (ThreadKilled, throwTo, myThreadId)

-- throwTo :: ThreadId -> SomeException -> IO ()
-- Throws an exception to another thread

-- Common use: timeout
-- timeout :: Int -> IO a -> IO (Maybe a)
-- Runs the IO action; if it takes longer than n microseconds,
-- throws ThreadKilled and returns Nothing

-- bracket :: IO a -> (a -> IO b) -> (a -> IO c) -> IO c
-- bracket acquire cleanup action
-- 1. Runs acquire
-- 2. Runs action (the main computation)
-- 3. Runs cleanup (whether action succeeded or threw an exception)

withFile' :: FilePath -> (Handle -> IO a) -> IO a
withFile' path action = do
  handle <- openFile path ReadMode
  result <- action handle
  hClose handle
  return result
  -- BUG: if action throws an exception, hClose is never called

-- Correct with bracket:
withFileSafe :: FilePath -> (Handle -> IO a) -> IO a
withFileSafe path action =
  bracket
    (openFile path ReadMode)
    hClose
    action

-- Timeout a computation
fetchWithTimeout :: Int -> String -> IO (Maybe String)
fetchWithTimeout timeoutMs url = timeout (timeoutMs * 1000) (fetchUrl url)

import Control.Exception (mask, mask_, uninterruptibleMask)

-- mask :: ((forall a. IO a -> IO a) -> IO b) -> IO b
-- Disables async exceptions for the duration of the inner action
-- except during unmask

-- mask_ :: IO a -> IO a
-- Completely masks async exceptions

-- uninterruptibleMask :: ((forall a. IO a -> IO a) -> IO b) -> IO b
-- Cannot be interrupted at all (even during unmask)

safeUpdate :: MVar Int -> (Int -> IO Int) -> IO Int
safeUpdate mvar f = mask $ \restore -> do
  val <- takeMVar mvar
  result <- restore (f val) `onException` putMVar mvar val
  putMVar mvar (result)
  return result

import Control.Parallel.Strategies

-- rseq :: Strategy a
-- Evaluate to weak head normal form (immediate, cheap)

-- rpar :: Strategy a
-- Start evaluating in parallel (spark a new thread)

-- using :: a -> Strategy a -> a
-- Apply a strategy to a value

-- Evaluate a pair in parallel
parallelPair :: (Int, Int)
parallelPair = runEval $ do
  a <- rpar (fib 35)
  b <- rpar (fib 36)
  return (a, b)

-- using combinator
parallelSum :: [Int] -> Int
parallelSum xs = sum xs `using` parList rseq
-- parList rseq sparks each element evaluation

-- parMap: parallel version of map
parallelMap :: (a -> b) -> [a] -> [b]
parallelMap f xs = map f xs `using` parList rseq

import Control.Parallel

-- par :: a -> b -> b
-- "Spark" the evaluation of the first argument,
-- then return the second argument

-- pseq :: a -> b -> b
-- Evaluate the first argument to WHNF,
-- then return the second

-- Parallel fibonacci
fib :: Int -> Integer
fib 0 = 0
fib 1 = 1
fib n =
  let x = fib (n - 1)
      y = fib (n - 2)
  in x `par` (y `pseq` x + y)
-- x is sparked for parallel evaluation
-- y is evaluated before the addition
-- Both branches may run on different cores

import Control.Parallel.Strategies

-- Parallel map
parallelMap :: (a -> b) -> [a] -> [b]
parallelMap f xs = map f xs `using` parList rseq

-- Parallel fold (after parallel map)
parallelFoldMap :: (a -> b) -> (b -> b -> b) -> b -> [a] -> b
parallelFoldMap f g z xs =
  foldl g z (map f xs) `using` parList rseq

-- Parallel traversal of a tree
parTree :: Tree a -> Tree a
parTree (Leaf x)   = Leaf x
parTree (Branch l r) = l' `par` r' `pseq` Branch l' r'
  where
    l' = parTree l
    r' = parTree r

-- Evaluating a list of independent computations
results :: [Int]
results = map slowComputation [1..8] `using` parList rdeepseq
-- rdeepseq evaluates to normal form (fully)