Parallel.ForEachAsync like forks concurrency control

[djm132]

Hi,

Searching for good examples how to implement the following task:

1. I have some computation which I need to run parallel at max N tasks at time, it looks like this

private static K<M, Unit> StreamBalances(
    Channel<ValidatorBalance> channel,
    Network network,
    string stateId,
    IEnumerable<string>? pubKeys = null) =>
    from options in Json.Map(t => t.Options)
    from _1 in Log<RT>.debug("Requesting {network} validator balances: {state}", network.Name, stateId)
    from uri in GetValidatorBalancesUri(network, stateId, pubKeys)
    from stream in Http<M, RT>.GetStream(uri)
    from res in stream
        .StreamFirstPropertyArray<ValidatorBalance>("data", options)
        .StreamToChannel(channel)
    select res;

2. These computations must be run at parallel from this code (fork just run-and-forget here) which returns multi-writer producer from channel

public static SourceT<M, ValidatorBalance> GetBalances(Network network, IEnumerable<ulong> slots) =>
    from baseUri in GetBaseUri(network)
    from channel in M.Pure(Channel.CreateUnbounded<ValidatorBalance>(new() { SingleWriter = false }))
    from _ in
        slots
            .Select(s => StreamBalances(channel, network, s.ToString()))
            .ProcessParallel<M, ValidatorBalance, RT>(channel, 5)
            .Fork()
    from source in SourceT.lift<M, ValidatorBalance>(channel)
    select source;

3. Here is the problematic code which I trying to resolve without success

public static IO<Unit> ProcessParallel<M, T, RT>(
    this IEnumerable<K<M, Unit>> computations,
    Channel<T> channel,
    int parallelism = 1) where M : MonadUnliftIO<M> =>
    IO.liftAsync(async env =>
    {
        await Parallel.ForEachAsync(
            computations,
            new ParallelOptions { MaxDegreeOfParallelism = parallelism, CancellationToken = env.Token },
            async (comp, ct) =>
            {
                // TODO: Unsafe cast, remove
                var c = comp as Eff<RT, Unit>;

                // TODO: How to execute awaited computation here to get Task/ValueTask to await ??

                return;
            });

        channel.Writer.Complete();
        return unit;
    });

Can't find the way to elegantly unlift computation and execute it. Any ideas ?

Thanks.

[micmarsh]

There's no way to execute a generic MonadLiftIO, and to Run an Eff<RT> you'd need access to an instance of the runtime, which you wouldn't have in imperative async code.

If you do depend on the computation actually being an Eff, then you could/should remove a lot of the generic type params throughout the whole call stack, and you can just do

    public static Eff<RT, Unit> ProcessParallel<RT>(
        this IEnumerable<Eff<RT, Unit>> computations,
        int parallelism = 1) =>
        Eff.lift<RT, Unit>(
            async (rt) =>
               {
                   await Parallel.ForEachAsync(
                        computations, 
                           new ParallelOptions { MaxDegreeOfParallelism = parallelism },
                           async (eff, ct) =>
                           {
                               await eff.RunIO(rt).RunAsync(EnvIO.New(token: ct));
                           });
                   return unit;
               });

otherwise, I would start with this

    public static K<M, Unit> ProcessParallel<M>(
        this IEnumerable<K<M, Unit>> computations,
        int parallelism = 1) where M : MonadUnliftIO<M>
    {
        //todo something that will probably involve messing with a mutable collection internally
    }

and try to fill in the blanks from there. It's an interesting problem, but I personally don't have the time for it at the moment.

The business with passing around and closing the channel is likely better handled by the caller, and since you are using a Channel, are you aware of Conduit? I haven't used it directly myself, but from what I can tell it seems to be a wholesale replacement for Channel, managing its own Source and Sink.
Upon further inspection (thinking about the Channel usage), you may be looking for ForkIO.Await? That could simplify things a lot.

[djm132]

Thanks for your reply.

I've figured some things out

1. It's impossible to execute any given computation depending on RT alone.
2. RT should be carried in some monad. I am using Eff<RT, A> app-wide which is K<Eff<RT>, A> and have RT backed in.
3. So I've rewritten my parallel code and it works

/// <summary>
/// Run tasks in parallel up to a given parallelism.
/// </summary>
/// <typeparam name="RT">Runtime environment type.</typeparam>
/// <typeparam name="A">Computation result.</typeparam>
/// <param name="tasks">List of tasks.</param>
/// <param name="parallelism">Parallelism limit.</param>
public static Eff<RT, Unit> ToParallel<RT, A>(this IEnumerable<Eff<RT, A>> tasks, int parallelism) =>
    Eff.lift<RT, Unit>(rt =>
        Parallel
            .ForEachAsync(
                tasks,
                new ParallelOptions { MaxDegreeOfParallelism = parallelism },
                async (eff, ct) =>
                {
                    await eff.RunIO(rt).RunAsync(ct);
                })
            .ToUnit());

4. I've also discovered Conduit yesterday which is wrapper around Channel and made these

Consumer side

/// <summary>
/// Write item to queue.
/// </summary>
/// <typeparam name="A">Item type.</typeparam>
/// <param name="queue">Queue.</param>
/// <returns>Pipe consumer.</returns>
private static Consumer<RT, A, Unit> writeToQueue<A>(Conduit<A, A> queue) =>
    from x in Consumer.awaiting<RT, A>()
    from _ in queue.Post(x)
    select unit;

Pipe definition

/// <summary>
/// Pump items from source to queue.
/// </summary>
/// <typeparam name="A">Item type.</typeparam>
/// <param name="source">Source of items.</param>
/// <param name="queue">Target queue.</param>
private static Effect<RT, Unit> pump<A>(Source<A> source, Conduit<A, A> queue) =>
    from effect in source.ToProducer<RT>() | writeToQueue(queue)
    select unit;

And task which should be parallel

/// <summary>
/// Stream validator balances.
/// </summary>
/// <param name="queue">Queue to pipe balances to.</param>
/// <param name="network">Network.</param>
/// <param name="stateId">State id.</param>
/// <param name="date">Balances date.</param>
/// <param name="pubKeys">Optional public keys to limit request.</param>
public static Eff<RT, Unit> pumpBalances(
    Conduit<Balance, Balance> queue,
    Network network,
    string stateId,
    LocalDate date,
    Seq<string> pubKeys = default) =>
    from _1 in Log<RT>.debug("Requesting {network} balances at {date:R}: {state}", network.Name, date, stateId)
    from uri in getValidatorBalancesUri(network, stateId, pubKeys)
    from stream in Http<RT>.getStream(uri)
    from src in Json<RT>.streamPropertyArray<ValidatorBalance>(stream, "data")
    from _2 in pump(src.Map(b => new Balance(b.Index, date, b.Balance)), queue).Run()
    select unit;

So running computations in parallel is solved.
But there are other problems now related to pipes and probably my misunderstanding of them

1. Nothing pushed to conduit (i've also tried recursive consumer without success)
2. I think there are problems with how pipe is defined and run.

Debugging it now.

[micmarsh]

Great, glad to see you're unblocked (at least on the "parallelize this" front)! Do you have a background in Haskell? Unfortunately I have no experience with pipes (in either language), since "the basic" Source, Sink and Conduit queuing mechanism has always been more than sufficient for my needs, but hopefully someone else can help out if you need it.

[djm132]

Don't have any experience in Haskell, only F#. Also checked pipes docs and in "theory" I have understanding how it works, but the devil hides in technical details, as usual )) Rewritten pipes with raw channel as optimization (there is a tight loop with processing 150M+ records).

BTW, how you push and pull many data to/from conduit ? Fold/Reduce ?

[micmarsh]

Post is how to push to a Conduit (or its Sink) and Fold/Reduce is typically how you'd "pull" data from a Source (which is also associated with Conduits). You can use Transducers to transform a Source (without explicitly pulling) as well, but you basically need experience with Clojure to start working with those.

If I'm understanding your intent correctly (just streaming all data from a source to a sink), unless you're just intentionally trying to use pipes as a way of learning them, you can define this helper

    public static IO<Unit> pump<A>(Source<A> source, Sink<A> sink) =>
        source.ReduceIO(unit, (_, a) => sink.Post(a) * Reduced.Continue);

and then the last line of your LINQ statement could just be

    from _2 in pump(src.Map(b => new Balance(b.Index, date, b.Balance)), queue.Sink)

without any pipes at all.

[djm132]

FYI, created my pipes issue #1558