lifeLOG + REPOsitory: タスクキューを使う - 内部リンクを置き換える #3 (Blogger Glass)

内部リンクの置き換えをデータストアを使って単純に実装すると、「一覧画面」表示リクエストの処理に時間がかかるようになってしまい(データストアへの書き込みが発生するときのみ)、GAE の管理コンソール(の Current Load)に赤文字が出るようになってしまった(→「CPU 使い過ぎ？ - GAE 管理コンソール上の警告」参照)。

今回は、GAE のタスクキュー API を使って「一覧画面」リクエストの処理時間を短くすることに挑戦してみた。

タスクキュー API の使い方

GAE でタスクキュー API を利用して何らかの処理を行う際には以下の 3 つが必要になる。

タスクを登録するキューの設定
タスクを処理するハンドラの定義
タスクを登録するコードの追加

以下で、これらを順番に今回の実装を例に説明する。ただし、公式ドキュメントにもあるように、タスクキュー API はまだ experimental 扱いなので、この先、仕様変更の可能性は高い。ここに書かれている内容もいつまでも正しいとは限らない。

タスクを登録するキューの設定

GAE のタスクキュー API で使用するキューには、名前(name)、処理速度(rate)、実行量(bucket_size) を設定することができる。詳細については「Python Task Queue Configuration」を参照。設定は queue.yaml という名前の YAML ファイルに記述する。1 つの GAE アプリが持つことのできるキューは 10 までとなっている。

タスクキューにはデフォルトキューが用意されており、これを(デフォルトの設定のままで)使う分には特に設定しなくて良い(つまり、queue.yaml を作らなくても良い)。以下にデフォルトキューの(デフォルトの)設定を示す。

Default Queue Settings
項目	設定値
name	default
rate	5/s
bucket_size	5

簡単に言うと、デフォルトキューでは 1 秒間に 5 つのタスクを処理するようになっている。

今回の実装では、デフォルトキューをデフォルトのまま使っている。このため queue.yaml は作っていない。

タスクを処理するハンドラの定義

タスクを処理するコードは、通常のリクエストハンドラと同様に定義する(google.appengine.ext.webapp.RequestHandler を継承)。タスク登録時に指定しない限り、タスクは HTTP の POST メソッドで処理される。よって、タスクのハンドラでは最低限 post() 関数を定義することになる。

また、他のリクエストハンドラと同様に app.yaml にも登録する。以下は今回の実装で追加した定義だ。login: admin はリクエスト実行にアプリ管理者の権限を要求する設定。こうすることで、ユーザがタスクハンドラを直接実行することを防ぐことができる(タスクハンドラを呼び出すタスクキューは管理者権限を持っている)。

- url: /store_pol
  script: store_pol.py
  login: admin

今回、実装したタスクハンドラは以下の通り。リクエストパラメータとして permalink と Blog ID を受け取り、さらにひもづけられた Post ID を memcache から取り出して、データストアに保存している。また、データストアへの書き込みはトランザクションとして実行するようになっている。これはタスクが並列で処理されるからだ。

model.bind_post_id_with_permalink() 関数は以前に示したものから変わっていない。

# store_pol.py: process tasks to store a pair of links to the data store.

from google.appengine.api import memcache
from google.appengine.ext import db
from google.appengine.ext import webapp
from google.appengine.ext.webapp.util import run_wsgi_app

import model
import util

class StorePolHandler(webapp.RequestHandler):
    def post(self):
        permalink = self.request.get('permalink')
        blog_id = self.request.get('blog_id')
        post_id = util.get_post_id_from_memcache(permalink, blog_id)
        if post_id:
            def txn():
                model.bind_post_id_with_permalink(permalink, blog_id, post_id)
            db.run_in_transaction(txn)

def main():
    application = webapp.WSGIApplication([('/store_pol', StorePolHandler)],
                                         debug=True)
    run_wsgi_app(application)


if __name__ == '__main__':
    main()

タスクを登録するコードの追加

タスクを登録するには google.appengine.api.labs.taskqueue.add() 関数を使う(デフォルトキューの場合)。今回の実装では、permalink と Post ID の「ひもづけ情報」を登録する関数にこれを組み込んでいる(src/util.py 185 〜 186行目)。

def register_pair_of_links(entry):
    """
    Register a binding of a permalink and a post_id.
    The registration won't change, since the binding must be permanent.
    """
    permalink = entry.get_html_link().href
    blog_id = entry.get_blog_id()
    post_id = entry.get_post_id()
    # save the pair into the memcache, if it hasn't been stored.
    if memcache.get(permalink) is None:
        memcache.add(permalink, (blog_id, post_id))
    # issue a task, if the pair hasn't been stored yet in the data store.
    if model.get_post_id(permalink, blog_id) is None:
        taskqueue.add(url='/store_pol',
                      params={'permalink': permalink, 'blog_id': blog_id})

引数 url がタスクを処理するハンドラの URL 、params はハンドラに渡されるパラメータだ。また、URL は通常のリクエスト URL と同様、app.yaml に定義したタスクハンドラの URL と一致していなければならない。