Кооперативные потоки с нуля в 33 линии на Хаскеле

Тип

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

• Потоки должны расширять существующие последовательности инструкций
• Потоки должны поддерживать набор операций: разветвление, передача контроля и завершение
• Потоки должны разрешать различные типы планировщиков

Теперь мы переводим эти понятия в Хаскель:

• Когда вы слышите «несколько интерпретаторов / планировщики / бэкэнды» вы должны думать «свободный» (как в «свободном объекте»)
• Когда вы слышите «последовательность команд» вы должны думать: «монады».
• Когда вы хотите что-то «расширить» Вы должны думать: «трансформеры».

Объедините эти слова вместе, и вы получите правильное математическое решение: «свободный монадный трансформер».

Синтаксическое дерево

«Свободный монадный трансформер» — причудливое название для математического абстрактного синтаксического дерева, где последовательность играет важную роль. Мы обеспечиваем его набором инструкций и он строит нам синтаксическое дерево из этих инструкций.

Мы сказали, что мы хотим, чтобы наш поток мог или разветвляться, или передавать контроль, или прекращаться, так что давайте сделаем тип данных с вилками, отдачами, и завершением:

{-# LANGUAGE DeriveFunctor #-}  data ThreadF next = Fork  next next                   | Yield next                   | Done                   deriving (Functor) 

ThreadF представляет наш набор инструкций. Мы хотели добавить три новых инструкции, поэтому ThreadF имеет три конструктора, по одному для каждой команды: Fork, Yield, и Done.

Наш тип ThreadF представляет один узел в синтаксическом дереве. next поля из конструкторов представляют, куда дети узлов должны идти. Fork создает два пути исполнения, поэтому он имеет двоих детей. Done завершает текущий путь выполнения, поэтому он не имеет детей. Yield ни ветвится, ни прекращается, поэтому он имеет одного ребенка. deriving (Functor) часть просто говорит свободному монадному трансформеру, что next поля, это туда, куда дети должны пойти.

instance Functor ThreadF where       f `fmap` (Fork  next next) = Fork  (f next) (f next)       f `fmap` (Yield next)      = Yield (f next)       f `fmap` Done              = Done 

Теперь свободный монадный трансформер FreeT может построить синтаксическое дерево наших команд. Мы будем называть это дерево потоком (Thread):

-- из `free` пакета import Control.Monad.Trans.Free    type Thread = FreeT ThreadF 

Опытный Хаскель-программист будет читать этот код, как бы проговаривая «поток Thread является синтаксическим деревом построенным из инструкций ThreadF».

Инструкции

Теперь нам нужны примитивные инструкции. free пакет предоставляет liftF операцию, которая преобразует одну команду в синтаксическое дерево на один узел глубже:

yield :: (Monad m) => Thread m () yield = liftF (Yield ())  done :: (Monad m) => Thread m r done = liftF Done  cFork :: (Monad m) => Thread m Bool cFork = liftF (Fork False True) 

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

• yield команда сохраняет () в качестве своего ребенка, поэтому и возвращаемое значения функции равно ()
• done команда не имеет детей, так что компилятор выводит, что она имеет полиморфное возвращаемое значение (т.е. r), это означает, что оно никогда не завершится
• Команда cFork хранит логические значения в качестве детей, поэтому она возвращает Bool

cFork получила свое название потому, что она ведет себя как fork функция из C, это значит, возвращённое логическое значение говорит нам, на какой мы ветви находимся после разветвления. Если получаем False, то мы находимся на левой ветви и если мы получаем True, то мы находимся на правой ветви.

Мы можем объединить cFork и done заново, реализовав fork в более традиционном стиле Хаскеля, используя соглашение, что левая ветвь является «родителем» и правая ветвь является «ребёнком»:

import Control.Monad  fork :: (Monad m) => Thread m a -> Thread m () fork thread = do     child <- cFork     when child $ do         thread         done 

Приведенный выше код вызывает cFork, а потом как-бы говорит: «Если я ребенок, запустите раздвоенное действие, а затем остановитесь, в противном случае просто продолжайте как обычно».

Свободные монады

Обратите внимание как случилось нечто необычное в последнем фрагменте кода. Мы собрали из примитивных инструкций потока Thread функций cFork и done с использованием обозначений do, и мы получили новый поток Thread обратно. Это потому, что Хаскель позволяет нам использовать do нотацию любого типа, реализующего интерфейс монад (Monad) и наш свободный монадный трансформер автоматически определяет нужный instance монады для Thread. Восхитительно!

На самом деле, наш свободный монадный трансформер совсем не супер-умный. Когда мы собираем свободный монадный трансформер с использованием do нотации, все что делается, это подключаются эти примитивные синтаксические деревья в один узел глубиной (т.е. инструкции) в большее синтаксическое дерево. Последовательность двух команд:

do yield    done 

… обессахаривается просто в хранение второй команды (т.е. done) в качестве дочернего первой команды (т.е. yield).

Циклический диспетчер потоков

Теперь мы собираемся написать наш собственный планировщик потоков. Это будет наивный циклический планировщик:

-- Очередь с O(1) операциями над головой и хвостом import Data.Sequence   roundRobin :: (Monad m) => Thread m a -> m () roundRobin t = go (singleton t)  -- Начнём с единственного потока   where     go ts = case (viewl ts) of         -- Если очередь пуста: готово!         EmptyL   -> return ()          -- Очередь не пуста: Начинаем работать с первого потока         t :< ts' -> do             x <- runFreeT t  -- Запускаем эффекты этого потока             case x of                 -- Новые потоки идут в конец очереди                 Free (Fork t1 t2) -> go (t1 <| (ts' |> t2))                  -- Отдающие потоки идут в конец очереди                 Free (Yield   t') -> go (ts' |> t')                  -- Поток выполнен: Избавляем очередь от потока                 Free  Done        -> go ts'                 Pure  _           -> go ts' 

… и готово! Нет, правда, это всё! Это полная потоковая реализация.

Пользовательские потоки

Давайте попробуем нашу новую храбрую потоковую систему. Начнём с чего-то простого

mainThread :: Thread IO () mainThread = do     lift $ putStrLn "Forking thread #1"     fork thread1     lift $ putStrLn "Forking thread #1"     fork thread2  thread1 :: Thread IO () thread1 = forM_ [1..10] $ \i -> do     lift $ print i     yield  thread2 :: Thread IO () thread2 = replicateM_ 3 $ do     lift $ putStrLn "Hello"     yield 

Каждый из этих потоков имеет тип Thread IO (). Thread — это «монадный трансформер», а это означает, что он расширяет существующую монаду дополнительной функциональностью. В нашем случае мы расширяем IO монаду пользовательскими потоками, а это, в свою очередь, означает, что каждый раз, когда нам необходимо вызвать IO действие, мы используем lift, чтобы вставить это действие в поток Thread.

Когда мы вызываем функцию roundRobin, мы вытаскиваем наш Thread монадный трансформер, и наша потоковая программа коллапсирует до линейной последовательности инструкций в IO

>>> roundRobin mainThread :: IO () Forking thread #1 Forking thread #1 1 Hello 2 Hello 3 Hello 4 5 6 7 8 9 10 

Более того, наша потоковая система чистая! Мы можем расширять другие монады, не только IO, и всё равно получать потоковые эффекты! Например, мы можем построить потоковые Writer вычисления, где Writer — одна из многочисленных чистых монад (детальней о ней, см. на хабре):

import Control.Monad.Trans.Writer  logger :: Thread (Writer [String]) () logger = do     fork helper     lift $ tell ["Abort"]     yield     lift $ tell ["Fail"]  helper :: Thread (Writer [String]) () helper = do     lift $ tell ["Retry"]     yield     lift $ tell ["!"] 

В этот раз функция roundRobin производит чистое Writer действие, когда мы запускаем logger:

roundRobin logger :: Writer [String] () 

… и мы можем извлечь результаты команды-логирования чисто тоже:

execWriter (roundRobin logger) :: [String] 

Заметьте, как тип вычисляет чистое значение, список String в нашем случае. И мы всё ещё можем получить реальные потоки логированных значений:

>>> execWriter (roundRobin logger) ["Abort","Retry","Fail","!"] 

Заключение

Вы можете подумать, что я читер, что основная работа досталась библиотеке free, но всю функциональность, которую я использовал можно уместить в 12 строчек очень общего кода, годного к вторичному использованию.

data FreeF f a x = Pure a | Free (f x)  newtype FreeT f m a =     FreeT { runFreeT :: m (FreeF f a (FreeT f m a)) }  instance (Functor f, Monad m) => Monad (FreeT f m) where     return a = FreeT (return (Pure a))     FreeT m >>= f = FreeT $ m >>= \v -> case v of         Pure a -> runFreeT (f a)         Free w -> return (Free (fmap (>>= f) w))  instance MonadTrans (FreeT f) where     lift = FreeT . liftM Pure  liftF :: (Functor f, Monad m) => f r -> FreeT f m r liftF x = FreeT (return (Free (fmap return x))) 

Это частая тенденция в Хаскеле: когда мы используем теорию, мы получаем часто использованное, элегантное и мощное решение в шокирующие малом количестве кода.

Написание статьи было вдохновлено статьёй Пенг Ли и Стива Ждантевича «Методы языка для объединения потоков и событий». Основное отличие состоит в том, что методы продолжения были заменены на более простые методы свободной монады.