マルチコア時代のプログラマは関数脳になろう〜Java8のススメ〜

松井暢之 (2013年8月12日 10:00)

CPUのクロックアップに限界が訪れ、マルチコア化することで処理性能向上を目指す時代になりました。これからのプログラマには、マルチコアで処理性能が向上するプログラム＝マルチスレッドで並列処理が可能なプログラムを書く能力が必要になります。今回は「関数型」でプログラムを書くことによって、いとも簡単に並列化ができることを実例を元に解説します。

関数型プログラミングと並列処理

「関数型でプログラムを書くことで簡単に並列化できる」と書きましたが、そもそもここで言う「関数」とは何なのでしょうか？

関数型プログラミングの特徴

関数型プログラミングの「関数」を理解するためには、数学における「関数」を想像するとわかりやすいでしょう。
例えば三角関数を用いた y=cos(x) という式を考えてみます。この式に入力値 x=0 を与えた場合、いつでも必ず結果は y=1 になります。x= π/3 の場合は y = 0.5 ですね。この値は誰がいつどのような状況で計算しようが、必ず同じです。計算する環境によって値が変わったりしません。

同様に、関数型プログラミングにおける「関数」とは、 「入力した値に対して必ず同じ結果を返すモノ」 を意味します。CやJava等の手続き型（オブジェクト指向）言語における「一連の手続きをまとめて名前を付けたモノ」である関数（メソッド）とは概念が違いますので、注意が必要です。

関数型プログラミングにおける「関数」には、下記の特徴があります。

参照透過性：ある関数に同じ入力値を与えれば、いつ評価しても必ず同じ値が得られる。
- データベースや外部の可変な変数、コンソール入力など「状況によって値が変わる外部環境」に依存しない。
- データベースや外部の可変な変数、コンソール等へ値を出力して、外部環境の状態を書き換えない。
- 変数は一度定義したら二度と変えない（再代入操作で状態を書き換えない）。

第一級関数：言語の基本的な操作を制限なく利用できるモノ（第一級オブジェクト）として、関数を扱う。
- 変数に代入したり、別の関数の引数や戻り値になったりすることができる。
- 「関数の配列」のようなデータ構造に含めることができる。
- ある関数と別の関数を合成するなど、プログラム実行時に関数を生成することもできる。
- コレクションの要素に対して一斉にある関数を適用したい場合、関数が第一級である関数型プログラミングであれば、 結果のコレクション = 元のコレクション.map(要素に適用したい関数) のように、適用したい関数を引数にコレクション操作関数を呼び出すことで、一斉に処理を行わせることができる。
- 例えばJavaの場合、メソッド定義そのものを引数にしたり変数に代入したりはできないため、Javaのメソッド定義は第一級関数とは言えない。一方JavaScriptの"function"は、第一級関数と考えることができる。

結局、シンプルな関数を組み合わせて「引数の集合に関数を適用して結果を得て、その結果に別の関数を適用して、・・・」と関数の連結することで問題を解決するのが「関数型プログラミング」のスタイルとなります。
分岐と繰り返しと逐次処理を駆使して命令文を実行する「手続き型（オブジェクト指向）プログラミング」のスタイルとは、問題解決へのアプローチが全く異なるのです。

関数型と手続き型.png

関数型プログラミングスタイルのメリット

・・・関数型プログラミングってなんだか面倒で難しそうですが、大きなメリットがあります。「処理が簡潔になり共通化やテストしやすいこと」や「並列化しやすいこと」などです。

処理が簡潔になり共通化やテストしやすい

関数型プログラミングのスタイルで「関数」を書くということは、「入力を出力に変換するルールを記述する」ことを意味します。関数の外部にある可変な変数やデータベースの状態や他のスレッドの状態・・・などは関数の中身を実装する際、一切気にする必要がありません。
よって記述がシンプルになり、その関数に対するユニットテストも簡単に書くことができます。

並列化しやすい

ある関数を評価する（入力を出力に変換する）際、外部の状態には一切影響されません。そのため、例えば入力値の集合に対して一斉に何等かの処理を行う関数の場合、入力値の集合を分割して複数のスレッドにばらまいて処理を行わせ、結果を集めて出力しても全く問題が無いことになります。
よってプログラマがスレッド操作と同期処理を意識しなくとも、言語レベルで簡単に並列化が可能となるのです。

関数型プログラミングの妥協点

関数型プログラミングには素晴らしいパワーがありますが、一方で純粋な関数型プログラミングだけで実践的なアプリケーションを実装するのは、かなり面倒です。
実践的なアプリケーションは、必ず外界との情報のやり取りを伴います。画面との入出力であったり、セッションの状態であったり、データベースやファイル、連携システムとのI/Oであったり。これらは参照透過性を求める関数型プログラミングとは非常に相性が悪いのです。

よってこれからのプログラマには、「I/Oや状態を操作する副作用がある部分」と「参照透過でテストしやすく並列化しやすい部分」を上手く切り分け、副作用がある部分を局所化し、処理の根幹部分を関数型プログラミングスタイルで実装する能力が必要になるのです。

やってみよう！

と理論をいっぱい並べても楽しくありませんので、実際にプログラムを書いて実践してみましょう。今回は従来の「手続き型（オブジェクト指向）スタイル」で書いたJavaプログラムを、Java8で導入される関数型スタイルで書き直し、並列化して効果測定します。
Java8はHaskellのような純粋な関数型言語ではありませんが、Javaエンジニアが関数型プログラミングのスタイルを取り入れる第一歩としては十分な機能を有しています。

お題

関数型言語の "Hello World!" といえばフィボナッチ数列の計算ですが、実践的なプログラムとは言い難いのは確かです。
よって今回は、

ファイルシステムからテキストファイルを読み込む
形態素解析を行う
名詞・動詞・形容詞・副詞以外の単語を排除する
単語を辞書上の基本形に変換する（辞書に無い単語はそのまま）
- 例えば「走れ」「メロス」は、「走る」「メロス」になります
単語の出現頻度を数える
出現頻度が高い順にソートして出力する

というプログラムを記述することにします。形態素解析はそれなりに重い処理なので、マルチコアによる並列化の効果が計測しやすいのではないか？という目論みです。

環境

以下の環境でプログラムを書き、timeコマンドによって処理時間を計測しました。

CPU	AMD Phenom™ II X6 1065T Processor （6コア 2.9GHz）
memory	16GB
OS	Ubuntu 12.04.2 LTS (64bit) 3.5.0-36-generic
形態素解析エンジン	Kuromoji 0.7.7
ビルドツール	Apache Maven 3.0.5
Java7	ORACLE Java SE 7u25 1.7.0_25-b15
Java8	JDK™ 8 Early Access Releases 1.8.0-ea-b96

Kuromoji はJavaで実装された形態素解析エンジンです。Apacheの全文検索ツールキット Apache Lucene 、及び全文検索プラットフォームの Apache Solr にも日本語解析エンジンとして組み込まれています。Kuromojiは辞書も内包しているため、Mavenリポジトリから取得したjarを組み込むだけでJavaプログラムから形態素解析を行うことができ、非常に便利です。

また解析対象として、青空文庫で公開されている宮本百合子氏の道標を12回連結し、20mb程度にしたテキストファイル用いました。

ソースコードの公開

ソースコード（及びビルドコンフィグ等）は、githubにて公開しています。完全なソースコードはココから取得してください。

手続き型（オブジェクト指向）スタイル

では、Java7で従来通りの手続き型（オブジェクト指向）スタイルでお題を実装してみます。

ソースコード（のコア部分）

完全なソースコード

/* Java7 非並列化版 */
public class App {
  private static List<Item> wordCount(String file, String enc) {
    Tokenizer tokenizer = Tokenizer.builder().build();
    List<String> text;
    try {
      // テキストファイルを読み込みメモリ展開
      text = Files.readAllLines(Paths.get(file), Charset.forName(enc));
    } catch (IOException e) {
      throw new RuntimeException(e);
    }

    Map<String, Integer> words = new HashMap<String, Integer>(); // 形態素解析した結果を格納するMapを準備
    for (String line : text) {
      for (Token token : tokenizer.tokenize(line)) { // 各行ごとに形態素解析を実施
        String pos = token.getPartOfSpeech().split(",")[0]; // 品詞を取得
        if ("名詞".equals(pos) || "動詞".equals(pos) || "形容詞".equals(pos) || "副詞".equals(pos)) { // 名詞・動詞・形容詞・副詞 のチェック
          String word = (token.isKnown()) ? token.getBaseForm() : token.getSurfaceForm(); // 辞書にあれば基本形、なければ字句そのままを取得
          if (!words.containsKey(word)) {
            words.put(word, 1); // 最初に出現した場合はカウント１で結果に登録
          } else {
            words.put(word, words.get(word) + 1); // 二回目以降はカウントを＋１
          }
        }
      }
    }

    List<Map.Entry<String, Integer>> sortedWords = new LinkedList<Map.Entry<String, Integer>>(words.entrySet()); // ソート用にMapをLinkedListへ変換
    Collections.sort(sortedWords, new Comparator<Map.Entry<String, Integer>>() {
      public int compare(Map.Entry<String, Integer> o1, Map.Entry<String, Integer> o2) {
        return o2.getValue() - o1.getValue(); 
      }
    }); // 出現頻度の降順でソート実施

    List<Item> result = new ArrayList<Item>();
    for(Map.Entry<String, Integer> entry : sortedWords) {
      result.add(new Item(entry.getKey(), entry.getValue()));
    } // ソート結果をListに格納

    return result;
  }
  // 結果のコンテナクラス
  private static class Item {
    public String word;
    public int count;
    public Item(String word, int count) {
      this.word = word;
      this.count = count;
    }
  }
}

/* Java7 非並列化版 */

public class App {

private static List<Item> wordCount(String file, String enc) {

Tokenizer tokenizer = Tokenizer.builder().build();

List<String> text;

try {

// テキストファイルを読み込みメモリ展開

text = Files.readAllLines(Paths.get(file), Charset.forName(enc));

} catch (IOException e) {

throw new RuntimeException(e);

}

Map<String, Integer> words = new HashMap<String, Integer>(); // 形態素解析した結果を格納するMapを準備

for (String line : text) {

for (Token token : tokenizer.tokenize(line)) { // 各行ごとに形態素解析を実施

String pos = token.getPartOfSpeech().split(",")[0]; // 品詞を取得

if ("名詞".equals(pos) || "動詞".equals(pos) || "形容詞".equals(pos) || "副詞".equals(pos)) { // 名詞・動詞・形容詞・副詞のチェック

String word = (token.isKnown()) ? token.getBaseForm() : token.getSurfaceForm(); // 辞書にあれば基本形、なければ字句そのままを取得

if (!words.containsKey(word)) {

words.put(word, 1); // 最初に出現した場合はカウント１で結果に登録

} else {

words.put(word, words.get(word) + 1); // 二回目以降はカウントを＋１

}

List<Map.Entry<String, Integer>> sortedWords = new LinkedList<Map.Entry<String, Integer>>(words.entrySet()); // ソート用にMapをLinkedListへ変換

Collections.sort(sortedWords, new Comparator<Map.Entry<String, Integer>>() {

public int compare(Map.Entry<String, Integer> o1, Map.Entry<String, Integer> o2) {

return o2.getValue() - o1.getValue();

}

}); // 出現頻度の降順でソート実施

List<Item> result = new ArrayList<Item>();

for(Map.Entry<String, Integer> entry : sortedWords) {

result.add(new Item(entry.getKey(), entry.getValue()));

} // ソート結果をListに格納

return result;

}

// 結果のコンテナクラス

private static class Item {

public String word;

public int count;

public Item(String word, int count) {

this.word = word;

this.count = count;

}

いかにもJavaっぽいプログラムですね。

ビルドと実行

maven-assembly-pluginを用い、以下のコマンドで単独で実行可能なjarを作ります。（具体的なpom.xmlは、コレを参照してください）

$ mvn package
$ mvn assembly:single

1 2	$ mvn package $ mvn assembly:single

実際に実行してみると、以下のような結果が得られました。

$ time java -Xms8192m -Xmx8192m -XX:NewSize=4096m -jar target/nonpar-0.1-jar-with-dependencies.jar ../../docs/dohyo_twelvefold.txt Shift-JIS
（省略）
押し : 12
廃寺 : 12
花街 : 12
計量 : 12
嬰児 : 12
あきの : 12
碑 : 12
画境 : 12
絹織物 : 12
沈鬱 : 12

real	0m32.304s
user	0m33.730s
sys	0m2.240s

$ time java -Xms8192m -Xmx8192m -XX:NewSize=4096m -jar target/nonpar-0.1-jar-with-dependencies.jar ../../docs/dohyo_twelvefold.txt Shift-JIS

（省略）

押し : 12

廃寺 : 12

花街 : 12

計量 : 12

嬰児 : 12

あきの : 12

碑 : 12

画境 : 12

絹織物 : 12

沈鬱 : 12

real 0m32.304s

user 0m33.730s

sys 0m2.240s

見事に１スレッドしか動いていません。そりゃそうですね。

関数型スタイル（非並列化）

では次に、Java8から導入された関数型のスタイルで同じ処理を実装してみます。

ソースコード（のコア部分）

完全なソースコード

/* Java8 非並列化版 */
public class App {
  private static List<Item> wordCount(String file, String enc) {
    Tokenizer tokenizer = Tokenizer.builder().build();
    List<String> text;
    try {
      // テキストファイルを読み込みメモリ展開（ここはJava7と同じ）
      // 副作用があるファイルI/Oをこの部分に局所化している
      text = Files.readAllLines(Paths.get(file), Charset.forName(enc));
    } catch (IOException e) {
      throw new RuntimeException(e);
    }

    // 以下の部分は、変数への再代入も外部へのI/Oも行わず、参照透過性を保ったまま
    // 実際の処理を行なっている
    List<Item> result = text.stream() // Stream型へ変換
      .flatMap(line -> tokenizer.tokenize(line).stream()) // 各行を形態素解析して結果をStream型へ変換
      .filter(token -> 
        Arrays.asList("名詞", "動詞", "形容詞", "副詞").contains(token.getPartOfSpeech().split(",")[0])) // 名詞、動詞、形容詞、副詞のみを抽出
      .map(token -> (token.isKnown()) ? token.getBaseForm() : token.getSurfaceForm()) // 各単語を、辞書にあれば基本形、なければ字句そのままに変換
      .collect(Collectors.groupingBy(String::toString)) // 同じ単語でグループ化
      .entrySet().stream() // Key=単語、Value=単語のList へ変換しStream型へ
      .map(entry -> new Item(entry.getKey(), entry.getValue().size())) // Itemクラスへ変換
      .sorted((l, r) -> r.count - l.count) // 出現頻度の降順でソート
      .collect(Collectors.toList()); // 最終的な結果をJava本来のListへ格納

    return result;
  }
  // 結果のコンテナクラス
  private static class Item {
    public String word;
    public int count;
    public Item(String word, int count) {
      this.word = word;
      this.count = count;
    }
  }
}

/* Java8 非並列化版 */

public class App {

private static List<Item> wordCount(String file, String enc) {

Tokenizer tokenizer = Tokenizer.builder().build();

List<String> text;

try {

// テキストファイルを読み込みメモリ展開（ここはJava7と同じ）

// 副作用があるファイルI/Oをこの部分に局所化している

text = Files.readAllLines(Paths.get(file), Charset.forName(enc));

} catch (IOException e) {

throw new RuntimeException(e);

}

// 以下の部分は、変数への再代入も外部へのI/Oも行わず、参照透過性を保ったまま

// 実際の処理を行なっている

List<Item> result = text.stream() // Stream型へ変換

.flatMap(line -> tokenizer.tokenize(line).stream()) // 各行を形態素解析して結果をStream型へ変換

.filter(token ->

Arrays.asList("名詞", "動詞", "形容詞", "副詞").contains(token.getPartOfSpeech().split(",")[0])) // 名詞、動詞、形容詞、副詞のみを抽出

.map(token -> (token.isKnown()) ? token.getBaseForm() : token.getSurfaceForm()) // 各単語を、辞書にあれば基本形、なければ字句そのままに変換

.collect(Collectors.groupingBy(String::toString)) // 同じ単語でグループ化

.entrySet().stream() // Key=単語、Value=単語のList へ変換しStream型へ

.map(entry -> new Item(entry.getKey(), entry.getValue().size())) // Itemクラスへ変換

.sorted((l, r) -> r.count - l.count) // 出現頻度の降順でソート

.collect(Collectors.toList()); // 最終的な結果をJava本来のListへ格納

return result;

}

// 結果のコンテナクラス

private static class Item {

public String word;

public int count;

public Item(String word, int count) {

this.word = word;

this.count = count;

}

Streamとは、List等のJava Collectionを関数型スタイルで使えるようにする仲介型です（遅延コレクション化と内部イテレータの提供など）。Stream以外にも、ラムダ式（無名の関数を記述する記法）やmap, collectといった高階関数（関数を引数や戻り値にする関数）など、従来のJavaでは見慣れないモノがたくさんあるかと思います。

一方で、随分すっきりと処理内容が記述できていることにも驚かされます。
二重ループや二重のIF、Comparatorを使ったソートなど「何をするのか（How）」をステップごとに記述していた従来のソースコードとは異なり、「入力されたコレクションをどのように変換するのか（What）」を連結して処理が記述されています。ループやIFはどこにもありません。
これこそが、シンプルな関数を連結して問題を解決する関数型プログラミングのスタイルとなります。

※ Java8はまだAPIが固まっていないため、Java8のバージョンによってはこのソースコードはそのまま動作しない可能性があります。ご了承ください。

ビルドと実行

Java7と同様の手順でビルドします。
この際Java8を使うようにPATHを設定するだけでなく、Java8から追加された記法を有効にするために、 pom.xml のmaven-compiler-pluginのsourceとtargetを1.8にすることを忘れないでください。

実際に実行してみると、以下のような結果が得られました。

$ time java -Xms8192m -Xmx8192m -XX:NewSize=4096m -jar target/nonpar-0.1-jar-with-dependencies.jar ../../docs/dohyo_twelvefold.txt Shift-JIS
・・・
押し : 12
廃寺 : 12
花街 : 12
計量 : 12
嬰児 : 12
あきの : 12
碑 : 12
画境 : 12
絹織物 : 12
沈鬱 : 12

real	0m32.511s
user	0m35.842s
sys	0m2.224s

$ time java -Xms8192m -Xmx8192m -XX:NewSize=4096m -jar target/nonpar-0.1-jar-with-dependencies.jar ../../docs/dohyo_twelvefold.txt Shift-JIS

・・・

押し : 12

廃寺 : 12

花街 : 12

計量 : 12

嬰児 : 12

あきの : 12

碑 : 12

画境 : 12

絹織物 : 12

沈鬱 : 12

real 0m32.511s

user 0m35.842s

sys 0m2.224s

やはり同時には１スレッドしか動作していないことがわかります。

関数型スタイル（並列化）

最後に、Streamを並列化してみます。

ソースコード（のコア部分）

完全なソースコード

/* Java8 並列化版 */
public class App {
  private static List<Item> wordCount(String file, String enc) {
    Tokenizer tokenizer = Tokenizer.builder().build();
    List<String> text;
    try {
      text = Files.readAllLines(Paths.get(file), Charset.forName(enc));
    } catch (IOException e) {
      throw new RuntimeException(e);
    }

    List<Item> result = text.stream().parallel() // StreamをParallel化する
      .flatMap(line -> tokenizer.tokenize(line).stream().parallel()) // ここでもStreamをParallel化する
      .filter(token -> 
        Arrays.asList("名詞", "動詞", "形容詞", "副詞").contains(token.getPartOfSpeech().split(",")[0]))
      .map(token -> (token.isKnown()) ? token.getBaseForm() : token.getSurfaceForm())
      .collect(Collectors.groupingByConcurrent(String::toString))
      .entrySet().stream().parallel() // ここでもStreamをParallel化する
      .map(entry -> new Item(entry.getKey(), entry.getValue().size()))
      .sorted((l, r) -> r.count - l.count)
      .collect(Collectors.toList());

    return result;
  }

  private static class Item {
    public String word;
    public int count;
    public Item(String word, int count) {
      this.word = word;
      this.count = count;
    }
  }
}

/* Java8 並列化版 */

public class App {

private static List<Item> wordCount(String file, String enc) {

Tokenizer tokenizer = Tokenizer.builder().build();

List<String> text;

try {

text = Files.readAllLines(Paths.get(file), Charset.forName(enc));

} catch (IOException e) {

throw new RuntimeException(e);

}

List<Item> result = text.stream().parallel() // StreamをParallel化する

.flatMap(line -> tokenizer.tokenize(line).stream().parallel()) // ここでもStreamをParallel化する

.filter(token ->

Arrays.asList("名詞", "動詞", "形容詞", "副詞").contains(token.getPartOfSpeech().split(",")[0]))

.map(token -> (token.isKnown()) ? token.getBaseForm() : token.getSurfaceForm())

.collect(Collectors.groupingByConcurrent(String::toString))

.entrySet().stream().parallel() // ここでもStreamをParallel化する

.map(entry -> new Item(entry.getKey(), entry.getValue().size()))

.sorted((l, r) -> r.count - l.count)

.collect(Collectors.toList());

return result;

}

private static class Item {

public String word;

public int count;

public Item(String word, int count) {

this.word = word;

this.count = count;

}

streamに.parallel()を適用し、並列処理対応のStreamにしているだけです。なんとこれだけで、JVMが並列可能な部分を判断し、勝手にマルチスレッドで動作します。関数型バンザイ！

ビルドと実行

ビルド方法も実行方法も、非並列化バージョンと全く同じです。何も気にする必要がありません。

$ time java -Xms8192m -Xmx8192m -XX:NewSize=4096m -jar target/par-0.1-jar-with-dependencies.jar ../../docs/dohyo_twelvefold.txt Shift-JIS
・・・
パンルヴェ : 12
ユタン : 12
菌 : 12
とりちがえる : 12
安気 : 12
カカーヤ : 12
手飼 : 12
厚手 : 12
スケール : 12
アジト : 12

real	0m10.826s
user	0m50.107s
sys	0m1.288s

$ time java -Xms8192m -Xmx8192m -XX:NewSize=4096m -jar target/par-0.1-jar-with-dependencies.jar ../../docs/dohyo_twelvefold.txt Shift-JIS

・・・

パンルヴェ : 12

ユタン : 12

菌 : 12

とりちがえる : 12

安気 : 12

カカーヤ : 12

手飼 : 12

厚手 : 12

スケール : 12

アジト : 12

real 0m10.826s

user 0m50.107s

sys 0m1.288s

実行結果をみると、すべてのコアが使われていることがわかります。また実処理時間（real）に対してCPUのユーザモード処理時間（user）が5倍ほどあることからも、複数コアで同時処理が行われていることが見て取れます。

非並列化の場合と比較すると、実処理時間（real）は1/3に減りました。並列化できない部分（分割された単語をgroupbyで集約している部分など）もありますし、並列化すること自体にもオーバーヘッドがあるため、6コアあっても1/6まで処理時間が削減できるわけでは無いのでしょう。逆にCPUのユーザモード処理時間（user）は1.4倍程度まで増えています。これはStreamを分割する処理や結果を結合する処理、及びスレッド自体を扱う処理などのオーバーヘッドのためだと思われます。

いずれにしろ、「.parallel()」と三箇所書いただけで実処理時間が1/3に減るというこの結果は、関数型プログラミングのパワーを実感させてくれます。関数型バンザイ！！

一方で、処理結果の単語の順序は非並列化バージョンと同じになりませんでした。複数コアに分割して形態素解析し、結果を集めてくる、という処理を行なっているため順序が異なったのだと思われます。Listの順序が重要な場合、注意すべきポイントとなるでしょう。

まとめ

いかがだったでしょうか。関数脳になれそうでしょうか？
・・・え？Java8は2014年までリリースされないじゃないかって？？
わかりました。次回はJava7のjvm上でも動作する代表的な関数型言語である、ScalaとClojureの並列化について、紹介しましょう。お楽しみに！