DataCollector in a functional way

[mozesa]

Dear @louthy,

I would like to discuss on the snippet below.

I would like to rewrite it using functional approach.

Originally, I implemented it in a functional way, but I gave up as I was unsure with the parallel processing and the cancellation.

My main difficulties that I don't know what is the up-to-date information with the cancellation.
I have bumped into PR where Eff also got some kind of cancellation support, how it is possible at all, that a sync process gets cancelled? - Or am I wrong?

I know it is much, but would it be possible to show us what would be your approach?

Thank you for all the effort bringing functional programming closer to C#.

static class Foo
{
    public static Task MainAsync(IEnumerable<string> ips, CancellationToken ct) =>
        Parallel.ForEachAsync(ips, ct, Loop); 

    static async ValueTask Loop(string ip, CancellationToken ct)
    {
        while (!ct.IsCancellationRequested)
        {
            using (var local = CancellationTokenSource.CreateLinkedTokenSource(ct))
            {
                _ = await Connect(ip, local.Token, async stream =>
                {
                   // for health check
                    var task = Task.Run(async () =>
                    {
                        while (!local.IsCancellationRequested)
                        {
                            await stream.WriteAsync(Encoding.ASCII.GetBytes("||>GET DEVICE.NAME\r\n"), local.Token);
                            await Task.Delay(TimeSpan.FromSeconds(5), local.Token);
                        }
                    }, local.Token);

                    // call external APIs
                    await foreach (var reading in ReadingsAsync(stream, timeout: TimeSpan.FromSeconds(10), ct: local.Token))
                    {
                        var timestamp = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss");

                        switch (reading)
                        {
                            case "||[105]":
                                goto Disconnect;

                            case {} s when s.StartsWith("DM"):
                                await PostConnected(timestamp, ip);
                                break;

                            case { Length: > 0 and <= 200 }:
                                await PostScanData(reading, timestamp, ip);
                                break;
                        }
                    }

                    Disconnect:
                    local.Cancel();

                    try { await task; }
                    catch (TaskCanceledException) {}

                    return true; 
                });
            }

            await PostDisconnected(DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"), ip);
            await Task.Delay(TimeSpan.FromSeconds(5), ct);
        }
    }
}

[louthy]

Currently CancellationToken is only supported in Aff<RT, A> (the runtime version of Aff), that's because the token is managed automatically by the runtime, no specific passing around of CancellationToken in every method.

There is some work going on to add a simple runtime to Aff<A> that only contains the CancellationToken, but that won't be ready for a while. Eff won't get the CancellationToken support at all. It was discussed, but I realise it's not needed as combining an Eff with an Aff always promotes the resulting expression to an Aff, and therefore only Aff needs the CancellationToken support.

I don't have enough time to make your above code functional, but you can always capture something like the above code within an Aff<RT, A> and get the cancellation-token directly out of the runtime.

static class Foo<RT>
    where RT: struct, HasCancel<RT>
{
    public static Aff<RT, Unit> MainAsync(IEnumerable<string> ips) =>
        ips.SequenceParallel(Loop)
           .Map(_ => unit);

    static Aff<RT, Unit> Loop(string ip) =>
        Aff<RT, Unit>(rt =>
        {
            using (var local = CancellationTokenSource.CreateLinkedTokenSource(rt.CancellationToken))
            {
                _ = await Connect(ip, local.Token, async stream =>
                {
                    // for health check
                    var task = Task.Run(async () =>
                    {
                        while (!local.IsCancellationRequested)
                        {
                            await stream.WriteAsync(Encoding.ASCII.GetBytes("||>GET DEVICE.NAME\r\n"), local.Token);
                            await Task.Delay(TimeSpan.FromSeconds(5), local.Token);
                        }
                    }, local.Token);

                    // call external APIs
                    await foreach (var reading in ReadingsAsync(stream, timeout: TimeSpan.FromSeconds(10),
                                       ct: local.Token))
                    {
                        var timestamp = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss");

                        switch (reading)
                        {
                            case "||[105]":
                                goto Disconnect;

                            case { } s when s.StartsWith("DM"):
                                await PostConnected(timestamp, ip);
                                break;

                            case {Length: > 0 and <= 200}:
                                await PostScanData(reading, timestamp, ip);
                                break;
                        }
                    }

                    Disconnect:
                    local.Cancel();

                    try
                    {
                        await task;
                    }
                    catch (TaskCanceledException)
                    {
                    }

                    return unit;
                });
            }

            await PostDisconnected(DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"), ip);
            await Task.Delay(TimeSpan.FromSeconds(5), ct);
            return unit;
        });
}

Obviously that means you spend all your time writing boilerplate like while (!local.IsCancellationRequested) which is automatically done for you within the Aff monad. But it can be a starting point. Usually if I was to do a refactor like this, I'd break each part of the large function up into key expressions and then compose them with LINQ.

The while loops are obviously completely imperative. In FP we use recursion, or you'd need to look at using the Pipes functionality, for proper streaming composition. I don't know what you're connecting to or streaming from, so it's hard for me to comment on the exact behaviour here.

[mozesa]

Thanks for the help.
I have started to refactor.

When I try to access to CancellationToken, I have an error The type arguments cannot be inferred from the query., without from ct in cancelToken<RT>(), it is fine.

[CS8016] Transparent identifier member access failed for field 'ln' of 'int'. Does the data being queried implement the query pattern?

@louthy could you help me figuring out, what could be the problem?

public static class DataCollector<RT>
    where RT : struct, HasCancel<RT>, HasConsole<RT>
{
    public static Aff<RT, Unit> Main(IEnumerable<string> ips) =>
        ips.SequenceParallel(Loop)
            .Map(_ => unit);

    static Aff<RT, Unit> Loop(string ip) =>
        use(Eff(() => new TcpClient()), client =>
            from ct in cancelToken<RT>()
            from _ in client.ConnectAsync(ip, 23, ct).ToUnit().ToAff()
            let stream = client.GetStream()
            from cancel in fork(Health(stream))
            from __ in TcpConnect(stream) | readLine | writeLine
            select unit)
        | @catch(error => Console<RT>.writeLine(error.Message).ToAff());

    static Aff<RT, Unit> Health(Stream stream) =>
        from ct in cancelToken<RT>()
        let cmd = Encoding.ASCII.GetBytes("||>GET DEVICE.NAME\r\n")
        from _ in Aff(() => stream.WriteAsync(cmd, ct).ToUnit()).Repeat(Schedule.spaced(5 * second) | Schedule.RepeatForever)
        select unit;

    static Producer<RT, TextReader, Unit> TcpConnect(Stream stream) =>
        from reader in use<RT, StreamReader>(SuccessEff(new StreamReader(stream)))
        from _ in Producer.yield<RT, TextReader>(reader)
        select unit;

    static Pipe<RT, TextReader, string, Unit> readLine
    {
        get
        {
            return from tr in awaiting<TextReader>()
                from ct in cancelToken<RT>()
                from ln in enumerate2(go(tr))
                from __ in yield(ln)
                select unit;

            static async IAsyncEnumerable<string> go(TextReader reader)
            {
                while (true)
                {
                    var line = await reader.ReadLineAsync().ConfigureAwait(false);
                    if(line == null) yield break;
                    yield return line;
                }
            }
        }
    } 
    
    static Consumer<RT, string, Unit> writeLine =>
        from l in awaiting<string>()
        from _ in Console<RT>.writeLine(l)
        select unit;
}

Update

Using enumerate instead of enumerate2 solves the problem, but don't know why... 😃

static Pipe<RT, TextReader, string, Unit> readLine
{
    get
    {
        return from tr in awaiting<TextReader>()
            from ct in cancelToken<RT>()
            from ln in enumerate(go(tr))
            // from _ in yield(ln)
            select unit;

        static async IAsyncEnumerable<string> go(TextReader reader)
        {
            while (true)
            {
                var line = await reader.ReadLineAsync().ConfigureAwait(false);
                if (line == null) yield break;

                yield return line;
            }
        }
    }
}