When to use Parallel.ForEach and when to PLINQ
- Transfer
Introduction
Usually, when optimizing a program for multi-core computers, the first step is to find out the possibility of dividing the algorithm into parts that run in parallel. If in order to solve the problem it is necessary to process separate elements from a large data set in parallel, the first candidates will be new parallel features in the .NET Framework 4: Parallel.ForEach and Parallel LINQ ( PLINQ )
Parallel.ForEach
The Parallel class contains the ForEach method , which is a multithreaded version of a regular foreach loop in C #. Like regular foreach, Parallel.ForEach iterates over enumerable data, but using multiple threads. One of the more commonly used Parallel.ForEach overloads is as follows:
public static ParallelLoopResult ForEach(
IEnumerable source,
Action body)
Ienumerable indicates the sequence to be iterated over, and the Action body sets the delegate to invoke for each element. A complete list of Parallel.ForEach overloads can be found here .
Plinq
Related to Parallel.ForEach PLINQ is a programming model for parallel data operations. The user defines an operation from a standard set of operators, including projections, filters, aggregation, etc. Like Parallel.ForEach, PLINQ achieves parallelism by breaking the input sequence into parts and processing elements in different threads.
The article highlights the differences between these two approaches to parallelism. Understanding usage scenarios in which it is best to use Parallel.ForEach instead of PLINQ and vice versa.
Independent operations
If you need to perform lengthy calculations on the elements of a sequence and the results are independent, it is preferable to use Parallel.ForEach. PLinq, in turn, will be too heavy for such operations. In addition, the maximum number of threads is indicated for Parallel.ForEach , that is, if ThreadPool has few resources and fewer threads are available than specified in ParallelOptions.MaxDegreeOfParallelism , the optimal number of threads will be used, which can increase as it runs. For PLINQ, the number of threads to execute is strictly specified.
Parallel operations with data order preservation
PLINQ to keep order
If your conversions require preserving the input order, then you will most likely find it easier to use PLINQ than Parallel.ForEach . For example, if we want to convert RGB color video frames to black and white, at the output, the frame order should naturally be preserved. In this case, it is better to use PLINQ and the AsOrdered () function , which in the depths of PLINQ partitions the input sequence, performs conversions, and then arranges the result in the correct order.
public static void GrayscaleTransformation(IEnumerable Movie)
{
var ProcessedMovie =
Movie
.AsParallel()
.AsOrdered()
.Select(frame => ConvertToGrayscale(frame));
foreach (var grayscaleFrame in ProcessedMovie)
{
// Movie frames will be evaluated lazily
}
}
Why not use Parallel.ForEach here?
Except in trivial cases, implementing parallel operations on serial data using Parallel.ForEach requires a significant amount of code. In our case, we can use the overload of the Foreach function to repeat the effect of the AsOrdered () operator:
public static ParallelLoopResult ForEach(
IEnumerable source,
Actionbody)
In the overloaded version of Foreach, the index parameter of the current element has been added to the data action delegate. Now we can write the result to the output collection at the same index, make costly calculations in parallel, and finally get the output sequence in the correct order. The following example illustrates one way to maintain order with Parallel.ForEach :
public static double [] PairwiseMultiply( double[] v1, double[] v2)
{
var length = Math.Min(v1.Length, v2.Lenth);
double[] result = new double[length];
Parallel.ForEach(v1,
(element, loopstate, elementIndex) =>
result[elementIndex] = element * v2[elementIndex]);
return result;
}
However, shortcomings of this approach are immediately discovered. If the input sequence is an IEnumerable type and not an array, then there are 4 ways to implement order preservation:
- The first option is to call IEnumerable.Count (), which will cost O (n). If the number of elements is known, you can create an output array to save the results at a given index
- The second option is to materialize the collection (by turning it, for example, into an array). If there is a lot of data, then this method is not very suitable.
- The third option is to think carefully about the output collection. The output collection may be a hash, then the amount of memory required to store the output value will be at least 2 times the input memory in order to avoid hash collisions; if there is a lot of data, the data structure for the hash will be prohibitively large, in addition, you can get a performance drop due to false sharing and the garbage collector.
- And the last option is to save the results with their original indexes, and then apply your own algorithm for sorting the output collection.
In PLINQ, the user simply asks for the preservation of the order, and the query engine manages all the routine details of ensuring the correct order of the results. The PLINQ framework allows the AsOrdered () operator to process streaming data, in other words, PLINQ supports lazy materialization. In PLINQ, materializing the entire sequence is the worst solution, you can easily avoid the above problems and perform parallel operations on the data simply using the AsOrdered () operator .
Parallel Streaming
Using PLINQ to process a stream
PLINQ offers the ability to process a request as a request over a stream. This feature is extremely valuable for the following reasons:
- 1. The results do not materialize in the array, so there is no redundancy in storing data in memory.
- 2. You can enumerate results in a single stream of computation as you receive new data.
Continuing with the example of the analysis of securities, imagine that you want to calculate the risk of each paper from a portfolio of securities, giving out only securities that meet the criteria for risk analysis, and then perform some calculations on the filtered results. In PLINQ, the code will look something like this:
public static void AnalyzeStocks(IEnumerable Stocks)
{
var StockRiskPortfolio =
Stocks
.AsParallel()
.AsOrdered()
.Select(stock => new { Stock = stock, Risk = ComputeRisk(stock)})
.Where(stockRisk => ExpensiveRiskAnalysis(stockRisk.Risk));
foreach (var stockRisk in StockRiskPortfolio)
{
SomeStockComputation(stockRisk.Risk);
// StockRiskPortfolio will be a stream of results
}
}
In this example, the elements are distributed into parts ( partitions ), processed by several threads, then reordered; it is important to understand that these steps are performed in parallel, as filtering results appear, a single-threaded consumer in the foreach loop can perform calculations. PLINQ is optimized for performance, not latency, and uses buffers internally; it may happen that although a partial result has already been obtained, it will remain in the output buffer until the output buffer is completely saturated and does not allow further processing. The situation can be corrected using the PLINQ WithMergeOptions extension method , which allows you to specify output buffering. MethodWithMergeOptions accepts the ParallelMergeOptions enumeration as a parameter , you can specify how the query returns the final result that will be used by a single stream. The following options are offered:
- ParallelMergeOptions.NotBuffered - Indicates that each processed item is returned from each thread as soon as it is processed.
- ParallelMergeOptions.AutoBuffered - indicates that the elements are collected in the buffer, the buffer is periodically returned to the consumer stream
- ParallelMergeOptions.FullyBuffered - indicates that the output sequence is fully buffered, this allows you to get results faster than using other options, but then the consumer thread will have to wait a long time to receive the first element for processing.
UsingMergeOptions example available on MSDN
Why not Parallel.ForEach?
Put aside the shortcomings of Parallel.ForEach to preserve the order of the sequence. For unordered computations over a stream using Parallel.ForEach, the code will look like this:
public static void AnalyzeStocks(IEnumerable Stocks)
{
Parallel.ForEach(Stocks,
stock => {
var risk = ComputeRisk(stock);
if(ExpensiveRiskAnalysis(risk)
{
// stream processing
lock(myLock) { SomeStockComputation(risk) };
// store results
}
}
This code is almost identical to the PLINQ example, with the exception of explicit blocking and less elegant code. Note that in this situation, Parallel.ForeEach means storing the results in a thread-safe style, while PLINQ does it for you.
To save the results, we have 3 ways: the first is to save the values in a stream-unsafe collection and require a lock on each record. The second is to save it to a thread-safe collection, fortunately, the .NET Framework 4 provides a set of such collections in the System.Collections.Concurrent namespace and you don’t have to implement it yourself. The third way is to use Parallel.ForEach with thread-localstorage, which will be described later. Each of these methods requires explicit control of the third-party effects of writing to the collection, while PLINQ allows us to abstract from these operations.
Operations on two collections
Using PLINQ for operations on two collections
The PLINQ ZIP operator specifically performs parallel computations on two different collections. Since it can be combined with other queries, you can simultaneously perform complex operations on each collection before combining the two collections. For instance:
public static IEnumerable Zipping(IEnumerable a, IEnumerable b)
{
return
a
.AsParallel()
.AsOrdered()
.Select(element => ExpensiveComputation(element))
.Zip(
b
.AsParallel()
.AsOrdered()
.Select(element => DifferentExpensiveComputation(element)),
(a_element, b_element) => Combine(a_element,b_element));
}
The example above demonstrates how each data source is processed in parallel by different operations, then the results from both sources are combined by the Zip operator.
Why not Parallel.ForEach?
A similar operation can be performed with Parallel.ForEach overload using indexes, for example:
public static IEnumerable Zipping(IEnumerable a, IEnumerable b)
{
var numElements = Math.Min(a.Count(), b.Count());
var result = new T[numElements];
Parallel.ForEach(a,
(element, loopstate, index) =>
{
var a_element = ExpensiveComputation(element);
var b_element = DifferentExpensiveComputation(b.ElementAt(index));
result[index] = Combine(a_element, b_element);
});
return result;
}
However, there are potential traps and shortcomings described in the application of Parallel.ForEach with preservation of the data order, one of the disadvantages includes viewing the entire collection to the end and explicit index management.
The local flow condition ( Thread-Local State )
Using Parallel.ForEach to Access the Local State of a Stream
Although PLINQ provides more concise means for parallel operations on data, some processing scenarios are better suited to using Parallel.ForEach , for example, operations that support the local state of a stream. The signature of the corresponding Parallel.ForEach method looks like this:
public static ParallelLoopResult ForEach(
IEnumerable source,
Func localInit,
Func body,
Action localFinally)
It should be noted that there is an overload of the Aggregate operator , which allows access to the local state of the stream and can be used if the data processing template can be described as a decrease in dimension. The following example illustrates how to exclude numbers that are not prime from a sequence:
public static List Filtering(IEnumerable source)
{
var results = new List();
using (SemaphoreSlim sem = new SemaphoreSlim(1))
{
Parallel.ForEach(source, () => new List(),
(element, loopstate, localStorage) =>
{
bool filter = filterFunction(element);
if (filter)
localStorage.Add(element);
return localStorage;
},
(finalStorage) =>
{
lock(myLock)
{
results.AddRange(finalStorage)
};
});
}
return results;
}
Such functionality could be achieved much more easily with PLINQ, the purpose of the example is to show that using Parallel.ForEach and the local state of the stream can greatly reduce synchronization costs. However, in other scenarios, local flow states become absolutely necessary; the following example demonstrates such a scenario.
Imagine that you, as a brilliant computer scientist and mathematician, have developed a statistical model for analyzing securities risks; this model you think will break all other risk models to the nines. In order to prove this, you need data from sites with information about stock markets. But loading the data sequence will be very long and is a bottleneck for an eight-core computer. Although useParallel.ForEach is an easy way to load data in parallel using WebClient , each stream will be blocked at every download, which can improve the use of asynchronous I / O; more information is available here . For performance reasons, you decided to use Parallel.ForEach to iterate through the collection of URLs and upload data in parallel. The code looks something like this:
public static void UnsafeDownloadUrls ()
{
WebClient webclient = new WebClient();
Parallel.ForEach(urls,
(url,loopstate,index) =>
{
webclient.DownloadFile(url, filenames[index] + ".dat");
Console.WriteLine("{0}:{1}", Thread.CurrentThread.ManagedThreadId, url);
});
}
Surprisingly, we get an exception at runtime: “System.NotSupportedException -> WebClient does not support concurrent I / O operations.” Having realized that many threads cannot access the same WebClient at the same time, you decide to create a WebClient for every download.
public static void BAD_DownloadUrls ()
{
Parallel.ForEach(urls,
(url,loopstate,index) =>
{
WebClient webclient = new WebClient();
webclient.DownloadFile(url, filenames[index] + ".dat");
Console.WriteLine("{0}:{1}", Thread.CurrentThread.ManagedThreadId, url);
});
}
This code allows the program to create more than a hundred web clients; the program will throw a timeout exception in WebClient. You will understand that the computer is not running a server operating system, so the maximum number of connections is limited. Then you can guess that using Parallel.ForEach with the local state of the stream will solve the problem:
public static void downloadUrlsSafe()
{
Parallel.ForEach(urls,
() => new WebClient(),
(url, loopstate, index, webclient) =>
{
webclient.DownloadFile(url, filenames[index]+".dat");
Console.WriteLine("{0}:{1}", Thread.CurrentThread.ManagedThreadId, url);
return webclient;
},
(webclient) => { });
}
}
In this implementation, each data access operation is independent of another. At the same time, the access point is neither independent nor thread safe. Using local stream storage allows us to be sure that the number of WebClient instances created is as many as required, and each WebClient instance belongs to the stream that created it.
Why is PLINQ bad here?
If you implement the previous example using ThreadLocal and PLINQ objects, the code is as follows:
public static void downloadUrl()
{
var webclient = new ThreadLocal(()=> new WebClient ());
var res =
urls
.AsParallel()
.ForAll(
url =>
{
webclient.Value.DownloadFile(url, host[url] +".dat"));
Console.WriteLine("{0}:{1}",
Thread.CurrentThread.ManagedThreadId, url);
});
}
While the implementation achieves the same goals, it is important to understand that in any scenario, using ThreadLocal <> is significantly more expensive than the corresponding Parallel.ForEach overload . Note that in this scenario, the cost of creating ThreadLocal <> instances is negligible compared to the time it took to download a file from the Internet.
Exit operations
Using Parallel.ForEach to exit operations
In a situation where control over the execution of operations is essential, it is important to understand that exiting the Parallel.ForEach cycle allows you to achieve the same effect as checking the conditions for the need to continue computing inside the body of the cycle. One of the Parallel.ForEach overloads that allow you to track ParallelLoopState looks like this:
public static ParallelLoopResult ForEach(
IEnumerable source,
Action body)
ParallelLoopState provides support for interrupting loop execution in two different ways, described below.
ParallelLoopState.Stop ()
Stop () informs the loop about the need to stop iterations; The ParallelLoopState.IsStopped property allows each iteration to determine whether some other iteration has called the Stop () method . The Stop () method is usually useful if the loop performs an unordered search and should exit as soon as the item is found. For example, if we want to find out if an object is present in the collection, the code will look something like this:
public static boolean FindAny(IEnumerable TSpace, T match) where T: IEqualityComparer
{
var matchFound = false;
Parallel.ForEach(TSpace,
(curValue, loopstate) =>
{
if (curValue.Equals(match) )
{
matchFound = true;
loopstate.Stop();
}
});
return matchFound;
}
Functionality can also be achieved using PLINQ, this example demonstrates how to use ParallelLoopState.Stop () to control the flow of execution.
ParallelLoopState.Break ()
Break () informs the loop that the elements preceding the current element should be processed, but for subsequent elements of the iteration it is necessary to stop. The lower iteration value can be obtained from the ParallelLoopState.LowestBreakIteration property . Break () is usually useful if you are searching through ordered data. In other words, there is a certain criterion for the need for data processing. For example, for a sequence containing non-unique elements in which it is necessary to find the lower index of a matching object, the code will look like this:
public static int FindLowestIndex(IEnumerable TSpace, T match) where
T: IEqualityComparer
{
var loopResult = Parallel.ForEach(source,
(curValue, loopState, curIndex) =>
{
if (curValue.Equals(match))
{
loopState.Break();
}
});
var matchedIndex = loopResult.LowestBreakIteration;
return matchedIndex.HasValue ? matchedIndex : -1;
}
In this example, the loop is executed until an object is found, the Break () signal means that only elements with a lower index than the found object should be processed; if one more matching instance is found, the Break () signal will be received again, this is repeated until there are elements, if the object was found, the LowestBreakIteration field points to the first index of the matching object.
Why not PLINQ?
Although PLINQ provides support for quitting query execution, the differences in the exit mechanisms of PLINQ and Parallel.ForEach are significant. In order to exit the PLINQ request, the request must be provided with a cancellation token, as described here . C Parallel.ForEach exit flags are polled at each iteration. In the case of PLINQ, you cannot rely on a canceled request to stop quickly.
Conclusion
Parallel.ForEach and PLINQ are powerful tools for quickly introducing concurrency in your applications without the need for deep immersion in the mechanisms of their work. However, to choose the right tool for solving a specific problem, remember the differences and tips described in this article.
Useful links:
Threading in C #
RSDN: Work with threads in C #. Parallel Programming
Microsoft Samples for Parallel Programming with the .NET Framework
