Parallel IO Example

[glendonjklassen]

I'm looking through the v5 language-ext Samples folder for an example of parallel processing but I can't find any references to forkIO and awaitAll in that context. (I also might be looking for the wrong commands). Would someone be able to share a simple example of how to perform multiple IOs in parallel and collect the results?

What I'm trying to get at is something equivalent to TestTask below in TestIO.

[Fact]
public async Task TestTask()
{
    var tasks = new List<Task<DateTime>>
                {
                    Task.FromResult(DateTime.Now),
                    Task.FromResult(DateTime.Now),
                    Task.FromResult(DateTime.Now)
                };
    var all = await Task.WhenAll(tasks);
}

[Fact]
public void TestIO()
{
    var test =
        // Does not compile, accepts Seq<IO<ForkIO<T>>> not Seq<ForkIO<T>>
        from t in awaitAll<DateTime>(LanguageExt.List.generate(3, _ => Async.fork(() => DateTime.Now)).ToSeq())
        select t;
    var result = test.Run();
}

Thanks for your time!

Glendon

[louthy]

To make any operation run in parallel you can use IO.fork, it works with IO<A> or any K<M, A> where the M trait-implementation supports MonadIO.ToIO.

You can use it like so:

var delayed    = IO.liftAsync(async () =>
                             {
                                 await Task.Delay(3000);
                                 return unit;
                             });
var readKey    = IO.lift(Console.ReadKey);
var startTimer = IO.lift(Stopwatch.StartNew);
var writeLine  = (object? value) => IO.lift(() => Console.WriteLine(value));

var op = from watch in startTimer
         from fork1 in IO.fork(delayed)
         from fork2 in IO.fork(delayed)
         from fork3 in IO.fork(delayed)
         from _1    in fork1.Await >> fork2.Await >> fork3.Await
         from _2    in writeLine($"Elapsed: {watch.ElapsedMillisecon}")
         select unit;

op.Run().Ignore();

I realise that there's no useful awaitAll for working with raw ForkIO values, only IO<ForkIO> values, so that's why there's the sequenced awaits:

fork1.Await >> fork2.Await >> fork3.Await

I'll add the awaitAll overload for the next release

Another way that it can work is:

awaitAll(IO.fork(delayed), IO.fork(delayed), IO.fork(delayed))

Note, all of the above is for parallel processing. If you just want concurrent processing, then use Traverse (on collections of IO), and Zip (on tuples of IO). See the IO.Extensions.cs.

I notice you're using Async.*. Avoid that. It's really just to help migrate away from Task. It's nothing to do with the IO monad. I may even remove it, because I'm not sure it really adds anything, even for those migrating from v4 to v5. If you want to make any function into a parallel IO operation, you must first lift the function:

// Take any function
Func<X> f = ...;

// Make it into an IO operation (synchronous)
IO<X> synchronousIO = IO.lift(f);

// Fork it to launch on another thread
IO<ForkIO<X>> parallelIO = IO.fork(synchronousIO);

If the function already returns a Task then use IO.liftAsync(f).

[glendonjklassen]

That's very helpful thank you. I'll adapt my code to remove the Async.*. Cheers.