11
\$\begingroup\$

I have a class that tests a list of proxies for validity concurrently, which I have created in 2 different ways, one utilizing Parallel.ForEach and the other using a TPL Dataflow ActionBlock.

I am just wondering which is one is preferred, and if I am overlooking something important. As a side note, I created the Dataflow version because I want to be able to queue proxy tests, with the Parallel.ForEach version I could probably achieve that but it would take more code and some refactoring.

I still need some add some functionality to these methods. I realize the 'tests' parameter isn't being used.

Parallel.ForEach version:

public class ProxyTester
{
    public async Task ValidateProxiesAsync(IEnumerable<Proxy> proxies, IEnumerable<ProxyJudge> judges, IEnumerable<ProxyTest> tests)
    {
        await Task.Run(async () =>
        {
            foreach (var proxy in proxies)
            {
                proxy.Status = ProxyStatus.Queued;
            }
            await proxies.ProcessWithParallelAsync(new Action<Proxy>((proxy =>
            {
                proxy.Status = ProxyStatus.Testing;
                proxy.Status = proxy.TestValidity(judges.ElementAt(0));
            })));
        });
    }
}

ProcessWithParallelAsync Extension method:

public static async Task ProcessWithParallelAsync<T>(this IEnumerable<T> list, Action<T> action, int maxConcurrency = 100)
{
    await Task.Run(() => {
        var options = new ParallelOptions();
        options.MaxDegreeOfParallelism = maxConcurrency;
        Parallel.ForEach(list, options, action);
    });
}

TPL Dataflow version (supports queuing):

class ProxyTester2
{

    private ActionBlock<(Proxy, IProxyTest)> actionBlock;

    public ProxyTester2()
    {
        actionBlock = new ActionBlock<(Proxy proxy, IProxyTest test)>(tup =>
        {
            tup.proxy.Status = ProxyStatus.Testing;
            tup.proxy.Status = tup.proxy.TestValidity(tup.test);
        }, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 100 });
    }

    public void QueueProxyTest(Proxy proxy, IProxyTest test)
    {
        proxy.Status = ProxyStatus.Queued;
        actionBlock.Post((proxy, test));
    }

    public void QueueProxyTests(IEnumerable<Proxy> proxies, IEnumerable<ProxyJudge> judges, IEnumerable<ProxyTest> tests)
    {
        foreach (var proxy in proxies)
        {
            QueueProxyTest(proxy, judges.ElementAt(0));
        }
    }
}
\$\endgroup\$
1
  • 1
    \$\begingroup\$ What is your reason behind setting MaxDegreeOfParallelism to 100? Parallel.ForEach should be capable of queuing your actions very well. Provided there are more items in the enumeration than maximum number of actions that can be executed in parallel. \$\endgroup\$ – sbecker May 8 '18 at 7:39
1
\$\begingroup\$

You are starting two Task in ValidateProxiesAsync - one in the method and another in the extension method. Really seems it should be one Task.

Also judges is an IEnumerable that you are constantly doing ElementAt. Should either do First and store that value for each call or pass in an IList or Array is that's what it really is if you need to access it by element

As for Parallel.ForEach vs TPL DataFlow I think that depends on if you are building a data flow mesh (TPL DataFlow) or just want to process items in Parallel only (Parallel.ForEach)

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.