Application Crash Consistency and Performance with CCFSを読んだ

FOLIOアドベントカレンダー 9日目です。 昨日は quantroさんの バリュー投資とかグロース投資とかの整理 でした。

NewSQLを調べつつ最近はストレージ周りにも手を出してみたいバックエンドエンジニアの @matsu_chara です。

今回はFAST '17のBest PaperであるApplication Crash Consistency and Performance with CCFS を読んだので紹介や感想を書きたいと思います。 自分が面白いと思ったところを中心にしたり、個人的に調べた内容が入っていたりするので正確かつ完全な情報は上記論文や関連文献等を直接読んでいただければと思います。

ざっくり

アプリケーションレベルでのCrash Consistencyの正確性を向上させるための仕組みを持ったファイルシステム CCFS(Crash-Consistent File System)の提案です。

重要な情報や機能を持ったアプリケーションでは突然の電源断やカーネルのバグなどによるクラッシュに対してロバストであることが求められます。(例:DBのトランザクション

ところがアプリケーションレベルでは、クラッシュに備えた処理に問題がある実装がよくあります。1 これらの間違った実装による影響はファイルシステムの振る舞いによっては影響を軽減したり、無くしたりすることが出来ます。2

ファイルシステムでクラッシュを考える際は、処理のatomicityとorderingが重要となります。

一貫性の観点からは、この2つが保証されていることが重要です。 このうち、atomicityについては既存のファイルシステムでも上手くサポートされています。

しかしorderingを保証しようとすると性能上のボトルネックになることがあるため、モダンなファイルシステムでは強力な保証は行わず制約を緩めていることが多いです。

例えばext4, xfs, btrfsなどではwriteの順番を並べ替えることがありますし、btrfsではディレクトリ操作の順番が入れ替わることがあります。

このような順序の入れ替えはシーク時間の削減につながるなど性能の面では有意義です。 一方でアプリケーションから見ると、書き込み順が入れ替わることにより生じる問題をケアすることが求められるため、誤った実装をしてしまいやすくなるという問題があります。

書き込み順の入れ替わりによって生じる問題の多くはテストが難しく、設定によっては正しくリカバリーされない実装のアプリケーションが多く存在します。

本論文では性能とordering保証を両立させるための鍵となるStream abstractionと、Stream abstractonを実装したファイルシステムであるCCFSが提案されています。

Stream abstractonの特長は以下のようなものです。

  • ファイルシステムレベルのorderingを少ないオーバーヘッドで提供
  • ユーザーコードの変更を最小限に抑えて利用を開始可能
  • さらにコードを変えると、さらなる性能を得られるといった柔軟性

CCFSの特長は以下です。

  • ext4をベースにStream abstractionを実装したファイルシステム
  • ext4と比べ遜色ない高性能を達成
  • アプリケーションに対し、より強力なcrash consistencyを提供

Background

前述したとおり、クラッシュを考慮する際のファイルシステムの性質を考える上ではatomicityとorderingの2つが重要となります。

一口にatomicity, orderingと言っても、「atomicityはシステムコールレベルで?セクタレベルで?」、「orderingはメタデータのみ?それとも全て?」といった議論すべき点がいくつかあります。

次のセクションでは、どのようなatomicity, orderingが適切かを検討していきます。

理想的な挙動

論文では"ordering", “weak atomicity”が達成されていれば、既存のアプリケーションの大部分が上手くリカバリーできるだろうとされています。(The Ordering Hypothesis)

この仮説は具体的には以下のようになります。

  • 全てのファイルシステムに対する更新はin-orderで行われる(ordering)
  • 書き込みはセクタ単位でatomicに、その他は全てatomicに行われる。(weak atomicity)

これらの制約はアプリケーションで考えなければいけない状態の数を制限するために存在します。

例えば Nセクタに対する更新がある場合、orderingとweak atomicityがあればクラッシュ時に取りうる状態はNです。 一方で並べ替えがある場合、クラッシュ時に取りうる状態は一気に 2N まで増加します。

実際のファイルシステムを見てみると、ext4, btrfs, xfsなどのモダンなファイルシステムでは"weak atomicity"については既に提供されています。 一方でorderingについての保証は稀です。(ext4, ext3のData-journaling modeでは提供されている) 保証が稀なのはorderingの保証を行うことによって性能が減ってしまうことに起因しています。

performance overhead

orderingを保証する場合の性能低下の原因としてFalse ordering dependenciesが挙げられます。

次節ではこのFalse ordering dependenciesについて検討していきます。

False ordering dependencies

例えば全く関係のないApplication A, Application Bが同時に動いているとします。 この時、以下のような動作が順番に起こることを想定します。

  1. Application Aで数百MB程度の大きなwriteがある
  2. Application Bで数Byte程度の小さいwriteを行った後、fsyncを行う

Application Bは通常一瞬でfsyncが終わりますが、orderingがグローバルに保証されているファイルシステムではApplication Aのwriteが終わるまで書き込みを開始することが出来ません。 そのため、結果として長い時間がかかってしまいます。

一方でApplication Bの一貫性の観点からはApplication Aの書き込みを待つ必要はありません。 たしかにApplication Aの書き込みのほうが順番としては早いですが、この2つはそもそも関係のないアプリケーションなのでそれぞれの書き込み順序の保証は本来不要のはずです。

このような不要な順序依存がFalse ordering dependenciesの正体です。 グローバルに順序を保証するファイルシステムではこのような不要な順序依存があちこちで起こってしまい性能が低下してしまいます。3

Stream abstraction

このような性能低下を防ぐために、Stream という概念を導入します。

具体的にはまず、各アプリケーションで行われる更新をいくつかのStreamに分割します。 そして、1つのStreamの中での更新は全て順序どおりに行われることを保証します。

例えば先程の例で、Application AではStream A, Application BではStream Bを使うと言った具合にStreamを分割します。 順序が保証されるのは同一Streamの中だけであるため、Stream Aが大きなデータをwriteしていても、Stream Bが順番待ちになることはありません。

このようにアプリケーション側でFalse ordering dependenciesを回避できるようにするための仕組みがStreamです。

Streamを作成するためには set_stream() を利用します。 Streamを利用するプログラムが既存のコードと異なる点は writeの前に set_stream(A)set_stream(B) を呼び出す点だけです。 set_streamを呼び出すと、その後の更新は全て指定したStreamに含まれるようになります。4

基本的にはアプリケーション起動時に、そのアプリケーション用のStreamを作成することで不要な順序依存をある程度取り除くことが可能です。 また、Streamはアプリケーション内に複数存在しても良いので、例えばスレッドごとにStreamを割り当てたり、一つのスレッドがStreamを切り替えたりするような最適化も可能です。

この柔軟性により、「粒度の大きいStreamを用意することにして既存のコードは殆ど変えずに安全性を保証したい」といった要求や「細粒度のStreamを用意して適宜使い分けて性能を向上させたい」といった要求など、様々なケースに対応出来るようになります。

CCFS

CCFSはext4のdata-journaling modeをベースに上記のStreamの概念が取り入れられたファイルシステムです。

ext4のjournaingではメモリ上に "Running Transaction", ディスク上に"journal"を保持します。 更新はRunning Transactionに保存され、コミット時にjournalに保存されます。

CCFSではTransactionの保存領域をStreamごとに分け、コミット時にはStream単位でjournalに格納します。 このようにすることで、全ての更新の順序を保存するのではなくStreamごとに順序を保証することが可能になります。5

Evaluation

実装の試験としてext4とCCFSの比較が載っています。

まず、クラッシュ時の挙動に関する試験です。 ext4とCCFSで発見された問題は以下のようになります。

ext4では9個の問題が発見されましたが、CCFSでは2つ(Mercurialでdirstateが破損)に留められました。

ext4 CCFS
LevelDB 1 0
SQLite-Roll 0 0
Git 2 0
Mercurial 5 2
ZooKeeper 1 0

また性能面での試験も行われています。 詳細は割愛しますが性能面も同等レベルになっています。 性能比較の結果は論文Figure 6などを参照してください。

まとめ

streamごとの順序を保証することでアプリケーションを堅牢に保つCCFSを紹介しました。 本記事ではあっさりした内容になっていますが、論文ではStreamを導入したときの様々な最適化について詳細が記してあります。それらを紹介しようとするとext4で行われている最適化についても述べる必要があり、記事が長大になってしまうので割愛しました。 Block Level Journalingやdelta journaling、Pointer-less data structuresなどの面白い要素がたくさん紹介されているのでぜひ論文を読んでいただければ幸いです。


  1. 例えばファイルシステムは書き込みの順番を入れ替えることがありますがアプリケーションレベルでのfsyncによる制御などが順序の入れ替えを考慮しきれていない場合に間違った実装となります。この場合クラッシュしたタイミングによってはアプリケーションが起動しないなどの問題に発展する可能性があります。 詳細は https://www.usenix.org/node/186195 で。

  2. filesystemの種類・設定で決まります。最悪ケースでは60種の問題が発見されるようなテストでも、良い種類・設定では10種に収まったという結果が論文で報告されています。

  3. 不要な順序依存の他にも、順序入れ替えによる最適化が行えないといった問題もあります。CCFSには、これらについての工夫もたくさん取り込まれていますが今回は割愛しています。

  4. 後方互換性を保つためにset_streamは古いファイルシステムに対しては何もしません。そのためstreamが導入されたアプリケーションは古いファイルシステムでも正常に動作します。(もちろんその場合は順序の保証は行われません)

  5. 同じブロックを異なるStreamで同時に更新した場合にどうするのかみたいな話はあるのですが、その辺はがっつり省いています。気になる方は論文をお読みください。

quillのio free monadがマージされてたので試してみた

この辺の事柄です。

quillにだいぶ前からIO monad入れようというissueが立っていたのですが、先日めでたくマージされたようです。 まだリリースされてないのでSNAPSHOTをビルドして試してみました。

環境などはこちら: https://github.com/matsu-chara/quill-free-example

機能とか

現状のquillだとqueryをrunした時点で副作用が発生してしまうので、うまく切り離したいというのが目的のようです。 また、クエリをtransaction内で実行させる機能やRead/Writeなどの副作用の種類を型で表現するためのEffect trackingの機能が付いています。

詳しくは GitHub - getquill/quill at edaa68bf438b0ca861d9e2bdb0893e22a8dcf70a

使い方

まず普通のバージョン

def findById(id: Long)(implicit ctx: MysqlJdbcContext[SnakeCase]): Option[Person] = {
  import ctx._
  val q = quote {
    query[Person].filter(_.id == lift(id))
  }
  run(q).headOption
}

そして、IO版。といっても runrunIO に置き換えただけです。(返り値の型が変わっているのに注目)

def findById(id: Long)(implicit ctx: MysqlJdbcContext[SnakeCase]): IO[Option[Person], Effect.Read] = {
  import ctx._
  val q = quote {
    query[Person].filter(_.id == lift(id))
  }
  runIO(q).map(_.headOption)
}

IO を呼ぶときは performIO を呼んでやればOKです。

以下のようにIOにtransactional Effectをつけてやれば、performIOを呼んだときにTransaction内で実行されるようです。 https://github.com/matsu-chara/quill-free-example/blob/2e2ad58e0970d56bfbc28bf05ecb0feb65949710/src/main/scala/matsu_chara/quill_free/Main.scala#L44

performIOの第二引数であるtransactional: Booleanをtrueにしてもtransactionで実行されるようです(内部的に Transaction() で包まれたIOがあったら transaction = true になる。)

Transactionalに実行しないと正しく動かないqueryをperformIOする場所で忘れないようにtransactionに包むのではなく、IO自体で表現できるのはうれしいですね。

Effect tracking

readmeに以下のリンクが参照されていました。 Put your writes where your master is: Compile-time restriction of Slick effect types - Daniel Westheide

ReadならSlave or Masterにクエリを向けたいけど、Writeを間違ってSlaveに向けて実行してしまうと困るので型で表現して守ろうという趣旨です。

参考:ReadWriteに対するpermissionを型で表現するといえばfujitaskなどもあります。

quillのIOにもEffectがあるので参考までに同じようなものを作ってみました。 https://github.com/matsu-chara/quill-free-example/blob/2e2ad58e0970d56bfbc28bf05ecb0feb65949710/src/main/scala/matsu_chara/quill_free/quill/RoleDb.scala

以下のようなWriteを含むクエリは roleDb の型が RoleDb[Master] なら通りますが、RoleDb[Slave]ではコンパイルエラーになります。

val personFreeRepository = new PersonFreeRepository
val ioOp = for {
  _ <- personFreeRepository.deleteAll() // Write
  _ <- personFreeRepository.insert(Person(id = 1, state = 0)) // Write
  p <- personFreeRepository.findById(1) // Read
} yield p
val personOpt = roleDb.roleBasedPerformIO(ioOp.transactional) // Read with Write with Transaction

もちろんEffect.Readだけを含むqueryならRoleDb[Slave]で実行することができます。

val ioOp = for {
  p <- personFreeRepository.findById(1) // Read
} yield p
val personOpt = roleDb.roleBasedPerformIO(ioOp) // Read

https://github.com/matsu-chara/quill-free-example/blob/2e2ad58e0970d56bfbc28bf05ecb0feb65949710/src/main/scala/matsu_chara/quill_free/Main.scala#L73-L76

WriteならMasterといった条件だけではなくReadAndWriteならTransactionalに実行しなければならない、みたいな条件まで細かく表現できるので良さそうです。

感想

好きなEffectを追加して、面白い制約(上述の記事だとExpensiveReadなどが挙げられていました。)を加えられないかなと思ったんですがsealed traitなので追加できなさそうでした。 quill/IOMonad.scala at edaa68bf438b0ca861d9e2bdb0893e22a8dcf70a · getquill/quill · GitHub

個々の機能を見ると良いんですが、インターフェースがqullの IO になってしまうので、quill依存をある程度閉じ込めて使いたい人はどうにかする必要がありそうです。

IO部分だけ切り出してすごく小さな安定したライブラリになれば依存するという判断も良さそうに感じますが、そもそも作り的にquillの色々なものに依存しているのでそのままだと色々不都合が生じそうな気がします。(まだリリースされていないSNAPSHOT版に対する感想なのでリリース版では欠点も含めて色々変わっている可能性があります。)

今後 IO 一本になる感じではなさそう?(明言はされているコメントは見当たらなかったので個人的にそういう気配を感じているだけ)ですが、しばらく様子を見ようかなと思っています。

FinagleのConnectionPool

すぐに忘れるので参照できるように雑にまとめ

Clients — Finagle 6.44.0 documentation とかを見るとすべて書いてあるので、基本的には公式を見たほうが良いです。 実装を読んだものも含んでいるのでfinagle 6.45以外で成り立つかは不明です。

種類について

以下の4種類がある。

  • Buffering Pool
  • Watermark Pool
  • Caching Pool
  • SingletonPool

Buffering Pool

finagle/BufferingPool.scala at finagle-6.45.0 · twitter/finagle · GitHub

bufferSizeを1以上にするとONになるデフォルトはOFF。 Watermarkの方が小回りがきくので基本的には使わなくてよいはず・・?

Watermark Pool

finagle/WatermarkPool.scala at finagle-6.45.0 · twitter/finagle · GitHub

low(デフォルト0), high(デフォルト Int.MaxValue)maxWaiters(デフォルト Int.MaxValue)の3つを設定して使う。

基本的にはlow ~ highの区間でコネクションを保持するもの。 最初のコネクションは0個なので常にこの区間にあるわけではなく、コネクションを作るときにlow個まではコネクションを保持し、それ以上は保持しないという動作。

high個以上のリクエストについてはそもそもコネクションを作らずwaitersキューに入れるという動作になっている。 このときlowを超えた分は基本的に随時connectしたりcloseしたりされるがwaiterが居る場合はcloseせずに使いまわす。 また、waitersキューに入れる際にキューサイズがmaxWaitersを超えていたらキューには追加せずTooManyWaitersExceptionが発生するようになっている。

Caching Pool

finagle/CachingPool.scala at finagle-6.45.0 · twitter/finagle · GitHub

WatermarkPoolだけだとlowを超えた瞬間に接続したり切断したりが繰り返されてしまう。 これを防ぐためにidleTimeに従って時間ベースでコネクションをキャッシュするのがCachingPool。 idleTimeというだけあって、コネクションが活用されている間はevictionされないようになっている。 ただし、idleTimeは目安であって実際には[ttl, ttl * 2)のレンジでevictされる様子

キャッシュされる個数はhigh - low個になっている。 デフォルトだとLong.MaxValueナノ秒キャッシュされるので、適宜調節しても良いかもしれない。

Singleton Pool

finagle/SingletonPool.scala at finagle-6.45.0 · twitter/finagle · GitHub

コネクションプールと言いつつ一つのコネクションのみを保持するプール ThriftMuxはコネクションを多重化しているのでプールいらないよねという事情により存在する。

使われ方

DefaultPoolを見ると、BufferPool/WatermarkPool/CachingPoolは組み合わせて利用するようになっている模様。 BufferPool(defaultオフ) => WatermarkPool => CachingPool のように順番にキャッシュ判定などがされるようになっている。(SingletonPoolは別枠)

どの条件で使われるかについては現状だと以下のようになっている

プール名 ONになる条件 備考
BufferPool bufferSize > 0 defaultでは0で、基本的に変える必要はなさそう。
WatermarkPool true 常に有効になっている
CachingPool idleTime > 0.seconds && high > low defaultだとLong.MaxValueナノ秒保持される

役立つmetrics

Metrics — Finagle 6.44.0 documentation を見るのが良い。 pool_waiterspool_num_waited を見ると待機キューが使われているかどうかなどを把握できる。 ちなみに pool_cached はキャッシュされているコネクションの数を表すが、現在使われているコネクションはこの中には入らないようなのでサーバーに負荷がかかっている状態では0になったりするっぽい。(cachからgetするとdequeからpopされるためsizeが減る)

手軽に自分用のショートカットが定義できるgolを作った

github.com

go-linkというものが前職にあって、それのパクリです。(ただし運用はしたくないのでlocalhostで動かす前提にした やっぱりみんなで使いたくなったのでdocker-compose up -dだけで運用できるようにした)
ワークフローがこれに依存しすぎていて無いと精神が不安定になるので作りました。

go get github.com/matsu-chara/gol で入ります。

golとは

urlに名前をつけられるブクマツールのようなものです。

  • あの雑に立てたツール(https://our-server02:9534)のポートいつも忘れるな・・・
  • あのよく見るコンフルのあのページ( https://confluence.nice-company.com/pages/viewpage.action?pageId=xxxxxxx )ってどこで見れるんだっけ・・・

といった事に悩まされる人用です。 以下のように登録しておくと gol open $key で簡単にページを開くことが出来ます。

gol add myproduct_admin_prod 'https://our-server02:9534'
gol add myteam_docs 'https://confluence.nice-company.com/pages/viewpage.action?pageId=xxxxxxx'
gol open myproduct_admin_prod # => https://our-server02:9534 が開く
gol open myteam_docs # =>  https://confluence.nice-company.com/pages/viewpage.action?pageId=xxxxxxx が開く

ブラウザ連携

gol server を立てた上でchrome検索エンジンに登録するとgol[tab]myteam_docs[enter]でリンクに飛べるのでshell操作なしで使えます。(むしろこっちが本来の使い方)

こんな感じ https://github.com/matsu-chara/gol/raw/master/sample/gol_chrome2.png

READMEには書いてないですが golm という名前で http://localhost:5656/myteam_%s のように登録すると golm[tab]docs[enter]だけで飛べます。(疑似名前空間) リンクが増えてきたらこの使い方もおすすめです。

さくっと作ったので穴が多い(ファイルに吐いてるので同時編集すると死ぬとか、ブラウザからは登録できないとか)ですが、個人で使う分には十分・・? サーバー立ててみんなで使ったりする場合はもうちょっと作り込む必要があると思います。CLI部分消してweb-ui作ればいいかな-と思います。web-ui作るの面倒なのでCLIにしたという経緯

go-linkだからgoでしょ!と思ってgoで書いたけど、goの作法わかってないところが多々ありそう

TwitterのFutureについてのざっくりまとめ

基本的に以下からそのまま取ってきています。

説明用のざっくりポインタとしてまとめる予定だったのに、あれもこれもと欲張ってしまった代物。(その割に全部あるわけではない。) playからfinagleに移行してきたりするとメソッド名が細かく違ったりするのに最初は戸惑いますが、Scala標準Futureとakkaのschedulerを使ったことがあれば、すぐ慣れつつ便利さを感じられると思います。

例ではThread.sleepを呼びまくっていますが、そのへんの事情はScala標準のFutureと同じなのでちゃんとやるときはFuture.sleepやtimer.doLaterなどを使ったほうが良いです。

目次

Futureの作成

基本的な作り方について。 この作り方だとapplyも含めて全部同期実行になる点に注意。

value/exception

import com.twitter.util.Future

// 基本の作り方
Future.value(1)
Future.exception(new RuntimeException)

apply/const

Scala標準Futureではapplyは非同期に実行されるが、TwitterFutureでのapplyは同期実行という違いに注意。

import com.twitter.util.Future

// applyを使うとTryで包むのでReturn/Throwに仕分けてくれる
// Scala標準のFutureと異なり、ただTryで包むだけ == 同期実行な点に注意。
Future(1)
Future(throw new RuntimeException)

// twitter Tryからの変換
Future.const(Try(1))
Future.const(Try(throw new RuntimeException))

Futureをまとめる

Scala標準のFutureと同じくfor式を使うだけだと並行実行されないことがあるので注意。

map/flatMap

join

ScalaFutureのzipと同じ http://qiita.com/mtoyoshi/items/f68beb17710c3819697f#zip コード例は略。

失敗のハンドリング

handle/rescue

import com.twitter.util.{Future, Return, Throw}

val successFuture = Future.value(1)
val failedFuture = Future.exception[Int](new RuntimeException)

// handle
// 失敗した例外をSuccessにできる。 failedに対するmap
// caseはPartialFunctionなのでcaseにマッチしない例外はそのまま例外として扱われる。
successFuture.handle {
  case e: RuntimeException => 0
}
failedFuture.handle {
  case e: RuntimeException => 0
}

// rescue
// 失敗した例外をSuccessにしたり別の例外に変換できる。 failedに対するflatMap
// caseはPartialFunctionなのでcaseにマッチしない例外はそのまま例外として扱われる。
successFuture.rescue {
  case e: RuntimeException => Future.value(0)
}
failedFuture.rescue {
  case e: RuntimeException => Future.value(0)
}

transform

import com.twitter.util.{Future, Return, Throw}

val successFuture = Future.value(1)
val failedFuture = Future.exception[Int](new RuntimeException)

// transform
// rescueと異なり成功時の値も同時に変換できる
successFuture.transform {
  case Return(a) if a == 1 => Future.exception(new RuntimeException)
  case Return(a) => Future.value(a)
  case Throw(e: RuntimeException) => Future.value(0)
  case Throw(e) => Future.exception(e)
}
failedFuture.transform {
  case Return(a) => Future.value(a)
  case Throw(e: RuntimeException) => Future.value(0)
  case Throw(e) => Future.exception(e)
}

// transformedByというメソッドもあるが、こちらはFutureTransformerを受け取る。
// FutureTransformerはJavaFriendlyと書いてあるので基本的にはtransformを使えば良い。
// 今回は割愛。

FuturePool

非同期実行したい場合はFuturePoolの力が必要。

FuturePool.unboundedPool

スレッドプール内部のExecutorServiceはglobalのもの(https://github.com/twitter/util/blob/util-6.45.0/util-core/src/main/scala/com/twitter/util/FuturePool.scala#L70-L72) が利用される。

import com.twitter.util.FuturePool

// 非同期実行されるプール
// 処理の実行方法はFuture.applyと同じくpoolのapplyに処理を渡せばOK
// unboundedなので際限なく拡張される。
val unboundedPool = FuturePool.unboundedPool
unboundedPool(1)
unboundedPool(throw new RuntimeException)

FuturePool(dbExecutorService).apply

import java.util.concurrent.ForkJoinPool

import com.twitter.util.FuturePool

// 非同期実行されるプールを自分で作る。
val dbExecutorService = new ForkJoinPool(50)
val myPool = FuturePool(dbExecutorService)

myPool { Thread.sleep(1); 1 }
myPool(throw new RuntimeException)

// interruptibleUnboundedPoolというキャンセルに対応したPoolもある。

Timer系列

Timerの種類を例ごとに変えてあるが使い方はどれも同じ。Timerの種類については後述。

sleep

import com.twitter.util.{Future, Await}
import com.twitter.conversions.time._

implicit val timer = com.twitter.finagle.util.DefaultTimer

// 3秒後にUnitが返る
val f = Future.sleep(3.seconds)

Await.result(f)

delayed

完了が遅れるだけで計算自体はすぐ行われる点に注意。

import com.twitter.util.{Future, Await}
import com.twitter.conversions.time._

implicit val timer = new com.twitter.util.JavaTimer(isDaemon = true)

// 3秒後にIntが返る
val f = Future {
  println("in future")
  1
}.delayed(3.seconds).foreach(_ => println("done"))

Await.result(f)

Timer#schedule

Futureのメソッドではないがついでなので紹介。

import com.twitter.util.{Await, Future, Time}
import com.twitter.conversions.time._

implicit val timer = new com.twitter.util.ScheduledThreadPoolTimer(poolSize = 5, makeDaemons = true)

// schedule 1秒ごとに何度も実行する 。キャンセル可能なTimerTaskを返す。
val timerTask = timer.schedule(1.seconds) {
  println("1sec!")
}

Thread.sleep(3000)
Await.result(timerTask.close())

// doLater 2秒後に1回実行する。Futureを返す。
val f1 =  timer.doLater(2.seconds) {
  println("2sec!")
}
Await.result(f1)

// doAt 具体的な時刻を指定する。Futureを返す。
val f2 =  timer.doAt(Time.Now + 3.seconds) {
  println("3sec!")
}
Await.result(f2)

timerの話

tl;dr;

  • finagleを使っているならcom.twitter.finagle.util.DefaultTimerを使えばOKだが、blockingな重い処理をするなら自前で定義したほうがよいかも。
    • finagle-netty4が依存パスにあればNetty4HashedWheelTimerが使われる。
    • 無くてもJavaTimer(isDaemon=true)が使われるので安心。
  • finagleは使って無くてtwitter/utilだけ使っている場合はJavaTimerかScheduledThreadPoolTimerが選択肢
    • 単に new JavaTimer とするとユーザースレッドとして起動するため、明示的にcancelを呼ばないとtimerがGCされるまでプロセスが終了しなくなることがあるのでisDaemon=trueを指定するとよい。

細かい話

twitter/utilのこの手のメソッドはcom.twitter.util.Timerを要求してくる。

スケジューリングを無視して即時実行するNullTimerやテスト用のMockTimerがあるが、 finagleを使っていない場合、基本的にはJavaTimer(isDaemon=true)を、ある程度の性能が欲しい場合はScheduledThreadPoolTimerを使えば良さそう。

finagleを使っている場合は、もう少し性能が出るタイマーがDefaultTimerとしてfinagle自体に用意されているのでそちらを使う方が良さそう。 ただ、DefaultTimerのインスタンスは共通なのでblockingな処理を行う際は、そのスレッドがブロックされる可能性があるので分けたほうが良いかもしれない。(未検証)

com.twitter.finagle.util.DefaultTimerの実装はServiceLoaderで一番最初に見つかったクラスを使う。 ServiceLoaderで見つからなかった場合はwarningログが出つつcom.twitter.util.JavaTimerをdaemonThreadをONにした上で使うようになっている。 https://github.com/twitter/finagle/blob/finagle-6.45.0/finagle-core/src/main/scala/com/twitter/finagle/util/DefaultTimer.scala#L31-L36

finagle-netty4(finagle-{thrift, http}の依存にある)がNetty4HashedWheelTimerを指定しているので大抵の場合はこれが読み込まれる気がする。  https://github.com/twitter/finagle/blob/finagle-6.45.0/finagle-netty4/src/main/resources/META-INF/services/com.twitter.finagle.util.DefaultTimer

もう一つcom.twitter.finagle.util.HashedWheelTimer.Defaultというややこしいものが存在するが、こちらはNetty3ベースのHashedWheelTimerを使っている。 Netty3,4間のHashedWheelTimerの差はよくわかっていないが新しい方が良さそうなので基本的にはcom.twitter.finagle.util.DefaultTimerを使うのが良いだろう。

Timerのロードにサービスローダーを使うようになったのはfinagle6.45から。経緯とかは https://github.com/twitter/finagle/commit/d047b4568e07a56b481b5f7c193b0e8c5ec6ff71 のコミットに書いてある通り、finagle-coreからnetty3依存を剥がすためにそうなっているらしい。

複数の処理

select/or

import com.twitter.util.{Future, Try}

// selectは一番最初に終わったFutureの値と残りのFutureを返す。
val fs = (1 to 5).map(Future(_))
val (firstDoneTry, others): (Try[Int], Seq[Future[Int]]) = Await.result(Future.select(fs))
println(firstDoneTry) // Return(1)
println(others) // Seq(Future(Return(2)), Future(Return(3)), Future(Return(4)), Future(Return(5)))


// orはselectの2つ版。selectと異なり、先に終わった値が含まれるFutureのみを返す。
Future(1).or(Future(2))

// selectIndexという一番最初に終わったSeq[Future[A]]のindexを返すメソッドもあるが割愛。

traverseSequentially/collect/collectToTry

traverseSequentiallyは前のFutureが終了してから次を実行し、collectは同時に実行する。

実装を見るとcollectでも結果に含まれる要素の順番は引数と同じになるっぽい。順序についてはドキュメントやコメントにはのってない気がする。

collectToTryはcollectと異なり一部が失敗したときも成功し、一連の結果がTryで取得できる。

import com.twitter.util.{Await, Future}

// 順番に実行される
val f = Future.traverseSequentially(1 to 5) { i =>
  Future(i)
}

// 並列に実行される
val f = Future.collect((1 to 5).map { i =>
  Future {
    val wait = scala.util.Random.nextInt(300)
    println(s"$i => $wait ms")
    Thread.sleep(wait)
    i
  }
})
println(Await.result(f)) // 順序は同じ。ArraySeq(1, 2, 3, 4, 5)

// 並列に実行される。失敗も補足できる
val f2 = Future.collectToTry((1 to 5).map { i =>
  Future {
    if(i % 2 == 0) throw new RuntimeException else i
  }
})

println(Await.result(f2)) // 順序は同じ。ArraySeq(Return(1), Throw(java.lang.RuntimeException), Return(3), Throw(java.lang.RuntimeException), Return(5))

batched

大量の非同期処理をいくつかのグループに分けて実行する君

同時実行数の制限という意味だと com.twitter.concurrent.AsyncSemaphore を使う手もある。 Futures — Util 21.5.0 documentation

batchedは複数リクエストをまとめて送るのが前提になっている(Seq[In]=> Seq[Out]で処理を記述する)などの違いがある。

import com.twitter.conversions.time._
import com.twitter.util.{Await, Future}

implicit val timer = new com.twitter.util.JavaTimer(isDaemon = true)

// まずbatcherを作る
val batcher = Future.batched[Int, String](
  sizeThreshold = 10, // 実行しきい値は10。正確にはsizeThreshold*sizePercentile個のジョブがエンキューされるまで実行を待つようになっている。
  timeThreshold = 500.milliseconds, // sizeの条件を満たさなくてもenqueueからtimeThresholdを超過したらジョブが実行される
  sizePercentile = 0.3f  // sizeThresholdと合わせて最低ジョブ実行数を決める。
                         // この例では固定値を入れているが名前渡しになっているのでRandom.nextFloat()とか入れるとバッチサイズを都度変化できるようになっている。
                         // 必要がなければ指定しないでデフォルト値(1.0f)を使えば良さそう。
) { ids =>
  // Seq[In]を受け取って、Seq[Out]を返す関数を書く
  Future.sleep(scala.util.Random.nextInt(50).milliseconds).map { _ =>
    println(s"${ids.mkString(", ")} are inserted")
    ids.map(_.toString)
  }
}

val userIds = 1 to 50

// applyに渡してジョブをenqueueする。中でsynchronizedするのでcollectで呼ぶ意味はあまりない。
// バッチグループごとにsleepを入れたりする機能は無さそうなので、enqueue自体のタイミングで制御すると良さそう。
val insertedFuture = Future.collect(userIds.map(batcher)) 

println(Await.result(insertedFuture))

// thresholdに関係なく全リクエストの実行を開始したいときはflushBatchを呼ぶ。
batcher.flushBatch() 

// sizeThreshold = 1でuserIdsを10個投入した場合の出力
// 8 are inserted
// 5 are inserted
// 2 are inserted
// 6 are inserted
// 3 are inserted
// 4 are inserted
// 1 are inserted
// 10 are inserted
// 9 are inserted
// 7 are inserted
// ArraySeq(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)

// sizeThreshold = 100でuserIdsを10個投入した場合の出力
// timeThreshold秒経過してから以下が出力される。
// 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 are inserted
// ArraySeq(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)

// sizeThreshold = 20でuserIdsを10個投入した場合の出力(sizePercentile=0.3fなので20*0.3=6個ずつ実行される。
// ただし7~10個目は数が足りないのでtimeThreshold秒経過してから出力される。
// 1, 2, 3, 4, 5, 6 are inserted
// 7, 8, 9, 10 are inserted
// ArraySeq(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)

便利const

import com.twitter.util.Future

// 以下の2つは同じ
Future.Unit
Future.Done

// 便利定数
Future.True
Future.False
Future.None
Future.Nil
Future.???

// Java用なので使わなくて良い
Future.Void

callback

コールバックよりはmap/flatMapでつなぎたい。ensureはありかも。

onSuccess/onFailure

省略

respond/ensure

import com.twitter.util.Future

// respond
// 完了した際のコールバックを設定する。
// 主にライブラリなどの汎用コード向けとコメントにある。
// respondは結果表示やリソースの後始末などの副作用を起こす前提となる。(promiseを解決したりしているコードもちょくちょく見かける。)
val f1 = Future.value(1)
val f2 = f1.respond {
  case Return(a) => println(a)
  case Throw(e) => println(e)
}

// ensure
// respondとほぼ同じだが、引数として計算の結果を受け取らない。
// 成功しても失敗してもいいから単にリソースをcloseしたい場合などに使える。
val f1 = Future.value(1)
val f2 = f1.ensure {
  println("f1 finished")
}

キャンセル

raise

import com.twitter.util.{Await, Future, FuturePool}

// Future.valueは即時評価なのでraiseできない。
// またinterruptibleUnboundedPoolを使っても、state=Doneになるとraiseを呼んでも正常系の値が返ってくるのでsleepでごまかしている
val f1 = FuturePool.interruptibleUnboundedPool {
  Thread.sleep(100)
  1
}
f1.raise(new RuntimeException("interrupt"))
Await.result(f) //  java.util.concurrent.CancellationException

// FuturePool.unboundedPoolを使う場合はFuture#interruptibleを使うとinterruptできるようになる。逆にinterruptibleを呼ばないとcancel出来ない。
val f2 = FuturePool.unboundedPool {
  Thread.sleep(100)
  1
}.interruptible()
f2.raise(new RuntimeException("interrupt"))
Await.result(f2)

raiseWithin

N秒以内に終わらないとタイムアウトといった指定が出来る。

import com.twitter.util.{Await, Future, FuturePool}
import com.twitter.conversions.time._

implicit val timer = new com.twitter.util.JavaTimer(isDaemon = true)

val f = FuturePool.interruptibleUnboundedPool {
  Thread.sleep(3000)
  1
}

// 2秒後にraiseされる
f.raiseWithin(2.seconds)

Await.result(f)

within/by

winthinとbyはDurationを受け取るかTimeを受け取るかの違いしか無い。

raseWithinとwithin/byには処理自体のfutureをraiseするか、withinなどの呼び出しの返り値のみをraiseするかの微妙な違いがある。詳細は以下のコメントを参照。

import com.twitter.util.{Await, Future, FuturePool}
import com.twitter.conversions.time._

implicit val timer = new com.twitter.util.JavaTimer(isDaemon = true)

val f1 = FuturePool.interruptibleUnboundedPool {
  Thread.sleep(3000)
  1
}

// 2秒後にf2がraiseされるが、raseWithinではf1,f2の両方raiseされるのに対し、within/byはf2のみがraiseされるためf1自体の結果は普通に取得することが出来る。
val f2 = f1.within(2.seconds)

println(Await.result(f1)) // 1が表示される

Await.result(f2)

monitored

Promise使いつつネストしていると辛くなるケースを救えるらしい。いまのところ使ったことは無い。

import java.util.concurrent.ForkJoinPool
import com.twitter.util.{Future, FuturePool, Try, Return, Throw, Promise}

import scala.util.control.NonFatal

val exception = new RuntimeException("test")

// 以下のようなケースを考えると、notMonitoredは決して終了しない。
val inner1 = new Promise[Int]
val inner2 = new Promise[Int]

val notMonitored: Future[Int] = {
  inner1.ensure {
    throw exception
    inner2.update(Return(2))
  }
  inner2
}

// このようなケースを防ぐために内部で起きた例外を伝搬してくれるのがFuture.monitored
val monitored: Future[Int] = Future.monitored {
  inner1.ensure {
    throw exception
    inner2.update(Return(2))
  }
  inner2
}

// before
inner1 // state=Waiting
inner2 // state=Waiting
notMonitored.poll // None
monitored.poll // None

inner1.update(Return(1))

// after
inner1 // state=Done
inner2 // state=Interuppted
notMonitored.poll // None (永久に終わらない)
monitored.poll // Some (例外が伝搬されるので終わる)

おまけ: Futureの再帰

ドキュメントにちゃんと実装してあるから再帰してもサンプルにあるようなコードではスタックオーバーフローにならないよと書いてある。

Futures — Util 21.5.0 documentation