Pipelines

The pipeline operator $(\mathbin{\mathtt{>>>}})$: the same composition, but read left to right, the way the data flows. It is defined for any instance of $\mathtt{Category}$:

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

-- arrows of the Hask category:
clean :: String -> String
clean = trim >>> lowercase >>> collapseSpaces

-- arrows of the Kleisli category:
pipeline :: Kleisli Maybe String User
pipeline = Kleisli parse >>> Kleisli validate >>> Kleisli build

So far our pipeline is a single pipe: $(\mathbin{\mathtt{>>>}})$ joins the stages strictly one after another. But since we have taken to steering the flow of data, why not run the goods through the pipes at full throttle: feed a pair in and process its halves separately. This is what the $\mathtt{Arrow}$ type class is for:

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')

There is no getting by without explanations here: $\mathtt{arr}$ lifts a pure function into the pipeline; $\mathtt{first}$ runs the first component of a pair through the arrow while carrying the second past it, untouched, along a bypass channel — this is the “data past a stage”; $\mathtt{second}$ is the mirror twin of $\mathtt{first}$: it processes the second component while the first rides by; $(\mathbin{\mathtt{***}})$ runs two arrows over the two halves of a pair in parallel; $(\mathbin{\mathtt{\&\&\&}})$ forks the flow: a single input is duplicated into both arrows, and the output is the pair of results. In fact, two class methods are enough: $\mathtt{second}$, $(\mathbin{\mathtt{***}})$ and $(\mathbin{\mathtt{\&\&\&}})$ are all derived from $\mathtt{arr}$ and $\mathtt{first}$.

Now a couple of instances — the plain function and the Kleisli arrow:

-- substitute a := (->) into the signature of 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))   -- prefix (->) x y is infix x -> y

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

-- the same for a Kleisli arrow, 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))      -- with the newtype wrapper removed

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))

A practical example — a summary of response times: the mean and the maximum in one pipeline. Fig. 2.10 shows how it flows — note that $(\mathbin{\mathtt{\&\&\&}})$ appears twice: not only on the outside but also inside ($\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)

Pure functions are not the whole story: the $\mathtt{Arrow}$ combinators work for any arrow. Let us move the $\mathtt{summary}$ pipeline into the Kleisli category: the response times no longer sit in a ready-made list but are read from a file — the pipeline now lives in IO. The first step is a genuine Kleisli arrow, then $\mathtt{arr}$ lifts the pure $\mathtt{mean}$ and $\mathtt{maximum}$ into the same category, and the $(\mathbin{\mathtt{\&\&\&}})$ fork repeats the $\mathtt{summary}$ line word for word:

-- the same summary, but the response times now come from a file
readTimes :: Kleisli IO FilePath [Double]
readTimes = Kleisli readFile >>> arr (lines >>> map read)

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

runKleisli summaryIO "times.log"   -- a file with lines 12, 8, 40 -> (20.0, 40.0)

Now a few words not about the pair but about $\mathtt{Either}$; working with it is moved out into a separate class, $\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

Each method is a twin of an Arrow method: $\mathtt{left}$ and $\mathtt{right}$ mirror $\mathtt{first}$ and $\mathtt{second}$, only they process one branch of an $\mathtt{Either}$ instead of one half of a pair, letting the other branch pass through untouched; if $(\mathbin{\mathtt{***}})$ processes both halves of a pair, its twin $(\mathbin{\mathtt{+++}})$ processes only the branch that actually arrived, while $(\mathbin{\mathtt{|||}})$ — the twin of $(\mathbin{\mathtt{\&\&\&}})$ — additionally merges the two branches into a common output, routing the flow. The instance for the plain function arrow $\mathtt{(\to)}$:

-- substitute a := (->) into the signature of 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)   -- the left branch arrived — process it
  left f (Right y) = Right y      -- the right one passes through untouched

-- the remaining methods are derived from left:
f +++ g = left f >>> right g
f ||| g = (f +++ g) >>> arr (either id id)

A practical example — the same summary, now guarded against an empty log: $(\mathbin{\mathtt{+++}})$ sends each branch down its own pipeline, and $(\mathbin{\mathtt{|||}})$ merges them into a common output:

checkLog :: [Double] -> Either String [Double]
checkLog xs = if null xs then Left "empty log" else Right xs

report :: [Double] -> String
report = checkLog >>> (("no summary: " ++) +++ summary) >>> (id ||| show)

report [12, 8, 40]   -- "(20.0,40.0)"
report []            -- "no summary: empty log"