Difference between the behaviours of FoldWhile and FoldWhileIO

[mysticfall]

Hi.

Maybe I'm fundamentally misunderstanding how SourceT<IO, A> is supposed to work, but I noticed SourceT.FoldWhileIO doesn't work as I expected:

var sum1 = Source
    .forever(1)
    .FoldWhile(
        (s, x) => s + x,
        (s, _) => s <= 10,
        0
    )
    .Take(1)
    .Last()
    .Run();

Console.WriteLine(sum1);

var sum2 = SourceT
    .forever<IO, int>(1)
    .FoldWhileIO(
        0,
        (s, x) => s + x,
        x => x.State <= 10
    )
    .As()
    .Take(1)
    .Last()
    .Run();

Console.WriteLine(sum2);

When I run this example, it prints out the value of sum1 as expected, but it never reaches the second Console.WriteLine.

Is this an expected behaviour? If so, how can I use an infinite stream of SourceT<A>, if methods like FoldWhileIO, or Take doesn't stop consuming it?

[mysticfall]

Maybe it's not FoldWhileIO that's causing the symptom but SourceT itself is. I just found that the following example also never completes:

var sum = SourceT
    .forever<IO, int>(1)
    .FoldWhile((s, x) => s + x, (s, x) => s >= 10, 0)
    .Take(1)
    .Last()
    .Run();

Console.WriteLine(sum);

[mysticfall]

I just stumbled across this closed PR which seems to be relevant to the problem.

I'll subscribe to the linked issue, assuming it's the root cause of the symptom I see.

[louthy]

@mysticfall This should be resolved in v5.0.0-beta-69. The usual caveats apply in that StreamT has minimal testing right now. I'll get to that once I've completed the Megaparsec work.

Oh, and btw, use FoldWhile on StreamT, not FoldWhileIO. FoldWhileIO is for repeatedly running the MonadUnliftIO supporting type. I'm going to look to remove MonadUnliftIO support from StreamT as I think it's unnecessary (because there's already the bespoke Fold operators), it's confusing (two sets of fold functions), and finally I am not convinced the implementation is lawful, which I'm looking into now.

[mysticfall]

As you expected, it turns out I completely misunderstood how FoldWhileIO works. I managed to rewrite my code using FoldWhile as follows, but it didn't solve the issue for me, unfortunately:

public IO<Seq<IObservation>> Think(Duration duration)
{
    Seq<IObservation> init = Empty;

    var timeout = SourceT.liftIO<IO, Seq<IObservation>>(
        IO.yieldFor(duration).Map(init)
    );

    return SourceT.merge(Observations.Map(Seq), timeout)
        .FoldWhile(
            (s, o) => s + o,
            x => !x.Value.IsEmpty,
            init
        )
        .As()
        .Take(1)
        .Last()
        .As();
}

Observations is a never-ending stream, and what I tried to achieve was collecting its items during the specified period. So, I couldn't use TimeoutIO since it would discard the aggregation rather than returning the collected items when the time out is reached.

I suspect I might have another misunderstanding about how the SourceT works. But I suppose I've wasted enough of your time already during a holiday season.

Thanks much for your help, and I hope you enjoy the holidays with your family! 🙂

[louthy]

Yes, there is a way to do this, although it's a little ugly. Essentially you need build a reducer of your own that knows how to complete once a cancellation-token event has fired (the cancellation-token is triggered by the timeout event). And because that's non-obvious, I've added an extra operator to StreamT, namely TakeFor(Duration) and TakeFor(TimeSpan) (available now in v5.0.0-beta-74).

If I create a StreamT that never ends:

static SourceT<IO, DateTime> tickTockIO()
{
    return from token in IO.token
           from value in SourceT.lift<IO, DateTime>(go(token))
           select value;
    
    static async IAsyncEnumerable<DateTime> go(CancellationToken token)
    {
        while (true)
        {
            await Task.Delay(TimeSpan.FromMilliseconds(100), token);
            yield return DateTime.Now;
        }
    }
}

NOTE: Your never ending stream needs to respect the CancellationToken for this to work

Then we need an IObservation maker that takes a DateTime from the stream and turns it into an IObservation:

static IObservation MakeObservation(DateTime dt)
{
    Console.WriteLine($"Making observation: {dt}");
    return new Obs(dt);
}

We can then map the never ending tickTockIO stream to a never ending stream of observations:

static SourceT<IO, IObservation> Observations =>
    tickTockIO().Map(MakeObservation);

Finally, we can use the new TakeFor operator:

public static IO<Option<IObservation>> Think(Duration duration) =>
    Observations
        .Take(1)
        .TakeFor(duration)
        .Last()
        .Map(Some)
    | NoObservations;

NoObservations is simply a pre-calculated IO(None):

static readonly IO<Option<IObservation>> NoObservations = 
    IO.pure<Option<IObservation>>(None);

Which makes the Think expression much more elegant.

If you instead wanted to collect all values yielded over a period of time, you could use something like this:

static IO<Seq<IObservation>> Think2(Duration duration) =>
    +Observations
        .TakeFor(duration)
        .Collect();

If you're interested in how this works behind the scenes (and to understand what you would have needed to do to implement this), then you can take a look at the source of TakeForSourceT:

record TakeForSourceT<M, A>(SourceT<M, A> Source, TimeSpan Duration) : SourceT<M, A>
    where M : MonadIO<M>, Fallible<Error, M>, Alternative<M>
{
    public override K<M, Reduced<S>> ReduceInternalM<S>(S state, ReducerM<M, K<M, A>, S> reducer) =>
        from sofar in atomIO(state)
        from value in M.TimeoutIOMaybe(reduce(state, reducer, sofar), Duration)
                        | @catch(ErrorCodes.TimedOut, timedOut(sofar))
                        | @catch(ErrorCodes.Cancelled, timedOut(sofar))
        select value;

    K<M, Reduced<S>> timedOut<S>(Atom<S> state) =>
        M.LiftIO(state.ValueIO).Map(Reduced.Done);
    
    K<M, Reduced<S>> reduce<S>(S state, ReducerM<M, K<M, A>, S> reducer, Atom<S> sofar) =>
        Source.ReduceInternalM(
            state,
            (s, ma) => from ns in reducer(s, ma)
                       from _ in sofar.SwapIO(_ => ns.Value)
                       select ns);
}

The way it works is to create an Atom<S> that keeps a mutable version of the state of the reducer. It is updated within the reduce function. The reason for this is that we can't 'look inside' the reducer function, so we need to try and keep the latest value just in case a cancellation event happens (timeout).

The whole inner reducer is wrapped around a M.TimeoutIOMaybe call which is what implements the timeout behaviour (by creating a local CancellationTokenSource with a timeout value.

The @catch handlers look for either a timeout or a cancellation and simply get the latest state from the Atom<S> and returns a Reduced.Done value, indicating we want the reducer to stop reducing.

So yeah, not entirely obvious! It's definitely better wrapped up in a bespoke operator.

[mysticfall]

I tested it again after updating to the beta 74 version, and TakeFor worked great. I really like how it made the code more intuitive. Thanks much! 🙂

There also was some interesting problem, however.

Before the update, the original code never yielded any value. Now it does, but it only yields items from the timeout source:

• Any emitted item from the Observations does not reach FoldWhile in the original code.
• At the end of the specified duration, the pipeline yields an empty Seq.
• If I replace the SourceT.merge with just Observations.Map(Seq), the FoldWhile gets its items but the pipeline doesn't yield a result, which is expected.

Although TakeFor would eliminate the need for FoldWhile and make the timeout source redundant, I feel curious why SourceT.merge behave as it does.

And the actual requirements for the Think involves accepting an optional predicate, so that it can stop collecting either when the timeout is reached or some interesting observation that matches the predicate was made, which might be only possible using an operator like FoldWhile, or FoldUntil.

But at least now I can make progress with the addition of TakeFor. And your detailed explanation has been great help for me to learn C# and functional programming. Thanks!

[louthy]

I've made some improvements in v5.0.0-beta-75:

• The merge function now forces evaluation of the monad, so this should cause lazy sequences to start yielding.
• I've added a Delay combinator that allows delaying the yielding of the items in a SourceT
• As well as some other additions and fixes relating to subscribing to events and tidying up of resources

In your original example you never needed to fold at all, but if you wanted to merge two streams and take the first item yielded, then this will work:

public static IO<Seq<IObservation>> Think(Duration duration)
{
    var empty        = SourceT.pure<IO, Seq<IObservation>>([]);
    var timeout      = empty.Delay(duration);
    var observations = Observations.Map(Seq);
    
    return +SourceT.merge(observations, timeout)
                   .Take(1)
                   .Last();
}

You could use Option instead of Seq to represent success or not:

public static IO<Option<IObservation>> Think(Duration duration)
{
    var empty        = SourceT.pure<IO, Option<IObservation>>(None);
    var timeout      = empty.Delay(duration);
    var observations = Observations.Map(Some);
    
    return +SourceT.merge(observations, timeout)
                   .Take(1)
                   .Last();
}

[mysticfall]

With v5.0.0-beta-75, the code works perfectly!

I need folding because the requirement involves collecting all observations until the condition (either the timeout or predicate) meets. But now both merge and FoldWhile/FoldUntil work as expected so it's not a problem anymore.

Here's the current version of my code:

public IO<Seq<IObservation>> Think(
    Duration duration,
    Func<IObservation, bool>? predicate = null
) => SourceT.merge(
        Observations.Map(Seq),
        SourceT.pure<IO, Seq<IObservation>>(Empty).Delay(duration)
    )
    .FoldUntil(
        (s, o) => s + o,
        x => x.Value.IsEmpty || 
             x.Value.Exists(y => predicate?.Invoke(y) ?? false),
        Seq<IObservation>()
    )
    .Take(1)
    .Last()
    .As();

I also like how the new Delay method makes the code much cleaner.

Thanks again for the support! I'm not really a C# programmer but LanguageExt is one of the reasons that make me still enjoying working with the language.

[mysticfall]

By the way, I found something weird while trying out the new TakeFor method. If I rewrite the above code as follows, the observation source is ignored (it only times out with an empty seq):

public IO<Seq<IObservation>> Think(
    Duration duration,
    Func<IObservation, bool>? predicate = null
)
{
    var observations = Observations
        .FoldUntil(
            (s, o) => s + o,
            x => predicate?.Invoke(x.Value) ?? false,
            Seq<IObservation>()
        )
        .TakeFor(duration) | SourceT.pure<IO, Seq<IObservation>>(Empty);

    return observations.Last().As();
}

If I remove the fallback (i.e. | SourceT.pure<IO, Seq<IObservation>>(Empty)), it throws an AggregateException without any inner exception, when the timeout is reached.

Now I don't have an immediate need to use TakeFor since the existing code works as it is with the new version. But I feel curious if it is an intended behaviour or I'm missing something.

[louthy]

Not sure of your intention, but the | operator (when the operands are StreamT) means 'merge', i.e. "yield a value whether it comes from the left-hand stream OR the right-hand stream". Whereas the | operator for IO is like a failure coalescing operator.

It's important to note where the expression switches from SourceT<IO> to IO (i.e. the reducer stage, or Last in your examples so far).

So, in your example you're merging the Observations.FoldUntil(...).TakeFor(...) stream with the singleton SourceT.pure([]) stream and then reducing it using Last. So, it has to wait for the merged streams to complete before returning the last value yielded.

If Observations.FoldUntil(...) yields nothing then the only value that will come downstream is [] from the right-hand stream. If it yields something, you will still need to wait for the TakeFor operation to complete (because nothing else terminates the stream, like Take(1) in earlier examples).

Removing the fallback means that when no values are yielded, the Last reducer needs to find a value from somewhere. It does that by using M.Empty(), which for the IO monad yields IO.fail(Errors.Empty) (an empty AggregateException basically).

A good example is this:

public static K<M, IObservation> Think<M>(Duration duration)
    where M : MonadIO<M>, Fallible<M>, Alternative<M>
{
    var empty        = SourceT.liftM(M.Empty<IObservation>());
    var timeout      = empty.Delay(duration);
    var observations = Observations<M>();

    return (observations | timeout)
               .Take(1)
               .Last();
}

This generalises the Think function to work with any M as long as it supports the MonadIO, Fallible, and Alternative traits.

For example:

Eff<IObservation> r1 = +Think<Eff>(1000);
IO<IObservation>  r2 = +Think<IO>(1000);

Console.WriteLine(r1.Run());
Console.WriteLine(r2.Run());

If we run that and make sure Observations doesn't yield a value within 1 second, what we get is:

Fail([])
Unhandled exception. []

The Eff monad can return Fin.Fail(Errors.Empty) for its M.Empty implementation, so we don't get an exception. The IO monad has no valid M.Empty it can use other than throwing an exception.

If you take a look at the implementaion of the Last reducer, you'll see:

public K<M, A> Last() =>
    ma.As()
      .FoldReduce(Option<A>.None, (_, x) => Some(x))
      .Bind(ma => ma switch
                  {
                      { IsSome: true, Case: A value } => M.Pure(value),
                      _                               => M.Empty<A>()
                  });

It folds into an Option<A>, which if None at the end of the reducer, returns M.Empty<A>().

The reason why exceptions are allowed in IO, but almost nowhere else in language-ext is, mostly, it is expected that IO is at the 'edges' of your application and any exceptions would mean terminating the process. If you don't want to terminate then either coalescing the failure with | and @catch is the way to go.

You could add your own LastOrNone() reducer if you really wanted to avoid the change that M.Empty<A>() yields an exception:

extension<M, A>(K<SourceT<M>, A> ma) 
    where M : MonadIO<M>
{
    public K<M, Option<A>> LastOrNone() =>
        ma.As().FoldReduce(Option<A>.None, (_, x) => Some(x));
}

Or, provide a Last reducer with a default-value if nothing has yielded:

extension<M, A>(K<SourceT<M>, A> ma) 
    where M : MonadIO<M>
{
    public K<M, A> Last(A defaultValue) =>
        ma.As().FoldReduce(defaultValue, (_, x) => x);
}

There are some general improvements in v5.0.0-beta-77 which would be worth getting by the way.

[mysticfall]

Sorry, I though | in Stream API was a fallback operator. I still have some difficulty understanding why it yielded an empty Seq when the observation stream emitted an item.

But I feel I need to spend more time trying to understand how it works before asking more questions. I'm feeling guilty for having taken so much of your time on a holiday season already.

There are some general improvements in v5.0.0-beta-77 which would be worth getting by the way.

Updated now. Thanks! :)