# Parallel stream processing with zero-copy fan-out and sharding

**URL:** <https://discourse.haskell.org/t/parallel-stream-processing-with-zero-copy-fan-out-and-sharding/8632>\
**Category:** Show and Tell\
**Created:** [January 24, 2024, 10:57am UTC](https://discourse.haskell.org/t/parallel-stream-processing-with-zero-copy-fan-out-and-sharding/8632 "2024-01-24T10:57:18Z")\
**Posts on this page:** 9\
**Page:** 1

<div class="post-metadata">

**Author:** ![stevan](https://avatars.discourse-cdn.com/v4/letter/s/45deac/32.png) [@stevan](https://discourse.haskell.org/u/stevan)\
**Post date:** [January 24, 2024, 10:57am UTC](https://discourse.haskell.org/t/parallel-stream-processing-with-zero-copy-fan-out-and-sharding/8632/1 "2024-01-24T10:57:18Z")

</div>

I’m interested in parallel stream processing. Not the map/reduce kind of  
parallelism, but rather pipeline parallelism like in manufacturing and CPU  
instruction pipelines.

I’ve recently made some progress using an `Arrow` interface and leveraging  
associated types to avoid copying data between queues that connect the different  
stages of the pipeline:

[Parallel stream processing with zero-copy fan-out and  
sharding](https://stevana.github.io/parallel_stream_processing_with_zero-copy_fan-out_and_sharding.html)

This is heavily inspired by Martin Thompson et al’s work on the LMAX Disruptor  
and Aeron. As I was writing things up I also stumbled upon an interview with the  
late Jim Gray where he [advocated](https://www.youtube.com/watch?v=U3eo49nVxcA&t=1949s) for pipeline parallelism.

As far as I know, no streaming library in Haskell uses this approach, but I’d be  
happy to be proven wrong. I would also be happy to try to answer any questions  
or listen to any kind of feedback, cheers!

---

<div class="post-metadata">

**Author:** ![turion](https://sea2.discourse-cdn.com/flex002/user_avatar/discourse.haskell.org/turion/32/989_2.png) [@turion](https://discourse.haskell.org/u/turion)\
**Post date:** [January 24, 2024, 8:49pm UTC](https://discourse.haskell.org/t/parallel-stream-processing-with-zero-copy-fan-out-and-sharding/8632/2 "2024-01-24T20:49:53Z")

</div>

One thing I often run into with such approaches is that arrow notation desugars into a lot of `arr` and `>>>`, but never `***` even if it could. For example:

```haskell
proc (a, b) -> do
  c <- f -< a
  d <- g -< b
  return (c, d)

```

Semantically, this is ` ***`, but it will desugar as some mess like `first f >>> second g` decorated with needless `arr`s. But there is no way to shard `>>>`, only `*** ` and `&&&`, right? Or is your approach able to parallelize `>>>` well? If not, do you avoid arrow notation then?

---

<div class="post-metadata">

**Author:** ![stevan](https://avatars.discourse-cdn.com/v4/letter/s/45deac/32.png) [@stevan](https://discourse.haskell.org/u/stevan)\
**Post date:** [January 24, 2024, 9:22pm UTC](https://discourse.haskell.org/t/parallel-stream-processing-with-zero-copy-fan-out-and-sharding/8632/3 "2024-01-24T21:22:59Z")

</div>

Yeah, you are right that arrow notation desugaring is bad/wrong, I briefly touch  
upon this in [footnote 3](https://stevana.github.io/parallel_stream_processing_with_zero-copy_fan-out_and_sharding.html#fn3), an alternative could also be to do your own desugaring  
a la the `arrowp-qq` library. Either way, for now one would have to use the  
combinators explicitly, unfortunately.

You can think of sharding (what Jim Gray calls partition parallelism) as taking  
a copy of a whole pipeline and diverge half of the traffic (even indices) to the  
first copy and the other half (odd indices) to the second copy. This works  
irregardless of what combinators the pipeline is composed out of, and we can  
deterministically retain the order of the outputs (because we’ve split by  
even/odd).

Pipeline parallelism (which is orthogonal to sharding/partition parallelism) is  
introduced by `(>>>)`, where by its first argument gets deployed on one thread  
and the second argument on another thread and the two stages get connected with  
a queue. You can see this in the `runQueueSleep` example (it uses `&&&`, but  
that’s not important, parallelism would happen despite it) and you can also compare  
it with the `runQueueSleepSharded` example, which adds sharding.

---

<div class="post-metadata">

**Author:** ![olf](https://sea2.discourse-cdn.com/flex002/user_avatar/discourse.haskell.org/olf/32/3813_2.png) [@olf](https://discourse.haskell.org/u/olf)\
**Post date:** [January 24, 2024, 10:00pm UTC](https://discourse.haskell.org/t/parallel-stream-processing-with-zero-copy-fan-out-and-sharding/8632/4 "2024-01-24T22:00:12Z")

</div>

Just an observation: I think what you have defined with `P` is a free arrow (see below) and deployment is mapping the free arrow to a less free one. I’m not exactly an expert on this, but there exists research on arrows that uses the arrow laws to study program transformation, whose goals may include parallelism.

Recently I attempted to use the arrow idiom to define a complex model that should be checkable against a graphical specification, but gave up eventually because code became incomprehensible. Do-notation for Kleisli arrows with let bindings turned out to be superior for this task. (Plus it turns out that there exists no automatic translation from Haskell arrows back to diagrams, where one would easily _see_ the parallelism.)

```haskell
-- | Abstract syntax tree over arrow combinators.
-- In @FreeArrow arr@ the type @arr@ contains 
-- The special capabilities we want our workflow to have. 
data FreeArrow :: (Type -> Type -> Type) -> Type -> Type -> Type where
    Arr :: (x -> y) -> FreeArrow arr x y
    PureArr :: arr x y -> FreeArrow arr x y
    ComposeArrow :: FreeArrow arr y z -> FreeArrow arr x y -> FreeArrow arr x z
    First :: FreeArrow arr x y -> FreeArrow arr (x,z) (y,z)
    Second :: FreeArrow arr x y -> FreeArrow arr (z,x) (z,y)
    Pair :: FreeArrow arr x y -> FreeArrow arr a b -> FreeArrow arr (x,a) (y,b)
    FanOut :: FreeArrow arr x y -> FreeArrow arr x z -> FreeArrow arr x (y,z)

instance Category (FreeArrow arr) where
    id = Arr (Data.Function.id)
    f . g = ComposeArrow f g
instance Arrow (FreeArrow arr) where
    arr = Arr
    first = First
    second = Second
    (***) = Pair
    (&&&) = FanOut

-- | interpret the abstract syntax in a concrete category.
fromFreeWith :: Arrow arr' =>
    (forall x y. arr x y -> arr' x y) ->
    FreeArrow arr a b -> arr' a b
fromFreeWith h (Arr f) = arr f
fromFreeWith h (ComposeArrow f g) = fromFreeWith h f <<< fromFreeWith h g
fromFreeWith h (First f ) = first (fromFreeWith h f)
fromFreeWith h (Second f ) = second (fromFreeWith h f)
fromFreeWith h (Pair f g ) = fromFreeWith h f *** fromFreeWith h g
fromFreeWith h (FanOut f g ) = fromFreeWith h f &&& fromFreeWith h g
fromFreeWith h (PureArr f ) = h f

```

---

<div class="post-metadata">

**Author:** ![stevan](https://avatars.discourse-cdn.com/v4/letter/s/45deac/32.png) [@stevan](https://discourse.haskell.org/u/stevan)\
**Post date:** [January 25, 2024, 6:31am UTC](https://discourse.haskell.org/t/parallel-stream-processing-with-zero-copy-fan-out-and-sharding/8632/5 "2024-01-25T06:31:41Z")

</div>

Or with other words: `first f >>> second g` will be parallel, but not as  
parallel as `f *** g`.

In the first case, we’ll have two threads working on the queue in a pipeline  
fashion, i.e. once `first f` is done with the first item in the queue it will  
pass it on to `second g` while at the same time starting to work on the second  
item, thus introducing parallelism (assuming the two threads run on different  
CPUs/cores).

In the second case, we first need to note that a Disruptor queue of pairs will  
be represented as a pair of queues (similar to how `vector` or array of structs  
vs struct of arrays works). So here we will again have two threads, but this  
time without any dependency between them. The first thread gets the first queue  
(containing all the `fst` components of the pair, without having to do any copying of items) while the second thread gets the second queue, and in the end we form a pair of queues again. Here the `fst`  
part of the first item is processed in parallel with the `snd` part of the first  
item, so this should perform better than the first case. (It’s important to note  
that any consumer of the output of `f *** g` may only consume items that have  
been handled by both `f` and `g`, this is done by keeping track of which index  
both components are at and then taking the minimum.)

---

<div class="post-metadata">

**Author:** ![stevan](https://avatars.discourse-cdn.com/v4/letter/s/45deac/32.png) [@stevan](https://discourse.haskell.org/u/stevan)\
**Post date:** [January 25, 2024, 6:33am UTC](https://discourse.haskell.org/t/parallel-stream-processing-with-zero-copy-fan-out-and-sharding/8632/6 "2024-01-25T06:33:16Z")

</div>

> Just an observation: I think what you have defined with P is a free arrow (see  
> below) and deployment is mapping the free arrow to a less free one.

Thanks! I currently have `deploy`ment and visualisation as a combined  
transformation, it would be nice to be more clear about `P` being a free arrow  
and separating the two transformations.

> I’m not exactly an expert on this, but there exists research on arrows that  
> uses the arrow laws to study program transformation, whose goals may include  
> parallelism.

Do you have any reference?

> Recently I attempted to use the arrow idiom to define a complex model that  
> should be checkable against a graphical specification, but gave up eventually  
> because code became incomprehensible. Do-notation for Kleisli arrows with let  
> bindings turned out to be superior for this task. (Plus it turns out that  
> there exists no automatic translation from Haskell arrows back to diagrams,  
> where one would easily see the parallelism.)

Interesting, do you have any code to share?

I also tried to visualise my pipelines, see the section on [observability](https://stevana.github.io/parallel_stream_processing_with_zero-copy_fan-out_and_sharding.html#observability), but granted only for small examples.

I’m not sure if it’s any use to you or perhaps you are already aware of it, but  
Conal Elliott’s work on [compiling to categories](http://conal.net/papers/compiling-to-categories/) also visualises parallelism.

---

<div class="post-metadata">

**Author:** ![olf](https://sea2.discourse-cdn.com/flex002/user_avatar/discourse.haskell.org/olf/32/3813_2.png) [@olf](https://discourse.haskell.org/u/olf)\
**Post date:** [January 25, 2024, 7:23am UTC](https://discourse.haskell.org/t/parallel-stream-processing-with-zero-copy-fan-out-and-sharding/8632/7 "2024-01-25T07:23:42Z")

</div>

> [@stevan](#):
>
> > I’m not exactly an expert on this, but there exists research on arrows that  
> > uses the arrow laws to study program transformation, whose goals may include  
> > parallelism.
> 
> Do you have any reference?

There is for example a paper on how [point-free style](https://repositorium.sdum.uminho.pt/bitstream/1822/14350/1/frapfpt.pdf) can help in program transformation, but it does not mention arrows. (But arrows are point-free syle.) More relevant may be [this preprint](https://arxiv.org/pdf/1801.02216.pdf). Judging from section 4.1 it seems the authors are concerned with classical parallel evaluation like

```haskell
[arr x y] -> arr [x] [y]

```

where a list models a vector of ad-hoc accessible data rather than a stream of values that become available one after the other.

> [@stevan](#):
>
> Interesting, do you have any code to share?

Unfortunately not at the moment, since it is company property. But the `FreeArrow` definition I showed arose in that attempt. I am concerned with self-documenting computations and arrows could be a great tool for that. The idea is that one first defines the algorithm in a free arrow, hiding a chosen level of implementation details inside the `PureArrow` constructor. The `FreeArrow` AST could be used to generate graphical representations, apply program transformations such as identify potential parallelism and machine-check the algorithm against graphical specifications. For actually running the algorithm the AST is mapped to a concrete and more efficient `Arrow` category.

I asked the author of the Haskell [arrows page](https://www.haskell.org/arrows/) how the diagrams were generated and he said they were done by hand. Disappointing. Do you think Conal’s _compiling to categories_ has the necessary toolkit for that?

---

<div class="post-metadata">

**Author:** ![olf](https://sea2.discourse-cdn.com/flex002/user_avatar/discourse.haskell.org/olf/32/3813_2.png) [@olf](https://discourse.haskell.org/u/olf)\
**Post date:** [January 25, 2024, 7:49am UTC](https://discourse.haskell.org/t/parallel-stream-processing-with-zero-copy-fan-out-and-sharding/8632/8 "2024-01-25T07:49:06Z")

</div>

The SVG viewer ist cool! I also use SVG to visualize data because it provides interactive features when embedded in HTML, which can be shared as a static file. But it never occurred to me I could overlay completely different pictures in the same canvas. I use the blaze family of packages to generate both the SVG and the surrounding HTML with JavaScript from the same Haskell program. I embed `data-whatever` attributes in the elements and have a JavaScript routine that collects their values and performs computation. That is a limited way of embedding functions into a picture.

---

<div class="post-metadata">

**Author:** ![stevan](https://avatars.discourse-cdn.com/v4/letter/s/45deac/32.png) [@stevan](https://discourse.haskell.org/u/stevan)\
**Post date:** [January 25, 2024, 8:05am UTC](https://discourse.haskell.org/t/parallel-stream-processing-with-zero-copy-fan-out-and-sharding/8632/9 "2024-01-25T08:05:12Z")

</div>

> [@olf](#):
>
> Do you think Conal’s _compiling to categories_ has the necessary toolkit for that?

I’m not sure if you can get diagrams exactly like those on the arrows wiki, also note that Conal works with cartesian closed categories rather than arrows, but I’d guess it can be adapted for your purpose.