yieldRepeatIO creates memory leak when not using IO

[cheerio-pixel]

As the title says. Here is a code that exemplifies this problem:

global using static AppPrelude;
using LanguageExt.Pipes;
using LanguageExt;
using static LanguageExt.Prelude;
using LanguageExt.Traits;
using System.Runtime.CompilerServices;
using System.Net.Sockets;
using System.Net;
using System.Text;
using LanguageExt.Common;
using LanguageExt.Sys;
using M = Application;

var effect = Module<M>.producer | Module<M>.pipe | Module<M>.consumer;
// effect.Run().As().Run();
effect.Run().As().Run.Run(new (AtomHashMap(("", "")))).Run().Ignore();

public static class AppPrelude
{

    [MethodImpl(MethodImplOptions.AggressiveInlining)]
    public static Application<A> As<A>(this K<Application, A> self) =>
        (Application<A>)self;
}

public record ApplicationState(AtomHashMap<string, string> Cache);

public record Application<A>(StateT<ApplicationState, IO, A> Run) : K<Application, A>;

public class Application
: Deriving.Monad<Application, StateT<ApplicationState, IO>>,
  Deriving.Stateful<Application, StateT<ApplicationState, IO>, ApplicationState>,
  MonadIO<Application>
{
    public static K<Application, A> LiftIO<A>(IO<A> ma)
    {
        return new Application<A>(
                StateT.liftIO<ApplicationState, IO, A>(ma)
                );
    }

    public static K<StateT<ApplicationState, IO>, A> Transform<A>(K<Application, A> fa)
    {
        return fa.As().Run;
    }

    public static K<Application, A> CoTransform<A>(K<StateT<ApplicationState, IO>, A> fa)
    {
        return new Application<A>(fa.As());
    }
}

public static class Module<M>
where M : MonadIO<M>
{
    public static ProducerT<int, M, Unit> producer =>
        from _2 in IO.lift(() => Console.WriteLine("Producer"))
        from _1 in ProducerT.yieldRepeatIO<M, int>(
                IO.pure(1)
                    )
        select unit;

    public static PipeT<int, int, M, Unit> pipe =
        from i in PipeT.awaiting<M, int, int>()
        from _2 in IO.lift(() => Console.WriteLine("Pipe"))
        from _ in PipeT.yieldRepeatIO<M, int, int>(IO.pure(i))
        select unit;

    public static ConsumerT<int, M, Unit> consumer =
        from x in ConsumerT.awaiting<M, int>()
        from _2 in IO.lift(() => Console.WriteLine("Consumer"))
        select unit;
}

This is in the latest version (language v5-beta54)
The commented run is for IO, which if you substitue for M is going to work as expected.

[cheerio-pixel]

Forgot to add, the problem with the code is that it halts on the first yieldRepeatIO and if we substitute for a normal yield then it halts on the second (duh, they are internally the same type) and it seems it accumulates memory until there is no more memory left but the program doesn't crash or I just haven't waited long enough.

[louthy]

You appear to be using IO, so not sure what the title means?

The pipe only needs to yield, not yieldRepeat.

Make sure you test in a Release build, as sometimes the debugging info can cause a the leak.

If the leak remains then you can avoid it by manually implementing Applicative.Actions

Finally, just lifting IO.pure may well cause an issue as the internal binding of IO may not bounce on the trampoline, so try it with file streaming, or some other real IO rather than already-evaluated IO.

Let me know how you get on.

[cheerio-pixel]

Sorry for not being clear, what I mean by "not using IO" is setting M to be StateT<S, IO> instead of IO. Also this is a problem that I originally enconterred with this code

    public static ProducerT<Socket, M, Unit> createServer(IPAddress address, int port) =>
        from server in IO.lift(() =>
                {
                    Socket server = new(
                            AddressFamily.InterNetwork,
                            SocketType.Stream,
                            ProtocolType.IP
                            );
                    // This is only set to fix a small problem that the
                    // challenge has with opening and closing connections
                    // really quickly
                    server.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress, true);
                    server.Bind(new IPEndPoint(address, port));
                    server.Listen(128);
                    return server;
                })
        from _1 in ProducerT.yieldRepeatIO<M, Socket>(
                 from client in IO.liftAsync(server.AcceptAsync)
                 from _ in IO.lift(() => Log.Information("New client"))
                 select client
            )
        select unit;

The pure there is just to simplify the example. Also running it in Release doesn't make a change. Will come back later (or tomorrow) with the Actions override.

[louthy]

I tested with a bespoke Actions and the memory-usage stays flat:

public static K<Application, A> Actions<A>(IEnumerable<K<Application, A>> fas) =>
    new Application<A>(
        from s in StateT.get<IO, ApplicationState>()
        from r in fas.Select(fa => fa.As().Run.Run(s)).Actions()
        from _ in StateT.put<IO, ApplicationState>(r.State)
        select r.Value);

One thing to note is that you only get the last ApplicationState, there's no state propagation for each Action. If you need that you could use a Atom<ApplicationState> to update a reference after each IO operation.

The reason this works

If you look at the source for PipeTYieldAll<IN, OUT, M, A>, which is the DSL entry for dealing with yieldRepeat and yieldRepeatIO, you'll see the call to Actions() in the RunAsync method.

The IO implementation of Actions supports recursion, whereas the default Actions implementation (provided by the Applicative interface) looks like this:

public static virtual K<F, A> Actions<A>(IEnumerable<K<F, A>> fas)
{
    K<F, A>? ra = null;
    foreach (var fa in fas)
    {
        ra = ra is null 
                 ? fa
                 : F.Action(ra, fa);
    }
    if(ra is null) throw Exceptions.SequenceEmpty;
    return ra;
}

That is a naive implementation that simply chains many Action calls together. It is very likely to create stack pressure (memory leak). So, because we know about our own monad-transformer stack components, we can Run the stack to get to the inner IO (which isn't available in the trait definitions, so can't be generalised), then call Actions on IEnumerable<IO<?>>.

Now that I have the Iterable and Iterator types, there may be a better default Actions that I could provide that may just work when combined with monad-transformer stacks that have IO in them, so I'll have a think about how I could improve that.

[cheerio-pixel]

With TailRecM you can do that for any monad as long as it has an implementation. It will look something like this

    public static K<Application, A> Actions<A>(IEnumerable<K<Application, A>> fas) =>
        TailRecM(
                Seq(fas),
                a => a.Case switch {
                (K<Application, A> x, Seq<K<Application, A>> xs) => x.Map(_ => Left<Seq<K<Application, A>>, A>(xs)),
                K<Application, A> x => x.Map(r => Right<Seq<K<Application, A>>, A>(r)),
                null => throw Exceptions.SequenceEmpty,
                _ => throw new Exception("Unahndled case " + a.Case)
                });

    public static K<Application, B> TailRecM<A, B>(A seed, Func<A, K<Application, Either<A, B>>> f)
    {
        return new Application<B>(
            new StateT<ApplicationState, IO, B>(state =>
            {
                return IO<(B Value, ApplicationState State)>.Lift(() =>
                {
                    var currentState = state;
                    var currentValue = seed;

                    while (true)
                    {
                        var result = f(currentValue).As().Run
                            .Run(currentState)
                            .Run();

                        currentState = result.State;

                        if (result.Value.IsLeft)
                        {
                            currentValue = (A)result.Value;
                        }
                        else
                        {
                            return ((B)result.Value, currentState);
                        }
                    }
                });
            })
        );
    }

As you can see we can substitute Application for M and it would work the same.

[louthy]

But it doesn't exist, so you can't ;)

[cheerio-pixel]

hahahaha, I know I was just point it out for a use case of TailRecM