C# - Concurrent Foreach Like Thread System

Question

IObservable<Match> IObservableArray = new Regex("(.*):(.*)").Matches(file).OfType<Match>().ToList().ToObservable();
var query = IObservableArray.SelectMany(s => Observable.Start(() => {
    //do stuff
}));

Working Code Above's Explanation: The code above uses Observable with Reactive to do a Concurrent Multi-Threading system while retaining s as a Match.

---

My issue is that it seems to need to load everything into memory before even starting doing //do stuff since IObservableArray is a big array of Matches - this takes up a lot of the memory causing it to do a OutOfMemory Exception.

I have been researching for more than a month and all I can find is .Buffer() which if I put it before the .SelectMany() and then foreach Match over the s, im able to load 1000 Matches into memory at a time causing the memory overall to be much better.

But, since I have to resort to using a foreach to go through all 1000 in the buffer at a time, it isnt concurrent - meaning im basically checking 1 after the other.

Is there a way to do similar code below, yet have it Concurrent/Multi-Threaded? (Have at least 150 running concurrently, but don't load all of it to memory, using 1000 at the moment.)

yes I tried using thread.start etc, using them makes it trigger the finished code much earlier since technically it does finish since it had done what it was told which was make them all into a new thread

IObservable<Match> IObservableArray = new Regex("(.*):(.*)").Matches(file).OfType<Match>().ToList().ToObservable();
var query = IObservableArray.Buffer(1000).SelectMany(s => Observable.Start(() => {
    //do stuff
}));
query.ObserveOn(ActiveForm).Subscribe(x =>
{
    //do finish stuff
});

Answer 1

You're not actually telling Start() which scheduler to use, which may be the reason you're not getting the concurrency you want. You can specify the desired scheduler as the second argument:

var query = IObservableArray.Buffer(1000).SelectMany(s => Observable.Start(() => {
    //do stuff
}, TaskPoolScheduler.Default));

If you know that //do stuff is going to take more than 500ms, I'd consider using the ThreadPoolScheduler instead. The task pool won't spawn a new thread until a Task has blocked a thread for at least 500ms, so if you know you're going to do a lot of heavy work and need a lot of threads, you can use ThreadpoolScehduler.Instance instead of TaskPoolScheduler.Default.

Answer 2

For this kind of work an IEnumerable<T> is a better match than an IObservable<T>. An enumerable is something that you can unwind on demand, and take its values when you are ready to process them. On the contrary an observable is something that forcefully pushes its values onto you, whether you are able to handle the load or not.

There are numerous ways to process an IEnumerable<T> in parallel, with a specific degree of parallelism. Before suggesting anything, the first question to ask is whether the stuff you have to do with each Match is synchronous or asynchronous. For synchronous work the most commonly used tools are the Parallel class, the PLINQ, and the TPL Dataflow library. Below is a PLINQ example:

IEnumerable<Match> matches = RegexFindAllMatches(file, "(.*):(.*)");
Partitioner
    .Create(matches, EnumerablePartitionerOptions.NoBuffering)
    .AsParallel()
    .WithDegreeOfParallelism(Environment.ProcessorCount)
    .ForAll(match =>
    {
        // Do stuff
    });

/// <summary>
/// Provides an enumerable whose elements are the successful matches found by
/// iteratively applying a regular expression pattern to the input string.
/// </summary>
public static IEnumerable<Match> RegexFindAllMatches(
    string input, string pattern, RegexOptions options = RegexOptions.None,
    TimeSpan matchTimeout = default)
{
    if (matchTimeout == default) matchTimeout = Regex.InfiniteMatchTimeout;
    var match = Regex.Match(input, pattern, options, matchTimeout);
    while (match.Success)
    {
        yield return match;
        match = match.NextMatch();
    }
}

The above implementation avoids the use of the Regex.Matches method, and subsequently the MatchCollection class, because although this class evaluates lazily the next Match during the enumeration, it then stores each found Match in an internal ArrayList (source code). This could cause massive memory allocation, proportional to the total numbers of matches.

For asynchronous work the Parallel class and the PLINQ are not good options (unless you are willing to wait for the .NET 6 Parallel.ForEachAsync), but you can still use the TPL Dataflow library. You can also find plenty of custom options here or here. Searching for C# ForEachAsync should reveal even more options.