Gushwell's C# Dev Notes

この記事は、C# Advent Calender 2012 http://atnd.org/events/33905 への参加記事です。
今年もC# Advent Calender が開始されました。

なかなか書くネタが見つからず、参加しようかどうか悩んでいるうちに、気が付いたら、空いている日が、
初日の 12/1 と 最終日の 12/25 の２日だけになってました。さすがに最終日はプレッシャーがかかると
思ったので、初日である 12/1 にエントリーさせてもらいました。

さて、お題は、「.NET Framework4.5 での非同期I/O処理 事始め」です。
既に多くの方がブログで async, awaitの記事を書いているので、２番煎じ、３番煎じの感は
否めませんが、まあ、このC# Advent Calenderは、ゆるいイベントということなので、大目
に見てください(^^;

さて、ここからが本題です。

.NET Framework4.5から、非同期IOをサポートする新しいasync系メソッドが追加されています。
これにより、従来よりもより簡単に非同期IO処理が書けるようになりました。

■.NET Framework4.0 以前の非同期I/O処理

まずは、.NET Framework4.0 以前はどうだったかを振り返ってみましょう。
これまでの Streamクラスには、BeginRead, EndRead といった非同期IOをサポートするメソッドが
用意されていました。
従来の非同期IOの簡単なサンプルを以下に示します。
このコードは、Sample.txtファイルを非同期IOで読み込み、TextBox にその内容を表示しています。

※ 今回は、WindowsFormsを使いましたが、WPFでもなんら変わることはありません。

public partial class Form1 : Form { public Form1() { InitializeComponent(); } private byte[] buff = new byte[200]; private FileStream fs; private void button1_Click(object sender, EventArgs e) { textBox1.Text = ""; fs = new FileStream("Sample.txt",FileMode.Open, System.IO.FileAccess.Read, FileShare.ReadWrite); fs.BeginRead(buff, 0, 200, Callback, fs); } void Callback(IAsyncResult result) { int lenght = fs.EndRead(result); if (lenght > 0) { string s = Encoding.UTF8.GetString(buff, 0, lenght); this.Invoke(new MethodInvoker(() => textBox1.Text += s )); fs.BeginRead(buff, 0, 20, Callback, fs); } else { fs.Close(); } } }
ファイルのサイズが小さければ、こんな面倒なコードを書かずに、File.ReadAllTextで一気に
読み込んでしまって良いと思いますが、巨大なファイルの場合は、UIがフリーズするのを防止できません。
UIのフリーズを防止するには、スレッドを起動するとか、BackgroundWorkerクラスを利用するとか、Taskを使うとか、
いろいろと方法はありますが、同期処理に比べると面倒です。

それと、上記のコードは、テキストファイルを扱うのに、ReadLine系のメソッドを使えないため、byte配列に
読み込んでいて、いまいち感がぬぐえません。

■.NET Framework4.5 での非同期I/O処理

では、C#5.0 + .NET Framework4.5ではどうなるかを見てみましょう。

public partial class Form1 : Form { public Form1() { InitializeComponent(); } private void button1_Click(object sender, EventArgs e) { ReadTextData(); } private async void ReadTextData() { FileStream fs = new FileStream("Sample.txt", FileMode.Open, System.IO.FileAccess.Read, FileShare.ReadWrite); StreamReader sr = new StreamReader(fs); string s = await sr.ReadLineAsync(); while (s != null) { textBox1.Text += s + Environment.NewLine; s = await sr.ReadLineAsync(); } } }
上記コードの注目点は、以下の４点です。

1. メソッドに、async キーワードが付いている。
2. ReadLineAsync() という非同期IOメソッドを使っている。
3. この非同期IOメソッドに、await キーワードをつけている。
4. 同期処理のように、UIへのアクセスを普通に行っている。

メソッドに、await キーワードを使う場合には、必ずメソッドにasync キーワードをつける必要があります。
そして、await を使うことで、非同期処理をあたかも同期処理のように書くことが
可能になります。
いやー、これはほんとうに便利ですね。

■IProgressによるロジックとUIの分離

でも、ファイルIO処理の中で、UIにアクセスするコードを書くのは嫌だ！ と思う人も多いと思います。
大丈夫です。.NET Framework4.5では、そういった要求にも簡単にこたえることができます。
コードを示します。

private void button1_Click(object sender, EventArgs e) { ReadTextData(new Progress<string>( line => textBox1.Text += line + Environment.NewLine )); } static async void ReadTextData(IProgress<string> progress) { FileStream fs = new FileStream("Sample.txt", FileMode.Open, System.IO.FileAccess.Read, FileShare.ReadWrite); StreamReader sr = new StreamReader(fs); string s = await sr.ReadLineAsync(); while (s != null) { progress.Report(s); s = await sr.ReadLineAsync(); } }
ReadTextDataメソッドの引数に、Progress オブジェクトを渡しています。
このProgress のコンストラクタの引数に、コールバックされるラムダ式を記述します。
ここで、UI処理を書くわけですね。
ReadTextData メソッドの中では、progress.Report(s) と、Reportメソッドを呼び出しています。
このタイミングで、先ほどのコールバックが呼び出されることになります。
これで、ReadTextDataメソッドは、完全にUIとは分離されたので、Formクラスに書く必要もなくなりましたね。

Write系のコードも載せようと思ったのですが、まあ、上記とほとんど同じ感覚で書けるので、
あえて載せる必要もないでしょう。

  

staticフィールドは、複数のスレッドで共有されるので、
スレッド間で情報を共有する際に利用できますが、
スレッド毎に独立したstaticフィールドを持ちたい場合があります。

そんな時に、フィールドに ThreadStatic属性を使います。
しばらくぶりに、このThreadStatic属性を使いました。
あまりにもしばらくぶりだったので、その存在自体忘れてました(^^;
  

BackgroundWorkerのサンプルコード(1)
BackgroundWorkerのサンプルコード(2)
BackgroundWorkerのサンプルコード(3)

BackgroundWorkerのサンプルコード(3)で掲載したコードとほとんど同じだけど、バックグラウンド処理の結果を受け取れるようにしました。
終了していないのに、結果を取得しようとした場合は、終了するまで待つようにしてあります。
ついでにExecuteメソッドでは、引数を渡せるようにしてあります。
同期するために利用していたManualResetEventですが、代わりにAutoResetEventを使ってみました。


class Program { static void Main(string[] args) { MyWorker worker1 = new MyWorker("Hoo"); Console.WriteLine("Hoo 開始"); worker1.Execute(150); MyWorker worker2 = new MyWorker("Bar"); Console.WriteLine("Bar 開始"); worker2.Execute(7); Console.WriteLine("Hoo " + worker1.Result.ToString()); Console.WriteLine("Bar " + worker2.Result.ToString()); Console.ReadLine(); } } class MyWorker { private BackgroundWorker _worker = new BackgroundWorker(); private string _name; private AutoResetEvent _are = new AutoResetEvent(false); private long _result; public MyWorker(string name) { _worker.DoWork += new DoWorkEventHandler(worker_DoWork); _worker.RunWorkerCompleted += new RunWorkerCompletedEventHandler(worker_RunWorkerCompleted); this._name = name; } public void Execute(int n) { _worker.RunWorkerAsync(n); } public void Join() { _are.WaitOne(); _are.Set(); } public long Result { get { _are.WaitOne(); _are.Set(); return this._result; } } private void worker_RunWorkerCompleted(object sender, RunWorkerCompletedEventArgs e) { if ( !e.Cancelled && e.Error == null ) { this._result = (long)e.Result; } Console.WriteLine(_name + " 終了 "); _are.Set(); } private void worker_DoWork(object sender, DoWorkEventArgs e) { int count = (int)e.Argument; long a = 1; long b = 1; long c = 1; for ( int i = 3; i <= count; i++ ) { c = a + b; a = b; b = c; } System.Threading.Thread.Sleep(1000); // デモ用・本来は不要 e.Result = c; } }   

BackgroundWorkerのサンプルコード(1)
BackgroundWorkerのサンプルコード(2)


BackgroundWorkerは、Formが無くたって使えます。
今回のサンプルコードは、ManualResetEventを使って、終了まで待てるようにしています。
クラスMyWorkerをもうちょっと汎用的にしたいところですが、BackgroundWorkerの サンプルを示すことが目的なので、これで良しとします。

このプログラムを実行してみればわかりますが、worker1とworker2がパラレルに実行されているのが確認できます。

class Program { static void Main(string[] args) { MyWorker worker1 = new MyWorker("Hoo"); worker1.Execute(); MyWorker worker2 = new MyWorker("Bar"); worker2.Execute(); worker1.Join(); worker2.Join(); Console.Write("終了 "); } } class MyWorker { private BackgroundWorker worker = new BackgroundWorker(); private string name; private object lockObject = new object(); ManualResetEvent mre = new ManualResetEvent(false); public MyWorker(string name) { worker.DoWork += new DoWorkEventHandler(worker_DoWork); this.name = name; } public void Execute() { worker.RunWorkerAsync(); } public void Join() { mre.WaitOne(); } private static System.Random rnd = new Random(); private void worker_DoWork(object sender, DoWorkEventArgs e) { mre.Reset(); for ( int i = 0; i < 50; i++ ) { Console.Write(name + i.ToString() + "\n"); System.Threading.Thread.Sleep(rnd.Next(1,100)); } mre.Set(); } }   

BackgroundWorkerのサンプルコード(1)

今度は、処理の進行状況をプログレスバーで表示する例です。
途中で、処理をキャンセルする例にもなっています。

Form1のボタンをクリックすると、Form2が開き、処理を開始します。
処理の進行状況は、プログレスバーで表示され、処理が終わると 自動的にForm2が閉じるというものです。
Form2には、キャンセルボタンが配置されていて、キャンセルボタンを押すと 処理を中止します。


public partial class Form1 : Form { public Form1() { InitializeComponent(); } private void button1_Click(object sender, EventArgs e) { using ( Form2 fm = new Form2() ) { if ( fm.ShowDialog() == DialogResult.Cancel ) { MessageBox.Show("キャンセルされました"); } } } }
public partial class Form2 : Form { public Form2() { InitializeComponent(); } BackgroundWorker _bw = new BackgroundWorker(); private void Form2_Load(object sender, EventArgs e) { _bw.DoWork += new DoWorkEventHandler(_bw_DoWork); _bw.ProgressChanged += new ProgressChangedEventHandler(_bw_ProgressChanged); _bw.RunWorkerCompleted += new RunWorkerCompletedEventHandler(_bw_RunWorkerCompleted); _bw.WorkerSupportsCancellation = true; // キャンセルを可能にする _bw.WorkerReportsProgress = true; _bw.RunWorkerAsync(); progressBar1.Maximum = 90; } private void Form2_Shown(object sender, EventArgs e) { _bw.WorkerSupportsCancellation = true; } private void cancelButton_Click(object sender, EventArgs e) { _bw.CancelAsync(); this.DialogResult = DialogResult.Cancel; } void _bw_DoWork(object sender, DoWorkEventArgs e) { // 素直に100%まで処理をし、ProgressBarに表示すると、 // この例では、プログレスバーが100%に行かずに // フォームが閉じたように見える。 // ここでは、そのため小手先の対応をしている。 for (int i = 1; i <= 100; i++) { if (_bw.CancellationPending) { e.Cancel = true; return; } _bw.ReportProgress(Math.Min(i,progressBar1.Maximum), null); System.Threading.Thread.Sleep(50); } } void _bw_ProgressChanged(object sender, ProgressChangedEventArgs e) { progressBar1.Value = e.ProgressPercentage; } void _bw_RunWorkerCompleted(object sender, RunWorkerCompletedEventArgs e) { this.DialogResult = DialogResult.OK; } }
  

BackgroundWorkerで検索して、このブログに来ている人が結構いるようなので、何回かに分けて、BackgroundWorkerを使ったサンプルを載せたいと思います。

まずは、バックグラウンド処理で得られたデータをListBoxに順に表示する例です。
ついでに、ListBoxの自動スクロールもやってます。

当然、ListBoxに項目を追加している間でも、ListBoxの操作（選択など）が可能です。

public partial class Form1 : Form { private BackgroundWorker worker = new BackgroundWorker(); public Form1() { InitializeComponent(); worker.DoWork += worker_DoWork; worker.ProgressChanged += worker_ProgressChanged; worker.RunWorkerCompleted += worker_RunWorkerCompleted; worker.WorkerReportsProgress = true; } private void button1_Click(object sender, EventArgs e) { this.worker.RunWorkerAsync(); } private void worker_DoWork(object sender, DoWorkEventArgs e) { BackgroundWorker bw = (BackgroundWorker)sender; // ここで時間のかかる処理 for ( int i = 0; i < 40; i++ ) { System.Threading.Thread.Sleep(100); // 処理をするたびに、ReportProgressを呼び出す。ProgressChangedイベントが発行される bw.ReportProgress(0, i); } } void worker_RunWorkerCompleted(object sender, RunWorkerCompletedEventArgs e) { label1.Text = "終了しました"; } private void worker_ProgressChanged(object sender, ProgressChangedEventArgs e) { int i = (int)e.UserState; listBox1.Items.Add("Item " + i.ToString()); listBox1.TopIndex = listBox1.Items.Count - listBox1.Height / listBox1.ItemHeight; } private void listBox1_SelectedIndexChanged(object sender, EventArgs e) { ListBox lb = (ListBox)sender; if (lb.SelectedItem != null) label1.Text = lb.SelectedItem.ToString(); } }
  

AutoResetEvent の簡単なサンプルです。

ManualResetEvent の簡単なサンプルと見比べてみると、その違いがわかります。

// 非シグナル状態 AutoResetEvent are = new AutoResetEvent(false); private void button3_Click(object sender, EventArgs e) { Thread th = new Thread(ThreadProc2); th.Start(); } // スレッドを起動する void ThreadProc2() { for (int i = 0; i < 10; i++) { // Set(シグナル状態になるまで）を待つ are.WaitOne(); // 自動的に非シグナル状態になる // Invokeでメインスレッド内で、GUI処理を行う this.Invoke(new MethodInvoker(delegate { listBox1.Items.Add(i.ToString()); })); } } private void button4_Click(object sender, EventArgs e) { // シグナル状態 are.Set(); }   

ManualResetEvent の簡単なサンプルです。

// 非シグナル状態 ManualResetEvent mre = new ManualResetEvent(false); // スレッドを起動する private void button1_Click(object sender, EventArgs e) { Thread th = new Thread(ThreadProc); th.Start(); } // button1で起動されるスレッド void ThreadProc() { for (int i = 0; i < 10; i++ ) { // Set(シグナル状態になるまで）を待つ mre.WaitOne(); // 非シグナル状態にする mre.Reset(); // InvokeでGUI処理をする this.Invoke(new MethodInvoker(delegate { listBox1.Items.Add(i.ToString()); })); } } private void button2_Click(object sender, EventArgs e) { // シグナル状態 mre.Set(); }
button2をクリックすることで、ThreadProc内の処理が進みます。
  

備忘メモ

「スレッド プールに許可されるワーカー スレッドの最大数」 と「スレッド プールに許可される非同期 I/O スレッドの最大数」

int workerThreads; int completionPortThreads; System.Threading.ThreadPool.GetMaxThreads(out workerThreads, out completionPortThreads);   

ThreadStaticAttribute こんな属性があることを知らずに、スレッド毎に一つのインスタンスが保障されるコードを一生懸命C#で書いていました。
会社の後輩にこの属性があることを教えてもらったので、備忘のために書いておきます。
  

ThreadPool.GetMaxThreads(out workerThreads, out completionPortThreads);

を実行したら、workerThreadsの数が　500 だった。
CPU×25だと思っていたのだけど、
CPU×250 みたいだ。いつから変わったのかな？  

今まで、C#で、AutorResetEventを使ったコードを書いたことがなかったので、
動きを確かめるため、書いてみた。

using System.Threading;
...

 private List<AutoResetEvent> areList = new List<AutoResetEvent>();

 private void Run() {
     for (int i = 0; i < 3; i++) {
         AutoResetEvent are = new AutoResetEvent(false);
         areList.Add(are);
         Thread th = new Thread(Start);
         th.Start(are);
     }
     AutoResetEvent.WaitAll(areList.ToArray());
     Console.WriteLine("Thread End");
     Console.ReadLine();
 }

 private Random rnd = new Random();
 private void Start(object obj) {
     Console.WriteLine("Thread Start");
     Thread.Sleep(1000*rnd.Next(1,4));
     Console.WriteLine("Thread Stop");
     ((AutoResetEvent)obj).Set();
 }

このコードは、３つのスレッドを起動し、すべてが終わるまで待つというコード。
どれかひとつを待つのならば、WaitAny staticメソッドか、WaitOneメソッドを使えばよい。

WaitOneの場合は、AutoResetEvent のインスタンスはひとつだけでよいので、
コードも簡単だ。（コードは省略）

ManualResetEvent の場合は、Resetをするコードも必要なので、スレッドの終了待ちの
ようなコードは、AutoResetEventの方が楽ですね。  

先日は、

>BackgroundWorker コンポーネントは、System.Threading 名前空間に代わると
>共に追加の機能を提供します。

とMSDN ドキュメントを引用しましたが、

BackgroundWorkerは、名前のとおり、ＧＵＩの裏（バックグラウンド）で処理をさせたいときに利用するためのコンポーネントであって、僕が今やろうとしているスレッドプールを利用した多重処理を実現させるには、ちょっと利用するのが難しそうです。

ただ、WindowsFormで、時間のかかる処理を走らせるにはとても手軽に使えて良いですね。プログレスバーの表示のための仕組みも組み込まれています。
BeginInvoke を使う必要もありません。

先日のコードだと、このあたりが分からないので、再度コードをアップします。

  BackgroundWorker backgroundWorker1 = new BackgroundWorker();
  backgroundWorker1.DoWork += new DoWorkEventHandler(MyTask);
  backgroundWorker1.RunWorkerCompleted += MyTaskCompleted;
  backgroundWorker1.ProgressChanged += 
      new ProgressChangedEventHandler(MyTaskProgressChanged);
  // バックグラウンド処理の起動
  backgroundWorker1.WorkerReportsProgress = true;
  
// プログレスバー表示処理
private void MyTaskProgressChanged(object sender,ProgressChangedEventArgs e) {
    this.progressBar1.Value = e.ProgressPercentage;
}

// バックグラウンド処理
void MyTask(object sender, DoWorkEventArgs e) {
    BackgroundWorker bw = (BackgroundWorker)sender;
    ...
    for ( ... ) {
       ...
       bw.ReportProgress(percentage);
    }
}

// バックグラウンド処理が終了したときに呼び出される 
void MyTaskCompleted(object sender, RunWorkerCompletedEventArgs e) {
   ...
}

※ ReportProgress には、ユーザデータを渡せるメソッドがオーバーロードされていますので、パーセント以外の情報も表示できます。  

MSDN のドキュメントを読んでたら、こんな記述が目にとまった。

BackgroundWorker コンポーネントは、System.Threading 名前空間に代わると共に追加の機能を提供します。

ということは、BackgroundWorker コンポーネントを利用することが推奨されているってことかな。
たしかに、簡単です。

backgroundWorker1.DoWork += MyTask;
backgroundWorker1.RunWorkerCompleted += MyTaskCompleted;
backgroundWorker1.RunWorkerAsync();

private void MyTask(object sender, DoWorkEventArgs e) {
    ...
}
        
private void MyTaskCompleted(object sender, 
                          RunWorkerCompletedEventArgs e) {
    ...
}

  

.NET Framework2.0 で ThreadPool クラスに SetMaxThreads メソッド が追加されています。

  ThreadPool.SetMaxThreads(workerThreads, completionPortThreads);

  workerThreads
     スレッド プール内のワーカー スレッドの最大数。 
  completionPortThreads
     スレッド プール内の非同期 I/O スレッドの最大数。 

これで、同時にアクティブにできるスレッドの数を設定できるようになりました。
最適なスレッド数を実測して求めるにの利用できそうです。

workerThreads の デフォルト値は、シングルCPUの場合 ２５ですが、２５より大きな値を入れても大丈夫みたいです。
まあ、そんなに大きな値を指定しても、パフォーマンスが悪くなるので、小さく制限することはあっても、25よりも大きな値を設定することは、無いと思いますが...