Конвейеры

Оператор конвейера $(\mathbin{\mathtt{>>>}})$: та же композиция, но читается слева направо, по движению данных. Определён он для любого инстанса $\mathtt{Category}$:

(>>>) :: Category cat => cat a b -> cat b c -> cat a c
f >>> g = g . f

-- стрелки категории Hask:
clean :: String -> String
clean = trim >>> lowercase >>> collapseSpaces

-- стрелки категории Клейсли:
pipeline :: Kleisli Maybe String User
pipeline = Kleisli parse >>> Kleisli validate >>> Kleisli build

Пока наш конвейер — одна труба: $(\mathbin{\mathtt{>>>}})$ соединяет шаги строго друг за другом. Но раз уж мы взялись управлять потоком данных, почему бы не пустить добро по трубам на полную катушку: подать на вход пару и обработать её половинки порознь. Для этого существует класс типов $\mathtt{Arrow}$:

class Category a => Arrow a where
  arr    :: (b -> c) -> a b c
  first  :: a b c -> a (b, d) (c, d)
  second :: a b c -> a (d, b) (d, c)
  (***)  :: a b c -> a b' c' -> a (b, b') (c, c')
  (&&&)  :: a b c -> a b c'  -> a b (c, c')

Тут никуда без пояснений: $\mathtt{arr}$ поднимает чистую функцию в конвейер; $\mathtt{first}$ пропускает через стрелку первую компоненту пары, а вторую ведёт обводным каналом нетронутой — это и есть «данные мимо шага»; $\mathtt{second}$ — зеркальный близнец $\mathtt{first}$: обрабатывает вторую компоненту, первая едет мимо; $(\mathbin{\mathtt{***}})$ запускает две стрелки по половинкам пары параллельно; $(\mathbin{\mathtt{\&\&\&}})$ раздваивает поток: один вход дублируется в обе стрелки, на выходе — пара результатов. При этом двух методов класса достаточно: $\mathtt{second}$, $(\mathbin{\mathtt{***}})$ и $(\mathbin{\mathtt{\&\&\&}})$ выводятся из $\mathtt{arr}$ и $\mathtt{first}$.

Теперь пара инстансов — обычная функция и стрелка Клейсли:

-- подставим a := (->) в сигнатуру first:
first :: a    b c -> a    (b, d) (c, d)
first :: (->) b c -> (->) (b, d) (c, d)   -- a := (->)
first :: (b -> c) -> ((b, d) -> (c, d))   -- префиксное (->) x y = инфиксное x -> y

instance Arrow (->) where
  arr f          = f
  first f (x, y) = (f x, y)

-- то же для стрелки Клейсли, a := Kleisli m:
first :: a         b c -> a         (b, d) (c, d)
first :: Kleisli m b c -> Kleisli m (b, d) (c, d)   -- a := Kleisli m
first :: (b -> m c)    -> ((b, d) -> m (c, d))      -- сняли newtype-обёртку

instance Monad m => Arrow (Kleisli m) where
  arr f              = Kleisli (pure . f)
  first (Kleisli f)  = Kleisli (\(x, y) -> do c <- f x; pure (c, y))

Практический пример — сводка по временам ответа: среднее и максимум одним конвейером. На рис. 2.10 покажем, как он идёт, — заметьте, что $(\mathbin{\mathtt{\&\&\&}})$ там дважды: не только снаружи, но и внутри ($\mathtt{sum} \mathbin{\mathtt{\&\&\&}} \mathtt{genericLength}$):

mean :: [Double] -> Double
mean = (sum &&& genericLength) >>> uncurry (/)

summary :: [Double] -> (Double, Double)
summary = mean &&& maximum

summary [12, 8, 40]   -- (20.0, 40.0)

На чистых функциях свет клином не сошёлся: комбинаторы класса $\mathtt{Arrow}$ работают для любой стрелки. Перенесём сводку $\mathtt{summary}$ в категорию Клейсли: времена ответа теперь не лежат готовым списком, а читаются из файла — конвейер живёт в IO. Первый шаг — настоящая стрелка Клейсли, дальше $\mathtt{arr}$ поднимает чистые $\mathtt{mean}$ и $\mathtt{maximum}$ в ту же категорию, и развилка $(\mathbin{\mathtt{\&\&\&}})$ повторяет строчку $\mathtt{summary}$ один в один:

-- та же сводка, но времена ответа теперь читаются из файла
readTimes :: Kleisli IO FilePath [Double]
readTimes = Kleisli readFile >>> arr (lines >>> map read)

summaryIO :: Kleisli IO FilePath (Double, Double)
summaryIO = readTimes >>> arr mean &&& arr maximum   -- ср.: summary = mean &&& maximum

runKleisli summaryIO "times.log"   -- файл со строками 12, 8, 40 -> (20.0, 40.0)

Теперь несколько слов скажем не про пару, а про $\mathtt{Either}$ — работа с ним вынесена в отдельный класс $\mathtt{ArrowChoice}$:

class Arrow a => ArrowChoice a where
  left  :: a b c -> a (Either b d) (Either c d)
  right :: a b c -> a (Either d b) (Either d c)
  (+++) :: a b c -> a b' c' -> a (Either b b') (Either c c')
  (|||) :: a b d -> a c d   -> a (Either b c) d

Каждый метод — двойник метода из Arrow: $\mathtt{left}$ и $\mathtt{right}$ — как $\mathtt{first}$ и $\mathtt{second}$, только вместо половинки пары обрабатывают одну ветвь $\mathtt{Either}$, а другую пропускают нетронутой; если $(\mathbin{\mathtt{***}})$ обрабатывает обе половинки пары, то его двойник $(\mathbin{\mathtt{+++}})$ — лишь ту ветвь, которая реально пришла, а $(\mathbin{\mathtt{|||}})$ — двойник $(\mathbin{\mathtt{\&\&\&}})$ — вдобавок сливает обе ветви в общий выход, маршрутизируя поток. Инстанс для обычной стрелки $\mathtt{(\to)}$:

-- подставим a := (->) в сигнатуру left:
left :: a    b c -> a    (Either b d) (Either c d)
left :: (->) b c -> (->) (Either b d) (Either c d)   -- a := (->)
left :: (b -> c) -> (Either b d -> Either c d)

instance ArrowChoice (->) where
  left f (Left  x) = Left (f x)   -- пришла левая ветвь — обрабатываем
  left f (Right y) = Right y      -- правая проходит мимо нетронутой

-- остальные методы выводятся из left:
f +++ g = left f >>> right g
f ||| g = (f +++ g) >>> arr (either id id)

Практический пример — та же сводка, но с защитой от пустого лога: $(\mathbin{\mathtt{+++}})$ ведёт каждую ветвь своим конвейером, а $(\mathbin{\mathtt{|||}})$ сливает их в общий выход:

checkLog :: [Double] -> Either String [Double]
checkLog xs = if null xs then Left "пустой лог" else Right xs

report :: [Double] -> String
report = checkLog >>> (("нет сводки: " ++) +++ summary) >>> (id ||| show)

report [12, 8, 40]   -- "(20.0,40.0)"
report []            -- "нет сводки: пустой лог"