要素数1千万のint型配列(値はランダム)から、最大値を1000回求めた時の処理時間は、MathクラスよりVectorクラスのほうが2倍以上速い結果になった
青色グラフはシングルスレッド
赤色グラフは配列を分割してマルチスレッド
緑色グラフは1000回ループを分割してマルチスレッド
環境は
OS Window 10 64bit
CPU Ryzen 5 2400G 4コア8スレッド
MEM DDR4-2666
計測アプリ
ダウンロード
20200125_VectorMinMax.zip
ギットハブ
github.com
アプリの操作
取得(ループ)回数を設定して、それぞれのボタンを押して待つだけ
結果
ループなし(回数1)のときの結果
ループ回数1000回の結果
仕様
右上の4つのボタンのうちLinqAsParallelMaxを除く3つはCPUによっては間違った値が表示されることがある。配列の分割をCPUのスレッド数で割っていて、割り切れなかった場合のあまりの要素を処理していないから。もしその要素に最大値が入っていた場合は、間違った値が表示される。要素数100にして、例えばスレッド数が3のCPUだと100/3=33…余り1で、この最後の1個は無視される、6コア12スレッドCPUだと100/12=8…余り4で最後から4つの要素が無視される。
System.Numerics.Vector
docs.microsoft.com
.NET Core3.1なら普通に使えるので便利、.NET Frameworkだと参照の追加が必要
Vector作成、Vector<int>でint型を扱うのができる
中は配列みたいになっている?
Countプロパティ
中に入る値の個数が返ってくる、これはOSやCPU環境によって違ってくるのかも?今の環境では8が返ってきた、これ大事
値1つから作成
1つから作ったけど中には同じ値が4つ並んで、その後の4つは0?よくわからん
配列から作成1
要素数がCountプロパティで返ってきた8未満だとエラーで作成できなかった、8より多い場合はエラーにはならないけど、どうなっているのかなと中身が気になる。中の値は普通の配列みたいに取り出せたので見てみると
頭から8個の値が入っていた。9番目からは無視されているようで、インデックス8を指定した場合はエラーになった
配列から作成2
指定したインデックスの要素から作成
11個の要素の配列とインデックス2を指定、指定通り途中の要素から8個分で作成された
Maxメソッド
2つのVectorを渡すと、要素のペアごとに比較して大きい方で作ったVectorを返す?使ってみると
{ 1, 2, 3, 4, 5, 6, 7, 8 }
{ 1, 1, 1, 1, 100, 1, 1, 1 }
この2つの比較の結果は
{ 1, 2, 3, 4, 100, 6, 7, 8 }
こうなった、なるほどねえ
あとは結果の8個の中からの最大値は普通のループで求めればいい
できた
System.Numerics.Vector.Maxを使って配列から最大値を求める
270行目、Countプロパティを求める、今の環境では8だった
271行目のmyLastは目印、10,000,000 - 8 = 9,999,992
272行目、Vector<int>をintの一番小さい値、int.MinValueで作成、これに配列の要素8個づつから作成したVector<int>をMaxメソッドで比較していくのが273行目のFor
できあがったVector<int>の中の8つの値の最大値を比較が278行目のFor
282行目のForはCountプロパティで割り切れなかった余りの要素との比較
ForIf
普通に不等号で比較、意外に遅くない、今まではこの方法を使っていた
MathMax
意外に速くない、不等号で比較より誤差程度だけど遅い。
LinqMax
遅い、今回の計測の中では一番遅いけど、画像処理とかじゃなくて普通のアプリなら十分な速度だし、書くのもラク、1行で済む
VectorMax
Vectorってベクトルの計算に使うものみたいなので、今回のような使い方があっているのかわかんないけど、速い!けどコードが長くなる、けどやっぱり速い!
ここまではシングルスレッド
ここからはマルチスレッド
ForIfMT
配列をCPUのスレッド数で分割して、それぞれをTask(よくわかってない)を使って並列処理
10,000,000 / 8=1,250,000なので、1250000要素の配列8個を並列処理
スレッド数取得はEnvironment.ProcessorCountプロパティ、これはCPUの論理コア数になるみたい
Task.Runの中でさっきのシングルスレッドのForIfを181行目で呼び出している。
マルチスレッドだからこちらのほうが速くなると思ってたんだけど結果は逆で
3倍以上も遅くなってしまった、Taskの使い方や書き方が間違っているのかわからんけど遅い
処理中のCPU使用率は50%前後、本当のマルチスレッドなら100%になるはずなんだけどねえ
MathMaxMT
同じように配列分割してのマルチスレッドをMathクラスのMaxメソッドを使ったもの、結果も同じく3倍遅くなった
LinqMaxAsParallel
これはLinqのAsParallelのMaxメソッドを使ったもの、Maxの前にAsParallel()って書くだけでマルチスレッドで処理してくれる、とてもラク。処理時間は速くないけど、AsParallelじゃないMaxに比べると46.6/11.01=4.23...4倍以上も速くなった!コア数多いCPUならもっと差が出るはず。素晴らしい
この時のCPU使用率
だいたい85%意外に100%にならないんだなあ
VectorMaxMT
これもMathMaxMTのように遅くなってしまった
10.22/2=5.11、5倍以上も時間かかった、やっぱり分割の部分が比較部分より重たいのかなあ
ここまでは配列の分割して、それぞれを並列処理
ここからは配列はそのままで、取得回数のループをParallelクラスを使っての並列処理。最大値を求めるなら1回の取得でいいんだから意味のないことなんだけど、並列処理(マルチスレッド)で速くなるところを見たかった
シングルスレッド処理の最大値を求めるメソッドを取得回数分ループさせているだけなので、ForをParallel.Forに書き換えただけ
3倍以上速くなった、4コアCPUでこれだと16コアCPUなら10倍以上速くなるんだろうなあ
シングルスレッドのMathMaxからマルチスレッドのVectorMaxだと
5.77 / 0.6 ≒ 9.6で9倍以上!
最小値も同時に求める
Vector.Minメソッドを使って最小値も同時に求めてみた結果、0.66秒
最大値だけのときは0.60秒なので差は0.06秒と誤差程度しか遅くなっていない、ってことはやっぱり比較部分(Vector)はめちゃくちゃ速い
右上3つのボタン
10,000,001要素の配列の最後の要素にint型の最大値、int.MaxValueの2147483647を入れて処理した結果
右上3つのボタンの最大値の結果が間違っているのは
10,000,001をCPUスレッド数8で割った余りは1、この余りの要素を比較処理していないからこうなる。処理するように直してもこの3つは遅くて使う意味がないのでこのまま
日記
この前の距離を求めるVectorクラスのDistanceメソッドが速かったから、同じVectorクラスで最大値を求めるっぽいMaxメソッドがVectorじゃなくて普通のintやbyte型に使えないかなあと思ってぐぐってみたら、できそうな出来なさそうな感じだったので試してみたらできた、と思うけど実際どうなんだろうねえ。
あと速くしたい配列処理は、平均値、中央値、分散、同じ値のカウント、これらもSIMDを使うSystem.Numericsクラスでできるのがあればいいなあ
参照したところ
配列の中から指定した範囲の要素を抜き出す - .NET Tips(VB.NET, C#...)
dobon.net
SIMDのVector処理を試してみる: 第十四工房
http://c5d5e5.asablo.jp/blog/2017/09/27/8685145
Vector, System.Numerics C# (CSharp) Code Examples - HotExamples
https://csharp.hotexamples.com/examples/System.Numerics/Vector/-/php-vector-class-examples.html
[WPF]DependencyObject の子孫要素を型指定ですべて列挙する
mseeeen.msen.jp
コード全部
<Window xClass="_20200125_VectorMinMax.MainWindow"
xmlns="http://schemas.microsoft.com/winfx/2006/xaml/presentation"
xmlnsx="http://schemas.microsoft.com/winfx/2006/xaml"
xmlnsd="http://schemas.microsoft.com/expression/blend/2008"
xmlnsmc="http://schemas.openxmlformats.org/markup-compatibility/2006"
xmlnslocal="clr-namespace:_20200125_VectorMinMax"
mcIgnorable="d"
Title="MainWindow" Height="520" Width="600">
<WindowResources>
<Style TargetType="StackPanel">
<Setter Property="Margin" Value="4"/>
</Style>
<Style TargetType="Button">
<Setter Property="Width" Value="140"/>
</Style>
</WindowResources>
<Grid>
<GridColumnDefinitions>
<ColumnDefinition/>
<ColumnDefinition/>
</GridColumnDefinitions>
<StackPanel>
<StackPanel Orientation="Horizontal">
<Button xName="ButtonForIf" Content="ForIf"/>
<TextBlock xName="tbForIf"/>
</StackPanel>
<StackPanel Orientation="Horizontal">
<Button xName="ButtonMathMax" Content="MathMax"/>
<TextBlock xName="tbMathMax"/>
</StackPanel>
<StackPanel Orientation="Horizontal">
<Button xName="ButtonLinqMax" Content="LinqMax"/>
<TextBlock xName="tbLinqMax"/>
</StackPanel>
<StackPanel Orientation="Horizontal">
<Button xName="ButtonVectorMax" Content="VectorMax"/>
<TextBlock xName="tbVectorMax"/>
</StackPanel>
<Border Height="1" Background="Orange" UseLayoutRounding="True" Margin="0,0,0,4"/>
<Button xName="ButtonRandomArray" Content="配列リセット"/>
<Button xName="ButtonSetLoopCount1" Content="ループ回数1にする"/>
<Button xName="ButtonSetLoopCount10" Content="ループ回数10にする"/>
<Button xName="ButtonSetLoopCount100" Content="ループ回数100にする"/>
<Button xName="ButtonSetLoopCount1000" Content="ループ回数1000にする"/>
<StackPanel Orientation="Horizontal">
<Button xName="ButtonVectorMinMaxMT2" Content="VectorMinMaxMT2"/>
<TextBlock xName="tbVectorMinMaxMT2"/>
</StackPanel>
</StackPanel>
<StackPanel GridColumn="1">
<StackPanel Orientation="Horizontal">
<Button xName="ButtonForIfMT" Content="ForIfMT"/>
<TextBlock xName="tbForIfMT"/>
</StackPanel>
<StackPanel Orientation="Horizontal">
<Button xName="ButtonMathMaxMT" Content="MathMaxMT"/>
<TextBlock xName="tbMathMaxMT"/>
</StackPanel>
<StackPanel Orientation="Horizontal">
<Button xName="ButtonLinqAsParallelMax" Content="LinqAsParallelMax"/>
<TextBlock xName="tbLinqAsParallelMax"/>
</StackPanel>
<StackPanel Orientation="Horizontal">
<Button xName="ButtonVectorMaxMT" Content="VectorMaxMT"/>
<TextBlock xName="tbVectorMaxMT"/>
</StackPanel>
<Border Height="1" Background="Orange" UseLayoutRounding="True"/>
<StackPanel Orientation="Horizontal">
<Button xName="ButtonForIfMT2" Content="ForIfMT2"/>
<TextBlock xName="tbForIfMT2"/>
</StackPanel>
<StackPanel Orientation="Horizontal">
<Button xName="ButtonMathMaxMT2" Content="MathMaxMT2"/>
<TextBlock xName="tbMathMaxMT2"/>
</StackPanel>
<StackPanel Orientation="Horizontal">
<Button xName="ButtonLinqAsParallelMax2" Content="LinqMaxMT2"/>
<TextBlock xName="tbLinqAsParallelMax2"/>
</StackPanel>
<StackPanel Orientation="Horizontal">
<Button xName="ButtonVectorMaxMT2" Content="VectorMaxMT2"/>
<TextBlock xName="tbVectorMaxMT2"/>
</StackPanel>
<StackPanel Orientation="Horizontal">
<Button xName="ButtonTest" Content="test"/>
<TextBlock xName="tbTest"/>
</StackPanel>
</StackPanel>
</Grid>
</Window
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Windows;
using System.Windows.Controls;
using System.Windows.Data;
using System.Windows.Documents;
using System.Windows.Input;
using System.Windows.Media;
using System.Windows.Media.Imaging;
using System.Windows.Navigation;
using System.Windows.Shapes;
using System.Diagnostics;
using System.Numerics;
namespace _20200125_VectorMinMax
{
<summary>
</summary>
public partial class MainWindow : Window
{
private int[] MyArray;
const int MY要素数 = 10_000_000;
private int Myループ回数;
public MainWindow()
{
InitializeComponent();
=<int>
SetLoopCount(100);
MyArray = new int[MY要素数];
SetMyArray(int.MinValue, int.MaxValue);
ButtonRandomArray.Click += (s, e) => { SetMyArray(int.MinValue, int.MaxValue); };
ButtonSetLoopCount1.Click += (s, e) => { SetLoopCount(1); };
ButtonSetLoopCount10.Click += (s, e) => { SetLoopCount(10); };
ButtonSetLoopCount100.Click += (s, e) => SetLoopCount(100);
ButtonSetLoopCount1000.Click += (s, e) => SetLoopCount(1000);
ButtonForIf.Click += (s, e) => { MyTime(ForIf, tbForIf); };
ButtonMathMax.Click += (s, e) => { MyTime(MathMax, tbMathMax); };
ButtonLinqMax.Click += (s, e) => { MyTime(LinqMax, tbLinqMax); };
ButtonVectorMax.Click += (s, e) => { MyTime(VectorMax, tbVectorMax); };
ButtonForIfMT.Click += (s, e) => { MyTime(ForIfMT, tbForIfMT); };
ButtonMathMaxMT.Click += (s, e) => { MyTime(MathMaxMT, tbMathMaxMT); };
ButtonLinqAsParallelMax.Click += (s, e) => { MyTime(LinqMaxAsParallel, tbLinqAsParallelMax); };
ButtonVectorMaxMT.Click += (s, e) => { MyTime(VectorMaxMT, tbVectorMaxMT); };
ButtonForIfMT2.Click += (s, e) => { MyTimeParallel(ForIf, tbForIfMT2); };
ButtonMathMaxMT2.Click += (s, e) => { MyTimeParallel(MathMax, tbMathMaxMT2); };
ButtonLinqAsParallelMax2.Click += (s, e) => { MyTimeParallel(LinqMax, tbLinqAsParallelMax2); };
ButtonVectorMaxMT2.Click += (s, e) => { MyTimeParallel(VectorMax, tbVectorMaxMT2); };
ButtonVectorMinMaxMT2.Click += (s, e) => MyTimeParallelMinMax(VectorMinMax, tbVectorMinMaxMT2);
}
private void SetMyArray(int min, int max)
{
var r = new Random();
for (int i = 0; i < MyArray.Length; i++)
{
MyArray[i] = r.Next(min, max);
}
r.Next(min, max);
}
private void SetLoopCount(int loopCount)
{
Myループ回数 = loopCount;
this.Title = $"int型配列(要素数10,000,000)から最大値を、{Myループ回数}回取得するまでの処理時間";
foreach (var item in EnumerateDescendantObjects2<TextBlock>(this))
{
item.Text = "";
}
}
#region コントロールの列挙
private static IEnumerable<T> EnumerateDescendantObjects<T>(DependencyObject obj) where T : DependencyObject
{
foreach (var child in LogicalTreeHelper.GetChildren(obj))
{
if (child is T cobj)
{
yield return cobj;
}
if (child is DependencyObject dobj)
{
foreach (var cobj2 in EnumerateDescendantObjects<T>(dobj))
{
yield return cobj2;
}
}
}
}
private static List<T> EnumerateDescendantObjects2<T>(DependencyObject obj) where T : DependencyObject
{
var l = new List<T>();
foreach (object child in LogicalTreeHelper.GetChildren(obj))
{
if (child is T)
{
l.Add((T)child);
}
if (child is DependencyObject dobj)
{
foreach (T cobj2 in EnumerateDescendantObjects2<T>(dobj))
{
l.Add(cobj2);
}
}
}
return l;
}
#endregion
private int ForIf(int[] intAry)
{
int max = intAry[0];
for (int i = 1; i < intAry.Length; i++)
{
if (max < intAry[i]) max = intAry[i];
}
return max;
}
private int ForIfMT(int[] intAry)
{
int cpuThread = Environment.ProcessorCount;
int windowSize = intAry.Length / cpuThread;
int[] mm = Task.WhenAll(Enumerable.Range(0, cpuThread).Select(n => Task.Run(() =>
{
var a = intAry.Skip(windowSize * n).Take(windowSize);
return ForIf(a.ToArray());
}))).GetAwaiter().GetResult();
return mm.Max();
}
#region 未使用
private int ForIfMT最後の要素まできっちり処理(int[] intAry)
{
int cpuThread = Environment.ProcessorCount;
int windowSize = intAry.Length / cpuThread;
int[] mm = Task.WhenAll(Enumerable.Range(0, cpuThread).Select(n => Task.Run(() =>
{
var a = intAry.Skip(windowSize * n).Take(windowSize);
return ForIf(a.ToArray());
}))).GetAwaiter().GetResult();
int last = windowSize * cpuThread;
int[] nokori = intAry.Skip(last).ToArray();
return nokori.Concat(mm).Max();
}
private int ForIfParallelFor(int[] intAry)
{
int cpuThread = Environment.ProcessorCount;
int windowSize = intAry.Length / cpuThread;
int[][] temp = new int[cpuThread][];
for (int i = 0; i < cpuThread; i++)
{
temp[i] = intAry.Skip(i * windowSize).Take(windowSize).ToArray();
}
int[] neko = new int[cpuThread];
Parallel.For(0, cpuThread, i => { neko[i] = ForIf(temp[i]); });
return neko.Max();
}
#endregion
private int MathMax(int[] intAry)
{
int max = intAry[0];
for (int i = 1; i < intAry.Length; i++)
{
max = Math.Max(max, intAry[i]);
}
return max;
}
private int MathMaxMT(int[] intAry)
{
int cpuThread = Environment.ProcessorCount;
int windowSize = intAry.Length / cpuThread;
int[] mm = Task.WhenAll(Enumerable.Range(0, cpuThread).Select(n => Task.Run(() =>
{
var a = intAry.Skip(windowSize * n).Take(windowSize);
return MathMax(a.ToArray());
}))).GetAwaiter().GetResult();
return mm.Max();
}
private int LinqMax(int[] intAry)
{
return intAry.Max();
}
private int LinqMaxAsParallel(int[] intAray)
{
return intAray.AsParallel().Max();
}
private int VectorMax(int[] intAry)
{
int simdLength = Vector<int>.Count;
int myLast = intAry.Length - simdLength;
var vMax = new Vector<int>(int.MinValue);
for (int i = 0; i < myLast; i += simdLength)
{
vMax = System.Numerics.Vector.Max(vMax, new Vector<int>(intAry, i));
}
int max = int.MinValue;
for (int i = 0; i < simdLength; i++)
{
if (max < vMax[i]) max = vMax[i];
}
for (int i = myLast; i < intAry.Length; i++)
{
if (max < intAry[i]) max = intAry[i];
}
return max;
}
private int VectorMaxMT(int[] intAry)
{
int cpuThread = Environment.ProcessorCount;
int windowSize = intAry.Length / cpuThread;
int[] mm = Task.WhenAll(Enumerable.Range(0, cpuThread).Select(n => Task.Run(() =>
{
int[] a = new int[windowSize];
int typeSize = System.Runtime.InteropServices.Marshal.SizeOf(
intAry.GetType().GetElementType());
Buffer.BlockCopy(intAry, n * windowSize * typeSize, a, 0, windowSize * typeSize);
return VectorMax(a);
}))).GetAwaiter().GetResult();
return mm.Max();
}
private (int min, int max) VectorMinMax(int[] intAry)
{
int simdLength = Vector<int>.Count;
int myLast = intAry.Length - simdLength;
var vMin = new Vector<int>(int.MaxValue);
var vMax = new Vector<int>(int.MinValue);
for (int i = 0; i < myLast; i += simdLength)
{
var v = new Vector<int>(intAry, i);
vMin = System.Numerics.Vector.Min(vMin, v);
vMax = System.Numerics.Vector.Max(vMax, v);
}
int max = int.MinValue;
int min = int.MaxValue;
for (int i = 0; i < simdLength; i++)
{
if (max < vMax[i]) max = vMax[i];
if (min > vMin[i]) min = vMin[i];
}
for (int i = myLast; i < intAry.Length; i++)
{
if (max < intAry[i]) max = intAry[i];
if (min > intAry[i]) min = intAry[i];
}
return (min, max);
}
#region 時間計測
private void MyTime(Func<int[], int> func, TextBlock textBlock)
{
var sw = new Stopwatch();
int max = int.MinValue;
sw.Start();
for (int i = 0; i < Myループ回数; i++)
{
max = func(MyArray);
}
sw.Stop();
textBlock.Text =
$"{System.Reflection.RuntimeReflectionExtensions.GetMethodInfo(func).Name}" +
$"\n最大値 = {max}\n処理時間:{sw.Elapsed.TotalSeconds.ToString("00.00")}秒";
}
private void MyTimeParallel(Func<int[], int> func, TextBlock textBlock)
{
var sw = new Stopwatch();
int max = int.MinValue;
int[] mm = new int[Myループ回数];
var option = new ParallelOptions();
option.MaxDegreeOfParallelism = Environment.ProcessorCount;
sw.Start();
Parallel.For(0, Myループ回数, option, n =>
{
mm[n] = func(MyArray);
});
max = mm.Max();
sw.Stop();
textBlock.Text =
$"{System.Reflection.RuntimeReflectionExtensions.GetMethodInfo(func).Name}" +
$"\n最大値 = {max}\n処理時間:{sw.Elapsed.TotalSeconds.ToString("00.00")}秒";
}
private void MyTimeParallelMinMax(Func<int[], (int min, int max)> func, TextBlock textBlock)
{
var sw = new Stopwatch();
int min = int.MaxValue;
int max = int.MinValue;
int[] mins = new int[Myループ回数];
int[] maxs = new int[Myループ回数];
var option = new ParallelOptions();
option.MaxDegreeOfParallelism = Environment.ProcessorCount;
sw.Start();
Parallel.For(0, Myループ回数, option, n =>
{
(mins[n], maxs[n]) = func(MyArray);
});
min = mins.Min();
max = maxs.Max();
sw.Stop();
textBlock.Text =
$"{System.Reflection.RuntimeReflectionExtensions.GetMethodInfo(func).Name}" +
$"\n最小値 = {min}\n最大値 = {max}\n処理時間:{sw.Elapsed.TotalSeconds.ToString("00.00")}秒";
}
#endregion
}
}
関連記事
1ヶ月後
Intrinsics.Vector vs Numerics.Vector
10日後
gogowaten.hatenablog.com
前回は6日前
1週間前
System.Numerics.VectorのDistanceメソッドと参照の追加