방법: 고정 개수의 파티션을 사용하여 파티셔너 구현
다음 예제에서는 정적 분할을 수행하는 PLINQ에 대한 간단한 사용자 지정 파티셔너를 구현하는 한 가지 방법을 보여 줍니다. 이 파티셔너는 동적 파티션을 지원하지 않기 때문에 Parallel.ForEach에서 사용할 수 없습니다. 이 파티셔너는 각 요소의 처리에 많은 시간이 소요되는 데이터 소스에 대해 기본 범위 파티셔너보다 빠른 속도를 제공합니다.
예제
// A static range partitioner for sources that require
// a linear increase in processing time for each succeeding element.
// The range sizes are calculated based on the rate of increase
// with the first partition getting the most elements and the
// last partition getting the least.
class MyPartitioner : Partitioner<int>
{
int[] source;
double rateOfIncrease = 0;
public MyPartitioner(int[] source, double rate)
{
this.source = source;
rateOfIncrease = rate;
}
public override IEnumerable<int> GetDynamicPartitions()
{
throw new NotImplementedException();
}
// Not consumable from Parallel.ForEach.
public override bool SupportsDynamicPartitions
{
get
{
return false;
}
}
public override IList<IEnumerator<int>> GetPartitions(int partitionCount)
{
List<IEnumerator<int>> _list = new List<IEnumerator<int>>();
int end = 0;
int start = 0;
int[] nums = CalculatePartitions(partitionCount, source.Length);
for (int i = 0; i < nums.Length; i++)
{
start = nums[i];
if (i < nums.Length - 1)
end = nums[i + 1];
else
end = source.Length;
_list.Add(GetItemsForPartition(start, end));
// For demonstratation.
Console.WriteLine("start = {0} b (end) = {1}", start, end);
}
return (IList<IEnumerator<int>>)_list;
}
/*
*
*
* B
// Model increasing workloads as a right triangle / |
divided into equal areas along vertical lines. / | |
Each partition is taller and skinnier / | |
than the last. / | | |
/ | | |
/ | | |
/ | | | |
/ | | | |
A /______|____|___|__| C
*/
private int[] CalculatePartitions(int partitionCount, int sourceLength)
{
// Corresponds to the opposite side of angle A, which corresponds
// to an index into the source array.
int[] partitionLimits = new int[partitionCount];
partitionLimits[0] = 0;
// Represent total work as rectangle of source length times "most expensive element"
// Note: RateOfIncrease can be factored out of equation.
double totalWork = sourceLength * (sourceLength * rateOfIncrease);
// Divide by two to get the triangle whose slope goes from zero on the left to "most"
// on the right. Then divide by number of partitions to get area of each partition.
totalWork /= 2;
double partitionArea = totalWork / partitionCount;
// Draw the next partitionLimit on the vertical coordinate that gives
// an area of partitionArea * currentPartition.
for (int i = 1; i < partitionLimits.Length; i++)
{
double area = partitionArea * i;
// Solve for base given the area and the slope of the hypotenuse.
partitionLimits[i] = (int)Math.Floor(Math.Sqrt((2 * area) / rateOfIncrease));
}
return partitionLimits;
}
IEnumerator<int> GetItemsForPartition(int start, int end)
{
// For demonstration purpsoes. Each thread receives its own enumerator.
Console.WriteLine("called on thread {0}", Thread.CurrentThread.ManagedThreadId);
for (int i = start; i < end; i++)
yield return source[i];
}
}
class Consumer
{
public static void Main2()
{
var source = Enumerable.Range(0, 10000).ToArray();
Stopwatch sw = Stopwatch.StartNew();
MyPartitioner partitioner = new MyPartitioner(source, .5);
var query = from n in partitioner.AsParallel()
select ProcessData(n);
foreach (var v in query) { }
Console.WriteLine("Processing time with custom partitioner {0}", sw.ElapsedMilliseconds);
var source2 = Enumerable.Range(0, 10000).ToArray();
sw = Stopwatch.StartNew();
var query2 = from n in source2.AsParallel()
select ProcessData(n);
foreach (var v in query2) { }
Console.WriteLine("Processing time with default partitioner {0}", sw.ElapsedMilliseconds);
}
// Consistent processing time for measurement purposes.
static int ProcessData(int i)
{
Thread.SpinWait(i * 1000);
return i;
}
}
이 예제의 파티션은 각 요소에 대한 처리 시간이 선형적으로 증가한다는 가정을 기반으로 합니다. 실제로는 처리 시간을 이러한 방식으로 예측하기 어려울 수 있습니다. 정적 파티셔너를 특정 데이터 소스와 함께 사용할 경우에는 해당 소스에 대한 분할 형식을 최적화하고 부하 분산 논리를 추가거나 방법: 동적 파티션 구현에서 설명된 청크 분할 방법을 사용할 수 있습니다.