🔬

重い初期化、並列処理, Singletonの罠, そしてInstancePoolへ

2022/06/11に公開

はじめに

初期化処理が重いインスタンスってありますよね。代表的な例だとJDBCコネクションとか。Factoryパターンで初期化を隠蔽するような奴は概ね重いのはどの言語でも一緒だと思います。
こういった重い処理を例えばループなどで大量に初期化してしまうのはかなりのコストになってしまい、特にバッチのようなチリツモな処理では無視できないほど実行時間に影響を与えてしまう恐れがあります。
そういった場合はSingletonなどを使ってインスタンスの生成コストを隠蔽しますが、マルチスレッド環境下だと影響が出る場合もあります。以前、そういうケースでバグってるのに出くわした事があるので、記憶の整理がてらまとめておこうと思います。

コード例はJavaですが、あまりJavaかどうかには依存しない考え方になっていると思うので、他の言語が得意な人は疑似コード程度に思っておいてください。

なお、JDBCを始めとした一般的に重いとされているインスタンスには枯れたPoolの実装が通常はあるので、そちらを使う方が良いでしょう。

一番最初の素朴な実装

さて、まずはベースとなるサンプルコードを示しましょう。1から100までのループの中でCounterクラスを初期化して、2回ほどインクリメントした後に合計値を出しています。Counterの初期値は0なので合計値は200を返します。一応計測用に三回処理を回してる感じです。

privatestaticvoidinvoke(){int total=IntStream.range(0,100).map(i->{var counter=newCounter();                    counter.increment();                    counter.increment();return counter.get();}).reduce((xs, x)-> xs+ x).getAsInt();System.out.println("Total: "+ total);}publicstaticvoidmain(String[] args){System.out.println("param-size: "+ForkJoinPool.commonPool().getParallelism());for(int i=0; i<3; i++){long s=System.nanoTime();System.out.println("Loop: "+ i);invoke();long e=System.nanoTime();System.out.println(((e- s)/1000/1000)+" ms");}}

呼び出しているCounterクラスはこんな感じ。実際は何か色々してて初期化とincrementが重いのですが、とりあえずSleepを入れています。ファイルアクセスとかDBアクセスとか外部APIとかそういうのです。たぶん。

publicclassCounter{privateint count=0;publicCounter(){try{System.out.println("Counter:create:"+this.hashCode());Thread.sleep(300L);}catch(InterruptedException ex){            ex.printStackTrace();}}publicvoidincrement(){try{Thread.sleep(10L);}catch(InterruptedException ex){            ex.printStackTrace();}        count++;}publicvoidclear(){        count=0;}publicintget(){return count;}}

それではこのクラスを実行してみましょう。単純な処理ですが初期化とincrementにそれなりに時間が掛かるので結構時間を使ってしまいます。

Loop: 0Counter:create:245257410Counter:create:1705736037Counter:create:455659002～中略～Total: 20034199 msLoop: 1～中略～Total: 20034187 msLoop: 2～中略～Total: 20034150 ms

どの処理も34秒程度で終わり、合計値が200ですね。中略していますが、毎回インスタンスが生成されているのが分かると思います。

それでは、この処理を高速化していきましょう。

並列処理による高速化

まず最初に考えるのはマルチスレッドを使った並列処理による高速化ですよね？時代はメニーコア！ 1CPU/1Coreなんて今は昔です。我々もフリーランチに甘えるばかりではありません。まあ、若干クラウド環境とかだとコア数先祖返りしてるケースもあるけど、それはそれ。
Javaには多くの並列化の仕組みがありますが、今回はもっともお手軽な並列ストリームを使います。

int total=IntStream.range(0,100).parallel().map(i->{var counter=newCounter();    counter.increment();    counter.increment();return counter.get();}).reduce((xs, x)-> xs+ x).getAsInt();System.out.println("Total: "+ total);

parallelって足すだけだから簡単ですね！この書き方だと並列度は実行環境に依存しますが、私のマシンでは11並列になります。実行結果は以下の通り。

param-size: 11Loop: 0Counter:create:1364922542Counter:create:1048596955Counter:create:1178848130...～中略～Total: 2003048 msLoop: 1～中略～Total: 2003044 msLoop: 2～中略～Total: 2003396 ms

3秒前後とかなり速くなりましたね！スリープ処理だから並列化の影響をダイレクトに受けて気持ちが良いですねー。実際はこんなにスケールしないケースも多いですが、それでも並列化の効果は大きいですよね。

それでも、まだ物足りない。もっと高速化を、というあなたは本題のインスタンス生成コストの省略を次の章で試しましょう。

シングルトン、並列処理、そしてバグ

そもそもインスタンスの生成に300msもかかっているので、これを毎回生成するのが何よりの無駄ですよね？無駄を省くのは改善の基本です。インスタンスの生成の数を抑えるテクニックと言えば定番はSingletonですね。さっそく実装してみましょう。

以下のようにCounterインスタンスを保持するSingletonを作ります。良くある書き方ですね？

publicclassCounterSource{privatestaticCounter counter=null;privateCounterSource(){}publicstaticCounterget(){if(counter==null){                counter=newCounter();}return counter;}}

続いて、直接newするのではなく、CounterSourceを経由して取得するようにコードを変更します。

int total=IntStream.range(0,100).parallel().map(i->{var counter=CounterSource.get();    counter.increment();    counter.increment();return counter.get();}).reduce((xs, x)-> xs+ x).getAsInt();System.out.println("Total: "+ total);

そして実行結果は以下になります。

param-size: 11Loop: 0Counter:create:781607405Counter:create:1955827078Counter:create:62803695Counter:create:1283928880Counter:create:1364922542Counter:create:582939092Counter:create:1048596955Counter:create:1184496519Counter:create:1060245958Counter:create:761384723Counter:create:552016626Counter:create:121354012Total: 7798606 msLoop: 1Total: 26265280 msLoop: 2Total: 45361294 ms

280ms～600ms前後と爆速になりましたが...奇妙な所がいくつかありますね？まずトータル値がおかしいです。200のはずが26265とか良く分からない値になっています。こんなとんでもないバグがあってはリリース出来ませんね。さあ、デバックの時間だ！

バグの原因を追え！

ここで勘の良い人は思うかもしれません。「あ、シングルトンだから状態を保持してるのでループ毎にcounterをクリアしなきゃダメなんだ！」 と。確かにそれは怪しそうですよね？早速修正してみましょう。counter.clear()でCounterSourceから取得する度に初期化しています。計算前に毎回初期化してるから必ずincrementの前はcount=0だと思いますよね？　よし、勝ったな。風呂入ってこよう。

int total=IntStream.range(0,100).parallel().map(i->{var counter=CounterSource.get();    counter.clear();    counter.increment();    counter.increment();return counter.get();}).reduce((xs, x)-> xs+ x).getAsInt();System.out.println("Total: "+ total);

さて、お風呂に行ってる間の実行結果がこちらとなります。

param-size: 11Loop: 0Counter:create:1184496519Counter:create:1364922542Counter:create:1955827078Counter:create:664380515Counter:create:295530567Counter:create:1878446100Counter:create:781607405Counter:create:552016626Counter:create:1060245958Counter:create:62803695Counter:create:1048596955Counter:create:121354012Total: 583597 msLoop: 1Total: 707311 msLoop: 2Total: 639322 ms

値がバグってるままですね！そんな、毎回初期化してるのに… こんなの絶対おかしいよ！と思ってしまうのも無理は無いかもしれません。
実は、counter.clear()が必要なこと自体はあっています。ただ、一手足りないのです。ここで注目したいのは画面に出ているCounter:create:...です。シングルトンなのに、なんで複数のインスタンスが出来ているのでしょう？？？この原因はCounterSourceがスレッドセーフでは無いためです。

if(counter==null){   counter=newCounter();}return counter;

上記のコードはnullの時に初期化という実装でシングルスレッドでは問題無いのですが、マルチスレッドでは複数スレッドから同時にアクセスしたときにnullを見たのにその後にそれぞれのスレッドが別のインスタンスを初期化するという現象が発生します。そのため、マルチスレッドで状態を変更するコードは必ずスレッドセーフに実装する必要があります。

という分けでスレッドセーフに対応させましょう。Javaではsynchronizedを使う事で並列アクセスの処理をシリアライズできます。利用側のコードは一切変更が要りません。簡単ですね。

publicclassCounterSource{privatestaticCounter counter=null;privateCounterSource(){}publicstaticsynchronizedCounterget(){if(counter==null){                counter=newCounter();}return counter;}}

さて、それでは結果を見ていきましょう。Counter:create:2003749087と想定通り、インスタンスが一つだけ生成されていますね。いやー、ちょっと苦労したけど楽勝でしたね！これで今日は旨い酒が飲めそうだ...って良く見たらトータル値がやっぱりバグってるというツラい現実が目に入ってきますね。ふう、今夜は長い夜になりそうだ....

param-size: 11Loop: 0Counter:create:2003749087Total: 423617 msLoop: 1Total: 755280 msLoop: 2Total: 852283 ms

インスタンスプール実装への道

さて、先ほどのバグの原因は何でしょうか？処理毎にclearも付けたし、スレッドセーフにもしたのにいったいどこに問題があったのでしょうか。実は原因はやはりスレッドセーフに対応していない事です。CounterSourceはスレッドセーフに実装したけど、Counter自体はスレッドセーフに対応してないままですね？synchronizedはあくまでそのメソッドをスレッドセーフにするだけで、その後の一連の処理の保証をするものではありません。

単純に考えると、Counterクラスをスレッドセーフに修正してしまうという案が思いつきます。それも一つの手ですが、これが自前のクラスでは無く標準ライブラリやサードパーティのライブラリだとどうでしょうか？　修正できませんよね？　こういったケースで役立つのがInstance Poolです。

Instance PoolはSingletonあるいはFlyweightパターンのように特定の数のインスタンスをstaticにプールしておいて、それを利用者の要求に応じて払い出す仕組みです。これによって、都度インスタンスの生成をする事を避けつつ、インスタンスが複数スレッドで同時に共有されることを防ぐのです。

Instance Poolで最も代表的なものはJDBC Connectionですね。所謂コネクションプールがこれにあたります。他にもEJBなんかでも使われていますし、他の言語でもそれなりによく使われるテックニックだと思います。

今回はかなりシンプルに実装してみました。まずCounterクラスを包むCounterWrapperを以下のように実装します。isActiveメソッドを持ち、この状態でプールから払い出されているか否かを判定します。takeでコネクションを取得して、releaseで解放ですね。Closeableの実装は必須では無いのですが入れとくと閉じ忘れを防げて便利です。

publicclassCounterWrapperimplementsCloseable{privateboolean isActive;privateCounter counter;publicCounterWrapper(){this.isActive=false;this.counter=newCounter();}publicvoidtake(){this.isActive=true;            counter.clear();}publicvoidrelease(){this.isActive=false;}publicCounterget(){returnthis.counter;}@Overridepublicvoidclose()throwsIOException{this.release();}}

つづいて、CounterSourceを以下のように変更します。変更というかもはや原型がほぼ無いですが...

Counterでは無く、CounterWrapperをプールとしてSet型で持ちます。そして、maxPoolSizeの数まではインスタンスを作成しますが、それ以上はしません。CounterSource#get()が呼ばれたら、poolに存在するアクティブでは無いインスタンスを返します。見つかるまで無限ループをさせているので、maxPoolSizeを超えたリクエストを受けた場合はリリースされるまで待ちます。

publicclassCounterSource{privatestaticSet<CounterWrapper> pool=newHashSet<>();privatestaticint maxPoolSize=3;publicstaticvoidsetMaxPoolSize(int maxPoolSize){CounterSource.maxPoolSize= maxPoolSize;}publicstaticsynchronizedCounterWrapperget(){if(pool.size()< maxPoolSize){var wrap=newCounterWrapper();            pool.add(wrap);};while(true){for(var x: pool){if(x.isActive==false){    x.take();return x;}}}}}

最後に呼び出し側も一部修正が必要になります。まずCounterSourceからCounterWrapperを取得して、その後takeメソッドで実際のCounterインスタンスを取得しています。リリース漏れを防ぐためにtry-catch-resourceの構文でCounterWrapperの取得を行っていますが、それ以外はほぼ同じコードですね。

int total=IntStream.range(0,100).parallel().map(i->{try(var wrap=CounterSource.get()){var counter= wrap.get();                counter.increment();                counter.increment();return counter.get();}catch(IOException ex){thrownewUncheckedIOException(ex);}}).reduce((xs, x)-> xs+ x).getAsInt();System.out.println("Total: "+ total);

それでは実行してみましょう。CounterインスタンスはCounterSourceのデフォルト値の3で作成され、性能が単なる並列処理の時より改善しつつも、トータル値が正しい事が分かります。

param-size: 11Loop: 0Counter:create:1324119927Counter:create:857327643Counter:create:1799308595Total: 2001953 msLoop: 1Total: 2001037 msLoop: 2Total: 2001049 ms

プールサイズを最適化してみましょう。今回のコードは11並列なのでプルーサイズが3だと待ち時間が発生してしまいます。そのため同数の11にプール数も変更する事で性能改善が期待できます。

CounterSource.setMaxPoolSize(11);int total=IntStream.range(0,100).parallel().map(i->{...

以下が実行結果です。かなり速くなりましたね！

param-size: 11Loop: 0中略Total: 2003666 msLoop: 1Total: 200314 msLoop: 2Total: 200312 ms

ただベンチマークの作りの都合上、最初のテスト実行にだけ初期化が入ってるので実際の性能改善の具合が少し分かりづらいので、ループ数を1万回に増やし有意差が分かりやすいようにしてみました。
結果は以下の通り。プール化だけでだいたい10倍以上の性能アップですね。これはループ回数、つまり一般的なバッチだとデータ量が増えれば増えるほど影響が大きくなります。なお、素朴なシーケンシャル処理での1万回は事前のテストから10倍程度の差と分かっているので想定値を入れています。終わるの待ってられないので...

実装	時間(sec)
素朴な実装	3000(推定)
並列化	321
プール化	34

表: 1万回当たりの実装別実行時間

グラフにするとより際立ちますね？

図: 1万回当たりの実装別実行時間

まとめ

とりあえず初期化が重いインスタンスを含むバッチの高速化をユースケースに、並列処理でSingletonを使うときの注意点とPoolingによる解決をまとめてみました。スレッドセーフの話はマルチスレッドが基本リアルタイム系ではお約束で、だからこそFWとかがサポートしてるんですが、バッチはそういうケースが少ないのでうっかり考慮漏れのコードが紛れ込む事があります。
特にシングルスレッドを想定していたコードを性能改善で並列処理に書き換えたりするケースでは、こういったどっかにシングルトンが紛れてて謎の挙動をするってのはありがちなので注意をしたいですね。共通ライブラリでSingletonとかも。