Конкурентность и STM

В уроке 34 мы закончили на обещании: отмена потока это асинхронное исключение, а значит исключения и конкурентность это одна тема, разрезанная на два урока. Сегодня вторая половина. У Haskell одна из самых приятных моделей конкурентности среди всех языков: зелёные потоки настолько дешёвы, что их порождают миллионами, а транзакционная память убирает ручные локи и связанный с ними класс багов. Пройдём стек снизу вверх: от forkIO и общей памяти без защиты через MVar к STM, разберём, что именно делает atomically под капотом, поднимемся до структурной конкурентности через async и ki, а в конце честно сравним эту модель с каналами Go и акторами Erlang, потому что у каждой свои сильные стороны и своя цена. bracket, mask и throwTo из прошлого урока уже в руках, они нам понадобятся.

Идея: зелёные потоки

Начнём с того, что делает всю тему возможной. Поток в Haskell это не поток операционной системы. Когда ты пишешь forkIO action, рантайм GHC создаёт зелёный поток: лёгкую сущность, которой управляет он сам, а не ядро. Стоит такой поток около килобайта кучи, а не мегабайта стека, как нативный, и создаётся за микросекунды. Поэтому миллион потоков это нормально, а не катастрофа: классический тест Саймона Марлоу из книги Parallel and Concurrent Programming in Haskell запускает их именно столько.

Как же миллион потоков исполняется на четырёхъядерной машине? Рантайм мультиплексирует их по схеме M:N: множество зелёных потоков на небольшое число настоящих. Единицей исполнения у GHC служит capability, виртуальный процессор; их количество задаёт флаг RTS -N. Чтобы это вообще работало параллельно, программу собирают с флагом -threaded, иначе весь процесс крутится на одном ОС-потоке.

import Control.Concurrent (forkIO, threadDelay, getNumCapabilities)

main :: IO ()
main = do
  caps <- getNumCapabilities      -- сколько ядер реально работает (флаг -N)
  putStrLn ("capabilities: " <> show caps)
  mapM_ (\i -> forkIO (worker i)) [1 .. 100000]   -- сто тысяч потоков, это дёшево
  threadDelay 1000000

Дальше три детали, важные на практике.

Переключение происходит на аллокации. Зелёный поток вытесняется не по таймеру ядра, а в точке, где он выделяет память: рантайм там проверяет флаг и при необходимости отдаёт capability другому потоку. Отсюда ловушка: тугой цикл, который ничего не аллоцирует (например, считает в аккумуляторе без выделения), может надолго захватить capability и не дать соседям работать. Случается это редко, но если поток вдруг “завис”, первый подозреваемый это неаллоцирующий цикл.

Потоки мигрируют между ядрами. Простаивающая capability подтягивает работу с чужих очередей, это балансировка нагрузки. Когда нужно прибить поток к конкретному ядру (например, ради локальности кэша), есть forkOn n, который привязывает поток к capability n на всю жизнь.

Связанные потоки для FFI. Обычный forkIO-поток делает иностранные вызовы с произвольного ОС-потока. Это ломает библиотеки с thread-local состоянием (OpenGL, некоторые C-библиотеки). Для них есть связанный поток forkOS: ему выделен персональный ОС-поток. Платишь дороговизной, поэтому берёшь только когда библиотека требует.

Если из соседних разделов ты помнишь модель GMP планировщика Go (горутина, ОС-поток, логический процессор), то у GHC всё устроено родственно: capability это аналог Go-шного P, миграция это тот же work stealing. И там, и там идея одна: дешёвый поток как единица абстракции, а не дорогой ресурс ОС. Это же роднит модель с событийным циклом из урока про асинхронность, только здесь конкурентность выглядит как обычный последовательный код, без callback и await.

forkIO и общая память

Самый низкий уровень это forkIO плюс общая изменяемая ячейка. Он же самый опасный, и именно поэтому его редко используют напрямую. Покажем почему.

import Control.Concurrent
import Data.IORef

main :: IO ()
main = do
  counter <- newIORef (0 :: Int)
  let bump = modifyIORef counter (+ 1)
  mapM_ (\_ -> forkIO (mapM_ (const bump) [1 .. 1000])) [1 .. 100]
  threadDelay 1000000
  readIORef counter >>= print   -- ожидаем 100000, получим меньше

Сто потоков по тысяче инкрементов каждый, итог должен быть 100000. Но modifyIORef это не атомарная операция: прочитать, прибавить, записать это три шага, и между ними может вклиниться другой поток. Два потока читают одно значение, оба прибавляют единицу, оба записывают, один инкремент потерян. Это гонка данных в чистом виде. Чистый код Haskell от гонок защищён неизменяемостью, но как только мы вышли в IO с разделяемой ячейкой, мы вернулись в общий для всех языков мир состояний гонки.

Для одной ячейки лекарство: атомарный read-modify-write.

import Data.IORef (atomicModifyIORef')

let bump = atomicModifyIORef' counter (\n -> (n + 1, ()))   -- атомарно и строго

atomicModifyIORef' делает обновление одной ячейки неделимым. Штрих важен: он форсирует новое значение, иначе в ячейке копятся thunk и получается утечка памяти ровно в духе урока 32. Но как только согласованно менять надо две ячейки сразу, atomicModifyIORef' бессилен: атомарность одной ячейки не растягивается на несколько. Здесь и начинается восхождение к MVar и STM.

И вторая ловушка forkIO, про которую мы уже говорили в уроке 34: дочерний поток глотает исключения молча. Если внутри forkIO что-то падает, рантайм печатает ошибку в stderr, поток тихо умирает, а родитель не узнаёт ничего. Поэтому голый forkIO почти никогда не используют для важной работы: для неё есть async, к которому мы придём.

MVar, Chan и честная блокировка

Следующий уровень это MVar: ячейка, которая бывает пустой или полной.

newMVar  :: a -> IO (MVar a)   -- создать полной
newEmptyMVar :: IO (MVar a)    -- создать пустой
takeMVar :: MVar a -> IO a     -- взять; если пусто, заблокироваться
putMVar  :: MVar a -> a -> IO ()  -- положить; если полно, заблокироваться

MVar это швейцарский нож конкурентности, он играет три роли. Как замок: держишь значение внутри, takeMVar берёт его и закрывает доступ, putMVar возвращает. Как одноместный канал: один поток кладёт, другой забирает, получается рандеву. Как общее состояние: modifyMVar читает, меняет, кладёт обратно под защитой.

-- MVar как замок вокруг общего состояния
withLockedCounter :: MVar Int -> IO ()
withLockedCounter box = modifyMVar_ box (\n -> pure (n + 1))   -- атомарно

У MVar есть ценное свойство честности, которое стоит знать. Когда ячейка освобождается, GHC будит ровно один ожидающий поток (никакого thundering herd, когда просыпаются все и дерутся), и будит в порядке очереди, FIFO. Гарантия из документации: ни один поток не заблокируется на MVar навсегда, если только другой поток не держит её навсегда. Это делает MVar-абстракции справедливыми.

Для очередей сообщений поверх MVar построен Chan: неограниченный канал, в который один поток пишет, другой читает, с поддержкой широковещания через dupChan.

import Control.Concurrent.Chan

main = do
  ch <- newChan
  forkIO (mapM_ (writeChan ch) [1 .. 5 :: Int])   -- производитель
  replicateM_ 5 (readChan ch >>= print)            -- потребитель

И тонкость, которая связывает тему с прошлым уроком: рантайм умеет ловить тупик. Если поток навсегда заблокирован на пустой MVar, до которой больше никто не дотянется, GHC бросает ему BlockedIndefinitelyOnMVar. Работает это через сборщик мусора: если ячейка с ожидающим потоком недостижима от живых корней, разбудить поток уже физически некому, и рантайм будит его исключением. Аналогичный BlockedIndefinitelyOnSTM есть и для транзакций. Это диагностика, а не гарантия (GC может не успеть заметить тупик), но в проде она не раз спасала от тихо зависшего рабочего потока.

MVar мощнее IORef, но у него своя слабость, та же, что у локов в любом языке: согласовать несколько MVar сразу можно только захватывая их по очереди, а это прямой путь к дедлоку при неправильном порядке захвата. Чтобы атомарно работать с несколькими ячейками без локов и без порядка захвата, нужен качественно другой инструмент.