AsParallel () and Any()?

Question

I've seen this code which check a condition using AsParallel() and Any() :

bool IsAnyDeviceConnected()
{
   return m_devices.Any(d => d.IsConnected);
}

and to make it faster :

bool IsAnyDeviceConnected()
{
   return m_devices.AsParallel().Any(d => d.IsConnected);
}

But looking at Any() :

 internal static bool Any<T>(this IEnumerable<T> source, Func<T, bool> predicate) {
            foreach (T element in source) {
                if (predicate(element)) {
                    return true;
                }
            }
            return false;
        }

I don't see (obviously) - that it does care about cancellation of other workers - once found.

However - this (other) code does "finish - soon as possible" + cancel other future work :

bool IsAnyDeviceConnected()
{
   var res = Parallel.ForEach(m_devices,
      (d,loopState) => {  
         if (d.IsConnected) 
            loopState.Stop();
      });
   return !res.IsCompleted;
}

Question :

Does my diagnostics correct? Does Any() - once found item , doesn't cancel other threads ( in AsParallel context)

nb , my fear is that I might looking at the wrong source code.

Answer 1

AsParallel() returns a ParallelQuery, so if you call AsParallel().Any(...) you're not calling Enumerable.Any, but ParallelEnumerable.Any.

The reference source code for ParallelEnumerable.Any is here.

When you dig e.g. into the AnyAllSearchOperatorEnumerator class, you see that a flag called resultFoundFlag is used to tell other workers that a result is found so they can stop searching.

Answer 2

You're looking at the wrong code. AsParallel returns a ParallelQuery<TSource>, and ParellelQuery has another overload for Any.

'Any' creates a new AnyAllSearchOperator object and aggregates it. If you dig deeper into that chain of method calls and objects you'll find that the QueryOpeningEnumerator does support cancellation.

---

Unfortunately the reference source links to those particular member functions are bugged.

Answer 3

You are looking at the wrong code. ParallelEnumerable.AsParallel returns a ParallelQuery<>. ParallelEnumerable also defines its own Any extension method.

In order to specify cancellation, degrees of parallelism etc you need to use ParallelEnumerable's WithXXX extension methods like WithCancellation and WithDegreeOfParallelism. ParallelEnumerable.Any doesn't allow you to specify those options to preserve a similar signature as Enumerable.Any