Pipes refactor and Cofunctors

[louthy]

LanguageExt Pipes Background

Part of the v5 refresh was to migrate the Pipes functionality to be a proper monad-transformer (in v4 it's a transformer too, but it can only lift Eff<RT, A>, rather than the more general K<M, A> where M : Monad<M>). I completed the generalisation work a while back, but it had some problems:

For any users of pipes it was going to be a big upheaval

Obviously, v5 is a big change, but where possible I want the migrations to be quite mechanical - it wasn't going to be. That doesn't mean I shouldn't 'go for it', but I'm trying to make sure that every bit of pain a user has to go through to move from v4 to v5 is strongly justified and will lead to a better experience once migrated.

It was inconsistently named

The core type Proxy, and the derived types: Producer, Consumer, Pipe, etc. don't follow the monad-transformer naming convention of having a T suffix. Really, if they're going to be generalised for any monad then they should be called ProducerT, ConsumerT, PipeT, ...

Pipes is hard to use

This is not a new problem with v5. I made Pipes into a 1-for-1 clone of the Haskell Pipes library. Even in Haskell they can be quite hard to use as you chase alignment of generics. The desire for pipes to support: producers, pipes, clients, servers, and more seems (in hindsight) to be too greedy.

Hard to retrofit

The generalisation process wasn't working well in some areas. The Producer.merge was blocking and fixing it with the original code was challenging to say the least.

LanguageExt Pipes Refresh

So, I decided to take a step back. Instead of trying to make an exact clone of the Haskell version, I thought I'd build it from scratch in a way that's more 'csharpy', consistent, and simpler. In particular I looked at the techniques I used to refactor the IO monad (to support recursion, asynchrony, etc.) and brought them into a new Pipes implementation.

I also decided to drop support for Client, Server, Request, Response, and all of the other stuff that I suspect nobody used because they were too hard.

That means:

• There's no need for an underlying Proxy<A1, A, B1, B, M, R> interface. This was only needed to support all flavours of client, server, producer, consumer, etc.
• The base-type of all pipes related types is: PipeT<IN, OUT, M, R>
  • This is clearly easier to understand
  • A ProducerT<OUT, M, R> is simply a pipe with the input set to Unit:
    • PipeT<Unit, OUT, M, R>
  • A ConsumerT<IN, M, R> is simply a pipe with the output set to Void:
    • PipeT<IN, Void, M, R>
  • A EffectT<M, R> is simply a pipe with the input set to Unit and the output set to Void. This enclosed effect is the result of fusing producer, pipe, and consumers together:
    • PipeT<Unit, Void, M, R>

Those four types: ProducerT, PipeT, ConsumerT, and EffectT are the new simplified and, fully generalised, version of pipes.

Now that the generalised implementation follows the naming convention of having a T suffix for transformers, we can use the original names Producer, Pipe, Consumer, and Effect to provide a more specialised version that only works with Eff<RT, A> (like the original pipes).

So,

• Producer<RT, OUT, R> is (internally) a ProducerT<OUT, Eff<RT>, R>
• Pipe<RT, IN, OUT, R> is (internally) a PipeT<IN, OUT, Eff<RT>, R>
• Consumer<RT, IN, R> is (internally) a ConsumerT<IN, Eff<RT>, R>
• Effect<RT, IN, R> is (internally) a EffectT<IN, Eff<RT>, R>

The good thing about this refactor is that there really is only one implementation of the pipes functionality and it all sits in the PipesT.DSL.cs . This focused DSL is much easier to manage than before - it was implemented in a similar way before, but it's now just much easier for a C# dev to consume. I have put a real effort into making the interfaces, modules, preludes, etc. consistent for all types.

Pipes concurrency

Concurrency wasn't front-and-centre in the original implementation. In some senses it was 'bolted on'. You got concurrency from the lifted Eff type and from the Producer.merge function, but that was it.

Now pipes has first-class support for concurrency:

• Support for IEnumerable and IAsyncEnumerable with ProducerT.yieldAll, Producer.yieldAll, PipeT.yieldAll, and Pipe.yieldAll.
• Unlike the original, the core DSL supports the lifting of tasks
  • Which means direct support from: PipeT.liftT, PipeT.liftM, Pipe.liftT, Pipe.liftM, ProducerT.liftT, ProducerT.liftM, Producer.liftT, Producer.liftM, ConsumerT.liftT, ConsumerT.liftM, Consumer.liftT, Consumer.liftM, EffectT.liftT, EffectT.liftM, Effect.liftT, and Effect.liftM!

Mailbox, Inbox, and Outbox

Inspired by the original Pipes.Concurrency library, I implemented Mailbox, Inbox, and Outbox. It's not a clone of the original, just inspired by. A Mailbox consists of an Inbox and an Outbox. The inbox receives values posted to it. The outbox yields values posted to the inbox upon request.

Backing the Mailbox is a System.Threading.Channels.Channel. You can create a Mailbox like so:

var mailbox = Mailbox.spawn<string>();

A mailbox is simply a record with an Inbox and Outbox:

public record Mailbox<A, B>(Inbox<A> Inbox, Outbox<B> Outbox)

You can Post to the Mailbox and you can Read from the Mailbox. But, even more critically, you can call:

• mailbox.ToConsumer<M>() - to get a consumer of values being posted into the Inbox
• mailbox.ToProducer<M>() - to get a producer of values being yielded into the Outbox

A good example of why this is useful is the new Producer.merge function:

public static ProducerT<OUT, M, Unit> merge<OUT, M>(Seq<ProducerT<OUT, M, Unit>> producers) where M : Monad<M> =>
    from mailbox in Pure(Mailbox.spawn<OUT>())
    from forks   in forkEffects(producers, mailbox)
    from _       in mailbox.ToProducerT<M>()
    from x       in forks.Traverse(f => f.Cancel).As()
    select unit;

static K<M, Seq<ForkIO<Unit>>> forkEffects<M, OUT>(
    Seq<ProducerT<OUT, M, Unit>> producers,
    Mailbox<OUT, OUT> mailbox)
    where M : Monad<M> =>
    producers.Map(p => (p | mailbox.ToConsumerT<M>()).Run())
             .Traverse(ma => ma.ForkIO());

The merge function gets a collection of producers. What we want is for those to run concurrently so we can receive the values as they happen. Then we want to produce a single merged stream of values.

This creates the merged stream Mailbox:

    from mailbox in Pure(Mailbox.spawn<OUT>())

In forkEffects we process each producer p and pipe its values to mailbox.ToConsumerT:

p | mailbox.ToConsumerT<M>()

So, we get a ConsumerT for the merged-stream's Mailbox. It consumes every value from p, fusing into an EffectT. We then Run() that EffectT which gives us the underlying M monad:

(p | mailbox.ToConsumerT<M>()).Run()

We do this for every ProducerT, which means the merged-values Mailbox gets every value yielded from upstream.

producers.Map(p => (p | mailbox.ToConsumerT<M>()).Run())

Finally, we ForkIO each EffectT so that it can run in parallel.

.Traverse(ma => ma.ForkIO())

Back to the merge function, we then access the other side of the mailbox by asking for the Outbox producer, using ToProducerT:

from _ in mailbox.ToProducerT<M>()

This will then yield all of the merged values downstream (whilst there are values to yield). Once complete, we tidy up the forks:

forks.Traverse(f => f.Cancel).As()

Cofunctor

Mailbox is pretty powerful in its own right and doesn't need pipes to function. This is a quick example of a loop that reads every value posted to a Mailbox and writes it to the console:

static IO<Unit> consumeAll(Mailbox<string, string> mailbox) =>
        from x in mailbox.Read()
        from _ in IO.lift(() => Console.WriteLine(x))
        from r in consumeAll(mailbox)
        select r;

Mailbox<A, B> has two type parameters: A represents the values coming in and B represents the values being yielded.

    A -> B

Values of type A are posted to Mailbox.Inbox and values of type B are yielded from Mailbox.Outbox.

If you call mailbox.Map<C>((B b) => ...) on Mailbox then you could imagine Mailbox being represented like this:

    A -> B -> C

The result is a Mailbox<A, C>, but internally there's a mapping of the values as they flow through.

Subsequent calls to Map<D>, and the like, would continue to transform the value being yielded from the Mailbox.Outbox:

    A -> B -> C -> D

But what if we wanted to transform the values being posted into the Mailbox.Inbox (the A value). Map doesn't work here, because it transforms an existing value, we'd have to Map the A to something else. But, to do that, we'd have to have an A value.

So, there's no way we can change the values coming in? Well there is, but not with Functor and Map. We need Contravariant Functors; colloquially known as 'Co-functors'.

When it comes to category-theoretic concepts, 'co', can usually be read to mean 'reverse the arrows'. Or, in other words, find the 'dual' of. So a co-functor is a functor with the arrows reversed.

Functor, looks like this:

F<B> Map<B>(F<A> fa, Func<A, B> f);

It maps an A -> B. This can be seen as mapping the values that F<A> yields after they've been yielded.

Let's reverse the arrows:

F<A> Contramap<B>(F<B> fb, Func<A, B> f);

Now, it takes an F<B> and function from A -> B and returns an F<A>. This may seem batshit crazy. How can we get a value of A out of an F<B> to pass to the f function?

We can't. And we won't be doing that. F in this case is not a type that yields values, but a type that receives values. It's a sink rather than a stream. So, the f is being used to transform values coming into the F<B> (before arrival), not transforming values being yielded (after leaving).

Mailbox.Inbox is a Cofunctor and so, you can call Contramap on the Mailbox to transform values before they are posted into the Inbox.

Custom Mailboxes

Because Mailbox is simply a record that takes an Inbox and an Outbox, you can build your own without using Mailbox.spawn. Inbox is currently created from a System.Threading.Channels.ChannelWriter:

record InboxWriter<A>(ChannelWriter<A> Writer, string Label) : Inbox<A>
{
    public override Inbox<B> Contramap<B>(Func<B, A> f) => 
        new InboxContraMap<A, B>(f, this);

    public override IO<Unit> Post(A value) =>
        from f in IO.liftVAsync(e => Writer.WaitToWriteAsync(e.Token))
        from r in f ? IO.liftVAsync(() => Writer.WriteAsync(value).ToUnit())
                    : IO.fail<Unit>(Errors.NoSpaceInInbox)
        select r;

    public override IO<Unit> Complete() =>
        IO.lift(() => Writer.Complete());    

    public override IO<Unit> Fail(Error error) =>
        IO.lift(() => Writer.Complete(error.ToException()));    
}

This simply writes to the Channel when a value is posted.

And, Inbox is created from System.Threading.Channels.ChannelReader:

record OutboxReader<A>(ChannelReader<A> Reader, string Label) : Outbox<A>
{
    public override Outbox<B> Map<B>(Func<A, B> f) => 
        new OutboxMap<A, B>(this, f);

    public override Outbox<B> Bind<B>(Func<A, Outbox<B>> f) => 
        new OutboxBind<A, B>(this, f);

    public override Outbox<B> ApplyBack<B>(Outbox<Func<A, B>> ff) => 
        new OutboxApply<A, B>(this, ff);

    public override IO<A> Read() =>
        IO.liftVAsync(e => Reader.WaitToReadAsync(e.Token))
          .Bind(f => f ? IO.liftVAsync(e => Reader.ReadAsync(e.Token))
                       : IO.fail<A>(Errors.OutboxChannelClosed));
    
    internal override ValueTask<bool> ReadyToRead(CancellationToken token) =>
        Reader.WaitToReadAsync(token);
}

Which simply reads a value from Channel when one is available.

You could extend Inbox<A> and Outbox<A> to work with any sink or source-type you like. Channel works pretty well and has good control over buffer-size and back-pressure, but there are other options too.

Divisible and Decidable contravariant functors

Another powerful aspect is that Inbox<A> is Divisible and Decidable.

A Divisible contravariant functor is the contravariant analogue of Applicative.

Continuing the intuition that Contravariant functors consume input, a Divisible contravariant functor also has the ability to be composed "beside" another contravariant functor.

F<A> Divide<A, B, C>(Func<A, (B Left, C Right)> f, F<B> fb, F<C> fc)

If you 'follow the arrows' here, then you'll see that fb and fc get somehow composed using the f function that takes an A value, turns them into a pair (B, C) and then passes them on to fb and fc.

Visually:

                              B --> F<B>     
                             /
    F<A> --> A --> (B, C) -->
                             \
                              C --> F<C>     

So, Divide allows a single F<A> to represent the splitting of values and the routing into two new sinks (F<B> and F<C>). With Inbox this allows an Inbox<A> to be a sink of values of A that then route to other (hidden) Inbox structures.

Decidable contravariant functors are very similar to Divisible contravariant functors. But, instead of generating a tuple of (B, C) to route the incoming values to two other contravariant functors at the same time, Decidable contravariant functors return an Either<B, C>, which means we route the values to only one contravariant functor.

F<A> Route<A, B, C>(Func<A, Either<B, C>> f, F<B> fb, F<C> fc);

Again, Inbox<A> is a Decidable contravariant functor and so you can call Route to direct the values downstream.

Visually, if B is returned by f:

                          B --> F<B>     
                        /
    F<A> --> A --> B | C 

If C is returned by f

    F<A> --> A --> B | C 
                         \
                          C --> F<C>     

---

This discussion was created from the release Pipes refactor.