save/load

gistfile1.txt

ユースケース
============

 1. ヒューマンエラーなどの意図せぬ状態変更からの復元
 2. システムとしてのバックアップ
 3. システム状態のコピー(テスト用など)


想定される取得タイミング
========================

 1. ユーザーによる手動実行
 2. 自動実行


要求事項(goal)
==============

 * スナップショットからシステム状態を「安全に」復元したい
   * 「安全に」とは、機械学習的、システム的に正しく動作する状態ということ
   * client から keeper がリクエストを受け付けた瞬間の全ノードの状態を出力したい
      -> 完全性を保持するためには、クラスタ全体の update を止める?
      => [参考] HDFSのスナップショット
                 http://hadoop.apache.org/docs/r1.0.4/hdfs_design.html#Snapshots
                 http://hortonworks.com/blog/snapshots-for-hdfs/
                 https://issues.apache.org/jira/browse/HDFS-2802
                  -> 3.0.0 でリリース予定
 * 簡単な方が良い。


決めるべきこと
==============

 * 完全性の確保
    -> どこまでの完全性を確保するか
    -> ノード間の時間差は許容されるのか、時間差により問題は発生しないのか
    -> 不完全な snapshot と redoログ的なもの との組み合わせ?


スナップショット方法(案)
========================

A) 完全なスナップショットを取得する
   [1] save リクエスト実行
   [2] keeper から zookeeper のマスターロックを取得
      1) すべての keeper で update リクエストをブロック(saveは可能) (メンテナンスコードをレスポンス?)
      2) server 間の mix もブロック
   [3] ready-to-save 状態となる
   [4] save処理実行
      1) zookepper から membership 情報の取得/出力
      2) zookeeper から config 情報の取得/出力
      3) server から serialize した model の出力 (非同期)
   [5] マスターロック解除
   [6] finished-to-save 状態となる

  メリット: 特定のタイミングでの完全なスナップショットをとれる
  デメリット: 可用性を失う、update 性能へ影響を与える

B) 適切なスナップショットを取得する
   [1] save リクエスト実行
      1) zookepper から membership 情報の取得/出力
      2) zookeeper から config 情報の取得/出力
      3) server から serialize した model の出力 (非同期)

  メリット: 可用性を維持できる
  デメリット: update や mix により、特定のタイミングでの完全なスナップショットとはならない
              (運用で update を止めないと、完全性を保証できない)
              (mix を一時的に止める仕組みの提供も必要となる)


保存対象
========

 1. モデル ... 学習の状態
 2. fv_converter の状態
 3. config
     -> 分散構成時は、zk へ保存したものを各ノードで再ロードする
 4. クラスタ構成
     -> cht の場合、クラスタ構成に依存してデータの格納ノードが変わる
 5. その他、状態を保持するもの (global_id など)


保存先ディレクトリ構造(案)
==========================

 /datadir/snapshot-ID(or UUID)
　+ Jubatus_TYPE_NAME (下記は記録する情報、NAMEがない場合は、_もなし) ※形式はJSONが楽かも
  |  - バージョン ... loadする際に起動中のServerのバージョンと比較する
  |  - タイムスタンプ ... 特に利用しない。付加情報。
  |  - プロセス名 ... 付加情報
  |  - config ... loadする際に起動時に指定されたconfigと比較する
  |  - membership ... zk上のactor情報と比較する
　|  - global_id
  + model ... 現状の形式を踏襲
  + fv_converter

 ※同一の情報を利用することを保証するため、UUIDなど一意性を保証できるIDを付与する(現行の ID のままで、運用ルール化?)


対応スケジュール
================

 0.4.1 (or 2) : 対象情報を save できるようにする (案B)
 0.5.0 以降   : A案のような排他制御機構を実装し、完全なスナップショットを取得可能とする



過去の記述(メモ)
================

* やりたいこと
   + save: 
      - Jubatus のインメモリ情報を永続化する
      - Jubatus の処理性能へ影響を与えないよう実現する (プロセスを fork して出力?)
      - モデル情報を可視化したい (#178)
      - 定期実行するようなツールを用意する?
   + load:
      - 情報を読み込むことで、save したときの状態を復元する
      - 引数指定することで起動時に読み込むことを可能とする (#222)
      - config や model に変更があったことは確認する??

* 出力内容
  1. config ... 現状: json形式 (string, json_config)
    + 読み込んだ際の string型 のものであれば、get_config で取得可能 ... 起動時読み込みなので get_config で十分という認識
    + json型 に変換されたら unordered_map
       -> print or pritty で string型 等に取得はできる
    + classifier_serv_config のような構造体に変換されたものを json型 に戻す?
       -> 本来の意味での load された config はこれになる

  2. model ... 現状: pficommon::data::serialization::binary_iarchive で serialize されたもの
    + 各 storage で変換を実施
    + 解読可能にするには外側から デシリアライズ をする? 
    + strage ごとの構造体を人間が解釈可能なものへ変換する

  3. 状態を持つものは全体的に保存したい(global_id?,  )

  4. そのうち window データとか? 
    + どんな状態をどんな構造で持つのか?

* save/load 拡張案
  1. serialize したものをとりまとめて、出力する
  2. ディレクトリを切って、それぞれ別ファイルに出力する

* 可視化案
  1. save とは別に可視化用の口を作る(dump)
      -> format も検討
  2. save したものを可視化するツールを作る ... 難易度高
  3. save そのものをserialize ではなく、プレーンな形式にする

「まずはB案を元に実装して、必要ならA案も実装していく」に賛成します。

スナップショット方法 A についてですが、守りたいものをもう少し検討しなければいけない気がします。

すべての keeper で update リクエストをブロック(saveは可能)

• 「完全なスナップショット」というのが「クライアントの投げる 1 つの RPC リクエストを粒度として一貫性を保証した状態」であれば、すべての update リクエストで keeper が ZK ロックを取得する必要があるように思えます (ある keeper に save リクエストが到着した時点で、別の keeper が仕掛中の update リクエストが存在する可能性がある)。
• 「完全なスナップショット」というのが「Keeper の投げる 1 つの RPC リクエストを粒度として一貫性を保証した状態」（つまり、サーバ間ではモデルに一貫性がないかもしれない）であれば、現状の save でよいのではと思います (提案されている手法では、現状の save よりもちょっとだけ一貫性を保証できる可能性が上がってはいますが、上記の「仕掛中の update リクエストが存在する」問題があるので本質的には変わらない気がします)。

「完全なスナップショット」は前者の想定でした。
ある時間 t 以前の update 処理をすべて完了した状態。つまり、仕掛中の update がすべて終わった状態。

save におけるマスターロックを 仮に save-lock とする と、以下のようにしないといけないですね。

• save を受け取った keeper : save-lock へ write-lock を取得
• update を受け取った keeper: save-lock へ read-lock を取得、lock が取れない状態だと、update は reject

後者だと、ご指摘の通り B案 と同じですね。

最終的に目指すのは、 A案の「書込み中のもの」がない状態での save (完璧なスナップショット) ですが、実現すると、性能等への影響が大きそうですね。

membership ... zk上のactor情報と比較する

CHT で「IP アドレスからハッシュを生成しているので違う環境で復元できない問題」は、ストレージのリバランス機能が実装されれば自然に解決しそうですね (全部のデータがガガっと一気に移動するかもしれませんが)。

モデルの load に関しては、個人的な直感としては、config 同様に起動時のみロード可能とするのが筋が良いように思います (オンラインで load したい、という強いユースケースが無い限りは)。
クラスタの運用という観点でも、共有ストレージがない環境でオンライン load を行うとミス起こしやすそうです。
考慮不足だったので無視して下さい。

コメントを受けて、修正しました。(rev.21)

最終的に目指すのは、 A案の「書込み中のもの」がない状態での save (完璧なスナップショット)

これを実現するために「すべての update リクエストで keeper が ZK ロックを取得」すると、サーバをどれだけスケールさせても update リクエストが同時に 1 本しか走らないことになるので選択肢としてはダメな気がします。

A 案の実現ために、より現実的な選択肢として、update リクエストごとに keeper でユニーク ID を発番 (global ID generator) して、サーバ側でモデルの世代番号管理をするという手はありそうです (その場合は二相コミットで compare-and-save という感じ)。

A案、大きく分けて2種類の実現方法があります。ロックベースな物とノンブロッキングな物です。

ロックベースな手法は、分散システムではかなり実装コストが高いです。
何故ならロック抱え込み死を想定しないといけないからで、タイムアウト付きのゴツい実装するか、コーディネータが離脱する事に耐性のあるトランザクションを書く必要があり、労力半端ないです。

ノンブロッキングな物はkmaehashiさんの言うようにバージョン番号を振る方法です。
updateが走るたびにモデルのバージョン番号が単調増加するようにしておき
そのバージョン番号を取得するメソッド get_version()と、バージョン番号：モデルデータのペアを取得するget_verison_and_model()を実装して

• 全員にget_version()を投げてバージョン番号取得
• 揃ったらget_version_and_model()を投げてバージョン番号とモデルデータのペア取得
• 1回目と2回のバージョンが全て一致したら案Aの完璧なスナップショット取得成功
• バージョン番号の差の合計がk未満なら許容の案B（このkがどれぐらいが良いかは悩みどころ
• 差がk以上の場合はget_version_and_model()を再び投げて取得

という感じで全体のスナップショットは行けます。k=1とすれば案Aになりますし、k=∞でも実質困らない気もします。最後の判断はユーザーに委ねるなんて手もあって、rerely updateなユースケースならk=1でもまともに動くと思います。
kが1でも試行回数の上限をT回以下に抑えるというwait-free snapshotも研究としては存在しますが、大雑把に言うとupdateの度にsnapshotを取るという手法なので性能ペナルティ半端ないです。auto-savingするような場合を除いてあまり手を出したくないです。

それより問題なのは、各プロセスの持つデータがバラバラであることを設計に含んでいる物です。
recommenderのデコード情報とか。Jubatusの設計思想上、クライアントからサーバの台数や詳しい所在を意識させない事にしているので、Jubatusが1台でもn台でも同じsave結果である必要があると思います。100台のJubatusに学習させた結果を2台のクラスタ構成でloadする必要のあるケースもあり、load側は遅かれ早かれkmaehashiさんの言うようにリバランス機能が搭載される必要があります。CHTをJubatusプロセス側が認識しないといけない時期になってると思います。
save側は、MPI_Gatherのような通信を行わせて、それぞれで保存させれば良いと思います。

各々のノードで save させるのが得策なのかどうかという議論も必要かなと思いました。以下、メモレベルであまり練れていませんが。

• 各々のノードで save する (現行)
  • 100台でsaveして2台でloadというユースケースの解決方法が自明でない
  • ノード間で重複データ (レプリカ) が生じるのでディスク容量が無駄
  • ディスク起因で save に失敗するケースの考慮が必要
• 管理ノードが各々のノードからモデルを集約し、管理ノードに単一ファイルを save する
  • 台数を意識せずにloadできる
  • 重複データは生じないので省ディスク容量
  • サーバのローカルディスクの状態について考慮しなくてよい
  • モデルファイルのハンドリングが容易
  • NW 帯域のコストは高い

そうですね、可能であればconfigファイル同様にZK側に保存するのが筋かもしれません。
特に、モデルデータとconfig.jsonは対で保持しますが、load時には全部のマシンのconfigが一致する事を期待するので、その読み出し対象のファイルが各マシンローカルにあるというのも片手落ちな感じです。
一方で、学習モデルデータ、本気出すとギガ級になるのでほいっとZKに置けるわけではなさそうなのが悩みどころですね。
実装コストを度外視するならHDFSのような抽象的な分散ストレージ層を作って共有するのが一番良さそうな気がしてきました。

kumagiさん、kmaehashiさん、コメントありがとうございます。

ロックベースな手法におけるロックの抱え込み死の考慮、確かにそうですね。
zk上のロックは、ロック取得プロセスのダウンとともに解除される(はず)ですが、様々なパターンを想定した設計が必要です。

実装コスト、可用性を犠牲にすること(この点が一番の問題点で、利用ケースによっては update をブロックすることが許されないかもしれない)を考えると、ノンブロッキングな方式に倒すべきなのかもしれません。
※load の際のフローについては、検討不足ですが、全プロセスの load が終わるまで、update/mix をブロックする必要がある気がします。(むしろ、ブロックしなくてはいけない状況/ユースケースなので、許容されるのか?)

k=0 にすることができない update を頻繁に行うケースについては、k がある範囲になる形で諦めるしか無いのかもしれないです。(正しく動くという保証は...)

「1つの save結果であるべき」「管理ノードで集約して save する」ということについては、ユースケースの 3 と ディスクの問題の点では確かにその通りだと思います。
100台 save、2台 load に限らず、同一構成で load する場合も、ユースケース 3 の場合は、IPアドレスが違う可能性が高いので、リバランスが必要ですね。

[以下はあまり関係ないです]
現状の想定ユースケースは「ある時点 T の状態を復元する」という点ですが、「Jubatus に予期せぬ障害が発生しても、障害発生時点(最新の状態) の状態に復元可能」という点では、"場合によっては"、update の都度、永続化するような機構も必要とされてくるのかもしれません。(または、クラスタ内でのデータを多重度を選べるようにするとか)

zkが死活監視をするというの、実際のところ「zkに対して必ず指定時間以内に応答する」という条件を満たしているだけであって、たまたま知らずにzkへの応答が遅れて離脱と見做されアンロックされる場合を想定して処理を書くとなると更に面倒になります。特にデータセンタなどを跨ぐ将来を考えるとzkへの応答時間に制約を設けるのはあまり好ましくないなぁと考えます。

で、kがいくつずれても良いかという議論ですが、分散ストレージの議論を持ってくるなら「因果関係が壊れなければどこまでずれてもOK」というのが答えです。あるクライアントがA→B→Cの順番で一個ずつ保存していく過程なら、それに対するスナップショットは、[],[A],[A,B],[A,B,C]のどれかしか許されません。つまり例えば[B]というスナップショットを取ることはNGなわけです。が、A・BがCHTで別のサーバ（それぞれα・β）に割り当てられる場合に、それらを保存する場合に

• αを観測→[]　（これを①とする
• A,Bを順に保存。αはAを保持、βはBを保持する。
• βを観測→[B]　（これを②とする

という過程を経て、①と②を合成して[B]を全体のスナップショットとしてしまう事があるのは不味いです。
Jubatusの学習においてもそうやって外から因果関係を期待するケースはあると思うのでそこをどう許容するかですね。クライアント側にカウンタを付けてロックしましょうか…。

あとWait-free Snapshotですが、N回に1回の周期でupdate時にSnapshotを取ってから保存、という方法を取ればSnapshot側のstarvationを回避しつつパフォーマンスをある程度犠牲にせずに済む上国内学会の論文ネタぐらいにはなるんじゃないかと思うのですが微妙ですかね？

「zkに対して必ず指定時間以内に応答する」という条件を満たしているだけ

「コーディネータから離脱したがクラスタに参加している」というケースを想定する必要があるということでしょうか。
(すみません、こちらの理解が追い付いていません。。)

Jubatusの学習においてもそうやって外から因果関係を期待するケースはある

現時点では、updateリクエスト処理がすべて完了すればコンバージする（というか因果関係が保たれた状態になる）ことが保証されていて（＝「ゆるやかな一貫性」？）、これはこれで良いような気がしています。
一方で、因果関係が崩れた状態で save されたモデルを load してもコンバージョンが保証できないので、これは問題です。

今出ている中で A 案の実現方法を考えてみると、以下のような感じでしょうか。(松竹梅は実装コスト)

• 松案: cht IDの更新単位(row)とモデルに世代番号を付与 (ノンブロッキング)
  • モデルの世代番号は全ノードで共有されない(updateによってモデルが更新されるのはそのrowを持つノードのみ)なので、rowごとに世代番号を付与
  • rowは削除される可能性があるため、各ノードごとのモデルにも世代番号を付与
  • save中の可用性劣化なし
• 竹案: Keeperのupdateリクエスト処理区間をトランザクション化 + モデルに世代番号を付与 (ノンブロッキング)
  • save中の可用性劣化なし
• 梅案: ブロッキング(ロックベース)の手法
  • 単純に、mix/updateの新規開始/受付を禁止 -> 既存のmix/updateの完了待ち合わせ -> save
  • save中の可用性は劣化(analysisは継続可能)

Wait-free Snapshot

ユースケース次第ですが、頻繁(ないし自動的)なsave が必要であれば、あってもよいと思います。

先行して実装を行う 保存対象の拡張 という点について、 https://gist.github.com/6ae462fbe27aacb53e4c に整理中。

「コーディネータから離脱したがクラスタに参加している」というケースを想定する必要があるということでしょうか。

そうです。
実際には反応が遅れただけなのに離脱したと見做されて勝手にロックが開放される事まで考慮した実装をするのは面倒そうという。確率の問題なので運用でカバーというのは無理ではないのですが。

ちょっと思いついたのですが
linear_mixerの場合はmixの瞬間に全体の一貫性を持ったデータが手に入るわけで、数秒に1度の頻度でmixするなら「一番最近のmixデータ」だけは常時保存し続けても良いんじゃないだろうか？と。
save対象がどこまで最新でなくてはいけないかは再考の余地がありそうです。

mix のタイミングで取得

確かに、liner_mixer であれば、一貫性のとれたモデルデータが取得できますね。
気になるのは liner_mixer 以外の mix戦略 をとったときには?? という点ですが、A案の実現方法として候補に入れさせて頂きます。ありがとうございます。

save対象が最新でなくてはいけないか

jubatus の緩い共有というコンセプトから考えても、保証する範囲を再考する必要がありますね。

皆様から頂いたコメントをもとに、今一度、整理を行いたいと思います。

以下にまとめました。なかなかのボリュームになりました。
https://gist.github.com/4633358