Terminology usage: I will use chunk, page, partition, or subrange interchangeably.
I frequently use and promote using the System.Collections.Concurrent.Partitioner
class (see link). For my particular usage, I always use it for a full indexed collection, i.e. IList<T>
. Sometimes I may use it for single-threaded processing lists with fixed range size per chunk, or I may process in parallel threads where I care more about the number of chunks rather than a specific range size.
Recently I posted an answer using Partitioner
to this CR question. However, it caused me to think of ways I could enhance a partitioner customized more to my needs and preferred usage.
For starters, I wish the boring tuple property names of Item1
and Item2
had more descriptive names such as FromInclusive
and ToExclusive
.
For logging purposes, it would be nice sometimes to know what chunk number I am currently processing. There could be some other nice things to know such as how many total chunks there are or perhaps whether I am on the last chunk.
I almost always work with an item count, meaning the overall range is from 0 to item count – 1
. But the manner in how those chunks are formed and how many chunks there will be depends upon whether I pass in a range size, or a desired chunk count. With Partitioner
, if I want a specific chunk count, I must first use my desired chunk count to determine the applicable range size, which is what Partitioner
is expecting.
CLASS PartitionedIndexRange
I am most interested in feedback or a review of this class.
public class PartitonedIndexRange
{
private PartitonedIndexRange(int fromInclusive, int toExclusive, int partitionIndex, int partitionsCount)
{
this.FromInclusive = fromInclusive;
this.ToExclusive = toExclusive;
this.PartitionIndex = partitionIndex;
this.PartitionsCount = partitionsCount;
}
public int FromInclusive { get; }
public int ToExclusive { get; } = -1;
public int PartitionIndex { get; } = -1;
public int PartitionsCount { get; }
public int LastPartitionIndex => PartitionsCount - 1;
public bool IsFirstPartition => PartitionIndex == 0;
public bool IsLastPartition => PartitionIndex == LastPartitionIndex;
public int Size => ToExclusive - FromInclusive;
public static void CheckCannotBeNegative(string name, int value)
{
if (value < 0)
{
throw new ArgumentOutOfRangeException(name, $"{name} cannot be negative.");
}
}
public static void CheckMustBePositive(string name, int value)
{
if (value <= 0)
{
throw new ArgumentOutOfRangeException(name, $"{name} must be greater than 0.");
}
}
public static void Check2CannotBeLessThan1(string name1, int value1, string name2, int value2)
{
if (value2 < value1)
{
throw new ArgumentOutOfRangeException(name2, $"{name2} cannot be less than {name1}.");
}
}
public static IEnumerable<PartitonedIndexRange> GetPartitionsByRangeSize(int itemCount, int rangeSize)
{
CheckCannotBeNegative(nameof(itemCount), itemCount);
CheckMustBePositive(nameof(rangeSize), rangeSize);
if (itemCount == 0)
{
yield break;
}
int partitionCount = GetChunkSize(itemCount, rangeSize);
foreach (var partition in GetPartitions(0, itemCount, partitionCount, rangeSize))
{
yield return partition;
}
}
public static IEnumerable<PartitonedIndexRange> GetPartitionsByRangeSize(int fromInclusive, int toExclusive, int rangeSize)
{
CheckCannotBeNegative(nameof(fromInclusive), fromInclusive);
Check2CannotBeLessThan1(nameof(fromInclusive), fromInclusive, nameof(toExclusive), toExclusive);
CheckMustBePositive(nameof(rangeSize), rangeSize);
if (toExclusive == fromInclusive)
{
yield break;
}
int partitionCount = GetChunkSize(toExclusive - fromInclusive, rangeSize);
foreach (var partition in GetPartitions(fromInclusive, toExclusive, partitionCount, rangeSize))
{
yield return partition;
}
}
public static IEnumerable<PartitonedIndexRange> GetPartitionsByPartitionCount(int itemCount, int partitionCount)
{
CheckCannotBeNegative(nameof(itemCount), itemCount);
CheckMustBePositive(nameof(partitionCount), partitionCount);
if (itemCount == 0)
{
yield break;
}
int rangeSize = GetChunkSize(itemCount, partitionCount);
foreach (var partition in GetPartitions(0, itemCount, partitionCount, rangeSize))
{
yield return partition;
}
}
public static IEnumerable<PartitonedIndexRange> GetPartitionsByPartitionCount(int fromInclusive, int toExclusive, int partitionCount)
{
CheckCannotBeNegative(nameof(fromInclusive), fromInclusive);
Check2CannotBeLessThan1(nameof(fromInclusive), fromInclusive, nameof(toExclusive), toExclusive);
CheckMustBePositive(nameof(partitionCount), partitionCount);
if (toExclusive == fromInclusive)
{
yield break;
}
int rangeSize = GetChunkSize(toExclusive - fromInclusive, partitionCount);
foreach (var partition in GetPartitions(fromInclusive, toExclusive, partitionCount, rangeSize))
{
yield return partition;
}
}
private static IEnumerable<PartitonedIndexRange> GetPartitions(int fromInclusive, int toExclusive, int partitionCount, int rangeSize)
{
int inclusiveStart = fromInclusive;
for (int partitionIndex = 0; partitionIndex < partitionCount; partitionIndex++)
{
bool isLast = (partitionIndex + 1) == partitionCount;
int exclusiveEnd = isLast ? toExclusive : inclusiveStart + rangeSize;
var partition = new PartitonedIndexRange(inclusiveStart, exclusiveEnd, partitionIndex, partitionCount);
yield return partition;
inclusiveStart = exclusiveEnd;
}
}
private static int GetChunkSize(int itemCount, int inputSize)
{
// <quote>Context means everything.</quote>
// If inputSize is a Range Size, then outputSize is the Partition Count
// If inputSize is a Partition Count, then outputSize is the Range Size
int outputSize = itemCount / inputSize;
return (itemCount % inputSize == 0) ? outputSize : 1 + outputSize;
}
public override string ToString()
{
return $"Partition[{PartitionIndex}] Range [{FromInclusive} - {ToExclusive}) Size {Size}";
}
}
I intentionally made the constructor private restricting usage to the static create methods: GetPartitionsByRangeSize
and GetPartitionsByPartitionCount
.
STATIC CLASS UsageExample
Note I am not interested in any feedback or review on the UsageExample
class.
internal static class UsageExample
{
private static Random _random { get; } = new Random();
public static void RunSimple()
{
// Simple example does not need an actual collection.
// All we really need is to know the size of a collection.
Console.WriteLine();
Console.WriteLine("SIMPLE EXAMPLE using itemCount");
var itemCount = _random.Next(60_000, 200_000);
var rangeSize = 12_500;
var partitionsByRangeSize = PartitonedIndexRange.GetPartitionsByRangeSize(itemCount, rangeSize);
Console.WriteLine();
Console.WriteLine($"BY RANGE SIZE. ItemCount={itemCount} RangeSize={rangeSize}");
foreach (var partition in partitionsByRangeSize)
{
Console.WriteLine(partition);
}
var partitionCount = 9;
var partitionsByPartitionCount = PartitonedIndexRange.GetPartitionsByPartitionCount(itemCount, partitionCount);
Console.WriteLine();
Console.WriteLine($"BY PARTITION COUNT. ItemCount={itemCount} PartitionCount={partitionCount}");
foreach (var partition in partitionsByPartitionCount)
{
Console.WriteLine(partition);
}
}
public static void RunSimple2()
{
Console.WriteLine();
Console.WriteLine("SIMPLE EXAMPLE using toInclusive & fromExclusive");
var fromInclusive = 1_000;
var toExclusive = 12_500;
var rangeSize = 1_000;
var partitionsByRangeSize = PartitonedIndexRange.GetPartitionsByRangeSize(fromInclusive, toExclusive, rangeSize);
Console.WriteLine();
Console.WriteLine($"BY RANGE SIZE. FromInclusive={fromInclusive} ToExclusive={toExclusive} RangeSize={rangeSize}");
foreach (var partition in partitionsByRangeSize)
{
Console.WriteLine(partition);
}
var partitionCount = 10;
var partitionsByPartitionCount = PartitonedIndexRange.GetPartitionsByPartitionCount(fromInclusive, toExclusive, partitionCount);
Console.WriteLine();
Console.WriteLine($"BY PARTITION COUNT. FromInclusive={fromInclusive} ToExclusive={toExclusive} PartitionCount={partitionCount}");
foreach (var partition in partitionsByPartitionCount)
{
Console.WriteLine(partition);
}
}
public static void RunParallel()
{
Console.WriteLine();
Console.WriteLine("PARALLEL EXAMPLE");
var arraySize = _random.Next(88_888, 111_111);
var maxDegreeOfParallelism = Environment.ProcessorCount + 1;
var example = SumArray.Create(arraySize, maxDegreeOfParallelism);
Console.WriteLine();
Console.WriteLine($"Running multi-threaded for ArraySize={arraySize} MaxDegreeOfParallelism={maxDegreeOfParallelism}");
Console.WriteLine("Custom partitioner with guaranteed order will be called.");
Console.WriteLine("Parallel.ForEach order is NOT guaranteed.");
var elapsed = example.RunMultiThreaded();
Console.WriteLine($" Elapsed = {elapsed}");
}
private class SumArray
{
// Modified my answer from this original Code Review post:
// https://codereview.stackexchange.com/questions/270338/sum-of-2-arrays-using-multi-threading/270355#270355
public int[] ArrayA { get; private set; } = new int[0];
public int[] ArrayB { get; private set; } = new int[0];
public int[] ArrayC { get; private set; } = new int[0];
public int MaxDegreeOfParallelism { get; private set; } = 1;
public int ArraySize { get; private set; } = 0;
private SumArray(int size, int maxDegreeOfParallelism)
{
// There are probably more checks that could be done but here are the minimum.
if (size <= 0)
{
throw new ArgumentOutOfRangeException(nameof(size), "Array size must be greater than 0");
}
if (maxDegreeOfParallelism <= 0)
{
throw new ArgumentOutOfRangeException(nameof(maxDegreeOfParallelism), "MaxDegreeOfParallelism must be greater than 0");
}
// While I could call Initialize() here, my philosophy is that a constructor
// should do the BARE MINIMUM, which is only to set some properties and
// then return as quickly as possible.
// To that end, I mark the constructor as private, and require accessing
// it via the public Create().
this.ArraySize = size;
this.MaxDegreeOfParallelism = maxDegreeOfParallelism;
}
public static SumArray Create(int size, int maxDegreeOfParallelism)
{
var instance = new SumArray(size, maxDegreeOfParallelism);
// Initialize is intentionally run after construction.
instance.Initialize();
return instance;
}
private void Initialize()
{
ArrayA = new int[ArraySize];
ArrayB = new int[ArraySize];
ArrayC = new int[ArraySize];
// Magic number replacements.
// https://en.wikipedia.org/wiki/Magic_number_(programming)
// Consider increasing with bigger arrays.
// Consider making parameters or properties rather than constants.
const int minA = 1;
const int maxA = 9999;
const int minB = 100_000;
const int maxB = 900_000;
for (int i = 0; i < ArraySize; i++)
{
// Consider increasing the magic num
ArrayA[i] = _random.Next(minA, maxA);
ArrayB[i] = _random.Next(minB, maxB);
}
}
public TimeSpan RunSingleThreaded()
{
var watch = Stopwatch.StartNew();
for (int i = 0; i < ArraySize; i++)
{
ArrayC[i] = ArrayA[i] + ArrayB[i];
}
watch.Stop();
return watch.Elapsed;
}
public TimeSpan RunMultiThreaded()
{
var watch = Stopwatch.StartNew();
var options = new ParallelOptions() { MaxDegreeOfParallelism = MaxDegreeOfParallelism };
// CUSTOM PARTITIONER CALL HERE
var partitions = PartitonedIndexRange.GetPartitionsByPartitionCount(ArraySize, MaxDegreeOfParallelism);
// While the partitioner sends the partitions in order,
// the task scheduler is not guaranteed to process them in the same order.
Parallel.ForEach(partitions, options, partition =>
{
LockedWriteLine($" START {partition}");
for (var i = partition.FromInclusive; i < partition.ToExclusive; i++)
{
ArrayC[i] = ArrayA[i] + ArrayB[i];
}
LockedWriteLine($" END Partition[{partition.PartitionIndex}]");
});
watch.Stop();
return watch.Elapsed;
}
private object _lockObject = new object();
private void LockedWriteLine(string text)
{
// I know Console.WriteLine is supposed to be thread safe,
// but I am being extra cautious here to remove any doubt.
lock (_lockObject)
{
Console.WriteLine(text);
}
}
}
}
This requires using System.Diagnostics ;
for the Stopwatch
. There is a subclass named SumArray
, which is a modified example of the answer I provided to this POST. The RunMultiThreaded
method uses the custom partitioner. I included some Console.WriteLine
to illustrate when a thread starts and ends, which can produce some very interesting results regarding ordering.
SAMPLE CONSOLE OUTPUT
SIMPLE EXAMPLE
BY RANGE SIZE. ItemCount=77032 RangeSize=12500
Partition[0] Range [0 - 12500) Size 12500
Partition[1] Range [12500 - 25000) Size 12500
Partition[2] Range [25000 - 37500) Size 12500
Partition[3] Range [37500 - 50000) Size 12500
Partition[4] Range [50000 - 62500) Size 12500
Partition[5] Range [62500 - 75000) Size 12500
Partition[6] Range [75000 - 77032) Size 2032
BY PARTITION COUNT. Item Count=77032 PartitionCount=9
Partition[0] Range [0 - 8560) Size 8560
Partition[1] Range [8560 - 17120) Size 8560
Partition[2] Range [17120 - 25680) Size 8560
Partition[3] Range [25680 - 34240) Size 8560
Partition[4] Range [34240 - 42800) Size 8560
Partition[5] Range [42800 - 51360) Size 8560
Partition[6] Range [51360 - 59920) Size 8560
Partition[7] Range [59920 - 68480) Size 8560
Partition[8] Range [68480 - 77032) Size 8552
PARALLEL EXAMPLE
Running multi-threaded for ArraySize=103493 MaxDegreeOfParallelism=7
Custom partitioner with guaranteed order will be called.
Parallel.ForEach order is NOT guaranteed.
START Partition[0] Range [0 - 14785) Size 14785
END Partition[0]
START Partition[1] Range [14785 - 29570) Size 14785
START Partition[4] Range [59140 - 73925) Size 14785
END Partition[4]
START Partition[2] Range [29570 - 44355) Size 14785
END Partition[1]
START Partition[6] Range [88710 - 103493) Size 14783
END Partition[6]
START Partition[3] Range [44355 - 59140) Size 14785
END Partition[3]
START Partition[5] Range [73925 - 88710) Size 14785
END Partition[5]
END Partition[2]
Elapsed = 00:00:00.0364611
Simple Observations: notice the size of the last partition in the Simple example, chunking by Partition Count produces more balanced chunks than by Range Size.
Parallel Observations: although the custom partitioner creates the partitions in a very ordered, sequential manner, the Parallel.ForEach
does not guarantee the order of processing. Note that # 2 starts after # 4 but before # 6, 3 and 5, and that # 2 also is the last to complete. Note as well that # 0 started and ended before the others began.
Granted that’s just from one sample run. I encourage you to run it several times since the results change. Here is output from another run.
SIMPLE EXAMPLE
BY RANGE SIZE. ItemCount=130810 RangeSize=12500
Partition[0] Range [0 - 12500) Size 12500
Partition[1] Range [12500 - 25000) Size 12500
Partition[2] Range [25000 - 37500) Size 12500
Partition[3] Range [37500 - 50000) Size 12500
Partition[4] Range [50000 - 62500) Size 12500
Partition[5] Range [62500 - 75000) Size 12500
Partition[6] Range [75000 - 87500) Size 12500
Partition[7] Range [87500 - 100000) Size 12500
Partition[8] Range [100000 - 112500) Size 12500
Partition[9] Range [112500 - 125000) Size 12500
Partition[10] Range [125000 - 130810) Size 5810
BY PARTITION COUNT. ItemCount=130810 PartitionCount=9
Partition[0] Range [0 - 14535) Size 14535
Partition[1] Range [14535 - 29070) Size 14535
Partition[2] Range [29070 - 43605) Size 14535
Partition[3] Range [43605 - 58140) Size 14535
Partition[4] Range [58140 - 72675) Size 14535
Partition[5] Range [72675 - 87210) Size 14535
Partition[6] Range [87210 - 101745) Size 14535
Partition[7] Range [101745 - 116280) Size 14535
Partition[8] Range [116280 - 130810) Size 14530
PARALLEL EXAMPLE
Running multi-threaded for ArraySize=98424 MaxDegreeOfParallelism=7
Custom partitioner with guaranteed order will be called.
Parallel.ForEach order is NOT guaranteed.
START Partition[1] Range [14061 - 28122) Size 14061
START Partition[0] Range [0 - 14061) Size 14061
START Partition[3] Range [42183 - 56244) Size 14061
END Partition[1]
START Partition[2] Range [28122 - 42183) Size 14061
END Partition[3]
START Partition[4] Range [56244 - 70305) Size 14061
END Partition[0]
START Partition[6] Range [84366 - 98424) Size 14058
END Partition[2]
END Partition[4]
END Partition[6]
START Partition[5] Range [70305 - 84366) Size 14061
END Partition[5]
Elapsed = 00:00:00.0644807
Here is another example of parallel usage where the last partition starts before the first one!
PARALLEL EXAMPLE
Running multi-threaded for ArraySize=93916 MaxDegreeOfParallelism=7
Custom partitioner with guaranteed order will be called.
Parallel.ForEach order is NOT guaranteed.
START Partition[1] Range [13417 - 26834) Size 13417
START Partition[6] Range [80502 - 93916) Size 13414
START Partition[3] Range [40251 - 53668) Size 13417
START Partition[2] Range [26834 - 40251) Size 13417
START Partition[5] Range [67085 - 80502) Size 13417
START Partition[4] Range [53668 - 67085) Size 13417
START Partition[0] Range [0 - 13417) Size 13417
END Partition[1]
END Partition[6]
END Partition[3]
END Partition[2]
END Partition[5]
END Partition[0]
END Partition[4]
Elapsed = 00:00:00.0346445
CONCERNS
While I like UsageExample
as a nice way of demonstrating the unpredictable order of parallel tasks, I have little concern about a review of its code. My primary concern is a review of the PartitionedIndexRange
class. So far, I like it enough to have it find a welcome on my developer’s toolkit. Having meta data about the partition is nice from a logging perspective in my many applications, most of which are Console apps running unattended via Windows Task Scheduler.
I chose to use longer descriptive names for the meta data to clarify the context of the indices. There are the pair of FromInclusive
and ToExclusive
, which do not contain the word “Index” and save for the capitalization match the old partitioner names. For the meta properties, I could have gone with “Id” or “Index” but chose the longer “PartitionIndex” to be clear. Ditto for “PartitionCount” instead of “Count”.