Add and remove Parallel.Invoke actions during runtime

Question

I'm using Parallel.Invoke to run certain methods simultaneously and collect the results when all methods are finished.

PROBLEM

As you can see on the "horrible code" section, the list of actions is hardcoded to three elements, and will be totally useless if the detectedDevicesList.Count != 3.

TRIED SOLUTIONS

I tried to dynamically create an Actions[] array and pass it as an argument for Parallel.Invoke, but I'm unable to convert my existing methods into Tasks, and then into Actions.

NON WORKING CODE

public async Task<Task<String>> callWithEveryConnectedDevice(ListBox listBoxLog, Boolean sendAlarms)
{        
    String TEST_CALLS_COMPLETED = "All test calls completed.";
    String TEST_CALLS_FAILED = "One or more test cals failed";

    return Task.Run(async () =>
    {
        List<MobileEquipment> detectedDevicesList = await GetConnectedDevices.getAsync();

        if (detectedDevicesList.Count == 0)
        {
            UpdateGui.listboxAddItem(listBoxLog, "No devices are connected.", true);
            return TEST_CALLS_FAILED;
        }

        Console.WriteLine("Executing test calls...");

        List<Task<MobileEquipment>> results = new List<Task<MobileEquipment>>();

        //Horrible code begins...
        Parallel.Invoke(() =>
        {
            results.Add(new MakePhoneCall().call(detectedDevicesList[0], listBoxLog));
        },
        () =>
        {
            results.Add(new MakePhoneCall().call(detectedDevicesList[1], listBoxLog));
        },
        () =>
        {
            results.Add(new MakePhoneCall().call(detectedDevicesList[2], listBoxLog));
        });

        //Horrible code ends...
        foreach (Task<MobileEquipment> mobileEquipment in results)
        {
            UpdateGui.listboxAddItem(listBoxLog, "Test call result for " + mobileEquipment.Result.serial + " " + mobileEquipment.Result.operador + ": " + mobileEquipment.Result.callSuccess, true);

            if (!mobileEquipment.Result.callSuccess && sendAlarms)
            {                      
                await SendEmail.sendAlarmEmailsAsync(libreta, asunto, mensaje);
            }
        }
                      

        UpdateGui.listboxAddItem(listBoxLog, TEST_CALLS_COMPLETED, true);

        return TEST_CALLS_COMPLETED;
    });
}

EDIT: USEFUL INFORMATION FOR THE READER AND LESSONS LEARNED

Following the excellent answers and comments received, I added some originally missing code that may help you to safely interact with Windows Form objects from parallel tasks.

public static void ListboxAddItem(ListBox listBox, String argText, Boolean useTimestamp)
    {
        String timeStamp = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss");

        if (useTimestamp)
        {
            argText = timeStamp + ": " + argText;
        }

        if (Thread.CurrentThread.IsBackground)
        {
            listBox.Invoke(new Action(() =>
            {
                listBox.Items.Add(argText);
                listBox.SelectedIndex = listBox.Items.Count - 1;
            }));
        }
        else
        {
            listBox.Items.Add(argText);
            listBox.SelectedIndex = listBox.Items.Count - 1;
        }
    }

Also, refrain from blindly following IntelliSense suggestions, to prevent shenanigans like Task<Task>, or using Java-like casing on C#.

Is hard to pick the best-proposed answer, because all of them work correctly and without any visible performance difference (the MakePhoneCall().call makes automated phone calls with Android devices connected via ADB). Check which answer works best for your specific application.

Maybe use `Parallel.For`? But be careful with a parallel results.Add. I doubt List can handle parallel adds. To be safe use something from concurrent https://learn.microsoft.com/de-de/dotnet/api/system.collections.concurrent?view=net-5.0 — Klamsi, Oct 22 '21 at 07:49
Whenever you have to write a return type like this: `Task>` it is a good sign that you need to revise your solution. — Peter Csala, Oct 22 '21 at 08:03
Is the `MakePhoneCall().call` method I/O or CPU bound? `Parallel.XYZ` were designed for CPU bound operations. — Peter Csala, Oct 22 '21 at 08:09
Camel-case method names aren't standard in C#. Pascal-case is the norm. — Enigmativity, Oct 22 '21 at 08:10
What does `MakePhoneCall.call()` do? If it makes any kind of IO `Parallel.Invoke` is the *wrong* method. In fact, even with CPU-bound code there are better alternatives like using ActionBlock or Channel — Panagiotis Kanavos, Oct 22 '21 at 11:50

JonasH · Answer 1 · 2021-10-22T08:14:02.087

Parallel.For or Parallel.Foreach and a concurrent collection. should be more appropriate:

ConcurrentStack<Task<MobileEquipment>> results = new ();
Parallel.Foreach(detectedDevicesList, d => results.Add(new MakePhoneCall().call(d, listBoxLog));

Another alternative would be parallel Linq

var result = detectedDevicesList.AsParallel(
    d => results.Add(new MakePhoneCall().call(d, listBoxLog).ToList();

However, it looks like the Call returns a task, so are you sure it is a slow blocking call? If no, it might be better to use a regular loop to start the calls, and use Task.WaitAll to (a)wait for them to complete. It looks like your current solution could block on mobileEquipment.Result.

Also note that listBoxLog looks like a UI object, and accessing UI objects from worker threads is not allowed. Using background threads for processing is much easier if the method is 'Pure' and objects that are immutable. I.e. avoid side effects that may not be thread-safe. As a general rule, I recommend avoiding multi threaded programming unless, a) there is good reason to expect some improvement, b) you are well aware of the dangers of thread safety.

You might also consider using Dataflow to setup a pipeline that does each step of the processing in a parallel and asynchronous way.

The DataFlow library looks excellent. It may require a code rewriting, but is worth it for future projects, — Felipe La Rotta, Oct 22 '21 at 16:43

score 1 · Accepted Answer · answered Oct 22 '21 at 08:21

You should use Microsoft's Reactive Framework (aka Rx) - NuGet System.Reactive and add using System.Reactive.Linq; - then all of your ugly code becomes this:

IObservable<MobileEquipment> query =
    from detectedDevicesList in Observable.FromAsync(() => GetConnectedDevices.getAsync())
    from detectedDevice in detectedDevicesList.ToObservable()
    from mobileEquipment in Observable.FromAsync(() => new MakePhoneCall().call(detectedDevice, listBoxLog))
    select mobileEquipment;

The full method now correctly returns Task<String>, rather than Task<Task<String>>.

Here it is:

public async Task<String> callWithEveryConnectedDevice(ListBox listBoxLog, Boolean sendAlarms)
{
    String TEST_CALLS_COMPLETED = "All test calls completed.";
    String TEST_CALLS_FAILED = "One or more test cals failed";

    IObservable<MobileEquipment> query =
        from detectedDevicesList in Observable.FromAsync(() => GetConnectedDevices.getAsync())
        from detectedDevice in detectedDevicesList.ToObservable()
        from mobileEquipment in Observable.FromAsync(() => new MakePhoneCall().call(detectedDevice, listBoxLog))
        select mobileEquipment;
        
    IList<MobileEquipment> results = await query.ToList();

    if (results.Count == 0)
    {
        UpdateGui.listboxAddItem(listBoxLog, "No devices are connected.", true);
        return TEST_CALLS_FAILED;
    }

    foreach (MobileEquipment mobileEquipment in results)
    {
        UpdateGui.listboxAddItem(listBoxLog, "Test call result for " + mobileEquipment.serial + " " + mobileEquipment.operador + ": " + mobileEquipment.callSuccess, true);

        if (!mobileEquipment.callSuccess && sendAlarms)
        {
            await SendEmail.sendAlarmEmailsAsync(libreta, asunto, mensaje);
        }
    }

    UpdateGui.listboxAddItem(listBoxLog, TEST_CALLS_COMPLETED, true);

    return TEST_CALLS_COMPLETED;
}

The Reactive Framework looks awesome Enigmativity, very much appreciated. I only wonder how could I prevent the query execution, if detectedDevicesList returns a size of 0. My mediocre approach would be adding if(GetConnectedDevices.Conunt>0) but I'm pretty sure Reactive has a more decent way. — Felipe La Rotta, Oct 23 '21 at 18:41
@FelipeLaRotta - If `detectedDevicesList` is empty then the query immediately stops. There's no need to check. That's why I removed it from the code. — Enigmativity, Oct 23 '21 at 22:19

score 0 · Answer 3 · answered Oct 22 '21 at 10:57

The Parallel.Invoke is not the correct tool to use in this case because your workload is asynchronous, and the Parallel.Invoke is not async-friendly. Your problem can be solved by just creating all the CallAsync tasks at once, and then await all of them to complete with the Task.WhenAll method. After the await you are back on the UI thread, and you can update safely the UI with the results.

A handy tool for projecting the detected devices to tasks, is the Select LINQ operator.

public static async Task CallWithEveryConnectedDevice(ListBox listBoxLog, Boolean sendAlarms)
{
    List<MobileEquipment> detectedDevicesList = await GetConnectedDevices.GetAsync();

    Task<MobileEquipment>[] tasks = detectedDevicesList
        .Select(device => new MakePhoneCall().CallAsync(device))
        .ToArray();

    MobileEquipment[] results = await Task.WhenAll(tasks);

    foreach (var mobileEquipment in results)
    {
        UpdateGui.ListboxAddItem(listBoxLog,
            $"Test call result for {mobileEquipment.Serial} {mobileEquipment.Operador}: {mobileEquipment.CallSuccess}", true);
    }

    foreach (var mobileEquipment in results)
    {
        if (!mobileEquipment.CallSuccess && sendAlarms)
        {
            await SendEmail.SendAlarmEmailsAsync(libreta, asunto, mensaje);
        }
    }
}

Add and remove Parallel.Invoke actions during runtime

PROBLEM

TRIED SOLUTIONS

NON WORKING CODE

3 Answers3