F#の土台を越えて - 非同期ワークフロー

F#とは？
F#は、.NET Frameworkをターゲットとした静的型付け関数型プログラミング言語です。人気が高い別の関数型言語OCamlと共通のコア言語を持ち、HaskellやErlang、C#など、他の多数のプログラミング言語からアイデアを採り入れています。簡単に言えばF#は、インタラクティブにコードの実行が可能なことから、いささかスクリプティングのような感じもしますが、コンパイラ型言語のすべての型安全性とパフォーマンスを持つ、素晴らしく簡潔なシンタックスを有するプログラミング言語です。この記事はF#の紹介を意図したものではありませんが、Web上にはF#を簡単に習得することを目的とした情報資源がたくさんあります。私が書いたF#に関する最初の記事のサイドバーにある「F# Resources」をご覧ください。

非同期プログラミングモデル

.NET BCLを扱っていると、すべてのI/Oオペレーションで同期モデルあるいは非同期モデルの2つのうちどちらかのモデルを使います。非同期モデルは共通のプログラミングパターンを通じてサポートされますが、このパターンでは1対のBeginXXXとEndXXXのメソッドが提供され、プログラマーがBeginXXXを呼び出すと、BeginXXXがオペレーションを開始させ、復帰します。プログラマーは、非同期オペレーション終了を知らされたら、直ちにEndXXXメソッドを呼び出す必要があります。

私の経験では、ほとんどのプログラマーが同期モデルを使う傾向にありますが、その理由は単純性と、BCLの多数のクラスが同期モデルだけをサポートしているからです。しかし多くの場合、非同期プログラミングモデルを使えば、より応答性の高い、いっそうスケーラブルなアプリケーションを制作できます。非同期モジュールの難しさを例証するために、ファイルを開き、そのファイルからバイトを読み取るという単純な例を見てみましょう。以下は同期モデルでのコード例です。

#light
open System.IO

let openFile() =
     use fs = new FileStream(@"C:\Program Files\Internet Explorer\iexplore.exe",
                             FileMode.Open, FileAccess.Read, FileShare.Read)
     let data = Array.create (int fs.Length) 0uy
     let bytesRead = fs.Read(data, 0, data.Length)
     printfn "Read Bytes: %i, First bytes were: %i %i %i ..." 
         bytesRead data.(1) data.(2) data.(3)

 openFile()

BCLには「File.ReadAllBytes」など、より単純な方法がありますが、非同期に同等のやり方があるものとしては、上記が最も単純な方法です。ファイルから読み込むオペレーションは非常に単純で、ファイルストリームを開き、データを保持するための配列を作成し、そして最後にその配列にすべてのデータを読み込みます。ファイルストリーム作成時にどのようにして「use」バインディングを使うかに注目してください。C#におけるusingステートメントにほぼ相当し、ファイルストリームが範囲外になったら配置されることを意味します。

次に、非同期プログラミングモデルを使った同等物を見てみましょう。

#light
open System.IO

let openFile() =
     let fs = new FileStream(@"C:\Program Files\Internet Explorer\iexplore.exe",
                             FileMode.Open, FileAccess.Read, FileShare.Read)
     let data = Array.create (int fs.Length) 0uy
     let callback ar =
         let bytesRead = fs.EndRead(ar)
         fs.Dispose()
         printfn "Read Bytes: %i, First bytes were: %i %i %i ..."
             bytesRead data.(1) data.(2) data.(3)
     fs.BeginRead(data, 0, data.Length, (fun ar -> callback ar), null) |> ignore

  openFile()

ファイルを開くこの単純な例は、今のところはかなり理解できるものですが、事態は確かに複雑化しています。最初の2段階はおおかた同じであり、ファイルストリームを開き、データ保持のための配列を作成しています。しかしその後は事態が悪化し、「EndRead」メソッドの呼び出しを処理するためにコールバックを定義しなければならず、そしてこのコールバックを状態オブジェクト（ここでは必要ないのでnull）と一緒に「BeginRead」メソッドの中に渡す必要があります。「use」バインディングを使えないことも重要です。なぜなら「BeginRead」メソッドの存在時には、ファイルストリームが範囲外になり、つまりファイルストリームの配置が早すぎて、「EndRead」メソッドが呼び出される時には使用不可であることを意味するからです。すなわち、その「Dispose」（配置）メソッドに呼び出しを追加する必要があり、finally blockの中で呼び出していた安全性を失うことを意味します。ファイルを開くような単純な例では、こうした余分な複雑性も合理的に思われますが、アプリケーションにより多くの機能や、非同期の読み取りを追加していくにつれ、すぐに問題に直面することになります。

非同期ワークフロー

非同期ワークフローは、特にこの問題に対処するために導入されてきました。では、非同期ワークフローバージョンを見てみましょう。

#light

open System.IO
open  Microsoft.FSharp.Control.CommonExtensions

let  openFile =
     async { use  fs = new  FileStream(@"C:\Program Files\Internet Explorer\iexplore.exe",
                                     FileMode.Open, FileAccess.Read, FileShare.Read)
             let  data = Array.create (int fs.Length) 0uy
             let!  bytesRead = fs.ReadAsync(data, 0, data.Length)
             do  printfn "Read Bytes: %i, First bytes were: %i %i %i ..."
                 bytesRead data.(1) data.(2) data.(3) }

 Async.Run openFile

ワークフローバージョンで最も重要なことは、同期バージョンとの相違がわずか数文字という点です。「async { ... }」というワークフロー宣言が追加されていますが、もっと重要なことは、ファイルから読み込む行を以下のように変えたことです。

let! bytesRead = fs.ReadAsync(data, 0, data.Length)

letキーワードに感嘆符（!）も追加し、「Read」ではなく今度は「ReadAsync」を呼び出しています。非同期ワークフローで「let!」が命じているのは、非同期にバインドを作成し、また、BeginReadとEndReadメソッドがどのように呼び出されるかという仕様をReadAsync関数が提供するということです。非同期関数を使わないと、コンパイルタイプエラーになることに注意してください。では、「ReadAsync」は一体どこからやって来たのでしょうか。「FileStream」クラスで通常利用可能な関数ではありません。観察力の鋭い読者は「open Microsoft.FSharp.Control.CommonExtensions」に気付かれるでしょうが、これによりF#型拡張を多数含んだネーム空間が開きます。これはC#の拡張メソッドに非常に類似しており、既存のクラスにさらに関数を追加できるようにし、ネーム空間「Microsoft.FSharp.Control.CommonExtensions」により、非同期ワークフローで使用する多くの拡張機能が提供されます。

ファイル配置に今までどおり「use」バインディングを使えることも重要であり、ファイルが別のスレッド上に配置されることを意味するとしても、別に気にすることはなく、useが機能するというだけのことです。

他の注目に値する変更はその実行方法です。「fileOpen」識別子は直ちにファイルを開いたりはしませんが、それはこれがワークフローであり、これから発生するアクションを待っているからです。このアクションを実行するために、「Async.Run」関数を使う必要があり、この関数がワークフローを1つ実行し、その結果を待ちます。

呼び出された「ReadAsync」の両側に小さなデバッギング関数の呼び出しを追加すると、非同期がどのように機能するかを理解するのに役立ちます。プログラムがどのスレッドを実行中かを把握し、また、スレッドスタックトレースを理解できるようになりますが、これについては読者への課題として残しておきます。:

let printThreadDetails() =
     Console.WriteLine("Thread ID {0}", Thread.CurrentThread.ManagedThreadId)
     Console.WriteLine((new StackTrace()).ToString())

ワークフローに関する私の元々の記事（http://www.infoq.com/articles/pickering-fsharp-workflow)に戻って、「let!」が継続関数にシンタックスシュガー化された経緯を理解するのも良いでしょう。非同期ワークフローがどのようにして「let!」後の別のスレッドで、とりたてて何もしないのに再度開始できるのかを理解するのに役立つでしょう。

パフォーマンスゲインの数量化

では、非同期ワークフローを使うと、どのようなパフォーマンスゲインを期待できるのでしょうか。ほとんどすべてのパフォーマンス関連の疑問がそうであるように、実験もしないで回答するのは難しいことです。プログラムは概して計算バウンドか、I/Oバウンドかのどちらかであり、非同期ワークフローを使うことにより、一般に両ケースで改善が見られるでしょう。しかし、これには使用するハードウェアが大きな影響をもたらし、タスクがI/Oバウンドであるなら、ディスクの同時アクセスが良くない限り、あまり改善が見られないことを指摘しておくべきであり、非常に良い同時アクセスを提供するディスクも出回ってはいますが、ラップトップやデスクトップよりもむしろ、高仕様のサーバー向けという傾向にあります。タスクがプロセッサバウンドなら、一般にパフォーマンスゲインは良くなるでしょう。なぜなら、ほとんどの最新ラップトップやデスクトップがデュアルコアプロセッサを装備しているからで、間もなく登場するクアドコアモデルを注文しようと考えている読者も多数いるかもしれません。つまり、非同期ワークフローを正しく使うことは、割増の処理能力を活用できることを意味するのです。

I/Oと計算の両方を行うタスク例を最初から最後まで処理し、どのようなパフォーマンスゲインが得られるのかを見てみましょう。分析したいアスキーのテキストがあるとします。最初のステップで総単語数を数え、次に固有の単語数を数えるとしましょう。ファイルを開いて読むのはI/Oの仕事であり、単語を数えて固有の単語を計算することにより、計算のオーバーヘッドが発生します。このテスト用として、Project Gutenberg（プロジェクト・グーテンベルグ）からヘンリー・フィールディングのすべての作品(リンク)をダウンロードすることに決めました（フィールディングを選んだ主な理由は、シェークスピアやディケンズとは異なり、すべてダウンロードしても腹が立たないほど、フィールディングの作品数が少ないからである）。

まず、作品を同期的に分析するスクリプトが必要です。

#light
open System
open  System.Diagnostics
open  System.IO
open  System.Text.RegularExpressions

let  path = @"C:\Users\robert\Documents\Fielding"
 let  readFile filePath =
     // open and read file
     let  fileStream = File.OpenText(filePath)
     let  text = fileStream.ReadToEnd()

     // find all the "words" using a regex
     let  word = new  Regex("\w+")
     let  matches = word.Matches(text)
     let  words  = { for m in  matches -> m.Value }

     // count unique words using a set
     let uniqueWords = Set.of_seq words

     // print the results
     let name = Path.GetFileNameWithoutExtension(filePath)
     Console.WriteLine("{0} - Words: {1} Unique words: {2} ",
                       name, matches.Count, uniqueWords.Count)

 let main() =
     let filePaths = Directory.GetFiles(path)
     for filePath in filePaths do readFile filePath

お分かりのように、このスクリプトは非常に単純で、最初にファイルを開いて読み込み、次に正規表現を使ってすべての単語を数え（ここでは、1つ以上の連続した文字を単語として定義）、次にセットを作成することにより、固有の単語を数えます。「Set」型はF#ネイティブのライブラリの一部であり、計算でセットをモデル化する不変のデータ構造であって、ドキュメント内の固有の単語を計算する上では効率のいい仕事をするでしょうが、それでも計算上かなり無骨なやり方でしょう。

さて、非同期バージョンを考察してみましょう。

#light
open System
open System.IO
open System.Text.RegularExpressions
open Microsoft.FSharp.Control.CommonExtensions

let path = @"C:\Users\robert\Documents\Fielding"

let readFileAsync filePath =
     async { // open and read file
             let fileStream = File.OpenText(filePath)
             let! text = fileStream.ReadToEndAsync()

             // find all the "words" using a regex
             let word = new Regex("\w+")
             let matches = word.Matches(text)
             let words  = { for m in matches -> m.Value }

             // count unique words using a set
             let uniqueWords = Set.of_seq words
             // print the results
             let name = Path.GetFileNameWithoutExtension(filePath)
             do Console.WriteLine("{0} - Words: {1} Unique words: {2} ",
                                  name, matches.Count, uniqueWords.Count) }

let main() =
     let filePaths = Directory.GetFiles(path)
     let tasks = [ for filePath in filePaths -> readFileAsync filePath ]
     Async.Run (Async.Parallel tasks)

 main()

見て分かるように、ファイル読み取り関数は、「async { ... }」ワークフローで関数をラップし、「ReadToEnd」の代わりに「ReadToEndAsync」関数を呼び出している以外、ほとんど変わっていません。「主要」関数に対する変更の方がより興味深く、ここでは最初にファイルのリストを非同期ワークフローのリストにマップした後、「tasks」識別子にそれをバインドします。この時点ではワークフローがまだ実行されていないことに留意し、ワークフローを実行するために、「Async.Parallel」を使ってタスクのリストを単一のワークフローに変換しますが、このワークフローが並行実行されることになります。次に「Async.Run」を使い、並行でタスクを実行します。

F#をインタラクティブに使い、優秀な計時機能を備えた自分のラップトップ（デュアルコア）でテストを行いました。やり方は単純です。両スクリプトを1度走らせた後、その結果を破棄し（ディスクキャッシング効果を排除するため）、その後、各スクリプトを3回走らせました。

デュアルコアマシンでは、非同期バージョンは同期バージョンより約22%高速で動作していますが、わずか2行変えただけの修正ということを考慮すると悪くない結果だといえます。けれどもなぜ100%の高速化にならないのでしょうか。答えは非常に簡単です。このタスクは完全な計算バウンドではないということであり、各単語の発生数を数えるとか、類似単語のグループを作ってみるなど、アルゴリズムにさらに計算タスクを加えたなら、高速化のパーセンテージが増加するでしょう。ファイルの読み取りと処理だけが非同期ワークフローが役に立つ唯一のタスクというわけではなく、ネットワークプログラミングにも使えます。実際、ネットワークデータアクセスの方がディスクアクセスより遅い傾向にあるため、ネットワークアクセスが完了する間、スレッドのブロックを回避するために非同期ワークフローを使えば、ファイルアクセスに非同期ワークフローを使うより、多数のメリットをもたらすことができます。

結論

.NET Framework上で利用可能な最も洗練された解決策を提供することにより、非同期ワークフローは、.NET非同期プログラミングモデルをいかにして正しく使用するかという、非常に特化した問題に対処します。非同期プログラミングモデルを使えば、アプリケーションがいっそうスケーラブルになり、また、非同期ワークフローにより、さらに容易にできるようになります。

さらに勉強したい場合は

Jeffery Richter氏が、C#を使って非同期プログラミングモデルを実装する場合の非同期プログラミングモデルやその問題、解決策を取り上げています(リンク)。

非同期ワークフローについては、『Expert F#』（APress、2007年12月刊）の13章で取り上げており、14章ではさらに例が掲載されています。