Hatena::ブログ(Diary)

スティルハウスの書庫 このページをアンテナに追加 RSSフィード

2009-10-05

Memcacheでスピンロックを実装してTask Queue処理結果を集約してみるテスト

TaskQueueで分散処理した結果をまとめるときは、排他を考慮する必要があります。Datastoreを使う場合なら、単に結果を新規エンティティとして追加したり、エンティティグループの楽観排他を使ったりすればOKです。一方、やっぱりMemcacheでスピーディーに集約したいよ、という場合は、Memcache上で排他を実装する必要があります。以前のエントリにちょろっと書いたMemcacheService#incrementでスピンロックという方法について説明してほしいというコメントをid:miztakaさんよりいただいたので、ここに改めて書きたいと思います。

Memcacheでは排他制御が必要

ご存じのとおり、App Engineはデフォルトで複数のApp Serverによるクラスタが構成されており、またMemcacheサービスはクラスタ全体で共有されるグローバルなキャッシュとして機能します。よって、TaskQueueのように複数のリクエストを大量に並列処理し、それらがMemcache上の1つのエントリに値を書き込む場合は、何らかの方法で排他制御を行わないと問題が生じます。

例えば単純なカウンターを実装する場合、

  • Cache#get("counter")で現在の値を取得
  • 1たす
  • Cache#put("counter")で現在の値を書き込み

という一連の処理になります。RDBやDatastoreならこれ全体をトランザクションとすることで排他制御できますが、Memcacheにはそれがないのがやっかいです。2つのリクエストやタスクが上記コードをまったく同時に実行すると、どちらか一方のカウントが反映されず上書きされてしまいます。

MemcacheService#incrementでmutexを実装

ここで超役に立つ(というより唯一の頼みの綱)のが、MemcacheService#incrementメソッドです。MemcacheService#incrementは「キーで指定したMemcache上のLong値を読み取り、第2引数のLong値を加算し、Memcacheに書き込む」という操作をアトミックに実行します。よって、例えば2つのスレッドや2台のサーバーが、例えばincrement("counter", 1)をまったく同時に実行しても、それぞれの加算処理が必ず反映させる仕組みになっています(ちなみにこういうアトミックな演算ってたしかx86プロセッサの命令にもありましたよね)。

これを使って、以下のようなコードを書きました。ホントはコードをそのままコピペした方が分かりやすいのですが、そういうわけにも行かないので、大まかな流れだけ紹介します。

  • acquireLockメソッド(引数key:ロックを識別する文字列。戻り値:trueならロック取得OK)
    • MemcacheService#increment(key, 1)でlong値を取得
      • 例外出たら0で初期化してfalseを返す
    • l = 1ならtrueを返す
    • l != 0ならMemcacheService#increment(key, -1)し、falseを返す
  • releaseLockメソッド
    • MemcacheService#increment(key, -1)

acquireLockメソッドは、任意のキーで名付けたロック(これってmutexに分類されるとおもいますたぶん)を取得してtrueを返します。誰かが同じキーのロックを取得済みならば、falseを返します。これを使って、以下のようなコードを書きます。

  • aquireLock("counter_lock")でロック取得
  • Cache#get("counter")で現在の値を取得
  • 1たす
  • Cache#put("counter")で現在の値を書き込み
  • releaseLock("counter_lock")でロック取得

これで、上述の競合書き込みの問題はなくなります。

synchronized節っぽくして使いやすくする

とはいえ、このままではちょっと使いにくいです。

  • ロック取得に失敗しても、何回かはリトライしたい
  • ロック取得中に例外出たり、書き忘れたりしてreleaseLockを呼び忘れると永遠にロックされたままになりお客さまから怒られる

そこで、以下のようなメソッドを書きました。

  • runSynchronizedメソッド(引数key:ロック識別用文字列、Runnable:ロック中に実行したいロジックを実装)
    • acquireLockを呼び出し
      • ロック取得に失敗したら、500msスリープして、リトライ。10回くらい失敗したらあきらめて例外をthrow
    • Runnableのrunを呼び出し
      • try/finallyで括って、finallyの中でreleaseLock呼び出し

つまり、ロック取得に失敗しても、ループで10回くらいリトライするスピンロックです(OSカーネルの本物のスピンロックはスリープ挟まないと思うのでちょっと違いますが)。これを使うと、例えばカウンターの例は以下のように書けます。

runSynchronized("counter_lock", new Runnable() {
    public void run() {
        Cache#get("counter")で現在の値を取得
        1たす
        Cache#put("counter")で現在の値を書き込み        
    }
});

こうして、めでたくMemcacheを複数タスクや複数リクエストで共有して書き込み可能になりました。いまのところ快調に動いているようです。

takezakitakezaki 2009/10/06 14:17 おせわになっております。大変参考になる記事、ありがとうございます。現在、同様の排他制御を考えているのですが、Memcache の信頼性の低さを考慮すると意外と難しいことがわかって悩んでおります。
http://code.google.com/intl/ja/appengine/docs/java/memcache/overview.html#How_Cached_Data_Expiresに「アプリケーション側ではキャッシュの値が常に利用可能だと仮定すべきではありません。」とありますように、counterは有効期限やその他の理由で消えてしまうことを考慮しなければならないようです。(かといってDatastoreに保存するのもどうかと思いますが。)
 リクエスターが都度チェックして、もし消えていた場合に、counterの初期化するというロジックを入れたとすると、同時に初期化するものが存在してしまう可能性があり、排他制御が機能しなくなる可能性もあると思います。そのあたりをどう考えましょうか。

kazunori_279kazunori_279 2009/10/06 19:19 ええ、memcacheはあくまでキャッシュとして使うべきと考えます。今回私が適用した要件では、もしmemcacheに保存した値が消えた場合は全体の処理をやり直すことになります。もっとも、これまで繰り返しテストしてきた中では、memcacheに保存した値が消える現象はまだ見たことがありません。

より信頼性が必要な用途には、memcacheとdatastoreを組み合わせたライトバック/ライトスルーキャッシュとすべきですね(実装がややこしくなりますが)。memcacheに書いたら、TQ経由でdatastoreにバックアップとか。

takezakitakezaki 2009/10/06 23:06 了解です。ありがとうございました。

はてなユーザーのみコメントできます。はてなへログインもしくは新規登録をおこなってください。